RomanNum3ral
/
paperclip
mirror of https://github.com/paperclipai/paperclip.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

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
This commit is contained in:
Dotta 2026-09-04 17:54:44 -05:00 committed by GitHub
parent bb920fb859
commit 263f181fed
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
20 changed files with 1040 additions and 321 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_backend.rs
View File

@ -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,
})))
}

3
packages/paperclip-runner/runner/crates/runner-core/src/managed_provider_backend.rs
View File

@ -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,
})))

3
packages/paperclip-runner/runner/crates/runner-core/src/provider_backend.rs
View File

@ -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,
})))

7
packages/paperclip-runner/runner/crates/runner-core/tests/codex_provider.rs
View File

@ -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);

121
packages/paperclip-runner/src/backends/codex-native-backend.ts
View File

@ -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<void>;
transportFactory?: (context?: {
providerRecoveryPolicy?: PersistedNativeSession["providerRecoveryPolicy"];
persistedSession?: Pick<
PersistedHarnessSession,
| "driverSessionId"
| "providerSessionId"
| "providerIdentity"
| "activeTurnId"
>;
}) => CodexAppServerTransport;
dynamicTools?: readonly Readonly<Record<string, unknown>>[];
dynamicToolHandler?: (call: {
@ -34,9 +45,7 @@ export interface CodexNativeSessionBackendOptions {
}) => Promise<unknown>;
}
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,
}));
);
}
/**

5
packages/paperclip-runner/src/backends/harness-driver-backend.ts
View File

@ -438,6 +438,11 @@ class HarnessNativeSession implements NativeSession {
this.#explicitlyCancelled = false;
}
async detachControllerForRestart(): Promise<void> {
if (this.#session.detachControllerForRestart === undefined) return;
await this.#session.detachControllerForRestart();
}
async *events(): AsyncIterable<PrpEvent> {
let sourceInstanceId: string | null = null;
let lastSourceSequence = 0;

8
packages/paperclip-runner/src/backends/native-backend-factory.ts
View File

@ -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;

57
packages/paperclip-runner/src/contracts/harness-driver.ts
View File

@ -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<string, unknown> | null {
: null;
}
function parseAnswers(value: unknown): Record<string, { answers: string[] }> | null {
function parseAnswers(
value: unknown,
): Record<string, { answers: string[] }> | null {
const fields = plainRecord(value);
if (fields === null || Object.keys(fields).length === 0) return null;
const parsed: Record<string, { answers: string[] }> = {};
@ -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<void>;
interrupt?(input: { turnId?: string; reason?: string; signal?: AbortSignal }): Promise<void>;
}): Promise<{
turnId: string;
effectiveCollaborationMode?: "default" | "plan";
}>;
steer?(input: {
turnId: string;
message: NativeUserMessage;
correlationId?: string;
}): Promise<void>;
interrupt?(input: {
turnId?: string;
reason?: string;
signal?: AbortSignal;
}): Promise<void>;
pendingRuntimeRequests?(): HarnessRuntimeRequest[];
resolveRuntimeRequest?(input: {
requestId: string;
@ -503,6 +536,8 @@ export interface HarnessSession {
usage?(): Promise<Record<string, unknown> | null>;
transcript?(): Promise<HarnessTranscriptSnapshot>;
snapshot(): Promise<PersistedHarnessSession>;
/** Relinquish controller authority without semantically closing the session. */
detachControllerForRestart?(): Promise<void>;
close(input: { reason: string; force?: boolean }): Promise<void>;
}

22
packages/paperclip-runner/src/contracts/native-session-backend.ts
View File

@ -91,15 +91,27 @@ export interface NativeSession {
identity(): NativeRunIdentity;
capabilities(): Promise<NativeSessionCapabilities>;
attachRun?(input: { identity: NativeRunIdentity }): Promise<void>;
/** Relinquish controller authority without suspending provider execution. */
detachControllerForRestart?(): Promise<void>;
events(input?: { afterCursor?: string | null }): AsyncIterable<PrpEvent>;
startTurn(input: {
message: NativeUserMessage;
requestedCollaborationMode?: "default" | "plan";
}): Promise<{ turnId: string; effectiveCollaborationMode?: "default" | "plan" }>;
steer?(input: { turnId: string; message: NativeUserMessage; correlationId?: string }): Promise<void>;
}): Promise<{
turnId: string;
effectiveCollaborationMode?: "default" | "plan";
}>;
steer?(input: {
turnId: string;
message: NativeUserMessage;
correlationId?: string;
}): Promise<void>;
interrupt?(input: { turnId?: string; reason?: string }): Promise<void>;
/** 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<Record<string, unknown> | null>;
snapshot(options?: NativeSessionSnapshotOptions): Promise<PersistedNativeSession>;
snapshot(
options?: NativeSessionSnapshotOptions,
): Promise<PersistedNativeSession>;
/**
* Idempotently stop provider work and release every pending `events().next()`
* before this promise resolves. Implementations must settle every promise

6
packages/paperclip-runner/src/drivers/codex/app-server-transport.ts
View File

@ -46,6 +46,8 @@ export interface CodexAppServerTransport {
resolution: HarnessRuntimeRequestResolution;
}): Promise<void>;
close(): Promise<void>;
/** Relinquish controller authority while leaving durable runner work alive. */
detachControllerForRestart?(): Promise<void>;
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", () => {

214
packages/paperclip-runner/src/drivers/codex/codex-app-server-driver-impl.ts
View File

@ -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<object, string>();
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<string> | 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)

197
packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.lifecycle.test.ts
View File

@ -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,
});
});
});

7
packages/paperclip-runner/src/drivers/codex/codex-driver-types.ts
View File

@ -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<Record<string, unknown>>[];

82
packages/paperclip-runner/src/drivers/codex/codex-harness-session.ts
View File

@ -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<Record<string, unknown>> {
@ -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<Record<string, unknown> | null> {
this.requireCapability("usage");
return this.usageSnapshot === null ? null : structuredClone(this.usageSnapshot);
return this.usageSnapshot === null
? null
: structuredClone(this.usageSnapshot);
}
async snapshot(): Promise<PersistedHarnessSession> {
@ -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<void> {
if (this.transport.detachControllerForRestart === undefined) return;
await this.transport.detachControllerForRestart();
}
}

79
packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts
View File

@ -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<typeof createCapabilityRunnerdCodexTransport> | null =
null;
try {
let opened: Record<string, unknown>;
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<string, unknown>;
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<void>((resolveClose) => server.close(() => resolveClose()));
await new Promise<void>((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<void>(() => undefined);

137
packages/paperclip-runner/src/live/runnerd-codex-transport.ts
View File

@ -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<string, unknown> | null = null;
#checkpointProviderIdentityExpectation: {
driverSessionId: string;
providerSessionId: string;
providerIdentity: Record<string, unknown> | 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;

54
server/src/__tests__/heartbeat-process-recovery.test.ts
View File

@ -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 })

125
server/src/services/heartbeat.ts
View File

@ -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 =

140
server/src/services/native-runtime/native-session-executor.test.ts
View File

@ -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<unknown>;
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<string, unknown>) => ({
@ -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,

90
server/src/services/native-runtime/native-session-executor.ts
View File

@ -120,6 +120,32 @@ export class NativeCancellationPendingRecoveryError extends Error {
}
const activeNativeSessions = new Map<string, ActiveNativeSession>();
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<void> {
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" &&