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"); }