mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-11 05:11:49 +01:00
fix(native): make Retry work after a turn start cut off once Codex recorded the turn (#20516)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - A native Codex run sends `turn/start` through its runner. A control-plane restart can kill the runner and Codex while that request is in flight > - #20507 lets the runner resume by itself when Codex's thread history proves the turn was never created. When Codex did record the turn, the turn may have done work, so the runner still fails closed and nothing replays it > - That branch must still end in a Retry that works. The next run on the same conversation is refused at the runner-state fence before it asks for a runner. #20507 let Retry continue after that run by matching the refusal's error text > - Driving that branch end to end with the real runner exposed a second dead end. When the owner's runner dies, its provider-session close cannot prove suspension, so this server process can keep an operator-required cleanup quarantine for the task. Retry was then refused outright ("Send a new message") > - This pull request replaces the text match with a server-authored run-log event, lets Retry take the same verified-stop path as a new message, and adds a real-runner test of the whole branch: the cut-off owner fails closed, the follow-up is refused with structured proof, Retry is admitted, and the fresh Retry run completes > - The benefit is that this restart window cannot strand a task, and the Retry gate no longer depends on error wording ## Linked Issues or Issue Description **What happened?** Follow-up to #20507 (restart-chaos wave `w-20261110T033426-s108`, t03). #20507 left one item open: prove that a person's Retry produces a working run when the restart cut off `turn.start` after Codex recorded the turn. It also matched the fence refusal by its error text. **Expected behavior** After that cut-off, the owner stays failed closed and the turn is never started again by itself. A person's Retry starts a new run that completes. It never fails with `quarantine_owner_unverified` and never returns 202 with no run. **Steps to reproduce** 1. Run `pnpm --filter @paperclipai/server exec vitest run src/services/native-runtime/native-cut-turn-start-retry.integration.test.ts`. 2. Without the new event, the test fails: the refused follow-up has no proof that it never launched, so Retry after it would wait on `process_identity_missing`. ## What Changed - `native-session-executor.ts`: when the runner-state fence refuses a run before launch (`prior_owner_*` or `legacy_prior_owner_*`), the executor records the server-authored event `native.prior_owner_fence_refused` with the closed refusal `reason`, then fails the run as before. A failed write leaves the run held (the safe side). - `native-local-process-stop.ts`: `nativeRunNeverLaunched` accepts a failed run that has this event (and still no launch, identity, `adapter.invoke` or runner-sourced event). It no longer reads the error text. - `explicit-native-continuation.ts`: when the task's in-process cleanup quarantine is operator-required, a Retry no longer returns `cleanup_quarantined` at once. Like a new message, it must first pass `verifyStoppedNativeSessionForContinuation`, which releases the quarantine and records `native.stopped_conversation_verified`. A native Retry always starts a fresh conversation (`forceFreshSession`), so it never resumes the interrupted turn. If the proof fails, Retry still returns `cleanup_quarantined`. - `verifyStoppedNativeSessionForContinuation` (`native-session-executor.ts`), three narrow additions: - a run with `native.prior_owner_fence_refused` may stand in for the saved state's owner, like a run refused by the cleanup quarantine (both were refused before launch); - when the session root is gone, it reads the fence's single `identity_indeterminate` quarantined copy of the owner's state (several copies stay ambiguous); - a provider startup attempt in phase `initialization_failed` whose direct child exited and whose process and process group are gone no longer blocks the proof. A same-run restore that fails closed leaves exactly this record. Those PIDs are also re-checked before the quarantine is released. - New real-runnerd integration test `native-cut-turn-start-retry.integration.test.ts` (real runnerd and fake Codex with `--hang-turn-start after`): 1. The owner run's `turn/start` hangs after Codex recorded the turn; the runner and Codex are killed. The owner's provider state keeps the ambiguous start, and the thread history keeps the turn. 2. The same-run resume restores the runner and fails closed with "cannot safely recover an ambiguous Codex turn start". Codex saw exactly one `turn/start`. 3. A follow-up on the same conversation is refused with `prior_owner_*`, never `quarantine_owner_unverified`. It records `native.prior_owner_fence_refused`, launches nothing, and `nativeRunNeverLaunched` is true for it. 4. The owner's operator-required in-process cleanup quarantine is pinned (whether the real close lands there is timing-dependent; about half of the local runs did). 5. Retry of the refused follow-up passes `admitExplicitNativeContinuation` (not a dry run) with no block reason, releases the quarantine and records `native.stopped_conversation_verified`. 6. The Retry run, prepared with a fresh conversation as task Retry does (`forceFreshSession`), uses a different runner-state root and completes with exit code 0. Codex saw one new `thread/start` and one new `turn/start`; the cut-off turn was never started again. - `explicit-native-continuation.test.ts`: the fence rows now insert the event. A new row shows that the refusal text alone keeps the run held. The "worker loss (retry)" case now expects a verified Retry to be admitted like a new message. - `native-recovered-owner-followup.test.ts`: a fixture event moves to a free sequence number, because the refused follow-up now writes its own event. - Docs: `doc/run-log-events.md` (new event) and `packages/paperclip-runner/docs/durable-recovery.md`. ## Verification - Red/green: - Without the event write, the integration test fails (no fence event, so the follow-up is not proven unlaunched). - Without the gate and proof changes, it fails with `expected [ 'cleanup_quarantined' ] to deeply equal []`: Retry is refused while the owner's quarantine is held. - With all changes, it passes. - `vitest run` of `native-recovered-owner-followup`, `explicit-native-continuation` (427 tests), `quarantined-task-retry`, `issue-confirm-run-stopped-routes`, `native-finalization-recovery`, `task-interruption-query`, `execution-projection`, `chat-run-publications`, `closed-native-receipt-accounting`, `agent-live-run-routes` and the new integration test pass. - Server `tsc --noEmit` passes. - Runner code and the pack `contractDigest` do not change. ## Risks - Low-medium risk. The new event is written only on the fence refusal path, which already fails the run. It is local run-log data (no Telemetry or OpenTelemetry). - Retry after an operator-required cleanup quarantine is a behaviour change: it was always refused, and now it is admitted only after the same exact stop proof a new message needs. That proof checks the owner's runner stop record, every recorded provider and startup PID and process group, and the unchanged state fingerprint before it releases the quarantine. - A run refused by a build between #20507 and this change has only the error text. It stays held on `process_identity_missing` until a person sends a new message or the board confirms the stop. #20507 has not shipped in a release yet. - The test models the restart in-process: it kills the runner and Codex and calls the executor again. It does not drive the HTTP Retry route; the route's own behaviour (fresh conversation, admission gate) is covered by existing tests and by `nativeRunNeverLaunched` here. - Open decision, unchanged: a turn Codex recorded before the restart is never replayed or adopted. Continuing is a person's decision. > For core feature work, check [`ROADMAP.md`](ROADMAP.md) first and discuss it in `#dev` before opening the PR. Feature PRs that overlap with planned core work may need to be redirected — check the roadmap first. See `CONTRIBUTING.md`. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), via Claude Code with tool use (shell, file edits, test execution). ## 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 - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge Refs #20507 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude <noreply@anthropic.com>
This commit is contained in:
DottaandClaude authored and GitHub committed
1 parent
93faca77ef
commit
bf45700ca3
10 files changed
+374
-29
No files matched your search
@@ -495,6 +495,15 @@ server-authored event among these three types; provider
|
||||
source events cannot supply stop authority. These records stay in the local run
|
||||
log and do not add Telemetry or OpenTelemetry data.
|
||||
|
||||
The server writes `native.prior_owner_fence_refused` when its runner-state
|
||||
check refuses a run before any launch, because the conversation's saved runner
|
||||
state belongs to an earlier run that cannot be proven finished. The payload
|
||||
holds only the closed refusal `reason` (for example
|
||||
`prior_owner_terminal_state_indeterminate`). Continuation admission reads it,
|
||||
with no launch, identity, `adapter.invoke` or runner-sourced event, as proof
|
||||
that the failed run never had a runner, so Retry and new input need no stop
|
||||
record for it.
|
||||
|
||||
Explicit continuation writes `native.stopped_conversation_verified` after it
|
||||
verifies the retained worker and provider processes stopped. Its payload binds
|
||||
the requesting `runId`, retained `ownerRunId`, native session, runner instance,
|
||||
|
||||
@@ -214,8 +214,18 @@ 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.
|
||||
runner, so it never had a process. The server records
|
||||
`native.prior_owner_fence_refused` for it, and Retry and new messages after it
|
||||
continue without waiting for a stop record of that run. A task Retry always
|
||||
starts a fresh provider conversation, so it never resumes or trips over the
|
||||
stuck conversation's saved or quarantined runner state.
|
||||
If this server process still holds the owner's provider-session cleanup
|
||||
quarantine (its close could not prove suspension once the runner died),
|
||||
Retry, like a new message, first proves the owner's runner, Codex and tools
|
||||
stopped and then releases that quarantine (`native.stopped_conversation_verified`).
|
||||
That proof also reads the fence's single quarantined copy of the owner's
|
||||
state, and accepts a provider restore that failed its own admission once its
|
||||
process and process group are gone.
|
||||
|
||||
Saved OpenCode and Cursor Cloud provider state binds the digest of its own
|
||||
qualified launch profile (command, proxy script and, for OpenCode, the
|
||||
|
||||
@@ -329,6 +329,7 @@ const pluginEvictionSuitePath = "server/src/services/native-runtime/plugin-runne
|
||||
const dotRunnerSuitePath = "server/src/__tests__/dot-runner.test.ts";
|
||||
const restartRecoverySuitePath =
|
||||
"server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts";
|
||||
const cutTurnStartRetrySuitePath = "server/src/services/native-runtime/native-cut-turn-start-retry.integration.test.ts";
|
||||
const restartWaitSuites = ["lease-renewal", "preparation", "checkpoint-cost", "diskless"]
|
||||
.map(name => `server/src/services/native-runtime/native-runner-restart-${name}.integration.test.ts`);
|
||||
|
||||
@@ -372,12 +373,13 @@ test("12 PR without-chat shards plus the dedicated chat and native-runner lanes
|
||||
assert.ok(!files.includes(ingressRetrySuitePath));
|
||||
assert.ok(!files.includes(dotRunnerSuitePath));
|
||||
assert.ok(!files.includes(restartRecoverySuitePath));
|
||||
assert.ok(!files.includes(cutTurnStartRetrySuitePath));
|
||||
assert.ok(!files.includes(usageDeliverySuitePath));
|
||||
assert.ok(!files.includes(warmCheckpointOwnershipSuitePath));
|
||||
assert.ok(!files.includes(pluginWarmOwnerSuitePath));
|
||||
assert.ok(!files.includes(pluginReleaseResumeSuitePath));
|
||||
assert.ok(!files.includes(pluginEvictionSuitePath));
|
||||
assert.deepEqual([...files, chatSuitePath, nativeRunnerSuitePath, ingressRetrySuitePath, dotRunnerSuitePath, restartRecoverySuitePath, ...restartWaitSuites, usageDeliverySuitePath, warmCheckpointOwnershipSuitePath, pluginWarmOwnerSuitePath, pluginReleaseResumeSuitePath, pluginEvictionSuitePath, "server/src/services/native-runtime/plugin-runner-probe.test.ts"].sort(), full.selectedGeneralServerSuites.sort());
|
||||
assert.deepEqual([...files, chatSuitePath, nativeRunnerSuitePath, ingressRetrySuitePath, dotRunnerSuitePath, restartRecoverySuitePath, ...restartWaitSuites, cutTurnStartRetrySuitePath, usageDeliverySuitePath, warmCheckpointOwnershipSuitePath, pluginWarmOwnerSuitePath, pluginReleaseResumeSuitePath, pluginEvictionSuitePath, "server/src/services/native-runtime/plugin-runner-probe.test.ts"].sort(), full.selectedGeneralServerSuites.sort());
|
||||
assert.equal(new Set(files).size, files.length);
|
||||
const defaultRun = dryRunJson([], prEnv);
|
||||
assert.ok(defaultRun.generalServerSuiteCount === full.generalServerSuiteCount);
|
||||
@@ -432,6 +434,7 @@ test("the native-runner lane runs exactly the cargo-dependent suites", () => {
|
||||
dotRunnerSuitePath,
|
||||
restartRecoverySuitePath,
|
||||
...restartWaitSuites,
|
||||
cutTurnStartRetrySuitePath,
|
||||
usageDeliverySuitePath,
|
||||
warmCheckpointOwnershipSuitePath,
|
||||
pluginWarmOwnerSuitePath,
|
||||
|
||||
@@ -103,6 +103,7 @@ const nativeRunnerSuites = [
|
||||
"server/src/services/native-runtime/native-runner-restart-preparation.integration.test.ts",
|
||||
"server/src/services/native-runtime/native-runner-restart-checkpoint-cost.integration.test.ts",
|
||||
"server/src/services/native-runtime/native-runner-restart-diskless.integration.test.ts",
|
||||
"server/src/services/native-runtime/native-cut-turn-start-retry.integration.test.ts",
|
||||
"server/src/services/native-runtime/native-usage-delivery.integration.test.ts",
|
||||
"server/src/services/native-runtime/native-warm-checkpoint-ownership.integration.test.ts",
|
||||
"server/src/services/native-runtime/plugin-runner-task-warm.integration.test.ts",
|
||||
|
||||
@@ -10,7 +10,7 @@ import { issueTreeControlService } from "./issue-tree-control.js";
|
||||
import { createReleaseIssueExecution } from "../modules/wake-queue/application/use-cases.js";
|
||||
import * as nativeExecutor from "./native-runtime/native-session-executor.js";
|
||||
import { appendHeartbeatRunEvent } from "./heartbeat-run-events.js";
|
||||
import { recordNativeLocalProcessStop, hasNativeLocalProcessStop, PROCESS_START_REQUESTED } from "./native-local-process-stop.js";
|
||||
import { recordNativeLocalProcessStop, hasNativeLocalProcessStop, PRIOR_OWNER_FENCE_REFUSED, PROCESS_START_REQUESTED } from "./native-local-process-stop.js";
|
||||
import { remoteTerminationReceipt } from "./remote-execution-termination.js";
|
||||
import { createHash, randomBytes, randomUUID } from "node:crypto";
|
||||
import { mkdtemp, mkdir, writeFile, rm } from "node:fs/promises";
|
||||
@@ -117,13 +117,16 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes
|
||||
// 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 },
|
||||
prepared: true, fenced: 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 },
|
||||
prepared: true, coordinator: true, fenced: 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" },
|
||||
prepared: true, fenced: 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" },
|
||||
runnerEvent: true, prepared: true, fenced: true, error: PRIOR_OWNER_REFUSAL, comment: "process_identity_missing", retry: "process_identity_missing" },
|
||||
// The fence's own server-authored event is the proof, never the error text.
|
||||
{ shape: "carries the prior-owner refusal text without the fence event", errorCode: "adapter_failed", launched: false,
|
||||
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" },
|
||||
@@ -134,8 +137,8 @@ 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;
|
||||
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 }) => {
|
||||
error?: string; fenced?: 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, error, fenced, 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
|
||||
@@ -156,6 +159,9 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes
|
||||
if (runnerEvent) await db.insert(heartbeatRunEvents).values({ companyId: f.companyId, runId: f.sourceRunId, agentId: f.agentId,
|
||||
seq: 1, eventType: "session.started", stream: "system", level: "info", message: "runner event",
|
||||
sourceEventId: "runner-event-1", payload: {} });
|
||||
if (fenced) await db.insert(heartbeatRunEvents).values({ companyId: f.companyId, runId: f.sourceRunId, agentId: f.agentId,
|
||||
seq: 2, eventType: PRIOR_OWNER_FENCE_REFUSED, stream: "system", level: "warn", message: "fence refused before launch",
|
||||
payload: { reason: "prior_owner_terminal_state_indeterminate" } });
|
||||
if (prepared) await db.update(heartbeatRuns).set({ nativeIssueId: f.issueId, nativeSessionId: randomUUID(), runnerInstanceId: randomUUID() })
|
||||
.where(eq(heartbeatRuns.id, f.sourceRunId));
|
||||
const { result, blocked } = await readAdmission(f, false);
|
||||
@@ -1194,13 +1200,15 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes
|
||||
? await db.transaction(tx => admitExplicitNativeContinuation({ ...f, db: tx as unknown as typeof db,
|
||||
reason: "retry_failed_run", failedRunId: f.sourceRunId }))
|
||||
: await admit(f, mode === "dry_run");
|
||||
expect(Boolean(result)).toBe(["verified", "dry_run", "duplicate"].includes(mode));
|
||||
expect(retire).toHaveBeenCalledTimes(["verified", "changed", "duplicate"].includes(mode) ? 1 : 0);
|
||||
// A native Retry starts a fresh conversation, so the same verified stop
|
||||
// admits it; it never resumes the interrupted turn.
|
||||
expect(Boolean(result)).toBe(["verified", "dry_run", "duplicate", "retry"].includes(mode));
|
||||
expect(retire).toHaveBeenCalledTimes(["verified", "changed", "duplicate", "retry"].includes(mode) ? 1 : 0);
|
||||
const actions = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, f.issueId));
|
||||
expect(Boolean(actions[0].evidence.explicitUserContinuation)).toBe(["verified", "duplicate"].includes(mode));
|
||||
expect(Boolean(actions[0].evidence.explicitUserContinuation)).toBe(["verified", "duplicate", "retry"].includes(mode));
|
||||
const events = await db.select().from(heartbeatRunEvents).where(and(eq(heartbeatRunEvents.runId, f.sourceRunId),
|
||||
eq(heartbeatRunEvents.eventType, "native.stopped_conversation_verified")));
|
||||
expect(events).toHaveLength(["verified", "duplicate"].includes(mode) ? 1 : 0);
|
||||
expect(events).toHaveLength(["verified", "duplicate", "retry"].includes(mode) ? 1 : 0);
|
||||
if (mode === "duplicate") {
|
||||
expect(await admit(f)).toBeNull();
|
||||
expect(retire).toHaveBeenCalledTimes(1);
|
||||
|
||||
@@ -532,11 +532,14 @@ export async function admitExplicitNativeContinuation(input: {
|
||||
return blocked("local_cleanup", "Waiting for the previous provider session to finish cleanup. Your message will start automatically.");
|
||||
if (!remote && run.runtimeMode === "native" &&
|
||||
(run.errorCode === "native_session_cleanup_quarantined" || retainedCleanup === "operator_required")) {
|
||||
// This is new user input, never permission to retry the interrupted turn.
|
||||
if (retry) return blocked("cleanup_quarantined", "Send a new message after the previous provider has stopped.");
|
||||
// Never permission to resume the interrupted turn: new input and a
|
||||
// native Retry both start a fresh provider conversation, and only after
|
||||
// the old runner, provider and tools are proven stopped.
|
||||
const { verifyStoppedNativeSessionForContinuation } = await import("./native-runtime/native-session-executor.js");
|
||||
const stopped = await verifyStoppedNativeSessionForContinuation(db, run);
|
||||
if (!stopped) return blocked("local_cleanup", "Waiting for the previous provider and its tools to stop. Your message is saved.");
|
||||
if (!stopped) return retry
|
||||
? blocked("cleanup_quarantined", "Send a new message after the previous provider has stopped.")
|
||||
: blocked("local_cleanup", "Waiting for the previous provider and its tools to stop. Your message is saved.");
|
||||
stoppedSessions.push(stopped);
|
||||
}
|
||||
if (nativeQueuedSend && await hasPendingRunCleanup(db, companyId, run.id))
|
||||
|
||||
@@ -12,6 +12,9 @@ import { proveOwnerGone } from "./owner-proof.js";
|
||||
|
||||
export const PROCESS_START_REQUESTED = "native.process_start_requested";
|
||||
export const PROCESS_IDENTITY_RECORDED = "native.process_identity_recorded";
|
||||
/** Server-authored: the runner-state fence refused this run before it asked for
|
||||
* a runner, because a prior run's saved state could not be proven finished. */
|
||||
export const PRIOR_OWNER_FENCE_REFUSED = "native.prior_owner_fence_refused";
|
||||
const LOCAL_PROCESS_STOPPED = "native.local_process_stopped";
|
||||
const SETTLED_RUN_STATUSES = ["failed", "cancelled", "timed_out", "interrupted"];
|
||||
const SETTLED_COORDINATOR_PHASES = ["committed", "terminal_failure"];
|
||||
@@ -243,8 +246,9 @@ function record(value: unknown): Record<string, unknown> {
|
||||
* 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
|
||||
* owner could not be proven finished. The executor records
|
||||
* PRIOR_OWNER_FENCE_REFUSED before it fails the run. 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
|
||||
@@ -252,16 +256,20 @@ function record(value: unknown): Record<string, unknown> {
|
||||
* 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_/;
|
||||
async function hasRunEvent(db: Db, run: Pick<typeof heartbeatRuns.$inferSelect, "id" | "companyId">, eventType: string) {
|
||||
const [event] = await db.select({ id: heartbeatRunEvents.id }).from(heartbeatRunEvents).where(and(
|
||||
eq(heartbeatRunEvents.companyId, run.companyId), eq(heartbeatRunEvents.runId, run.id),
|
||||
isNull(heartbeatRunEvents.sourceEventId), eq(heartbeatRunEvents.eventType, eventType),
|
||||
)).limit(1);
|
||||
return Boolean(event);
|
||||
}
|
||||
|
||||
export async function nativeRunNeverLaunched(db: Db, run: Pick<typeof heartbeatRuns.$inferSelect,
|
||||
"id" | "companyId" | "runtimeMode" | "status" | "errorCode" | "processPid" | "processGroupId" |
|
||||
"nativeSessionId" | "runnerInstanceId" | "nativeIssueId" | "error">): Promise<boolean> {
|
||||
"nativeSessionId" | "runnerInstanceId" | "nativeIssueId">): Promise<boolean> {
|
||||
if (run.runtimeMode !== "native" || !SETTLED_RUN_STATUSES.includes(run.status) || run.processPid || run.processGroupId) return false;
|
||||
const refusedBeforeLaunch = run.status === "failed" && (run.errorCode === "native_session_cleanup_quarantined" ||
|
||||
PRIOR_OWNER_FENCE_REFUSAL.test(run.error ?? ""));
|
||||
await hasRunEvent(db, run, PRIOR_OWNER_FENCE_REFUSED));
|
||||
if (!refusedBeforeLaunch) {
|
||||
// A coordinator row requires the run's native issue (foreign key), so the
|
||||
// native issue also rules out a coordinator.
|
||||
|
||||
@@ -0,0 +1,256 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { existsSync } from "node:fs";
|
||||
import { mkdir, readFile, readdir, rm, writeFile } from "node:fs/promises";
|
||||
import { join } from "node:path";
|
||||
import { and, eq } from "drizzle-orm";
|
||||
import { completionContracts, heartbeatRunEvents, heartbeatRuns, issueRecoveryActions, issues, nativeRunFinalizations } from "@paperclipai/db";
|
||||
import { expect, vi } from "vitest";
|
||||
import { NativeSessionCloseUnrecoverableError, executeNativeSession, parseNativeExecutionInput, nativeSessionCleanupBlock, createNativeSessionBackend, createRunnerdCodexTransport, canonicalNativeRuntimeContextDigest } from "../../vendor/paperclip-runner/index.js";
|
||||
import { describeEmbeddedPostgres, nativeRestartFixture, processAlive, realProcessIt, waitForCondition } from "../../__tests__/helpers/native-restart.js";
|
||||
import { buildNativeExecutionInput } from "./native-execution-input.js";
|
||||
import { buildNativeCompletionContract } from "./completion-contracts.js";
|
||||
import { executePaperclipNativeSession, scopedRunnerdStateRoot } from "./native-session-executor.js";
|
||||
import { prepareNativeHeartbeatRun } from "./prepare-native-run.js";
|
||||
import { materializeAsset } from "./runtime-context.js";
|
||||
import { nativeRuntimeContextFixture } from "./runtime-context.test-fixture.js";
|
||||
import { nativeRunNeverLaunched, PRIOR_OWNER_FENCE_REFUSED, recordNativeLocalProcessStop } from "../native-local-process-stop.js";
|
||||
import { admitExplicitNativeContinuation } from "../explicit-native-continuation.js";
|
||||
|
||||
/**
|
||||
* w-20261110T033426-s108/t03, the branch the runner cannot settle by itself:
|
||||
* a restart kills the runner and Codex while `turn/start` is in flight, after
|
||||
* Codex already recorded the turn. That turn may have done work, so the
|
||||
* runner keeps failing closed and nothing replays it. The task must still
|
||||
* continue through a person's Retry:
|
||||
* - a follow-up on the same conversation is refused at the runner-state fence
|
||||
* before it asks for a runner, and that refusal leaves structured proof that
|
||||
* it never had a process (so Retry after it is admitted, never a 202 with no
|
||||
* run);
|
||||
* - the Retry run starts a fresh provider conversation (task Retry always sets
|
||||
* forceFreshSession) and completes with the real runner, without touching or
|
||||
* tripping over the stuck conversation's saved or quarantined state.
|
||||
*/
|
||||
describeEmbeddedPostgres("Retry after a turn start cut off once Codex recorded the turn", () => {
|
||||
const environment = nativeRestartFixture();
|
||||
|
||||
realProcessIt("keeps the cut-off turn closed and completes a fresh Retry run", async () => {
|
||||
const fixture = await environment.seedRun("CUT-TURN-START");
|
||||
const { db, issueId } = fixture;
|
||||
const { companyId, agentId, runtimeRoot } = environment.current;
|
||||
const stateBase = join(runtimeRoot, `cut-turn-start-state-${fixture.runId}`);
|
||||
const sandbox = join(runtimeRoot, `cut-turn-start-workspace-${fixture.runId}`);
|
||||
const codexHome = join(runtimeRoot, `cut-turn-start-codex-home-${fixture.runId}`);
|
||||
await mkdir(sandbox, { recursive: true });
|
||||
await mkdir(codexHome, { recursive: true });
|
||||
vi.stubEnv("PAPERCLIP_RUNNER_STATE_DIR", stateBase);
|
||||
const [issue] = await db.select().from(issues).where(eq(issues.id, issueId));
|
||||
const [contract] = await db.select().from(completionContracts).where(eq(completionContracts.issueId, issueId));
|
||||
const runtimeContext = nativeRuntimeContextFixture();
|
||||
runtimeContext.instructions.bundle = await materializeAsset([{ path: "AGENTS.md", mode: 0o444,
|
||||
content: Buffer.from("Complete only the controlled fixture task.\n") }]);
|
||||
runtimeContext.aggregateDigest = canonicalNativeRuntimeContextDigest(runtimeContext);
|
||||
const executionFor = (runId: string, normalizedSessionId: string) => buildNativeExecutionInput({
|
||||
companyId, agentId, runId, provider: "codex",
|
||||
issue: { id: issueId, identifier: "NRR-CUT", title: issue!.title, description: null, workMode: "standard" },
|
||||
taskPrompt: "Write the brief once.",
|
||||
workspace: { id: runId, cwd: sandbox, repoUrl: null, repoRef: null, branchName: null },
|
||||
normalizedSessionId, runtimeContext, lifecyclePolicy: { mode: "per_turn", idleTimeoutMs: null },
|
||||
completionContract: { id: contract!.id, sha256: contract!.canonicalSha256, schemaVersion: "paperclip.completion-contract.v1",
|
||||
contract: buildNativeCompletionContract(issue!, { revision: contract!.revision }) },
|
||||
});
|
||||
const resultFile = join(sandbox, "fixture-result.json");
|
||||
const hangFile = join(sandbox, "hang-turn-start");
|
||||
const calls = join(sandbox, "provider-calls.log");
|
||||
const execution = executionFor(fixture.runId, fixture.native.normalizedSessionId);
|
||||
await writeFile(resultFile, JSON.stringify({ schema: "paperclip.run_result.v1", reportedWorkDisposition: "done",
|
||||
summary: "Controlled Retry completed.", completionClaim: { contractRevision: execution.completionContract.contract.revision,
|
||||
objectiveSatisfied: true, criteria: execution.completionContract.contract.criteria.map(criterion => ({
|
||||
criterionId: criterion.id, status: "satisfied", evidenceRefs: [] })), remainingWork: [] },
|
||||
evidence: [], verification: [], artifacts: [], attentionRequests: [] }));
|
||||
await db.update(heartbeatRuns).set({ runnerProfileJson: { ...fixture.preparedProfile, nativeExecutionInput: execution } })
|
||||
.where(eq(heartbeatRuns.id, fixture.runId));
|
||||
|
||||
const bundles: ReturnType<typeof createRunnerdCodexTransport>[] = [];
|
||||
const run = (runExecution: typeof execution, runnerInstanceId: string, environmentLeaseId: string) => {
|
||||
const root = scopedRunnerdStateRoot(runExecution);
|
||||
const identity = { runnerInstanceId, environmentLeaseId, runId: runExecution.binding.runId,
|
||||
normalizedSessionId: runExecution.session.normalizedSessionId!,
|
||||
turnId: `turn-${runExecution.binding.runId}`, itemId: `item-${runExecution.binding.runId}` };
|
||||
const backend = createNativeSessionBackend(runExecution, { runnerInstanceId,
|
||||
environment: { CODEX_HOME: codexHome, PATH: process.env.PATH },
|
||||
codexTransportFactory: () => {
|
||||
const options = environment.transportOptions(fixture, root);
|
||||
const bundle = createRunnerdCodexTransport({ ...options,
|
||||
codexArgs: ["--state-file", join(sandbox, "provider.json"), "--terminal-result-file", resultFile,
|
||||
"--call-log", calls, "--durable-turn-ids", "--hang-turn-start", hangFile],
|
||||
runtimeContext, runnerRuntimeContext: runtimeContext, sourceCodexHome: codexHome,
|
||||
environment: { CODEX_HOME: codexHome, PATH: process.env.PATH }, closeGraceMs: 1_000, turnStartTimeoutMs: 20_000,
|
||||
prpIdentity: identity, lifecyclePolicy: { mode: "per_turn", idleTimeoutMs: null } });
|
||||
bundles.push(bundle);
|
||||
return bundle.transport;
|
||||
} });
|
||||
return { root, outcome: executePaperclipNativeSession({ db, execution: runExecution, backend, useRunnerd: true,
|
||||
runnerInstanceId, turnTimeoutMs: 15_000 }) };
|
||||
};
|
||||
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 {
|
||||
// 1. The owner's turn start is in flight when the restart kills its
|
||||
// runner and Codex, after Codex recorded the turn.
|
||||
await writeFile(hangFile, "after");
|
||||
const owner = run(execution, fixture.native.runnerInstanceId, fixture.native.environmentLeaseId);
|
||||
const ownerOutcome = owner.outcome.catch((error: unknown) => error);
|
||||
await waitForCondition("Codex to record the owner's turn", async () => existsSync(join(sandbox, "provider.json")) &&
|
||||
JSON.parse(await readFile(join(sandbox, "provider.json"), "utf8")).activeTurnId === "provider-turn-1", 20_000);
|
||||
const killed = bundles[0]!.evidence();
|
||||
kill(killed.providerPid);
|
||||
kill(killed.runnerPid);
|
||||
await waitForCondition("the owner's runner and Codex to exit", () =>
|
||||
!processAlive(killed.runnerPid) && !processAlive(killed.providerPid));
|
||||
await rm(hangFile, { force: true });
|
||||
expect(await ownerOutcome).toBeInstanceOf(Error);
|
||||
const providerState = JSON.parse(await readFile(join(owner.root, "runner", "codex-provider-state.json"), "utf8"));
|
||||
expect(providerState.ambiguousTurnStartPending).toBe(true);
|
||||
// Codex's process-loss history: the thread keeps the turn, not its run.
|
||||
const fake = JSON.parse(await readFile(join(sandbox, "provider.json"), "utf8"));
|
||||
await writeFile(join(sandbox, "provider.json"), JSON.stringify({ ...fake, activeTurnId: null }));
|
||||
expect((await readFile(join(sandbox, "provider.json.turns"), "utf8")).trim()).toBe("provider-turn-1");
|
||||
// The same-run resume restores the owner's runner and fails closed: the
|
||||
// recorded turn may have done work, and it is never started again.
|
||||
const resumedOwner = createRunnerdCodexTransport({ ...environment.transportOptions(fixture, owner.root),
|
||||
codexArgs: ["--state-file", join(sandbox, "provider.json"), "--call-log", calls, "--durable-turn-ids"],
|
||||
runtimeContext, runnerRuntimeContext: runtimeContext, sourceCodexHome: codexHome, resumeDynamicTools: [],
|
||||
environment: { CODEX_HOME: codexHome, PATH: process.env.PATH }, closeGraceMs: 1_000,
|
||||
prpIdentity: { runnerInstanceId: fixture.native.runnerInstanceId, environmentLeaseId: fixture.native.environmentLeaseId,
|
||||
runId: fixture.runId, normalizedSessionId: fixture.native.normalizedSessionId,
|
||||
turnId: `turn-${fixture.runId}`, itemId: `item-${fixture.runId}` },
|
||||
lifecyclePolicy: { mode: "per_turn", idleTimeoutMs: null } });
|
||||
bundles.push(resumedOwner);
|
||||
await expect(resumedOwner.transport.request("thread/read", {}))
|
||||
.rejects.toThrow(/cannot safely recover an ambiguous Codex turn start/);
|
||||
await resumedOwner.transport.close().catch(() => undefined);
|
||||
// What restart recovery records for the owner: its runner identity, then
|
||||
// the verified stop of that exact process.
|
||||
await db.update(heartbeatRuns).set({ status: "failed", errorCode: "native_session_interrupted", finishedAt: new Date(),
|
||||
processPid: killed.runnerPid, processGroupId: killed.runnerProcessGroupId }).where(eq(heartbeatRuns.id, fixture.runId));
|
||||
const [stoppedOwner] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, fixture.runId));
|
||||
expect(await db.transaction(tx => recordNativeLocalProcessStop(tx as never, stoppedOwner!))).toBe(true);
|
||||
await db.update(nativeRunFinalizations).set({ phase: "terminal_failure", leaseOwner: null, leaseExpiresAt: null,
|
||||
nextAttemptAt: null, recoveryState: null, failureCode: "native_continuation_requires_reconciliation" })
|
||||
.where(eq(nativeRunFinalizations.runId, fixture.runId));
|
||||
// The owner's in-process session close cannot prove suspension once its
|
||||
// runner died, so this process can keep an operator-required cleanup
|
||||
// quarantine for the task. Whether the real close lands there is
|
||||
// timing-dependent; pin that state so Retry is checked against it.
|
||||
if (nativeSessionCleanupBlock({ companyId, issueId }) !== "operator_required") {
|
||||
const capabilities = { resume: true, typedEvents: true, steering: false, interruption: false, structuredResult: true };
|
||||
const identity = { runId: fixture.runId, sessionId: fixture.native.normalizedSessionId, companyId, issueId, agentId };
|
||||
await expect(executeNativeSession({ input: parseNativeExecutionInput({ schema: "paperclip.native-execution-input.v1",
|
||||
provider: { kind: "codex", model: null }, binding: execution.binding,
|
||||
task: { identifier: "NRR-CUT", title: issue!.title, description: null, prompt: "Write the brief once.", workMode: "standard" },
|
||||
workspace: { cwd: sandbox, repoUrl: null, repoRef: null, branchName: null },
|
||||
session: { normalizedSessionId: fixture.native.normalizedSessionId, driverKind: "codex_app_server", protocolVersion: 1,
|
||||
lifecyclePolicy: { mode: "per_turn", idleTimeoutMs: null } },
|
||||
completionContract: execution.completionContract, interactionResponses: [], credentialBindings: [] }),
|
||||
backend: { descriptor: async () => ({ kind: "runner", name: "cut-turn-start-owner", version: "1", capabilities }),
|
||||
openSession: async () => ({ identity: () => identity, capabilities: async () => capabilities,
|
||||
async *events() { throw new Error("owner runner exited"); },
|
||||
startTurn: async () => ({ turnId: "owner-turn" }), result: async () => null,
|
||||
snapshot: async () => ({ backendKind: "runner", sessionId: fixture.native.normalizedSessionId, identity,
|
||||
providerSessionId: "owner-thread", cursor: null, activeTurnId: null, pendingRuntimeRequests: [], lineage: [] }),
|
||||
close: async () => { throw new NativeSessionCloseUnrecoverableError(); } }) },
|
||||
controlPlane: { openRun: async () => {}, checkpointSession: async () => {},
|
||||
appendEvent: async () => ({ cursor: 0, highestContiguousSourceSeq: 0, disposition: "committed" }),
|
||||
replayEvents: async () => ({ events: [], highestContiguousSourceSeq: 0 }), completeRun: async () => {} },
|
||||
runnerInstanceId: fixture.native.runnerInstanceId, controlPlaneInstanceId: "cut-turn-start-owner" })).rejects.toThrow("owner runner exited");
|
||||
}
|
||||
expect(nativeSessionCleanupBlock({ companyId, issueId })).toBe("operator_required");
|
||||
const startsBeforeRetry = (await readFile(calls, "utf8")).match(/^turn\/start$/gm)?.length ?? 0;
|
||||
expect(startsBeforeRetry).toBe(1);
|
||||
|
||||
// 2. A follow-up that continues the same conversation is refused before
|
||||
// it asks for a runner, and records structured proof of that.
|
||||
const followUpId = randomUUID();
|
||||
const followUp = await admitRun(followUpId, fixture.native.normalizedSessionId);
|
||||
const followUpExecution = executionFor(followUpId, fixture.native.normalizedSessionId);
|
||||
const refused = await run(followUpExecution, followUp.runnerInstanceId, followUp.environmentLeaseId).outcome
|
||||
.then(() => null, (error: unknown) => error);
|
||||
expect(String(refused)).toContain("runner_state_identity_mismatch: prior_owner_");
|
||||
expect(String(refused)).not.toContain("quarantine_owner_unverified");
|
||||
await db.update(heartbeatRuns).set({ status: "failed", errorCode: "adapter_failed", error: String(refused), finishedAt: new Date() })
|
||||
.where(eq(heartbeatRuns.id, followUpId));
|
||||
// What run failure handling records for it: a settled coordinator and
|
||||
// the "Automatic recovery stopped" action that offers Retry.
|
||||
await db.update(nativeRunFinalizations).set({ phase: "terminal_failure",
|
||||
failureCode: "native_continuation_requires_reconciliation" }).where(eq(nativeRunFinalizations.runId, followUpId));
|
||||
await db.insert(issueRecoveryActions).values({ companyId, sourceIssueId: issueId, kind: "active_run_watchdog",
|
||||
cause: "native_continuation_requires_reconciliation", fingerprint: followUpId, status: "resolved", outcome: "blocked",
|
||||
nextAction: "Automatic recovery stopped.", evidence: { runId: followUpId, automaticRecovery: { replay: "blocked" } } });
|
||||
const [refusedRun] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, followUpId));
|
||||
expect(await db.select({ type: heartbeatRunEvents.eventType }).from(heartbeatRunEvents).where(and(
|
||||
eq(heartbeatRunEvents.runId, followUpId), eq(heartbeatRunEvents.eventType, PRIOR_OWNER_FENCE_REFUSED)))).toHaveLength(1);
|
||||
expect(await nativeRunNeverLaunched(db, refusedRun!)).toBe(true);
|
||||
// Owner launch and its resume only: the refused follow-up launched nothing.
|
||||
expect(bundles).toHaveLength(2);
|
||||
|
||||
// 3. The person's Retry starts a fresh conversation and completes once.
|
||||
// The latest run is the refused follow-up, so the person retries it.
|
||||
// Admission must not wait for a stop record of a process it never had.
|
||||
const retryId = randomUUID();
|
||||
const blocked: string[] = [];
|
||||
const admitted = await admitExplicitNativeContinuation({ db, companyId, issueId, agentId, actorType: "user",
|
||||
actorId: "local-board", reason: "retry_failed_run", failedRunId: followUpId, commentId: null,
|
||||
successorRunId: retryId, onBlocked: reason => blocked.push(reason) });
|
||||
expect(blocked).toEqual([]);
|
||||
expect(admitted).toMatchObject({ previousRunId: followUpId });
|
||||
// Admission proved the owner's runner, Codex and tools stopped and
|
||||
// released the task's cleanup quarantine; nothing old is resumed.
|
||||
expect(nativeSessionCleanupBlock({ companyId, issueId })).toBeNull();
|
||||
expect(await db.select({ type: heartbeatRunEvents.eventType }).from(heartbeatRunEvents).where(and(
|
||||
eq(heartbeatRunEvents.runId, followUpId), eq(heartbeatRunEvents.eventType, "native.stopped_conversation_verified"))))
|
||||
.toHaveLength(1);
|
||||
const retry = await admitRun(retryId, null, followUpId);
|
||||
expect(retry.normalizedSessionId).not.toBe(fixture.native.normalizedSessionId);
|
||||
const retryExecution = executionFor(retryId, retry.normalizedSessionId);
|
||||
const retried = run(retryExecution, retry.runnerInstanceId, retry.environmentLeaseId);
|
||||
expect(retried.root).not.toBe(owner.root);
|
||||
await expect(retried.outcome).resolves.toMatchObject({ exitCode: 0, timedOut: false });
|
||||
const providerCalls = await readFile(calls, "utf8");
|
||||
// The fresh conversation opened its own thread; the cut-off turn was
|
||||
// never started again.
|
||||
expect(providerCalls.match(/^turn\/start$/gm)).toHaveLength(2);
|
||||
expect(providerCalls.match(/^thread\/start$/gm)).toHaveLength(2);
|
||||
const quarantine = join(stateBase, "quarantine");
|
||||
const quarantined = existsSync(quarantine) ? await readdir(quarantine) : [];
|
||||
expect(quarantined.every(name => !name.startsWith(retried.root.split("/").at(-1)!))).toBe(true);
|
||||
} finally {
|
||||
for (const bundle of bundles) {
|
||||
await bundle.transport.close().catch(() => undefined);
|
||||
const evidence = bundle.evidence();
|
||||
kill(evidence.providerPid);
|
||||
kill(evidence.runnerPid);
|
||||
}
|
||||
vi.unstubAllEnvs();
|
||||
await rm(stateBase, { recursive: true, force: true });
|
||||
}
|
||||
|
||||
async function admitRun(runId: string, nativeSessionId: string | null, retryOf?: string) {
|
||||
const [row] = await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, status: "running",
|
||||
runtimeMode: "native", nativeIssueId: issueId, nativeSessionId, invocationSource: "on_demand", triggerDetail: "manual",
|
||||
runnerInstanceId: nativeSessionId ? fixture.native.runnerInstanceId : null,
|
||||
retryOfRunId: retryOf ?? null,
|
||||
contextSnapshot: { issueId, ...(retryOf ? { forceFreshSession: true, previousRunId: retryOf } : {}) } })
|
||||
.returning();
|
||||
await db.update(issues).set({ executionRunId: runId }).where(eq(issues.id, issueId));
|
||||
const native = await prepareNativeHeartbeatRun({ db, run: row!,
|
||||
issue: { id: issueId, title: issue!.title, description: null, reviewPolicy: null },
|
||||
environmentLeaseId: `lease-${runId}` });
|
||||
await db.insert(nativeRunFinalizations).values({ runId, companyId, issueId, phase: "observed" });
|
||||
await db.update(heartbeatRuns).set({ runnerProfileJson: { ...(row!.runnerProfileJson ?? {}),
|
||||
nativeExecutionInput: executionFor(runId, native.normalizedSessionId) } }).where(eq(heartbeatRuns.id, runId));
|
||||
return native;
|
||||
}
|
||||
}, 120_000);
|
||||
});
|
||||
@@ -645,7 +645,7 @@ describe.skipIf(!recoveryPostgresSupport.supported)("follow-up after a recovered
|
||||
await db.update(heartbeatRuns).set({ processPid: 4_242 }).where(eq(heartbeatRuns.id, failed));
|
||||
}
|
||||
if (scenario === "provider event") {
|
||||
await db.insert(heartbeatRunEvents).values({ companyId, agentId, runId: failed, seq: 1, eventType: "session.started",
|
||||
await db.insert(heartbeatRunEvents).values({ companyId, agentId, runId: failed, seq: 100, eventType: "session.started",
|
||||
stream: "system", payload: {} });
|
||||
}
|
||||
if (scenario === "controller checkpoint") {
|
||||
|
||||
@@ -115,7 +115,7 @@ import { stoppedCodexTurnIsTextOnly } from "./stopped-codex-turn.js";
|
||||
import { prepareVerifiedRemoteProviderPack, useVerifiedPreinstalledRemoteProviderPack, shouldUseBundledAcpxImageAssets } from "./remote-provider-pack.js";
|
||||
import { uploadVerifiedProviderPack } from "./provider-pack-upload.js";
|
||||
import { selectRemotePiCompanion } from "./remote-pi-companion.js";
|
||||
import { readNativeLocalProcessStop, recordedLocalExecutionStopped, PROCESS_IDENTITY_RECORDED, PROCESS_START_REQUESTED, withReleasedNativeRunProcess } from "../native-local-process-stop.js";
|
||||
import { readNativeLocalProcessStop, recordedLocalExecutionStopped, PRIOR_OWNER_FENCE_REFUSED, PROCESS_IDENTITY_RECORDED, PROCESS_START_REQUESTED, withReleasedNativeRunProcess } from "../native-local-process-stop.js";
|
||||
import { remoteLeaseCleanupScope } from "../remote-execution-termination.js";
|
||||
import { connectorToolDefinitionsForSkills } from "../connector-runtime.js";
|
||||
import { NATIVE_PREPARATION_FENCING_VERSION, setNativeExecutionActive } from "./native-active-runs.js";
|
||||
@@ -3143,6 +3143,28 @@ export function rebaseRetainedNativeCleanupProviderHome(
|
||||
* Retain the old durable root permanently; only the exact in-memory cleanup
|
||||
* owner is retired, and the successor must use a fresh normalized session.
|
||||
*/
|
||||
/** A provider launch that failed its own admission (for example a restore
|
||||
* that refused an ambiguous turn start), whose direct child exited and whose
|
||||
* process and process group are gone, owns nothing. Any other recorded
|
||||
* launch may own a process that never reported its session identity. */
|
||||
function failedStartupRetired(attempt: unknown): boolean {
|
||||
if (attempt == null) return true;
|
||||
const startup = record(attempt);
|
||||
return startup.phase === "initialization_failed" && startup.directChildExitObserved === true &&
|
||||
cleanupProcessAbsent(startup.processId) &&
|
||||
(startup.processGroupId == null || cleanupProcessAbsent(startup.processGroupId));
|
||||
}
|
||||
|
||||
/** The session's state root, or, when only the fence's quarantine holds it,
|
||||
* that one `identity_indeterminate` copy. Several copies are ambiguous. */
|
||||
function quarantinedOwnerStateRoot(scoped: string): string {
|
||||
if (existsSync(scoped)) return scoped;
|
||||
const quarantine = resolve(runnerdStateBase(), "quarantine");
|
||||
if (!isSafeNativeStateDirectory(quarantine)) return scoped;
|
||||
const copies = readdirSync(quarantine).filter(name => name.startsWith(`${basename(scoped)}.identity_indeterminate.`));
|
||||
return copies.length === 1 ? resolve(quarantine, copies[0]!) : scoped;
|
||||
}
|
||||
|
||||
export async function verifyStoppedNativeSessionForContinuation(
|
||||
db: Db,
|
||||
run: Pick<typeof heartbeatRuns.$inferSelect, "id" | "companyId" | "agentId" | "runtimeMode" |
|
||||
@@ -3165,14 +3187,21 @@ export async function verifyStoppedNativeSessionForContinuation(
|
||||
eq(environmentLeases.companyId, run.companyId), eq(environmentLeases.heartbeatRunId, run.id)));
|
||||
if (leases.some(lease => lease.provider !== "local" || !lease.releasedAt ||
|
||||
lease.status === "pending_cleanup" || lease.cleanupStatus === "failed")) return null;
|
||||
const root = scopedRunnerdStateRoot(execution);
|
||||
// A run refused at the prior-owner fence moved the owner's state to
|
||||
// quarantine before it failed. That single exact copy is the owner's state.
|
||||
const root = quarantinedOwnerStateRoot(scopedRunnerdStateRoot(execution));
|
||||
const providerFile = runnerProviderStateFilename(execution);
|
||||
const snapshot = cleanupStateSnapshot(root, providerFile);
|
||||
const identity = record(snapshot.control.identity);
|
||||
let owner = run;
|
||||
let ownerExecution = execution;
|
||||
if (identity.runId !== run.id) {
|
||||
if (run.errorCode !== "native_session_cleanup_quarantined" || typeof identity.runId !== "string") return null;
|
||||
// Only a run refused before launch can stand in for the saved state's
|
||||
// owner: the cleanup quarantine or the prior-owner runner-state fence.
|
||||
const [fenced] = run.errorCode === "native_session_cleanup_quarantined" ? [] : await db.select({ id: heartbeatRunEvents.id })
|
||||
.from(heartbeatRunEvents).where(and(eq(heartbeatRunEvents.companyId, run.companyId), eq(heartbeatRunEvents.runId, run.id),
|
||||
isNull(heartbeatRunEvents.sourceEventId), eq(heartbeatRunEvents.eventType, PRIOR_OWNER_FENCE_REFUSED))).limit(1);
|
||||
if ((run.errorCode !== "native_session_cleanup_quarantined" && !fenced) || typeof identity.runId !== "string") return null;
|
||||
const [started] = await db.select({ id: heartbeatRunEvents.id }).from(heartbeatRunEvents).where(and(
|
||||
eq(heartbeatRunEvents.companyId, run.companyId), eq(heartbeatRunEvents.runId, run.id),
|
||||
inArray(heartbeatRunEvents.eventType, [PROCESS_START_REQUESTED, PROCESS_IDENTITY_RECORDED,
|
||||
@@ -3230,7 +3259,7 @@ export async function verifyStoppedNativeSessionForContinuation(
|
||||
snapshot.runner.schema !== RUNNERD_STATE_SCHEMA ||
|
||||
snapshot.provider.schema !== (execution.provider.kind === "codex" ? "paperclip.runner.codex-provider-state.v1" : ACPX_PROVIDER_STATE_SCHEMA) ||
|
||||
!["turn_active", "prepared", "suspended", "session_open"].includes(String(snapshot.provider.lifecycle)) ||
|
||||
snapshot.provider.startupAttempt != null) return null;
|
||||
!failedStartupRetired(snapshot.provider.startupAttempt)) return null;
|
||||
const pending = JSON.stringify([snapshot.runner.outbox, snapshot.provider.pendingEvents, snapshot.provider.queuedEvents]);
|
||||
if (/session\.(started|resumed|reconciled)/.test(pending)) return null;
|
||||
const events = snapshot.control.committedEvents.map(entry => record(record(record(entry).envelope).payload));
|
||||
@@ -3290,6 +3319,8 @@ export async function verifyStoppedNativeSessionForContinuation(
|
||||
if (!cleanupProcessAbsent(pid)) return null;
|
||||
providerPids.add(pid);
|
||||
}
|
||||
const startup = record(snapshot.provider.startupAttempt);
|
||||
for (const pid of [startup.processId, startup.processGroupId]) if (pid != null) providerPids.add(pid as number);
|
||||
const unchanged = () => idle() && !activeNativeSessions.has(owner.id) && cleanupProcessAbsent(stopped.processPid) &&
|
||||
[...providerPids].every(cleanupProcessAbsent) && cleanupStateSnapshot(root, providerFile).fingerprint === snapshot.fingerprint;
|
||||
return {
|
||||
@@ -5761,6 +5792,20 @@ async function rollBackUndeliveredAttachRotation(root: string, execution: Native
|
||||
`[paperclip-runner] Run ${execution.binding.runId} stopped before its runner received the session; resuming from the previous turn.\n`);
|
||||
}
|
||||
|
||||
/** The fence refused this run before any launch, because another run's saved
|
||||
* state could not be proven finished. Retry and new input read this event as
|
||||
* proof that the run never had a runner (native-local-process-stop.ts). A
|
||||
* failed write leaves the run held, which is the safe side. */
|
||||
async function recordPriorOwnerFenceRefusal(input: { db: Db; execution: NativeExecutionInput }, reason: string): Promise<void> {
|
||||
const { binding } = input.execution;
|
||||
await appendHeartbeatRunEvent(input.db, {
|
||||
companyId: binding.companyId, runId: binding.runId, agentId: binding.agentId,
|
||||
eventType: PRIOR_OWNER_FENCE_REFUSED, stream: "system", level: "warn",
|
||||
message: "The saved runner state of a previous run could not be proven finished; this run started no runner.",
|
||||
payload: { reason },
|
||||
}).catch(() => undefined);
|
||||
}
|
||||
|
||||
async function migrateRunnerdStateRootForExecution(input: {
|
||||
db: Db;
|
||||
execution: NativeExecutionInput;
|
||||
@@ -5949,6 +5994,7 @@ async function migrateRunnerdStateRootForExecution(input: {
|
||||
: "identity_indeterminate",
|
||||
);
|
||||
}
|
||||
await recordPriorOwnerFenceRefusal(input, `prior_owner_${verification}`);
|
||||
throw new Error(`runner_state_identity_mismatch: prior_owner_${verification}${retirementRefusal}`);
|
||||
}
|
||||
}
|
||||
@@ -5990,6 +6036,7 @@ async function migrateRunnerdStateRootForExecution(input: {
|
||||
// untouched because the legacy name may still belong to them.
|
||||
quarantineRunnerdStateRoot(legacy, "identity_indeterminate");
|
||||
}
|
||||
await recordPriorOwnerFenceRefusal(input, `legacy_prior_owner_${verification}`);
|
||||
throw new Error(`runner_state_identity_mismatch: legacy_prior_owner_${verification}`);
|
||||
}
|
||||
verifiedPriorRunId = identity.runId;
|
||||
|
||||
Reference in new issue
Block a user