From b21884bdcf51e547e669aea0227bbd60dba28733 Mon Sep 17 00:00:00 2001 From: Nicky Leach Date: Fri, 6 Nov 2026 20:45:02 -0800 Subject: [PATCH] Release and resume retained plugin Runner sessions in Core (#19611) --- doc/plugins/PLUGIN_ENVIRONMENT_TASKS.md | 53 ++- .../run-vitest-stable-shard.test.mjs | 6 +- scripts/run-vitest-stable.mjs | 1 + .../native-session-executor.test.ts | 322 +++++++++++++++- .../native-runtime/native-session-executor.ts | 345 ++++++++++++++---- ...-runner-release-resume.integration.test.ts | 315 ++++++++++++++++ .../remote-warm-session-recovery.ts | 3 +- 7 files changed, 965 insertions(+), 80 deletions(-) create mode 100644 server/src/services/native-runtime/plugin-runner-release-resume.integration.test.ts diff --git a/doc/plugins/PLUGIN_ENVIRONMENT_TASKS.md b/doc/plugins/PLUGIN_ENVIRONMENT_TASKS.md index 987edb8386a..5460fe8a443 100644 --- a/doc/plugins/PLUGIN_ENVIRONMENT_TASKS.md +++ b/doc/plugins/PLUGIN_ENVIRONMENT_TASKS.md @@ -117,20 +117,47 @@ warm or source recovery capabilities do not imply support. Older strict Core schemas reject this description field, so coordinate advertisement with a compatible Core version. -Core's lease-bound task owner exposes capability-gated `release` and `resume` -methods for both Codex and ACPX. They require `releaseResumeVersion: 1`, warm-task -support and a selected account, and validate the operation result and exact task -ID with the SDK parser. These methods are a forwarding facade; Core does not yet -call them automatically or change its warm-session timeout policy. +Core uses this pair only for explicitly supported warm tasks with a selected +account. It releases after the turn's required events, tool results, journal +writes and cleanup have settled. It keeps the exact session busy until the host +confirms release and Core records that source. Existing tasks without the +capability retain their ordinary warm lifecycle. -The caller must serialize settled release against new work and disposal, quiet -its owned connection's liveness checks and reconnect loop before release, and -retain the same pending ticket for ambiguous resume retries. Ordinary owner -rebind must first check the new run's account, credentials and project scope. -After resume, authenticate the source with the fresh ticket before warm readiness -and ordinary `run.attach`. An RPC error does not prove that release or resume had -no effect; it must not trigger reconnect, a replacement ticket or turn replay. -The facade does not supply this orchestration or enable provider retention. +Before release, Core stops connection pings and reconnect work while keeping the +existing receive/ACK path open. A confirmed release closes the route and stops +its idle polling. An ambiguous release stays with the original run for bounded +reconciliation. It cannot pass to a new run's task scope. Cancellation or an +expired reconciliation budget uses ordinary host stop and physical cleanup; +it does not change an already completed business result. + +Resume first checks the new run's account, credential revision and project +scope. It uses one fresh transient ticket, preserving that exact ticket through +ambiguous replies. Core authenticates the source before warm readiness, provider +preparation and ordinary `run.attach`; no new turn can pass that handover. +Generic RPC errors do not prove refusal or permit a new ticket, reconnect or +turn replay. A known `unavailable` result still requires ordinary physical +cleanup. Transparent cold fallback remains a separate qualification requirement. + +A normal Core cache timeout detaches only Core's quiet local owner. The host +keeps its retained resources under its own retention policy. Core's existing +completed-session record stores only the confirmed source, physical task and a +digest of the selected account, credential revision and workspace scope. The +transport returns the exact source after the host's release receipt; Core does +not infer it from a local checkpoint marker, which may still name an older run. +An invalid source receipt or failed proof write keeps cleanup under the same +owner until physical retirement finishes. Later +cache adoption validates that Core-authored proof before claiming or +reconnecting. A stale or missing proof never licenses a claim. A changed +selection remains blocked until guarded retirement/recovery; it does not imply +that a cold launch is safe. Confirmed claim or physical cleanup clears only the +matching proof. Raw tickets are never stored in this record. + +These Core paths do not advertise the capability or opt ordinary tasks into a +private host channel. Existing durable lease and plugin-worker idle checks still +apply; retaining a released Runner does not by itself make a tenant safe to sleep. +Provider recovery, bounded host retention and deployment qualification are still +required before activation. Recovery from a Core crash +between claim and source authentication is not qualified by this integration. ## Error codes diff --git a/scripts/__tests__/run-vitest-stable-shard.test.mjs b/scripts/__tests__/run-vitest-stable-shard.test.mjs index 5a083997bf5..e3e923635ce 100644 --- a/scripts/__tests__/run-vitest-stable-shard.test.mjs +++ b/scripts/__tests__/run-vitest-stable-shard.test.mjs @@ -324,6 +324,7 @@ const ingressRetrySuitePath = "server/src/services/native-runtime/native-ingress const usageDeliverySuitePath = "server/src/services/native-runtime/native-usage-delivery.integration.test.ts"; const warmCheckpointOwnershipSuitePath = "server/src/services/native-runtime/native-warm-checkpoint-ownership.integration.test.ts"; const pluginWarmOwnerSuitePath = "server/src/services/native-runtime/plugin-runner-task-warm.integration.test.ts"; +const pluginReleaseResumeSuitePath = "server/src/services/native-runtime/plugin-runner-release-resume.integration.test.ts"; const dotRunnerSuitePath = "server/src/__tests__/dot-runner.test.ts"; const restartRecoverySuitePath = "server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts"; @@ -373,7 +374,8 @@ test("12 PR without-chat shards plus the dedicated chat and native-runner lanes assert.ok(!files.includes(usageDeliverySuitePath)); assert.ok(!files.includes(warmCheckpointOwnershipSuitePath)); assert.ok(!files.includes(pluginWarmOwnerSuitePath)); - assert.deepEqual([...files, chatSuitePath, nativeRunnerSuitePath, ingressRetrySuitePath, dotRunnerSuitePath, restartRecoverySuitePath, ...restartWaitSuites, usageDeliverySuitePath, warmCheckpointOwnershipSuitePath, pluginWarmOwnerSuitePath, "server/src/services/native-runtime/plugin-runner-probe.test.ts"].sort(), full.selectedGeneralServerSuites.sort()); + assert.ok(!files.includes(pluginReleaseResumeSuitePath)); + assert.deepEqual([...files, chatSuitePath, nativeRunnerSuitePath, ingressRetrySuitePath, dotRunnerSuitePath, restartRecoverySuitePath, ...restartWaitSuites, usageDeliverySuitePath, warmCheckpointOwnershipSuitePath, pluginWarmOwnerSuitePath, pluginReleaseResumeSuitePath, "server/src/services/native-runtime/plugin-runner-probe.test.ts"].sort(), full.selectedGeneralServerSuites.sort()); assert.equal(new Set(files).size, files.length); const defaultRun = dryRunJson([], prEnv); assert.ok(defaultRun.generalServerSuiteCount === full.generalServerSuiteCount); @@ -395,6 +397,7 @@ for (const [caller, envOverrides] of [["Release", { GITHUB_WORKFLOW: "Release" } assert.ok(files.includes(restartRecoverySuitePath)); assert.ok(files.includes(warmCheckpointOwnershipSuitePath)); assert.ok(files.includes(pluginWarmOwnerSuitePath)); + assert.ok(files.includes(pluginReleaseResumeSuitePath)); assert.deepEqual([...files, chatSuitePath].sort(), full.selectedGeneralServerSuites.sort()); assert.equal(new Set(files).size, files.length); }); @@ -429,6 +432,7 @@ test("the native-runner lane runs exactly the cargo-dependent suites", () => { usageDeliverySuitePath, warmCheckpointOwnershipSuitePath, pluginWarmOwnerSuitePath, + pluginReleaseResumeSuitePath, ]); }); diff --git a/scripts/run-vitest-stable.mjs b/scripts/run-vitest-stable.mjs index 5c5709af2d1..897f0563ef9 100644 --- a/scripts/run-vitest-stable.mjs +++ b/scripts/run-vitest-stable.mjs @@ -106,6 +106,7 @@ const nativeRunnerSuites = [ "server/src/services/native-runtime/native-usage-delivery.integration.test.ts", "server/src/services/native-runtime/native-warm-checkpoint-ownership.integration.test.ts", "server/src/services/native-runtime/plugin-runner-task-warm.integration.test.ts", + "server/src/services/native-runtime/plugin-runner-release-resume.integration.test.ts", ]; // In the PR workflow (pr.yml, the caller of pr-trusted.yml — reusable // workflows inherit the caller's GITHUB_WORKFLOW), the last Verify Paperclip diff --git a/server/src/services/native-runtime/native-session-executor.test.ts b/server/src/services/native-runtime/native-session-executor.test.ts index 93f8c455e77..e251185499b 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -6,6 +6,7 @@ import * as harnessStorage from "./native-harness-backup-storage.js"; import * as stagedAttachmentRecovery from "./native-staged-attachment-recovery.js"; import * as controllerRecovery from "./native-controller-recovery.js"; import { isSettledRemoteWarmControlState } from "./remote-warm-session-recovery.js"; +import * as remoteWarmRecoveryRuntime from "./remote-warm-session-recovery.js"; import { DurablePrpControlPlane } from "../../vendor/paperclip-runner/index.js"; import * as environmentTaskRuntime from "../environment-task-runtime.js"; import * as pluginTaskRuntime from "./plugin-runner-task.js"; @@ -171,7 +172,7 @@ type BackendFactoryOptions = { type RunnerTransportOptions = { authorityRunId?: string; onAccountingEvent?: (event: PrpEvent) => Promise; - onRecoveryAccountingEvent?: unknown; + onRecoveryAccountingEvent?: (event: PrpEvent, identity: import("../../vendor/paperclip-runner/index.js").DurableRecoveryIdentity) => Promise; environment?: NodeJS.ProcessEnv; acpxRuntimeDirectory?: string; adoptExistingRunner?: { pid: number; isAlive: () => Promise | boolean }; @@ -8381,6 +8382,13 @@ describe("native terminal-turn accounting", () => { expect(firstCapture).toHaveBeenCalledTimes(2); expect(secondCapture).toHaveBeenCalledTimes(1); await expect(retainedHook({ ...firstReport, runId: "unrelated-run" })).rejects.toThrow("native_usage_run_binding_mismatch"); + const recoveryHook = state.createTransport.mock.calls.at(-1)![0].onRecoveryAccountingEvent!; + const source = { runnerInstanceId: "warm-producer", environmentLeaseId: "warm-usage-workspace", + runId: first.binding.runId, normalizedSessionId: first.session.normalizedSessionId!, turnId: "source-turn", itemId: "source-item" }; + await recoveryHook(firstReport, source); + expect(firstCapture).toHaveBeenCalledTimes(3); + expect(secondCapture).toHaveBeenCalledTimes(1); + await expect(recoveryHook({ ...firstReport, runId: "foreign-source" }, source)).rejects.toThrow("native_recovery_usage_source_unproven"); } finally { if (previous === undefined) delete process.env.PAPERCLIP_RUNNER_STATE_DIR; else process.env.PAPERCLIP_RUNNER_STATE_DIR = previous; await rm(directory, { recursive: true, force: true }); @@ -15857,6 +15865,318 @@ describe("runnerd provider runtime wiring", () => { } finally {owner.mockRestore();await rm(home,{recursive:true,force:true});} }); + it.each((["codex", "claude"] as const).flatMap(provider => [[false, false, false, false], [true, false, false, false], [false, true, false, false], [false, false, true, false], ...(provider === "codex" ? [[false, false, false, true]] : [])].map(([unknown, shutdown, journalFailure, deadline]) => [provider, unknown, shutdown, journalFailure, deadline] as const)))( + "releases settled %s only under its current owner and detaches its cache locally (unknown=%s, shutdown=%s, journalFailure=%s, deadline=%s)", async (provider, unknown, shutdown, journalFailure, deadline) => { + await closeIdleWarmNativeSessionsForRestart(); + expect(readNativeSessionIdleWork()).toBe("none"); + const previousHome = process.env.PAPERCLIP_HOME; + process.env.PAPERCLIP_HOME = isolatedStateDirectory; + const current = scopedNativeExecution(`released-${provider}-${unknown}`); + current.session.lifecyclePolicy = { mode: "warm", idleTimeoutMs: unknown ? 750 : 250 }; + if (provider === "claude") current.provider = { kind: "acpx", agent: "claude", model: "fixture", permissionMode: "deny-all" } as never; + const source = { runnerInstanceId: "released-runner", environmentLeaseId: "released-lease", runId: current.binding.runId, + normalizedSessionId: current.session.normalizedSessionId!, turnId: "settled-turn", itemId: "settled-item" }; + const root = await initializeScopedRunnerRoot(current, source.runnerInstanceId); + await writeRunnerAuthority(root, source, "ready"); + const home = join(isolatedStateDirectory, "selected-account"); + await mkdir(home, { recursive: true }); + await writeFile(join(home, "auth.json"), JSON.stringify({ OPENAI_API_KEY: "synthetic-fixture" }), { mode: 0o600 }); + const task = vi.fn(async (_operation: unknown) => ({ kind: "description", taskId: "retained-task", + launchPreparationVersion: 1, projectAuthCheckVersion: 1, warmTaskVersion: 1, releaseResumeVersion: 1, + runner: { version: "fixture", digest: `sha256:${"a".repeat(64)}` } })); + const target = { kind: "remote" as const, transport: "sandbox" as const, remoteCwd: ".", environmentId: "retained-environment", + leaseId: source.environmentLeaseId, runnerTask: task, sandboxLeaseAcquisition: { outcome: "resumed" as const, providerLeaseId: "retained-task" } }; + const owners = vi.spyOn(pluginTaskRuntime, "createPluginRunnerTaskOwner"); + let allowRelease!: () => void; + const receipt = new Promise(resolve => { allowRelease = resolve; }); + const updates: Array<{ table: unknown; values: Record }> = []; + let attempts = 0; + const release = vi.fn(async () => { + expect(owners.mock.calls.at(-1)![0].authorityRunId).toBe(current.binding.runId); + expect(updates.some(update => update.values.leaseOwner === null && update.values.recoveryState === null)).toBe(true); + if (unknown && attempts++ === 0) throw new Error("fixture lost release reply"); + await receipt; + return structuredClone(source); + }); + let allowPhysicalStop!: () => void; + const physicalStop = new Promise(resolve => { allowPhysicalStop = resolve; }); + const close = vi.fn(async () => { if (shutdown || deadline || journalFailure) await physicalStop; }), detachControllerForRestart = vi.fn(async () => {}); + const session = { release, close, detachControllerForRestart }; + const validated = validatePrpStructuredRunResult({ schema: "paperclip.run_result.v1", reportedWorkDisposition: "done", + summary: "Completed", completionClaim: { contractRevision: current.completionContract.contract.revision, + objectiveSatisfied: true, criteria: [], remainingWork: [] }, evidence: [], verification: [], artifacts: [], attentionRequests: [] }); + if (!validated.ok) throw new Error("Invalid completion fixture"); + state.execute.mockReset().mockImplementationOnce(async options => { + await options.onCheckpoint?.({ backendKind: "runner", driverKind: current.session.driverKind, sessionId: "driver-session", + identity: { ...current.binding, sessionId: source.normalizedSessionId }, providerSessionId: "provider-session", + activeTurnId: null, pendingRuntimeRequests: [] }); + await options.onSession?.(session); + return { result: validated.result, terminal: { runTerminalState: "succeeded" }, turnId: source.turnId, + normalizedSessionId: source.normalizedSessionId, providerSessionId: "provider-session", driverKind: "test", driverVersion: "1", + nativeEventCount: 1, highestContiguousSourceSeq: 1, usage: null }; + }); + const execute = executePaperclipNativeSession({ db: leaseDb(current, {}, {}, updates), execution: current, + runnerInstanceId: source.runnerInstanceId, useRunnerd: true, runnerIngressAuthorized: true, runnerExecutionTarget: target, + managedAiCredentialHome: home, managedAiCredentialIdentity: "selected-revision", managedAiAccountIdentity: "selected-account", + runnerEnvironment: provider === "claude" ? { ANTHROPIC_API_KEY: "synthetic-fixture" } : {} }); + void execute.catch(() => undefined); + try { + await vi.waitFor(() => expect(release, inspect(updates.map(update => update.values.failureDetail))).toHaveBeenCalledTimes(unknown ? 2 : 1)); + expect(readNativeSessionIdleWork()).toBe("present"); + const next = { ...current, binding: { ...current.binding, runId: `${current.binding.runId}-next` } }; + const nextTask = vi.fn(); + await expect(executePaperclipNativeSession({ db: leaseDb(next), execution: next, + runnerInstanceId: source.runnerInstanceId, useRunnerd: true, runnerIngressAuthorized: true, + runnerExecutionTarget: { ...target, runnerTask: nextTask } })).rejects.toThrow("native_session_supervisor_busy"); + expect(nextTask).not.toHaveBeenCalled(); + if (provider === "codex" && !unknown && !shutdown && !journalFailure && !deadline) { + await expect(executePaperclipNativeSession({ db: leaseDb(next), execution: next, + runnerInstanceId: source.runnerInstanceId, useRunnerd: true, runnerIngressAuthorized: true, + runnerExecutionTarget: { ...target, environmentId: "competing-environment", runnerTask: nextTask, + sandboxLeaseAcquisition: { ...target.sandboxLeaseAcquisition, providerLeaseId: "competing-task" } } })).rejects.toThrow("native_session_supervisor_busy"); + expect(nextTask).not.toHaveBeenCalled(); + expect(release).toHaveBeenCalledTimes(1); + expect(close).not.toHaveBeenCalled(); + expect(readNativeSessionIdleWork()).toBe("present"); + } + if (shutdown || deadline) { + if (shutdown) expect(await closeIdleWarmNativeSessionsForRestart()).toMatchObject({ busy: 1 }); + await vi.waitFor(() => expect(close).toHaveBeenCalledOnce()); + expect(readNativeSessionIdleWork()).toBe("present"); + expect(detachControllerForRestart).not.toHaveBeenCalled(); + // A late host receipt cannot publish a released owner during retirement. + allowRelease(); + await receipt; + await Promise.resolve(); + expect(readNativeSessionIdleWork()).toBe("present"); + await expect(executePaperclipNativeSession({ db: leaseDb(next), execution: next, + runnerInstanceId: source.runnerInstanceId, useRunnerd: true, runnerIngressAuthorized: true, + runnerExecutionTarget: { ...target, runnerTask: nextTask } })).rejects.toThrow("native_session_supervisor_busy"); + expect(nextTask).not.toHaveBeenCalled(); + allowPhysicalStop(); + await expect(execute).resolves.toMatchObject({ sessionId: source.normalizedSessionId }); + expect(readNativeSessionIdleWork()).toBe("none"); + const directory = join(resolvePaperclipInstanceRoot(), "runtime/paperclip-runner/sessions"); + expect(JSON.parse(await readFile(join(directory, (await readdir(directory))[0]!), "utf8")).releasedRunner).toBeUndefined(); + return; + } + if (journalFailure) { + const directory = join(resolvePaperclipInstanceRoot(), "runtime/paperclip-runner/sessions"); + const path = join(directory, (await readdir(directory))[0]!); + await rename(path, `${path}.saved`); + await mkdir(path); + let finished = false; + void execute.finally(() => { finished = true; }).catch(() => undefined); + allowRelease(); + await new Promise(resolve => setImmediate(resolve)); + expect(finished).toBe(false); + await expect(executePaperclipNativeSession({ db: leaseDb(next), execution: next, + runnerInstanceId: source.runnerInstanceId, useRunnerd: true, runnerIngressAuthorized: true, + managedAiCredentialHome: home, managedAiCredentialIdentity: "selected-revision", managedAiAccountIdentity: "selected-account", + runnerExecutionTarget: { ...target, remoteCwd: "different-workspace", runnerTask: nextTask } })).rejects.toThrow("native_session_supervisor_busy"); + expect(nextTask).not.toHaveBeenCalled(); + await vi.waitFor(() => expect(close).toHaveBeenCalledOnce()); + expect(readNativeSessionIdleWork()).toBe("present"); + expect(finished).toBe(false); + await rm(path, { recursive: true }); + await rename(`${path}.saved`, path); + expect(JSON.parse(await readFile(path, "utf8")).releasedRunner).toBeUndefined(); + allowPhysicalStop(); + await expect(execute).resolves.toMatchObject({ sessionId: source.normalizedSessionId }); + expect(readNativeSessionIdleWork()).toBe("none"); + expect(detachControllerForRestart).not.toHaveBeenCalled(); + return; + } + allowRelease(); + await expect(execute).resolves.toMatchObject({ sessionId: source.normalizedSessionId }); + expect(readNativeSessionIdleWork()).toBe("none"); + const directory = join(resolvePaperclipInstanceRoot(), "runtime/paperclip-runner/sessions"); + const files = await readdir(directory); + const path = join(directory, files[0]!); + const envelope = JSON.parse(await readFile(path, "utf8")); + expect(envelope.releasedRunner).toMatchObject({ source, taskId: "retained-task", scopeDigest: expect.any(String) }); + expect(JSON.stringify(envelope)).not.toContain("synthetic-fixture"); + if (!unknown) { + const calls = owners.mock.calls.length; + for (const changed of [{ remoteCwd: "different-workspace" }, { runnerTaskProjectIds: ["different-project"] }]) { + await expect(executePaperclipNativeSession({ db: leaseDb(next), execution: next, + runnerInstanceId: source.runnerInstanceId, useRunnerd: true, runnerIngressAuthorized: true, + managedAiCredentialHome: home, managedAiCredentialIdentity: "selected-revision", managedAiAccountIdentity: "selected-account", + runnerExecutionTarget: { ...target, ...changed, runnerTask: nextTask }, + runnerEnvironment: provider === "claude" ? { ANTHROPIC_API_KEY: "synthetic-fixture" } : {} })).rejects.toThrow("released_owner_scope_mismatch"); + } + expect(owners).toHaveBeenCalledTimes(calls); + expect(nextTask).not.toHaveBeenCalled(); + expect(state.execute).toHaveBeenCalledTimes(1); + expect(JSON.parse(await readFile(path, "utf8")).releasedRunner).toEqual(envelope.releasedRunner); + } + await vi.waitFor(() => expect(detachControllerForRestart).toHaveBeenCalledOnce()); + expect(close).not.toHaveBeenCalled(); + expect(task.mock.calls.map(([operation]) => (operation as { kind: string }).kind)).toEqual(["describe"]); + expect(JSON.parse(await readFile(path, "utf8")).releasedRunner).toEqual(envelope.releasedRunner); + if (provider === "codex" && !unknown) { + const inspected = vi.spyOn(remoteWarmRecoveryRuntime, "inspectRemoteWarmSessionRecovery").mockImplementation(async request => { + request.loadCheckpoint(current); + throw new Error("fixture completed-source proof checked"); + }); + try { + const calls = owners.mock.calls.length; + const base = { db: leaseDb(next), execution: next, runnerInstanceId: source.runnerInstanceId, + useRunnerd: true, runnerIngressAuthorized: true, runnerExecutionTarget: target, + managedAiCredentialHome: home, managedAiCredentialIdentity: "selected-revision", managedAiAccountIdentity: "selected-account" }; + await expect(executePaperclipNativeSession(base)).rejects.toThrow("fixture completed-source proof checked"); + for (const changed of [ + { managedAiAccountIdentity: "different-account" }, { managedAiCredentialIdentity: "different-revision" }, + { runnerExecutionTarget: { ...target, runnerTaskProjectIds: ["different-project"] } }, + { runnerExecutionTarget: { ...target, remoteCwd: "different-workspace" } }, + { runnerExecutionTarget: { ...target, sandboxLeaseAcquisition: { ...target.sandboxLeaseAcquisition, providerLeaseId: "different-task" } } }, + ]) await expect(executePaperclipNativeSession({ ...base, ...changed })).rejects.toThrow("released_owner_scope_mismatch"); + expect(owners).toHaveBeenCalledTimes(calls); + expect(state.execute).toHaveBeenCalledTimes(1); + } finally { inspected.mockRestore(); } + } + await owners.mock.calls.at(-1)![0].onResumed!(source); + expect(JSON.parse(await readFile(path, "utf8")).releasedRunner).toBeUndefined(); + const replacement = { ...envelope, snapshot: { ...envelope.snapshot, identity: { ...envelope.snapshot.identity, runId: "new-owner" } }, + releasedRunner: { ...envelope.releasedRunner, source: { ...source, runId: "new-owner" } } }; + await writeFile(path, JSON.stringify(replacement)); + await expect(owners.mock.calls.at(-1)![0].onResumed!(source)).rejects.toThrow("released_source_mismatch"); + expect(JSON.parse(await readFile(path, "utf8")).releasedRunner).toEqual(replacement.releasedRunner); + } finally { + allowRelease(); allowPhysicalStop(); await execute.catch(() => undefined); await closeIdleWarmNativeSessionsForRestart(); owners.mockRestore(); + if (previousHome === undefined) delete process.env.PAPERCLIP_HOME; else process.env.PAPERCLIP_HOME = previousHome; + } + }); + + it.each((["codex", "claude"] as const).flatMap(provider => (["current", "stale", ...(provider === "codex" ? ["foreign runner", "foreign lease"] : [])] as const) + .map(receipt => [provider, receipt] as const)))( + "records the confirmed %s source after external checkpoint registration and attachment (receipt=%s)", async (provider, receipt) => { + const invalidReceipt = receipt !== "current"; + await closeIdleWarmNativeSessionsForRestart(); + const previousHome = process.env.PAPERCLIP_HOME; + process.env.PAPERCLIP_HOME = isolatedStateDirectory; + const { permissionPolicy: _permissionPolicy, ...claudeProfile } = QUALIFIED_ACPX_PROFILES.claude; + let first: NativeExecutionInput = scopedNativeExecution(`released-external-${provider}-${receipt.replaceAll(" ", "-")}`); + first.session.lifecyclePolicy = { mode: "warm", idleTimeoutMs: 5_000 }; + if (provider === "claude") first = { ...first, schema: "paperclip.native-execution-input.v5", executionMode: "default", planningContext: null, + runtimeContext: nativeRuntimeContextFixture(), session: { ...first.session, driverKind: "acpx_runtime" }, + provider: { kind: "acpx", agent: "claude", model: "claude-sonnet-4-6", permissionMode: "approve-all", profile: claudeProfile } }; + parseNativeExecutionInput(first); + const next = { ...first, binding: { ...first.binding, runId: `${first.binding.runId}-next` } }; + const source = { runnerInstanceId: "external-runner", environmentLeaseId: "external-lease", runId: first.binding.runId, + normalizedSessionId: first.session.normalizedSessionId!, turnId: `turn-${first.binding.runId}`, itemId: `item-${first.binding.runId}` }; + const successor = { ...source, runId: next.binding.runId, turnId: "attached-turn", itemId: "attached-item" }; + const root = await initializeScopedRunnerRoot(first, source.runnerInstanceId); + await writeRunnerAuthority(root, source, "ready"); + const home = join(isolatedStateDirectory, "external-selected-account"); + await mkdir(home, { recursive: true }); + await writeFile(join(home, "auth.json"), JSON.stringify({ OPENAI_API_KEY: "synthetic-fixture" }), { mode: 0o600 }); + const task = vi.fn(async () => ({ kind: "description", taskId: "external-task", launchPreparationVersion: 1, + projectAuthCheckVersion: 1, warmTaskVersion: 1, releaseResumeVersion: 1, + runner: { version: "fixture", digest: `sha256:${"a".repeat(64)}` } })); + const target = { kind: "remote" as const, transport: "sandbox" as const, remoteCwd: ".", environmentId: "external-environment", + leaseId: source.environmentLeaseId, runnerTask: task, sandboxLeaseAcquisition: { outcome: "resumed" as const, providerLeaseId: "external-task" } }; + const createOwner = pluginTaskRuntime.createPluginRunnerTaskOwner; + const owners = vi.spyOn(pluginTaskRuntime, "createPluginRunnerTaskOwner").mockImplementation(input => createOwner({ ...input, adoptIdentity: source })); + const saved = new Map(); + const restore = vi.spyOn(runnerCheckpoint, "restoreNativeRunnerCheckpoint").mockImplementation(async (_db, binding, directory) => { + const checkpoint = saved.get(binding.runId); + if (!checkpoint) throw new Error("Missing fixture controller checkpoint"); + checkpoint.restoreSessionDirectory(directory); + return checkpoint; + }); + let authority: DurablePrpControlPlane | undefined; + let register: import("../../vendor/paperclip-runner/index.js").RunnerExecutionDriver["controlPlaneRegistration"]; + let releasePhysical!: () => void; + const physical = new Promise(resolve => { releasePhysical = resolve; }); + const close = vi.fn(async () => { if (invalidReceipt) await physical; }); + const release = vi.fn(async () => { + await authority!.flushCheckpoint(); + const identity = authority!.store.state.identity; + if (identity.runId !== successor.runId || receipt === "current") return structuredClone(identity); + if (receipt === "stale") return structuredClone(source); + return { ...identity, ...(receipt === "foreign runner" ? { runnerInstanceId: "successor-candidate-runner" } + : { environmentLeaseId: "successor-controller-lease" }) }; + }); + const session = { close, release, detachControllerForRestart: vi.fn(async () => {}) }; + const validated = validatePrpStructuredRunResult({ schema: "paperclip.run_result.v1", reportedWorkDisposition: "done", summary: "Completed", + completionClaim: { contractRevision: first.completionContract.contract.revision, objectiveSatisfied: true, criteria: [], remainingWork: [] }, + evidence: [], verification: [], artifacts: [], attentionRequests: [] }); + if (!validated.ok) throw new Error("Invalid completion fixture"); + state.execute.mockReset().mockImplementation(async options => { + const current = options.input.binding.runId === first.binding.runId ? first : next; + if (!authority) { + state.createBackend.mock.calls.at(-1)![1].codexTransportFactory!(); + const transport = state.createTransport.mock.calls.at(-1)![0] as RunnerTransportOptions & { + controlPlaneRegistration: NonNullable; + }; + await rm(join(root, "control-plane"), { recursive: true }); + authority = new DurablePrpControlPlane({ stateDirectory: join(root, "control-plane"), identity: source, + expectedRunnerVersion: "fixture", expectedRunnerDigest: `sha256:${"a".repeat(64)}` }); + const persist = authority.setCheckpointPersistence.bind(authority); + // Only the backing store is modeled; registration and the fail-closed marker are real. + vi.spyOn(authority, "setCheckpointPersistence").mockImplementation(() => persist(async checkpoint => { saved.set(checkpoint.runId, checkpoint); })); + register = transport.controlPlaneRegistration; + await register(authority, source); + expect(JSON.parse(await readFile(join(root, "control-plane/control-plane-state.json"), "utf8")).schema) + .toBe("paperclip.runner.external-control-plane-state.v1"); + } else { + expect(options.existingSession).toBe(session); + await owners.mock.calls[0]![0].onResumed!(source); + authority.rotateRunIdentity(successor); + await register!(authority, successor); + await writeFile(join(root, "runner/runner-state.json"), JSON.stringify(durableRunnerState(successor, "ready"))); + expect(JSON.parse(await readFile(join(root, "control-plane/control-plane-state.json"), "utf8")).identity.runId).toBe(source.runId); + } + await options.onCheckpoint?.({ backendKind: "runner", driverKind: current.session.driverKind, sessionId: "driver-session", + identity: { ...current.binding, sessionId: source.normalizedSessionId }, providerSessionId: "provider-session", activeTurnId: null, pendingRuntimeRequests: [] }); + await options.onSession?.(session); + return { result: validated.result, terminal: { runTerminalState: "succeeded" }, turnId: "provider-turn", + normalizedSessionId: source.normalizedSessionId, providerSessionId: "provider-session", driverKind: "test", driverVersion: "1", + nativeEventCount: 1, highestContiguousSourceSeq: 1, usage: null }; + }); + const run = (execution: NativeExecutionInput) => { + const db = leaseDb(execution, { leaseOwner: "external-controller", controllerGeneration: 1 }); + if (execution.binding.runId === next.binding.runId) { + const select = db.select.bind(db); + db.select = ((fields?: Record) => fields?.status === heartbeatRuns.status && fields?.runnerProfileJson === heartbeatRuns.runnerProfileJson + ? priorNativeRunDb(first).select(fields as never) : select(fields as never)) as Db["select"]; + } + return executePaperclipNativeSession({ db, leaseOwner: "external-controller", execution, + runnerInstanceId: execution === next ? "successor-candidate-runner" : source.runnerInstanceId, useRunnerd: true, runnerIngressAuthorized: true, + runnerExecutionTarget: execution === next ? { ...target, leaseId: "successor-controller-lease" } : target, + managedAiCredentialHome: home, managedAiCredentialIdentity: "selected-revision", + managedAiAccountIdentity: "selected-account", runnerEnvironment: provider === "claude" ? { ANTHROPIC_API_KEY: "synthetic-fixture" } : {} }); + }; + let following: ReturnType | undefined; + try { + await expect(run(first)).resolves.toMatchObject({ sessionId: source.normalizedSessionId }); + expect(release).toHaveBeenCalledOnce(); + const directory = join(resolvePaperclipInstanceRoot(), "runtime/paperclip-runner/sessions"); + const path = join(directory, (await readdir(directory))[0]!); + expect(JSON.parse(await readFile(path, "utf8")).releasedRunner.source).toEqual(source); + following = run(next); + void following.catch(() => undefined); + if (invalidReceipt) { + await vi.waitFor(() => expect(close).toHaveBeenCalledOnce()); + expect(readNativeSessionIdleWork()).toBe("present"); + expect(JSON.parse(await readFile(path, "utf8")).releasedRunner).toBeUndefined(); + releasePhysical(); + await expect(following).resolves.toMatchObject({ sessionId: source.normalizedSessionId }); + expect(readNativeSessionIdleWork()).toBe("none"); + } else { + await expect(following).resolves.toMatchObject({ sessionId: source.normalizedSessionId }); + expect(release).toHaveBeenCalledTimes(2); + expect(JSON.parse(await readFile(path, "utf8")).releasedRunner.source).toEqual(successor); + expect(close).not.toHaveBeenCalled(); + } + } finally { + releasePhysical(); await following?.catch(() => undefined); await closeIdleWarmNativeSessionsForRestart(); + await authority?.stop(); owners.mockRestore(); restore.mockRestore(); + if (previousHome === undefined) delete process.env.PAPERCLIP_HOME; else process.env.PAPERCLIP_HOME = previousHome; + } + }); + it.each(["codex", "opencode"] as const)("keeps assigned MCP credentials on the control plane for %s", async (providerKind) => { const selectedExecution: NativeExecutionInput = providerKind === "opencode" ? { ...execution, provider: { kind: "opencode", model: "openrouter/deepseek/deepseek-v4.1-flash-1017", permissionMode: "allow" } } diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 7f69d7ef5cf..a50289f7f8c 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -45,7 +45,7 @@ import { notifyExecutionRunWork } from "../execution-control-notifications.js"; import { createRunUsageRecorder } from "../usage-receipts.js"; import { registerPluginTaskCredentialHandoff, type PluginTaskCredentialWritebackBinding } from "../plugin-task-credential-writeback.js"; import { resolvePluginTaskResume } from "../environment-task-runtime.js"; -import { environmentTaskResultSchema } from "@paperclipai/plugin-sdk"; +import { environmentTaskOperationSchema, environmentTaskResultSchema } from "@paperclipai/plugin-sdk"; import { preparePluginRunnerCredentials } from "./plugin-runner-credentials.js"; import { createPluginRunnerTaskOwner, verifyPluginRunnerTaskReattachment } from "./plugin-runner-task.js"; import { NativeExecutionContext } from "./native-execution-context.js"; @@ -669,6 +669,8 @@ type WarmNativeSession = { companyId: string; environmentId: string | null; busy: boolean; + released?: boolean; + finishRelease?: () => Promise; closeOnReleaseReason?: string; idleTimer: ReturnType | null; lastActivityAt: string; @@ -716,7 +718,9 @@ const pluginTaskSessionOwners = new WeakMap>(); export function readNativeSessionIdleWork(): IdleLocalWork { - return warmNativeSessions.size !== 0 || closingWarmNativeSessions.size !== 0 ? "present" : "none"; + return closingWarmNativeSessions.size !== 0 || [...warmNativeSessions.values()].some(entry => + entry.busy || entry.preparingRunId || entry.instructionPreparationRunId || entry.closeOnReleaseReason || !entry.released) + ? "present" : "none"; } registerIdleLocalWorkProbe(readNativeSessionIdleWork); @@ -876,7 +880,8 @@ async function closeIdleWarmNativeSessions(input: { if (input.environmentId !== undefined && entry.environmentId !== input.environmentId) { continue; } - if (entry.busy || entry.preparingRunId || executingRunnerdSessionScopes.has(sessionId)) { + if (entry.busy || entry.preparingRunId || executingRunnerdSessionScopes.has(sessionId) + || (input.canRetire && (entry.instructionPreparationRunId || closingWarmNativeSessions.has(sessionId)))) { // A busy turn can complete while another idle session is checkpointing. // Fence that entry now so its eventual release cannot leave a new idle // owner behind after the shutdown sweep has already passed it. @@ -884,6 +889,8 @@ async function closeIdleWarmNativeSessions(input: { busy += 1; continue; } + // The host owns retention after release; a quiet local handle needs no retirement for sleep. + if (input.canRetire && entry.released && !entry.closeOnReleaseReason) continue; if (entry.idleTimer !== null) clearTimeout(entry.idleTimer); // Keep a fenced owner until both retirement and stopped collection succeed. // Failure must not make a live directory eligible for a replacement owner. @@ -1927,6 +1934,10 @@ class SessionAuthorityEpoch { if (!isNativeUsageAccountingEvent(event)) return; this.#assertCurrent(); if (event.runId === this.runId) throw new Error("native_recovery_usage_source_unproven"); + if (event.runId === this.#drainingRun?.runId) { + await this.acceptAccountingEvent(event); + return; + } await capture(); this.#assertCurrent(); } @@ -7324,27 +7335,92 @@ function legacyNativeSessionCheckpointPath( ); } +type ReleasedWarmRunner = { source: DurableRecoveryIdentity; taskId: string; scopeDigest: string }; + +function releasedWarmScope(input: Parameters[0]): string { + const target = input.runnerExecutionTarget; + return nativeSha256({ companyId: input.execution.binding.companyId, agentId: input.execution.binding.agentId, + environmentId: target?.kind === "remote" ? target.environmentId ?? null : null, workspace: nativeSessionWorkspaceScope(input.execution), + cwd: input.execution.workspace.cwd, remoteCwd: target?.kind === "remote" ? target.remoteCwd : null, + projectIds: target?.kind === "remote" && target.transport === "sandbox" ? target.runnerTaskProjectIds ?? [] : [], + accountIdentity: input.managedAiAccountIdentity ?? null, credentialIdentity: input.managedAiCredentialIdentity ?? null }); +} + +function requireReleasedWarmScope(input: Parameters[0], + proof: ReleasedWarmRunner | undefined, identity: unknown): DurableRecoveryIdentity { + const target = input.runnerExecutionTarget; + if (!proof || target?.kind !== "remote" || target.transport !== "sandbox" || !target.runnerTask + || proof.taskId !== target.sandboxLeaseAcquisition?.providerLeaseId + || proof.scopeDigest !== releasedWarmScope(input) || canonicalJson(proof.source) !== canonicalJson(identity)) { + throw new Error("native_session_released_owner_scope_mismatch"); + } + return proof.source; +} + +function readWarmNativeEnvelope(execution: NativeExecutionInput): Record | null { + const path = nativeSessionCheckpointPath(execution); + if (!existsSync(path)) return null; + const envelope = record(JSON.parse(readBoundedNativeFile(path, NATIVE_WARM_CHECKPOINT_MAX_BYTES, + "native_session_supervisor_checkpoint_too_large").toString("utf8"))); + if (envelope.schema !== "paperclip.native-session-supervisor.v1") throw new Error("native_session_supervisor_checkpoint_mismatch"); + return envelope; +} + +function releasedWarmRunner(envelope: Record | null): ReleasedWarmRunner | undefined { + if (!envelope || envelope.releasedRunner === undefined) return undefined; + const proof = record(envelope.releasedRunner), source = record(proof.source); + const snapshot = record(envelope.snapshot), identity = record(snapshot.identity); + for (const key of ["runnerInstanceId", "environmentLeaseId", "runId", "normalizedSessionId", "turnId", "itemId"]) { + if (typeof source[key] !== "string" || !source[key]) throw new Error("native_session_released_source_mismatch"); + } + if (typeof proof.taskId !== "string" || !proof.taskId || typeof proof.scopeDigest !== "string" + || identity.runId !== source.runId || identity.sessionId !== source.normalizedSessionId) { + throw new Error("native_session_released_source_mismatch"); + } + return proof as ReleasedWarmRunner; +} + +function writeWarmNativeEnvelope(execution: NativeExecutionInput, envelope: Record): void { + const path = nativeSessionCheckpointPath(execution); + const temporary = `${path}.${process.pid}.${randomUUID()}.tmp`; + writeFileSync(temporary, JSON.stringify(envelope), { encoding: "utf8", mode: 0o600 }); + renameSync(temporary, path); + chmodSync(path, 0o600); +} + +function updateReleasedWarmRunner(execution: NativeExecutionInput, source: DurableRecoveryIdentity, + taskId: string, proof: ReleasedWarmRunner | null): void { + const envelope = readWarmNativeEnvelope(execution); + const previous = releasedWarmRunner(envelope); + if (!envelope) { + if (proof) throw new Error("native_session_supervisor_checkpoint_missing"); + return; + } + if (!proof && !previous) return; + const identity = record(record(envelope.snapshot).identity); + if (identity.companyId !== execution.binding.companyId || identity.agentId !== execution.binding.agentId + || identity.runId !== source.runId || identity.sessionId !== source.normalizedSessionId + || (previous && (previous.taskId !== taskId || canonicalJson(previous.source) !== canonicalJson(source)))) { + throw new Error("native_session_released_source_mismatch"); + } + if (proof) envelope.releasedRunner = proof; + else delete envelope.releasedRunner; + writeWarmNativeEnvelope(execution, envelope); +} + function persistWarmNativeCheckpoint( execution: NativeExecutionInput, configDigest: string, snapshot: PersistedNativeSession, processConfigDigest?: string, ): void { - const path = nativeSessionCheckpointPath(execution); - const temporary = `${path}.${process.pid}.${randomUUID()}.tmp`; - writeFileSync( - temporary, - JSON.stringify({ - schema: "paperclip.native-session-supervisor.v1", - configDigest, - processConfigDigest, - updatedAt: new Date().toISOString(), - snapshot, - }), - { encoding: "utf8", mode: 0o600 }, - ); - renameSync(temporary, path); - chmodSync(path, 0o600); + const previous = releasedWarmRunner(readWarmNativeEnvelope(execution)); + writeWarmNativeEnvelope(execution, { + schema: "paperclip.native-session-supervisor.v1", configDigest, processConfigDigest, + updatedAt: new Date().toISOString(), snapshot, + ...(previous && previous.source.runId === snapshot.identity.runId + && previous.source.normalizedSessionId === snapshot.identity.sessionId ? { releasedRunner: previous } : {}), + }); } function needsCoreGitHubAccess(input: Pick[0], "managedGitHub" | "runnerExecutionTarget">): boolean { @@ -7497,10 +7573,17 @@ function armWarmNativeSessionIdleTimer( return; } current.busy = true; - current.closeOnReleaseReason = "warm native session idle timeout"; - const closing = closeWarmNativeSession(current, current.closeOnReleaseReason) + const released = current.released === true; + if (!released) current.closeOnReleaseReason = "warm native session idle timeout"; + const closing = (released ? (async () => { + if (!current.session.detachControllerForRestart) throw new Error("native_session_released_detach_unavailable"); + await current.session.detachControllerForRestart(); + await current.githubAccess?.stop(); + })() : closeWarmNativeSession(current, current.closeOnReleaseReason!)) .then(() => { forgetWarmNativeSession(sessionId, current); }) - .catch(() => {}) + .catch(() => { + if (released) current.closeOnReleaseReason = "released Runner local detach is unconfirmed"; + }) .finally(() => { current.busy = false; if (closingWarmNativeSessions.get(sessionId) === closing) closingWarmNativeSessions.delete(sessionId); @@ -9098,29 +9181,8 @@ export async function executePaperclipNativeSession(input: { startupUnsettled = !(await startup.executionContext.drain()); startup.executionContext.finish(); } - if (ownsPhysicalIngress && physicalIngressKey && executingNativeIngressAllocations.get(physicalIngressKey) === ownerToken) { - executingNativeIngressAllocations.delete(physicalIngressKey); - } - if (executingNativeOwnerScopes.get(ownerScope) === ownerToken) { - executingNativeOwnerScopes.delete(ownerScope); - } - startup.resolve(null); - startup.settle(); - if (nativeSessionStartups.get(runId) === startup) { - nativeSessionStartups.delete(runId); - setNativeExecutionActive(runId, false); - } - if ( - ownsSessionScope && - sessionScopeId !== null && - executingRunnerdSessionScopes.get(sessionScopeId) === - input.execution.binding.runId - ) { - executingRunnerdSessionScopes.delete(sessionScopeId); - } - // Preparation can fail before the execution loop owns its normal cleanup. - // Remove the settled startup before yielding, so detachment cannot reacquire - // its old resolved session while renewal cleanup drains. + // Join preparation cleanup before release; keep cancellation reachable + // until the exact retained owner has finished the handover. await startup.executionLeaseRenewal?.stop().catch(() => undefined); if (startup.unstartedClaim && !nativeRunsDetachingForRestart.has(runId)) { await releaseUnstartedNativeExecutionClaim({ db: input.db, runId, companyId: input.execution.binding.companyId, @@ -9128,9 +9190,11 @@ export async function executePaperclipNativeSession(input: { logger.warn({ err: error, runId }, "Native preparation claim release failed; the control deadline will settle it"); }); } + let stagedCleanupFailed = false; try { await cleanupStagedAttachments(); } catch (cleanupError) { + stagedCleanupFailed = true; // Cleanup is a confidentiality incident, but it occurs after the native // provider may already have completed the turn. Reclassifying that turn // as failed could replay provider side effects. Emit a private runtime @@ -9152,6 +9216,38 @@ export async function executePaperclipNativeSession(input: { companyId: input.execution.binding.companyId, agentId: input.execution.binding.agentId, }, { companyId: input.execution.binding.companyId, runId: input.execution.binding.runId, issueId: input.execution.binding.issueId }); + const pendingRelease = ownsSessionScope && sessionScopeId + && executingRunnerdSessionScopes.get(sessionScopeId) === runId + ? warmNativeSessions.get(sessionScopeId) : undefined; + try { + if (pendingRelease?.runId === runId && pendingRelease.finishRelease) { + if (executionFailure !== undefined || startupUnsettled || stagedCleanupFailed) { + pendingRelease.closeOnReleaseReason = "warm native session final cleanup failed"; + } + await pendingRelease.finishRelease(); + } + } finally { + if (ownsPhysicalIngress && physicalIngressKey && executingNativeIngressAllocations.get(physicalIngressKey) === ownerToken) { + executingNativeIngressAllocations.delete(physicalIngressKey); + } + if (executingNativeOwnerScopes.get(ownerScope) === ownerToken) { + executingNativeOwnerScopes.delete(ownerScope); + } + if ( + ownsSessionScope && + sessionScopeId !== null && + executingRunnerdSessionScopes.get(sessionScopeId) === + input.execution.binding.runId + ) { + executingRunnerdSessionScopes.delete(sessionScopeId); + } + startup.resolve(null); + startup.settle(); + if (nativeSessionStartups.get(runId) === startup) { + nativeSessionStartups.delete(runId); + setNativeExecutionActive(runId, false); + } + } if (startupUnsettled) throw new NativeStartupCleanupPendingError(startup.executionContext!.id, startup.executionContext!.deadlineAtMs, startup.leaseOwner!, startup.executionContext!.pluginSupported, input.runnerExecutionTarget?.kind === "remote" ? input.runnerExecutionTarget.leaseId ?? undefined : undefined); @@ -9270,6 +9366,9 @@ async function executePaperclipNativeSessionWithinScope( } let retainedTransition: VerifiedWarmTransitionBinding | undefined; let remoteWarmRecovery: RemoteWarmSessionRecovery | undefined; + const cachedReleasedOwner = warmNativeSessions.get(nativeSessionScopeKey(input.execution)); + if (cachedReleasedOwner?.released) requireReleasedWarmScope(input, releasedWarmRunner(readWarmNativeEnvelope(input.execution)), + readRunnerdDurableIdentity(scopedRunnerdStateRoot(input.execution))); if (input.useRunnerd && !input.restartRecovery && !input.backend && input.runnerExecutionTarget?.kind === "remote" && input.execution.session.lifecyclePolicy.mode === "warm" && @@ -9295,6 +9394,7 @@ async function executePaperclipNativeSessionWithinScope( } if (durableIdentityMatchesSession(identity, input.execution) && identity.runId !== input.execution.binding.runId && runnerdAuthorityLifecycleWithVerifiedBackup({ root, identity, execution: input.execution, allowVerifiedBackup: true }) !== "suspended") { + let releasedSource: DurableRecoveryIdentity | undefined; remoteWarmRecovery = await inspectRemoteWarmSessionRecovery({ db: input.db, execution: input.execution, target: input.runnerExecutionTarget, identity, readControlState: () => record(JSON.parse(readBoundedNativeFile( @@ -9302,7 +9402,13 @@ async function executePaperclipNativeSessionWithinScope( "runner_durable_identity_too_large").toString("utf8"))), matchesExecution: previous => nativeSessionScopeKey(previous) === nativeSessionScopeKey(input.execution) && nativeSessionConfigDigest(previous, "remote") === nativeSessionConfigDigest(input.execution, "remote"), - loadCheckpoint: previous => loadWarmNativeCheckpoint(previous, nativeSessionConfigDigest(previous, "remote"), "remote"), + loadCheckpoint: previous => { + const proof = releasedWarmRunner(readWarmNativeEnvelope(previous)); + if (proof) { + releasedSource = requireReleasedWarmScope(input, proof, identity); + } + return loadWarmNativeCheckpoint(previous, nativeSessionConfigDigest(previous, "remote"), "remote"); + }, canReuseProcess: previous => { const expected = warmNativeProcessConfigDigest(input); if (!expected) return false; @@ -9311,6 +9417,7 @@ async function executePaperclipNativeSessionWithinScope( return envelope.processConfigDigest === expected; }, }); + remoteWarmRecovery.releasedSource = releasedSource; } } if (input.useRunnerd) { @@ -9517,7 +9624,7 @@ async function executePaperclipNativeSessionWithinScope( const target = input.runnerExecutionTarget; const retained = warmNativeSessions.get(nativeSessionScopeKey(input.execution)); if (!input.restartRecovery && target?.kind === "remote" && target.transport === "sandbox" && target.runnerTask && target.leaseId && - retained?.pluginTaskId && idleWarmNativeSessionOwnerRunId(input) !== undefined && + ((retained?.pluginTaskId && idleWarmNativeSessionOwnerRunId(input) !== undefined) || remoteWarmRecovery?.releasedSource) && (input.execution.provider.kind === "codex" || (input.execution.provider.kind === "acpx" && input.execution.provider.agent === "claude"))) { claimedStartupContext = new NativeExecutionContext(input.execution, Math.min(executionLeaseTtlMs, coldAdmissionTimeoutMs(input.execution.provider.kind, input.execution.provider.kind === "acpx" ? input.execution.provider.agent : undefined, false))); @@ -10216,7 +10323,9 @@ async function executePaperclipNativeSessionWithinScope( }; if (warmSessionId !== null && warmConfigDigest !== null && !input.restartRecovery) { const entry = warmNativeSessions.get(warmSessionId); - if (entry) { + if (entry?.released) requireReleasedWarmScope(input, releasedWarmRunner(readWarmNativeEnvelope(input.execution)), + readRunnerdDurableIdentity(scopedRunnerdStateRoot(input.execution))); + if (entry || remoteWarmRecovery?.releasedSource) { if (input.execution.provider.kind === "codex" || (input.execution.provider.kind === "acpx" && input.execution.provider.agent === "claude")) { executionContext = claimedStartupContext ?? new NativeExecutionContext(input.execution, @@ -10225,10 +10334,10 @@ async function executePaperclipNativeSessionWithinScope( startupForAdmission!.executionContext = executionContext; if (startupForAdmission!.stopRequested) executionContext.abort(new Error("native_session_cancelled")); } - if (entry.preparingRunId && entry.preparingRunId !== input.execution.binding.runId) throw new Error("native_session_supervisor_busy"); - entry.preparingRunId = undefined; + if (entry?.preparingRunId && entry.preparingRunId !== input.execution.binding.runId) throw new Error("native_session_supervisor_busy"); + if (entry) entry.preparingRunId = undefined; const taskTarget=input.runnerExecutionTarget; - if (entry.pluginTaskId && taskTarget?.kind === "remote" && taskTarget.transport === "sandbox" && taskTarget.runnerTask) { + if ((entry?.pluginTaskId || remoteWarmRecovery?.releasedSource) && taskTarget?.kind === "remote" && taskTarget.transport === "sandbox" && taskTarget.runnerTask) { const description=environmentTaskResultSchema.parse(await (executionContext ? executionContext.wait(() => taskTarget.runnerTask!({kind:"describe"}, { executionContext })) : taskTarget.runnerTask({kind:"describe"}))); @@ -10237,6 +10346,8 @@ async function executePaperclipNativeSessionWithinScope( if (executionContext && description.executionContextVersion === 1) executionContext.pluginSupported = true; else if (executionContext) await finishStartupContextForFallback(); } + } + if (entry) { // Old run-scoped environments still require process replacement. A // session-owned broker can change run authority without replacing it. const hasBrokerCapability = Boolean( @@ -10301,6 +10412,7 @@ async function executePaperclipNativeSessionWithinScope( entry.instructionWorkingCopy = input.instructionWorkingCopy?.checkpointWarm ? undefined : input.instructionWorkingCopy; entry.busy = true; entry.ownerToken = warmSessionOwnerToken; + entry.released = false; entry.environmentId = input.runnerExecutionTarget?.environmentId ?? null; if (entry.idleTimer !== null) clearTimeout(entry.idleTimer); @@ -10329,7 +10441,7 @@ async function executePaperclipNativeSessionWithinScope( ); } } - if ((claimedStartupContext && executionContext !== claimedStartupContext) || (executionContext && !existingWarmSession)) { + if ((claimedStartupContext && executionContext !== claimedStartupContext) || (executionContext && !existingWarmSession && !remoteWarmRecovery?.releasedSource)) { await finishStartupContextForFallback(); } async function resolvePendingGovernedWait() { @@ -10497,9 +10609,10 @@ async function executePaperclipNativeSessionWithinScope( }) : null; if (remoteWarmRecovery) { - await remoteWarmRecovery.revalidate(leaseOwner); + if (executionContext) await executionContext.wait(() => remoteWarmRecovery.revalidate(leaseOwner)); + else await remoteWarmRecovery.revalidate(leaseOwner); const recovered = await runnerdBackend?.recoverSession?.(remoteWarmRecovery.checkpoint, { - signal: new AbortController().signal, + signal: executionContext?.signal ?? new AbortController().signal, }); if (!recovered?.recovered || !recovered.session) throw new Error("runner_state_identity_mismatch: remote_warm_session_unproven"); if (remoteWarmRecovery.reuseProcess) { @@ -10509,6 +10622,7 @@ async function executePaperclipNativeSessionWithinScope( persistedWarmSession = null; } else { await recovered.session.close({ reason: "recovered warm session requires current process configuration" }); + if (executionContext) await finishStartupContextForFallback(); // Closing an externally persisted controller leaves an identity marker, // not its attachment template. The next backend must rotate the saved // PRP authority from the committed suspension checkpoint. @@ -10710,8 +10824,10 @@ async function executePaperclipNativeSessionWithinScope( rmSync(resolve(scopedRunnerdStateRoot(input.execution), "plugin-runner-state.json"), { force: true }); } if (session && session === existingWarmSession) remoteWarmRecoveryPending = false; - if (session && input.managedAiCredentialHome) { - managedCredentialSession = runnerdBackend?.bindManagedSession(session) ?? session; + if (session && (input.managedAiCredentialHome || (input.runnerExecutionTarget?.kind === "remote" + && input.runnerExecutionTarget.transport === "sandbox" && input.runnerExecutionTarget.runnerTask))) { + const bound = runnerdBackend?.bindManagedSession(session, input.managedAiCredentialHome ? undefined : { releaseResumeOnly: true }) ?? session; + if (input.managedAiCredentialHome) managedCredentialSession = bound; } releaseRegisteredGoalController(); liveQuestions.close(); @@ -11662,11 +11778,101 @@ async function executePaperclipNativeSessionWithinScope( if (collectInstructions && ownedSession) { ownedSession.closeOnReleaseReason = "registered instruction edits require stopped-provider collection"; } - await timeController("warm_release", () => releaseWarmNativeSession( - warmSessionId, - warmSessionOwnerToken, - lifecyclePolicy.idleTimeoutMs, - !retainSession, + if (retainSession && ownedSession?.ownerToken === warmSessionOwnerToken && !ownedSession.closeOnReleaseReason + && pluginTaskSessionOwners.get(ownedSession.session)?.driver.releaseResume) { + const scope = warmSessionId, entry = ownedSession, token = warmSessionOwnerToken; + entry.finishRelease = async () => { + const current = () => warmNativeSessions.get(scope) === entry && entry.ownerToken === token; + if (!current()) return; + try { + if (nativeSessionStartups.get(input.execution.binding.runId)?.stopRequested) entry.closeOnReleaseReason = "warm native session cancelled"; + if (entry.closeOnReleaseReason) { + await closeWarmNativeSession(entry, entry.closeOnReleaseReason); + forgetWarmNativeSession(scope, entry); + return; + } + const deadline = Date.now() + Math.min(lifecyclePolicy.idleTimeoutMs, + coldAdmissionTimeoutMs(input.execution.provider.kind, + input.execution.provider.kind === "acpx" ? input.execution.provider.agent : undefined, false)); + let released: DurableRecoveryIdentity | false | undefined; + for (;;) { + let gateTimer: ReturnType | undefined; + try { + const gate = new Promise((_resolve, reject) => { + const check = () => { + if (!current() || Date.now() >= deadline || entry.closeOnReleaseReason + || nativeSessionStartups.get(input.execution.binding.runId)?.stopRequested) { + reject(new Error("native_session_release_reconciliation_stopped")); + } else gateTimer = setTimeout(check, Math.min(25, deadline - Date.now())); + }; + check(); + }); + // Observe late replies, but only this reservation may publish their result. + released = await Promise.race([Promise.resolve().then(() => { + if (!current() || Date.now() >= deadline || entry.closeOnReleaseReason + || nativeSessionStartups.get(input.execution.binding.runId)?.stopRequested) { + throw new Error("native_session_release_reconciliation_stopped"); + } + return entry.session.release?.(); + }), gate]); + break; + } + catch { + if (!current()) return; + if (Date.now() >= deadline || entry.closeOnReleaseReason + || nativeSessionStartups.get(input.execution.binding.runId)?.stopRequested) { + entry.closeOnReleaseReason ??= "warm Runner release reconciliation expired"; + await closeWarmNativeSession(entry, entry.closeOnReleaseReason); + forgetWarmNativeSession(scope, entry); + return; + } + await new Promise(resolve => setTimeout(resolve, 250)); + } + finally { clearTimeout(gateTimer); } + } + if (!current()) return; + if (nativeSessionStartups.get(input.execution.binding.runId)?.stopRequested) entry.closeOnReleaseReason = "warm native session cancelled"; + if (entry.closeOnReleaseReason) { + await closeWarmNativeSession(entry, entry.closeOnReleaseReason); + forgetWarmNativeSession(scope, entry); + return; + } + if (released !== false && released !== undefined) { + try { + const operation = environmentTaskOperationSchema.parse({ kind: "release", identity: released }); + const target = input.runnerExecutionTarget; + if (operation.kind !== "release" || operation.identity.runId !== input.execution.binding.runId + || operation.identity.normalizedSessionId !== nativeSessionKey(input.execution) + || operation.identity.runnerInstanceId !== effectiveRunnerInstanceId + || target?.kind !== "remote" || operation.identity.environmentLeaseId !== (durableRunnerBinding?.environmentLeaseId ?? target.leaseId) + || target.transport !== "sandbox" || !entry.pluginTaskId + || entry.pluginTaskId !== target.sandboxLeaseAcquisition?.providerLeaseId) { + throw new Error("native_session_released_source_mismatch"); + } + const source = operation.identity; + updateReleasedWarmRunner(input.execution, source, entry.pluginTaskId, + { source, taskId: entry.pluginTaskId, scopeDigest: releasedWarmScope(input) }); + } catch { + entry.closeOnReleaseReason = "released Runner source proof could not be verified or saved"; + await closeWarmNativeSession(entry, entry.closeOnReleaseReason); + forgetWarmNativeSession(scope, entry); + return; + } + entry.released = true; + } + } catch { + // The business turn already settled. An unknown release stays owned and active. + logger.warn({ runId: input.execution.binding.runId }, "Retained Runner release remains unconfirmed"); + } finally { + if (current()) { + delete entry.finishRelease; + entry.busy = false; + armWarmNativeSessionIdleTimer(scope, entry, lifecyclePolicy.idleTimeoutMs); + } + } + }; + } else await timeController("warm_release", () => releaseWarmNativeSession( + warmSessionId, warmSessionOwnerToken, lifecyclePolicy.idleTimeoutMs, !retainSession, )); if (collectInstructions && !collectedByOwner) await timeController("instruction_collection", () => instructionCopy!.collectStopped()); } @@ -13741,7 +13947,7 @@ async function prepareRemoteExternalRunner({ input, target, runner, root, identi } type ManagedRunnerdBackend = NativeSessionBackend & { - bindManagedSession(session: NativeSession): NativeSession; + bindManagedSession(session: NativeSession, options?: { releaseResumeOnly: true }): NativeSession; bindDiagnosticSession?(session: NativeSession): void; sessionCheckpointPersisted?(snapshot: PersistedNativeSession): void; providerAdmissionPersisted?(): Promise; @@ -13941,8 +14147,11 @@ async function createRunnerdBackendWithinSessionClaim( } : undefined, onRecoveryAccountingEvent: input.onAccountingEvent ? async (event: PrpEvent, source: Parameters>[1]) => { const current = sessionToolAuthorityEpochs.get(sessionScopeId); - if (!current || !recoveryUsage) throw new Error("native_recovery_usage_source_unproven"); - await current.acceptRecoveryAccountingEvent(event, () => recoveryUsage(event, source)); + if (!current) throw new Error("native_recovery_usage_source_unproven"); + await current.acceptRecoveryAccountingEvent(event, () => { + if (!recoveryUsage) throw new Error("native_recovery_usage_source_unproven"); + return recoveryUsage(event, source); + }); } : undefined, }; const effectiveRunnerInstanceId = @@ -14003,6 +14212,10 @@ async function createRunnerdBackendWithinSessionClaim( onCredentials: input.execution.session.lifecyclePolicy.mode === "warm" ? undefined : credentials.onCredentials, warmTaskVersion: description.warmTaskVersion, releaseResumeVersion: description.releaseResumeVersion, + authorityRunId: description.releaseResumeVersion === 1 ? input.execution.binding.runId : undefined, + releasedSource: input.remoteWarmRecovery?.releasedSource, + onResumed: async source => updateReleasedWarmRunner(input.execution, source, description.taskId, null), + onRetired: async source => updateReleasedWarmRunner(input.execution, source, description.taskId, null), adoptIdentity: adoptionIdentity ?? (adopting && prior ? { runnerInstanceId: prior.runnerInstanceId, environmentLeaseId: prior.environmentLeaseId, runId: prior.runId, normalizedSessionId: prior.normalizedSessionId, @@ -14031,6 +14244,7 @@ async function createRunnerdBackendWithinSessionClaim( }); }, onDispatched: input.onTaskDispatched, }; + if (input.remoteWarmRecovery?.releasedSource && description.releaseResumeVersion !== 1) throw new Error("Plugin Runner release/resume unavailable"); if (input.execution.session.lifecyclePolicy.mode === "warm" && (!credentials.accountIdentity || description.warmTaskVersion !== 1)) throw new Error("Plugin warm task capability or account unavailable"); const owner = retainedOwner ?? createPluginRunnerTaskOwner(ownerInput); if (retainedOwner) retainedOwner.rebind(ownerInput); @@ -14107,7 +14321,10 @@ async function createRunnerdBackendWithinSessionClaim( sessionToolAuthorityEpochs.set(sessionScopeId, authorityEpoch); return { ...backend, descriptor: () => backend.descriptor(), openSession: session => backend.openSession(session), recoverSession: (snapshot, options) => backend.recoverSession!(snapshot, options), - bindManagedSession: session => { pluginTaskSessionOwners.set(session, owner); return session; } }; + bindManagedSession: (session, options) => { + if (!options?.releaseResumeOnly || description.releaseResumeVersion === 1) pluginTaskSessionOwners.set(session, owner); + return session; + } }; } if (input.execution.schema === "paperclip.native-execution-input.v6" || input.execution.schema === "paperclip.native-execution-input.v8") { if (target.kind === "remote" && input.dotWorkspaceRoot) throw new Error("dot_remote_workspace_access_unavailable"); diff --git a/server/src/services/native-runtime/plugin-runner-release-resume.integration.test.ts b/server/src/services/native-runtime/plugin-runner-release-resume.integration.test.ts new file mode 100644 index 00000000000..d94697eba8f --- /dev/null +++ b/server/src/services/native-runtime/plugin-runner-release-resume.integration.test.ts @@ -0,0 +1,315 @@ +import { execFileSync } from "node:child_process"; +import { createHash, randomUUID } from "node:crypto"; +import { existsSync } from "node:fs"; +import { mkdir, readFile, readdir, writeFile } from "node:fs/promises"; +import https, { createServer } from "node:https"; +import { createServer as createTcpServer } from "node:net"; +import { join, resolve } from "node:path"; +import { getCACertificates, setDefaultCACertificates } from "node:tls"; +import { eq } from "drizzle-orm"; +import WebSocket, { WebSocketServer, type RawData } from "ws"; +import { expect, it, vi } from "vitest"; +import { completionContracts, environmentLeases, environments, executionWorkspaces, heartbeatRuns, issues, nativeRunFinalizations, nativeRunResults } from "@paperclipai/db"; +import { environmentTaskOperationSchema, type PluginEnvironmentTaskOperation } from "@paperclipai/plugin-sdk"; +import { readRunnerdArtifactBinding, defaultCapabilityRunnerdBinary, + parseNativeExecutionInput, parseNativeRuntimeContext, QUALIFIED_ACPX_PROFILES, type DurableRecoveryIdentity, type RunnerProcessHandle, type RunnerProcessLaunchSpec } from "../../vendor/paperclip-runner/index.js"; +import { createRunnerProviderLaunch, runnerLaunchRequirementsDigest, canonicalNativeRuntimeContextDigest } from "@paperclipai/paperclip-runner"; +import { materializeAsset } from "./runtime-context.js"; +import { spawnRunner } from "../../vendor/paperclip-runner/testing.js"; +import { describeEmbeddedPostgres, nativeRestartFixture, realProcessIt, fakeCodexAppServer, waitForCondition } from "../../__tests__/helpers/native-restart.js"; +import { executePaperclipNativeSession, cancelNativeSession, closeIdleWarmNativeSessionsForRestart, readNativeSessionIdleWork } from "./native-session-executor.js"; +import { resolvePaperclipInstanceRoot } from "../../home-paths.js"; +import * as remoteWarmRecovery from "./remote-warm-session-recovery.js"; +import { reconcileNativeFinalizations } from "./native-finalization-reconciler.js"; +import { buildNativeCompletionContract } from "./completion-contracts.js"; +import { nativeRuntimeContextFixture } from "./runtime-context.test-fixture.js"; + +const { createPrivateHostControlFixture } = await import(new URL("../../../../packages/paperclip-runner/test/support/private-host-control.ts", import.meta.url).href) as { + createPrivateHostControlFixture(): { + launch(spec: RunnerProcessLaunchSpec): RunnerProcessHandle; + request(frame: Record): Promise<{ id: string; status: string }>; + close(): Promise; + }; +}; + +const fakeAcpxSidecar = resolve(import.meta.dirname, "../../../../packages/paperclip-runner/runner/target/debug/fake-acpx-sidecar"); +const sha = (value: string) => createHash("sha256").update(value).digest("hex"); +const parkId = (source: DurableRecoveryIdentity) => `park_${sha(JSON.stringify(Object.fromEntries(Object.entries(source).sort(([a], [b]) => a.localeCompare(b)))))}`; + +// The only replaced service is the external environment-task/Launcher edge. +// Runtime, backend, owner, PRP ingress, checkpoint persistence and native Runner are real. +describeEmbeddedPostgres("plugin Runner release/resume through the Core execution entry", () => { + const environment = nativeRestartFixture(); + const nativeIt = process.platform === "win32" ? it.skip : realProcessIt; + nativeIt.for([...(["codex", "claude"] as const).flatMap(harness => (["cached", "adopted"] as const).map(mode => [harness, mode] as const)), ["codex", "adopted cancellation"] as const, ["codex", "adopted legacy cancellation"] as const])("completes %s runs through a retained %s process", { timeout: 60_000 }, async ([harness, mode], context) => { + if (harness === "claude" && !existsSync(fakeAcpxSidecar)) context.skip(); + const adopted = mode.startsWith("adopted"), cancellation = mode.includes("cancellation"), qualifiedCancellation = mode === "adopted cancellation"; + const first = await environment.seedRun(`RELEASE-ENTRY-${harness}-${mode}`); + const { db, runId, issueId, native } = first; + const { companyId, agentId, runtimeRoot } = environment.current; + const root = join(runtimeRoot, runId), workspace = join(root, "workspace"), home = join(root, "credential-home"); + await mkdir(workspace, { recursive: true }); await mkdir(home, { recursive: true }); + await writeFile(join(home, "auth.json"), JSON.stringify({ OPENAI_API_KEY: "synthetic-fixture" }), { mode: 0o600 }); + const workspaceId = randomUUID(), environmentId = randomUUID(), leaseA = randomUUID(), taskId = randomUUID(); + await db.insert(executionWorkspaces).values({ id: workspaceId, companyId, sourceIssueId: issueId, + mode: "isolated", strategyType: "directory", name: "Release fixture", cwd: workspace }); + await db.insert(environments).values({ id: environmentId, name: environmentId, driver: "plugin" }); + await db.insert(environmentLeases).values({ id: leaseA, companyId, environmentId, heartbeatRunId: runId, issueId, + executionWorkspaceId: workspaceId, provider: "plugin:fixture:tasks", providerLeaseId: taskId, status: "active", leasePolicy: "reuse_by_environment", metadata: { remoteCwd: workspace } }); + const [issue] = await db.select().from(issues).where(eq(issues.id, issueId)); + const [contract] = await db.select().from(completionContracts).where(eq(completionContracts.issueId, issueId)); + const runtimeContext = nativeRuntimeContextFixture(); + runtimeContext.instructions.bundle = await materializeAsset([{ path: "AGENTS.md", mode: 0o444, content: Buffer.from("Complete only the controlled task.\n") }]); + runtimeContext.aggregateDigest = canonicalNativeRuntimeContextDigest(runtimeContext); + const { permissionPolicy: _permissionPolicy, ...claudeProfile } = QUALIFIED_ACPX_PROFILES.claude; + const execution = parseNativeExecutionInput({ schema: "paperclip.native-execution-input.v5", executionMode: "default", planningContext: null, + runtimeContext, binding: { companyId, agentId, issueId, runId, executionWorkspaceId: workspaceId }, + provider: harness === "codex" ? { kind: "codex", model: "fixture-model", approvalPolicy: "never" } + : { kind: "acpx", agent: "claude", model: "claude-sonnet-4-6", permissionMode: "approve-all", profile: claudeProfile }, + task: { identifier: "RELEASE-ENTRY", title: issue!.title, description: null, prompt: "Complete the controlled task.", workMode: "standard" }, + workspace: { cwd: workspace, repoUrl: null, repoRef: null, branchName: null }, + session: { normalizedSessionId: native.normalizedSessionId, driverKind: harness === "codex" ? "codex_app_server" : "acpx_runtime", protocolVersion: 1, + lifecyclePolicy: { mode: "warm", idleTimeoutMs: adopted ? 5_000 : 60_000 } }, + completionContract: { id: contract!.id, sha256: contract!.canonicalSha256, schemaVersion: "paperclip.completion-contract.v1", + contract: buildNativeCompletionContract(issue!, { revision: contract!.revision }) }, interactionResponses: [], credentialBindings: [] }); + const resultFile = join(root, "result.json"), calls = join(root, "provider-calls"); + await writeFile(resultFile, JSON.stringify({ schema: "paperclip.run_result.v1", reportedWorkDisposition: "done", summary: "Controlled task complete.", + completionClaim: { contractRevision: execution.completionContract.contract.revision, objectiveSatisfied: true, + criteria: execution.completionContract.contract.criteria.map(c => ({ criterionId: c.id, status: "satisfied", evidenceRefs: [] })), remainingWork: [] }, + evidence: [], verification: [], artifacts: [], attentionRequests: [] })); + const sidecar = join(root, "sidecar"), sidecarScript = join(root, "sidecar-entry"); + if (harness === "claude") { + const quote = (value: string) => `'${value.replaceAll("'", "'\"'\"'")}'`; + await writeFile(sidecarScript, "qualified fixture entry"); + await writeFile(sidecar, `#!/bin/sh\nexec ${quote(fakeAcpxSidecar)} --mode source-reconcile --profile-digest ${quote(claudeProfile.commandDigest)} --journal ${quote(calls)} --source-fault ${quote(join(root, "fault"))} --wait-completion-feedback\n`, { mode: 0o700 }); + } + const key = join(root, "tls.key"), certificate = join(root, "tls.crt"); + execFileSync("openssl", ["req", "-x509", "-newkey", "rsa:2048", "-nodes", "-keyout", key, "-out", certificate, + "-days", "1", "-subj", "/CN=localhost", "-addext", "subjectAltName=IP:127.0.0.1,DNS:localhost"], { stdio: "ignore" }); + const oldCAs = getCACertificates("default"), oldAgent = https.globalAgent; + const certificates = [...oldCAs, await readFile(certificate, "utf8")]; + const probe = createTcpServer(); await new Promise(done => probe.listen(0, "127.0.0.1", done)); + const address = probe.address(); if (!address || typeof address === "string") throw new Error("Fixture listener unavailable"); + const runnerPort = address.port; await new Promise(done => probe.close(() => done())); + const tls = createServer({ key: await readFile(key), cert: await readFile(certificate) }); + const sockets = new Set(), websocket = new WebSocketServer({ noServer: true }); + let listenPath = ""; + const wireEvents: string[] = []; + tls.on("upgrade", (request, socket, head) => { + if (request.headers["x-fixture-token"] !== "synthetic-ingress") { socket.destroy(); return; } + websocket.handleUpgrade(request, socket, head, front => { + const back = new WebSocket(`ws://127.0.0.1:${runnerPort}${listenPath}`, { perMessageDeflate: false }); + const pending: Array<{ data: RawData; binary: boolean }> = []; sockets.add(front); sockets.add(back); + front.on("message", (data, binary) => { + if (!binary) { const frame = JSON.parse(data.toString()); wireEvents.push(`core:${frame.kind ?? frame.type}:${frame.payload?.type ?? frame.payload?.eventType ?? frame.error?.code ?? ""}`); } + if (back.readyState === WebSocket.OPEN) back.send(data, { binary }); else pending.push({ data, binary }); + }); + back.on("open", () => { for (const message of pending.splice(0)) back.send(message.data, { binary: message.binary }); }); + back.on("message", (data, binary) => { + if (!binary) { const frame = JSON.parse(data.toString()); wireEvents.push(`runner:${frame.kind ?? frame.type}:${frame.payload?.type ?? frame.payload?.eventType ?? frame.error?.code ?? ""}`); } + if (front.readyState === WebSocket.OPEN) front.send(data, { binary }); + }); + front.on("close", () => { sockets.delete(front); back.terminate(); }); back.on("close", () => { sockets.delete(back); front.terminate(); }); + front.on("error", () => back.terminate()); back.on("error", () => front.terminate()); + }); + }); + await new Promise(done => tls.listen(0, "127.0.0.1", done)); + const tlsAddress = tls.address(); if (!tlsAddress || typeof tlsAddress === "string") throw new Error("Fixture TLS unavailable"); + const host = createPrivateHostControlFixture(), runnerRoot = join(root, "native"), binary = defaultCapabilityRunnerdBinary(); + const artifact = readRunnerdArtifactBinding(binary), preparations = new Map; launch: { configurationId: string } }>(); + const releases: DurableRecoveryIdentity[] = [], claims: DurableRecoveryIdentity[] = []; + let handle: RunnerProcessHandle | undefined, submissions = 0, stops = 0, completes = 0; + let releaseClaimReply!: () => void, observeHeldClaim!: () => void; + const claimReply = new Promise(done => { releaseClaimReply = done; }); + const heldClaim = new Promise(done => { observeHeldClaim = done; }); + let pendingFollowup: Promise | undefined; + const providerStarts = async () => (await readFile(calls, "utf8")).trim().split(/\r?\n/) + .filter(line => harness === "codex" ? line === "process-start" : JSON.parse(line).processStart).length; + const state = async () => JSON.parse(await readFile(join(runnerRoot, "runner-state.json"), "utf8")); + const sameSource = (a: DurableRecoveryIdentity, b: DurableRecoveryIdentity) => Object.entries(a).every(([key, value]) => b[key as keyof DurableRecoveryIdentity] === value); + const task = async (raw: unknown, authorityRunId: string): Promise => { + const operation = environmentTaskOperationSchema.parse(raw); + if (operation.kind === "describe") return { kind: "description", taskId, launchPreparationVersion: 1, + projectAuthCheckVersion: 1, warmTaskVersion: 1, releaseResumeVersion: 1, runner: artifact, ...(qualifiedCancellation ? { executionContextVersion: 1 } : {}) }; + if (operation.kind === "prepare") { + expect(operation.requirements.identity.runId).toBe(authorityRunId); + expect(operation.accountIdentity).toBe("fixture-account"); + if (operation.priorIdentity) expect(sameSource(operation.priorIdentity, await state())).toBe(true); + expect(operation.requirements.settings.managedProfile).toBeUndefined(); + expect(operation.requirements.settings.agentCoreProfile).toBeUndefined(); + expect(operation.requirements.settings.acpxAgent).toBe(harness === "claude" ? "claude" : undefined); + const requirements = { ...operation.requirements, runtimeContext: operation.requirements.runtimeContext ? parseNativeRuntimeContext(operation.requirements.runtimeContext) : null, settings: { ...operation.requirements.settings, acpxAgent: harness === "claude" ? "claude" as const : undefined, managedProfile: undefined, agentCoreProfile: undefined } }; + const launch = { configurationId: `configuration-${operation.requirements.identity.runId}`, requirementsDigest: runnerLaunchRequirementsDigest(requirements), + provider: createRunnerProviderLaunch(requirements, { options: { codexArgs: ["--state-file", join(root, "provider.json"), "--terminal-result-file", resultFile, + "--call-log", calls, "--record-process-start"] }, root, cwd: workspace, codexHome: home, codexCommand: fakeCodexAppServer, + providerNodeCommand: harness === "claude" ? sidecar : process.execPath, opencodeProxyPath: "", acpxSidecarPath: harness === "claude" ? sidecarScript : "" }) }; + preparations.set(launch.configurationId, { operation, launch }); return { kind: "prepared", taskId, launch }; + } + if (operation.kind === "check_project_auth") return { kind: "project_auth", taskId, configurationId: operation.configurationId, provider: operation.provider, allowed: true }; + if (operation.kind === "submit") { + if (!operation.preparedLaunchId || !operation.launch) throw new Error("Fixture structured launch unavailable"); + const prepared = preparations.get(operation.preparedLaunchId)!; submissions++; + expect(operation.runner).toMatchObject({ runnerId: prepared.operation.requirements.identity.runnerInstanceId, + leaseId: prepared.operation.requirements.identity.environmentLeaseId, runId: authorityRunId, + sessionId: prepared.operation.requirements.identity.normalizedSessionId, turnId: prepared.operation.requirements.identity.turnId, + itemId: prepared.operation.requirements.identity.itemId }); + listenPath = `/api/runner/v1/connect/${operation.runner.runId}`; + handle = spawnRunner({ ...operation.launch, stateDirectory: runnerRoot, identity: prepared.operation.requirements.identity, ticket: operation.bootstrapTicket, + runnerBinaryPath: binary, maxOutboxBytes: operation.launch.maxOutboxBytes, p0ReserveBytes: operation.launch.p0ReserveBytes, + connection: { mode: "listen", listenAddress: "0.0.0.0", listenPort: runnerPort, listenPath }, processLauncher: host.launch, + environment: { PATH: process.env.PATH }, + ...(harness === "claude" ? { acpxLaunchProfile: { authorityDigest: `sha256:${sha("fixture authority")}`, command: sidecar, + commandSha256: `sha256:${sha(await readFile(sidecar, "utf8"))}`, sidecarScript, + sidecarScriptSha256: `sha256:${sha(await readFile(sidecarScript, "utf8"))}` } } : {}) }); + return { kind: "accepted", taskId }; + } + if (operation.kind === "connection") return { kind: "connection", taskId, endpoint: { kind: "authenticated_websocket", + websocketUrl: `wss://127.0.0.1:${tlsAddress.port}/runner`, secretHeaders: [{ name: "X-Fixture-Token", value: "synthetic-ingress" }], generation: "fixture" } }; + if (operation.kind === "status") return { kind: "status", taskId, phase: handle?.isExitConfirmed?.() ? "cancelled" : "running", executionStopped: handle?.isExitConfirmed?.() ?? false }; + if (operation.kind === "read_state") { + const live = await state(); + const previous = [...preparations.values()].some(({ operation: prepared }) => prepared.priorIdentity && sameSource(prepared.priorIdentity, operation.identity)); + expect(sameSource(operation.identity, live) || previous).toBe(true); + return { kind: "state", taskId, state: live }; + } + if (operation.kind === "release") { + const live = await state(); const reply = await host.request({ op: "park", id: parkId(operation.identity), identity: operation.identity, + lastControllerCommandSeq: live.lastControllerCommandSeq, ackedSourceSeq: live.ackedSourceSeq }); + if (reply.status !== "parked") throw new Error("Fixture native release refused"); + releases.push(operation.identity); return { kind: "released", taskId }; + } + if (operation.kind === "resume") { + const retainedProviderCount = await providerStarts(); + const reply = await host.request({ op: "claim", id: `claim_${sha(operation.bootstrapTicket)}`, parkId: parkId(operation.identity), bootstrapTicket: operation.bootstrapTicket }); + if (reply.status !== "claimed") throw new Error("Fixture native claim refused"); + expect(await providerStarts()).toBe(retainedProviderCount); + claims.push(operation.identity); + if (cancellation) { observeHeldClaim(); await claimReply; } + return { kind: "resume", taskId, outcome: "resumed" }; + } + if (operation.kind === "stop") { stops++; await host.close(); } + if (operation.kind === "complete") { expect(handle?.isExitConfirmed?.()).toBe(true); completes++; } + return { kind: "accepted", taskId }; + }; + const supervisor = async () => { + const directory = join(resolvePaperclipInstanceRoot(), "runtime/paperclip-runner/sessions"); + for (const file of await readdir(directory)) { + const envelope = JSON.parse(await readFile(join(directory, file), "utf8")); + if (envelope.snapshot?.identity.sessionId === native.normalizedSessionId) return envelope; + } + throw new Error("Fixture supervisor record unavailable"); + }; + const proof = async () => (await supervisor()).releasedRunner; + const execute = async (current: typeof execution, leaseId: string) => { + await db.update(heartbeatRuns).set({ runnerProfileJson: { nativeExecutionInput: current }, runnerInstanceId: native.runnerInstanceId, completionContractId: contract!.id, completionContractSha256: contract!.canonicalSha256, nativeSessionId: current.session.normalizedSessionId }) + .where(eq(heartbeatRuns.id, current.binding.runId)); + return executePaperclipNativeSession({ db, execution: current, runnerInstanceId: native.runnerInstanceId, useRunnerd: true, runnerIngressAuthorized: true, + runnerEnvironment: harness === "claude" ? { ANTHROPIC_API_KEY: "synthetic-fixture" } : {}, + managedAiCredentialHome: home, managedAiCredentialIdentity: "fixture-revision", managedAiAccountIdentity: "fixture-account", turnTimeoutMs: 15_000, + runnerExecutionTarget: { kind: "remote", transport: "sandbox", remoteCwd: workspace, environmentId, leaseId, providerKey: "plugin:fixture:tasks", runnerTask: operation => task(operation, current.binding.runId), + sandboxLeaseAcquisition: { outcome: leaseId === leaseA ? "created" : "resumed", providerLeaseId: taskId } } }); + }; + const adoption = vi.spyOn(remoteWarmRecovery, "inspectRemoteWarmSessionRecovery"); + try { + setDefaultCACertificates(certificates); https.globalAgent = new https.Agent({ ca: certificates }); + const a = await execute(execution, leaseA); expect(a).toMatchObject({ exitCode: 0, timedOut: false }); + await waitForCondition("automatic release A", () => releases.length === 1); + expect((await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, runId)))[0]).toMatchObject({ schemaStatus: "accepted", resultJson: { result: { reportedWorkDisposition: "done" } } }); + expect(await proof()).toMatchObject({ source: releases[0], taskId }); + const providerSessionId = (await supervisor()).snapshot.providerSessionId; + expect(providerSessionId).toEqual(expect.any(String)); + const pid = handle!.child.pid; + const { heartbeatService } = await import("../heartbeat.js"); + expect(await heartbeatService(db).prepareIdleWarmNativeSessionsForSleep({ canRetire: () => true })) + .toEqual({ closed: 0, busy: 0, failed: 0 }); + expect(stops).toBe(0); expect(handle!.isExitConfirmed?.()).toBe(false); + if (adopted) { + await new Promise(done => setTimeout(done, 5_200)); + expect(stops).toBe(0); expect(handle!.isExitConfirmed?.()).toBe(false); + expect(await proof()).toMatchObject({ source: releases[0], taskId }); + } + expect(readNativeSessionIdleWork()).toBe("none"); + await reconcileNativeFinalizations(db, [runId]); + expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, runId)))[0]).toMatchObject({ phase: "committed", leaseOwner: null }); + const admitNext = async () => { + const runB = randomUUID(), leaseB = randomUUID(); + await db.insert(heartbeatRuns).values({ id: runB, companyId, agentId, nativeIssueId: issueId, status: "running", runtimeMode: "native" }); + await db.insert(nativeRunFinalizations).values({ runId: runB, companyId, issueId, phase: "observed" }); + await db.update(issues).set({ executionRunId: runB, status: "in_progress" }).where(eq(issues.id, issueId)); + await db.insert(environmentLeases).values({ id: leaseB, companyId, environmentId, heartbeatRunId: runB, issueId, executionWorkspaceId: workspaceId, + provider: "plugin:fixture:tasks", providerLeaseId: taskId, status: "active", leasePolicy: "reuse_by_environment", metadata: { remoteCwd: workspace } }); + const next = { ...execution, binding: { ...execution.binding, runId: runB } }; + pendingFollowup = execute(next, leaseB); + if (cancellation) { + void pendingFollowup.catch(() => undefined); + await heldClaim; + const competitor = { ...next, binding: { ...next.binding, runId: randomUUID() } }; + await expect(execute(competitor, leaseB)).rejects.toThrow("native_session_supervisor_busy"); + const stopping = cancelNativeSession(runB, "Stop retained adoption", { db }); + void stopping.catch(() => undefined); + await waitForCondition("durable cancellation intent", async () => Boolean((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runB)))[0]?.resultJson?.nativeCancellation)); + if (qualifiedCancellation) await waitForCondition("Core-triggered native stop", () => stops > 0 && handle!.isExitConfirmed!()); + else await expect(execute(competitor, leaseB)).rejects.toThrow("native_session_supervisor_busy"); + expect(claims).toEqual([releases[0]]); + expect(releases).toHaveLength(1); + releaseClaimReply(); + expect(await pendingFollowup.catch(error => error)).toBeInstanceOf(Error); + await stopping.catch(() => undefined); + expect(completes).toBeGreaterThan(0); + expect(handle!.isExitConfirmed!()).toBe(true); + expect((await readFile(calls, "utf8")).split(/\r?\n/).filter(method => method === "turn/start")).toHaveLength(1); + expect((await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, runB)))).toHaveLength(0); + expect(releases).toHaveLength(1); expect((await proof())?.source.runId).not.toBe(runB); + return runB; + } + expect(await pendingFollowup).toMatchObject({ exitCode: 0, timedOut: false }); + expect((await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, runB)))[0]).toMatchObject({ schemaStatus: "accepted", resultJson: { result: { reportedWorkDisposition: "done" } } }); + await reconcileNativeFinalizations(db, [runB]); + return runB; + }; + const runB = await admitNext(); + if (cancellation) return; + await waitForCondition("automatic release B", () => releases.length === 2); + expect(await proof()).toMatchObject({ source: releases[1], taskId }); + expect((await supervisor()).snapshot.providerSessionId).toBe(providerSessionId); + expect(adoption).toHaveBeenCalledTimes(adopted ? 1 : 0); + if (adopted) expect(adoption.mock.calls[0]![0]).toMatchObject({ identity: releases[0], execution: { binding: { runId: runB } } }); + if (adopted) { + const runC = await admitNext(); + await waitForCondition("automatic release C", () => releases.length === 3); + expect(releases[2]!.runId).toBe(runC); expect(await proof()).toMatchObject({ source: releases[2], taskId }); + expect((await supervisor()).snapshot.providerSessionId).toBe(providerSessionId); + expect(adoption).toHaveBeenCalledTimes(1); + } + expect(claims).toEqual(releases.slice(0, -1)); expect(releases[1]!.runId).toBe(runB); + expect(releases[1]!.environmentLeaseId).toBe(leaseA); expect(handle!.child.pid).toBe(pid); expect(submissions).toBe(1); + const methods = (await readFile(calls, "utf8")).trim().split(/\r?\n/); + if (harness === "codex") { + expect(methods.filter(method => method === "process-start")).toHaveLength(1); + expect(methods.filter(method => method === "turn/start")).toHaveLength(adopted ? 3 : 2); + } else { + const journal = methods.map(line => JSON.parse(line)); + expect(journal.filter(entry => entry.processStart)).toHaveLength(adopted ? 3 : 2); + expect(journal.filter(entry => entry.command === "turn.start")).toHaveLength(adopted ? 3 : 2); + } + } catch (error) { + const live = await state().catch(() => ({})); + console.info("entry-fixture", { releases: releases.length, claims: claims.length, lifecycle: live.lifecycle, + lastCommand: live.lastControllerCommandSeq, ackedSource: live.ackedSourceSeq, outbox: live.outbox?.map((e: { envelope?: { payload?: { eventType?: string } } }) => e.envelope?.payload?.eventType), wire: wireEvents.slice(-70) }); + throw error; + } finally { + releaseClaimReply(); + await pendingFollowup?.catch(() => undefined); + try { + const cleanup = await Promise.allSettled([closeIdleWarmNativeSessionsForRestart(), host.close()]); + for (const socket of sockets) socket.terminate(); websocket.close(); + await new Promise(done => tls.close(() => done())); + const failures = cleanup.filter((result): result is PromiseRejectedResult => result.status === "rejected"); + if (failures.length) throw new AggregateError(failures.map(result => result.reason), "Fixture cleanup failed"); + } finally { + adoption.mockRestore(); + https.globalAgent.destroy(); https.globalAgent = oldAgent; setDefaultCACertificates(oldCAs); + } + } + }); +}); diff --git a/server/src/services/native-runtime/remote-warm-session-recovery.ts b/server/src/services/native-runtime/remote-warm-session-recovery.ts index bc0d7ed31d4..14bc3a791d7 100644 --- a/server/src/services/native-runtime/remote-warm-session-recovery.ts +++ b/server/src/services/native-runtime/remote-warm-session-recovery.ts @@ -1,7 +1,7 @@ import { and, eq } from "drizzle-orm"; import { agents, companies, environmentLeases, heartbeatRuns, issues, nativeRunFinalizations, type Db } from "@paperclipai/db"; import type { AdapterExecutionTarget } from "@paperclipai/adapter-utils/execution-target"; -import { parseNativeExecutionInput, type NativeExecutionInput, type PersistedNativeSession } from "../../vendor/paperclip-runner/index.js"; +import { parseNativeExecutionInput, type NativeExecutionInput, type PersistedNativeSession, type DurableRecoveryIdentity } from "../../vendor/paperclip-runner/index.js"; import { currentNativeControllerIdentity, evaluateNativeControllerTakeover } from "./native-restart-recovery.js"; import { NativeControllerRecoveryWaitingError, readNativeControllerRecovery } from "./native-controller-recovery.js"; @@ -10,6 +10,7 @@ export type RemoteWarmSessionRecovery = { checkpoint: PersistedNativeSession; previousExecution: NativeExecutionInput; reuseProcess: boolean; + releasedSource?: DurableRecoveryIdentity; remote: { providerLeaseId: string; remoteCwd: string }; revalidate: (leaseOwner: string) => Promise; };