Relay bounded Runner source refusal diagnostics (#19302)

## 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 <noreply@paperclip.ing>

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.

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<string, unknown>)[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"));
+}

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

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<PluginEnvironmentTaskParams> | 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?.();
     }

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<JsonRpcResponse>(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();
+  });
+});

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)]) {

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<typeof definePlugin>, 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<any>((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" });

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