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