diff --git a/packages/paperclip-runner/docs/durable-recovery.md b/packages/paperclip-runner/docs/durable-recovery.md index d33a00f7e46..ad5c6d6a09d 100644 --- a/packages/paperclip-runner/docs/durable-recovery.md +++ b/packages/paperclip-runner/docs/durable-recovery.md @@ -197,6 +197,26 @@ startup fence over a `prepared` state that holds a saved thread. The next runner retires that fence once the recorded provider process and its group have exited, as it does for an open or active session. +A restart can also kill the runner and Codex while `turn/start` is in flight. +The runner marks the start ambiguous before it sends the request, so after the +restart neither journal says whether Codex created the turn, and the resumed +thread reports no active turn either way. Codex records a turn in the thread +before it runs anything, so the restored runner reads the newest turn of the +thread history (`thread/turns/list`, one turn, metadata only). If there is none, +or it is a turn Paperclip already settled, the start never created a turn: the +runner clears the ambiguity, emits the `ambiguous_turn_start_not_created` +diagnostic and restores an idle session. The old `turn.start` stays settled as +`execution_indeterminate` and never runs again; the same-run resume then starts +the run's turn once, as for any recovered session without an active turn. If +the newest turn is one Paperclip never settled, Codex did create the turn and it +may have done work. The runner still fails closed and never starts that work +again by itself; continuing is a person's decision (Retry or a new message). + +A follow-up run that a prior owner's unproven runner state refuses +(`runner_state_identity_mismatch: prior_owner_*`) fails before it asks for a +runner, so it never had a process. Retry and new messages after it continue +without waiting for a stop record of that run. + Saved OpenCode and Cursor Cloud provider state binds the digest of its own qualified launch profile (command, proxy script and, for OpenCode, the executable, each by path and SHA-256). Cursor Cloud does not re-bind. A changed diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/bin/fake-codex-app-server.rs b/packages/paperclip-runner/runner/crates/runner-core/src/bin/fake-codex-app-server.rs index 382f76c02a3..2a92b5880ec 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/bin/fake-codex-app-server.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/bin/fake-codex-app-server.rs @@ -21,6 +21,41 @@ struct FakeState { goal: Option, } +/// Every turn this thread accepted, oldest first, one per line, like the turn +/// history a real Codex thread keeps across process loss. It sits beside the +/// state file so existing state files and their shape stay unchanged. +fn turn_history_path(state_path: &Path) -> PathBuf { + let mut name = state_path.as_os_str().to_owned(); + name.push(".turns"); + PathBuf::from(name) +} + +fn load_turn_history(state_path: &Path) -> Vec { + fs::read_to_string(turn_history_path(state_path)) + .map(|text| { + text.lines() + .filter(|line| !line.is_empty()) + .map(str::to_owned) + .collect() + }) + .unwrap_or_default() +} + +fn record_turn(state_path: &Path, turn_id: &str) -> io::Result<()> { + if load_turn_history(state_path) + .iter() + .any(|known| known == turn_id) + { + return Ok(()); + } + let mut file = OpenOptions::new() + .create(true) + .append(true) + .open(turn_history_path(state_path))?; + writeln!(file, "{turn_id}")?; + file.sync_all() +} + fn argument(args: &[String], name: &str) -> Option { args.iter() .position(|value| value == name) @@ -1443,9 +1478,30 @@ fn run() -> Result<(), Box> { send(json!({"id": id, "result": {"data": [], "nextCursor": "next-page"}}))?; continue; } - let turns = state.active_turn_id.as_ref() - .map(|turn_id| vec![json!({"id": turn_id, "status": "inProgress", "items": [], "itemsView": "notLoaded"})]) - .unwrap_or_default(); + // Newest first. The thread keeps every accepted turn; only + // the active one is in progress. + let mut history = load_turn_history(&state_path); + if let Some(active) = state.active_turn_id.as_ref() { + history.retain(|turn_id| turn_id != active); + history.push(active.clone()); + } + let limit = message + .pointer("/params/limit") + .and_then(Value::as_u64) + .unwrap_or(100) as usize; + let turns = history + .iter() + .rev() + .take(limit) + .map(|turn_id| { + let status = if state.active_turn_id.as_ref() == Some(turn_id) { + "inProgress" + } else { + "completed" + }; + json!({"id": turn_id, "status": status, "items": [], "itemsView": "notLoaded"}) + }) + .collect::>(); send(json!({"id": id, "result": {"data": turns, "nextCursor": null}}))?; } "thread/read" => { @@ -1654,6 +1710,17 @@ fn run() -> Result<(), Box> { }))?; } "turn/start" => { + // Models a provider killed while turn/start was in flight. The + // file names where it stops: "before" the turn exists, or + // "after" the thread recorded it but before the response. + let hang_turn_start = argument(&args, "--hang-turn-start") + .and_then(|path| fs::read_to_string(path).ok()) + .map(|stage| stage.trim().to_owned()); + if hang_turn_start.as_deref() == Some("before") { + loop { + std::thread::sleep(std::time::Duration::from_secs(60)); + } + } if require_external_sandbox && (message.pointer("/params/sandboxPolicy") != Some(&json!({ @@ -1730,7 +1797,13 @@ fn run() -> Result<(), Box> { ) }; state.active_turn_id = Some(provider_turn_id.clone()); + record_turn(&state_path, &provider_turn_id)?; save_state(&state_path, &state)?; + if hang_turn_start.as_deref() == Some("after") { + loop { + std::thread::sleep(std::time::Duration::from_secs(60)); + } + } if ambiguous_older_reused_turn && turn_start_count == 3 { send(json!({ "method": "turn/started", diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/codex_provider.rs b/packages/paperclip-runner/runner/crates/runner-core/src/codex_provider.rs index e67c96c0f2f..9c2aa55af90 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/codex_provider.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/codex_provider.rs @@ -2514,6 +2514,38 @@ impl CodexProvider { Ok(()) } + /// Provider evidence for a `turn/start` whose outcome a process loss hid. + /// Returns `true` only when this resumed Codex thread has no active turn + /// and its newest recorded turn is absent or one Paperclip already settled: + /// Codex records a turn in the thread before it runs anything, so the + /// ambiguous start never created one. Any other history, an incomplete + /// page, or another provider returns `false` and keeps the start ambiguous. + pub(crate) fn ambiguous_turn_start_never_created_a_turn( + &mut self, + settled_provider_turn_ids: &BTreeSet, + ) -> Result { + if self.config.provider != "codex" || self.active_provider_turn_id.is_some() { + return Ok(false); + } + let page = self.request("thread/turns/list", self.newest_turn_page_params())?; + Ok(newest_turn_page_proves_no_new_turn( + &page, + settled_provider_turn_ids, + )) + } + + /// The first `thread/turns/list` page `read_thread` requests, narrowed to + /// the newest turn's metadata. + fn newest_turn_page_params(&self) -> Value { + let mut params = serde_json::Map::new(); + params.insert("threadId".to_owned(), Value::String(self.thread_id.clone())); + params.insert("cursor".to_owned(), Value::Null); + params.insert("limit".to_owned(), Value::from(1)); + params.insert("sortDirection".to_owned(), Value::from("desc")); + params.insert("itemsView".to_owned(), Value::from("notLoaded")); + Value::Object(params) + } + pub(crate) fn read_thread_after_recovery_goal( &mut self, goal_is_dormant: fn(&Value) -> bool, @@ -5020,6 +5052,35 @@ fn codex_question_response( Ok(json!({"answers": native})) } +/// Codex turn statuses that mean the turn is over. `read_thread` accepts the +/// same set plus `inProgress`. +const ENDED_TURN_STATUSES: [&str; 4] = ["completed", "failed", "interrupted", "cancelled"]; + +/// Whether the newest page of a thread's turn history proves that no turn +/// exists beyond the ones Paperclip settled: the page is empty and complete, +/// or its newest turn is a settled one with an explicit ended status. Any +/// missing, malformed or unknown field proves nothing. +fn newest_turn_page_proves_no_new_turn( + page: &Value, + settled_provider_turn_ids: &BTreeSet, +) -> bool { + let Some(turns) = page.pointer("/data").and_then(Value::as_array) else { + return false; + }; + let Some(newest) = turns.first() else { + // An empty first page is the whole history only without a cursor. + return page.pointer("/nextCursor").is_none_or(Value::is_null); + }; + newest + .pointer("/id") + .and_then(Value::as_str) + .is_some_and(|turn_id| settled_provider_turn_ids.contains(turn_id)) + && newest + .pointer("/status") + .and_then(Value::as_str) + .is_some_and(|status| ENDED_TURN_STATUSES.contains(&status)) +} + #[cfg(test)] mod tests { @@ -5071,6 +5132,45 @@ mod tests { } use super::*; + #[test] + fn only_a_complete_settled_ended_history_proves_no_new_turn() { + let settled = BTreeSet::from(["turn-1".to_owned()]); + let page = |turns: Value, cursor: Value| json!({"data": turns, "nextCursor": cursor}); + let proves = |value: Value| newest_turn_page_proves_no_new_turn(&value, &settled); + assert!(proves(page(json!([]), Value::Null))); + assert!(proves(json!({"data": []}))); + assert!(proves(page( + json!([{"id": "turn-1", "status": "completed"}]), + Value::Null + ))); + assert!(proves(page( + json!([{"id": "turn-1", "status": "interrupted"}]), + Value::Null + ))); + // An incomplete page, an unsettled or in-progress turn, or a missing, + // null or unknown field keeps the start ambiguous. + assert!(!proves(page(json!([]), json!("next-page")))); + assert!(!proves(json!({"nextCursor": null}))); + assert!(!proves(page( + json!([{"id": "turn-2", "status": "completed"}]), + Value::Null + ))); + assert!(!proves(page( + json!([{"id": "turn-1", "status": "inProgress"}]), + Value::Null + ))); + assert!(!proves(page(json!([{"id": "turn-1"}]), Value::Null))); + assert!(!proves(page( + json!([{"id": "turn-1", "status": null}]), + Value::Null + ))); + assert!(!proves(page( + json!([{"id": "turn-1", "status": "paused"}]), + Value::Null + ))); + assert!(!proves(page(json!([{"status": "completed"}]), Value::Null))); + } + #[test] #[cfg(unix)] fn runtime_notification_policy_grants_only_discovery_files_and_ipc() { 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 b51cfce38a6..d3cc6e05690 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 @@ -2768,6 +2768,23 @@ impl CodexCommandExecutor { self.save_state()?; return Ok(()); } + // A process loss can hide whether an in-flight turn/start reached + // Codex. When the resumed thread has no active turn and its history + // ends at a turn Paperclip already settled, Codex never created the + // turn: the start is settled as not started, and the controller + // may start it once. Any other evidence keeps it ambiguous. + let ambiguous_start_never_created = ambiguous_turn_start_pending + && recovered_active_turn_id.is_none() + && previous_active_turn_id.is_none() + && provider + .ambiguous_turn_start_never_created_a_turn(&settled_provider_turn_ids) + .map_err(|error| { + DurableRunnerError::invalid(format!( + "ambiguous {provider_name} turn start history probe failed: {error}" + )) + })?; + let ambiguous_turn_start_pending = + ambiguous_turn_start_pending && !ambiguous_start_never_created; if ambiguous_turn_start_pending { let recovered_turn_id = recovered_active_turn_id.as_deref().ok_or_else(|| { DurableRunnerError::invalid( @@ -2844,6 +2861,19 @@ impl CodexCommandExecutor { state.provider_session_id = resumed_provider_session_id.clone(); state.settled_provider_turn_ids = settled_provider_turn_ids; state.settled_provider_turn_filter = settled_provider_turn_filter; + if ambiguous_start_never_created { + state.ambiguous_turn_start_pending = false; + state.push_event(NormalizedProviderEvent { + event_type: "harness.diagnostic".to_owned(), + priority: EventPriority::P0, + payload: json!({ + "provider": provider_label, + "code": "ambiguous_turn_start_not_created", + "message": format!("{provider_name} recovery proved from the resumed thread's history that the turn start a process loss interrupted never created a turn"), + "paperclipAccepted": false, + }), + })?; + } state.push_terminal_event(NormalizedProviderEvent { event_type: "session.resumed".to_owned(), priority: EventPriority::P0, 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 2d6733572e4..8b0d751e644 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 @@ -3296,6 +3296,103 @@ fn completed_ambiguous_replacement_fails_closed_after_process_loss() { fs::remove_dir_all(directory).expect("remove Codex integration-test directory"); } +/// The durable state a process loss leaves when it lands after the runner +/// journaled a second `turn.start` and marked it ambiguous, but before Codex +/// answered. `created` says whether Codex had already recorded that turn. +fn cut_off_second_turn_start(label: &str, created: bool) -> PathBuf { + let directory = temporary_directory(label); + let config = provider_config(&directory, &["--durable-turn-ids"]); + let mut executor = CodexCommandExecutor::new(&directory); + executor + .execute(&command( + "prepare", + 1, + "run.prepare", + json!({"provider": config}), + )) + .expect("prepare Codex provider"); + executor + .execute(&command("open", 2, "session.open", json!({}))) + .expect("open Codex session"); + executor + .execute(&command( + "first-turn", + 3, + "turn.start", + json!({"text": "Complete the first turn."}), + )) + .expect("start first provider turn"); + wait_for_executor_event(&mut executor, "turn.completed"); + executor.shutdown().expect("stop first provider process"); + drop(executor); + let provider_path = directory.join("codex-provider-state.json"); + let mut provider: Value = + serde_json::from_slice(&fs::read(&provider_path).expect("read provider state")) + .expect("parse provider state"); + provider["ambiguousTurnStartPending"] = json!(true); + fs::write(&provider_path, serde_json::to_vec(&provider).unwrap()) + .expect("write the cut-off provider state"); + if created { + // Codex recorded the second turn in its thread before it died. + use std::io::Write as _; + let mut history = fs::OpenOptions::new() + .append(true) + .open(directory.join("fake-state.json.turns")) + .expect("open fake turn history"); + writeln!(history, "provider-turn-2").expect("record the cut-off turn"); + } + directory +} + +#[test] +fn ambiguous_turn_start_that_never_created_a_turn_recovers_and_starts_once() { + let directory = cut_off_second_turn_start("durable-ambiguous-not-created", false); + let mut recovered = CodexCommandExecutor::new(&directory); + let snapshot = recovered + .execute(&command("snapshot", 5, "session.snapshot", json!({}))) + .expect("history proves the cut-off start never created a turn"); + assert_eq!(snapshot.result["status"], "session_open"); + assert!(snapshot.result["activeProviderTurnId"].is_null()); + let persisted: Value = serde_json::from_slice( + &fs::read(directory.join("codex-provider-state.json")).expect("read recovered state"), + ) + .expect("parse recovered state"); + assert_eq!(persisted["ambiguousTurnStartPending"], false); + assert_eq!(call_count(&directory, "thread/turns/list"), 1); + let started = recovered + .execute(&command( + "second-turn", + 6, + "turn.start", + json!({"text": "Start once."}), + )) + .expect("start the turn the cut-off start never created"); + assert_eq!(started.result["providerTurnId"], "provider-turn-2"); + wait_for_executor_event(&mut recovered, "turn.completed"); + assert_eq!(call_count(&directory, "turn/start"), 2); + recovered + .shutdown() + .expect("stop recovered provider process"); + fs::remove_dir_all(directory).expect("remove Codex integration-test directory"); +} + +#[test] +fn ambiguous_turn_start_that_created_an_unsettled_turn_still_fails_closed() { + let directory = cut_off_second_turn_start("durable-ambiguous-created", true); + let mut recovered = CodexCommandExecutor::new(&directory); + let error = recovered + .execute(&command("snapshot", 5, "session.snapshot", json!({}))) + .expect_err("a recorded turn Paperclip never settled may have done work"); + assert!( + error + .to_string() + .contains("cannot safely recover an ambiguous Codex turn start"), + "unexpected ambiguous recovery error: {error}" + ); + assert_eq!(call_count(&directory, "turn/start"), 1); + fs::remove_dir_all(directory).expect("remove Codex integration-test directory"); +} + #[test] fn ambiguous_replacement_completion_replaces_durable_turn_authority() { let directory = temporary_directory("durable-ambiguous-turn-completion"); diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport-cut-turn-start.test.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport-cut-turn-start.test.ts new file mode 100644 index 00000000000..ecadcd38f1a --- /dev/null +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport-cut-turn-start.test.ts @@ -0,0 +1,137 @@ +import { mkdtemp, readFile, rm, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join, resolve } from "node:path"; +import { expect, it, vi } from "vitest"; +import { createRunnerdNativeSessionBackend } from "../backends/codex-native-backend.js"; +import type { NativeExecutionInputV1 } from "../contracts/native-execution.js"; +import type { PrpEvent } from "../protocol/replay-contract.js"; +import { executeNativeSession } from "../native-session-runtime.js"; +import { createRunnerdCodexTransport, defaultCapabilityRunnerdBinary } from "./runnerd-codex-transport.js"; + +/** + * A restart can kill the runner and its Codex process while `turn/start` is in + * flight. The runner journaled the command and marked the start ambiguous, so + * the restored runner cannot know from its own state whether Codex created the + * turn, and the resumed thread has no active turn either way. + * + * - "before": Codex never created the turn. The resumed thread's history ends + * at a turn Paperclip already settled (here: no turn at all), which proves the + * start never happened. The same-run resume starts the turn exactly once. + * - "after": Codex recorded the turn, then died. Its history names a turn + * Paperclip never settled. The turn may have done work, so the runner still + * fails closed and never starts it again by itself. + */ +it.each(["before", "after"] as const)("resumes a run whose turn.start a restart cut off (provider stopped %s creating the turn)", async (stage) => { + const root = await mkdtemp(join(tmpdir(), `runnerd-cut-turn-start-${stage}-`)); + const callsPath = join(root, "calls.log"); + const hangPath = join(root, "hang-turn-start"); + const identity = { runId: "run-cut-start", sessionId: "session-cut-start", companyId: "company-cut-start", issueId: "issue-cut-start", agentId: "agent-cut-start" }; + const execution: NativeExecutionInputV1 = { + schema: "paperclip.native-execution-input.v1", + binding: { runId: identity.runId, companyId: identity.companyId, issueId: identity.issueId, agentId: identity.agentId, executionWorkspaceId: "workspace-cut-start" }, + session: { normalizedSessionId: identity.sessionId, driverKind: "codex_app_server", protocolVersion: 1, lifecyclePolicy: { mode: "per_turn", idleTimeoutMs: null } }, + task: { identifier: "TEST-7", title: "Write the brief", description: null, prompt: "Write the brief with the answer marker.", workMode: "standard" }, + workspace: { cwd: root, repoUrl: null, repoRef: null, branchName: null }, + provider: { kind: "codex", model: null }, + completionContract: { id: "contract-cut-start", sha256: "digest", schemaVersion: "paperclip.completion-contract.v1", contract: { + revision: "1", objective: "Write the brief", criteria: [{ id: "brief", requirement: "Write the brief" }], + } }, + interactionResponses: [], credentialBindings: [], + }; + const options = { + runnerBinary: defaultCapabilityRunnerdBinary(), + codexCommand: resolve(import.meta.dirname, "../../runner/target/debug/fake-codex-app-server"), + codexArgs: ["--state-file", join(root, "fake-state.json"), "--call-log", callsPath, "--durable-turn-ids", "--hang-turn-start", hangPath], + stateDirectory: root, + turnStartTimeoutMs: 8_000, + prpIdentity: { runnerInstanceId: "runner-cut-start", environmentLeaseId: "lease-cut-start", runId: identity.runId, normalizedSessionId: identity.sessionId, turnId: "turn-cut-start", itemId: "item-cut-start" }, + }; + const first = createRunnerdCodexTransport(options); + let restored: ReturnType | undefined; + const alive = (pid: number | null) => { + if (!pid) return false; + try { process.kill(pid, 0); return true; } catch { return false; } + }; + const kill = (pid: number | null) => { + if (!pid) return; + try { process.kill(pid, "SIGKILL"); } catch (error) { if ((error as NodeJS.ErrnoException).code !== "ESRCH") throw error; } + }; + const turnStarts = async () => ((await readFile(callsPath, "utf8")).match(/^turn\/start$/gm) ?? []).length; + try { + const backend = createRunnerdNativeSessionBackend(execution, { runnerInstanceId: "runner-cut-start", transportFactory: () => first.transport }); + const session = await backend.openSession({ identity, workingDirectory: root }); + // The last checkpoint the controller persisted: no active turn. + const checkpoint = await session.snapshot(); + expect(checkpoint.activeTurnId ?? null).toBeNull(); + await writeFile(hangPath, stage); + const starting = session.startTurn({ message: { role: "user", text: execution.task.prompt } }).catch((error: unknown) => error); + // The runner journaled turn.start, marked the start ambiguous and sent it. + await vi.waitFor(async () => { + const provider = JSON.parse(await readFile(join(root, "runner", "codex-provider-state.json"), "utf8")); + expect(provider.ambiguousTurnStartPending).toBe(true); + expect(await turnStarts()).toBe(1); + if (stage === "after") expect(JSON.parse(await readFile(join(root, "fake-state.json"), "utf8")).activeTurnId).toBe("provider-turn-1"); + }, { timeout: 8_000, interval: 25 }); + const owner = first.evidence(); + await first.detachControllerForRestart(); + kill(owner.runnerPid); + kill(owner.providerPid); + await vi.waitFor(() => { expect(alive(owner.runnerPid)).toBe(false); expect(alive(owner.providerPid)).toBe(false); }); + await starting; + await rm(hangPath, { force: true }); + // Codex's process-loss history: the thread remains, its in-memory turn does not. + const providerState = JSON.parse(await readFile(join(root, "fake-state.json"), "utf8")); + providerState.activeTurnId = null; + await writeFile(join(root, "fake-state.json"), JSON.stringify(providerState)); + + restored = createRunnerdCodexTransport({ ...options, resumeDynamicTools: [], resumeActiveTurnId: null }); + const recovering = createRunnerdNativeSessionBackend(execution, { runnerInstanceId: "runner-cut-start", transportFactory: () => restored!.transport }); + const events: PrpEvent[] = []; + const executing = executeNativeSession({ + input: execution, backend: recovering, persistedSession: checkpoint, resumeInterruptedTurn: true, + runnerInstanceId: "runner-cut-start", controlPlaneInstanceId: "controller-cut-start", timeoutMs: 15_000, + controlPlane: { + async openRun() {}, + async checkpointSession() {}, + async appendEvent(event) { if (!("sourceSeq" in event)) throw new Error("Expected a durable PRP event"); events.push(event); return { cursor: events.length, highestContiguousSourceSeq: event.sourceSeq, disposition: "committed" }; }, + async replayEvents(request) { + const source = events.filter(event => event.sourceInstanceId === request.sourceInstanceId); + return { events: source.filter(event => event.sourceSeq > request.afterSourceSeq), highestContiguousSourceSeq: source.at(-1)?.sourceSeq ?? 0 }; + }, + async completeRun() {}, + }, + }); + const control = async () => JSON.parse(await readFile(join(root, "control-plane", "control-plane-state.json"), "utf8")) as { + commands: Array<{ type: string; status: string; payload: { text?: string }; result?: { result?: { code?: string } } }>; + }; + if (stage === "before") { + const completion = await executing; + expect(completion).toMatchObject({ providerSessionId: checkpoint.providerSessionId, terminal: { runTerminalState: "succeeded" } }); + expect(events.find(event => event.eventType === "session.failed")).toBeUndefined(); + // The cut-off start stays settled as unknown and is never executed again; + // one new start carries the same task, and Codex saw exactly one more. + const starts = (await control()).commands.filter(command => command.type === "turn.start"); + expect(starts).toHaveLength(2); + expect(starts[0]).toMatchObject({ status: "failed", result: { result: { code: "execution_indeterminate" } } }); + expect(starts[1]).toMatchObject({ status: "completed" }); + expect(starts[1]!.payload.text).toContain(execution.task.prompt); + expect(await turnStarts()).toBe(2); + } else { + await expect(executing).rejects.toThrow(/cannot safely recover an ambiguous Codex turn start/); + // Possibly started work is never started again without a person. + expect(await turnStarts()).toBe(1); + const starts = (await control()).commands.filter(command => command.type === "turn.start"); + expect(starts).toHaveLength(1); + } + } finally { + await rm(hangPath, { force: true }); + await restored?.transport.close().catch(() => undefined); + for (const bundle of [first, restored]) { + if (!bundle) continue; + const evidence = bundle.evidence(); + kill(evidence.runnerPid); + kill(evidence.providerPid); + } + await rm(root, { recursive: true, force: true }); + } +}, 60_000); diff --git a/server/src/services/explicit-native-continuation.test.ts b/server/src/services/explicit-native-continuation.test.ts index 3df961605db..fb48824208a 100644 --- a/server/src/services/explicit-native-continuation.test.ts +++ b/server/src/services/explicit-native-continuation.test.ts @@ -94,6 +94,8 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes onBlocked: reason => blocked.push(reason) }); return { result, blocked, reads }; } + const PRIOR_OWNER_REFUSAL = "runner_state_identity_mismatch: prior_owner_terminal_state_indeterminate; " + + "retirement_refused=journal_commands_unsettled(turn.start.failed)"; it.each([ { shape: "refused by the cleanup quarantine before launch", errorCode: "native_session_cleanup_quarantined", launched: false, comment: "local_cleanup", retry: "cleanup_quarantined" }, @@ -109,6 +111,22 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes prepared: true, comment: "process_identity_missing", retry: "process_identity_missing" }, { shape: "failed without a process record after native selection", errorCode: "adapter_failed", launched: false, coordinator: true, comment: "process_identity_missing", retry: "process_identity_missing" }, + // w-20261110T033426-s108/t03: a restart cut off the previous run's turn + // start, so its saved runner state could not be proven finished. The next + // run was refused at that runner-state fence before it asked for a runner, + // and Retry then waited forever for a stop record of a process that never + // existed. + { shape: "was refused at the prior-owner runner-state fence before launch", errorCode: "adapter_failed", launched: false, + prepared: true, error: PRIOR_OWNER_REFUSAL, comment: null, retry: null }, + { shape: "was refused at the prior-owner runner-state fence with a coordinator", errorCode: "adapter_failed", launched: false, + prepared: true, coordinator: true, error: PRIOR_OWNER_REFUSAL, comment: null, retry: null }, + { shape: "was refused at the prior-owner runner-state fence after it requested a launch", errorCode: "adapter_failed", launched: true, + prepared: true, error: PRIOR_OWNER_REFUSAL, comment: "process_identity_missing", retry: "process_identity_missing" }, + { shape: "was refused at the prior-owner runner-state fence after a runner event", errorCode: "adapter_failed", launched: false, + runnerEvent: true, prepared: true, error: PRIOR_OWNER_REFUSAL, comment: "process_identity_missing", retry: "process_identity_missing" }, + { shape: "failed its own runner-state authority check", errorCode: "adapter_failed", launched: false, + prepared: true, error: "runner_state_identity_mismatch: authority_indeterminate", + comment: "process_identity_missing", retry: "process_identity_missing" }, { shape: "timed out before it requested a launch", status: "timed_out", errorCode: "timeout", launched: false, comment: null, retry: null }, { shape: "timed out after a runner event without a process record", status: "timed_out", errorCode: "timeout", launched: false, @@ -116,14 +134,14 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes { shape: "was interrupted without a process record after native selection", status: "interrupted", errorCode: "process_lost", launched: false, coordinator: true, comment: "process_identity_missing", retry: "process_identity_missing" }, ] as Array<{ shape: string; status?: "timed_out" | "interrupted"; errorCode: string; launched: boolean; runnerEvent?: boolean; prepared?: boolean; coordinator?: boolean; - comment: string | null; retry: string | null }>)("routes input after a run $shape to the gate that can clear it", async ({ - status, errorCode, launched, runnerEvent, prepared, coordinator, comment, retry }) => { + error?: string; comment: string | null; retry: string | null }>)("routes input after a run $shape to the gate that can clear it", async ({ + status, errorCode, launched, runnerEvent, prepared, coordinator, error, comment, retry }) => { // s7532/t04: a Retry's run failed at the quarantine fence before it asked // for a runner. Input then waited for a stop record of a process that // never existed ("process_identity_missing", Retry still offered) instead // of the quarantine's own gate, which names the action that clears it. const f = await seed(); - await db.update(heartbeatRuns).set({ processPid: null, errorCode, ...(status ? { status } : {}) }) + await db.update(heartbeatRuns).set({ processPid: null, errorCode, ...(status ? { status } : {}), ...(error ? { error } : {}) }) .where(eq(heartbeatRuns.id, f.sourceRunId)); // A run that failed before native selection has no native issue and no // coordinator. A coordinator row requires the run's native issue (foreign diff --git a/server/src/services/native-local-process-stop.ts b/server/src/services/native-local-process-stop.ts index 3715622b670..81c0bc48469 100644 --- a/server/src/services/native-local-process-stop.ts +++ b/server/src/services/native-local-process-stop.ts @@ -240,7 +240,11 @@ function record(value: unknown): Record { /** A settled native run that never had a runner has nothing to stop: * - an admission fence (a quarantined prior-session cleanup) refused it before - * it requested a launch; or + * it requested a launch; + * - the runner-state fence refused it before it requested a launch because a + * previous run still owns the conversation's saved runner state and that + * owner could not be proven finished (`prior_owner_*`). The root belonged + * to another run, so this run started nothing; or * - it failed before native selection (for example a refused budget * reservation or a failed preparation query). Selection records the run's * native issue, native session, runner instance and coordinator, and a @@ -248,12 +252,17 @@ function record(value: unknown): Record { * in any build. * Any launch request, identity record or runner event keeps the normal stop * proofs. */ +/** Thrown only by the executor's runner-state migration, before any launch, + * when the conversation's saved state names a different (prior) run. */ +const PRIOR_OWNER_FENCE_REFUSAL = /^runner_state_identity_mismatch: (?:legacy_)?prior_owner_/; + export async function nativeRunNeverLaunched(db: Db, run: Pick): Promise { + "nativeSessionId" | "runnerInstanceId" | "nativeIssueId" | "error">): Promise { if (run.runtimeMode !== "native" || !SETTLED_RUN_STATUSES.includes(run.status) || run.processPid || run.processGroupId) return false; - const quarantinedBeforeLaunch = run.status === "failed" && run.errorCode === "native_session_cleanup_quarantined"; - if (!quarantinedBeforeLaunch) { + const refusedBeforeLaunch = run.status === "failed" && (run.errorCode === "native_session_cleanup_quarantined" || + PRIOR_OWNER_FENCE_REFUSAL.test(run.error ?? "")); + if (!refusedBeforeLaunch) { // A coordinator row requires the run's native issue (foreign key), so the // native issue also rules out a coordinator. if (run.nativeSessionId || run.runnerInstanceId || run.nativeIssueId) return false;