From 263f181fed709ae734f6f29f8d941317cade5e4b Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Fri, 4 Sep 2026 17:54:44 -0500 Subject: [PATCH] fix(runner): complete live hot restart adoption (#12852) ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Paperclip Runner owns durable provider sessions and streams their work to the control plane. > - Pull request #12845 added native restart recovery for live and dead local runners. > - A real browser test found three live-adoption gaps after that pull request merged. > - Lazy runner process ownership was not always stored before restart. > - The old controller did not release its PRP authority without closing the provider turn. > - Reconnect events could arrive before the active provider turn was restored. > - This pull request closes those gaps and proves the same turn completes after a UI hot restart. ## Linked Issues or Issue Description Refs #12845 Related search results: #12646 covers indeterminate command results after a runner restart. It does not cover controller adoption or active-turn rebinding. No open duplicate pull request was found. ## What Changed - Store lazy runnerd process ownership after provider session creation, read, and resume. - Detach native PRP controller authority during coordinated hot shutdown. Keep the live provider turn running. - Restore the exact checkpointed provider session when bounded PRP identity events have been compacted. - Restore the active provider turn before reconnect events are replayed. This prevents `turn_binding_mismatch`. - Keep exact live ownership by the current controller out of generic orphan recovery. - Add driver, transport, and server regression tests for these paths. ## Verification - Ran 12 Codex driver lifecycle tests. - Ran 53 runnerd transport tests. - Ran 143 recovery and orphan-reaper server tests. - Ran all 8 real-process restart recovery scenarios. - Ran all 96 existing runner E2E unit tests. - Ran runner TypeScript typecheck. - Ran server TypeScript typecheck. - Ran the migration replay test and migration safety checks. - Tested the board UI on an isolated local instance. A real local Codex-backed turn entered a 120-second terminal wait. The UI `Restart now` action replaced the server and kept the same runner PID, process start time, run ID, native session ID, runner ID, provider session ID, and active turn. The original turn then completed. - Confirmed one heartbeat run, no retry row, one result, one proposed-result event, one terminal event, no protocol errors, no active recovery state, and no surviving runner or provider process. ## Risks - A live runner can continue provider work while no server owns the control route. Recovery fails closed when the process fingerprint or durable identity is ambiguous. - Provider identity can be restored from the database only for an exact verified adoption claim. An authenticated live `session.snapshot` validates that identity before the driver can resume. - The new detach path applies only to native sessions that expose restart detachment. Other adapters keep their existing shutdown behavior. - This follow-up does not change the database migration or `package.json`. The migration in #12845 remains replay-safe through `ADD COLUMN IF NOT EXISTS` and its embedded-Postgres idempotence test. The dedicated real-process command remains in `doc/DEVELOPING.md`. ## Model Used - OpenAI Codex based on GPT-5. The exact serving build and context-window size are not exposed. The run used extended reasoning, repository tools, shell execution, and in-app browser automation. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --- .../runner-core/src/acpx_provider_backend.rs | 4 + .../src/managed_provider_backend.rs | 3 + .../runner-core/src/provider_backend.rs | 3 + .../runner-core/tests/codex_provider.rs | 7 + .../src/backends/codex-native-backend.ts | 121 +++++----- .../src/backends/harness-driver-backend.ts | 5 + .../src/backends/native-backend-factory.ts | 8 + .../src/contracts/harness-driver.ts | 57 ++++- .../src/contracts/native-session-backend.ts | 22 +- .../src/drivers/codex/app-server-transport.ts | 6 +- .../codex/codex-app-server-driver-impl.ts | 214 ++++++++++-------- .../codex-app-server-driver.lifecycle.test.ts | 197 +++++++++++++--- .../src/drivers/codex/codex-driver-types.ts | 7 + .../drivers/codex/codex-harness-session.ts | 82 ++++--- .../src/live/runnerd-codex-transport.test.ts | 79 ++++++- .../src/live/runnerd-codex-transport.ts | 137 ++++++++++- .../heartbeat-process-recovery.test.ts | 54 ++++- server/src/services/heartbeat.ts | 125 ++++++---- .../native-session-executor.test.ts | 140 +++++++++++- .../native-runtime/native-session-executor.ts | 90 ++++++-- 20 files changed, 1040 insertions(+), 321 deletions(-) diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_backend.rs b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_backend.rs index 97ced9b875..b14a2098c0 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_backend.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_backend.rs @@ -1112,7 +1112,11 @@ impl AcpxCommandExecutor { "status": state.lifecycle, "provider": "acpx", "driver": "acpx_runtime", + "driverSessionId": state.identity.as_ref().map(|value| value.acpx_record_id.as_str()), "providerSessionId": state.identity.as_ref().map(|value| value.acpx_record_id.as_str()), + "sessionId": state.identity.as_ref().map(|value| value.agent_session_id.as_str()), + "providerAccountSessionId": state.identity.as_ref().map(|value| value.agent_session_id.as_str()), + "providerIdentity": state.identity, "activeProviderTurnId": state.active_turn_id, }))) } diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/managed_provider_backend.rs b/packages/paperclip-runner/runner/crates/runner-core/src/managed_provider_backend.rs index 5957b4c97d..5456fc9824 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/managed_provider_backend.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/managed_provider_backend.rs @@ -1305,7 +1305,10 @@ impl ManagedProviderCommandExecutor { "status": state.lifecycle, "provider": state.descriptor.provider_label(), "driver": state.descriptor.driver(), + "driverSessionId": state.provider_session_id, "providerSessionId": state.provider_session_id, + "sessionId": state.provider_session_id, + "providerAccountSessionId": state.provider_session_id, "activeProviderTurnId": state.active_turn_id, "durableEventCursor": state.durable_event_cursor, }))) diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/provider_backend.rs b/packages/paperclip-runner/runner/crates/runner-core/src/provider_backend.rs index d2b634c015..f11fe402df 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/provider_backend.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/provider_backend.rs @@ -2673,7 +2673,10 @@ impl CodexCommandExecutor { "status": state.lifecycle, "provider": state.config.provider, "driver": state.config.driver, + "driverSessionId": state.thread_id, "providerSessionId": state.thread_id, + "sessionId": state.provider_session_id, + "providerAccountSessionId": state.provider_session_id, "activeProviderTurnId": state.active_provider_turn_id, "cwd": state.config.cwd, }))) diff --git a/packages/paperclip-runner/runner/crates/runner-core/tests/codex_provider.rs b/packages/paperclip-runner/runner/crates/runner-core/tests/codex_provider.rs index 27c6d57fc9..3b653b9902 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/tests/codex_provider.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/tests/codex_provider.rs @@ -3393,6 +3393,13 @@ fn durable_backend_resumes_the_active_thread_without_restarting_the_turn() { .execute(&command("snapshot", 4, "session.snapshot", json!({}))) .expect("restore provider session"); assert_eq!(snapshot.result["status"], "turn_active"); + assert_eq!(snapshot.result["driverSessionId"], "codex-thread-1"); + assert_eq!(snapshot.result["providerSessionId"], "codex-thread-1"); + assert_eq!(snapshot.result["sessionId"], "codex-account-session"); + assert_eq!( + snapshot.result["providerAccountSessionId"], + "codex-account-session" + ); assert_eq!(snapshot.result["activeProviderTurnId"], "provider-turn-1"); assert_eq!(call_count(&directory, "turn/start"), 1); assert_eq!(call_count(&directory, "thread/resume"), 1); diff --git a/packages/paperclip-runner/src/backends/codex-native-backend.ts b/packages/paperclip-runner/src/backends/codex-native-backend.ts index 5a586d62c2..bde981c261 100644 --- a/packages/paperclip-runner/src/backends/codex-native-backend.ts +++ b/packages/paperclip-runner/src/backends/codex-native-backend.ts @@ -1,5 +1,6 @@ import { createCodexTaskEnvelope } from "../contracts/codex.js"; import type { NativeExecutionInput } from "../contracts/native-execution.js"; +import type { PersistedHarnessSession } from "../contracts/harness-driver.js"; import type { NativeSessionBackend, PersistedNativeSession, @@ -8,7 +9,10 @@ import type { CodexAppServerTransport } from "../drivers/codex/app-server-transp import { CodexAppServerDriver } from "../drivers/codex/codex-app-server-driver.js"; import type { CodexWorkingDirectoryAuthority } from "../drivers/codex/codex-boundaries.js"; import { HarnessDriverBackend } from "./harness-driver-backend.js"; -import { nativeSystemInstructions, nativeTaskConstraints } from "./runtime-context.js"; +import { + nativeSystemInstructions, + nativeTaskConstraints, +} from "./runtime-context.js"; export interface CodexNativeSessionBackendOptions { /** Effective provider environment, including the assigned workspace boundary. */ @@ -23,6 +27,13 @@ export interface CodexNativeSessionBackendOptions { }) => Promise; transportFactory?: (context?: { providerRecoveryPolicy?: PersistedNativeSession["providerRecoveryPolicy"]; + persistedSession?: Pick< + PersistedHarnessSession, + | "driverSessionId" + | "providerSessionId" + | "providerIdentity" + | "activeTurnId" + >; }) => CodexAppServerTransport; dynamicTools?: readonly Readonly>[]; dynamicToolHandler?: (call: { @@ -34,9 +45,7 @@ export interface CodexNativeSessionBackendOptions { }) => Promise; } -function transportDriverIdentity( - input: NativeExecutionInput, -): { +function transportDriverIdentity(input: NativeExecutionInput): { kind: | "codex_app_server" | "opencode_server" @@ -108,66 +117,68 @@ function createTransportBackedNativeSessionBackend( input.provider.kind === "opencode" || input.provider.kind === "acpx"; if ( - input.provider.kind === "codex" - && input.provider.approvalPolicy !== undefined - && input.provider.approvalPolicy !== "never" + input.provider.kind === "codex" && + input.provider.approvalPolicy !== undefined && + input.provider.approvalPolicy !== "never" ) { throw new Error( "paperclip_runner_codex_permission_mode_unqualified: set codexPermissionMode to never before starting or recovering this native run", ); } - return new HarnessDriverBackend(new CodexAppServerDriver({ - ...(input.provider.model ? { model: input.provider.model } : {}), - // Runnerd owns provider permissions for non-Codex facades. Their - // Codex-compatible surface must never open a second approval channel. - approvalPolicy: - input.provider.kind === "codex" - ? input.provider.approvalPolicy ?? "never" - : "never", - baseInstructions: nativeSystemInstructions(input), - includeSkillInstructions: isCodex && "runtimeContext" in input, - requestedCollaborationMode: - supportsCollaborativePlanning && "executionMode" in input - ? input.executionMode - : "default", - taskEnvelope: createCodexTaskEnvelope({ - objective: input.completionContract.contract.objective, - contractRevision: input.completionContract.contract.revision, - criteria: input.completionContract.contract.criteria, - constraints: [ - "Work only inside the supplied working directory.", - ...(supportsCollaborativePlanning && + return new HarnessDriverBackend( + new CodexAppServerDriver({ + ...(input.provider.model ? { model: input.provider.model } : {}), + // Runnerd owns provider permissions for non-Codex facades. Their + // Codex-compatible surface must never open a second approval channel. + approvalPolicy: + input.provider.kind === "codex" + ? (input.provider.approvalPolicy ?? "never") + : "never", + baseInstructions: nativeSystemInstructions(input), + includeSkillInstructions: isCodex && "runtimeContext" in input, + requestedCollaborationMode: + supportsCollaborativePlanning && "executionMode" in input + ? input.executionMode + : "default", + taskEnvelope: createCodexTaskEnvelope({ + objective: input.completionContract.contract.objective, + contractRevision: input.completionContract.contract.revision, + criteria: input.completionContract.contract.criteria, + constraints: [ + "Work only inside the supplied working directory.", + ...(supportsCollaborativePlanning && "executionMode" in input && input.executionMode === "plan" - ? [ - "Use native plan collaboration mode and do not modify workspace files.", - "Treat the supplied Paperclip planning context as the canonical pinned base revision.", - "Complete one structured provider plan item; Paperclip will synchronize it after completion.", - "Keep the final response to a short synchronization summary instead of repeating the full plan.", - ] - : []), - ...nativeTaskConstraints(input), - "Return one semantic completion result.", - ], + ? [ + "Use native plan collaboration mode and do not modify workspace files.", + "Treat the supplied Paperclip planning context as the canonical pinned base revision.", + "Complete one structured provider plan item; Paperclip will synchronize it after completion.", + "Keep the final response to a short synchronization summary instead of repeating the full plan.", + ] + : []), + ...nativeTaskConstraints(input), + "Return one semantic completion result.", + ], + }), + runnerInstanceId: + options.runnerInstanceId ?? `paperclip-native-${input.binding.runId}`, + onSpawn: options.onSpawn, + transportFactory: options.transportFactory, + dynamicTools: options.dynamicTools, + dynamicToolHandler: options.dynamicToolHandler, + environment: options.environment, + workingDirectoryAuthority: options.workingDirectoryAuthority, + driverIdentity, + capabilities: isCodex + ? {} + : { steering: false, goals: false, threadLineage: false }, + collaborationModes: supportsCollaborativePlanning + ? ["default", "plan"] + : ["default"], + requireProviderSessionIdentity: options.transportFactory !== undefined, }), - runnerInstanceId: - options.runnerInstanceId ?? `paperclip-native-${input.binding.runId}`, - onSpawn: options.onSpawn, - transportFactory: options.transportFactory, - dynamicTools: options.dynamicTools, - dynamicToolHandler: options.dynamicToolHandler, - environment: options.environment, - workingDirectoryAuthority: options.workingDirectoryAuthority, - driverIdentity, - capabilities: isCodex - ? {} - : { steering: false, goals: false, threadLineage: false }, - collaborationModes: supportsCollaborativePlanning - ? ["default", "plan"] - : ["default"], - requireProviderSessionIdentity: options.transportFactory !== undefined, - })); + ); } /** diff --git a/packages/paperclip-runner/src/backends/harness-driver-backend.ts b/packages/paperclip-runner/src/backends/harness-driver-backend.ts index 4d73c5b8ff..a31826c672 100644 --- a/packages/paperclip-runner/src/backends/harness-driver-backend.ts +++ b/packages/paperclip-runner/src/backends/harness-driver-backend.ts @@ -438,6 +438,11 @@ class HarnessNativeSession implements NativeSession { this.#explicitlyCancelled = false; } + async detachControllerForRestart(): Promise { + if (this.#session.detachControllerForRestart === undefined) return; + await this.#session.detachControllerForRestart(); + } + async *events(): AsyncIterable { let sourceInstanceId: string | null = null; let lastSourceSequence = 0; diff --git a/packages/paperclip-runner/src/backends/native-backend-factory.ts b/packages/paperclip-runner/src/backends/native-backend-factory.ts index 8f18951044..6395c7a985 100644 --- a/packages/paperclip-runner/src/backends/native-backend-factory.ts +++ b/packages/paperclip-runner/src/backends/native-backend-factory.ts @@ -1,4 +1,5 @@ import type { NativeExecutionInput } from "../contracts/native-execution.js"; +import type { PersistedHarnessSession } from "../contracts/harness-driver.js"; import type { NativeSessionBackend, PersistedNativeSession, @@ -21,6 +22,13 @@ export interface NativeBackendFactoryOptions extends Omit< > { codexTransportFactory?: (context?: { providerRecoveryPolicy?: PersistedNativeSession["providerRecoveryPolicy"]; + persistedSession?: Pick< + PersistedHarnessSession, + | "driverSessionId" + | "providerSessionId" + | "providerIdentity" + | "activeTurnId" + >; }) => CodexAppServerTransport; acpxRuntimeDirectory?: string; acpxEnvironment?: NodeJS.ProcessEnv; diff --git a/packages/paperclip-runner/src/contracts/harness-driver.ts b/packages/paperclip-runner/src/contracts/harness-driver.ts index a3c7b6792c..9e0957d996 100644 --- a/packages/paperclip-runner/src/contracts/harness-driver.ts +++ b/packages/paperclip-runner/src/contracts/harness-driver.ts @@ -135,9 +135,11 @@ export type HarnessRuntimeRequestResolution = response: PaperclipQuestionResponse; }; -export type HarnessRuntimeRequestAction = HarnessRuntimeRequestResolution["action"]; +export type HarnessRuntimeRequestAction = + HarnessRuntimeRequestResolution["action"]; -export type HarnessRuntimeRequestHandoffResult = "handed_off" | "already_settled"; +export type HarnessRuntimeRequestHandoffResult = + "handed_off" | "already_settled"; /** * A runtime-input handoff commits its durable state transition before the @@ -174,7 +176,9 @@ function plainRecord(value: unknown): Record | null { : null; } -function parseAnswers(value: unknown): Record | null { +function parseAnswers( + value: unknown, +): Record | null { const fields = plainRecord(value); if (fields === null || Object.keys(fields).length === 0) return null; const parsed: Record = {}; @@ -210,7 +214,12 @@ export function parseHarnessRuntimeRequestResolution( ); } const action = rawAction as HarnessRuntimeRequestAction; - if (action !== "submit" && ("answers" in candidate || "content" in candidate || "response" in candidate)) { + if ( + action !== "submit" && + ("answers" in candidate || + "content" in candidate || + "response" in candidate) + ) { throw new HarnessRuntimeRequestResolutionError( requestKind, `${action} does not carry submitted form data`, @@ -239,7 +248,10 @@ export function parseHarnessRuntimeRequestResolution( try { return { action, - response: parsePaperclipQuestionResponse(questionSet, candidate.response), + response: parsePaperclipQuestionResponse( + questionSet, + candidate.response, + ), }; } catch (error) { throw new HarnessRuntimeRequestResolutionError( @@ -344,10 +356,14 @@ export function harnessRuntimeRequestOutcome( itemId: request.itemId, ...(outcome.action ? { action: outcome.action } : {}), ...(outcome.reason ? { reason: outcome.reason } : {}), - ...(outcome.response ? { response: structuredClone(outcome.response) } : {}), + ...(outcome.response + ? { response: structuredClone(outcome.response) } + : {}), ...(request.input ? { - ...(request.origin?.adapter ? { adapter: request.origin.adapter } : {}), + ...(request.origin?.adapter + ? { adapter: request.origin.adapter } + : {}), requestType: "input" as const, } : {}), @@ -385,7 +401,13 @@ export function harnessRuntimeInputExpiredOutcome( export interface HarnessThreadGoal { threadId: string; objective: string; - status: "active" | "paused" | "blocked" | "usageLimited" | "budgetLimited" | "complete"; + status: + | "active" + | "paused" + | "blocked" + | "usageLimited" + | "budgetLimited" + | "complete"; tokenBudget: number | null; tokensUsed: number; timeUsedSeconds: number; @@ -480,9 +502,20 @@ export interface HarnessSession { startTurn(input: { message: NativeUserMessage; requestedCollaborationMode?: "default" | "plan"; - }): Promise<{ turnId: string; effectiveCollaborationMode?: "default" | "plan" }>; - steer?(input: { turnId: string; message: NativeUserMessage; correlationId?: string }): Promise; - interrupt?(input: { turnId?: string; reason?: string; signal?: AbortSignal }): Promise; + }): Promise<{ + turnId: string; + effectiveCollaborationMode?: "default" | "plan"; + }>; + steer?(input: { + turnId: string; + message: NativeUserMessage; + correlationId?: string; + }): Promise; + interrupt?(input: { + turnId?: string; + reason?: string; + signal?: AbortSignal; + }): Promise; pendingRuntimeRequests?(): HarnessRuntimeRequest[]; resolveRuntimeRequest?(input: { requestId: string; @@ -503,6 +536,8 @@ export interface HarnessSession { usage?(): Promise | null>; transcript?(): Promise; snapshot(): Promise; + /** Relinquish controller authority without semantically closing the session. */ + detachControllerForRestart?(): Promise; close(input: { reason: string; force?: boolean }): Promise; } diff --git a/packages/paperclip-runner/src/contracts/native-session-backend.ts b/packages/paperclip-runner/src/contracts/native-session-backend.ts index dfa4cd597b..5179cf71b3 100644 --- a/packages/paperclip-runner/src/contracts/native-session-backend.ts +++ b/packages/paperclip-runner/src/contracts/native-session-backend.ts @@ -91,15 +91,27 @@ export interface NativeSession { identity(): NativeRunIdentity; capabilities(): Promise; attachRun?(input: { identity: NativeRunIdentity }): Promise; + /** Relinquish controller authority without suspending provider execution. */ + detachControllerForRestart?(): Promise; events(input?: { afterCursor?: string | null }): AsyncIterable; startTurn(input: { message: NativeUserMessage; requestedCollaborationMode?: "default" | "plan"; - }): Promise<{ turnId: string; effectiveCollaborationMode?: "default" | "plan" }>; - steer?(input: { turnId: string; message: NativeUserMessage; correlationId?: string }): Promise; + }): Promise<{ + turnId: string; + effectiveCollaborationMode?: "default" | "plan"; + }>; + steer?(input: { + turnId: string; + message: NativeUserMessage; + correlationId?: string; + }): Promise; interrupt?(input: { turnId?: string; reason?: string }): Promise; /** Commit cancellation synchronously; the returned promise owns cleanup only. */ - cancel?(input: { reason: string; signal: AbortSignal }): NativeSessionCancellation; + cancel?(input: { + reason: string; + signal: AbortSignal; + }): NativeSessionCancellation; resolveRuntimeRequest?(input: { requestId: string; turnId: string; @@ -122,7 +134,9 @@ export interface NativeSession { turnId: string | null; } | null>; usage?(): Promise | null>; - snapshot(options?: NativeSessionSnapshotOptions): Promise; + snapshot( + options?: NativeSessionSnapshotOptions, + ): Promise; /** * Idempotently stop provider work and release every pending `events().next()` * before this promise resolves. Implementations must settle every promise diff --git a/packages/paperclip-runner/src/drivers/codex/app-server-transport.ts b/packages/paperclip-runner/src/drivers/codex/app-server-transport.ts index e331a60399..e7786f62ce 100644 --- a/packages/paperclip-runner/src/drivers/codex/app-server-transport.ts +++ b/packages/paperclip-runner/src/drivers/codex/app-server-transport.ts @@ -46,6 +46,8 @@ export interface CodexAppServerTransport { resolution: HarnessRuntimeRequestResolution; }): Promise; close(): Promise; + /** Relinquish controller authority while leaving durable runner work alive. */ + detachControllerForRestart?(): Promise; processInfo?(): CodexTransportProcessInfo; attachRun?(input: { runId: string; @@ -445,7 +447,9 @@ export class ProcessCodexAppServerTransport implements CodexAppServerTransport { ); this.#process.stdout.on("end", () => { this.#stdoutDecoder.end(); - this.#fatal(new Error("codex app-server stdout ended before transport closure")); + this.#fatal( + new Error("codex app-server stdout ended before transport closure"), + ); }); this.#process.stdout.on("error", (error) => this.#fatal(error)); this.#process.stdout.on("close", () => { diff --git a/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver-impl.ts b/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver-impl.ts index eed1977f56..70fc2fbead 100644 --- a/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver-impl.ts +++ b/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver-impl.ts @@ -93,8 +93,10 @@ function bootstrapCancellation( // completed. The cancellation reason remains authoritative if cleanup // itself reports a secondary failure. void close().then( - () => rejectAborted(signal?.reason ?? new Error("Codex bootstrap aborted")), - () => rejectAborted(signal?.reason ?? new Error("Codex bootstrap aborted")), + () => + rejectAborted(signal?.reason ?? new Error("Codex bootstrap aborted")), + () => + rejectAborted(signal?.reason ?? new Error("Codex bootstrap aborted")), ); }; signal?.addEventListener("abort", onAbort, { once: true }); @@ -120,6 +122,7 @@ function bootstrapCancellation( export class CodexAppServerDriver implements HarnessDriver { readonly #options: CodexAppServerDriverOptions; readonly #caps: CodexCapabilities; + readonly #persistedProcessIdentities = new WeakMap(); constructor(options: CodexAppServerDriverOptions) { this.#options = options; @@ -157,7 +160,11 @@ export class CodexAppServerDriver implements HarnessDriver { this.#options.driverIdentity?.displayName ?? "Codex app-server", version: this.#options.driverIdentity?.version ?? DRIVER_VERSION, protocolVersion: CODEX_CODEX_PROTOCOL_VERSION, - runtimeContextCapabilities: { instructions: "native", skills: "native", mcp: "native" }, + runtimeContextCapabilities: { + instructions: "native", + skills: "native", + mcp: "native", + }, capabilities: { resume: this.#caps.resume, typedEvents: true, @@ -210,44 +217,45 @@ export class CodexAppServerDriver implements HarnessDriver { const initialize = await cancellation.wait(this.#initialize(transport)); const requestedMode = this.#options.requestedCollaborationMode ?? "default"; - const response = await cancellation.wait(transport.request("thread/start", { - ...createSecuredCodexThreadParams( - workingDirectory, - requestedMode, - this.#options.includeCollaborationModeInstructions ?? true, - this.#options.includeSkillInstructions ?? false, - ), - approvalPolicy: this.#options.approvalPolicy ?? "untrusted", - ...(this.#options.model ? { model: this.#options.model } : {}), - ...(this.#direct() - ? {} - : { - baseInstructions: this.#baseInstructions(), - completionContract: { - revision: - this.#options.taskEnvelope.completionContract.revision, - criterionIds: - this.#options.taskEnvelope.completionContract.criteria.map( - (criterion) => criterion.id, - ), - }, - }), - dynamicTools: this.#direct() - ? [] - : this.#caps.dynamicTools - ? [ - ...(this.#options.dynamicTools ?? []), - ...codexSemanticToolSpecs(), - ] - : [], - experimentalRawEvents: false, - persistExtendedHistory: false, - })); - const collaborationMode = await cancellation.wait(this.#negotiateCollaborationMode( - transport, - response, - requestedMode, - )); + const response = await cancellation.wait( + transport.request("thread/start", { + ...createSecuredCodexThreadParams( + workingDirectory, + requestedMode, + this.#options.includeCollaborationModeInstructions ?? true, + this.#options.includeSkillInstructions ?? false, + ), + approvalPolicy: this.#options.approvalPolicy ?? "untrusted", + ...(this.#options.model ? { model: this.#options.model } : {}), + ...(this.#direct() + ? {} + : { + baseInstructions: this.#baseInstructions(), + completionContract: { + revision: + this.#options.taskEnvelope.completionContract.revision, + criterionIds: + this.#options.taskEnvelope.completionContract.criteria.map( + (criterion) => criterion.id, + ), + }, + }), + dynamicTools: this.#direct() + ? [] + : this.#caps.dynamicTools + ? [ + ...(this.#options.dynamicTools ?? []), + ...codexSemanticToolSpecs(), + ] + : [], + experimentalRawEvents: false, + persistExtendedHistory: false, + }), + ); + await cancellation.wait(this.#persistProcessOwnership(transport)); + const collaborationMode = await cancellation.wait( + this.#negotiateCollaborationMode(transport, response, requestedMode), + ); const opened = this.#openedThread( response, initialize, @@ -309,15 +317,24 @@ export class CodexAppServerDriver implements HarnessDriver { } const transport = this.#transport({ providerRecoveryPolicy: snapshot.providerRecoveryPolicy, + persistedSession: { + driverSessionId: snapshot.driverSessionId, + providerSessionId: snapshot.providerSessionId, + providerIdentity: snapshot.providerIdentity, + activeTurnId: snapshot.activeTurnId, + }, }); const cancellation = bootstrapCancellation(transport, options.signal); try { await cancellation.wait(this.#persistProcessOwnership(transport)); const initialize = await cancellation.wait(this.#initialize(transport)); - const existing = await cancellation.wait(transport.request("thread/read", { - threadId: snapshot.driverSessionId, - includeTurns: true, - })); + const existing = await cancellation.wait( + transport.request("thread/read", { + threadId: snapshot.driverSessionId, + includeTurns: true, + }), + ); + await cancellation.wait(this.#persistProcessOwnership(transport)); const existingThread = record(existing.thread); if (text(existingThread.id) !== snapshot.driverSessionId) { await cancellation.wait(cancellation.close()); @@ -331,26 +348,29 @@ export class CodexAppServerDriver implements HarnessDriver { this.#options.environment, this.#options.workingDirectoryAuthority, ); - const response = await cancellation.wait(transport.request("thread/resume", { - threadId: snapshot.driverSessionId, - ...createSecuredCodexThreadParams( - workingDirectory, + const response = await cancellation.wait( + transport.request("thread/resume", { + threadId: snapshot.driverSessionId, + ...createSecuredCodexThreadParams( + workingDirectory, + this.#options.requestedCollaborationMode ?? "default", + this.#options.includeCollaborationModeInstructions ?? true, + this.#options.includeSkillInstructions ?? false, + ), + baseInstructions: this.#direct() ? "" : this.#baseInstructions(), + approvalPolicy: this.#options.approvalPolicy ?? "untrusted", + ...(this.#options.model ? { model: this.#options.model } : {}), + persistExtendedHistory: false, + }), + ); + await cancellation.wait(this.#persistProcessOwnership(transport)); + const collaborationMode = await cancellation.wait( + this.#negotiateCollaborationMode( + transport, + response, this.#options.requestedCollaborationMode ?? "default", - this.#options.includeCollaborationModeInstructions ?? true, - this.#options.includeSkillInstructions ?? false, ), - baseInstructions: this.#direct() - ? "" - : this.#baseInstructions(), - approvalPolicy: this.#options.approvalPolicy ?? "untrusted", - ...(this.#options.model ? { model: this.#options.model } : {}), - persistExtendedHistory: false, - })); - const collaborationMode = await cancellation.wait(this.#negotiateCollaborationMode( - transport, - response, - this.#options.requestedCollaborationMode ?? "default", - )); + ); const opened = this.#openedThread( response, initialize, @@ -404,16 +424,14 @@ export class CodexAppServerDriver implements HarnessDriver { let reconcileUncheckpointedDispositionTurn = false; let providerTurnIds: Set | null = null; if ( - !this.#direct() - && snapshot.semanticResult == null - && recoveredActiveTurnId === null - && (snapshot.terminalTurns?.length ?? 0) > 0 + !this.#direct() && + snapshot.semanticResult == null && + recoveredActiveTurnId === null && + (snapshot.terminalTurns?.length ?? 0) > 0 ) { const providerHistory = existingThread.turns; const providerHistoryIsArray = Array.isArray(providerHistory); - const turns = providerHistoryIsArray - ? providerHistory.map(record) - : []; + const turns = providerHistoryIsArray ? providerHistory.map(record) : []; const terminalIds = new Set( (snapshot.terminalTurns ?? []).map((turn) => turn.turnId), ); @@ -432,14 +450,16 @@ export class CodexAppServerDriver implements HarnessDriver { .filter((turnId) => turnId.length > 0), ); } - const laterTurns = lastKnownTerminalIndex < 0 - ? [] - : turns.slice(lastKnownTerminalIndex + 1); + const laterTurns = + lastKnownTerminalIndex < 0 + ? [] + : turns.slice(lastKnownTerminalIndex + 1); if (laterTurns.length > 1) { await cancellation.wait(cancellation.close()); return { recovered: false, - reason: "provider exposed multiple uncheckpointed disposition recovery turns", + reason: + "provider exposed multiple uncheckpointed disposition recovery turns", }; } const uncheckpointedTurnId = text(laterTurns[0]?.id); @@ -447,13 +467,14 @@ export class CodexAppServerDriver implements HarnessDriver { await cancellation.wait(cancellation.close()); return { recovered: false, - reason: "provider exposed an unidentifiable disposition recovery turn", + reason: + "provider exposed an unidentifiable disposition recovery turn", }; } if (uncheckpointedTurnId.length > 0) { if ( - dispositionOnlyRecoveryTurnId !== null - && dispositionOnlyRecoveryTurnId !== uncheckpointedTurnId + dispositionOnlyRecoveryTurnId !== null && + dispositionOnlyRecoveryTurnId !== uncheckpointedTurnId ) { await cancellation.wait(cancellation.close()); return { @@ -473,19 +494,15 @@ export class CodexAppServerDriver implements HarnessDriver { } } if ( - dispositionOnlyRecoveryConsumed - && recoveredActiveTurnId === null - && !reconcileUncheckpointedDispositionTurn - && providerTurnIds !== null - && ( - dispositionOnlyRecoveryTurnId === null - || ( - !providerTurnIds.has(dispositionOnlyRecoveryTurnId) - && !(snapshot.terminalTurns ?? []).some( + dispositionOnlyRecoveryConsumed && + recoveredActiveTurnId === null && + !reconcileUncheckpointedDispositionTurn && + providerTurnIds !== null && + (dispositionOnlyRecoveryTurnId === null || + (!providerTurnIds.has(dispositionOnlyRecoveryTurnId) && + !(snapshot.terminalTurns ?? []).some( (terminal) => terminal.turnId === dispositionOnlyRecoveryTurnId, - ) - ) - ) + ))) ) { // Older or crash-raced checkpoints could persist the pre-request // one-shot marker, including a requested turn id, without an accepted @@ -536,6 +553,13 @@ export class CodexAppServerDriver implements HarnessDriver { #transport(context?: { providerRecoveryPolicy?: PersistedHarnessSession["providerRecoveryPolicy"]; + persistedSession?: Pick< + PersistedHarnessSession, + | "driverSessionId" + | "providerSessionId" + | "providerIdentity" + | "activeTurnId" + >; }): CodexAppServerTransport { return ( this.#options.transportFactory?.(context) ?? @@ -592,11 +616,14 @@ export class CodexAppServerDriver implements HarnessDriver { const processInfo: CodexTransportProcessInfo | undefined = transport.processInfo?.(); if (!processInfo || processInfo.exited || processInfo.pid === null) return; + const identity = `${processInfo.pid}:${processInfo.processGroupId ?? ""}:${processInfo.startedAt}`; + if (this.#persistedProcessIdentities.get(transport) === identity) return; await this.#options.onSpawn({ pid: processInfo.pid, processGroupId: processInfo.processGroupId, startedAt: processInfo.startedAt, }); + this.#persistedProcessIdentities.set(transport, identity); } async #initialize( @@ -653,7 +680,10 @@ export class CodexAppServerDriver implements HarnessDriver { if (threadId.length === 0) throw new Error("Codex thread response omitted thread.id"); const providerSessionId = text(thread.sessionId) || null; - if (this.#options.requireProviderSessionIdentity && providerSessionId === null) { + if ( + this.#options.requireProviderSessionIdentity && + providerSessionId === null + ) { throw new Error( `provider_initialize_protocol_error: provider=${this.#options.driverIdentity?.kind ?? "codex"} stage=session.open omitted provider session identity`, ); @@ -713,7 +743,9 @@ export class CodexAppServerDriver implements HarnessDriver { networkAccess: false, }, approvalPolicy: boundedCodexValue( - response.approvalPolicy ?? this.#options.approvalPolicy ?? "untrusted", + response.approvalPolicy ?? + this.#options.approvalPolicy ?? + "untrusted", ), baseInstructions: this.#baseInstructions(), instructionSources: Array.isArray(response.instructionSources) diff --git a/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.lifecycle.test.ts b/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.lifecycle.test.ts index 7ce0c83f3c..0d3b6f0d25 100644 --- a/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.lifecycle.test.ts +++ b/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.lifecycle.test.ts @@ -100,12 +100,14 @@ describe("Codex app-server Codex driver", () => { const cancelled = new Error("session open cancelled before admission"); controller.abort(cancelled); - await expect(driver.openSession({ - runId: "run-pre-aborted", - normalizedSessionId: "normalized-pre-aborted", - workingDirectory: WORKSPACE, - signal: controller.signal, - })).rejects.toBe(cancelled); + await expect( + driver.openSession({ + runId: "run-pre-aborted", + normalizedSessionId: "normalized-pre-aborted", + workingDirectory: WORKSPACE, + signal: controller.signal, + }), + ).rejects.toBe(cancelled); expect(transportFactory).not.toHaveBeenCalled(); }); @@ -124,9 +126,11 @@ describe("Codex app-server Codex driver", () => { const cancelled = new Error("session recovery cancelled before admission"); controller.abort(cancelled); - await expect(recoveryDriver.recoverSession(snapshot, { - signal: controller.signal, - })).rejects.toBe(cancelled); + await expect( + recoveryDriver.recoverSession(snapshot, { + signal: controller.signal, + }), + ).rejects.toBe(cancelled); expect(transportFactory).not.toHaveBeenCalled(); }); @@ -137,17 +141,20 @@ describe("Codex app-server Codex driver", () => { const cancelled = new Error("session open cancelled while blocked"); let settled = false; - const opening = driver.openSession({ - runId: "run-blocked-open", - normalizedSessionId: "normalized-blocked-open", - workingDirectory: WORKSPACE, - signal: controller.signal, - }).then( - () => ({ error: null }), - (error: unknown) => ({ error }), - ).finally(() => { - settled = true; - }); + const opening = driver + .openSession({ + runId: "run-blocked-open", + normalizedSessionId: "normalized-blocked-open", + workingDirectory: WORKSPACE, + signal: controller.signal, + }) + .then( + () => ({ error: null }), + (error: unknown) => ({ error }), + ) + .finally(() => { + settled = true; + }); await transport.blocked; controller.abort(cancelled); await transport.closeStarted; @@ -178,14 +185,17 @@ describe("Codex app-server Codex driver", () => { const cancelled = new Error("session recovery cancelled while blocked"); let settled = false; - const recovering = driver.recoverSession(snapshot, { - signal: controller.signal, - }).then( - (value) => ({ value, error: null }), - (error: unknown) => ({ value: null, error }), - ).finally(() => { - settled = true; - }); + const recovering = driver + .recoverSession(snapshot, { + signal: controller.signal, + }) + .then( + (value) => ({ value, error: null }), + (error: unknown) => ({ value: null, error }), + ) + .finally(() => { + settled = true; + }); await recoveryTransport.blocked; controller.abort(cancelled); await recoveryTransport.closeStarted; @@ -241,6 +251,112 @@ describe("Codex app-server Codex driver", () => { expect(transport.calls[0]?.method).toBe("initialize"); }); + it("persists process ownership after a lazy transport launches during session open", async () => { + const transport = new FakeCodexTransport(); + Object.assign(transport, { + processInfo: () => ({ + pid: transport.calls.some((call) => call.method === "thread/start") + ? 71_002 + : null, + processGroupId: 71_002, + startedAt: "2026-08-18T18:01:00.000Z", + exited: false, + exitCode: null, + signal: null, + }), + }); + const onSpawn = vi.fn(async () => undefined); + const driver = makeDriver([transport], { onSpawn }); + + await driver.openSession({ + runId: "run-lazy-owned", + normalizedSessionId: "normalized-lazy-owned", + workingDirectory: WORKSPACE, + }); + + expect(onSpawn).toHaveBeenCalledOnce(); + expect(onSpawn).toHaveBeenCalledWith({ + pid: 71_002, + processGroupId: 71_002, + startedAt: "2026-08-18T18:01:00.000Z", + }); + expect(transport.calls.map((call) => call.method)).toContain( + "thread/start", + ); + }); + + it("persists process ownership after a lazy transport launches during recovery", async () => { + const originalTransport = new FakeCodexTransport(); + const originalDriver = makeDriver([originalTransport]); + const original = await originalDriver.openSession({ + runId: "run-lazy-recovery-owned", + normalizedSessionId: "normalized-lazy-recovery-owned", + workingDirectory: WORKSPACE, + }); + const snapshot = await original.snapshot(); + snapshot.activeTurnId = "turn-recovery-race"; + await original.close({ reason: "prepare lazy ownership recovery" }); + + const recoveryTransport = new FakeCodexTransport(); + Object.assign(recoveryTransport, { + processInfo: () => ({ + pid: recoveryTransport.calls.some( + (call) => call.method === "thread/read", + ) + ? 71_003 + : null, + processGroupId: 71_003, + startedAt: "2026-08-18T18:02:00.000Z", + exited: false, + exitCode: null, + signal: null, + }), + }); + const onSpawn = vi.fn(async () => undefined); + const transportFactory = vi.fn(() => recoveryTransport); + const recoveryDriver = makeDriver([], { onSpawn, transportFactory }); + + const recovered = await recoveryDriver.recoverSession(snapshot); + + expect(recovered.recovered).toBe(true); + expect(transportFactory).toHaveBeenCalledWith({ + providerRecoveryPolicy: snapshot.providerRecoveryPolicy, + persistedSession: { + driverSessionId: snapshot.driverSessionId, + providerSessionId: snapshot.providerSessionId, + providerIdentity: snapshot.providerIdentity, + activeTurnId: snapshot.activeTurnId, + }, + }); + expect(onSpawn).toHaveBeenCalledOnce(); + expect(onSpawn).toHaveBeenCalledWith({ + pid: 71_003, + processGroupId: 71_003, + startedAt: "2026-08-18T18:02:00.000Z", + }); + expect(recoveryTransport.calls.map((call) => call.method)).toContain( + "thread/read", + ); + }); + + it("detaches restart authority without closing the provider transport", async () => { + const transport = new FakeCodexTransport(); + const detachControllerForRestart = vi.fn(async () => undefined); + Object.assign(transport, { detachControllerForRestart }); + const driver = makeDriver([transport]); + const session = await driver.openSession({ + runId: "run-hot-detach", + normalizedSessionId: "normalized-hot-detach", + workingDirectory: WORKSPACE, + }); + const close = vi.spyOn(transport, "close"); + + await session.detachControllerForRestart?.(); + + expect(detachControllerForRestart).toHaveBeenCalledOnce(); + expect(close).not.toHaveBeenCalled(); + }); + it("sends direct chat as plain text and permits a follow-up turn", async () => { const transport = new FakeCodexTransport(); const driver = makeDriver([transport], { conversationMode: "direct" }); @@ -289,7 +405,9 @@ describe("Codex app-server Codex driver", () => { it("forwards the persisted native model to the runner transport", async () => { const transport = new FakeCodexTransport(); - const driver = makeDriver([transport], { model: "qualified-provider-model" }); + const driver = makeDriver([transport], { + model: "qualified-provider-model", + }); await driver.openSession({ runId: "run-qualified-model", @@ -326,7 +444,9 @@ describe("Codex app-server Codex driver", () => { workingDirectory: WORKSPACE, }); - const threadStart = transport.calls.find((call) => call.method === "thread/start"); + const threadStart = transport.calls.find( + (call) => call.method === "thread/start", + ); expect(threadStart?.params).toMatchObject({ baseInstructions, config: { @@ -334,7 +454,9 @@ describe("Codex app-server Codex driver", () => { include_apps_instructions: false, }, }); - expect(JSON.stringify(threadStart?.params.input ?? null)).not.toContain(baseInstructions); + expect(JSON.stringify(threadStart?.params.input ?? null)).not.toContain( + baseInstructions, + ); }); it("passes the common typed-event contract and reports one provider turn terminal", async () => { @@ -465,7 +587,9 @@ describe("Codex app-server Codex driver", () => { expect( events.filter((event) => event.eventType === "run.terminal"), ).toHaveLength(0); - const workspaceEvents = events.filter((event) => event.eventType === "workspace.change.updated"); + const workspaceEvents = events.filter( + (event) => event.eventType === "workspace.change.updated", + ); expect(workspaceEvents[0]?.payload).toMatchObject({ schema: "paperclip.workspace.diff.v1", changeSetId: `${turn.turnId}:workspace`, @@ -477,9 +601,13 @@ describe("Codex app-server Codex driver", () => { complete: true, totals: { files: 1 }, }); - const planEvents = events.filter((event) => event.eventType === "plan.updated"); + const planEvents = events.filter( + (event) => event.eventType === "plan.updated", + ); expect(planEvents).toHaveLength(2); - expect(new Set(planEvents.map((event) => event.itemId))).toEqual(new Set([turn.turnId])); + expect(new Set(planEvents.map((event) => event.itemId))).toEqual( + new Set([turn.turnId]), + ); expect(planEvents.at(-1)?.payload).toMatchObject({ planId: turn.turnId, revision: 2, @@ -551,5 +679,4 @@ describe("Codex app-server Codex driver", () => { modelContextWindow: 128000, }); }); - }); diff --git a/packages/paperclip-runner/src/drivers/codex/codex-driver-types.ts b/packages/paperclip-runner/src/drivers/codex/codex-driver-types.ts index e79103009c..1d946e0fb6 100644 --- a/packages/paperclip-runner/src/drivers/codex/codex-driver-types.ts +++ b/packages/paperclip-runner/src/drivers/codex/codex-driver-types.ts @@ -30,6 +30,13 @@ export interface CodexAppServerDriverOptions { includeCollaborationModeInstructions?: boolean; transportFactory?: (context?: { providerRecoveryPolicy?: PersistedHarnessSession["providerRecoveryPolicy"]; + persistedSession?: Pick< + PersistedHarnessSession, + | "driverSessionId" + | "providerSessionId" + | "providerIdentity" + | "activeTurnId" + >; }) => CodexAppServerTransport; /** Additional control-plane tools exposed to the provider for this run. */ dynamicTools?: readonly Readonly>[]; diff --git a/packages/paperclip-runner/src/drivers/codex/codex-harness-session.ts b/packages/paperclip-runner/src/drivers/codex/codex-harness-session.ts index eaf526ed19..74d7ca9942 100644 --- a/packages/paperclip-runner/src/drivers/codex/codex-harness-session.ts +++ b/packages/paperclip-runner/src/drivers/codex/codex-harness-session.ts @@ -26,7 +26,10 @@ import { } from "../../contracts/codex.js"; import type { PrpEvent } from "../../protocol/replay-contract.js"; import { boundedCodexPayload as boundedPayload } from "./codex-boundaries.js"; -import { CodexRpcError, redactCodexDiagnostic } from "./app-server-transport.js"; +import { + CodexRpcError, + redactCodexDiagnostic, +} from "./app-server-transport.js"; import { runtimeRequestResponse } from "./codex-question-adapter.js"; import { CODEX_PLANNING_PERMISSION_PROFILE as PLANNING_PERMISSION_PROFILE, @@ -43,10 +46,16 @@ import { } from "./codex-session-state.js"; import { pumpNotifications } from "./codex-session-notifications.js"; import { handleServerRequest } from "./codex-session-server-requests.js"; -import { mapTerminalTurn, terminalReplayConflict } from "./codex-session-terminal.js"; +import { + mapTerminalTurn, + terminalReplayConflict, +} from "./codex-session-terminal.js"; import { boundedText, record, text, userInput } from "./codex-driver-values.js"; -export class CodexHarnessSession extends CodexSessionState implements HarnessSession { +export class CodexHarnessSession + extends CodexSessionState + implements HarnessSession +{ constructor(input: CodexSessionStateInput) { super(input); this.transport.setServerRequestHandler((request) => @@ -132,10 +141,10 @@ export class CodexHarnessSession extends CodexSessionState implements HarnessSes ? input.message.text : dispositionOnlyRecovery ? input.message.text - : JSON.stringify({ - task: this.taskEnvelope, - message: input.message.text, - }); + : JSON.stringify({ + task: this.taskEnvelope, + message: input.message.text, + }); const effectiveCollaborationMode = this.opened.context.collaborationMode; if ( input.requestedCollaborationMode && @@ -434,12 +443,13 @@ export class CodexHarnessSession extends CodexSessionState implements HarnessSes this.requireCapability("runtimeRequestResolution"); const pending = this.pendingRuntimeRequestMap.get(input.requestId); if ( - pending === undefined - || pending.request.input === undefined - || pending.request.turnId !== input.turnId - || this.activeTurnId !== input.turnId - || pending.settlingResolution !== undefined - ) return { result: "already_settled", cleanup: Promise.resolve() }; + pending === undefined || + pending.request.input === undefined || + pending.request.turnId !== input.turnId || + this.activeTurnId !== input.turnId || + pending.settlingResolution !== undefined + ) + return { result: "already_settled", cleanup: Promise.resolve() }; if (!this.pendingRuntimeRequestMap.delete(input.requestId)) { return { result: "already_settled", cleanup: Promise.resolve() }; } @@ -450,15 +460,19 @@ export class CodexHarnessSession extends CodexSessionState implements HarnessSes ); pending.settle(safeRequestResponse(pending.request.method, "cancel")); const cleanup = Promise.allSettled([ - Promise.resolve().then(() => this.transport.resolveRuntimeRequest?.({ - requestId: input.requestId, - turnId: input.turnId, - resolution: { action: "cancel" }, - })), - Promise.resolve().then(() => this.transport.request("turn/interrupt", { - threadId: this.opened.threadId, - turnId: input.turnId, - })), + Promise.resolve().then(() => + this.transport.resolveRuntimeRequest?.({ + requestId: input.requestId, + turnId: input.turnId, + resolution: { action: "cancel" }, + }), + ), + Promise.resolve().then(() => + this.transport.request("turn/interrupt", { + threadId: this.opened.threadId, + turnId: input.turnId, + }), + ), ]).then(() => undefined); return { result: "handed_off", cleanup }; } @@ -520,7 +534,9 @@ export class CodexHarnessSession extends CodexSessionState implements HarnessSes } lineage(): HarnessThreadLineageEntry[] { - return [...this.lineageByThread.values()].map((entry) => structuredClone(entry)); + return [...this.lineageByThread.values()].map((entry) => + structuredClone(entry), + ); } async read(): Promise> { @@ -557,7 +573,8 @@ export class CodexHarnessSession extends CodexSessionState implements HarnessSes const reconciledUsage = boundedPayload( record(thread.tokenUsage ?? snapshot.tokenUsage), ); - if (Object.keys(reconciledUsage).length > 0) this.usageSnapshot = reconciledUsage; + if (Object.keys(reconciledUsage).length > 0) + this.usageSnapshot = reconciledUsage; const activeTurns = turns.filter( (turn) => text(turn.status) === "inProgress", ); @@ -626,7 +643,9 @@ export class CodexHarnessSession extends CodexSessionState implements HarnessSes async usage(): Promise | null> { this.requireCapability("usage"); - return this.usageSnapshot === null ? null : structuredClone(this.usageSnapshot); + return this.usageSnapshot === null + ? null + : structuredClone(this.usageSnapshot); } async snapshot(): Promise { @@ -655,12 +674,11 @@ export class CodexHarnessSession extends CodexSessionState implements HarnessSes turnId, fingerprint, })), - dispositionOnlyRecoveryConsumed: - this.dispositionOnlyRecoveryConsumed, - dispositionOnlyRecoveryTurnId: - this.dispositionOnlyRecoveryTurnId, + dispositionOnlyRecoveryConsumed: this.dispositionOnlyRecoveryConsumed, + dispositionOnlyRecoveryTurnId: this.dispositionOnlyRecoveryTurnId, pendingRuntimeRequests: this.pendingRuntimeRequests(), - goal: this.currentGoal === null ? null : structuredClone(this.currentGoal), + goal: + this.currentGoal === null ? null : structuredClone(this.currentGoal), lineage: this.lineage(), lastSourceSequence: this.sourceSequence, }; @@ -672,4 +690,8 @@ export class CodexHarnessSession extends CodexSessionState implements HarnessSes await this.transport.close(); } + async detachControllerForRestart(): Promise { + if (this.transport.detachControllerForRestart === undefined) return; + await this.transport.detachControllerForRestart(); + } } 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 3b96d6b82c..c8613723a4 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts @@ -888,6 +888,17 @@ it("resolves canonical and legacy durable session identities", () => { threadId: "provider-thread-1", sessionId: "provider-account-1", }); + expect( + resolveRunnerdSessionIdentity({ + driverSessionId: "provider-thread-2", + providerSessionId: "provider-account-2", + processId: 4243, + }), + ).toEqual({ + processId: 4243, + threadId: "provider-thread-2", + sessionId: "provider-account-2", + }); expect( resolveRunnerdSessionIdentity({ threadId: "legacy-thread-1", @@ -1987,7 +1998,7 @@ it("cold-restores a suspended provider session under its durable run binding", a } }, 30_000); -it("adopts a live runner on the same durable authority without spawning a duplicate", async () => { +async function verifyLiveRunnerAdoption(mismatchedCheckpoint: boolean) { const stateDirectory = await mkdtemp(join(tmpdir(), "runnerd-live-adopt-")); const server = createServer(); let authority: DurablePrpControlPlane | null = null; @@ -2035,8 +2046,9 @@ it("adopts a live runner on the same durable authority without spawning a duplic let adopted: ReturnType | null = null; try { + let opened: Record; try { - await first.transport.request("thread/start", { + opened = await first.transport.request("thread/start", { cwd: tmpdir(), dynamicTools: codexSemanticToolSpecs(), }); @@ -2055,12 +2067,43 @@ it("adopts a live runner on the same durable authority without spawning a duplic await first.detachControllerForRestart(); expect(() => process.kill(runnerPid!, 0)).not.toThrow(); + const controlPlaneStatePath = join( + stateDirectory, + "control-plane", + "control-plane-state.json", + ); + const compactProviderIdentityEvents = async () => { + const controlPlaneState = JSON.parse( + await readFile(controlPlaneStatePath, "utf8"), + ) as { committedEvents: Array<{ eventType: string }> }; + controlPlaneState.committedEvents = + controlPlaneState.committedEvents.filter( + (event) => + event.eventType !== "harness.ready" && + event.eventType !== "session.started" && + event.eventType !== "session.resumed", + ); + await writeFile( + controlPlaneStatePath, + `${JSON.stringify(controlPlaneState, null, 2)}\n`, + { mode: 0o600 }, + ); + }; + await compactProviderIdentityEvents(); + const duplicateLauncher = vi.fn(() => { throw new Error("duplicate runner spawn attempted"); }); + const openedThread = opened.thread as Record; adopted = createCapabilityRunnerdCodexTransport({ ...sharedOptions, resumeDynamicTools: [], + resumeProviderSession: { + driverSessionId: String(openedThread.id), + providerSessionId: mismatchedCheckpoint + ? "wrong-provider-session" + : String(openedThread.sessionId), + }, runnerProcessLauncher: duplicateLauncher, adoptExistingRunner: { pid: runnerPid!, @@ -2076,6 +2119,14 @@ it("adopts a live runner on the same durable authority without spawning a duplic }, }, }); + if (mismatchedCheckpoint) { + await expect( + adopted.transport.request("thread/read", {}), + ).rejects.toThrow("native_adopted_provider_identity_mismatch"); + expect(duplicateLauncher).not.toHaveBeenCalled(); + expect(() => process.kill(runnerPid!, 0)).not.toThrow(); + return; + } await expect(adopted.transport.request("thread/read", {})).resolves.toEqual( expect.objectContaining({ thread: expect.objectContaining({ id: "codex-thread-1" }), @@ -2086,6 +2137,12 @@ it("adopts a live runner on the same durable authority without spawning a duplic expect(adopted.evidence().diagnostics).toContain( `adopted runner ${runnerPid} authenticated to its durable PRP authority`, ); + expect(adopted.evidence().diagnostics).toContain( + "restored adopted provider identity from the exact durable checkpoint after PRP event compaction; awaiting live confirmation", + ); + expect(adopted.evidence().diagnostics).toContain( + "confirmed adopted provider identity against authenticated session.snapshot", + ); } finally { await adopted?.transport.close().catch(() => undefined); if (runnerPid) { @@ -2097,11 +2154,25 @@ it("adopts a live runner on the same durable authority without spawning a duplic } server.closeAllConnections(); if (server.listening) { - await new Promise((resolveClose) => server.close(() => resolveClose())); + await new Promise((resolveClose) => + server.close(() => resolveClose()), + ); } await rm(stateDirectory, { recursive: true, force: true }); } -}, 30_000); +} + +it( + "adopts a live runner on the same durable authority without spawning a duplicate", + () => verifyLiveRunnerAdoption(false), + 30_000, +); + +it( + "rejects a live runner whose provider identity mismatches the compacted checkpoint", + () => verifyLiveRunnerAdoption(true), + 30_000, +); it("surfaces a runner exit while provider-ingress readiness is still pending", async () => { const neverReady = new Promise(() => undefined); diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts index db3e392a0d..4f893a89df 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts @@ -34,7 +34,10 @@ import type { DurableRecoveryCommittedEvent, DurableRecoveryIdentity, } from "../contracts/durable-recovery.js"; -import type { HarnessRuntimeRequestResolution } from "../contracts/harness-driver.js"; +import type { + HarnessRuntimeRequestResolution, + PersistedHarnessProviderIdentity, +} from "../contracts/harness-driver.js"; import { DurablePrpControlPlane, durableRecoveryInternals, @@ -691,6 +694,17 @@ export interface CapabilityRunnerdCodexTransportOptions { }; /** Provider turn recorded by the owner checkpoint when restoring an active run. */ resumeActiveTurnId?: string | null; + /** + * Exact provider identity from the database checkpoint. A verified adopted + * runner may use this when its original identity event has left the bounded + * PRP replay window. Recovery remains blocked until an authenticated live + * snapshot or a fresh identity event matches this checkpoint. + */ + resumeProviderSession?: { + driverSessionId: string; + providerSessionId?: string | null; + providerIdentity?: PersistedHarnessProviderIdentity; + }; /** Explicitly permits ACPX to rotate its provider-native session after a governed wait. */ providerRecoveryPolicy?: | "same_session_only" @@ -797,8 +811,14 @@ export function resolveRunnerdSessionIdentity(input: unknown): { descriptor.processId ?? started.processId ?? started.pid; - const threadId = started.threadId ?? started.providerSessionId; - const sessionId = started.sessionId ?? started.providerAccountSessionId; + const threadId = + started.threadId ?? started.driverSessionId ?? started.providerSessionId; + const sessionId = + started.sessionId ?? + started.providerAccountSessionId ?? + (started.driverSessionId === undefined + ? undefined + : started.providerSessionId); return { processId: typeof processId === "number" ? processId : null, threadId: @@ -1659,6 +1679,12 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { #threadId = ""; #sessionId: string | null = null; #providerIdentity: Record | null = null; + #checkpointProviderIdentityExpectation: { + driverSessionId: string; + providerSessionId: string; + providerIdentity: Record | null; + } | null = null; + #checkpointProviderIdentityConfirmed = false; #turnId = ""; #turnStartResponsePending = false; #turnStartResponseEpoch = 0; @@ -1813,11 +1839,15 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { } if (method === "thread/read") { if (this.#core === null) await this.#resume(); - // An authenticated recovered runner has already restarted the provider - // and emitted harness.ready with its exact durable thread identity. Ask - // runnerd for its provider snapshot rather than reading its filesystem: - // remote process owners need the same recovery contract as local ones. + // Ask the authenticated runner for its live provider snapshot rather + // than reading its filesystem. This both supports remote process owners + // and proves any identity restored after PRP event compaction before the + // checkpoint-backed thread is exposed to the driver. const snapshot = await this.#commandResult("session.snapshot", {}); + this.#confirmCheckpointProviderIdentity( + snapshot, + "authenticated session.snapshot", + ); const activeProviderTurnId = typeof snapshot.activeProviderTurnId === "string" && snapshot.activeProviderTurnId.length > 0 @@ -1895,7 +1925,17 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { await this.#command("session.destroy", params); return {}; } - if (method === "thread/resume") + if (method === "thread/resume") { + if ( + this.#checkpointProviderIdentityExpectation !== null && + !this.#checkpointProviderIdentityConfirmed + ) { + const snapshot = await this.#commandResult("session.snapshot", {}); + this.#confirmCheckpointProviderIdentity( + snapshot, + "authenticated session.snapshot", + ); + } return { thread: { id: this.#threadId, @@ -1905,6 +1945,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { : { providerIdentity: structuredClone(this.#providerIdentity) }), }, }; + } throw new Error( `PRP Codex transport does not expose provider method ${method}`, ); @@ -2137,9 +2178,18 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { clearInterval(this.#adoptedRunnerMonitor); this.#adoptedRunnerMonitor = null; this.#core?.disconnectActiveRunner(); - if (this.#controlPlaneRelease !== null) await this.#controlPlaneRelease(); - await this.#core?.stop(); + const release = this.#controlPlaneRelease; this.#controlPlaneRelease = null; + await Promise.resolve(release?.()).catch((error: unknown) => { + this.#diagnostic( + `controller route release failed during restart detach: ${String(error)}`, + ); + }); + await this.#core?.stop().catch((error: unknown) => { + this.#diagnostic( + `controller authority stop failed during restart detach: ${String(error)}`, + ); + }); this.#handle = null; this.#queue.close(); this.#diagnostic( @@ -2925,6 +2975,31 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { this.#applyProviderIdentityEvent( committedEvents[adoptedProviderIdentityIndex]!, ); + } else if ( + this.options.adoptExistingRunner && + exactAuthority && + adoptedProviderIdentityIndex < 0 && + this.options.resumeProviderSession?.driverSessionId.trim() && + this.options.resumeProviderSession.providerSessionId?.trim() + ) { + this.#threadId = this.options.resumeProviderSession.driverSessionId; + this.#sessionId = this.options.resumeProviderSession.providerSessionId; + if (this.options.resumeProviderSession.providerIdentity !== undefined) { + this.#providerIdentity = record( + structuredClone(this.options.resumeProviderSession.providerIdentity), + ); + } + this.#checkpointProviderIdentityExpectation = { + driverSessionId: this.#threadId, + providerSessionId: this.#sessionId, + providerIdentity: + this.#providerIdentity === null + ? null + : structuredClone(this.#providerIdentity), + }; + this.#diagnostic( + "restored adopted provider identity from the exact durable checkpoint after PRP event compaction; awaiting live confirmation", + ); } const registration = this.options.controlPlaneRegistration ? await this.options.controlPlaneRegistration(core) @@ -3115,7 +3190,8 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { this.#pumpEvents(); if ( this.#threadId.length > 0 && - (this.#evidence.providerExecutionKind === "remote_service" || + (this.#checkpointProviderIdentityExpectation !== null || + this.#evidence.providerExecutionKind === "remote_service" || this.#evidence.providerPid !== null) ) return; @@ -3404,6 +3480,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { sessionId, } = resolveRunnerdSessionIdentity(started); const providerIdentity = record(started.providerIdentity); + this.#confirmCheckpointProviderIdentity(started, event.eventType); if (pid !== null) { this.#evidence.providerPid = pid; this.#evidence.providerProcessStartedAt = readLocalProcessStartedAt(pid); @@ -3458,6 +3535,44 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { this.#publish(); } + #confirmCheckpointProviderIdentity(input: unknown, source: string): void { + const checkpointExpectation = this.#checkpointProviderIdentityExpectation; + if (checkpointExpectation === null) return; + const { threadId, sessionId } = resolveRunnerdSessionIdentity(input); + const providerIdentity = record(record(input).providerIdentity); + if ( + threadId === null && + sessionId === null && + typeof providerIdentity.kind !== "string" + ) { + return; + } + const providerIdentityMatches = + checkpointExpectation.providerIdentity === null || + (typeof providerIdentity.kind === "string" && + durableRecoveryInternals.canonicalJson(providerIdentity) === + durableRecoveryInternals.canonicalJson( + checkpointExpectation.providerIdentity, + )); + const mismatchFields = [ + ...(threadId === checkpointExpectation.driverSessionId + ? [] + : ["driverSessionId"]), + ...(sessionId === checkpointExpectation.providerSessionId + ? [] + : ["providerSessionId"]), + ...(providerIdentityMatches ? [] : ["providerIdentity"]), + ]; + if (mismatchFields.length > 0) { + throw new Error( + `native_adopted_provider_identity_mismatch: ${source} did not match the exact durable checkpoint (${mismatchFields.join(", ")})`, + ); + } + if (this.#checkpointProviderIdentityConfirmed) return; + this.#checkpointProviderIdentityConfirmed = true; + this.#diagnostic(`confirmed adopted provider identity against ${source}`); + } + #flushPendingTraceRehydrations(): void { const tracePath = this.options.environment?.PAPERCLIP_PROVIDER_TRACE_PATH; if (!tracePath) return; diff --git a/server/src/__tests__/heartbeat-process-recovery.test.ts b/server/src/__tests__/heartbeat-process-recovery.test.ts index 9ca17509de..b207227a63 100644 --- a/server/src/__tests__/heartbeat-process-recovery.test.ts +++ b/server/src/__tests__/heartbeat-process-recovery.test.ts @@ -117,6 +117,7 @@ import { redactDetectedSuccessfulRunProgressSummaryForBoard, redactSuccessfulRunHandoffEvidence, } from "../services/heartbeat.ts"; +import { currentNativeControllerIdentity } from "../services/native-runtime/native-restart-recovery.ts"; import { readHotRestartIntent, readProcessStartedAt, @@ -1445,6 +1446,48 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { expect(wakeup?.status).toBe("claimed"); }); + it("keeps a live native run owned by the current controller out of ambiguous recovery", async () => { + const child = spawnAliveProcess(); + childProcesses.add(child); + expect(child.pid).toBeTypeOf("number"); + + const { companyId, runId, issueId } = await seedRunFixture({ + adapterType: "paperclip_runner", + runtimeMode: "native", + processPid: child.pid ?? null, + }); + const controller = await currentNativeControllerIdentity(); + await db + .update(heartbeatRuns) + .set({ nativeIssueId: issueId, nativePhase: "observed" }) + .where(eq(heartbeatRuns.id, runId)); + await db.insert(nativeRunFinalizations).values({ + runId, + companyId, + issueId, + phase: "observed", + attempt: 1, + leaseOwner: "current-controller:test", + leaseExpiresAt: new Date(Date.now() + 60_000), + controllerBootId: controller.bootId, + controllerPid: controller.pid, + controllerProcessStartedAt: controller.processStartedAt, + controllerGeneration: 1, + }); + + const heartbeat = heartbeatService(db); + const result = await heartbeat.reapOrphanedRuns(); + + expect(result).toEqual({ reaped: 0, runIds: [] }); + expect(await heartbeat.getRun(runId)).toMatchObject({ + status: "running", + error: null, + errorCode: null, + processPid: child.pid, + }); + expect(mockTerminateLocalService).not.toHaveBeenCalled(); + }); + it("does not reap a retryable native run while its same-run recovery path owns it", async () => { const { companyId, agentId, runId, issueId, wakeupRequestId } = await seedRunFixture({ @@ -2577,12 +2620,11 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { }); it("suspends native Paperclip Runner ownership on graceful restart without cancelling or creating a retry run", async () => { - const { agentId, runId, issueId, wakeupRequestId } = - await seedRunFixture({ - adapterType: "paperclip_runner", - agentStatus: "running", - runtimeMode: "native", - }); + const { agentId, runId, issueId, wakeupRequestId } = await seedRunFixture({ + adapterType: "paperclip_runner", + agentStatus: "running", + runtimeMode: "native", + }); await db .update(heartbeatRuns) .set({ nativeIssueId: issueId }) diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 93d266bbd6..f10ef49e49 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -128,7 +128,9 @@ import { buildNativeRuntimeContext, cancelNativeSession, claimNativeRestartRecoveries, + currentNativeControllerIdentity, dispatchNativeSessionResumptions, + detachNativeSessionsForRestart, ensureNativeCompletionContract, executePaperclipNativeSession, finalizeNativeRun, @@ -12544,11 +12546,22 @@ export function heartbeatService( }); } + const nativeRunIds = activeRuns + .filter( + ({ run, adapterType }) => + adapterType === "paperclip_runner" && + isNativeSessionId(run.nativeSessionId), + ) + .map(({ run }) => run.id); + const detachedNativeSessions = + await detachNativeSessionsForRestart(nativeRunIds); + logger.info( { signal, previousServerPid: intent.previousServerPid, activeRunIds: snapshotRuns.map((run) => run.runId), + detachedNativeSessions, }, "hot-restart shutdown snapshot captured; skipping graceful run drain", ); @@ -12719,10 +12732,7 @@ export function heartbeatService( continue; } - if ( - run.runtimeMode === "native" && - adapterType === "paperclip_runner" - ) { + if (run.runtimeMode === "native" && adapterType === "paperclip_runner") { classify(candidate, "skipped", "native_restart_recovery_owned", patch); continue; } @@ -12987,7 +12997,9 @@ export function heartbeatService( ); }); activeRunExecutionPromises.add(execution); - void execution.finally(() => activeRunExecutionPromises.delete(execution)); + void execution.finally(() => + activeRunExecutionPromises.delete(execution), + ); } return { @@ -16965,6 +16977,11 @@ export function heartbeatService( adapterConfig: agents.adapterConfig, nativeCoordinatorPhase: nativeRunFinalizations.phase, nativeRecoveryState: nativeRunFinalizations.recoveryState, + nativeControllerBootId: nativeRunFinalizations.controllerBootId, + nativeControllerPid: nativeRunFinalizations.controllerPid, + nativeControllerProcessStartedAt: + nativeRunFinalizations.controllerProcessStartedAt, + nativeControllerLeaseExpiresAt: nativeRunFinalizations.leaseExpiresAt, }) .from(heartbeatRuns) .innerJoin(agents, eq(heartbeatRuns.agentId, agents.id)) @@ -17004,6 +17021,7 @@ export function heartbeatService( ); const reaped: string[] = []; + const currentNativeController = await currentNativeControllerIdentity(); for (const { run, @@ -17011,6 +17029,10 @@ export function heartbeatService( adapterConfig, nativeCoordinatorPhase, nativeRecoveryState, + nativeControllerBootId, + nativeControllerPid, + nativeControllerProcessStartedAt, + nativeControllerLeaseExpiresAt, } of activeRuns) { const nativeRun = run.runtimeMode === "native"; const nativeProcessPidAlive = @@ -17019,19 +17041,27 @@ export function heartbeatService( nativeRun && !!run.processGroupId && isProcessGroupAlive(run.processGroupId); + const coordinatorOwnedByCurrentController = + nativeRun && + nativeControllerBootId === currentNativeController.bootId && + nativeControllerPid === currentNativeController.pid && + nativeControllerProcessStartedAt?.getTime() === + currentNativeController.processStartedAt.getTime() && + !!nativeControllerLeaseExpiresAt && + nativeControllerLeaseExpiresAt.getTime() > now.getTime(); const locallyTracked = - runningProcesses.has(run.id) || activeRunExecutions.has(run.id); + runningProcesses.has(run.id) || + activeRunExecutions.has(run.id) || + coordinatorOwnedByCurrentController; if ( nativeRun && - ( - [ - "awaiting_evidence", - "awaiting_runner_reattach", - "resuming_session", - "bootstrap_incomplete", - ].includes(nativeRecoveryState ?? "") || - nativeCoordinatorPhase === "retryable_failure" - ) + ([ + "awaiting_evidence", + "awaiting_runner_reattach", + "resuming_session", + "bootstrap_incomplete", + ].includes(nativeRecoveryState ?? "") || + nativeCoordinatorPhase === "retryable_failure") ) { continue; } @@ -17050,9 +17080,10 @@ export function heartbeatService( // intentionally precedes resumedRunIds so a claim cannot bypass the // ownership check. if ( - nativeProcessPidAlive || - nativeProcessGroupAlive || - observedOwnerUnverified + !locallyTracked && + (nativeProcessPidAlive || + nativeProcessGroupAlive || + observedOwnerUnverified) ) { await markNativeOwnershipUnverified(run, { reason: @@ -20180,30 +20211,29 @@ export function heartbeatService( // recovery existed. Only an entirely unused replacement row may // inherit its source checkpoint; any process/provider evidence on the // replacement makes the ownership ambiguous and therefore ineligible. - const legacyRetrySource = - run.retryOfRunId - ? await db - .select({ - id: heartbeatRuns.id, - companyId: heartbeatRuns.companyId, - agentId: heartbeatRuns.agentId, - runnerInstanceId: heartbeatRuns.runnerInstanceId, - nativeSessionId: heartbeatRuns.nativeSessionId, - runnerProfileJson: heartbeatRuns.runnerProfileJson, - runtimeMode: heartbeatRuns.runtimeMode, - status: heartbeatRuns.status, - }) - .from(heartbeatRuns) - .where( - and( - eq(heartbeatRuns.id, run.retryOfRunId), - eq(heartbeatRuns.companyId, agent.companyId), - eq(heartbeatRuns.agentId, agent.id), - ), - ) - .limit(1) - .then((rows) => rows[0] ?? null) - : null; + const legacyRetrySource = run.retryOfRunId + ? await db + .select({ + id: heartbeatRuns.id, + companyId: heartbeatRuns.companyId, + agentId: heartbeatRuns.agentId, + runnerInstanceId: heartbeatRuns.runnerInstanceId, + nativeSessionId: heartbeatRuns.nativeSessionId, + runnerProfileJson: heartbeatRuns.runnerProfileJson, + runtimeMode: heartbeatRuns.runtimeMode, + status: heartbeatRuns.status, + }) + .from(heartbeatRuns) + .where( + and( + eq(heartbeatRuns.id, run.retryOfRunId), + eq(heartbeatRuns.companyId, agent.companyId), + eq(heartbeatRuns.agentId, agent.id), + ), + ) + .limit(1) + .then((rows) => rows[0] ?? null) + : null; const legacyRetryHasProviderEvidence = legacyRetrySource ? await db .select({ id: heartbeatRunEvents.id }) @@ -20233,18 +20263,17 @@ export function heartbeatService( }) ? legacyRetrySource : null; - const legacyRetrySessionId = compatibleLegacyRetrySource - ?.nativeSessionId; + const legacyRetrySessionId = + compatibleLegacyRetrySource?.nativeSessionId; const taskResumeRunId = taskSessionForRun?.lastRunId && taskSessionForRun.lastRunId !== run.id && isNativeSessionId(taskNativeSessionId) ? taskSessionForRun.lastRunId : null; - const resumableTaskSessionId = - taskResumeRunId - ? taskNativeSessionId - : legacyRetrySessionId ?? null; + const resumableTaskSessionId = taskResumeRunId + ? taskNativeSessionId + : (legacyRetrySessionId ?? null); const priorNativeRunId = taskResumeRunId ?? compatibleLegacyRetrySource?.id ?? null; const previousNativeRun = 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 8dfa0e97a0..feb8f4cc4e 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -32,7 +32,13 @@ type BackendFactoryOptions = { runnerInstanceId?: string; acpxRuntimeDirectory?: string; workingDirectoryAuthority?: "local_filesystem" | "remote_runner"; - codexTransportFactory?: () => unknown; + codexTransportFactory?: (recoveryContext?: { + persistedSession?: { + driverSessionId: string; + providerSessionId?: string | null; + activeTurnId?: string | null; + }; + }) => unknown; dynamicToolHandler?: (call: unknown) => Promise; onSpawn?: (meta: { pid: number; @@ -53,6 +59,12 @@ type RunnerTransportOptions = { opencodePermissionMode?: "allow" | "ask" | "deny"; acpxAgent?: "claude" | "codex"; acpxPermissionMode?: "approve-all" | "approve-reads" | "deny-all"; + resumeActiveTurnId?: string | null; + resumeProviderSession?: { + driverSessionId: string; + providerSessionId?: string | null; + activeTurnId?: string | null; + }; }; const durableControlPlaneState = (identity: Record) => ({ @@ -2339,6 +2351,100 @@ describe("native session same-turn steering", () => { }); describe("native warm session supervision", () => { + it("preserves the active turn when a warm checkpoint resumes the same run", async () => { + const stateBase = await mkdtemp( + join(tmpdir(), "paperclip-warm-same-run-recovery-"), + ); + const previousPaperclipHome = process.env.PAPERCLIP_HOME; + process.env.PAPERCLIP_HOME = stateBase; + const activeRun = { + ...execution, + binding: { + ...execution.binding, + runId: "run-warm-same-run-recovery", + executionWorkspaceId: "workspace-warm-same-run-recovery", + }, + session: { + ...execution.session, + normalizedSessionId: "session-warm-same-run-recovery", + lifecyclePolicy: { mode: "warm" as const, idleTimeoutMs: 20 }, + }, + } as NativeExecutionInputV1; + const checkpoint = { + identity: { + runId: activeRun.binding.runId, + sessionId: activeRun.session.normalizedSessionId, + companyId: activeRun.binding.companyId, + issueId: activeRun.binding.issueId, + agentId: activeRun.binding.agentId, + }, + sessionId: activeRun.session.normalizedSessionId, + driverSessionId: "driver-warm-same-run-recovery", + providerSessionId: "provider-warm-same-run-recovery", + activeTurnId: "provider-turn-warm-same-run-recovery", + semanticResult: null, + terminal: null, + terminalTurns: [], + pendingRuntimeRequests: [], + }; + const result = { + result: { summary: "completed" }, + terminal: { runTerminalState: "succeeded" }, + turnId: checkpoint.activeTurnId, + normalizedSessionId: activeRun.session.normalizedSessionId, + providerSessionId: checkpoint.providerSessionId, + driverKind: "test", + driverVersion: "1", + nativeEventCount: 1, + highestContiguousSourceSeq: 1, + usage: null, + }; + const firstClose = vi.fn(async () => undefined); + state.execute + .mockReset() + .mockImplementationOnce(async (options) => { + await options.onCheckpoint?.(checkpoint); + options.onSession?.({ close: firstClose }); + return result; + }) + .mockImplementationOnce(async (options) => { + expect(options.persistedSession).toEqual( + expect.objectContaining({ + identity: checkpoint.identity, + driverSessionId: checkpoint.driverSessionId, + providerSessionId: checkpoint.providerSessionId, + activeTurnId: checkpoint.activeTurnId, + }), + ); + return result; + }); + + try { + await executePaperclipNativeSession({ + db: leaseDb(activeRun), + execution: activeRun, + runnerInstanceId: "runner-warm-same-run-recovery", + }); + await vi.waitFor(() => expect(firstClose).toHaveBeenCalled(), { + timeout: 500, + }); + await expect( + executePaperclipNativeSession({ + db: leaseDb(activeRun), + execution: activeRun, + runnerInstanceId: "runner-warm-same-run-recovery", + }), + ).resolves.toBeDefined(); + } finally { + if (previousPaperclipHome === undefined) { + delete process.env.PAPERCLIP_HOME; + } else { + process.env.PAPERCLIP_HOME = previousPaperclipHome; + } + await rm(stateBase, { recursive: true, force: true }); + } + }); + it("reuses one session across distinct governed runs and closes it after idle expiry", async () => { const close = vi.fn(async () => undefined); const sharedSession = { close }; @@ -2470,6 +2576,7 @@ describe("native warm session supervision", () => { agentId: first.binding.agentId, }, providerSessionId: "provider-runnerd-warm", + activeTurnId: "provider-turn-runnerd-warm-first", }); options.onSession?.(firstSession); return result; @@ -2483,6 +2590,7 @@ describe("native warm session supervision", () => { sessionId: second.session.normalizedSessionId, }), providerSessionId: "provider-runnerd-warm", + activeTurnId: null, }), ); options.onSession?.(secondSession); @@ -3031,6 +3139,36 @@ describe("runnerd provider runtime wiring", () => { await rm(isolatedStateDirectory, { recursive: true, force: true }); }); + it("passes the run checkpoint active turn into restart recovery", async () => { + state.createBackend.mockClear(); + state.createTransport.mockClear(); + await createRunnerdBackend({ + db: leaseDb(execution), + execution, + runnerInstanceId: "runner-active-turn-recovery", + }); + + state.createTransport.mockClear(); + state.createBackend.mock.calls[0]![1].codexTransportFactory!({ + persistedSession: { + driverSessionId: "driver-session-active-turn", + providerSessionId: "provider-session-active-turn", + activeTurnId: "provider-turn-active", + }, + }); + + expect(state.createTransport).toHaveBeenCalledWith( + expect.objectContaining({ + resumeActiveTurnId: "provider-turn-active", + resumeProviderSession: expect.objectContaining({ + driverSessionId: "driver-session-active-turn", + providerSessionId: "provider-session-active-turn", + activeTurnId: "provider-turn-active", + }), + }), + ); + }); + it("rejects overlapping runs for the same runnerd provider session scope", async () => { const first = { ...execution, diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 5236048ee6..7434f9ce55 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -120,6 +120,32 @@ export class NativeCancellationPendingRecoveryError extends Error { } const activeNativeSessions = new Map(); + +export async function detachNativeSessionsForRestart( + runIds: readonly string[], +): Promise<{ + detachedRunIds: string[]; + inactiveRunIds: string[]; + unsupportedRunIds: string[]; +}> { + const detachedRunIds: string[] = []; + const inactiveRunIds: string[] = []; + const unsupportedRunIds: string[] = []; + for (const runId of new Set(runIds)) { + const active = activeNativeSessions.get(runId); + if (!active) { + inactiveRunIds.push(runId); + continue; + } + if (active.session.detachControllerForRestart === undefined) { + unsupportedRunIds.push(runId); + continue; + } + await active.session.detachControllerForRestart(); + detachedRunIds.push(runId); + } + return { detachedRunIds, inactiveRunIds, unsupportedRunIds }; +} const MAX_REMOTE_CHECKPOINT_ARCHIVE_BYTES = 64 * 1024 * 1024; const MAX_REMOTE_CHECKPOINT_EXPANDED_BYTES = 64 * 1024 * 1024; const MAX_REMOTE_CHECKPOINT_ENTRIES = 20_000; @@ -1524,7 +1550,11 @@ async function migrateRunnerdStateRootForExecution(input: { export function runnerdStateProvesIncompleteBootstrap(root: string): boolean { try { - const statePath = resolve(root, "control-plane", "control-plane-state.json"); + const statePath = resolve( + root, + "control-plane", + "control-plane-state.json", + ); const state = record( JSON.parse( readBoundedNativeFile( @@ -2158,21 +2188,29 @@ function loadWarmNativeCheckpoint( ) { throw new Error("native_session_supervisor_checkpoint_mismatch"); } - const resumed = { - ...envelope.snapshot, - identity: { - runId: execution.binding.runId, - sessionId: nativeSessionKey(execution), - companyId: execution.binding.companyId, - issueId: execution.binding.issueId, - agentId: execution.binding.agentId, - }, - semanticResult: null, - terminal: null, - activeTurnId: null, - terminalTurns: [], - pendingRuntimeRequests: [], - }; + const sameRunRecovery = + persistedIdentity.runId === execution.binding.runId && + persistedIdentity.issueId === execution.binding.issueId; + const resumed = sameRunRecovery + ? structuredClone(envelope.snapshot) + : { + ...envelope.snapshot, + identity: { + runId: execution.binding.runId, + sessionId: nativeSessionKey(execution), + companyId: execution.binding.companyId, + issueId: execution.binding.issueId, + agentId: execution.binding.agentId, + }, + // A warm provider can be rebound only after the previous run settled. + // Its provider identity survives, but run-scoped turn, result, and + // request authority must not cross into the new heartbeat run. + semanticResult: null, + terminal: null, + activeTurnId: null, + terminalTurns: [], + pendingRuntimeRequests: [], + }; if (path !== scopedPath) { // Copy the validated legacy checkpoint into the fully scoped location. // persistWarmNativeCheckpoint uses an atomic rename and leaving the old @@ -3174,7 +3212,8 @@ export async function renewNativeSessionExecutionLease(input: { controller?: NativeControllerIdentity; leaseTtlMs?: number; }): Promise { - const controller = input.controller ?? (await currentNativeControllerIdentity()); + const controller = + input.controller ?? (await currentNativeControllerIdentity()); const leaseTtlMs = input.leaseTtlMs ?? NATIVE_SESSION_EXECUTION_LEASE_TTL_MS; if ( !Number.isInteger(leaseTtlMs) || @@ -6887,13 +6926,13 @@ async function createRunnerdBackendWithinSessionClaim( runnerProcessLauncher: remoteProcessLauncher, runnerReconnectGraceMs: remoteTarget ? 120_000 : undefined, adoptExistingRunner: adoptedProcess - ? { - ...adoptedProcess, - isAlive: () => verifiedRecoveryProcessIsAlive(adoptedProcess), - signal: (signal) => - signalVerifiedRecoveryProcess(adoptedProcess, signal), - } - : undefined, + ? { + ...adoptedProcess, + isAlive: () => verifiedRecoveryProcessIsAlive(adoptedProcess), + signal: (signal) => + signalVerifiedRecoveryProcess(adoptedProcess, signal), + } + : undefined, environment: effectiveRunnerEnvironment, lifecyclePolicy: input.execution.session.lifecyclePolicy, runtimeContext: @@ -6913,6 +6952,9 @@ async function createRunnerdBackendWithinSessionClaim( (criterion) => criterion.id, ), }, + resumeActiveTurnId: + recoveryContext?.persistedSession?.activeTurnId ?? null, + resumeProviderSession: recoveryContext?.persistedSession, providerRecoveryPolicy: recoveryContext?.providerRecoveryPolicy ?? (input.execution.provider.kind === "acpx" &&