From 9a79384958fb1e9fcb3e8ea8e14251f58e166037 Mon Sep 17 00:00:00 2001 From: Nicky Leach Date: Thu, 5 Nov 2026 13:38:55 -0800 Subject: [PATCH] Relay bounded Runner source refusal diagnostics (#19302) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - An environment plugin can submit a Runner continuation from a retained source. > - The host can refuse that continuation when its source is unverified or has an unresolved operation. > - The worker and Core currently discard that bounded reason. > - Raw provider errors can contain credentials, text or foreign identities. > - This pull request relays only a closed diagnostic and keeps the existing admission and redaction rules. ## Linked Issues or Issue Description Refs #19134 and #19157. This independent fix is based on master. A search found no duplicate relay. **What happened?** A source submit can receive HTTP 409 with `SESSION_STATE_UNAVAILABLE`. The worker drops its structured data. Core then replaces the error with a generic message. The caller cannot distinguish the two known diagnostic reasons safely. **Expected behavior** Carry only the fixed status, code and reason through SDK worker RPC and Core. Keep all provider text and identities out of the relayed error. A diagnostic must not authorize a retry or change state. **Steps to reproduce** Submit an authorized retained source through a task provider. Have the provider throw the recognized state-unavailable error. Read the error after worker RPC and Core dispatch. ## What Changed - Export `PluginEnvironmentTaskErrorDiagnostic`, `environmentTaskErrorData` and `readEnvironmentTaskErrorDiagnostic`. - Accept only `httpStatus: 409`, `errorCode: "SESSION_STATE_UNAVAILABLE"` and reason `"unresolved_operation"` or `"source_state_unverified"`. - Attach the versioned envelope only to an environment task `submit` with `resumeFrom`. Replace its provider message with a fixed message. - Revalidate and reconstruct the closed error in Core. Keep other failures under the existing generic redaction. - Document older SDK behavior and the diagnostic-only boundary. No grant, retry, discard, lifecycle or admission rule changes. ## Verification - Rebased head `7d02bb3e4c` includes master’s UI fixture fix from #19342. The sole conflict was in the worker integration test imports. It now imports both the physical-task result schema from #19300 and the diagnostic reader and RPC error type from this PR. The 39 SDK/worker cases and 57 Core/runtime cases passed, including PostgreSQL worker integration. The SDK build and Runner/server TypeScript checks passed. [Current-head CI](https://github.com/paperclipai/paperclip/actions/runs/42111889388) passed without retries. All 56 check-rollup entries were green, including Snyk and Socket; [standard Greptile review 2](https://github.com/paperclipai/paperclip/pull/19302#issuecomment-6665823815) scored this exact head **5/5**, with zero unresolved threads. The PR was marked ready on November 5 at 21:37 UTC. - Both new worker relay cases fail against the baseline worker and pass with the fix. - 46 SDK/worker cases passed, including both reasons, malformed/foreign envelopes, extra identities, unsafe getters and non-source operations. - 51 Core/runtime cases passed. The integration uses real SDK serialization and dispatch with PostgreSQL through the existing authorized retained-source submit. It verifies both reasons, malformed RPC rejection, extra-field removal, preserved source outcome and existing dispatch bookkeeping. A valid envelope on an ordinary submit or status stays generic. - Runner and SDK builds and server TypeScript check passed. An initial Runner build lacked its eval-kernel build prerequisite; it passed after that prerequisite was built. - The first expanded DB fixture used a shared environment name and failed its unique constraint. It now uses its unique fixture ID; both complete focused files pass. - Diff and added-line secret scans passed. [GitHub workflow 42083410387](https://github.com/paperclipai/paperclip/actions/runs/42083410387) passed on its second attempt. All CI checks completed, and Snyk passed. Standard Greptile scored exact head `38559ec4df8c` 5/5 with no unresolved threads. That result applies to the previous head. - The first GitHub attempt failed in the unchanged concurrent-budget case in `decision-models.test.ts` (one provider call observed, two expected). That exact case passed locally without code or assertion changes. The workspace shard also passed 4,960 tests before the unchanged CLI company import/export fixture exceeded its 30-second child-server startup wait. No port-collision cause is claimed; that fixture deliberately reserves the requested port. Both initial logs are retained. One failed-job-only retry passed without source changes. ## Risks Older SDKs keep the generic failure. The envelope is diagnostic evidence only. Provider operation and call identities are never accepted. Specific user-visible operation reporting must use Core's retained event or receipt under its existing authority. Master does not contain the new claimed predecessor orchestration. This PR tests the existing retained-source submission boundary. It does not claim the #19157 source recovery or old-journal end-to-end qualification. Runtime pins and deployments remain separate. ## Model Used OpenAI GPT-6.1 Sol (`gpt-6.1-sol`), high reasoning. Used repository tools, TypeScript, SDK RPC and real PostgreSQL tests. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used with version and capability details - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have described the issue in-PR using the bug template - [x] I have not referenced internal or instance-local Paperclip issues or links - [x] My branch name describes the change - [x] I have run the focused tests locally and they pass - [x] I have added tests for the new behavior and rejection paths - [x] I have updated relevant documentation - [x] I have considered and documented risks - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no unresolved findings - [x] I will address reviewer comments before requesting merge Co-authored-by: Paperclip --- doc/plugins/PLUGIN_ENVIRONMENT_TASKS.md | 12 ++++ .../plugins/sdk/src/environment-task-error.ts | 35 ++++++++++ packages/plugins/sdk/src/index.ts | 2 + packages/plugins/sdk/src/worker-rpc-host.ts | 8 ++- .../sdk/tests/environment-task-error.test.ts | 57 ++++++++++++++++ .../environment-task-runtime.test.ts | 11 +++- .../plugin-runner-task.integration.test.ts | 66 +++++++++++++++++-- .../src/services/environment-task-runtime.ts | 8 ++- 8 files changed, 191 insertions(+), 8 deletions(-) create mode 100644 packages/plugins/sdk/src/environment-task-error.ts create mode 100644 packages/plugins/sdk/tests/environment-task-error.test.ts diff --git a/doc/plugins/PLUGIN_ENVIRONMENT_TASKS.md b/doc/plugins/PLUGIN_ENVIRONMENT_TASKS.md index 62d97ad739c..557f3a3c46a 100644 --- a/doc/plugins/PLUGIN_ENVIRONMENT_TASKS.md +++ b/doc/plugins/PLUGIN_ENVIRONMENT_TASKS.md @@ -46,6 +46,18 @@ See [the schemas](../../packages/plugins/sdk/src/environment-tasks.ts) for all f | `stop` | `accepted` | Stop the task and its descendants. Fence a delayed submission. | | `complete` | `accepted` | Release task resources after Core saves required results. | +For a refused `submit` with `resumeFrom`, a driver can throw an error with +`status: 409`, `code: "SESSION_STATE_UNAVAILABLE"`, and `reason` set to +`"unresolved_operation"` or `"source_state_unverified"`. The SDK worker and +Core preserve only these closed diagnostic fields. They replace the provider +message and omit provider identities and text. Read the validated RPC envelope +with `readEnvironmentTaskErrorDiagnostic`. + +This diagnostic does not authorize a retry, replacement, cleanup or state +change. Other task failures keep the existing generic error. An older SDK +does not carry this diagnostic. Operation and call identities, when needed, +must come from Core's own retained records. + Each result must contain the task ID from the request. Core rejects a result with a different ID or result kind. A WSS URL must have no user information, query, or fragment. Return authentication in the secret headers. diff --git a/packages/plugins/sdk/src/environment-task-error.ts b/packages/plugins/sdk/src/environment-task-error.ts new file mode 100644 index 00000000000..21455ef9ec1 --- /dev/null +++ b/packages/plugins/sdk/src/environment-task-error.ts @@ -0,0 +1,35 @@ +/** Diagnostic only. A refusal never authorizes replacement, retry or cleanup. */ +export interface PluginEnvironmentTaskErrorDiagnostic { + httpStatus: 409; + errorCode: "SESSION_STATE_UNAVAILABLE"; + reason: "unresolved_operation" | "source_state_unverified"; +} + +const SCHEMA = "paperclip/environment-task-error/v1"; + +function field(value: unknown, key: string): unknown { + try { return value && typeof value === "object" ? (value as Record)[key] : undefined; } + catch { return undefined; } +} + +function sanitize(httpStatus: unknown, errorCode: unknown, reason: unknown): PluginEnvironmentTaskErrorDiagnostic | null { + if (httpStatus !== 409 || errorCode !== "SESSION_STATE_UNAVAILABLE" || + (reason !== "unresolved_operation" && reason !== "source_state_unverified")) return null; + return { httpStatus, errorCode, reason }; +} + +/** Worker side: copy only the closed refusal fields, never provider text or identities. */ +export function environmentTaskErrorData(error: unknown): { + schema: string; diagnostic: PluginEnvironmentTaskErrorDiagnostic; +} | undefined { + const diagnostic = sanitize(field(error, "status"), field(error, "code"), field(error, "reason")); + return diagnostic ? { schema: SCHEMA, diagnostic } : undefined; +} + +/** Host side: validate the envelope again after the worker RPC. */ +export function readEnvironmentTaskErrorDiagnostic(error: unknown): PluginEnvironmentTaskErrorDiagnostic | null { + const data = field(error, "data"); + if (field(data, "schema") !== SCHEMA) return null; + const diagnostic = field(data, "diagnostic"); + return sanitize(field(diagnostic, "httpStatus"), field(diagnostic, "errorCode"), field(diagnostic, "reason")); +} diff --git a/packages/plugins/sdk/src/index.ts b/packages/plugins/sdk/src/index.ts index 034a931c4c1..221cdecf043 100644 --- a/packages/plugins/sdk/src/index.ts +++ b/packages/plugins/sdk/src/index.ts @@ -468,6 +468,8 @@ export { PluginEnvironmentStopError, environmentStopErrorData, readEnvironmentSt export type { PluginEnvironmentStopFailure, PluginEnvironmentStopFailureReason } from "./environment-stop-error.js"; export { environmentResumeErrorData, readEnvironmentResumeErrorDiagnostic } from "./environment-resume-error.js"; export type { PluginEnvironmentResumeErrorDiagnostic } from "./environment-resume-error.js"; +export { environmentTaskErrorData, readEnvironmentTaskErrorDiagnostic } from "./environment-task-error.js"; +export type { PluginEnvironmentTaskErrorDiagnostic } from "./environment-task-error.js"; export type { AiConnectionPool, AiConnectionPoolConfig, AiConnectionPoolMember, AiConnectionRouterRequest, AiConnectionRouterResult, AiConnectionRouterSelection } from "@paperclipai/shared"; diff --git a/packages/plugins/sdk/src/worker-rpc-host.ts b/packages/plugins/sdk/src/worker-rpc-host.ts index cc8b212230a..7c34deaf807 100644 --- a/packages/plugins/sdk/src/worker-rpc-host.ts +++ b/packages/plugins/sdk/src/worker-rpc-host.ts @@ -9,6 +9,7 @@ import { environmentSyncErrorData, withEnvironmentSyncErrorCapture } from "./env import { environmentRunnerIngressErrorData } from "./environment-runner-ingress-error.js"; import { environmentStopErrorData } from "./environment-stop-error.js"; import { environmentResumeErrorData } from "./environment-resume-error.js"; +import { environmentTaskErrorData } from "./environment-task-error.js"; import { createInvocationTracer } from "./invocation-tracer.js"; import { createPluginIdleDrain } from "./idle-drain.js"; /** @@ -1635,7 +1636,10 @@ export function startWorkerRpcHost(options: WorkerRpcHostOptions): WorkerRpcHost : await invoke(); sendMessage(createSuccessResponse(id, result ?? null)); } catch (err) { - const errorMessage = err instanceof Error ? err.message : String(err); + const input = params as Partial | undefined; + const taskError = method === "environmentTask" && input?.operation?.kind === "submit" && input.operation.resumeFrom + ? environmentTaskErrorData(err) : undefined; + const errorMessage = taskError ? "Plugin Runner source state unavailable" : err instanceof Error ? err.message : String(err); // Propagate specific error codes from handler errors (e.g. // METHOD_NOT_FOUND, METHOD_NOT_IMPLEMENTED) — fall back to // WORKER_ERROR for untyped exceptions. @@ -1650,7 +1654,7 @@ export function startWorkerRpcHost(options: WorkerRpcHostOptions): WorkerRpcHost : method === "environmentSyncOut" ? environmentSyncErrorData(err) : method === "environmentRunnerIngressEndpoint" ? environmentRunnerIngressErrorData(err) : method === "environmentStopLease" ? environmentStopErrorData(err) - : method === "environmentResumeLease" ? environmentResumeErrorData(err) : undefined)); + : method === "environmentResumeLease" ? environmentResumeErrorData(err) : taskError)); } finally { done?.(); } diff --git a/packages/plugins/sdk/tests/environment-task-error.test.ts b/packages/plugins/sdk/tests/environment-task-error.test.ts new file mode 100644 index 00000000000..27b16910dfb --- /dev/null +++ b/packages/plugins/sdk/tests/environment-task-error.test.ts @@ -0,0 +1,57 @@ +import { createInterface } from "node:readline"; +import { PassThrough } from "node:stream"; +import { describe, expect, it } from "vitest"; +import { definePlugin } from "../src/define-plugin.js"; +import { environmentTaskErrorData, readEnvironmentTaskErrorDiagnostic } from "../src/environment-task-error.js"; +import { createRequest, isJsonRpcResponse, JsonRpcCallError, parseMessage, serializeMessage, type JsonRpcResponse } from "../src/protocol.js"; +import { startWorkerRpcHost } from "../src/worker-rpc-host.js"; + +const submit = { kind: "submit", preparedLaunchId: "configuration", projectIds: [], bootstrapTicket: "fixture-ticket", resumeFrom: "source-task", + runner: { protocolMin: 1, protocolMax: 3, harness: "codex", runnerId: "runner", leaseId: "lease", runId: "run", + sessionId: "session", turnId: "turn", itemId: "item" } }; +const failure = (reason = "unresolved_operation") => Object.assign(new Error("synthetic-private-provider-text"), { + status: 409, code: "SESSION_STATE_UNAVAILABLE", reason, operationId: "forged-operation", callId: "forged-call", +}); + +async function invoke(error: unknown, operation: unknown = submit) { + const stdin = new PassThrough(), stdout = new PassThrough(), lines = createInterface({ input: stdout }); + const worker = startWorkerRpcHost({ stdin, stdout, plugin: definePlugin({ async setup() {}, async onEnvironmentTask() { throw error; } }) }); + try { + const response = new Promise(resolve => lines.on("line", line => { + const message = parseMessage(line); + if (isJsonRpcResponse(message)) resolve(message); + })); + stdin.write(serializeMessage(createRequest("environmentTask", { taskId: "task", operation }, "task-error"))); + const reply = await response; + if (!("error" in reply) || !reply.error) throw new Error("Expected RPC error"); + return reply.error; + } finally { worker.stop(); lines.close(); stdin.destroy(); stdout.destroy(); } +} + +describe("source task refusal diagnostics", () => { + it.each(["unresolved_operation", "source_state_unverified"])("carries only the closed %s diagnostic across real worker RPC", async reason => { + const reply = await invoke(failure(reason)); + expect(readEnvironmentTaskErrorDiagnostic(new JsonRpcCallError(reply))).toEqual({ httpStatus: 409, errorCode: "SESSION_STATE_UNAVAILABLE", reason }); + expect(reply.message).toBe("Plugin Runner source state unavailable"); + expect(JSON.stringify(reply)).not.toMatch(/synthetic-private|forged/); + }); + + it.each([{ kind: "status" }, { ...submit, resumeFrom: undefined }])("does not attach the envelope outside a source submit", async operation => { + expect(await invoke(failure(), operation)).not.toHaveProperty("data"); + }); + + it.each([null, {}, { ...failure(), status: 503 }, { ...failure(), code: "OTHER" }, { ...failure(), reason: "provider text" }])( + "rejects unrecognized failures %j", error => { expect(environmentTaskErrorData(error)).toBeUndefined(); }, + ); + + it("does not trust a malformed envelope or copy its extra fields", () => { + expect(readEnvironmentTaskErrorDiagnostic({ data: { schema: "foreign", diagnostic: failure() } })).toBeNull(); + expect(readEnvironmentTaskErrorDiagnostic({ data: { schema: "paperclip/environment-task-error/v1", diagnostic: { + httpStatus: 409, errorCode: "SESSION_STATE_UNAVAILABLE", reason: "provider text", + } } })).toBeNull(); + const data = environmentTaskErrorData(failure())!; + expect(readEnvironmentTaskErrorDiagnostic({ data: { ...data, diagnostic: { ...data.diagnostic, operationId: "forged" } } })) + .toEqual(data.diagnostic); + expect(environmentTaskErrorData(Object.defineProperty({}, "status", { get() { throw new Error("unsafe getter"); } }))).toBeUndefined(); + }); +}); diff --git a/server/src/__tests__/environment-task-runtime.test.ts b/server/src/__tests__/environment-task-runtime.test.ts index 8a175531594..38d455c57d4 100644 --- a/server/src/__tests__/environment-task-runtime.test.ts +++ b/server/src/__tests__/environment-task-runtime.test.ts @@ -1,7 +1,7 @@ import { beforeEach, describe, expect, it, vi } from "vitest"; import { PgDialect } from "drizzle-orm/pg-core"; import { executionWorkspaces, environmentLeases, type Db } from "@paperclipai/db"; -import { environmentTaskOperationSchema, parseEnvironmentTaskResult } from "@paperclipai/plugin-sdk"; +import { environmentTaskOperationSchema, parseEnvironmentTaskResult, JsonRpcCallError, readEnvironmentTaskErrorDiagnostic } from "@paperclipai/plugin-sdk"; import { executeEnvironmentTask, settleDispatchedEnvironmentTask } from "../services/environment-task-runtime.js"; import { NativeExecutionContext } from "../services/native-runtime/native-execution-context.js"; @@ -503,6 +503,15 @@ describe("plugin-provided Runner execution", () => { const workers = worker(); workers.call.mockRejectedValue(new Error("private-credential")); await expect(executeEnvironmentTask(database().db, workers as never, { companyId: "company", leaseId, operation: submit })).rejects.toThrow(/^Plugin-provided Runner execution operation unavailable; reconcile the same task before retrying$/); }); + it.each([submit, { kind: "status" } as const])("redacts even a valid source envelope outside a source submit (%j)", async operation => { + const workers = worker(); + workers.call.mockRejectedValue(new JsonRpcCallError({ code: -32000, message: "synthetic-private-text", data: { + schema: "paperclip/environment-task-error/v1", diagnostic: { httpStatus: 409, errorCode: "SESSION_STATE_UNAVAILABLE", reason: "unresolved_operation" }, + } })); + const error = await executeEnvironmentTask(database().db, workers as never, { companyId: "company", leaseId, operation }).catch(error => error); + expect(error.message).toBe("Plugin-provided Runner execution operation unavailable; reconcile the same task before retrying"); + expect(readEnvironmentTaskErrorDiagnostic(error)).toBeNull(); + }); it("accepts PRP identity characters and length limits", () => { for (const field of ["runnerId", "leaseId", "runId", "sessionId", "turnId", "itemId"]) { for (const value of ["session:part.1", "a".repeat(160)]) { diff --git a/server/src/__tests__/plugin-runner-task.integration.test.ts b/server/src/__tests__/plugin-runner-task.integration.test.ts index a9196cab70a..f612b295144 100644 --- a/server/src/__tests__/plugin-runner-task.integration.test.ts +++ b/server/src/__tests__/plugin-runner-task.integration.test.ts @@ -7,12 +7,12 @@ import { eq } from "drizzle-orm"; import { agents, companies, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, environmentLeases, environments, heartbeatRuns, plugins, projects } from "@paperclipai/db"; import { definePlugin } from "../../../packages/plugins/sdk/src/define-plugin.js"; import { startWorkerRpcHost } from "../../../packages/plugins/sdk/src/worker-rpc-host.js"; -import { createRequest, parseMessage, serializeMessage, isJsonRpcResponse, isJsonRpcSuccessResponse } from "../../../packages/plugins/sdk/src/protocol.js"; -import { environmentTaskOperationSchema, environmentTaskResultSchema, type PluginEnvironmentTaskOperation } from "@paperclipai/plugin-sdk"; +import { createRequest, parseMessage, serializeMessage, isJsonRpcResponse, isJsonRpcSuccessResponse, JsonRpcCallError } from "../../../packages/plugins/sdk/src/protocol.js"; +import { environmentTaskOperationSchema, environmentTaskResultSchema, readEnvironmentTaskErrorDiagnostic, type PluginEnvironmentTaskOperation } from "@paperclipai/plugin-sdk"; import type { PaperclipPluginManifestV1 } from "@paperclipai/shared"; import { agentEnvironmentTestService } from "../services/agent-environment-test.js"; import { heartbeatService } from "../services/heartbeat.js"; -import { remoteExecutionHasStopped } from "../services/remote-execution-termination.js"; +import { remoteExecutionHasStopped, remoteTerminationReceipt } from "../services/remote-execution-termination.js"; import { resolvePluginTaskResume } from "../services/environment-task-runtime.js"; import { environmentRuntimeService } from "../services/environment-runtime.js"; import { environmentRunOrchestrator } from "../services/environment-run-orchestrator.js"; @@ -41,7 +41,7 @@ function workerHarness(plugin: ReturnType, manifest: Paperc const waiter = pending.get(Number(message.id)); pending.delete(Number(message.id)); if (isJsonRpcSuccessResponse(message)) waiter?.resolve(message.result); - else waiter?.reject(new Error(message.error.message)); + else waiter?.reject(new JsonRpcCallError(message.error)); } }); const call = (method: string, params: unknown) => new Promise((resolve, reject) => { @@ -63,6 +63,64 @@ databaseSuite("Core plugin Runner task lifecycle", () => { }, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS); afterAll(async () => { await stopDb?.(); }); + it.each(["unresolved_operation", "source_state_unverified", "malformed_rpc", "extra_rpc_fields"] as const)("relays only the closed source refusal through Core (%s)", async mode => { + const reason = mode === "source_state_unverified" ? mode : "unresolved_operation"; + const companyId = randomUUID(), agentId = randomUUID(), pluginId = randomUUID(), environmentId = randomUUID(); + const sourceRunId = randomUUID(), runId = randomUUID(), sourceLeaseId = randomUUID(), leaseId = randomUUID(); + const manifest: PaperclipPluginManifestV1 = { id: pluginId, apiVersion: 1, version: "1.0.0", displayName: "Task diagnostics", + description: "Synthetic task provider", author: "Paperclip", categories: ["automation"], + capabilities: ["environment.drivers.register"], entrypoints: { worker: "fixture" }, + environmentDrivers: [{ driverKey: "tasks", displayName: "Tasks", supportsTasks: true, configSchema: { type: "object" } }] }; + await db.insert(companies).values({ id: companyId, name: "Diagnostic fixture", issuePrefix: `D${companyId.slice(0, 7).replace(/-/g, "").toUpperCase()}` }); + await db.insert(agents).values({ id: agentId, companyId, name: "Fixture", adapterType: "paperclip_runner" }); + await db.insert(environments).values({ id: environmentId, name: `Diagnostic ${environmentId}`, driver: "plugin", config: {} }); + await db.insert(plugins).values({ id: pluginId, pluginKey: pluginId, packageName: "fixture", version: "1.0.0", status: "ready", manifestJson: manifest }); + await db.insert(heartbeatRuns).values([ + { id: sourceRunId, companyId, agentId, invocationSource: "manual", status: "succeeded", resultJson: { outcome: "preserved" } }, + { id: runId, companyId, agentId, invocationSource: "manual", status: "running" }, + ]); + const pin = { driver: "plugin", pluginId, driverKey: "tasks", runnerTaskLease: true, + runnerTaskProjectAuthValidation: { configurationId: "fixture", provider: "openai", projectIds: [] } }; + const source = { id: sourceLeaseId, companyId, environmentId, heartbeatRunId: sourceRunId, provider: "plugin:fixture:tasks", providerLeaseId: "source-task" }; + await db.insert(environmentLeases).values([ + { ...source, status: "released", leasePolicy: "ephemeral", cleanupStatus: "success", releasedAt: new Date(), metadata: { ...pin, + runnerTaskSessionRetained: true, runnerTaskIdentity: { runnerId: "runner", leaseId: sourceLeaseId, runId: sourceRunId, sessionId: "session", turnId: "old-turn", itemId: "old-item" }, + remoteExecutionTermination: remoteTerminationReceipt(source, { providerLeaseId: "source-task", state: "stopped" }) } }, + { id: leaseId, companyId, environmentId, heartbeatRunId: runId, provider: source.provider, providerLeaseId: "current-task", status: "active", metadata: pin }, + ]); + const dispatched: string[] = []; + const rpc = workerHarness(definePlugin({ async setup() {}, async onEnvironmentTask(params) { + dispatched.push(params.operation.kind); + throw Object.assign(new Error("synthetic-private-provider-text"), { status: 409, code: "SESSION_STATE_UNAVAILABLE", reason, + operationId: "forged-operation", callId: "forged-call", taskId: "foreign-task" }); + } }), manifest); + const workers = { getWorker: () => ({ supportedMethods: ["environmentTask"] }), + call: (_id: string, method: string, params: unknown) => { + if (mode === "malformed_rpc" || mode === "extra_rpc_fields") throw new JsonRpcCallError({ code: -32000, + message: "synthetic-private-provider-text", data: { schema: "paperclip/environment-task-error/v1", diagnostic: { + httpStatus: 409, errorCode: "SESSION_STATE_UNAVAILABLE", reason: mode === "malformed_rpc" ? "provider text" : reason, + operationId: "forged-operation", callId: "forged-call", taskId: "foreign-task", + } } }); + return rpc.call(method, params); + } } as unknown as PluginWorkerManager; + try { + const runtime = environmentRuntimeService(db, { pluginWorkerManager: workers }); + const error = await runtime.task({ companyId, leaseId, operation: { kind: "submit", resumeFrom: "source-task", preparedLaunchId: "fixture", + projectIds: [], bootstrapTicket: "fixture-ticket", runner: { runnerId: "runner", leaseId: sourceLeaseId, runId, sessionId: "session", + turnId: "turn", itemId: "item", harness: "codex", protocolMin: 1, protocolMax: 3 } } }).catch(error => error); + expect(readEnvironmentTaskErrorDiagnostic(error)).toEqual(mode === "malformed_rpc" ? null : { httpStatus: 409, errorCode: "SESSION_STATE_UNAVAILABLE", reason }); + expect(error.message).toBe(mode === "malformed_rpc" + ? "Plugin-provided Runner execution operation unavailable; reconcile the same task before retrying" : "Plugin Runner source state unavailable"); + expect(JSON.stringify(error)).not.toMatch(/synthetic-private|forged|foreign-task/); + expect(dispatched).toEqual(mode === "malformed_rpc" || mode === "extra_rpc_fields" ? [] : ["submit"]); + const [savedSource] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, sourceRunId)); + expect(savedSource).toMatchObject({ status: "succeeded", resultJson: { outcome: "preserved" } }); + const [current] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, leaseId)); + expect(current.metadata).toMatchObject({ runnerTaskDispatched: true, runnerTaskExecutionStopped: false }); + expect(JSON.stringify(current.metadata)).not.toMatch(/synthetic-private|forged|unresolved_operation|source_state_unverified/); + } finally { rpc.stop(); } + }); + it("does not allocate agent-dependent setup leases before the agent is prepared", async () => { const companyId = randomUUID(), pluginId = randomUUID(), environmentId = randomUUID(), agentId = randomUUID(); await db.insert(companies).values({ id: companyId, name: "Setup lease fixture", issuePrefix: "SETUP" }); diff --git a/server/src/services/environment-task-runtime.ts b/server/src/services/environment-task-runtime.ts index 4b6489243cc..64af59a9ee2 100644 --- a/server/src/services/environment-task-runtime.ts +++ b/server/src/services/environment-task-runtime.ts @@ -9,6 +9,7 @@ import { settlePluginTaskCredentialHandoff } from "./plugin-task-credential-writ import { and, eq, inArray, sql } from "drizzle-orm"; import { environmentLeases, executionWorkspaces, environments, heartbeatRuns, issues, projects, type Db } from "@paperclipai/db"; import { freshWorkspaceSchema, environmentTaskOperationSchema, parseEnvironmentTaskResult, + environmentTaskErrorData, readEnvironmentTaskErrorDiagnostic, JsonRpcCallError, PLUGIN_RPC_ERROR_CODES, type HostToWorkerMethodName, type HostToWorkerMethods, type PluginEnvironmentTaskOperation } from "@paperclipai/plugin-sdk"; import { pluginRegistryService } from "./plugin-registry.js"; import { remoteTerminationReceipt } from "./remote-execution-termination.js"; @@ -453,7 +454,12 @@ export async function executeEnvironmentTask(db: Db, workers: PluginWorkerManage eq(environmentLeases.providerLeaseId, taskId), sql`${environmentLeases.metadata}->>'runnerTaskExecutionStopped' = 'true'`)); } return parsed; - } catch { + } catch (error) { + const diagnostic = operation.kind === "submit" && operation.resumeFrom ? readEnvironmentTaskErrorDiagnostic(error) : null; + if (diagnostic) throw new JsonRpcCallError({ code: PLUGIN_RPC_ERROR_CODES.WORKER_ERROR, + message: "Plugin Runner source state unavailable", + data: environmentTaskErrorData({ status: diagnostic.httpStatus, code: diagnostic.errorCode, reason: diagnostic.reason }), + }); // Worker errors and schema diagnostics can contain request credentials. throw new Error("Plugin-provided Runner execution operation unavailable; reconcile the same task before retrying"); }