diff --git a/doc/plugins/PLUGIN_RUNNER_EXECUTION.md b/doc/plugins/PLUGIN_RUNNER_EXECUTION.md index 117ccca2166..9e6f746fc77 100644 --- a/doc/plugins/PLUGIN_RUNNER_EXECUTION.md +++ b/doc/plugins/PLUGIN_RUNNER_EXECUTION.md @@ -222,6 +222,16 @@ Core permits read-only suspension proof for the exact saved identity of a successful warm run after its reuse lease is retained or superseded. This does not authorize new work or file access through that lease. Temporary credential homes are not needed after the originating run ends. +An explicit reset to a different normalized session can retire a warm owner +whose close failed only because it could not prove checkpoint suspension. Core +requires the exact lease-bound task's `executionStopped: true` receipt, a +successful `complete` handoff, and settled Core tool results first. It then +collects stopped instruction files before it releases the in-memory owner. The old checkpoint remains evidence; +it is not reused by the new session. A failed handoff, failed file collection, +unsettled Core tool result, other close error, or continuation of the same +session still blocks replacement. +A controller restart is not required for this stopped-owner reset path. + Termination receipts cover all run leases that owned the retired physical task. A stopped task gets a new provider task ID for checkpoint resumption. Inspection precedes in-memory session selection so a replacement cannot inherit the stopped diff --git a/packages/paperclip-runner/src/contracts/native-session-backend.ts b/packages/paperclip-runner/src/contracts/native-session-backend.ts index 28866778dbe..8457c47a396 100644 --- a/packages/paperclip-runner/src/contracts/native-session-backend.ts +++ b/packages/paperclip-runner/src/contracts/native-session-backend.ts @@ -143,7 +143,11 @@ export function isSemanticToolOutcomeUnknownError(error: unknown): error is Sema export class NativeSessionCloseUnrecoverableError extends Error { readonly code = "native_session_close_unrecoverable"; - constructor(readonly settlement?: Record) { + constructor( + readonly settlement?: Record, + /** Exact durable results exist for every admitted Core tool callback. */ + readonly semanticToolsSettled = false, + ) { super( "provider_transport_failed: runner did not durably suspend before checkpoint", ); diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts index 4b351a656f3..68fc16f53f7 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts @@ -1356,6 +1356,7 @@ it.each(["after_budget", "within_budget", "interrupted_within_budget", "persiste expect(closeFailure).toBeInstanceOf( NativeSessionCloseUnrecoverableError, ); + expect(closeFailure).toHaveProperty("semanticToolsSettled", false); expect(closeFailure).toHaveProperty("settlement.semanticTools.pending", expect.arrayContaining([ expect.objectContaining({ callId: expect.any(String), operationId: expect.any(String), sourceEventId: expect.any(String), inputDigest: expect.stringMatching(/^[a-f0-9]{64}$/) }), @@ -8110,9 +8111,13 @@ it("still fails closed when a real close grace period cannot fit a durable suspe cwd: tmpdir(), dynamicTools: [], }); - await expect(bundle.transport.close()).rejects.toThrow( - "runner did not durably suspend before checkpoint", - ); + const closed = bundle.transport.close(); + await expect(closed).rejects.toMatchObject({ + code: "native_session_close_unrecoverable", + semanticToolsSettled: true, + }); + // Retirement preserves the original rejected promise and its proof. + expect(bundle.transport.close()).toBe(closed); } finally { await rm(stateDirectory, { recursive: true, force: true }); } diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts index 23ab1f1721e..bba62831ee4 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts @@ -4928,7 +4928,11 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { } catch { // Keep the settlement failure authoritative if its observer fails. } - throw new NativeSessionCloseUnrecoverableError(settlement); + // Sample after ingress retirement and its drain. A result observed before + // that barrier cannot rule out one last admitted tool callback. + throw new NativeSessionCloseUnrecoverableError( + settlement, this.#core?.semanticToolResultsSettled() === true, + ); } if (!preparationConfirmed) { throw new Error(`Native runner ${attachmentPending ? "attachment" : "startup"} cleanup is unconfirmed; recovery state retained.`); 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 30d94805b98..4dabeda6954 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -75,6 +75,7 @@ import { agentInstructionWorkingCopies } from "@paperclipai/db"; import * as noLaunchProofModule from "./native-maintenance-no-launch.js"; import { NativeSessionCleanupQuarantinedError, + NativeSessionCloseUnrecoverableError, NativeProviderTerminalFailure, NativeSessionProtocolIntegrityError, } from "../../vendor/paperclip-runner/index.js"; @@ -9201,6 +9202,102 @@ describe("native warm session supervision", () => { await closeIdleWarmNativeSessionsForRestart(); }); + it.each(["retired", "unconfirmed", "other-error", "collection-failed", "same-session", "same-id-provider", "unbound", "pending-semantic"] as const)( + "handles a permanently rejected plugin close during reset (%s)", async mode => { + const { base, next, run } = fixture(`identity-poisoned-close-${mode}`); + const identity = { runnerInstanceId: "runner", environmentLeaseId: "lease", runId: base.binding.runId, + normalizedSessionId: base.session.normalizedSessionId!, turnId: "turn", itemId: "item" }; + let stopped = false; + const task = vi.fn(async (operation: any) => { + if (operation.kind === "describe") return { kind: "description", taskId: "old-task", launchPreparationVersion: 1, + projectAuthCheckVersion: 1, runner: { version: "fixture", digest: `sha256:${"a".repeat(64)}` } }; + if (operation.kind === "stop") stopped = true; + if (operation.kind === "status") return { kind: "status", taskId: "old-task", phase: stopped ? "cancelled" : "running", executionStopped: stopped }; + if (operation.kind === "complete" && mode === "unconfirmed") throw new Error("handoff unconfirmed"); + return { kind: "accepted", taskId: "old-task" }; + }); + const createOwner = pluginTaskRuntime.createPluginRunnerTaskOwner; + let owner!: ReturnType; + const ownerFactory = vi.spyOn(pluginTaskRuntime, "createPluginRunnerTaskOwner").mockImplementation(input => { + owner = createOwner({ ...input, adoptIdentity: identity }); + return owner; + }); + onTestFinished(() => ownerFactory.mockRestore()); + // Bind through the production backend so the receipt belongs to this + // exact session object. The task boundary supplies the stop proof. + const backend = await createRunnerdBackend({ db: leaseDb(base), + execution: { ...base, session: { ...base.session, lifecyclePolicy: { mode: "per_turn", idleTimeoutMs: null } } }, + runnerInstanceId: "runner", runnerIngressAuthorized: true, + runnerExecutionTarget: { kind: "remote", transport: "sandbox", remoteCwd: ".", leaseId: "lease", runnerTask: task } }); + const registration = await owner.registration({} as never, base.binding.runId); + const failure = mode === "other-error" ? new Error("controller retirement unconfirmed") : new NativeSessionCloseUnrecoverableError(undefined, mode !== "pending-semantic"); + let closePromise: Promise | undefined; + const close = vi.fn(() => closePromise ??= (async () => { + await registration.release(); + await registration.checkpoint("unsettled").catch(() => undefined); + throw failure; + })()); + const session = { close }; + if (mode !== "unbound") backend.bindManagedSession(session as never); + const collectStopped = vi.fn(async () => { + expect(owner.isRetired()).toBe(true); + if (mode === "collection-failed") throw new Error("file collection failed"); + }); + const retirementFailed = vi.fn(async () => {}); + state.execute.mockReset().mockImplementationOnce(async options => { + await options.onSession?.(session); return result; + }).mockImplementationOnce(async options => { + expect(options.existingSession).toBeUndefined(); + expect(collectStopped).toHaveBeenCalled(); + return result; + }); + try { + await executePaperclipNativeSession({ db: leaseDb(base), execution: base, runnerInstanceId: "runner", + instructionWorkingCopy: { hasChanges: async () => false, collectStopped, retirementFailed } }); + // First poison the retained owner through the real configuration + // retirement path. An explicit reset must not need a process restart. + const changed = { ...base, binding: { ...base.binding, runId: `${base.binding.runId}-changed` } }; + await expect(executePaperclipNativeSession({ db: leaseDb(changed), execution: changed, + runnerInstanceId: "runner", refreshTools: true })).rejects.toThrow(failure.message); + expect(close).toHaveBeenCalledOnce(); + expect(collectStopped).not.toHaveBeenCalled(); + expect(retirementFailed).toHaveBeenCalledOnce(); + retirementFailed.mockClear(); + const replacement = mode === "same-session" ? { ...next, session: base.session } + : mode === "same-id-provider" ? { ...next, session: base.session, + provider: { kind: "opencode" as const, model: "openai/test", permissionMode: "allow" as const } } : next; + const attempt = () => mode === "same-session" + ? executePaperclipNativeSession({ db: leaseDb(replacement), execution: replacement, runnerInstanceId: "runner", refreshTools: true }) + : run(replacement); + if (mode === "retired") { + await attempt(); + expect(state.execute).toHaveBeenCalledTimes(2); + expect(retirementFailed).not.toHaveBeenCalled(); + } else { + // Both attempts see the same rejected promise, just as the real + // transport does; no mock succeeds on the second close. + const expected = mode === "collection-failed" ? "file collection failed" : failure.message; + await expect(attempt()).rejects.toThrow(expected); + await expect(attempt()).rejects.toThrow(expected); + expect(state.execute).toHaveBeenCalledTimes(1); + if (mode !== "collection-failed") expect(collectStopped).not.toHaveBeenCalled(); + else { + collectStopped.mockResolvedValue(undefined); + await attempt(); + expect(state.execute).toHaveBeenCalledTimes(2); + } + } + expect(task.mock.calls.filter(([operation]) => operation.kind === "stop")).toHaveLength(1); + expect(task.mock.calls.some(([operation]) => operation.kind === "read_state")).toBe(false); + } finally { + collectStopped.mockResolvedValue(undefined); + close.mockResolvedValue(undefined); + await closeIdleWarmNativeSessionsForRestart(); + ownerFactory.mockRestore(); + } + }, + ); + it("retains failed retirement for retry and never launches over an uncontained owner", async () => { const { base, next, run } = fixture("identity-close-failure"); const close = vi.fn().mockRejectedValueOnce(new Error("containment unavailable")).mockResolvedValue(undefined); @@ -12675,7 +12772,7 @@ describe("runnerd provider runtime wiring", () => { const driver = { prepareLaunch: vi.fn(), runnerProcessLauncher: launcher, controlPlaneRegistration: registration }; state.toolAuthorityDefinitions.mockImplementation(async binding => binding.readRemoteTaskFile ? [{ name: "register_deliverable" }] as never : []); const readWorkspaceFile = vi.fn(async () => Buffer.from("proof")); - const owner = vi.spyOn(pluginTaskRuntime, "createPluginRunnerTaskOwner").mockReturnValue({ launcher, registration, driver, readWorkspaceFile, rebind:vi.fn() }); + const owner = vi.spyOn(pluginTaskRuntime, "createPluginRunnerTaskOwner").mockReturnValue({ launcher, registration, driver, readWorkspaceFile, isRetired: () => false, rebind:vi.fn() }); const task = vi.fn(async (operation: unknown) => { environmentTaskOperationSchema.parse(operation); return { kind: "description", launchPreparationVersion: 1, projectAuthCheckVersion: 1, ...(kind === "codex" ? { workspaceFileReadVersion: 1 } : {}), taskId: "task", runner: { version: "fixture", digest: `sha256:${"a".repeat(64)}` } }; diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 66b77f875a8..4f5c5ae99de 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -127,6 +127,7 @@ import type { import { NativeProviderTerminalFailure, NativeSessionCleanupQuarantinedError, + NativeSessionCloseUnrecoverableError, NativeSessionProtocolIntegrityError, completeRetainedNativeSessionCleanup, completeTerminatedLocalNativeSessionCleanup, @@ -581,6 +582,7 @@ type WarmNativeSession = { configDigest: string; configuredEnvironmentDigest: string; ownerScope: string; + normalizedSessionId: string; companyId: string; environmentId: string | null; busy: boolean; @@ -589,18 +591,31 @@ type WarmNativeSession = { lastActivityAt: string; }; -async function closeWarmNativeSession(entry: WarmNativeSession, reason: string, preserveInstructionsForRunId?: string) { +async function closeWarmNativeSession(entry: WarmNativeSession, reason: string, options?: { + preserveInstructionsForRunId?: string; + discardUnsettledSession?: boolean; +}) { // Revoke before awaiting process retirement/checkpoint IO. const stopping = entry.githubAccess?.stop(); try { try { await entry.session.close({ reason }); } catch (error) { - try { await entry.instructionWorkingCopy?.retirementFailed?.(); } - catch (receiptError) { throw new AggregateError([error, receiptError], "Warm instruction retirement and receipt both failed"); } - throw error; + // A new conversation does not consume the old checkpoint. Its failed + // suspension must not poison every later reset after Launcher has proved + // that the exact task stopped and completed its credential handoff. + // Other close errors can mean controller authority is still live. + const retiredWithoutCheckpoint = options?.discardUnsettledSession === true + && error instanceof NativeSessionCloseUnrecoverableError + && error.semanticToolsSettled + && pluginTaskSessionOwners.get(entry.session)?.isRetired() === true; + if (!retiredWithoutCheckpoint) { + try { await entry.instructionWorkingCopy?.retirementFailed?.(); } + catch (receiptError) { throw new AggregateError([error, receiptError], "Warm instruction retirement and receipt both failed"); } + throw error; + } } await entry.instructionWorkingCopy?.collectStopped(); - if (entry.instructionCopy?.runId !== preserveInstructionsForRunId) await entry.instructionCopy?.collectStopped(); + if (entry.instructionCopy?.runId !== options?.preserveInstructionsForRunId) await entry.instructionCopy?.collectStopped(); } finally { await stopping; } } @@ -1915,6 +1930,7 @@ function nativeSessionOwnerScope( async function retireSupersededWarmNativeSessions( ownerScope: string, nextSessionScope: string, + nextSessionId: string, ): Promise { for (const [scope, entry] of warmNativeSessions) { if (scope === nextSessionScope || entry.ownerScope !== ownerScope) continue; @@ -1926,7 +1942,9 @@ async function retireSupersededWarmNativeSessions( entry.busy = true; entry.closeOnReleaseReason = "warm native session identity changed"; try { - await closeWarmNativeSession(entry, entry.closeOnReleaseReason); + await closeWarmNativeSession(entry, entry.closeOnReleaseReason, { + discardUnsettledSession: entry.normalizedSessionId !== nextSessionId, + }); forgetWarmNativeSession(scope, entry); } finally { // A failed close retains the owner and prevents launch. A later attempt @@ -7939,7 +7957,7 @@ export async function executePaperclipNativeSession(input: { } executingNativeOwnerScopes.set(ownerScope, ownerToken); await retireSupersededWarmNativeSessions( - ownerScope, nativeSessionScopeKey(input.execution), + ownerScope, nativeSessionScopeKey(input.execution), nativeSessionKey(input.execution), ); if ((input.execution.schema === "paperclip.native-execution-input.v6" || input.execution.schema === "paperclip.native-execution-input.v8") && isCloudManagedInstance() && (!input.useRunnerd || input.runnerExecutionTarget?.kind !== "remote" @@ -9025,7 +9043,7 @@ async function executePaperclipNativeSessionWithinScope( entry.busy = true; entry.closeOnReleaseReason = "warm native session configuration changed"; try { - await closeWarmNativeSession(entry, entry.closeOnReleaseReason, input.execution.binding.runId); + await closeWarmNativeSession(entry, entry.closeOnReleaseReason, { preserveInstructionsForRunId: input.execution.binding.runId }); forgetWarmNativeSession(warmSessionId, entry); } finally { entry.busy = false; } if (preparedCopy) throw new Error("native_instruction_preparation_configuration_changed"); @@ -9481,6 +9499,7 @@ async function executePaperclipNativeSessionWithinScope( ownerScope: nativeSessionOwnerScope( input.execution, input.runnerExecutionTarget?.environmentId ?? null, ), + normalizedSessionId: nativeSessionKey(input.execution), companyId: input.execution.binding.companyId, environmentId: input.runnerExecutionTarget?.environmentId ?? null, diff --git a/server/src/services/native-runtime/plugin-runner-task.test.ts b/server/src/services/native-runtime/plugin-runner-task.test.ts index b419030d8d1..f592f4f421a 100644 --- a/server/src/services/native-runtime/plugin-runner-task.test.ts +++ b/server/src/services/native-runtime/plugin-runner-task.test.ts @@ -137,6 +137,39 @@ describe("plugin native Runner ownership", () => { expect(retained).toHaveBeenCalledWith(expect.objectContaining(requirements.identity)); expect(task.mock.calls.map(([op]) => op.kind)).toEqual(["status", "read_state", "connection", "stop", "status", "read_state", "complete"]); }); + it.each(["settled", "unsettled"] as const)("proves retirement independently of %s checkpoint reuse", async settlement => { + const { task } = provider(); + const retained = vi.fn(async () => {}); + const owner = createPluginRunnerTaskOwner({ task, projectIds: [], harness: "codex", + adoptIdentity: requirements.identity, onSessionRetained: retained }); + const registration = await owner.registration({} as never, "run"); + expect(owner.isRetired()).toBe(false); + await registration.release(); + expect(owner.isRetired()).toBe(false); + await registration.checkpoint(settlement); + expect(owner.isRetired()).toBe(true); + expect(retained).toHaveBeenCalledTimes(settlement === "settled" ? 1 : 0); + expect(task.mock.calls.map(([op]) => op.kind)).toEqual(settlement === "settled" + ? ["stop", "status", "read_state", "complete"] : ["stop", "status", "complete"]); + }); + + it.each(["credentials", "read_state", "complete"])("does not prove retirement when %s obligations fail after execution stops", async failingOperation => { + const { task } = provider(); + const perform = task.getMockImplementation()!; + task.mockImplementation(async operation => { + if (operation.kind === failingOperation) throw new Error("handoff failed"); + if (operation.kind === "credentials") return { kind: "credentials", taskId: "task", credential: null }; + return perform(operation); + }); + const owner = createPluginRunnerTaskOwner({ task, projectIds: [], harness: "codex", accountIdentity: "account", + adoptIdentity: requirements.identity, onCredentials: async () => {}, onSessionRetained: async () => {} }); + const registration = await owner.registration({} as never, "run"); + await registration.release(); + await expect(registration.checkpoint("settled")).rejects.toThrow("handoff failed"); + expect(await owner.driver.adoptExistingRunner!.isAlive()).toBe(false); + expect(owner.isRetired()).toBe(false); + }); + it("submits only a prepared identity and configuration ID", async () => { const { task } = provider(); const owner = createPluginRunnerTaskOwner({ projectAuthCheckVersion: 1, task, projectIds: [], harness: "codex", pollMs: 1 }); diff --git a/server/src/services/native-runtime/plugin-runner-task.ts b/server/src/services/native-runtime/plugin-runner-task.ts index 7b6b2410771..c39b9f7c4fe 100644 --- a/server/src/services/native-runtime/plugin-runner-task.ts +++ b/server/src/services/native-runtime/plugin-runner-task.ts @@ -59,6 +59,7 @@ export function createPluginRunnerTaskOwner(input: { let failReady!: (error: unknown) => void; let confirmStopped: (code: number) => void = () => {}; let stoppedState: Extract | undefined; + let completed = false; let rejectFailure!: (error: unknown) => void; const failure = new Promise((_resolve, reject) => { rejectFailure = reject; }); void failure.catch(() => undefined); @@ -284,7 +285,10 @@ export function createPluginRunnerTaskOwner(input: { } await input.onSessionRetained(state); } - if (submitted) await call({ kind: "complete" }); + if (submitted) { + await call({ kind: "complete" }); + completed = true; + } }, release: async (reason?: "detach") => { registrationReleased = true; @@ -328,6 +332,9 @@ export function createPluginRunnerTaskOwner(input: { return bytes; }; return { launcher, registration, driver, readWorkspaceFile, + // This is a lease-bound Launcher receipt, not a terminal phase or a sent + // stop request. Completion also proves credential/writeback obligations ran. + isRetired: () => stoppedState?.executionStopped === true && completed && !detached, rebind(next: typeof input) { if (!launched || stopRequested || stoppedState || detached || next.harness !== input.harness || input.accountIdentity !== next.accountIdentity || input.credentialIdentity !== next.credentialIdentity || diff --git a/server/src/vendor/paperclip-runner/index.ts b/server/src/vendor/paperclip-runner/index.ts index b94c0791176..94143837aee 100644 --- a/server/src/vendor/paperclip-runner/index.ts +++ b/server/src/vendor/paperclip-runner/index.ts @@ -96,6 +96,8 @@ export const runnerCodexDynamicToolsFit = runner.runnerCodexDynamicToolsFit; export const inspectWarmRunTransition = runner.inspectWarmRunTransition; export const readRunnerdArtifactBinding = runner.readRunnerdArtifactBinding; export const readLinuxProcessStartedAt = runner.readLinuxProcessStartedAt; +export const NativeSessionCloseUnrecoverableError = + runner.NativeSessionCloseUnrecoverableError; export const NativeSessionCleanupQuarantinedError = runner.NativeSessionCleanupQuarantinedError; export const NativeSessionProtocolIntegrityError =