diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 175a0108647..0253f8cb459 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -273,6 +273,24 @@ The cancellation records the board user's identity. Its acknowledged operator Stop receipt keeps recovery from promoting old queued input or creating a retry while the task closure commits. +A persisted run-only Stop can survive a controller restart before dispatch is +acknowledged. Recovery can claim an interrupted run only for its unfinished, +audited Stop and an authenticated existing remote session. A cleared task lock +does not authorize new task work or replacement of another task owner. Cleanup +reuses the run's original held budget reservation, including during a budget +pause. It cannot create a reservation, bootstrap a runner, or submit a provider +turn. The recovered session sends the existing Stop through the audited +cancellation producer. Only its acknowledgement closes the execution owner. +Cleanup skips attachment staging and warm-owner retirement. Its transport +retains authenticated pending semantic inputs without dispatch, recovery, or a +semantic reply. The server derives this purpose from current audited Stop, +lease, and held-reservation authority. Closing with unanswered input requires a +new exact provider-stop receipt confirming process exit, the durable event drain, +and identity-checked runner suspension. The input stays unanswered; the cleanup +purpose alone cannot certify termination. Ordinary execution retains its existing +semantic dispatcher requirement. Missing usage remains pending accounting debt; +a complete receipt settles the original reservation and charge once. + ## 4. Agent-Owned vs User-Owned Execution The execution model differs depending on assignee type. diff --git a/maps/agent-access-map/recovery.md b/maps/agent-access-map/recovery.md index 18894d62573..3650f12b166 100644 --- a/maps/agent-access-map/recovery.md +++ b/maps/agent-access-map/recovery.md @@ -125,6 +125,17 @@ can restore responses, safe reads recheck current authority, and uncertain mutations remain pending. The [remote continuity recipe](../../doc/testing-native-remote-continuity.md#replacing-the-main-server) separates real-process/scripted-provider evidence from hosted Daytona qualification. +Pending run-only Stop cleanup uses the server's exact audited intent and unfinished +coordinator. It grants no agent retry or new provider-turn authority. An existing +remote-session claim can reuse its original held reservation during budget pause; +fresh startup, dead-runner resume, foreign owners and missing audit proof remain +fenced. Cleanup transport retains pending semantic input without a tool dispatch +or reply, and grants no attachment staging or warm-owner retirement authority. +Closing with unanswered input still requires confirmed provider exit, durable +event drain, and exact runner suspension. +Local PostgreSQL tests exercise the cancellation producer and late usage +settlement. They do not prove hosted provider termination or deployed recovery. + ### `observe-automatic-attempt` API: `present` — GET /api/issues/{id}/recovery-actions; GET /api/issues/{id}; GET /api/heartbeat-runs/{runId}. Automatic recovery has no generic agent-enqueue endpoint. diff --git a/maps/feature-map/recovery.md b/maps/feature-map/recovery.md index 1ba9594a4d7..8d387551747 100644 --- a/maps/feature-map/recovery.md +++ b/maps/feature-map/recovery.md @@ -25,6 +25,17 @@ Agent access: [companion](../agent-access-map/recovery.md). Restart recovery respects live heartbeat preparation before workspace synchronization finishes. PostgreSQL regressions cover unchanged ownership, no duplicate dispatch, and later recovery after preparation leaves the registry. + An interrupted run with an audited pending Stop can reattach its exact remote + session only for cleanup. It reuses the original held reservation during a + budget pause and cannot start another provider turn. Local PostgreSQL + [Stop accounting](../../server/src/services/native-runtime/native-stop-accounting.integration.test.ts) + and [restart selection](../../server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts) + tests cover cleanup acknowledgement, fenced alternatives, and receipt debt. + Cleanup does not stage new attachments, retire another warm owner, or replay + pending semantic work. Authenticated Core intake retains that input unanswered. + Close still requires confirmed provider exit, durable event drain, and exact + runner suspension; the cleanup purpose alone cannot certify termination. + Hosted provider settlement remains separate acceptance. - `source-action`: `recovery/source-action` — source-scoped recovery actions remain attached to the task; independent repair work may have its own issue and return owner. - `exhaustion`: `recovery/exhaustion` — an exhausted infrastructure attempt offers ordinary Retry. diff --git a/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.test.ts b/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.test.ts index b55f34797cf..1de51aca17a 100644 --- a/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.test.ts +++ b/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.test.ts @@ -540,6 +540,80 @@ it("rebuilds a queued checkpoint after the preceding write fails", async () => { } }); +it.each(["get_task_context", "paperclip_finish", "paperclip_block"].flatMap(operationId => + [false, true].map(callbacksConfigured => ({ operationId, callbacksConfigured })), +))( + "retains unseen $operationId input without dispatch or reply (callbacks: $callbacksConfigured)", + async ({ operationId, callbacksConfigured }) => { + const root = mkdtempSync(resolve(tmpdir(), "runner-cleanup-semantic-intake-")); + const dispatch = vi.fn(async () => ({ result: "must not dispatch" })); + const recover = vi.fn(async () => ({ result: "must not recover" })); + // Keep intake, checkpointing, and cancellation on the actual authenticated + // Core. Retention must suppress even accidentally configured callbacks. + const core = new DurablePrpControlPlane({ + stateDirectory: root, identity, expectedRunnerVersion, expectedRunnerDigest, + semanticInputMode: "retain_only", + ...(callbacksConfigured ? { onSemanticToolInput: dispatch, recoverSemanticToolInput: recover } : {}), + }); + let client: AuthenticatedClient | null = null; + try { + await core.start(); + client = await authenticate(core, core.issueBootstrapTicket()); + expect(client).not.toBeNull(); + const event = semanticInputEvent(); + const payload = (event.payload as Record).payload as Record; + (payload.semantic_tool as Record).operationId = operationId; + sendSecure(client!, event); + expect(await receiveSecure(client!)).toMatchObject({ + kind: "ack", payload: { ackedSourceSeq: 1 }, + }); + expect(core.store.state.committedEvents).toHaveLength(1); + expect(core.store.state.committedEvents[0]).toMatchObject({ + eventType: "semantic_tool.input", + envelope: { payload: { payload: { semantic_tool: { operationId } } } }, + }); + expect(core.semanticToolResultsSettled()).toBe(false); + expect(core.store.state.commands).toEqual([]); + await core.recoverPendingSemanticInputs(); + await core.flushCheckpoint(); + expect(core.semanticToolResultsSettled()).toBe(false); + expect(core.store.state.commands).toEqual([]); + expect(dispatch).not.toHaveBeenCalled(); + expect(recover).not.toHaveBeenCalled(); + core.queueCommand("turn.stop", { reason: "cleanup" }, "cleanup-stop", true); + // The next real wire frame must be cancellation, never a synthetic tool + // reply or a provider-start command that could resume execution. + expect(await receiveSecure(client!)).toMatchObject({ + kind: "command", payload: { commandId: "cleanup-stop", type: "turn.stop" }, + }); + expect(core.store.state.commands.map(command => command.type)).toEqual(["turn.stop"]); + } finally { + client?.socket.destroy(); + await core.stop(); + rmSync(root, { recursive: true, force: true }); + } + }, +); + +it("rejects unseen semantic input without a dispatcher in the default execution mode", async () => { + const root = mkdtempSync(resolve(tmpdir(), "runner-missing-semantic-dispatch-")); + const core = new DurablePrpControlPlane({ stateDirectory: root, identity, expectedRunnerVersion, expectedRunnerDigest }); + let client: AuthenticatedClient | null = null; + try { + await core.start(); + client = await authenticate(core, core.issueBootstrapTicket()); + expect(client).not.toBeNull(); + sendSecure(client!, semanticInputEvent()); + expect(await receiveSecure(client!)).toBeNull(); + expect(core.store.state.committedEvents).toEqual([]); + expect(core.store.state.commands).toEqual([]); + } finally { + client?.socket.destroy(); + await core.stop(); + rmSync(root, { recursive: true, force: true }); + } +}); + it.each([true, false])("reconciles a saved semantic input only from an authoritative outcome (known=%s)", async known => { const root = mkdtempSync(resolve(tmpdir(), "runner-recovered-semantic-")); const dispatch = vi.fn(() => new Promise<{ result: unknown }>(() => undefined)); @@ -2984,7 +3058,7 @@ describe("DurablePrpControlPlane", () => { }, ); - it("latches an authenticated semantic integrity fault without ACK or dispatch and still permits suspension", async () => { + it.each(["dispatch", "retain_only"] as const)("latches an authenticated semantic integrity fault without ACK or dispatch and still permits suspension (%s)", async semanticInputMode => { const root = mkdtempSync(resolve(tmpdir(), "paperclip-prp-integrity-")); const onProtocolIntegrityError = vi.fn(); const onCommittedEvent = vi.fn(async () => undefined); @@ -2994,6 +3068,7 @@ describe("DurablePrpControlPlane", () => { identity, expectedRunnerVersion, expectedRunnerDigest, + semanticInputMode, onProtocolIntegrityError, onCommittedEvent, onSemanticToolInput, diff --git a/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts b/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts index fad19bfada3..f78c58fd57d 100644 --- a/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts +++ b/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts @@ -269,6 +269,7 @@ export interface DurablePrpControlPlaneOptions { readonly identity: DurableRecoveryIdentity; readonly warmTransitionId: string | null; }) => Promise; + semanticInputMode?: "dispatch" | "retain_only"; onSemanticToolInput?: (input: { readonly callId: string; readonly operationId: string; @@ -1921,6 +1922,7 @@ export class DurablePrpControlPlane { code: string; }> = []; #port: number | null = null; + #semanticInputMode: "dispatch" | "retain_only"; #onSemanticToolInput?: DurablePrpControlPlaneOptions["onSemanticToolInput"]; #onCommittedEvent?: DurablePrpControlPlaneOptions["onCommittedEvent"]; #recoverSemanticToolInput?: DurablePrpControlPlaneOptions["recoverSemanticToolInput"]; @@ -1975,6 +1977,7 @@ export class DurablePrpControlPlane { "Warm run transition requires its exact approved runner artifact.", ); } + this.#semanticInputMode = options.semanticInputMode ?? "dispatch"; this.#onSemanticToolInput = options.onSemanticToolInput; this.#onCommittedEvent = options.onCommittedEvent; this.#recoverSemanticToolInput = options.recoverSemanticToolInput; @@ -2126,7 +2129,7 @@ export class DurablePrpControlPlane { } async recoverPendingSemanticInputs(): Promise { - if (!this.#recoverSemanticToolInput) return; + if (this.#semanticInputMode === "retain_only" || !this.#recoverSemanticToolInput) return; const identity = this.#identity; for (const saved of [...this.#store.state.committedEvents]) { if (!unsettledSemanticInput(saved, this.#store.state)) continue; @@ -3867,7 +3870,7 @@ export class DurablePrpControlPlane { eventType === "semantic_tool.input" || eventType === "mcp_app.tool_input"; if ( isSemanticInput && - (this.#onSemanticToolInput === undefined || + ((this.#onSemanticToolInput === undefined && this.#semanticInputMode !== "retain_only") || semantic?.phase !== "input" || typeof semantic.callId !== "string" || typeof semantic.operationId !== "string" || @@ -4007,7 +4010,7 @@ export class DurablePrpControlPlane { } if (!isCurrentOwner()) return; - const liveCall = isSemanticInput && this.#onSemanticToolInput && semantic !== undefined + const liveCall = this.#semanticInputMode !== "retain_only" && isSemanticInput && this.#onSemanticToolInput && semantic !== undefined && typeof semantic.callId === "string" && typeof semantic.operationId === "string" ? { callId: semantic.callId, operationId: semantic.operationId, input: semantic.input, sourceEventId, sourceEventType: eventType, 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 ddedf75af16..3eff4e3b3e2 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts @@ -8785,6 +8785,8 @@ async function verifyLiveRunnerAdoption( adoptionAuthorization: "allow" | "deny" | "legacy" | "legacy-missing" | "missing-binding" = "allow", toolHandlerOrder?: "before" | "after" | "completion-before" | "completion-after", uncertainAtClose = false, + semanticRecoveryMode?: "denied-before-cancel" | "cleanup-only", + stopExitProofFault?: "missing" | "false", ) { const stateDirectory = await mkdtemp(join(tmpdir(), "runnerd-live-adopt-")); const pendingCompletion = toolHandlerOrder?.startsWith("completion-") === true; @@ -8814,7 +8816,7 @@ async function verifyLiveRunnerAdoption( authority = next; return { connectUrl: `ws://127.0.0.1:${address.port}/runner`, - ...(mismatchedArtifact || uncertainAtClose ? { checkpoint } : {}), + ...(mismatchedArtifact || uncertainAtClose || semanticRecoveryMode === "cleanup-only" ? { checkpoint } : {}), release: async () => { if (authority === next) authority = null; }, @@ -8857,7 +8859,19 @@ async function verifyLiveRunnerAdoption( const first = createCapabilityRunnerdCodexTransport(options); const pendingTool = vi.fn(() => new Promise>(() => undefined)); if (toolHandlerOrder) first.transport.setServerRequestHandler(pendingTool); - const recoveredTool = vi.fn(async () => ({ result: { receipt: "known-outcome" }, isError: false })); + let releaseRecovery!: () => void; + const recoveryGate = new Promise(resolve => { releaseRecovery = resolve; }); + const recoveryDenied = new Error("execution authority revoked after Stop"); + const recoveredTool = vi.fn(async () => { + if (semanticRecoveryMode) { + // Model server authority denying replay after Stop cleared the task lock. + // The gate holds that denial until reattachment has returned its session, + // before the caller invokes the eventual onSession cancellation callback. + await recoveryGate; + throw recoveryDenied; + } + return { result: { receipt: "known-outcome" }, isError: false }; + }); const redispatchedTool = vi.fn(async () => ({ success: true, contentItems: [] })); let runnerPid: number | null = null; let runnerStartedAt: string | null = null; @@ -8959,6 +8973,7 @@ async function verifyLiveRunnerAdoption( ...options, ...(preparedTask && !adoptionAuthorization.startsWith("legacy") ? { authorizeRunnerAdoption } : {}), ...(uncertainAtClose ? { closeGraceMs: 2_000 } : {}), + ...(semanticRecoveryMode === "cleanup-only" ? { recoveryPurpose: "stop_cleanup" as const } : {}), // Hash different stable bytes without replacing the real runner artifact // used by concurrent tests. Adoption must never execute this path. ...(mismatchedArtifact @@ -9055,6 +9070,87 @@ async function verifyLiveRunnerAdoption( }), ); if (toolHandlerOrder?.endsWith("after")) adopted.transport.setServerRequestHandler(redispatchedTool); + if (semanticRecoveryMode) { + const retained = authority as DurablePrpControlPlane | null; + expect(retained).not.toBeNull(); + expect(retained!.store.state.committedEvents.some(event => event.eventType === "semantic_tool.input")).toBe(true); + expect(retained!.store.state.commands.filter(command => command.type === "turn.start")).toHaveLength(1); + expect(retained!.store.state.commands.some(command => command.type === "turn.interrupt")).toBe(false); + if (semanticRecoveryMode === "denied-before-cancel") { + await vi.waitFor(() => expect(recoveredTool).toHaveBeenCalledOnce()); + // This is a real Core replay callback, not an artificial transport error. + // Deliver its denial before giving the caller its cancellation turn. + releaseRecovery(); + await vi.waitFor(async () => { + await expect(adopted!.transport.request("thread/read", {})).rejects.toBe(recoveryDenied); + }); + await expect(adopted.transport.request("turn/interrupt", {})).rejects.toBe(recoveryDenied); + expect(retained!.store.state.commands.some(command => command.type === "turn.interrupt")).toBe(false); + } else { + // Let all startup/handler microtasks and a real reconciliation read run. + // Cleanup must not enter replay even while its authority would deny it. + await adopted.transport.request("thread/read", {}); + expect(recoveredTool).not.toHaveBeenCalled(); + expect(redispatchedTool).not.toHaveBeenCalled(); + releaseRecovery(); + await expect(adopted.transport.request("turn/interrupt", {})).resolves.toEqual({}); + expect(retained!.store.state.commands.filter(command => command.type === "turn.interrupt")).toMatchObject([{ status: "completed" }]); + } + expect(redispatchedTool).not.toHaveBeenCalled(); + expect(retained!.store.state.commands.some(command => command.type === "semantic_tool.result")).toBe(false); + expect(retained!.store.state.commands.filter(command => command.type === "turn.start")).toHaveLength(1); + expect(duplicateLauncher).not.toHaveBeenCalled(); + if (semanticRecoveryMode === "cleanup-only") { + const faultedStopProof = vi.fn(); + if (stopExitProofFault) { + const queue = retained!.queueCommand.bind(retained!); + vi.spyOn(retained!, "queueCommand").mockImplementation((type, ...args) => { + const command = queue(type, ...args); + if (type === "turn.stop") { + let result = command.result; + Object.defineProperty(command, "result", { + enumerable: true, + configurable: true, + get: () => result, + set: value => { + const receipt = value as { result?: Record } | null; + if (receipt?.result && typeof receipt.result === "object") { + faultedStopProof(receipt.result.providerExitConfirmed); + const stopped = { ...receipt.result }; + if (stopExitProofFault === "missing") delete stopped.providerExitConfirmed; + else stopped.providerExitConfirmed = false; + result = { ...receipt, result: stopped }; + } else result = value; + }, + }); + } + return command; + }); + await expect(adopted.transport.close()).rejects.toBeInstanceOf(NativeSessionCloseUnrecoverableError); + expect(faultedStopProof).toHaveBeenCalledWith(true); + expect(checkpoint).toHaveBeenCalledExactlyOnceWith("unsettled"); + } else { + await expect(adopted.transport.close()).resolves.toBeUndefined(); + expect(checkpoint).toHaveBeenCalledExactlyOnceWith("settled"); + expect(retained!.store.state.commands.filter(command => command.type === "turn.stop")).toMatchObject([{ + status: "completed", result: { result: { providerExitConfirmed: true } }, + }]); + } + expect(retained!.semanticToolResultsSettled()).toBe(false); + expect(retained!.store.state.commands.some(command => command.type === "semantic_tool.result")).toBe(false); + expect(retained!.store.state.committedEvents.some(event => event.eventType === "semantic_tool.input")).toBe(true); + if (!stopExitProofFault) { + expect(retained!.store.state.commands.filter(command => command.type === "runner.suspend")).toMatchObject([{ status: "completed" }]); + const runnerState = JSON.parse(await readFile(join(stateDirectory, "runner", "runner-state.json"), "utf8")); + expect(runnerState).toMatchObject({ ...identity, lifecycle: "suspended" }); + } + const providerState = JSON.parse(await readFile(join(stateDirectory, "runner", "codex-provider-state.json"), "utf8")); + expect(providerState).toMatchObject({ activeProviderTurnId: null, pendingEvents: [], queuedEvents: [] }); + expect(recoveredTool).not.toHaveBeenCalled(); + expect(redispatchedTool).not.toHaveBeenCalled(); + } + return; + } if (toolHandlerOrder) { await vi.waitFor(() => { expect(pendingCompletion ? redispatchedTool : recoveredTool).toHaveBeenCalledOnce(); @@ -9143,6 +9239,7 @@ async function verifyLiveRunnerAdoption( expect(commandTypes).not.toContain("runner.suspend"); } } finally { + releaseRecovery(); await adopted?.transport.close().catch(() => undefined); if (uncertainAtClose) { const providerPid = adopted?.evidence().codexPid; @@ -9871,3 +9968,12 @@ it.each(["before", "after"] as const)("settles a saved tool outcome with the han it.each(["completion-before", "completion-after"] as const)("recovers a pending finish without an ordinary-tool lookup (%s)", order => verifyLiveRunnerAdoption(false, false, false, false, false, false, "allow", order), 30_000); + +it("exposes semantic recovery before cancellation when the authority has been revoked", () => + verifyLiveRunnerAdoption(false, false, false, false, false, false, "allow", "before", false, "denied-before-cancel"), 30_000); + +it.each(["before", "after", "completion-before", "completion-after"] as const)("allows cleanup-only adoption to cancel without replaying a pending semantic input (%s)", order => + verifyLiveRunnerAdoption(false, false, false, false, false, false, "allow", order, false, "cleanup-only"), 30_000); + +it.each(["missing", "false"] as const)("refuses cleanup settlement when the exact stop receipt has %s provider exit proof", fault => + verifyLiveRunnerAdoption(false, false, false, false, false, false, "allow", "before", false, "cleanup-only", fault), 30_000); diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts index 790b10fe8d4..ccdf72eaa11 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts @@ -1245,6 +1245,7 @@ export interface CapabilityRunnerdCodexTransportOptions extends RunnerExecutionD onAccountingEvent?: (event: PrpEvent) => Promise; onRecoveryAccountingEvent?: (event: PrpEvent, sourceIdentity: DurableRecoveryIdentity) => Promise; recoverSemanticToolInput?: DurablePrpControlPlaneOptions["recoverSemanticToolInput"]; + recoveryPurpose?: "execution" | "stop_cleanup"; opencodePermissionMode?: NativeOpenCodePermissionMode; acpxAgent?: QualifiedAcpxAgent; /** Explicit evaluation-only candidate selection, never derived from persisted session input. */ @@ -4111,7 +4112,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { } #recoverPendingSemanticInputsWhenReady(): void { - if (this.#startupComplete && this.#handlerInstalled && this.#sourceRecoveryIdentity === null && this.#pendingWarmRecoveryCompletion === null) { + if (this.options.recoveryPurpose !== "stop_cleanup" && this.#startupComplete && this.#handlerInstalled && this.#sourceRecoveryIdentity === null && this.#pendingWarmRecoveryCompletion === null) { void Promise.resolve().then(() => this.#core?.recoverPendingSemanticInputs()) .catch(error => this.#failTransport(error instanceof Error ? error : new Error(String(error)))); } @@ -4511,7 +4512,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { } } - async #stopProviderBeforeSuspend(deadline: number): Promise { + async #stopProviderBeforeSuspend(deadline: number): Promise { const state = this.#providerDrainState(); const core = this.#core; const inferredActiveProviderTurnId = @@ -4534,10 +4535,10 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { && (state === null || (state.activeProviderTurnId === null && state.providerSettled)); if ( state === "unreadable" || - (activeProviderTurnId === null && !stopIdlePi) || + (activeProviderTurnId === null && !stopIdlePi && this.options.recoveryPurpose !== "stop_cleanup") || core === null ) { - return; + return false; } const commandId = `command_close_stop_${randomUUID().replaceAll("-", "")}`; core.queueCommand( @@ -4557,29 +4558,32 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { ? "stopped idle Pi provider before runner suspension" : `stopped active provider turn ${activeProviderTurnId} before runner suspension`, ); - return; + return record(record(command.result).result).providerExitConfirmed === true; } if (command !== undefined && command.status !== "pending") { this.#diagnostic( `provider turn stop ${command.status} before runner suspension`, ); - return; + return false; } - if (await this.#runnerHasExited()) return; + if (await this.#runnerHasExited()) return false; await new Promise((resolveWait) => setTimeout(resolveWait, 5)); } this.#diagnostic("provider turn stop timed out before runner suspension"); + return false; } async #drainSettledProviderEventsBeforeSuspend( timeoutMs = 1_000, + providerStopConfirmed = false, ): Promise { const core = this.#core; if (!core) return false; try { const drained = await awaitProviderDrainBarrier({ readProviderState: () => this.#providerDrainState(), - semanticResultsSettled: () => core.semanticToolResultsSettled(), + semanticResultsSettled: () => core.semanticToolResultsSettled() + || (this.options.recoveryPurpose === "stop_cleanup" && providerStopConfirmed), commands: () => core.store.state.commands, queueDrain: (commandId) => { core.queueCommand("runner.drain", {}, commandId, true); @@ -4775,6 +4779,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { // `unsettled` and preserve its original bootstrap diagnostic. let runnerSuspended = false; let providerDrained = false; + let cleanupProviderStopConfirmed = false; let suspensionRequired = false; let lastSuspensionState: Record | null = null; let suspensionState: { @@ -4793,11 +4798,14 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { // fresh run authority never inherits the prior run's pending events. if (!(await this.#runnerHasExited())) { + if (this.options.recoveryPurpose === "stop_cleanup") { + cleanupProviderStopConfirmed = await this.#stopProviderBeforeSuspend(preparationDeadline); + } // Let an already-admitted tool result reach its original provider // before turn.stop can retire that tool-call identity. This shares the // close preparation deadline; a stuck callback never stalls cleanup. while ( - !this.#core.semanticToolResultsSettled() && + !this.#core.semanticToolResultsSettled() && !cleanupProviderStopConfirmed && Date.now() < preparationDeadline ) { this.#pumpEventsSafely(); @@ -4809,9 +4817,12 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { // trip may need to carry. Give both cases the same budget so a // slow-but-idle runner is not held to a tighter deadline than a // runner that just stopped a turn. - await this.#stopProviderBeforeSuspend(preparationDeadline); + if (this.options.recoveryPurpose !== "stop_cleanup") { + await this.#stopProviderBeforeSuspend(preparationDeadline); + } providerDrained = await this.#drainSettledProviderEventsBeforeSuspend( Math.min(5_000, Math.max(0, preparationDeadline - Date.now())), + cleanupProviderStopConfirmed, ); } // Local durable roots are reused too. Process exit alone cannot prove @@ -4920,7 +4931,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { const runnerSettled = preparationConfirmed && runnerSuspended && providerDrained && - this.#core?.semanticToolResultsSettled() === true && + (this.#core?.semanticToolResultsSettled() === true || cleanupProviderStopConfirmed) && (finalProviderState === null || (finalProviderState !== "unreadable" && finalProviderState.pendingEventCount === 0 && @@ -5282,8 +5293,9 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { expectedRunnerVersion: runnerArtifact.version, expectedRunnerDigest: runnerArtifact.digest, onProtocolIntegrityError: (error) => this.#failTransport(error), - onSemanticToolInput: (call) => this.#handleSemanticToolInput(call), - recoverSemanticToolInput: call => this.#recoverSemanticToolInput(call), + semanticInputMode: this.options.recoveryPurpose === "stop_cleanup" ? "retain_only" : "dispatch", + onSemanticToolInput: this.options.recoveryPurpose === "stop_cleanup" ? undefined : (call) => this.#handleSemanticToolInput(call), + recoverSemanticToolInput: this.options.recoveryPurpose === "stop_cleanup" ? undefined : call => this.#recoverSemanticToolInput(call), onCommittedEvent: async (event) => { if (this.#sourceRecoveryIdentity && event.runId === this.#sourceRecoveryIdentity.runId) { if (!this.options.onRecoveryAccountingEvent && this.options.onAccountingEvent) throw new Error("native_runner_source_accounting_unavailable"); @@ -5775,8 +5787,9 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { expectedRunnerVersion: runnerArtifact.version, expectedRunnerDigest: runnerArtifact.digest, onProtocolIntegrityError: (error) => this.#failTransport(error), - onSemanticToolInput: (call) => this.#handleSemanticToolInput(call), - recoverSemanticToolInput: call => this.#recoverSemanticToolInput(call), + semanticInputMode: this.options.recoveryPurpose === "stop_cleanup" ? "retain_only" : "dispatch", + onSemanticToolInput: this.options.recoveryPurpose === "stop_cleanup" ? undefined : (call) => this.#handleSemanticToolInput(call), + recoverSemanticToolInput: this.options.recoveryPurpose === "stop_cleanup" ? undefined : call => this.#recoverSemanticToolInput(call), onCommittedEvent: async (event) => { if (this.#sourceRecoveryIdentity && event.runId === this.#sourceRecoveryIdentity.runId) { if (!this.options.onRecoveryAccountingEvent && this.options.onAccountingEvent) throw new Error("native_runner_source_accounting_unavailable"); diff --git a/server/src/modules/run-dispatch/adapters/postgres.test.ts b/server/src/modules/run-dispatch/adapters/postgres.test.ts index 397d7cc001c..5b494545f79 100644 --- a/server/src/modules/run-dispatch/adapters/postgres.test.ts +++ b/server/src/modules/run-dispatch/adapters/postgres.test.ts @@ -841,6 +841,44 @@ describeEmbeddedPostgres("run-dispatch postgres adapter", () => { return { companyId, agentId, issueId, runId, owner, reservation }; } + it.each(["fresh", "bootstrap", "dead-runner", "expired", "foreign-owner", "foreign-intent", "missing-audit", "foreign-audit", "stale-generation", "terminal", "no-hold", "settled-hold"] as const)( + "keeps %s work fenced while a company is paused for budget", async mode => { + const f = await seedRecoveryAdmission(); + const intentAuditId = randomUUID(); + await db.insert(activityLog).values({ id: intentAuditId, companyId: f.companyId, actorType: "system", actorId: "native-session-cancellation", + action: "native.cancellation_intent_recorded", entityType: "heartbeat_run", entityId: f.runId, runId: f.runId, agentId: f.agentId }); + await db.update(nativeRunFinalizations).set({ controllerGeneration: 2, attempt: 1, recoveryState: "awaiting_runner_reattach", + recoveryHistory: [{ disposition: "reattach_remote_runner", controllerGeneration: 2, providerAttempt: 1 }], + ...(mode === "bootstrap" ? { recoveryState: "bootstrap_incomplete" } : {}), + ...(mode === "dead-runner" ? { recoveryState: "resuming_session" } : {}), + ...(mode === "expired" ? { leaseExpiresAt: new Date(Date.now() - 1) } : {}), + ...(mode === "foreign-owner" ? { leaseOwner: "another-owner" } : {}), + ...(mode === "stale-generation" ? { controllerGeneration: 3 } : {}), + ...(mode === "terminal" ? { phase: "terminal_failure" } : {}), + }).where(eq(nativeRunFinalizations.runId, f.runId)); + await db.update(heartbeatRuns).set({ resultJson: mode === "fresh" ? {} : { + startupCancellation: { beforeNativeSelection: false, errorCode: "cancelled", requestedAt: new Date().toISOString() }, + nativeCancellation: { schema: "paperclip.native-cancellation.v1", companyId: f.companyId, runId: f.runId, + issueId: mode === "foreign-intent" ? randomUUID() : f.issueId, scope: "run", reasonCode: "cancellation_run_only", + intentId: `native-cancellation:${randomUUID()}`, intentAuditId: mode === "missing-audit" ? randomUUID() : intentAuditId, + dispatchState: "pending", dispatched: false, effects: ["release_run_resources"] }, + } }).where(eq(heartbeatRuns.id, f.runId)); + if (mode === "foreign-audit") await db.update(activityLog).set({ entityId: randomUUID() }).where(eq(activityLog.id, intentAuditId)); + if (mode === "no-hold") await db.delete(budgetReservations).where(eq(budgetReservations.id, f.reservation.id)); + if (mode === "settled-hold") await db.update(budgetReservations).set({ state: "settled" }).where(eq(budgetReservations.id, f.reservation.id)); + await db.update(companies).set({ status: "paused", pauseReason: "budget" }).where(eq(companies.id, f.companyId)); + await expect(reserveRunBudget(db, f.companyId, f.runId, null, {}, f.owner)).rejects.toThrow(); + const dispatch = vi.fn(async () => undefined); + const gate = await createPostgresRunDispatchAdapter(db).dispatchResolvedInteractionIfCurrent({ companyId: f.companyId, runId: f.runId, + reservationId: f.reservation.id, recoveryLeaseOwner: f.owner, expectedStatus: "running", now: new Date(), dispatch }) + .catch(error => ({ dispatched: false, error })); + expect(gate.dispatched).toBe(false); + expect(dispatch).not.toHaveBeenCalled(); + expect(await db.select().from(budgetReservations).where(eq(budgetReservations.runId, f.runId))) + .toHaveLength(mode === "no-hold" ? 0 : 1); + expect(await db.select().from(costEvents).where(eq(costEvents.heartbeatRunId, f.runId))).toHaveLength(0); + }); + it.each(["reservation", "dispatch"] as const)("orders recovery %s behind a coordinator before taking finalization's issue and run locks", async boundary => { const f = await seedRecoveryAdmission(); let signalLocked!: () => void, allowFinalization!: () => void; diff --git a/server/src/modules/run-dispatch/adapters/postgres.ts b/server/src/modules/run-dispatch/adapters/postgres.ts index cb59e363c1a..26266084275 100644 --- a/server/src/modules/run-dispatch/adapters/postgres.ts +++ b/server/src/modules/run-dispatch/adapters/postgres.ts @@ -10,6 +10,7 @@ import { notifyUnsafeWorkspaceExportRecovery } from "../../../services/unsafe-wo import { notifyCancelledWorkspacePreservation } from "../../../services/cancelled-workspace-preservation-notifications.js"; import { notifyInstructionCollection } from "../../../services/instruction-collection-notifications.js"; import { notifyScheduledRetryCancellation } from "../../../services/scheduled-retry-cancellation-notifications.js"; +import { readClaimedNativeStopRecovery } from "../../../services/native-runtime/native-cancellation-request.js"; import { hasConversationContinuationPolicy } from "../../../services/conversation-continuation.js"; import { getExecutionBlocker } from "../../../services/execution-blocker.js"; import { validateReconstructedWorkspaceDispatch } from "../../../services/native-runtime/native-workspace-reconstruction.js"; @@ -1278,10 +1279,11 @@ export function createPostgresRunDispatchAdapter( input: DispatchResolvedInteractionInput, ): Promise> { const dispatchLockedRun = async (tx: Db, run: HeartbeatRun, publications: ActivityPublication[]) => { - if (run.status !== input.expectedStatus || run.resultJson?.cancellation || run.resultJson?.startupCancellation) { + const stopRecovery = await readClaimedNativeStopRecovery(tx, run, input.recoveryLeaseOwner); + if (run.status !== input.expectedStatus || ((run.resultJson?.cancellation || run.resultJson?.startupCancellation) && !stopRecovery)) { return { dispatched: false as const, cancellation: { outcome: "lost_race" as const } }; } - if (input.enforceContinuationOwnership !== false) { + if (!stopRecovery && input.enforceContinuationOwnership !== false) { const { issueId, decision: initialDecision } = await decideCurrentRunStaleness( tx, run, @@ -1366,7 +1368,8 @@ export function createPostgresRunDispatchAdapter( return withRunOwnershipLocks>(input, () => ({ dispatched: false as const, cancellation: { outcome: "lost_race" as const } }), async (tx, run) => { const claim = parseObject(parseObject(run.usageJson).budgetDispatch); - if (run.status !== input.expectedStatus || run.costAccountedAt || run.resultJson?.cancellation || run.resultJson?.startupCancellation + const stopRecovery = await readClaimedNativeStopRecovery(tx, run, input.recoveryLeaseOwner); + if (run.status !== input.expectedStatus || run.costAccountedAt || ((run.resultJson?.cancellation || run.resultJson?.startupCancellation) && !stopRecovery) || claim.reservationId !== admission.claim.reservationId || claim.recoveryLeaseOwner !== admission.claim.recoveryLeaseOwner || claim.admittedAt !== admission.claim.admittedAt || claim.handedOffAt) return { dispatched: false as const, cancellation: { outcome: "lost_race" as const } }; @@ -1381,13 +1384,13 @@ export function createPostgresRunDispatchAdapter( .where(eq(companies.id, run.companyId)); if (!scope || scope.reservationId !== input.reservationId) return { dispatched: false as const, admissionBlockReason: "Provider dispatch no longer owns a held budget reservation" }; - if (scope.projectId && !scope.currentProjectId) + if (!stopRecovery && scope.projectId && !scope.currentProjectId) return { dispatched: false as const, admissionBlockReason: "Reserved project no longer exists" }; - if (scope.companyStatus !== "active" || scope.projectPausedAt || scope.lifecycleState !== "ready" - || ["paused", "terminated", "pending_approval"].includes(scope.agentStatus)) + if (!stopRecovery && (scope.companyStatus !== "active" || scope.projectPausedAt || scope.lifecycleState !== "ready" + || ["paused", "terminated", "pending_approval"].includes(scope.agentStatus))) return { dispatched: false as const, admissionBlockReason: "Execution scope is paused or no longer ready for work" }; - if (input.enforceContinuationOwnership !== false) { + if (!stopRecovery && input.enforceContinuationOwnership !== false) { const contextSnapshot = parseObject(run.contextSnapshot); const issueId = readNonEmptyString(contextSnapshot.issueId); if (issueId) { @@ -1408,7 +1411,7 @@ export function createPostgresRunDispatchAdapter( cancellation: await cancelStaleRunInTx(tx, run, issueId, decision, input.expectedStatus, input.now, true) }; } } - const budgetBlock = await recheckAdmittedRunBudgetReadOnly(tx, { companyId: run.companyId, agentId: run.agentId, + const budgetBlock = stopRecovery ? null : await recheckAdmittedRunBudgetReadOnly(tx, { companyId: run.companyId, agentId: run.agentId, runId: run.id, projectId: scope.projectId, reservationId: input.reservationId }); if (budgetBlock) return { dispatched: false as const, admissionBlockReason: budgetBlock }; if (input.recoveryLeaseOwner) { diff --git a/server/src/services/budget-reservations.ts b/server/src/services/budget-reservations.ts index 3d9621f3581..9f771e37047 100644 --- a/server/src/services/budget-reservations.ts +++ b/server/src/services/budget-reservations.ts @@ -6,6 +6,7 @@ import { withAccountingTransaction } from "./accounting-transaction.js"; import { budgetServiceInTransaction } from "./budgets.js"; import { withBudgetSpan } from "./budget-tracing.js"; import type { ActivityPublication } from "./activity-log.js"; +import { readClaimedNativeStopRecovery } from "./native-runtime/native-cancellation-request.js"; /** Native finalization takes its coordinator before issue/run locks. Recovery * admission uses the same order after the company accounting lock, so its @@ -31,7 +32,8 @@ export async function recheckRunBudgetReservationInTransaction( reservationId: string, recoveryLeaseOwner?: string, ): Promise { - if (run.resultJson?.cancellation || run.resultJson?.startupCancellation) + const stopRecovery = await readClaimedNativeStopRecovery(tx, run, recoveryLeaseOwner); + if ((run.resultJson?.cancellation || run.resultJson?.startupCancellation) && !stopRecovery) return "Run cancellation already requested"; if (run.costAccountedAt || !["queued", "running"].includes(run.status)) return "Run can no longer start provider work"; @@ -48,6 +50,7 @@ export async function recheckRunBudgetReservationInTransaction( sql`${nativeRunFinalizations.leaseExpiresAt} > clock_timestamp()`)) : []; if (!owner) return "Native recovery no longer owns provider dispatch"; } + if (stopRecovery) return null; const { block, policies, observations } = await budgetServiceInTransaction(tx, publications) .evaluateAdmission(run.companyId, run.agentId, { projectId: reservation.projectId }); if (block) return block.reason; @@ -75,7 +78,8 @@ export async function reserveRunBudget(db: Db, companyId: string, runId: string, await lockRunBudgetRecoveryOwnerInTransaction(tx, companyId, runId, recoveryLeaseOwner); const [run] = await tx.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.id, runId), eq(heartbeatRuns.companyId, companyId))).for("update"); if (!run) throw notFound("Run not found"); - if (run.resultJson?.cancellation || run.resultJson?.startupCancellation) throw conflict("Run cancellation already requested"); + const stopRecovery = await readClaimedNativeStopRecovery(tx, run, recoveryLeaseOwner); + if ((run.resultJson?.cancellation || run.resultJson?.startupCancellation) && !stopRecovery) throw conflict("Run cancellation already requested"); if (run.costAccountedAt || !["queued", "running"].includes(run.status)) throw conflict("Run can no longer start provider work"); const [existing] = await tx.select().from(budgetReservations).where(and(eq(budgetReservations.companyId, companyId), eq(budgetReservations.runId, runId))); if (existing) { @@ -87,6 +91,7 @@ export async function reserveRunBudget(db: Db, companyId: string, runId: string, eq(nativeRunFinalizations.runId, runId), eq(nativeRunFinalizations.leaseOwner, recoveryLeaseOwner), sql`${nativeRunFinalizations.leaseExpiresAt} > clock_timestamp()`)) : []; if (owner) { + if (stopRecovery) return { ...existing, reused: true }; // Recovery has already made this run queued/running, so its own pending // accounting is excluded. Current overruns, other unfinished runs and // scope pauses still apply before any more paid work starts. @@ -100,6 +105,7 @@ export async function reserveRunBudget(db: Db, companyId: string, runId: string, } throw conflict("Provider dispatch has already reserved this run"); } + if (stopRecovery) throw conflict("Native Stop recovery requires the original held reservation"); const { block, policies, observations } = await budgetServiceInTransaction(tx, publications).evaluateAdmission(companyId, run.agentId, { projectId }); if (block) return { admissionBlockReason: block.reason }; // A budget release can start asynchronous resume hooks. Capacity alone diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index b9878f8ae20..7ab9773f517 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -774,6 +774,7 @@ export interface HeartbeatServiceOptions { */ nativeSessionBackendFactory?: ( execution: NativeExecutionInput, + context?: { recoveryPurpose: "execution" | "stop_cleanup" }, ) => NativeSessionBackend; /** Test seam for observing the native-session shutdown boundary before lease destruction. */ closeWarmNativeSessionsForRun?: (input: { diff --git a/server/src/services/heartbeat/runtime-execution.ts b/server/src/services/heartbeat/runtime-execution.ts index 268ebf4aa41..bcbe9cd71ba 100644 --- a/server/src/services/heartbeat/runtime-execution.ts +++ b/server/src/services/heartbeat/runtime-execution.ts @@ -1,3 +1,4 @@ +import { readClaimedNativeStopRecovery } from "../native-runtime/native-cancellation-request.js"; import { NativeStartupCleanupPendingError } from "../native-runtime/native-startup-cleanup.js"; import { NativeControllerRecoveryWaitingError } from "../native-runtime/native-controller-recovery.js"; @@ -113,7 +114,7 @@ export interface HeartbeatRuntimeExecutionInput { run: typeof heartbeatRuns.$inferSelect; agent: typeof agents.$inferSelect; options: { - nativeSessionBackendFactory?: (execution: NativeExecutionInput) => NativeSessionBackend; + nativeSessionBackendFactory?: (execution: NativeExecutionInput, context?: { recoveryPurpose: "execution" | "stop_cleanup" }) => NativeSessionBackend; pluginWorkerManager?: PluginWorkerManager; }; task: Pick & { @@ -605,6 +606,8 @@ export async function executeHeartbeatRuntime(db: Db, input: HeartbeatRuntimeExe nativeDispatchAtMs, }), ); + const stopCleanup = runOptions.nativeRestartRecovery?.kind === "reattach_remote_runner" + && await readClaimedNativeStopRecovery(db, run, runOptions.nativeLeaseOwner); const guardedDispatch = await dispatchWithTiming( (markDispatchStarted) => { @@ -622,7 +625,9 @@ export async function executeHeartbeatRuntime(db: Db, input: HeartbeatRuntimeExe restartRecovery: runOptions.nativeRestartRecovery, onExecutionLeaseAcquired: runOptions.onNativeExecutionLeaseAcquired, backend: - options.nativeSessionBackendFactory?.(nativeExecution), + stopCleanup + ? options.nativeSessionBackendFactory?.(nativeExecution, { recoveryPurpose: "stop_cleanup" }) + : options.nativeSessionBackendFactory?.(nativeExecution), useRunnerd: agent.adapterType === "paperclip_runner", adapterType: agent.adapterType, sessionGoalControl, diff --git a/server/src/services/heartbeat/runtime-selection.test.ts b/server/src/services/heartbeat/runtime-selection.test.ts index f5d4603d050..d3d3a8ba4c3 100644 --- a/server/src/services/heartbeat/runtime-selection.test.ts +++ b/server/src/services/heartbeat/runtime-selection.test.ts @@ -2,7 +2,7 @@ import { randomUUID } from "node:crypto"; import { mkdtemp, rm, writeFile } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; -import { agents, companies, completionContracts, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, heartbeatRuns, issues, nativeRunFinalizations, type Db } from "@paperclipai/db"; +import { activityLog, agents, companies, completionContracts, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, heartbeatRuns, issues, nativeRunFinalizations, type Db } from "@paperclipai/db"; import { eq, sql } from "drizzle-orm"; import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; @@ -14,6 +14,7 @@ import { resolveHeartbeatNativeRuntimeMode } from "../native-runtime/runtime-mod import { claimNativeRestartRecoveries } from "../native-runtime/native-restart-recovery.js"; import { buildNativeModelEnvelope, canonicalNativeRuntimeContextDigest } from "../../vendor/paperclip-runner/index.js"; import { selectHeartbeatRuntime, type HeartbeatRuntimeSelectionInput } from "./runtime-selection.js"; +import { reserveRunBudget } from "../budget-reservations.js"; import { legacyControllerClaim } from "../legacy-controller-lease.js"; // Keep the real contract, session, and runtime-selection transactions. Replace only @@ -266,6 +267,42 @@ describe.skipIf(!support.supported)("heartbeat runtime selection boundary", () = expect((await read(run.id)).contextSnapshot?.skillVersions).toEqual({ [key]: versionId }); }); + it.each(["exact", "fresh", "bootstrap", "dead-runner", "expired", "foreign-owner", "missing-audit"] as const)( + "admits only exact pending Stop cleanup at runtime selection (%s)", async mode => { + const { input, run, issue } = await fixture(); + const initial = await selectHeartbeatRuntime(db, input); + if (!initial.selected || !initial.nativeExecution) throw new Error("Expected original native selection"); + await reserveRunBudget(db, run.companyId, run.id, null); + const saved = await read(run.id), leaseOwner = `stop:${randomUUID()}`, intentAuditId = randomUUID(); + input.run = saved; + input.session.persistedNativeExecutionInput = initial.nativeExecution; + input.session.persistedRunnerProfile = saved.runnerProfileJson ?? {}; + await db.insert(activityLog).values({ id: intentAuditId, companyId: run.companyId, actorType: "system", actorId: "native-session-cancellation", + action: "native.cancellation_intent_recorded", entityType: "heartbeat_run", entityId: run.id, runId: run.id, agentId: run.agentId }); + await db.update(nativeRunFinalizations).set({ leaseOwner: mode === "foreign-owner" ? "another-owner" : leaseOwner, + leaseExpiresAt: new Date(Date.now() + (mode === "expired" ? -1 : 60_000)), controllerGeneration: 2, attempt: 1, + recoveryState: mode === "bootstrap" ? "bootstrap_incomplete" : mode === "dead-runner" ? "resuming_session" : "awaiting_runner_reattach", + recoveryHistory: [{ disposition: "reattach_remote_runner", controllerGeneration: 2, providerAttempt: 1 }], + }).where(eq(nativeRunFinalizations.runId, run.id)); + await db.update(heartbeatRuns).set({ resultJson: { + startupCancellation: { beforeNativeSelection: false, errorCode: "cancelled", requestedAt: new Date().toISOString() }, + executionCancellation: { state: "requested" }, + nativeCancellation: { schema: "paperclip.native-cancellation.v1", companyId: run.companyId, runId: run.id, issueId: issue.id, + scope: "run", reasonCode: "cancellation_run_only", intentId: `native-cancellation:${randomUUID()}`, + intentAuditId: mode === "missing-audit" ? randomUUID() : intentAuditId, dispatchState: "pending", dispatched: false, + effects: ["release_run_resources"] }, + } }).where(eq(heartbeatRuns.id, run.id)); + input.recovery = mode === "fresh" ? {} : { nativeLeaseOwner: leaseOwner, nativeRestartRecovery: { + kind: "reattach_remote_runner", runId: run.id, leaseOwner, controllerGeneration: 2, providerAttempt: 1, + restartKind: "hard", recoveryRequestId: null, remote: { providerLeaseId: "existing-sandbox", remoteCwd: "/workspace" }, + } }; + const selected = await selectHeartbeatRuntime(db, input); + expect(selected.selected).toBe(mode === "exact"); + expect(await coordinators(run.id)).toHaveLength(1); + expect((await read(run.id)).nativeSessionId).toBe(saved.nativeSessionId); + expect((await read(run.id)).runnerProfileJson?.nativeExecutionInput).toEqual(initial.nativeExecution); + }); + it.each(["cancelled", "startup_cancellation"])("does not create a native coordinator when %s wins preparation", async winner => { const { input, run } = await fixture(); input.beforeNativeRuntimeSelection = async () => { diff --git a/server/src/services/heartbeat/runtime-selection.ts b/server/src/services/heartbeat/runtime-selection.ts index 49d96cf762b..6fab2c82c42 100644 --- a/server/src/services/heartbeat/runtime-selection.ts +++ b/server/src/services/heartbeat/runtime-selection.ts @@ -1,3 +1,4 @@ +import { readClaimedNativeStopRecovery } from "../native-runtime/native-cancellation-request.js"; import { museRunnerBroker } from "../muse-runner-broker.js"; import { assertEphemeralNativeLifecycle, readRemoteWorkspaceAuthority } from "../remote-workspaces.js"; import { measureRunTransaction } from "../execution-transaction-timing.js"; @@ -927,7 +928,11 @@ export async function selectHeartbeatRuntime(db: Db, input: HeartbeatRuntimeSele if (!lockedRun) throw new Error("native_runtime_run_missing"); // Cancellation and runtime selection serialize on this row. A // stopped preparation must never create a new native coordinator. - if (lockedRun.status !== "running" || lockedRun.resultJson?.startupCancellation) return false; + const stopRecovery = runOptions.nativeRestartRecovery?.kind === "reattach_remote_runner" + && runOptions.nativeRestartRecovery.runId === lockedRun.id + && runOptions.nativeRestartRecovery.leaseOwner === runOptions.nativeLeaseOwner + && await readClaimedNativeStopRecovery(tx as unknown as Db, lockedRun, runOptions.nativeLeaseOwner); + if (lockedRun.status !== "running" || (lockedRun.resultJson?.startupCancellation && !stopRecovery)) return false; if (lockedRun.runtimeMode === "legacy" && lockedRun.controllerBootId && !(await renewLegacyControllerLease(tx as unknown as Db, lockedRun))) { onNativeOwnershipHeld(); diff --git a/server/src/services/native-runtime/native-cancellation-request.ts b/server/src/services/native-runtime/native-cancellation-request.ts index a7f04911425..6150f030c26 100644 --- a/server/src/services/native-runtime/native-cancellation-request.ts +++ b/server/src/services/native-runtime/native-cancellation-request.ts @@ -1,6 +1,6 @@ import { notifyScheduledRetryCancellation } from "../scheduled-retry-cancellation-notifications.js"; import { and, eq, sql } from "drizzle-orm"; -import { heartbeatRuns, nativeRunFinalizations, type Db } from "@paperclipai/db"; +import { activityLog, budgetReservations, heartbeatRuns, nativeRunFinalizations, type Db } from "@paperclipai/db"; import { badRequest, conflict } from "../../errors.js"; const UUID = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i; @@ -12,6 +12,57 @@ export function cancellationRequestId(value: unknown): string | undefined { } export const cancellationIntentId = (requestId: string) => `native-cancellation:${requestId}`; +type StopRun = Pick; +type StopRecoveryOwner = Pick; + +export function hasPendingNativeRunStop(run: StopRun): boolean { + const intent = record(record(run.resultJson).nativeCancellation); + return run.runtimeMode === "native" && !!run.nativeIssueId + && intent.schema === "paperclip.native-cancellation.v1" && intent.scope === "run" && intent.reasonCode === "cancellation_run_only" + && intent.runId === run.id && intent.companyId === run.companyId && intent.issueId === run.nativeIssueId + && typeof intent.intentId === "string" && intent.intentId.startsWith("native-cancellation:") + && UUID.test(intent.intentId.slice("native-cancellation:".length)) + && typeof intent.intentAuditId === "string" && UUID.test(intent.intentAuditId) + && intent.dispatchState === "pending" && intent.dispatched === false + && Array.isArray(intent.effects) && intent.effects.length === 1 && intent.effects[0] === "release_run_resources"; +} + +export function isClaimedNativeStopRecovery(run: StopRun, owner: StopRecoveryOwner | undefined, leaseOwner: string | undefined, now = Date.now()): boolean { + if (!leaseOwner || !hasPendingNativeRunStop(run) || run.status !== "running" || !owner + || owner.runId !== run.id || owner.companyId !== run.companyId || owner.issueId !== run.nativeIssueId + || owner.phase !== "observed" || owner.resultId || owner.leaseOwner !== leaseOwner + || !owner.leaseExpiresAt || owner.leaseExpiresAt.getTime() <= now || owner.recoveryState !== "awaiting_runner_reattach") return false; + const latest = record(Array.isArray(owner.recoveryHistory) ? owner.recoveryHistory.at(-1) : null); + return latest.disposition === "reattach_remote_runner" && latest.controllerGeneration === owner.controllerGeneration + && latest.providerAttempt === owner.attempt; +} + +export async function readPendingNativeRunStop(db: Db, run: StopRun): Promise { + if (!hasPendingNativeRunStop(run)) return false; + const intent = record(record(run.resultJson).nativeCancellation); + const [audit] = await db.select({ id: activityLog.id }).from(activityLog).where(and( + eq(activityLog.id, intent.intentAuditId as string), eq(activityLog.companyId, run.companyId), + eq(activityLog.runId, run.id), eq(activityLog.entityId, run.id), eq(activityLog.entityType, "heartbeat_run"), + eq(activityLog.action, "native.cancellation_intent_recorded"), + )).limit(1); + return !!audit; +} + +export async function readClaimedNativeStopRecovery(db: Db, run: StopRun, leaseOwner?: string): Promise { + if (!leaseOwner || !await readPendingNativeRunStop(db, run)) return false; + const [owner] = await db.select().from(nativeRunFinalizations).where(and( + eq(nativeRunFinalizations.runId, run.id), eq(nativeRunFinalizations.companyId, run.companyId), + eq(nativeRunFinalizations.issueId, run.nativeIssueId!), + )).limit(1); + if (!isClaimedNativeStopRecovery(run, owner, leaseOwner)) return false; + const [hold] = await db.select({ id: budgetReservations.id }).from(budgetReservations).where(and( + eq(budgetReservations.companyId, run.companyId), eq(budgetReservations.runId, run.id), + eq(budgetReservations.agentId, run.agentId), eq(budgetReservations.state, "held"), + )).limit(1); + return !!hold; +} + /** Run only while holding this company's run row lock. An earlier uncorrelated * startup fence is also a conflicting Stop, even before native intent exists. */ export function assertCancellationRequest(result: unknown, requestId: string, allowUnclaimed = false) { diff --git a/server/src/services/native-runtime/native-restart-recovery.ts b/server/src/services/native-runtime/native-restart-recovery.ts index 635808df5ef..a24fccec657 100644 --- a/server/src/services/native-runtime/native-restart-recovery.ts +++ b/server/src/services/native-runtime/native-restart-recovery.ts @@ -1,3 +1,4 @@ +import { readPendingNativeRunStop } from "./native-cancellation-request.js"; import { notifyLiveRunRecovery } from "../live-run-recovery-notifications.js"; import { NATIVE_RETAINED_OWNER_WAIT } from "./native-controller-recovery.js"; import { notifyNativeEnvironmentSettlement } from "../native-environment-settlement-notifications.js"; @@ -524,7 +525,10 @@ export async function claimNativeRestartRecoveries(input: { and( eq(heartbeatRuns.runtimeMode, "native"), eq(agents.adapterType, "paperclip_runner"), - inArray(heartbeatRuns.status, ["running", "failed"]), + or(inArray(heartbeatRuns.status, ["running", "failed"]), and( + eq(heartbeatRuns.status, "interrupted"), eq(nativeRunFinalizations.phase, "observed"), + sql`${heartbeatRuns.resultJson}->'nativeCancellation'->>'dispatchState' = 'pending'`, + )), isNull(nativeRunFinalizations.resultId), // A pre-provider retained-owner wait belongs to normal same-run // resumption, including when its deadline elapsed while we were down. @@ -599,6 +603,12 @@ export async function claimNativeRestartRecoveries(input: { if (!row) { return busy(); } + const stopCleanup = row.coordinator.phase === "observed" + && row.coordinator.companyId === row.run.companyId && row.coordinator.issueId === row.run.nativeIssueId + && await readPendingNativeRunStop(tx as unknown as Db, row.run); + if (!["running", "failed"].includes(row.run.status) && !(row.run.status === "interrupted" && stopCleanup)) { + return { kind: "blocked", runId: row.run.id, reason: "run_no_longer_recoverable" } as const; + } if (isNativeRunnerOwnershipHeld(row.run)) { return { kind: "blocked", @@ -616,7 +626,7 @@ export async function claimNativeRestartRecoveries(input: { if (!input.liveRunExecutions?.has(row.run.id) && await retireSupersededNativeRetry(tx as unknown as Db, { run: row.run, coordinator: row.coordinator, issueExecutionRunId: row.issueExecutionRunId, now, })) return { kind: "blocked", runId: row.run.id, reason: "native_retry_superseded" } as const; - if (row.issueExecutionRunId !== row.run.id) { + if (row.issueExecutionRunId !== row.run.id && !(stopCleanup && row.issueExecutionRunId === null)) { return { kind: "blocked", runId: row.run.id, @@ -921,6 +931,9 @@ export async function claimNativeRestartRecoveries(input: { nativeRestartInterruptedTurnId(checkpointRecord as unknown as PersistedNativeSession) === null, providerAttempt: row.coordinator.attempt, }); + if (stopCleanup && classification.claimKind !== "reattach_remote_runner") { + return { kind: "blocked", runId: row.run.id, reason: "native_stop_requires_existing_remote_session" } as const; + } const ownershipChanged = row.issueAssigneeAgentId !== row.run.agentId || ["done", "cancelled"].includes(row.issueStatus); diff --git a/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts b/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts index dd2f92e779a..4430e7c2121 100644 --- a/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts +++ b/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts @@ -10,6 +10,7 @@ import { tmpdir } from "node:os"; import { resolve } from "node:path"; import { and, eq, inArray, sql } from "drizzle-orm"; import { + activityLog, agents, authUsers, companies, @@ -1213,8 +1214,8 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", ( expect(run?.processPid).toBe(remotePid); }); - it("gives a recovery claim its full lease after an old batch observation", async () => { - const fixture = await seedRun("LATE-BATCH-CLAIM"); + it.each(["running", "interrupted-stop", "interrupted-no-intent", "interrupted-foreign-intent", "interrupted-missing-audit", "interrupted-foreign-owner", "interrupted-terminal", "interrupted-no-checkpoint"])("gives only eligible recovery a full lease after an old batch observation (%s)", async mode => { + const fixture = await seedRun(`LATE-BATCH-CLAIM-${mode}`); const binding = { companyId, issueId: fixture.issueId, runId: fixture.runId }; const root = resolve(runtimeRoot, fixture.runId); const observedAt = new Date(Date.now() - 90_000); @@ -1236,9 +1237,45 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", ( // Set up an abandoned predecessor. Never alter the successor's deadline. await fixture.db.update(nativeRunFinalizations).set({ leaseExpiresAt: new Date(observedAt.getTime() - 1_000) }) .where(eq(nativeRunFinalizations.runId, fixture.runId)); + if (mode !== "running") { + const intentAuditId = randomUUID(), requestedAt = new Date().toISOString(); + await fixture.db.insert(activityLog).values({ id: intentAuditId, companyId, actorType: "system", actorId: "native-session-cancellation", + action: "native.cancellation_intent_recorded", entityType: "heartbeat_run", entityId: fixture.runId, runId: fixture.runId, agentId }); + await fixture.db.update(heartbeatRuns).set({ status: "interrupted", finishedAt: new Date(), + errorCode: "lease_released_before_terminal", resultJson: { + cancelledByActorType: "user", cancelledByUserId: "test-board", + cancellation: { expected: true, initiator: { type: "user" }, reason: "Operator requested Stop" }, + startupCancellation: { beforeNativeSelection: false, requestedAt, errorCode: "cancelled" }, + executionCancellation: { state: "requested", requestedAt }, + nativeCancellation: { schema: "paperclip.native-cancellation.v1", ...binding, + ...(mode === "interrupted-foreign-intent" ? { issueId: randomUUID() } : {}), scope: "run", reasonCode: "cancellation_run_only", + intentId: `native-cancellation:${randomUUID()}`, intentAuditId: mode === "interrupted-missing-audit" ? randomUUID() : intentAuditId, dispatchState: "pending", dispatched: false, + effects: ["release_run_resources"] }, + } }).where(eq(heartbeatRuns.id, fixture.runId)); + await fixture.db.update(issues).set({ executionRunId: null, checkoutRunId: null }).where(eq(issues.id, fixture.issueId)); + if (mode === "interrupted-no-intent") await fixture.db.update(heartbeatRuns).set({ resultJson: {} }).where(eq(heartbeatRuns.id, fixture.runId)); + if (mode === "interrupted-foreign-owner") { + const other = await seedRun(`OTHER-${mode}`); + await fixture.db.update(issues).set({ executionRunId: other.runId }).where(eq(issues.id, fixture.issueId)); + } + if (mode === "interrupted-terminal") await fixture.db.update(nativeRunFinalizations).set({ phase: "terminal_failure" }) + .where(eq(nativeRunFinalizations.runId, fixture.runId)); + if (mode === "interrupted-no-checkpoint") await fixture.db.update(heartbeatRuns).set({ + runnerProfileJson: sql`${heartbeatRuns.runnerProfileJson} - 'sessionCheckpoint'`, + }).where(eq(heartbeatRuns.id, fixture.runId)); + } const controller = { ...successor, bootId: randomUUID() }; const [claim] = await claimNativeRestartRecoveries({ db: fixture.db, controller, restartKind: "hard", now: observedAt, runIds: [fixture.runId] }); + if (mode !== "running" && mode !== "interrupted-stop") { + expect(["reattach_existing_runner", "reattach_remote_runner", "resume_dead_runner", "bootstrap_incomplete"]).not.toContain(claim?.kind); + const [unchanged] = await fixture.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, fixture.runId)); + expect(unchanged.status).toBe("interrupted"); + const [owner] = await fixture.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, fixture.runId)); + expect(owner.controllerGeneration).toBe(1); + expect(owner.leaseOwner).toBe("previous-batch-owner"); + return; + } expect(claim).toMatchObject({ kind: "reattach_remote_runner", runId: fixture.runId }); if (!("leaseOwner" in claim)) throw new Error("Expected a recovery claim"); const [saved] = await fixture.db.execute(sql`select extract(epoch from (lease_expires_at - clock_timestamp())) as remaining 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 35845b35f5d..bd1844e8dd7 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -38,6 +38,8 @@ import { tmpdir } from "node:os"; import { inspect } from "node:util"; import { dirname, join } from "node:path"; import { + activityLog, + budgetReservations, heartbeatRuns, environmentLeases, heartbeatRunEvents, @@ -591,7 +593,7 @@ describe("remote controller restart adoption", () => { ).rejects.toThrow("runner_remote_recovery_identity_mismatch"); expect(execute).toHaveBeenCalledTimes(1); }); - it("wires remote adoption when only controller state survived on the host", async () => { + it.each(["execution", "stop_cleanup", "invalid_cleanup"] as const)("wires remote adoption when only controller state survived on the host (%s)", async purpose => { const stateBase = await mkdtemp( join(tmpdir(), "paperclip-remote-reattach-"), ); @@ -619,14 +621,33 @@ describe("remote controller restart adoption", () => { join(root, "control-plane", "control-plane-state.json"), JSON.stringify(durableControlPlaneState(identity)), ); - await createRunnerdBackend({ - db: leaseDb(current), + const stopIntent = { startupCancellation: { beforeNativeSelection: false }, nativeCancellation: { + schema: "paperclip.native-cancellation.v1", scope: "run", reasonCode: "cancellation_run_only", ...current.binding, + intentId: `native-cancellation:${randomUUID()}`, intentAuditId: randomUUID(), dispatchState: "pending", dispatched: false, + effects: ["release_run_resources"], + } }; + const stopDb = leaseDb(current, { leaseOwner: purpose === "invalid_cleanup" ? "another-owner" : claim.leaseOwner, + leaseExpiresAt: new Date(Date.now() + 60_000), controllerGeneration: 2, attempt: 1, + recoveryState: "awaiting_runner_reattach", recoveryHistory: [{ disposition: "reattach_remote_runner", controllerGeneration: 2, providerAttempt: 1 }], + }, stopIntent); + state.createTransport.mockClear(); + const adopted = createRunnerdBackend({ + db: purpose === "execution" ? leaseDb(current) : stopDb, execution: current, runnerInstanceId: "runner", runnerExecutionTarget: target as never, restartRecovery: claim, + ...(purpose === "execution" ? {} : { recoveryPurpose: "stop_cleanup" as const }), }); + if (purpose === "invalid_cleanup") { + await expect(adopted).rejects.toThrow("native_cancellation_pending_recovery"); + expect(state.createTransport).not.toHaveBeenCalled(); + return; + } + await adopted; state.createBackend.mock.calls.at(-1)![1].codexTransportFactory!(); + if (purpose === "stop_cleanup") expect(state.createTransport.mock.calls.at(-1)![0]).toHaveProperty("recoveryPurpose", "stop_cleanup"); + else expect(state.createTransport.mock.calls.at(-1)![0]).not.toHaveProperty("recoveryPurpose"); expect( state.createTransport.mock.calls.at(-1)![0].adoptExistingRunner?.pid, ).toBe(123); @@ -6437,6 +6458,8 @@ type LeaseCoordinator = { controllerBootId?: string; controllerPid?: number; controllerGeneration?: number; + recoveryState?: string; + recoveryHistory?: Array>; runId: string; companyId: string; issueId: string; @@ -6517,6 +6540,10 @@ function leaseDb( checkoutRunId: null, }, ] + : table === budgetReservations + ? [{ id: "original-held-reservation" }] + : table === activityLog + ? [{ id: (runResultJson.nativeCancellation as Record | undefined)?.intentAuditId }] : table === heartbeatRunEvents ? priorAccountingEvents.map((event, seq) => ({ seq, eventType: event.eventType, sourceSeq: event.sourceSeq, payload: { prpEvent: event } })) : []; @@ -6750,12 +6777,22 @@ describe("native startup cancellation fence", () => { expect(cancel).toHaveBeenCalledOnce(); expect(submit).not.toHaveBeenCalled(); }); - it.each(["current", "stale"] as const)("reattaches a %s remote recovery claim only to deliver an existing Stop", async mode => { + it.each([ + { mode: "current", startupFence: false }, + { mode: "stale", startupFence: false }, + { mode: "current", startupFence: true }, + ] as const)("reattaches a $mode remote recovery claim only to deliver an existing Stop (startup fence=$startupFence)", async ({ mode, startupFence }) => { const { currentNativeControllerIdentity } = await import("./native-restart-recovery.js"); const controller = await currentNativeControllerIdentity(); - const resultJson = { nativeCancellation: { schema: "paperclip.native-cancellation.v1", scope: "run", + const resultJson = { ...(startupFence ? { startupCancellation: { beforeNativeSelection: false, + requestedAt: new Date().toISOString(), errorCode: "cancelled" }, + cancellation: { source: "operator", expected: true, initiator: { type: "user" }, + reason: "Operator requested Stop", recordedAt: new Date().toISOString() }, + executionCancellation: { state: "requested", requestedAt: new Date().toISOString() } } : {}), + nativeCancellation: { schema: "paperclip.native-cancellation.v1", scope: "run", companyId: execution.binding.companyId, runId: execution.binding.runId, issueId: execution.binding.issueId, - dispatchState: "pending", intentId: "native-cancellation:recovered-stop", effects: ["release_run_resources"] } }; + dispatchState: "pending", dispatched: false, intentId: `native-cancellation:${randomUUID()}`, + intentAuditId: randomUUID(), reasonCode: "cancellation_run_only", effects: ["release_run_resources"] } }; const cancel = vi.fn(() => ({ cleanup: Promise.resolve() })), submit = vi.fn(); state.execute.mockReset().mockImplementationOnce(async options => { try { await options.onSession({ cancel }); submit(); } @@ -6763,7 +6800,9 @@ describe("native startup cancellation fence", () => { throw new Error("a stopped recovery must not submit a turn"); }); const db = leaseDb(execution, { leaseOwner: "stop-recovery", controllerBootId: controller.bootId, - controllerPid: controller.pid, controllerGeneration: 2, attempt: 1 }, resultJson); + controllerPid: controller.pid, controllerGeneration: 2, attempt: 1, + leaseExpiresAt: new Date(Date.now() + 60_000), recoveryState: "awaiting_runner_reattach", + recoveryHistory: [{ disposition: "reattach_remote_runner", controllerGeneration: 2, providerAttempt: 1 }] }, resultJson); await expect(executePaperclipNativeSession({ db, execution, runnerInstanceId: "recovered-stop", leaseOwner: "stop-recovery", restartRecovery: { kind: "reattach_remote_runner", runId: execution.binding.runId, leaseOwner: "stop-recovery", controllerGeneration: mode === "current" ? 2 : 1, providerAttempt: 1, @@ -6895,9 +6934,9 @@ describe("native startup restart detachment", () => { it.each(["authority", "questions"])("keeps a post-publication Stop owner through %s attachment and restart", async (seam) => { const restarting = structuredClone(execution); restarting.binding.runId = `restart-stop-after-publication-${seam}`; - let published!: () => void, releaseRead!: () => void, acknowledgeStop!: () => void; + let releaseRead!: () => void, acknowledgeStop!: () => void; let cleaning!: () => void, releaseCleanup!: () => void; - const publication = new Promise(resolve => { published = resolve; }); + const published = vi.fn(); const read = new Promise(resolve => { releaseRead = resolve; }); const receipt = new Promise(resolve => { acknowledgeStop = resolve; }); const cleanupStarted = new Promise(resolve => { cleaning = resolve; }); @@ -6907,12 +6946,20 @@ describe("native startup restart detachment", () => { const submit = vi.fn(); const db = leaseDb(restarting); const select = db.select.bind(db); + let publishingSession = false; db.select = ((fields?: Record) => { - if (seam === "authority" && fields && Object.keys(fields).length === 2 && "status" in fields && "resultJson" in fields) { - return { from: () => ({ where: () => ({ limit: () => { - published(); - return read.then(() => [{ status: "running", resultJson: {} }]); - } }) }) }; + if (seam === "authority" && !fields && publishingSession) { + const query = select(); + return { from: (table: unknown) => { + const source = query.from(table as never); + return table === heartbeatRuns ? { where: (condition: never) => { + const filtered = source.where(condition); + return { limit: (count: number) => { + published(); + return read.then(() => filtered.limit(count)); + } }; + } } : source; + } }; } if (seam === "questions" && !fields) { const query = select(); @@ -6924,6 +6971,7 @@ describe("native startup restart detachment", () => { }) as typeof db.select; state.execute.mockReset().mockImplementationOnce(async options => { try { + publishingSession = true; await options.onSession({ detachControllerForRestart: detach, cancel, resolveRuntimeRequest: vi.fn() }); submit(); } finally { @@ -6933,11 +6981,13 @@ describe("native startup restart detachment", () => { } }); const outcome = executePaperclipNativeSession({ db, execution: restarting, runnerInstanceId: "runner" }).catch(error => error); - await publication; - const stopping = cancelNativeSession(restarting.binding.runId, "operator Stop after publication"); const completed = vi.fn(); - const shutdown = detachNativeSessionsForRestart([restarting.binding.runId]).then(completed); + let stopping: Promise | undefined; + let shutdown: Promise | undefined; try { + await vi.waitFor(() => expect(published).toHaveBeenCalled(), { timeout: 5_000 }); + stopping = cancelNativeSession(restarting.binding.runId, "operator Stop after publication"); + shutdown = detachNativeSessionsForRestart([restarting.binding.runId]).then(completed); releaseRead(); await new Promise(resolve => setImmediate(resolve)); expect(cancel).toHaveBeenCalledOnce(); @@ -9772,6 +9822,66 @@ describe("native warm session supervision", () => { expect(state.execute).toHaveBeenCalledTimes(1); }); + it.each([false, true])("keeps warm owners and attachment staging untouched during exact Stop cleanup (invalidated=%s)", async invalidated => { + const { base, next, run } = fixture("identity-native-stop-cleanup"); + const close = vi.fn<() => Promise>(async () => { throw new Error("Stop cleanup must not retire another warm owner"); }); + const cancel = vi.fn(() => ({ cleanup: Promise.resolve() })); + const originalStaging = state.stageNativeRunnerWakeAttachments.getMockImplementation()!; + state.execute.mockReset().mockImplementationOnce(async options => { await options.onSession?.({ close }); return result; }); + await run(base); + await createRunnerdBackend({ db: leaseDb(next), execution: next, runnerInstanceId: "stop-cleanup" }); + state.createBackend.mock.calls.at(-1)![1].codexTransportFactory!(); + const root = state.createTransport.mock.calls.at(-1)![0].stateDirectory!; + const remoteIdentity = { runId: next.binding.runId, normalizedSessionId: next.session.normalizedSessionId!, + runnerInstanceId: "stop-cleanup", environmentLeaseId: next.binding.executionWorkspaceId, turnId: "stop-turn", itemId: "stop-item" }; + const authority = new DurablePrpControlPlane({ stateDirectory: join(root, "control-plane"), identity: remoteIdentity, + expectedRunnerVersion: "stop-fixture", expectedRunnerDigest: `sha256:${"a".repeat(64)}` }); + await authority.flushCheckpoint(); + await authority.stop(); + const controller = await import("./native-restart-recovery.js").then(module => module.currentNativeControllerIdentity()); + const owner = "exact-stop-cleanup"; + const resultJson = { cancelledByActorType: "user", cancelledByUserId: "test-board", + startupCancellation: { beforeNativeSelection: false }, nativeCancellation: { + schema: "paperclip.native-cancellation.v1", scope: "run", reasonCode: "cancellation_run_only", ...next.binding, + intentId: `native-cancellation:${randomUUID()}`, intentAuditId: randomUUID(), dispatchState: "pending", dispatched: false, + effects: ["release_run_resources"], + } }; + const db = leaseDb(next, { leaseOwner: owner, leaseExpiresAt: new Date(Date.now() + 60_000), + controllerBootId: controller.bootId, controllerPid: controller.pid, controllerGeneration: 2, attempt: 1, + recoveryState: "awaiting_runner_reattach", recoveryHistory: [{ disposition: "reattach_remote_runner", controllerGeneration: 2, providerAttempt: 1 }], + }, resultJson); + state.stageNativeRunnerWakeAttachments.mockClear().mockImplementation(async () => { throw new Error("Stop cleanup must not stage attachments"); }); + const submit = vi.fn(() => { throw new Error("Stop cleanup cannot submit a turn"); }); + state.execute.mockImplementationOnce(async options => { + if (invalidated) { + Reflect.deleteProperty(resultJson, "startupCancellation"); + Reflect.deleteProperty(resultJson, "nativeCancellation"); + } + try { await options.onSession?.({ cancel }); submit(); throw new Error("Stop cleanup cannot return to execution"); } + finally { await options.onSession?.(null); } + }); + try { + await expect(executePaperclipNativeSession({ db, execution: next, runnerInstanceId: "stop-cleanup", useRunnerd: true, + backend: {} as never, runnerExecutionTarget: { kind: "remote", transport: "sandbox", remoteCwd: "/workspace", + sandboxLeaseAcquisition: { outcome: "resumed", providerLeaseId: "existing-sandbox" }, runner: { execute: async request => ({ + exitCode: 0, signal: null, pid: 123, startedAt: new Date().toISOString(), timedOut: false, stderr: "", stdout: request.args?.[1]?.includes("base64") + ? Buffer.from(JSON.stringify({ ...remoteIdentity, lifecycle: "running" })).toString("base64") + : request.args?.[1]?.includes("cat --") ? "nonce\n123\n2026-09-19T10:00:00.000Z\nstop-cleanup\nlinux:abcd-1234:100\n" : "", + }) } }, leaseOwner: owner, restartRecovery: { kind: "reattach_remote_runner", runId: next.binding.runId, leaseOwner: owner, + controllerGeneration: 2, providerAttempt: 1, restartKind: "hard", recoveryRequestId: null, + remote: { providerLeaseId: "existing-sandbox", remoteCwd: "/workspace" } }, + })).rejects.toThrow("native_cancellation_pending_recovery"); + expect(close).not.toHaveBeenCalled(); + expect(state.stageNativeRunnerWakeAttachments).not.toHaveBeenCalled(); + expect(cancel).toHaveBeenCalledTimes(invalidated ? 0 : 1); + expect(submit).not.toHaveBeenCalled(); + } finally { + state.stageNativeRunnerWakeAttachments.mockReset().mockImplementation(originalStaging); + close.mockImplementation(async () => undefined); + await closeIdleWarmNativeSessionsForRestart(); + } + }); + it("does not retire an active turn when a new session identity arrives", async () => { const { base, next, run } = fixture("identity-active"); let finish!: () => void; diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 12f079da14a..fd3d9ccc29b 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -1,4 +1,5 @@ import { nativeCancellationHasUnsettledOwner, notifyNativeCancellation, waitForNativeCancellation } from "./native-cancellation-delivery.js"; +import { readClaimedNativeStopRecovery } from "./native-cancellation-request.js"; import { publishNativeHarnessBackup } from "./native-harness-backup-publication.js"; import { rehydratePinnedSkillAssets } from "./runtime-context.js"; import { createNativeUsageDelivery } from "./native-usage-delivery.js"; @@ -75,7 +76,7 @@ import { createArtifactInstaller } from "./artifact-transfer.js"; import { readVerifiedRemoteWorkspaceFile } from "./remote-deliverable-file.js"; import { copyBackCodexAuth } from "@paperclipai/adapter-codex-local/server"; import { nativeCompletionFeedback } from "./native-completion-feedback.js"; -import { hasAcknowledgedNativeReassignmentStopIntent, hasAcknowledgedNativeStopIntent } from "../acknowledged-native-stop.js"; +import { hasAcknowledgedNativeReassignmentStopIntent, hasAcknowledgedNativeStopIntent, hasAcknowledgedNativeRunStopReceipt } from "../acknowledged-native-stop.js"; import { stoppedCodexTurnIsTextOnly } from "./stopped-codex-turn.js"; import { prepareVerifiedRemoteProviderPack, useVerifiedPreinstalledRemoteProviderPack, shouldUseBundledAcpxImageAssets } from "./remote-provider-pack.js"; import { uploadVerifiedProviderPack } from "./provider-pack-upload.js"; @@ -8189,6 +8190,17 @@ export async function executePaperclipNativeSession(input: { if (input.execution.schema === "paperclip.native-execution-input.v6" || input.execution.schema === "paperclip.native-execution-input.v8") { return await executePaperclipNativeSessionWithinScope(input); } + if (input.restartRecovery?.kind === "reattach_remote_runner" + && input.restartRecovery.runId === input.execution.binding.runId + && input.restartRecovery.leaseOwner === input.leaseOwner) { + const [stoppedRun] = await input.db.select().from(heartbeatRuns).where(and( + eq(heartbeatRuns.id, input.execution.binding.runId), eq(heartbeatRuns.companyId, input.execution.binding.companyId), + eq(heartbeatRuns.agentId, input.execution.binding.agentId), eq(heartbeatRuns.nativeIssueId, input.execution.binding.issueId), + )).limit(1); + if (stoppedRun && await readClaimedNativeStopRecovery(input.db, stoppedRun, input.leaseOwner)) { + return await executePaperclipNativeSessionWithinScope(input, undefined, true); + } + } const workspaceRoot = input.execution.workspace.cwd; if (workspaceRoot === null) throw new Error("native_workspace_required"); const needsWarmRetirement = [...warmNativeSessions].some(([scope, entry]) => @@ -8325,6 +8337,7 @@ export async function executePaperclipNativeSession(input: { async function executePaperclipNativeSessionWithinScope( input: Parameters[0], prepareAfterClaim?: () => Promise[0]>, + cleanupOnly = false, ): Promise { if (input.execution.provider.kind === "acpx" && input.execution.provider.agent === "antigravity") { const candidate = resolveAntigravityCandidateAdmission({ @@ -8505,6 +8518,7 @@ async function executePaperclipNativeSessionWithinScope( const executionLeaseTtlMs = input.runnerExecutionTarget?.kind === "remote" ? REMOTE_CONTROLLER_LEASE_MS : NATIVE_SESSION_EXECUTION_LEASE_TTL_MS; let attempt: number; + let claimedStopCleanup = false; let controllerGeneration = 0; const providerPackAdmissionEpoch = randomUUID(); let providerPackAdmissionSource: NativeProviderPackAdmissionSource | undefined; @@ -8569,7 +8583,9 @@ async function executePaperclipNativeSessionWithinScope( } // A cancellation can win after heartbeat dispatch admission but // before this claim. Never revive a terminal run or a settled startup. - if (boundRun.status !== "running" || boundRun.resultJson?.startupCancellation || + const stopRecovery = input.restartRecovery?.kind === "reattach_remote_runner" + && await readClaimedNativeStopRecovery(tx as unknown as Db, { ...boundRun, id: input.execution.binding.runId }, leaseOwner); + if ((cleanupOnly && !stopRecovery) || boundRun.status !== "running" || (boundRun.resultJson?.startupCancellation && !stopRecovery) || coordinator.phase === "terminal_failure") { throw new NativeCancellationPendingRecoveryError(); } @@ -8596,7 +8612,7 @@ async function executePaperclipNativeSessionWithinScope( // Fresh execution, dead-runner resume and acknowledged Stops remain // fenced: none authorize launching another provider turn. if (input.restartRecovery?.kind !== "reattach_remote_runner" - || cancellationIntent.dispatchState !== "pending") { + || cancellationIntent.dispatchState !== "pending" || !stopRecovery) { throw new NativeCancellationPendingRecoveryError(); } } @@ -8711,6 +8727,7 @@ async function executePaperclipNativeSessionWithinScope( receipt: boundRun.runnerProfileJson?.nativeProviderPackAdmission, state: boundRun.runnerProfileJson?.nativeProviderPackAdmissionState }; controllerGeneration = nextControllerGeneration; + claimedStopCleanup = stopRecovery; await tx .update(heartbeatRuns) .set({ @@ -8761,7 +8778,7 @@ async function executePaperclipNativeSessionWithinScope( // Claim current native authority before the old warm checkpoint writer // must prove its same-controller successor. Shared staging stays after // strict predecessor suspension; it cannot run on a failed claim/Stop. - if (input.useRunnerd) await retireSupersededWarmNativeSessions( + if (input.useRunnerd && !claimedStopCleanup) await retireSupersededWarmNativeSessions( nativeSessionOwnerScope(input.execution, input.runnerExecutionTarget?.environmentId ?? null), nativeSessionScopeKey(input.execution), nativeSessionKey(input.execution), ); @@ -9563,6 +9580,7 @@ async function executePaperclipNativeSessionWithinScope( const create = () => createRunnerdBackend({ ...input, providerPackAdmissionSource, + recoveryPurpose: claimedStopCleanup ? "stop_cleanup" : "execution", executionContext, onAccountingEvent: producerUsage?.capture, controllerPersistence: input.runnerExecutionTarget?.kind === "remote" ? { @@ -9613,7 +9631,7 @@ async function executePaperclipNativeSessionWithinScope( persistWarmNativeCheckpoint(remoteWarmRecovery.previousExecution, nativeSessionConfigDigest(remoteWarmRecovery.previousExecution, "remote"), await recovered.session.snapshot()); persistedWarmSession = loadWarmNativeCheckpoint(input.execution, warmConfigDigest!, "remote"); - runnerdBackend = await createRunnerdBackend({ ...input, providerPackAdmissionSource, onAccountingEvent: producerUsage?.capture, runnerInstanceId: effectiveRunnerInstanceId, + runnerdBackend = await createRunnerdBackend({ ...input, providerPackAdmissionSource, recoveryPurpose: claimedStopCleanup ? "stop_cleanup" : "execution", onAccountingEvent: producerUsage?.capture, runnerInstanceId: effectiveRunnerInstanceId, controllerPersistence: input.runnerExecutionTarget?.kind === "remote" ? { bootId: controller.bootId } : undefined, durableEnvironmentLeaseId: durableRunnerBinding?.environmentLeaseId, trace, toolTrace }); } @@ -9911,10 +9929,7 @@ async function executePaperclipNativeSessionWithinScope( // session is still opening. Publishing the handle before this // read closes both sides of the race: earlier Stop is durable; // later Stop can cancel this exact active session. - const [currentRun] = await input.db.select({ - status: heartbeatRuns.status, - resultJson: heartbeatRuns.resultJson, - }).from(heartbeatRuns).where(and( + const [currentRun] = await input.db.select().from(heartbeatRuns).where(and( eq(heartbeatRuns.id, input.execution.binding.runId), eq(heartbeatRuns.companyId, input.execution.binding.companyId), eq(heartbeatRuns.agentId, input.execution.binding.agentId), @@ -9927,11 +9942,22 @@ async function executePaperclipNativeSessionWithinScope( await active.cancellationSettled; throw new NativeCancellationPendingRecoveryError(); } + if (claimedStopCleanup) { + if (!currentRun || !await readClaimedNativeStopRecovery(input.db, currentRun, leaseOwner)) { + throw new NativeCancellationPendingRecoveryError(); + } + await cancelNativeSession(input.execution.binding.runId, "Recovered durable native run cancellation", { db: input.db, scope: "run" }); + throw new NativeCancellationPendingRecoveryError(); + } if (!currentRun || currentRun.status !== "running" || currentRun.resultJson?.startupCancellation || (cancellation.scope === "run" && ["pending", "acknowledged"].includes(String(cancellation.dispatchState)))) { - await cancelNativeSession(input.execution.binding.runId, "Run stopped during native session startup"); + if (input.restartRecovery?.kind === "reattach_remote_runner") { + await cancelNativeSession(input.execution.binding.runId, "Run stopped during native session startup", { db: input.db, scope: "run" }); + } else { + await cancelNativeSession(input.execution.binding.runId, "Run stopped during native session startup"); + } // The execution-owned finally closes the session when this // callback fails; no provider turn may follow publication. throw new NativeCancellationPendingRecoveryError(); @@ -10098,7 +10124,7 @@ async function executePaperclipNativeSessionWithinScope( // another turn to perform completion bookkeeping. const stoppedBeforeFirstTurn = error instanceof Error && error.message === "native_session_cancelled"; if (!startupUnsettled && protocolIntegrityFailure === null && error instanceof Error && - (stoppedBeforeFirstTurn || error instanceof NativePermissionDeclinedError || error.message === "native_finalization_missing: session returned no semantic result")) { + (stoppedBeforeFirstTurn || error instanceof NativePermissionDeclinedError || error instanceof NativeCancellationPendingRecoveryError || error.message === "native_finalization_missing: session returned no semantic result")) { const [stoppedRun] = await input.db.select().from(heartbeatRuns).where(and( eq(heartbeatRuns.id, input.execution.binding.runId), eq(heartbeatRuns.companyId, input.execution.binding.companyId), @@ -10117,7 +10143,8 @@ async function executePaperclipNativeSessionWithinScope( stopIntent.scope === "run" && typeof stopIntent.intentAuditId === "string" && stopIntent.intentAuditId.length > 0 && ["pending", "acknowledged"].includes(String(stopIntent.dispatchState)); - if (stoppedRun && (pendingStartupStop || hasAcknowledgedNativeStopIntent(stoppedRun) || hasAcknowledgedNativeReassignmentStopIntent(stoppedRun))) { + if (stoppedRun && (pendingStartupStop || hasAcknowledgedNativeStopIntent(stoppedRun) || hasAcknowledgedNativeReassignmentStopIntent(stoppedRun) + || (error instanceof NativeCancellationPendingRecoveryError && hasAcknowledgedNativeRunStopReceipt(stoppedRun)))) { const [settled] = await writeExecutionControl(input.db, [DELIVERY_QUEUES.executionFinalization, DELIVERY_QUEUES.accounting], tx => tx.update(nativeRunFinalizations).set({ phase: "terminal_failure", failureCode: "native_retry_cancelled", nextAttemptAt: null, leaseOwner: null, leaseExpiresAt: null, controlDeadlineAt: null, recoveryState: null, @@ -12324,6 +12351,7 @@ export async function createRunnerdBackend(input: { runnerInstanceId: string; chatAttachmentReadScope?: NativeChatAttachmentReadScope; restartRecovery?: NativeRestartRecoveryClaim; + recoveryPurpose?: "execution" | "stop_cleanup"; durableEnvironmentLeaseId?: string; remoteWarmRecovery?: Omit & { revalidate: () => Promise }; retainedPluginTaskOwner?: ReturnType; @@ -12371,6 +12399,16 @@ export async function createRunnerdBackend(input: { }, ) => Promise; }): Promise { + if (input.recoveryPurpose === "stop_cleanup") { + const [run] = await input.db.select().from(heartbeatRuns).where(and( + eq(heartbeatRuns.id, input.execution.binding.runId), eq(heartbeatRuns.companyId, input.execution.binding.companyId), + eq(heartbeatRuns.agentId, input.execution.binding.agentId), eq(heartbeatRuns.nativeIssueId, input.execution.binding.issueId), + )).limit(1); + if (input.restartRecovery?.kind !== "reattach_remote_runner" || input.restartRecovery.runId !== input.execution.binding.runId + || !run || !await readClaimedNativeStopRecovery(input.db, run, input.restartRecovery.leaseOwner)) { + throw new NativeCancellationPendingRecoveryError(); + } + } const sessionScopeId = nativeSessionScopeKey(input.execution); const scopeOwner = executingRunnerdSessionScopes.get(sessionScopeId); if (scopeOwner && scopeOwner !== input.execution.binding.runId) { @@ -12814,6 +12852,7 @@ async function createRunnerdBackendWithinSessionClaim( return nativeCompletionFeedback(input.db, current.runId, result); }, codexTransportFactory: recovery => createRunnerdCodexTransport({ provider: provider.kind, stateDirectory: root, + ...(input.recoveryPurpose === "stop_cleanup" ? { recoveryPurpose: "stop_cleanup" as const } : {}), measureOperation: process.env.OTEL_EXPORTER_OTLP_ENDPOINT?.trim() ? measureRunnerOperation : undefined, recordTimingReceipt: process.env.OTEL_EXPORTER_OTLP_ENDPOINT?.trim() ? recordRunnerTimingReceipt : undefined, ...owner.driver, runnerArtifact: input.controllerPersistence?.runnerArtifact ?? description.runner, @@ -14176,6 +14215,7 @@ async function createRunnerdBackendWithinSessionClaim( ), codexTransportFactory: (recoveryContext) => { const bundle = createRunnerdCodexTransport({ + ...(input.recoveryPurpose === "stop_cleanup" ? { recoveryPurpose: "stop_cleanup" as const } : {}), measureOperation: process.env.OTEL_EXPORTER_OTLP_ENDPOINT?.trim() ? measureRunnerOperation : undefined, recordTimingReceipt: process.env.OTEL_EXPORTER_OTLP_ENDPOINT?.trim() ? recordRunnerTimingReceipt : undefined, onSpawn: input.onSpawn, diff --git a/server/src/services/native-runtime/native-stop-accounting.integration.test.ts b/server/src/services/native-runtime/native-stop-accounting.integration.test.ts index 904286a3634..d2ef4378083 100644 --- a/server/src/services/native-runtime/native-stop-accounting.integration.test.ts +++ b/server/src/services/native-runtime/native-stop-accounting.integration.test.ts @@ -1,14 +1,30 @@ import { randomUUID } from "node:crypto"; -import { eq } from "drizzle-orm"; +import { resolve } from "node:path"; +import { eq, sql } from "drizzle-orm"; import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; -import { agents, companies, completionContracts, costEvents, createDb, heartbeatRuns, issues, nativeRunFinalizations } from "@paperclipai/db"; -import type { ControlPlanePort, NativeExecutionInputV1, PrpEvent } from "@paperclipai/paperclip-runner"; +import { authUsers, companyMemberships, agents, companies, completionContracts, costEvents, createDb, budgetReservations, activityLog, environments, environmentLeases, heartbeatRuns, issues, nativeRunFinalizations } from "@paperclipai/db"; +import { executeNativeSession as executeActualNativeSession, type ControlPlanePort, type NativeExecutionInputV1, type PrpEvent } from "@paperclipai/paperclip-runner"; +import { createIsolatedTestHome } from "../../__tests__/helpers/isolated-test-home.js"; +import { registerNativeRunnerCheckpoint } from "./native-runner-checkpoint.js"; +import { DurablePrpControlPlane } from "../../vendor/paperclip-runner/index.js"; import { startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; const provider = vi.hoisted(() => ({ execute: vi.fn() })); vi.mock("../../vendor/paperclip-runner/index.js", async importOriginal => ({ ...(await importOriginal()), executeNativeSession: provider.execute, })); +import { createPostgresRunDispatchAdapter } from "../../modules/run-dispatch/adapters/postgres.js"; +import * as githubLaunchers from "../heartbeat-github-launchers.js"; +import * as executionTargets from "../environment-execution-target.js"; +import * as attachmentStaging from "./native-runner-file-handoff.js"; +import * as workspaceSync from "./native-workspace-sync.js"; +import { environmentRuntimeService, classifyEnvironmentCapabilities } from "../environment-runtime.js"; +import { remoteTerminationReceipt } from "../remote-execution-termination.js"; +import { heartbeatService } from "../heartbeat.js"; +import { reserveRunBudget } from "../budget-reservations.js"; +import { requestedRunCancellation } from "../run-cancellation.js"; +import { readClaimedNativeStopRecovery } from "./native-cancellation-request.js"; +import { currentNativeControllerIdentity } from "./native-restart-recovery.js"; import { executePaperclipNativeSession } from "./native-session-executor.js"; import { buildNativeCompletionContract } from "./completion-contracts.js"; import { prepareNativeHeartbeatRun } from "./prepare-native-run.js"; @@ -20,19 +36,24 @@ import { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js"; describe("native Stop accounting after the last usage receipt", () => { let temporary: Awaited>; let db: ReturnType; + let home: ReturnType; beforeAll(async () => { + home = createIsolatedTestHome(); + vi.stubEnv("PAPERCLIP_HOME", home.home); temporary = await startEmbeddedPostgresTestDatabase("native-stop-accounting-"); db = createDb(temporary.connectionString); + await db.insert(authUsers).values({ id: "test-board", name: "Test Board", email: "board@example.test", emailVerified: true, createdAt: new Date(), updatedAt: new Date() }); }, 60_000); - afterAll(async () => { await temporary?.cleanup(); }); + afterAll(async () => { await temporary?.cleanup(); home?.cleanup(); vi.unstubAllEnvs(); }); - it.each([true, false])("wakes settlement after Stop closure while preserving incomplete receipt debt (complete=%s)", async complete => { + async function fixture() { const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(), runId = randomUUID(); await db.insert(companies).values({ id: companyId, name: "Stopped accounting", issuePrefix: randomUUID().slice(0, 8) }); - await db.insert(agents).values({ id: agentId, companyId, name: "Stopped worker", status: "active", adapterType: "paperclip_runner" }); - const [issue] = await db.insert(issues).values({ id: issueId, companyId, title: "Resume after Stop", status: "in_progress", assigneeAgentId: agentId }).returning(); + await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: "test-board", status: "active", membershipRole: "owner" }); + await db.insert(agents).values({ id: agentId, companyId, name: "Stopped worker", status: "active", adapterType: "paperclip_runner", adapterConfig: { provider: "codex", model: "test-model", cwd: home.home } }); + const [issue] = await db.insert(issues).values({ id: issueId, companyId, title: "Resume after Stop", responsibleUserId: "test-board", status: "in_progress", assigneeAgentId: agentId }).returning(); const [run] = await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, status: "running", runtimeMode: "native", - nativeIssueId: issueId, invocationSource: "on_demand", costAccountingPending: true }).returning(); + nativeIssueId: issueId, issueId, scopeKind: "issue", contextSnapshot: { issueId }, invocationSource: "on_demand", costAccountingPending: true }).returning(); await db.update(issues).set({ executionRunId: runId }).where(eq(issues.id, issueId)); const native = await prepareNativeHeartbeatRun({ db, run, issue, environmentLeaseId: randomUUID() }); const [contract] = await db.select().from(completionContracts).where(eq(completionContracts.issueId, issueId)); @@ -41,7 +62,7 @@ describe("native Stop accounting after the last usage receipt", () => { schema: "paperclip.native-execution-input.v1", binding: { companyId, runId, issueId, agentId, executionWorkspaceId: runId }, provider: { kind: "codex", model: "test-model" }, task: { identifier: issueId, title: issue.title, description: null, prompt: "Keep the task context.", workMode: "standard" }, - workspace: { cwd: "/tmp", repoUrl: null, repoRef: null, branchName: null }, + workspace: { cwd: home.home, repoUrl: null, repoRef: null, branchName: null }, session: { normalizedSessionId: native.normalizedSessionId, driverKind: "codex_app_server", protocolVersion: 1, lifecyclePolicy: { mode: "per_turn", idleTimeoutMs: null } }, completionContract: { id: contract.id, sha256: contract.canonicalSha256, schemaVersion: "paperclip.completion-contract.v1", @@ -49,6 +70,11 @@ describe("native Stop accounting after the last usage receipt", () => { interactionResponses: [], credentialBindings: [], }; await db.update(heartbeatRuns).set({ runnerProfileJson: { nativeExecutionInput: execution } }).where(eq(heartbeatRuns.id, runId)); + return { companyId, agentId, issueId, runId, native, execution }; + } + + it.each([true, false])("wakes settlement after Stop closure while preserving incomplete receipt debt (complete=%s)", async complete => { + const { companyId, agentId, issueId, runId, native, execution } = await fixture(); const usage = await createRunUsageRecorder(db, { companyId, runId, adapterType: "paperclip_runner" }); const reconcile = vi.fn(createCostAccountingReconciler(db)); const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => { throw error; } }); @@ -109,4 +135,179 @@ describe("native Stop accounting after the last usage receipt", () => { expect(coordinator.nextWakeAt()).toBeNull(); } finally { worker.stop(); await coordinator.stop(); } }, 30_000); + it.each([{ actorType: "user", entry: "dispatch" }, { actorType: "system", entry: "dispatch" }, { actorType: "user", entry: "heartbeat" }] as const)("recovers the combined Stop fence through existing-session cleanup without new paid work ($actorType/$entry)", async ({ actorType, entry }) => { + const { companyId, agentId, issueId, runId, native, execution } = await fixture(); + const hold = await reserveRunBudget(db, companyId, runId, null); + const controller = await currentNativeControllerIdentity(), owner = `cleanup:${randomUUID()}`; + const intentAuditId = randomUUID(), requestedAt = new Date().toISOString(); + await db.insert(activityLog).values({ id: intentAuditId, companyId, actorType: "user", actorId: "test-board", + action: "native.cancellation_intent_recorded", entityType: "heartbeat_run", entityId: runId, runId, agentId }); + await db.update(nativeRunFinalizations).set({ controllerBootId: controller.bootId, controllerPid: controller.pid, + controllerGeneration: 2, attempt: 1, leaseOwner: owner, leaseExpiresAt: new Date(Date.now() + 60_000), + recoveryState: "awaiting_runner_reattach", recoveryHistory: [{ disposition: "reattach_remote_runner", controllerGeneration: 2, providerAttempt: 1 }] }) + .where(eq(nativeRunFinalizations.runId, runId)); + if (entry === "heartbeat") { + const [environment] = await db.insert(environments).values({ name: `Stop sandbox ${runId}`, driver: "sandbox", config: { provider: "daytona" } }).returning(); + await db.update(agents).set({ defaultEnvironmentId: environment.id }).where(eq(agents.id, agentId)); + await db.insert(environmentLeases).values({ id: native.environmentLeaseId, environmentId: environment.id, companyId, + issueId, heartbeatRunId: runId, status: "active", leasePolicy: "ephemeral", provider: "daytona", providerLeaseId: "existing-sandbox", + metadata: { driver: "sandbox", agentId, remoteCwd: "/workspace" } }); + const authority = new DurablePrpControlPlane({ stateDirectory: resolve(home.home, runId, "control-plane"), + identity: { runnerInstanceId: native.runnerInstanceId, environmentLeaseId: native.environmentLeaseId, + runId, normalizedSessionId: native.normalizedSessionId, turnId: native.turnId, itemId: native.itemId }, + expectedRunnerVersion: "stop-recovery-fixture", expectedRunnerDigest: `sha256:${"a".repeat(64)}` }); + try { await registerNativeRunnerCheckpoint({ db, binding: { companyId, issueId, runId }, bootId: controller.bootId, authority }); } + finally { await authority.stop(); } + await db.update(heartbeatRuns).set({ runnerProfileJson: sql`${heartbeatRuns.runnerProfileJson} || ${JSON.stringify({ + nativeWorkspaceSync: { schema: "paperclip.native-workspace-sync/v1", state: "prepared", descriptorSha256: "a".repeat(64), + baselineSha256: "b".repeat(64), workspaceId: runId, leaseId: native.environmentLeaseId, + providerLeaseId: "existing-sandbox", remoteCwd: "/workspace" }, + sessionCheckpoint: { backendKind: "mock", sessionId: native.normalizedSessionId, + identity: { companyId, agentId, issueId, runId, sessionId: native.normalizedSessionId }, + providerSessionId: "surviving-provider", workingDirectory: home.home, cursor: null, activeTurnId: native.turnId, + semanticResult: null, terminal: null, terminalTurns: [], pendingRuntimeRequests: [], lineage: [] }, + })}::jsonb` }).where(eq(heartbeatRuns.id, runId)); + } + await db.update(heartbeatRuns).set({ resultJson: { + cancelledByActorType: actorType, ...(actorType === "user" ? { cancelledByUserId: "test-board" } : {}), + cancellation: requestedRunCancellation({ cancelledByActorType: actorType }, "Operator requested Stop"), + startupCancellation: { beforeNativeSelection: false, errorCode: "cancelled", requestedAt }, + executionCancellation: { state: "requested", requestedAt }, + nativeCancellation: { schema: "paperclip.native-cancellation.v1", companyId, issueId, runId, scope: "run", + reasonCode: "cancellation_run_only", intentId: `native-cancellation:${randomUUID()}`, intentAuditId, dispatchState: "pending", dispatched: false, + effects: ["release_run_resources"] }, + }, usageJson: { accountingReceiptReady: false, accountingReceiptSequence: 0 } }).where(eq(heartbeatRuns.id, runId)); + await db.update(companies).set({ status: "paused", pauseReason: "budget" }).where(eq(companies.id, companyId)); + if (entry === "heartbeat") { + await db.update(heartbeatRuns).set({ status: "interrupted", finishedAt: new Date(), errorCode: "lease_released_before_terminal" }).where(eq(heartbeatRuns.id, runId)); + await db.update(issues).set({ executionRunId: null, checkoutRunId: null }).where(eq(issues.id, issueId)); + await db.update(nativeRunFinalizations).set({ controllerBootId: randomUUID(), leaseExpiresAt: new Date(Date.now() - 1) }) + .where(eq(nativeRunFinalizations.runId, runId)); + } + const reservation = entry === "dispatch" ? await reserveRunBudget(db, companyId, runId, null, {}, owner) : hold; + if (entry === "dispatch") expect(reservation).toMatchObject({ id: hold.id, reused: true }); + const usage = await createRunUsageRecorder(db, { companyId, runId, adapterType: "paperclip_runner" }); + let completeCleanup!: () => void; + const cleanup = new Promise(resolve => { completeCleanup = resolve; }); + const startTurn = vi.fn(() => { throw new Error("Stop cleanup must not start a provider turn"); }), cancel = vi.fn(() => ({ cleanup })); + const identity = { companyId, agentId, issueId, runId, sessionId: native.normalizedSessionId }; + const backend = { descriptor: async () => ({ kind: "mock" as const, name: "existing-session", version: "1", + capabilities: { resume: true, typedEvents: true, steering: false, interruption: true, structuredResult: true } }), + openSession: vi.fn(async () => ({ identity: () => identity, cancel, startTurn, close: async () => undefined, + snapshot: async () => ({ identity, sessionId: identity.sessionId, backendKind: "mock" as const }) })) }; + Object.assign(backend, { recoverSession: async () => ({ recovered: true, session: await backend.openSession() }) }); + provider.execute.mockReset().mockImplementationOnce(executeActualNativeSession); + const dispatch = vi.fn(() => executePaperclipNativeSession({ db, execution, runnerInstanceId: native.runnerInstanceId, + backend: backend as never, leaseOwner: owner, restartRecovery: { kind: "reattach_remote_runner", runId, leaseOwner: owner, + controllerGeneration: 2, providerAttempt: 1, restartKind: "hard", recoveryRequestId: null, + remote: { providerLeaseId: "existing-sandbox", remoteCwd: "/workspace" } }, + onUsage: receipt => usage.capture(receipt).then(() => undefined) })); + if (entry === "heartbeat") { + let reachedSelection = false; + const factory = vi.fn(() => backend as never); + const launchers = vi.spyOn(githubLaunchers, "prepareHeartbeatGitHubLaunchers").mockResolvedValue({ env: {}, cleanupLocation: null }); + const transport = vi.spyOn(executionTargets, "resolveEnvironmentExecutionTransport").mockResolvedValue(null); + const remoteIdentity = { runId, normalizedSessionId: native.normalizedSessionId, runnerInstanceId: native.runnerInstanceId, + environmentLeaseId: native.environmentLeaseId, turnId: native.turnId, itemId: native.itemId }; + const target = vi.spyOn(executionTargets, "resolveEnvironmentExecutionTarget").mockResolvedValue({ + kind: "remote", transport: "sandbox", providerKey: "daytona", remoteCwd: "/workspace", leaseId: native.environmentLeaseId, + sandboxLeaseAcquisition: { outcome: "resumed", providerLeaseId: "existing-sandbox" }, runner: { execute: async request => ({ + exitCode: 0, signal: null, pid: 123, startedAt: new Date().toISOString(), timedOut: false, stderr: "", stdout: request.args?.includes("paperclip-git-context") + ? "\0PAPERCLIP_GIT_CONTEXT_V1\0\0PAPERCLIP_GIT_CONTEXT_END\0" + : request.args?.[1]?.includes("base64") ? Buffer.from(JSON.stringify({ ...remoteIdentity, lifecycle: "running" })).toString("base64") + : request.args?.[1]?.includes("cat --") ? `nonce\n123\n2026-09-19T10:00:00.000Z\n${native.runnerInstanceId}\nlinux:abcd-1234:100\n` : "", + }) }, + }); + const localCwd = vi.spyOn(workspaceSync, "readNativeWorkspaceSyncLocalCwd").mockResolvedValue(home.home); + const sync = vi.spyOn(workspaceSync, "prepareNativeWorkspaceSync").mockResolvedValue(null); + const stage = vi.spyOn(attachmentStaging, "stageNativeRunnerWakeAttachments").mockRejectedValue(new Error("Stop cleanup must not stage attachments")); + const acquire = vi.fn(async () => { throw new Error("Stop cleanup cannot acquire a new sandbox"); }); + const runtime = environmentRuntimeService(db, { drivers: [{ driver: "sandbox", acquireRunLease: acquire, + resolveCapabilities: async () => classifyEnvironmentCapabilities({}), + realizeWorkspace: async () => ({ cwd: home.home }), + retryPendingSandboxTeardown: async ({ lease }) => { + await cleanup; + expect(cancel).toHaveBeenCalledOnce(); + return { providerLeaseId: lease.providerLeaseId, state: "stopped" }; + }, + releaseRunLease: async ({ lease }) => { + expect(cancel).toHaveBeenCalledOnce(); + const [released] = await db.update(environmentLeases).set({ status: "released", releasedAt: new Date(), cleanupStatus: "success", + metadata: { ...lease.metadata, remoteExecutionTermination: remoteTerminationReceipt(lease, { providerLeaseId: lease.providerLeaseId, state: "stopped" }) }, + }).where(eq(environmentLeases.id, lease.id)).returning(); + return { ...released, status: "released", leasePolicy: "ephemeral", cleanupStatus: "success" }; + } }] }); + const heartbeat = heartbeatService(db, { nativeSessionBackendFactory: factory, environmentRuntime: runtime, + beforeNativeRuntimeSelection: async () => { + const [current] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)); + const [owner] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, runId)); + const holds = await db.select().from(budgetReservations).where(eq(budgetReservations.runId, runId)); + expect(await readClaimedNativeStopRecovery(db, current, owner.leaseOwner ?? undefined)).toBe(true); + expect(holds).toMatchObject([{ id: hold.id, state: "held" }]); + reachedSelection = true; + } }); + try { + const recovery = await heartbeat.recoverNativeRunsAfterRestart(new Date(), [runId]); + expect(recovery.claims).toMatchObject([{ kind: "reattach_remote_runner", runId, providerAttempt: 1 }]); + await expect.poll(() => reachedSelection, { timeout: 5_000 }).toBe(true); + await expect.poll(() => cancel.mock.calls.length, { timeout: 5_000 }).toBe(1); + expect(factory).toHaveBeenCalledOnce(); + expect(factory).toHaveBeenCalledWith(expect.objectContaining({ binding: expect.objectContaining({ runId }) }), { recoveryPurpose: "stop_cleanup" }); + expect(stage).not.toHaveBeenCalled(); + expect((await db.select().from(issues).where(eq(issues.id, issueId)))[0].executionRunId).toBeNull(); + expect((await heartbeat.recoverNativeRunsAfterRestart(new Date(), [runId])).claims).toHaveLength(0); + } finally { + completeCleanup(); + await heartbeat.drainActiveRunExecutions(); + transport.mockRestore(); target.mockRestore(); localCwd.mockRestore(); sync.mockRestore(); stage.mockRestore(); launchers.mockRestore(); + } + expect(acquire).not.toHaveBeenCalled(); + expect((await heartbeat.getRun(runId))?.status).toBe("cancelled"); + } else { + const gate = await createPostgresRunDispatchAdapter(db).dispatchResolvedInteractionIfCurrent({ companyId, runId, + reservationId: reservation.id, recoveryLeaseOwner: owner, expectedStatus: "running", now: new Date(), dispatch }); + expect(gate.dispatched).toBe(true); + if (!gate.dispatched) throw new Error("Existing Stop cleanup was not admitted"); + try { + await expect.poll(() => cancel.mock.calls.length).toBe(1); + const duplicate = await createPostgresRunDispatchAdapter(db).dispatchResolvedInteractionIfCurrent({ companyId, runId, + reservationId: reservation.id, recoveryLeaseOwner: owner, expectedStatus: "running", now: new Date(), dispatch }); + expect(duplicate.dispatched).toBe(false); + expect(dispatch).toHaveBeenCalledOnce(); + const [pending] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)); + expect(pending.resultJson?.nativeCancellation).toMatchObject({ dispatchState: "pending", dispatched: false }); + expect(pending.costAccountingPending).toBe(true); + expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, runId)))[0].phase).toBe("observed"); + } finally { completeCleanup(); } + await expect(gate.resultPromise).rejects.toThrow("native_cancellation_pending_recovery"); + } + expect(cancel).toHaveBeenCalledOnce(); + expect(startTurn).not.toHaveBeenCalled(); + const [stopped] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)); + expect(stopped.resultJson?.nativeCancellation).toMatchObject({ dispatchState: "acknowledged", dispatched: true }); + expect(stopped.resultJson?.nativeCancellation).toHaveProperty("acknowledgementAuditId", expect.any(String)); + expect(stopped.costAccountingPending).toBe(true); + expect(await db.select().from(costEvents).where(eq(costEvents.heartbeatRunId, runId))).toHaveLength(0); + expect(await db.select().from(budgetReservations).where(eq(budgetReservations.runId, runId))).toMatchObject([{ id: hold.id, state: "held" }]); + expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, runId)))[0]) + .toMatchObject({ phase: "terminal_failure", failureCode: "native_retry_cancelled", leaseOwner: null }); + const cancelled = await heartbeatService(db).cancelRun(runId, "Recovered durable native run cancellation"); + expect(cancelled?.status).toBe("cancelled"); + const reconcile = createCostAccountingReconciler(db); + await reconcile(); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)))[0].costAccountingPending).toBe(true); + expect(await db.select().from(costEvents).where(eq(costEvents.heartbeatRunId, runId))).toHaveLength(0); + const finalUsage = entry === "heartbeat" ? await createRunUsageRecorder(db, { companyId, runId, adapterType: "paperclip_runner" }) : usage; + await finalUsage.capture({ complete: true, provider: "openai", model: "test-model", billingType: "metered_api", + usageBasis: "per_run", usage: { inputTokens: 120, outputTokens: 10 }, costUsd: 0.25 }); + await reconcile(); + await reconcile(); + const [settled] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)); + expect(settled).toMatchObject({ status: "cancelled", costAccountingPending: false }); + expect(settled.costAccountedAt).toBeInstanceOf(Date); + expect(await db.select().from(costEvents).where(eq(costEvents.heartbeatRunId, runId))).toMatchObject([{ costCents: 25 }]); + expect(await db.select().from(budgetReservations).where(eq(budgetReservations.runId, runId))).toMatchObject([{ id: hold.id, state: "settled" }]); + expect((await db.select().from(companies).where(eq(companies.id, companyId)))[0]).toMatchObject({ status: "paused", pauseReason: "budget" }); + }, 30_000); + });