mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-11 05:11:49 +01:00
fix(runner): continue restart-interrupted Codex turns (#15297)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Native runs retain their provider conversation across server restarts. > - A dead runner can restore a Codex conversation after its active turn is lost. > - The old recovery path synthesized a failed task result from that interruption. > - The task then required operator action even though its conversation and workspace were available. > - This pull request preserves the interruption cause and uses the admitted restart attempt for one continuation in the same conversation. > - The agent can reconcile unfinished actions and complete the current request without resending the original task. ## Linked Issues or Issue Description Related: #12845 added native restart recovery. #15042 covers admission during shutdown. #14796 covers legacy shutdown recovery. This change covers a lost native Codex turn after successful conversation restoration. **What happened?** After a server restart killed the local runner, Paperclip restored the saved Codex thread. Runnerd found that the old turn was no longer active. It synthesized a failed terminal and a needs-review result from the last progress message. The transport also discarded the terminal error when it reconstructed thread history. The task failed instead of continuing. **Expected behavior** After proving that the old process stopped and admitting a bounded recovery attempt, resume the current request in the same conversation. Preserve the workspace. Inspect unfinished actions before proceeding. Keep real provider failures, accepted results, intentional stops, unknown unreconciled effects, and exhausted attempts subject to their existing rules. **Steps to reproduce** 1. Start a local native Codex run and leave its turn active. 2. Kill the isolated runner and provider processes, as can happen during a server restart. 3. Restore the same provider thread with no active turn. 4. Observe the synthetic task failure. The new real-process regression reproduces this boundary with a scripted provider. ## What Changed - Record an explicit recoverable process-loss cause without inventing a task result. - Preserve terminal errors and prior turns in reconstructed provider history. Recover the authoritative saved result when adopting an accepted continuation. - Send one continuation in the same conversation for an admitted dead-runner recovery. Require reconciliation of unfinished commands and external actions. - Persist the interrupted terminal before submission and retain the existing recovery marker across another controller loss. - Keep provider attempt limits, terminal failures, and intentional cancellation behavior. - Add red/green regressions, real process-kill coverage, restart checkpoint coverage, and retry-budget coverage. Document the behavior and run-log evidence. ## Verification - Red: the new native runtime regression rejected with `NativeProviderTerminalFailure` on the original code; the Rust restore regression found a missing recovery cause. - Red/green: if restoring the conversation fails and replacement is allowed, the replacement receives the full task and fresh-session handoff. Both prepared and legacy execution inputs are covered. - Red: a second controller crash after the provider accepted the continuation caused an extra `turn/start`. The regression now proves there are exactly two submissions total: the original and its continuation. - Green: focused runtime, backend, driver recovery, and real-process restart suites (207 tests). After the final history/result changes, driver recovery and real-process restart suites passed again (36 tests). - Green: complete Codex transport suite (186 tests), server restart classification/database integration suites (34 tests), and Rust Codex provider suite (92 passed, 2 ignored). - Full `pnpm -r typecheck` and `pnpm build` passed on `60141e649`. The subsequent replacement-prompt guard passed the Runner TypeScript check and the complete runtime plus process-restart suites (148 tests). - Local full-suite attempt: `pnpm test:run` reported two failures in the untouched chat integration suite. Both passed individually, and the complete chat suite passed on rerun (1,063 tests). After all remote test shards passed, the duplicate serial local run was stopped with SIGINT; it is not claimed as a full local-suite pass. - Latest-head CI (`b1297dcd4`): 55 successful checks and 4 intentionally skipped checks, including all test shards, typecheck, build, native Runner verification, end-to-end tests, and canary dry run. [CI run](https://github.com/paperclipai/paperclip/actions/runs/37405346303). - Greptile: 5/5 on the latest head, with no unresolved review threads. The PR is mergeable. - The process tests use the real runner binary and a scripted Codex provider. They do not call a live model service. ## Risks - This changes local Codex recovery after process loss. A continuation can execute more work in the retained conversation. Its prompt requires state inspection before repeating an uncertain action; the system does not replay tool calls. - Recovery shares the existing three-attempt budget and one-shot continuation marker. Real failures and older unmarked failed checkpoints are not reopened. - No database migration or API change is required. ## Model Used - OpenAI Codex, based on GPT-6. The exact model ID and context-window size are not exposed in this session. - Capabilities: reasoning, source inspection, tool use, code editing, code execution, and test analysis. ## 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 (e.g. `docs/...`, `fix/...`) 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 --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip authored and GitHub committed
1 parent
a1ab55a56d
commit
63f3aa2dbf
18 files changed
+492
-19
No files matched your search
@@ -1000,6 +1000,19 @@ recording the proof and successor lineage. Retained provider files are not edite
|
||||
|
||||
Bootstrap retries, exact-checkpoint resumes, and fresh replacement sessions share three total provider attempts, including the original attempt. Linked run IDs, controller restarts, and duplicate wakes do not reset this budget. Automatic attempts retain the 30-second delay. Replacement scheduling and predecessor lineage commit together, with one successor per predecessor and admission through the normal task locks, authorization, pause, approval, and budget gates.
|
||||
|
||||
An exact local Codex restart can restore the conversation after losing its active
|
||||
turn. Runnerd records this as `provider_turn_lost_on_restore`, with no invented
|
||||
task result. The admitted restart attempt may send one continuation in that same
|
||||
conversation. It preserves the workspace and asks the agent to reconcile unfinished
|
||||
commands and external actions before proceeding. It does not resend the task or
|
||||
tool calls. An outcome that cannot be reconciled remains a blocker. The continuation
|
||||
uses the existing durable one-shot recovery marker; recovery adopts an accepted
|
||||
turn if the controller dies before checkpointing its ID. The interrupted terminal
|
||||
is checkpointed before submission. Real provider failures, accepted semantic
|
||||
results, intentional stops, and historical unmarked failures keep their existing
|
||||
terminal behavior. This continuation uses the already charged restart attempt and
|
||||
does not reset the provider-attempt budget.
|
||||
|
||||
Provider execution and control-plane finalization have different clocks. A healthy provider can think or execute a long tool without output. Once execution settles, recovery and finalization control steps have a 60-second deadline, checked on startup and every 15 seconds. With a healthy database and scheduler, an abandoned transition must be repaired or surfaced within 90 seconds. Terminal persistence must not wait on provider cleanup or publication; a late finalizer cannot change a reassigned or closed task or release another run's locks. Historical ambiguous runs are never automatically replayed after an upgrade.
|
||||
|
||||
Every continuation carries the triggering request, ordered user direction, interaction outcomes, completed work, and explicit history coverage. A delivered message remains part of the task's request after its connection or approval resolves. The original title is background; a completed Notion read does not satisfy a later Gmail request. Author and source-trust boundaries survive rendering into both native and legacy prompts. Missing required history must be fetched before dispatch rather than described as complete.
|
||||
|
||||
@@ -110,6 +110,14 @@ proofs, encryption keys, environment variables, provider credentials, command
|
||||
arguments, or an unsanitized stderr stream. Detailed failed-attempt diagnostics
|
||||
remain in the bounded `native_run_finalizations.recovery_history` ledger.
|
||||
|
||||
For a local Codex turn lost while runnerd was stopped, the canonical `turn.failed`
|
||||
event retains `error.code: "provider_turn_lost_on_restore"` and
|
||||
`error.recoverable: true`. It records the interrupted turn, not a final task
|
||||
outcome. There is no synthesized result for that interruption. An admitted
|
||||
same-session continuation emits its own `turn.submitted`, accepted turn ID, and
|
||||
terminal events; only the final outcome closes the run. Reconstructed thread
|
||||
history retains the original error so reconciliation and live delivery agree.
|
||||
|
||||
## Native Local Process Stop Evidence
|
||||
|
||||
The server writes `native.local_process_stopped` in the same transaction that
|
||||
|
||||
@@ -652,6 +652,9 @@ fn run() -> Result<(), Box<dyn std::error::Error>> {
|
||||
.iter()
|
||||
.any(|value| value == "--require-codex-home-auth");
|
||||
let durable_turn_ids = args.iter().any(|value| value == "--durable-turn-ids");
|
||||
let hold_first_durable_turn = args
|
||||
.iter()
|
||||
.any(|value| value == "--hold-first-durable-turn");
|
||||
let durable_tool_ids = args.iter().any(|value| value == "--durable-tool-ids");
|
||||
let expected_canonical_task_context_file =
|
||||
argument(&args, "--expected-canonical-task-context-file");
|
||||
@@ -1348,6 +1351,7 @@ fn run() -> Result<(), Box<dyn std::error::Error>> {
|
||||
.checked_add(1)
|
||||
.ok_or("fake provider turn sequence exhausted")?;
|
||||
}
|
||||
let hold_turn = hold_turn || (hold_first_durable_turn && state.next_turn == 1);
|
||||
if reject_second_turn_start && turn_start_count == 2 {
|
||||
send(json!({
|
||||
"method": "warning",
|
||||
|
||||
@@ -2103,17 +2103,29 @@ impl CodexCommandExecutor {
|
||||
"providerTurnId": previous_active_turn_id,
|
||||
"status": "failed",
|
||||
"providerTerminalObserved": false,
|
||||
"error": if provider_label == "codex" { json!({
|
||||
"code": "provider_turn_lost_on_restore",
|
||||
"recoverable": true,
|
||||
"message": "The runner restored the conversation after process loss, but the previous turn is no longer active.",
|
||||
}) } else { Value::Null },
|
||||
}),
|
||||
};
|
||||
let outcome = terminal_events(
|
||||
state,
|
||||
"turn.failed",
|
||||
state.goal.as_ref().map(|goal| goal.status.as_str()),
|
||||
);
|
||||
state.extend_terminal_events(with_terminal_outcome(
|
||||
vec![provider_terminal],
|
||||
outcome,
|
||||
))?;
|
||||
if provider_label == "codex" {
|
||||
// No provider result was observed. Preserve the lost
|
||||
// turn as a fact, but leave the run open for the
|
||||
// controller's bounded same-conversation recovery.
|
||||
state.push_terminal_event(provider_terminal)?;
|
||||
} else {
|
||||
let outcome = terminal_events(
|
||||
state,
|
||||
"turn.failed",
|
||||
state.goal.as_ref().map(|goal| goal.status.as_str()),
|
||||
);
|
||||
state.extend_terminal_events(with_terminal_outcome(
|
||||
vec![provider_terminal],
|
||||
outcome,
|
||||
))?;
|
||||
}
|
||||
}
|
||||
} else {
|
||||
state.push_event(reconciled)?;
|
||||
|
||||
@@ -3484,6 +3484,65 @@ fn durable_backend_replays_pending_tool_calls_without_mutating_the_event_queue()
|
||||
fs::remove_dir_all(directory).expect("remove Codex integration-test directory");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn restart_lost_turn_retains_a_recoverable_cause_without_inventing_a_result() {
|
||||
let directory = temporary_directory("restart-lost-turn");
|
||||
let config = provider_config(
|
||||
&directory,
|
||||
&["--durable-turn-ids", "--linger-after-turn-start"],
|
||||
);
|
||||
let mut first = CodexCommandExecutor::new(&directory);
|
||||
first
|
||||
.execute(&command(
|
||||
"prepare",
|
||||
1,
|
||||
"run.prepare",
|
||||
json!({"provider": config}),
|
||||
))
|
||||
.unwrap();
|
||||
first
|
||||
.execute(&command("open", 2, "session.open", json!({})))
|
||||
.unwrap();
|
||||
first
|
||||
.execute(&command(
|
||||
"start",
|
||||
3,
|
||||
"turn.start",
|
||||
json!({"text": "Watch the existing check run."}),
|
||||
))
|
||||
.unwrap();
|
||||
// The process disappeared while its command was in flight. Codex restores
|
||||
// the conversation, but cannot restore the running turn.
|
||||
drop(first);
|
||||
fs::write(
|
||||
directory.join("fake-state.json"),
|
||||
serde_json::to_vec(&json!({
|
||||
"threadId": "codex-thread-1", "activeTurnId": null, "nextTurn": 1,
|
||||
}))
|
||||
.unwrap(),
|
||||
)
|
||||
.unwrap();
|
||||
let mut restored = CodexCommandExecutor::new(&directory);
|
||||
restored
|
||||
.execute(&command("snapshot", 4, "session.snapshot", json!({})))
|
||||
.unwrap();
|
||||
let events = poll_and_ack(&mut restored).unwrap();
|
||||
let interrupted = events
|
||||
.iter()
|
||||
.find(|event| event.event_type == "turn.failed")
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
interrupted.payload["error"]["code"],
|
||||
"provider_turn_lost_on_restore"
|
||||
);
|
||||
assert_eq!(interrupted.payload["error"]["recoverable"], true);
|
||||
assert!(!events
|
||||
.iter()
|
||||
.any(|event| event.event_type == "run.result.proposed"));
|
||||
restored.shutdown().unwrap();
|
||||
fs::remove_dir_all(directory).unwrap();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn durable_backend_settles_pending_tools_when_recovery_finds_the_turn_ended() {
|
||||
let directory = temporary_directory("durable-tool-ended-offline");
|
||||
|
||||
@@ -673,6 +673,7 @@ class HarnessNativeSession implements NativeSession {
|
||||
// The provider has no active turn to interrupt yet. Do not launch work
|
||||
// whose events cancellation would suppress and leave the owner waiting.
|
||||
if (this.#explicitlyCancelled) throw new Error("native_session_cancelled");
|
||||
this.#terminal = null;
|
||||
try {
|
||||
const started = await this.#session.startTurn(input);
|
||||
this.#assertProtocolIntegrity();
|
||||
|
||||
@@ -63,7 +63,7 @@ export interface PersistedNativeSession {
|
||||
terminal?: PrpTerminalState | null;
|
||||
activeTurnId?: string | null;
|
||||
terminalTurns?: PersistedHarnessTurnTerminal[];
|
||||
/** Durable at-most-once marker for a resultless terminal recovery turn. */
|
||||
/** At-most-once marker for resultless disposition repair or restart continuation. */
|
||||
dispositionOnlyRecoveryConsumed?: boolean;
|
||||
dispositionOnlyRecoveryTurnId?: string | null;
|
||||
pendingRuntimeRequests?: HarnessRuntimeRequest[];
|
||||
@@ -77,6 +77,31 @@ export interface NativeSessionRecoveryResult {
|
||||
reason?: string;
|
||||
}
|
||||
|
||||
/** Runner-owned cause: the process was restored, but its active turn was lost. */
|
||||
export const NATIVE_RESTART_INTERRUPTION_CODE = "provider_turn_lost_on_restore";
|
||||
|
||||
export function isNativeRestartInterruption(error: unknown): boolean {
|
||||
if (!error || typeof error !== "object" || Array.isArray(error)) return false;
|
||||
const failure = error as Record<string, unknown>;
|
||||
return failure.code === NATIVE_RESTART_INTERRUPTION_CODE && failure.recoverable === true;
|
||||
}
|
||||
|
||||
/** The terminal fingerprint survives a crash between interruption and continuation. */
|
||||
export function nativeRestartInterruptedTurnId(snapshot: Pick<PersistedNativeSession,
|
||||
"driverKind" | "activeTurnId" | "semanticResult" | "terminalTurns" | "dispositionOnlyRecoveryConsumed" | "pendingRuntimeRequests" | "goal"
|
||||
>): string | null {
|
||||
if (snapshot.driverKind !== "codex_app_server" || snapshot.activeTurnId != null ||
|
||||
snapshot.semanticResult != null || snapshot.dispositionOnlyRecoveryConsumed ||
|
||||
snapshot.goal != null || snapshot.pendingRuntimeRequests?.length) return null;
|
||||
const terminal = Array.isArray(snapshot.terminalTurns) ? snapshot.terminalTurns.at(-1) : null;
|
||||
if (!terminal || typeof terminal.fingerprint !== "string" || !terminal.turnId) return null;
|
||||
try {
|
||||
const fingerprint = JSON.parse(terminal.fingerprint);
|
||||
return fingerprint?.terminalState === "failed" && fingerprint.result === null &&
|
||||
isNativeRestartInterruption(fingerprint.error) ? terminal.turnId : null;
|
||||
} catch { return null; }
|
||||
}
|
||||
|
||||
/**
|
||||
* Cancellation is a synchronous authority transition followed by passive
|
||||
* provider cleanup. Once `cancel` returns, the provider session must no longer
|
||||
|
||||
@@ -17,7 +17,7 @@ import type {
|
||||
PersistedHarnessSession,
|
||||
PersistedHarnessTurnTerminal,
|
||||
} from "../../contracts/harness-driver.js";
|
||||
import { NativeSessionProtocolIntegrityError } from "../../contracts/native-session-backend.js";
|
||||
import { NativeSessionProtocolIntegrityError, nativeRestartInterruptedTurnId } from "../../contracts/native-session-backend.js";
|
||||
import { HarnessReconciliationError } from "../../contracts/harness-driver.js";
|
||||
import {
|
||||
CODEX_CODEX_PROTOCOL_VERSION,
|
||||
@@ -528,6 +528,12 @@ export class CodexAppServerDriver implements HarnessDriver {
|
||||
turns.forEach((turn, index) => {
|
||||
if (terminalIds.has(text(turn.id))) lastKnownTerminalIndex = index;
|
||||
});
|
||||
if (nativeRestartInterruptedTurnId({
|
||||
...snapshot, semanticResult: null,
|
||||
}) && (!providerHistoryIsArray || lastKnownTerminalIndex < 0)) {
|
||||
await cancellation.wait(cancellation.close());
|
||||
return { recovered: false, reason: "restart interruption history is incomplete" };
|
||||
}
|
||||
// Releasing a consumed marker requires both an actual history array
|
||||
// and a checkpointed terminal that anchors its ordering. An array that
|
||||
// omits every durable terminal may be truncated or inconsistent, so
|
||||
|
||||
@@ -45,6 +45,29 @@ import {
|
||||
import { NativeSessionProtocolIntegrityError } from "../../contracts/native-session-backend.js";
|
||||
|
||||
describe("Codex app-server Codex driver", () => {
|
||||
it("does not resend a restart continuation when its history anchor is missing", async () => {
|
||||
const first = new FakeCodexTransport();
|
||||
const second = new FakeCodexTransport();
|
||||
const driver = makeDriver([first, second]);
|
||||
const original = await driver.openSession({
|
||||
runId: "run-restart-anchor", normalizedSessionId: "session-restart-anchor", workingDirectory: WORKSPACE,
|
||||
});
|
||||
await original.startTurn({ message: { role: "user", text: "Finish the request." } });
|
||||
first.push("turn/completed", { threadId: "thread-1", turn: {
|
||||
id: "turn-1", status: "failed", error: { code: "provider_turn_lost_on_restore", recoverable: true }, items: [],
|
||||
} });
|
||||
await collectUntilTerminal(original.events());
|
||||
const snapshot = await original.snapshot();
|
||||
await original.close({ reason: "controller lost before continuation checkpoint" });
|
||||
second.readResponse = { thread: { id: "thread-1", sessionId: "provider-session-1", cwd: WORKSPACE,
|
||||
turns: [{ id: "uncheckpointed-turn", status: "completed", items: [] }],
|
||||
} };
|
||||
await expect(driver.recoverSession?.(snapshot)).resolves.toMatchObject({
|
||||
recovered: false, reason: "restart interruption history is incomplete",
|
||||
});
|
||||
expect(second.calls.some(call => call.method === "turn/start")).toBe(false);
|
||||
});
|
||||
|
||||
it.each([null, "checkpointed-prior-turn"])("recovers an autonomous goal turn beyond checkpoint %s", async (checkpointTurnId) => {
|
||||
const first = new FakeCodexTransport();
|
||||
const second = new FakeCodexTransport();
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
import { HarnessReconciliationError } from "../../contracts/harness-driver.js";
|
||||
import { isNativeRestartInterruption } from "../../contracts/native-session-backend.js";
|
||||
import type { PrpStructuredRunResult } from "../../protocol/replay-contract.js";
|
||||
import { boundedCodexPayload as boundedPayload, isRetainableCodexPayload } from "./codex-boundaries.js";
|
||||
import type { CodexSessionState } from "./codex-session-state.js";
|
||||
@@ -168,6 +169,13 @@ export function mapTerminalTurn(
|
||||
{ turnId },
|
||||
);
|
||||
finalize(state, status);
|
||||
if (status === "failed" && isNativeRestartInterruption(turn.error) && state.result === null &&
|
||||
!state.dispositionOnlyRecoveryConsumed) {
|
||||
// Orchestration may use the paid restart attempt for one same-thread
|
||||
// continuation. Its accepted turn id uses the existing durable marker.
|
||||
state.terminal = false;
|
||||
state.dispositionOnlyRecoveryAvailable = true;
|
||||
}
|
||||
}
|
||||
|
||||
function terminalFingerprint(
|
||||
|
||||
@@ -3581,20 +3581,31 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
||||
if (method === "thread/turns/list") {
|
||||
data = turns.map(turn => ({ ...turn, items: [], itemsView: "notLoaded" }));
|
||||
} else {
|
||||
if (params.turnId !== this.#turnId) throw new Error("codex_history_unavailable: requested turn is outside the retained runner event window");
|
||||
if (!turns.some(turn => turn.id === params.turnId)) throw new Error("codex_history_unavailable: requested turn is outside the retained runner event window");
|
||||
const items = new Map<string, Record<string, unknown>>();
|
||||
let semanticResultItem: Record<string, unknown> | null = null;
|
||||
let observedTurn = "";
|
||||
let observedStart = false;
|
||||
for (const event of this.#core?.store.state.committedEvents ?? []) {
|
||||
if (event.envelope.runId !== this.#core?.store.state.identity.runId) continue;
|
||||
const payload = record(record(event.envelope.payload).payload);
|
||||
if (event.eventType === "turn.started") observedTurn = String(payload.providerTurnId ?? payload.turnId ?? record(payload.turn).id ?? "");
|
||||
if (event.eventType === "turn.started" && observedTurn === params.turnId) observedStart = true;
|
||||
if (event.eventType === "run.result.proposed" && observedTurn === params.turnId) {
|
||||
// Reconciliation can precede notification delivery. Recover the
|
||||
// runner's authoritative result with the exact retained turn,
|
||||
// rather than launching work again just to obtain a disposition.
|
||||
const id = `runner-result-${event.sourceSeq}`;
|
||||
semanticResultItem = { turnId: observedTurn, item: { id, type: "agentMessage", text: JSON.stringify(payload) } };
|
||||
}
|
||||
if (event.eventType !== "item.completed" || observedTurn !== params.turnId) continue;
|
||||
const item = record(rehydrateRunnerdItemNotification(payload, this.#threadId, observedTurn).item);
|
||||
if (typeof item.id === "string") items.set(item.id, { turnId: observedTurn, item });
|
||||
}
|
||||
if (!observedStart) throw new Error("codex_history_incomplete: requested turn start is outside the retained runner event window");
|
||||
data = [...items.values()];
|
||||
// Runner authority wins over schema-shaped prose in an assistant item.
|
||||
if (semanticResultItem) data.push(semanticResultItem);
|
||||
}
|
||||
if (params.sortDirection === "desc") data.reverse();
|
||||
const offset = params.cursor == null ? 0 : Number(params.cursor);
|
||||
@@ -3655,8 +3666,12 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
||||
await new Promise((resolveWait) => setTimeout(resolveWait, 10));
|
||||
}
|
||||
if (terminal !== undefined) {
|
||||
const payload = record(record(terminal.envelope.payload).payload);
|
||||
recoveredTurns.push({
|
||||
id: this.#turnId,
|
||||
// Reconciliation must retain the cause, including the runner's
|
||||
// explicit process-loss marker, rather than inventing error:null.
|
||||
error: payload.error ?? record(payload.turn).error ?? null,
|
||||
status:
|
||||
terminal.eventType === "turn.completed"
|
||||
? "completed"
|
||||
@@ -3668,6 +3683,24 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
||||
});
|
||||
}
|
||||
}
|
||||
// A controller can lose the checkpoint after a continuation is accepted.
|
||||
// Keep its prior terminal as the history anchor, so driver recovery can
|
||||
// adopt the later accepted turn instead of submitting it again. Items
|
||||
// remain lazy and fail closed if their start left the retained window.
|
||||
const priorTerminals = new Map<string, Record<string, unknown>>();
|
||||
for (const event of this.#core?.store.state.committedEvents ?? []) {
|
||||
if (event.envelope.runId !== this.#core?.store.state.identity.runId ||
|
||||
!["turn.completed", "turn.failed", "turn.interrupted", "turn.cancelled"].includes(event.eventType)) continue;
|
||||
const payload = record(record(event.envelope.payload).payload);
|
||||
const turnId = payload.providerTurnId ?? payload.turnId ?? record(payload.turn).id;
|
||||
if (typeof turnId !== "string" || !turnId || turnId === this.#turnId) continue;
|
||||
priorTerminals.set(turnId, {
|
||||
id: turnId,
|
||||
status: event.eventType.slice("turn.".length),
|
||||
error: payload.error ?? record(payload.turn).error ?? null,
|
||||
});
|
||||
}
|
||||
recoveredTurns.unshift(...priorTerminals.values());
|
||||
this.#recoveryTurnBindingPending = false;
|
||||
this.#pumpEvents();
|
||||
return {
|
||||
|
||||
@@ -0,0 +1,125 @@
|
||||
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 { PersistedNativeSession } from "../contracts/native-session-backend.js";
|
||||
import type { PrpEvent } from "../protocol/replay-contract.js";
|
||||
import { executeNativeSession } from "../native-session-runtime.js";
|
||||
import { createRunnerdCodexTransport, defaultCapabilityRunnerdBinary } from "./runnerd-codex-transport.js";
|
||||
|
||||
it.each(["none", "before-start", "after-start"] as const)("completes after the real runner and provider are killed (second crash: %s)", async (crash) => {
|
||||
const root = await mkdtemp(join(tmpdir(), "runner-restart-continuation-"));
|
||||
const identity = { runId: "run-restart", sessionId: "session-restart", companyId: "company-restart", issueId: "issue-restart", agentId: "agent-restart" };
|
||||
const execution: NativeExecutionInputV1 = {
|
||||
schema: "paperclip.native-execution-input.v1",
|
||||
binding: { runId: identity.runId, companyId: identity.companyId, issueId: identity.issueId, agentId: identity.agentId, executionWorkspaceId: "workspace-restart" },
|
||||
session: { normalizedSessionId: identity.sessionId, driverKind: "codex_app_server", protocolVersion: 1 },
|
||||
task: { identifier: "TEST-1", title: "Finish checks", description: null, prompt: "Watch the already running checks and finish the request.", workMode: "standard" },
|
||||
workspace: { cwd: root, repoUrl: null, repoRef: null, branchName: null },
|
||||
provider: { kind: "codex", model: null },
|
||||
completionContract: { id: "contract-restart", sha256: "digest", schemaVersion: "paperclip.completion-contract.v1", contract: {
|
||||
revision: "1", objective: "Finish checks", criteria: [{ id: "checks", requirement: "Check the outcome" }],
|
||||
} },
|
||||
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", join(root, "calls.log"), "--durable-turn-ids", "--hold-first-durable-turn"],
|
||||
stateDirectory: root,
|
||||
prpIdentity: { runnerInstanceId: "runner-restart", environmentLeaseId: "lease-restart", runId: identity.runId, normalizedSessionId: identity.sessionId, turnId: "turn-restart", itemId: "item-restart" },
|
||||
};
|
||||
const first = createRunnerdCodexTransport(options);
|
||||
let restored: ReturnType<typeof createRunnerdCodexTransport> | 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; }
|
||||
};
|
||||
try {
|
||||
const backend = createRunnerdNativeSessionBackend(execution, { runnerInstanceId: "runner-restart", transportFactory: () => first.transport });
|
||||
const session = await backend.openSession({ identity, workingDirectory: root });
|
||||
await session.startTurn({ message: { role: "user", text: execution.task.prompt } });
|
||||
const persisted = await session.snapshot();
|
||||
expect(persisted.activeTurnId).toBe("provider-turn-1");
|
||||
const durableEvents: PrpEvent[] = [];
|
||||
const initialEvents = session.events()[Symbol.asyncIterator]();
|
||||
while ((durableEvents.at(-1)?.sourceSeq ?? 0) < Number(persisted.cursor)) {
|
||||
const event = await initialEvents.next();
|
||||
if (event.done) throw new Error("initial event history ended early");
|
||||
durableEvents.push(event.value);
|
||||
}
|
||||
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); });
|
||||
// Simulate Codex's process-loss history: the thread remains, its in-memory
|
||||
// turn does not. Keep the turn counter to prevent provider identity reuse.
|
||||
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: persisted.activeTurnId });
|
||||
let recovering = createRunnerdNativeSessionBackend(execution, { runnerInstanceId: "runner-restart", transportFactory: () => restored!.transport });
|
||||
const events: PrpEvent[] = [];
|
||||
const checkpoints: PersistedNativeSession[] = [];
|
||||
let recoveryCheckpoint = persisted;
|
||||
let failCheckpoint = crash !== "none";
|
||||
const interruptedController = new Error("controller lost before continuation submission");
|
||||
const execute = () => executeNativeSession({
|
||||
input: execution, backend: recovering, persistedSession: recoveryCheckpoint, resumeInterruptedTurn: true,
|
||||
runnerInstanceId: "runner-restart", controlPlaneInstanceId: "controller-restart", timeoutMs: 15_000,
|
||||
controlPlane: {
|
||||
async openRun() {},
|
||||
async checkpointSession(snapshot) {
|
||||
if (failCheckpoint && crash === "after-start" && snapshot.dispositionOnlyRecoveryTurnId === "provider-turn-2") {
|
||||
failCheckpoint = false;
|
||||
throw interruptedController;
|
||||
}
|
||||
checkpoints.push(structuredClone(snapshot));
|
||||
recoveryCheckpoint = structuredClone(snapshot);
|
||||
if (failCheckpoint && crash === "before-start" && snapshot.terminal?.runTerminalState === "failed") {
|
||||
failCheckpoint = false;
|
||||
throw interruptedController;
|
||||
}
|
||||
},
|
||||
async appendEvent(event) { events.push(event); durableEvents.push(event); return { cursor: durableEvents.length, highestContiguousSourceSeq: event.sourceSeq, disposition: "committed" }; },
|
||||
async replayEvents(request) {
|
||||
const source = durableEvents.filter(event => event.sourceInstanceId === request.sourceInstanceId);
|
||||
return { events: source.filter(event => event.sourceSeq > request.afterSourceSeq), highestContiguousSourceSeq: source.at(-1)?.sourceSeq ?? 0 };
|
||||
},
|
||||
async completeRun() {},
|
||||
},
|
||||
});
|
||||
if (crash !== "none") {
|
||||
await expect(execute()).rejects.toBe(interruptedController);
|
||||
await restored.transport.close().catch(() => undefined);
|
||||
restored = createRunnerdCodexTransport({ ...options, resumeDynamicTools: [], resumeActiveTurnId: recoveryCheckpoint.activeTurnId });
|
||||
recovering = createRunnerdNativeSessionBackend(execution, { runnerInstanceId: "runner-restart", transportFactory: () => restored!.transport });
|
||||
}
|
||||
const completion = await execute();
|
||||
expect(completion).toMatchObject({ providerSessionId: persisted.providerSessionId, turnId: "provider-turn-2", terminal: { runTerminalState: "succeeded" } });
|
||||
expect(events.filter(event => event.eventType === "turn.failed")).toHaveLength(1);
|
||||
expect(events.filter(event => event.eventType === "run.terminal")).toHaveLength(1);
|
||||
if (crash !== "after-start") expect(checkpoints.some(snapshot => snapshot.activeTurnId === "provider-turn-2" && snapshot.terminal === null)).toBe(true);
|
||||
expect((await readFile(join(root, "calls.log"), "utf8")).match(/^turn\/start$/gm)).toHaveLength(2);
|
||||
const journal = JSON.parse(await readFile(join(root, "control-plane/control-plane-state.json"), "utf8"));
|
||||
const submissions = journal.commands.filter((command: { type: string }) => command.type === "turn.start");
|
||||
expect(submissions).toHaveLength(2);
|
||||
expect(submissions[1].payload.text).toContain("Reconcile any unfinished tool or command");
|
||||
} finally {
|
||||
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 });
|
||||
}
|
||||
}, 30_000);
|
||||
@@ -225,6 +225,71 @@ function highestContiguous(events: PrpEvent[]): number {
|
||||
}
|
||||
|
||||
describe("executeNativeSession recovery", () => {
|
||||
it.each([
|
||||
{ label: "continues a restart-lost Codex turn", allowed: true, error: { code: "provider_turn_lost_on_restore", recoverable: true } },
|
||||
{ label: "requires the control plane restart claim", allowed: false, error: { code: "provider_turn_lost_on_restore", recoverable: true } },
|
||||
{ label: "preserves real provider errors", allowed: true, error: { code: "model_not_found", recoverable: false } },
|
||||
{ label: "does not reopen historical error-null failures", allowed: true, error: null },
|
||||
])("$label without replaying the task", async ({ label, allowed, error }) => {
|
||||
const checkpoint: PersistedNativeSession = {
|
||||
backendKind: "runner", driverKind: "codex_app_server", sessionId: "thread-recovery", identity,
|
||||
providerSessionId: "provider-recovery", cursor: "0", activeTurnId: "turn-recovery",
|
||||
terminalTurns: [], semanticResult: null, pendingRuntimeRequests: [], lineage: [],
|
||||
};
|
||||
const failure = error;
|
||||
const interrupted: PersistedNativeSession = {
|
||||
...checkpoint, activeTurnId: null,
|
||||
terminalTurns: [{ turnId: "turn-recovery", fingerprint: JSON.stringify({ terminalState: "failed", error: failure, result: null }) }],
|
||||
};
|
||||
const startTurn = vi.fn<NativeSession["startTurn"]>(async () => ({ turnId: "turn-continuation" }));
|
||||
const session: NativeSession = {
|
||||
identity: () => identity,
|
||||
async capabilities() { return { resume: true, typedEvents: true, steering: false, interruption: true, structuredResult: true }; },
|
||||
async *events() {
|
||||
yield runnerEvent(1, "turn.failed", { status: "failed", error: failure });
|
||||
if (!startTurn.mock.calls.length) return;
|
||||
yield { ...runnerEvent(2, "turn.completed", { status: "completed" }), turnId: "turn-continuation" };
|
||||
},
|
||||
startTurn,
|
||||
async result() { return startTurn.mock.calls.length ? { result, terminal, turnId: "turn-continuation" } : null; },
|
||||
async snapshot() { return structuredClone(interrupted); },
|
||||
async close() {},
|
||||
};
|
||||
const events: PrpEvent[] = [];
|
||||
const checkpoints: PersistedNativeSession[] = [];
|
||||
const backend: NativeSessionBackend = {
|
||||
async descriptor() { return { kind: "runner", name: "restart", version: "1", capabilities: await session.capabilities() }; },
|
||||
async openSession() { throw new Error("must preserve the provider session"); },
|
||||
async recoverSession() { return { recovered: true, session }; },
|
||||
};
|
||||
const port: ControlPlanePort = {
|
||||
async openRun() {},
|
||||
async checkpointSession(snapshot) { checkpoints.push(structuredClone(snapshot)); },
|
||||
async appendEvent(event) { events.push(event); return { cursor: events.length, highestContiguousSourceSeq: event.sourceSeq, disposition: "committed" }; },
|
||||
async replayEvents() { return { events: [], highestContiguousSourceSeq: 0 }; },
|
||||
async completeRun() {},
|
||||
};
|
||||
const execution = executeNativeSession({
|
||||
input, backend, controlPlane: port, persistedSession: checkpoint,
|
||||
runnerInstanceId: "runner-recovery", controlPlaneInstanceId: "control-recovery",
|
||||
resumeInterruptedTurn: allowed,
|
||||
});
|
||||
if (label !== "continues a restart-lost Codex turn") {
|
||||
await expect(execution).rejects.toMatchObject({ code: "native_provider_terminal_failed" });
|
||||
expect(startTurn).not.toHaveBeenCalled();
|
||||
return;
|
||||
}
|
||||
await expect(execution).resolves.toMatchObject({ result, turnId: "turn-continuation", terminal: { runTerminalState: "succeeded" } });
|
||||
expect(startTurn).toHaveBeenCalledOnce();
|
||||
const submitted = startTurn.mock.calls[0]![0] as Parameters<NativeSession["startTurn"]>[0];
|
||||
expect(submitted.continuation).toBe(true);
|
||||
expect(submitted.message.text).toContain("Reconcile any unfinished tool or command");
|
||||
expect(submitted.message.text).not.toContain(input.task.prompt);
|
||||
expect(events.filter(event => event.eventType === "turn.failed")).toHaveLength(1);
|
||||
expect(events.filter(event => event.eventType === "run.terminal")).toHaveLength(1);
|
||||
expect(checkpoints.some(snapshot => snapshot.terminalTurns?.some(turn => turn.turnId === "turn-recovery"))).toBe(true);
|
||||
});
|
||||
|
||||
it.each([undefined, 0, 7 * 24 * 60 * 60 * 1000, 30 * 24 * 60 * 60 * 1000])(
|
||||
"honors long-lived turn duration independently of operation bounds (%s)",
|
||||
async (turnTimeoutMs) => {
|
||||
@@ -6234,7 +6299,12 @@ describe("executeNativeSession recovery", () => {
|
||||
expect(openRun).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it.each([false, true])("replaces a provider session that already ended with a failed terminal (prepared: %s)", async (preparedMode) => {
|
||||
it.each([
|
||||
{ preparedMode: false, interrupted: false },
|
||||
{ preparedMode: true, interrupted: false },
|
||||
{ preparedMode: false, interrupted: true },
|
||||
{ preparedMode: true, interrupted: true },
|
||||
])("replaces a failed provider with full task context (prepared: $preparedMode, interrupted: $interrupted)", async ({ preparedMode, interrupted }) => {
|
||||
const digest = "0".repeat(64);
|
||||
const skill = {
|
||||
key: "company/recovery-skill",
|
||||
@@ -6286,6 +6356,7 @@ describe("executeNativeSession recovery", () => {
|
||||
} as const;
|
||||
const checkpoint: PersistedNativeSession = {
|
||||
backendKind: "mock",
|
||||
driverKind: "codex_app_server",
|
||||
sessionId: "driver-failed",
|
||||
identity,
|
||||
providerSessionId: "provider-failed",
|
||||
@@ -6299,7 +6370,10 @@ describe("executeNativeSession recovery", () => {
|
||||
runTerminalState: "failed",
|
||||
reportedWorkDisposition: "yielded",
|
||||
},
|
||||
terminalTurns: [{ turnId: "turn-failed", fingerprint: "failed" }],
|
||||
terminalTurns: [{ turnId: "turn-failed", fingerprint: interrupted ? JSON.stringify({
|
||||
terminalState: "failed", result: null,
|
||||
error: { code: "provider_turn_lost_on_restore", recoverable: true },
|
||||
}) : "failed" }],
|
||||
pendingRuntimeRequests: [],
|
||||
lineage: [],
|
||||
};
|
||||
@@ -6335,8 +6409,8 @@ describe("executeNativeSession recovery", () => {
|
||||
async close() {},
|
||||
};
|
||||
const recoverSession = vi.fn(async () => ({
|
||||
recovered: true as const,
|
||||
session: replacementSession,
|
||||
recovered: false as const,
|
||||
reason: "provider thread is unavailable",
|
||||
}));
|
||||
const openReplacementSession = vi.fn(async () => replacementSession);
|
||||
const onContinuityBreak = vi.fn(async () => undefined);
|
||||
@@ -6397,7 +6471,7 @@ describe("executeNativeSession recovery", () => {
|
||||
).resolves.toMatchObject({ providerSessionId: "provider-replacement" });
|
||||
|
||||
expect(getFreshSessionHandoff).toHaveBeenCalledOnce();
|
||||
expect(recoverSession).not.toHaveBeenCalled();
|
||||
expect(recoverSession).toHaveBeenCalledTimes(interrupted ? 1 : 0);
|
||||
expect(openReplacementSession).toHaveBeenCalledOnce();
|
||||
const replacementEnvelope = JSON.parse(
|
||||
startTurn.mock.calls[0]![0].message.text,
|
||||
@@ -6419,7 +6493,7 @@ describe("executeNativeSession recovery", () => {
|
||||
expect(JSON.stringify(replacementEnvelope)).not.toContain("ONLY_NEW_COMMENT");
|
||||
expect(startTurn.mock.calls[0]![0]).not.toHaveProperty("continuation");
|
||||
expect(onContinuityBreak).toHaveBeenCalledWith({
|
||||
reason: "provider session ended with a failed terminal",
|
||||
reason: interrupted ? "provider thread is unavailable" : "provider session ended with a failed terminal",
|
||||
previousDriverSessionId: "driver-failed",
|
||||
previousProviderSessionId: "provider-failed",
|
||||
replacementDriverSessionId: "driver-replacement",
|
||||
|
||||
@@ -23,6 +23,8 @@ import {
|
||||
NativeSessionCloseUnrecoverableError,
|
||||
NativeSessionCleanupQuarantinedError,
|
||||
NativeSessionProtocolIntegrityError,
|
||||
isNativeRestartInterruption,
|
||||
nativeRestartInterruptedTurnId,
|
||||
} from "./contracts/native-session-backend.js";
|
||||
import {
|
||||
validatePrpStructuredRunResult,
|
||||
@@ -184,6 +186,8 @@ export interface NativeSessionGoalControl {
|
||||
}
|
||||
|
||||
export interface ExecuteNativeSessionOptions {
|
||||
/** The control plane proved the old local runner stopped and charged a recovery attempt. */
|
||||
resumeInterruptedTurn?: boolean;
|
||||
/** Read bounded task history only after recovery requires a fresh conversation. */
|
||||
getFreshSessionHandoff?: () => Promise<string | null>;
|
||||
/** Durable launch intent, after cleanup admission and before provider calls. */
|
||||
@@ -865,6 +869,7 @@ async function consumeTurn(
|
||||
requestId: string;
|
||||
},
|
||||
initialGoal?: HarnessThreadGoal | null,
|
||||
resumeInterruptedTurn?: (event: PrpEvent, signal: AbortSignal) => Promise<boolean>,
|
||||
) {
|
||||
let timer: ReturnType<typeof setTimeout> | undefined;
|
||||
const appendAbort = new AbortController();
|
||||
@@ -1174,6 +1179,8 @@ async function consumeTurn(
|
||||
}
|
||||
if (isTurnTerminal(event)) {
|
||||
if (providerFailure) throw providerFailure;
|
||||
if (semanticResultProposal === null && governedResult === null &&
|
||||
inputTimers.size === 0 && await resumeInterruptedTurn?.(event, appendAbort.signal)) continue;
|
||||
return {
|
||||
event,
|
||||
eventCount,
|
||||
@@ -1871,6 +1878,12 @@ export async function executeNativeSession(
|
||||
previousProviderSessionId: string | null;
|
||||
} | null = null;
|
||||
let reconciledRecoveryCheckpoint: PersistedNativeSession | null = null;
|
||||
const checkpointedInterruption = persistedSession && nativeRestartInterruptedTurnId(persistedSession);
|
||||
if (checkpointedInterruption && persistedSession) {
|
||||
// The turn failed because its process vanished; the conversation is still
|
||||
// usable. Keep its terminal fingerprint so replay cannot execute it twice.
|
||||
persistedSession = { ...persistedSession, terminal: null };
|
||||
}
|
||||
await options.onSessionAdmission?.();
|
||||
if (options.existingSession) {
|
||||
if (options.existingSession.attachRun === undefined) {
|
||||
@@ -2157,6 +2170,42 @@ export async function executeNativeSession(
|
||||
await persistCheckpoint(snapshot, signal);
|
||||
};
|
||||
const recoveredSnapshot = await session.snapshot();
|
||||
const restartContinuation = () => ({
|
||||
message: { role: "user" as const, text: JSON.stringify({
|
||||
schema: "paperclip.native-continuation.v1",
|
||||
events: [
|
||||
"Paperclip restarted and restored this same conversation. The previous turn was interrupted.",
|
||||
"Continue the current user request from the recorded progress and preserved workspace.",
|
||||
"Reconcile any unfinished tool or command before proceeding: inspect its current state and recorded outcomes.",
|
||||
"Do not repeat completed work or blindly rerun an action whose outcome is unknown. If an external action cannot be reconciled, report the specific blocker.",
|
||||
].join("\n"),
|
||||
completion: {
|
||||
revision: input.completionContract.contract.revision,
|
||||
criterionIds: input.completionContract.contract.criteria.map(criterion => criterion.id),
|
||||
},
|
||||
}) },
|
||||
continuation: true as const,
|
||||
requestedCollaborationMode: "executionMode" in input ? input.executionMode : "default" as const,
|
||||
});
|
||||
let restartContinuationStarted = false;
|
||||
const resumeInterruptedTurn = async (event: PrpEvent, signal: AbortSignal) => {
|
||||
if (restartContinuationStarted || !recovered || !options.resumeInterruptedTurn ||
|
||||
input.provider.kind !== "codex" || options.sessionGoalControl || options.resumeSessionGoalHeartbeat ||
|
||||
recoveredSnapshot.goal || persistedSession?.semanticResult || persistedSession?.pendingRuntimeRequests?.length ||
|
||||
event.eventType !== "turn.failed" || !isNativeRestartInterruption(event.payload.error) ||
|
||||
event.turnId !== persistedSession?.activeTurnId) return false;
|
||||
const snapshot = await session.snapshot();
|
||||
signal.throwIfAborted();
|
||||
if (nativeRestartInterruptedTurnId(snapshot) !== event.turnId || snapshot.pendingRuntimeRequests?.length) return false;
|
||||
restartContinuationStarted = true;
|
||||
// Persist the old terminal before launching. Recovery inspects provider
|
||||
// history to adopt an accepted continuation if its checkpoint was lost.
|
||||
await persistCheckpoint(snapshot, signal);
|
||||
signal.throwIfAborted();
|
||||
await session.startTurn(restartContinuation());
|
||||
await checkpoint(signal);
|
||||
return true;
|
||||
};
|
||||
const recoveredActiveTurnId = recovered
|
||||
? (recoveredSnapshot.activeTurnId ?? null)
|
||||
: (persistedSession?.activeTurnId ?? null);
|
||||
@@ -2294,6 +2343,7 @@ export async function executeNativeSession(
|
||||
}
|
||||
: undefined,
|
||||
recoveredSnapshot.goal,
|
||||
resumeInterruptedTurn,
|
||||
)
|
||||
: Promise.resolve({
|
||||
event: recoveryTerminal,
|
||||
@@ -2333,6 +2383,10 @@ export async function executeNativeSession(
|
||||
await session.goal({ action: "get", requestId });
|
||||
}
|
||||
await checkpoint();
|
||||
} else if (shouldStartFreshTurn && recovered && checkpointedInterruption) {
|
||||
restartContinuationStarted = true;
|
||||
await session.startTurn(restartContinuation());
|
||||
await checkpoint();
|
||||
} else if (shouldStartFreshTurn) {
|
||||
let modelEnvelope = recovered
|
||||
? buildNativeModelEnvelope(input, { resumedSession: true })
|
||||
|
||||
@@ -16,6 +16,7 @@ import { redactSensitiveText } from "../../redaction.js";
|
||||
import { issueRecoveryActionService } from "../issue-recovery-actions.js";
|
||||
import { reportRunFailure } from "../run-failure-report.js";
|
||||
import { isNativeRunnerOwnershipHeld } from "./native-runner-ownership.js";
|
||||
import { nativeRestartInterruptedTurnId, type PersistedNativeSession } from "../../vendor/paperclip-runner/index.js";
|
||||
|
||||
export type NativeControllerIdentity = {
|
||||
bootId: string;
|
||||
@@ -764,7 +765,8 @@ export async function claimNativeRestartRecoveries(input: {
|
||||
hasProviderEvidence,
|
||||
checkpointFailed:
|
||||
(checkpointRecord.terminal as Record<string, unknown> | undefined)
|
||||
?.runTerminalState === "failed",
|
||||
?.runTerminalState === "failed" &&
|
||||
nativeRestartInterruptedTurnId(checkpointRecord as unknown as PersistedNativeSession) === null,
|
||||
providerAttempt: row.coordinator.attempt,
|
||||
});
|
||||
const ownershipChanged =
|
||||
|
||||
@@ -1018,6 +1018,30 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", (
|
||||
]);
|
||||
});
|
||||
|
||||
it.each([2, 3])("recovers a checkpointed restart interruption within provider budget %s", async (attempt) => {
|
||||
const fixture = await seedRun(`INTERRUPTED-${attempt}`);
|
||||
await fixture.db.update(heartbeatRuns).set({ runnerProfileJson: { sessionCheckpoint: {
|
||||
driverKind: "codex_app_server",
|
||||
sessionId: fixture.native.normalizedSessionId,
|
||||
identity: { companyId, issueId: fixture.issueId, runId: fixture.runId, agentId, sessionId: fixture.native.normalizedSessionId },
|
||||
providerSessionId: "interrupted-provider",
|
||||
activeTurnId: null,
|
||||
semanticResult: null,
|
||||
terminal: { runTerminalState: "failed", turnTerminalState: "failed" },
|
||||
terminalTurns: [{ turnId: "interrupted-turn", fingerprint: JSON.stringify({
|
||||
terminalState: "failed", result: null,
|
||||
error: { code: "provider_turn_lost_on_restore", recoverable: true },
|
||||
}) }],
|
||||
} } }).where(eq(heartbeatRuns.id, fixture.runId));
|
||||
await fixture.db.update(nativeRunFinalizations).set({ attempt }).where(eq(nativeRunFinalizations.runId, fixture.runId));
|
||||
const dispositions = await claimNativeRestartRecoveries({
|
||||
db: fixture.db, controller: successor, restartKind: "hot", runIds: [fixture.runId],
|
||||
});
|
||||
expect(dispositions).toEqual([expect.objectContaining(attempt < 3
|
||||
? { kind: "resume_dead_runner", runId: fixture.runId, providerAttempt: attempt }
|
||||
: { kind: "blocked", runId: fixture.runId, reason: "execution_recovery_budget_exhausted" })]);
|
||||
});
|
||||
|
||||
it("terminalizes the recorded failed-checkpoint incident atomically instead of resuming it on upgrade", async () => {
|
||||
const fixture = await seedRun("FAILED-CHECKPOINT");
|
||||
await fixture.db.update(heartbeatRuns).set({ runnerProfileJson: { sessionCheckpoint: {
|
||||
|
||||
@@ -8338,6 +8338,7 @@ async function executePaperclipNativeSessionWithinScope(
|
||||
trace.activate(runnerSessionStartupScope);
|
||||
const result = await trace.run(runnerSessionStartupScope, () =>
|
||||
executeNativeSession({
|
||||
resumeInterruptedTurn: input.restartRecovery?.kind === "resume_dead_runner",
|
||||
getFreshSessionHandoff: input.getFreshSessionHandoff,
|
||||
onSessionAdmission: async () => {
|
||||
// Invalidate prior stop evidence before a backend can spawn.
|
||||
|
||||
+1
@@ -128,6 +128,7 @@ export const validatePrpStructuredRunResult =
|
||||
runner.validatePrpStructuredRunResult;
|
||||
|
||||
export const NativeProviderTerminalFailure = runner.NativeProviderTerminalFailure;
|
||||
export const nativeRestartInterruptedTurnId = runner.nativeRestartInterruptedTurnId;
|
||||
|
||||
export const completeTerminatedRemoteNativeSessionCleanup = runner.completeTerminatedRemoteNativeSessionCleanup;
|
||||
export const completeTerminatedLocalNativeSessionCleanup = runner.completeTerminatedLocalNativeSessionCleanup;
|
||||
Reference in new issue
Block a user