diff --git a/doc/run-log-events.md b/doc/run-log-events.md index 36a413d3ff8..7c24efe7ba1 100644 --- a/doc/run-log-events.md +++ b/doc/run-log-events.md @@ -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, diff --git a/packages/paperclip-runner/docs/durable-recovery.md b/packages/paperclip-runner/docs/durable-recovery.md index ad5c6d6a09d..e89d8bc438a 100644 --- a/packages/paperclip-runner/docs/durable-recovery.md +++ b/packages/paperclip-runner/docs/durable-recovery.md @@ -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 diff --git a/scripts/__tests__/run-vitest-stable-shard.test.mjs b/scripts/__tests__/run-vitest-stable-shard.test.mjs index 260fd83afee..c620041ecba 100644 --- a/scripts/__tests__/run-vitest-stable-shard.test.mjs +++ b/scripts/__tests__/run-vitest-stable-shard.test.mjs @@ -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, diff --git a/scripts/run-vitest-stable.mjs b/scripts/run-vitest-stable.mjs index 0c462a2750f..e2135993fab 100644 --- a/scripts/run-vitest-stable.mjs +++ b/scripts/run-vitest-stable.mjs @@ -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", diff --git a/server/src/services/explicit-native-continuation.test.ts b/server/src/services/explicit-native-continuation.test.ts index fb48824208a..9174f919a9c 100644 --- a/server/src/services/explicit-native-continuation.test.ts +++ b/server/src/services/explicit-native-continuation.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); diff --git a/server/src/services/explicit-native-continuation.ts b/server/src/services/explicit-native-continuation.ts index d831683c307..7232aa726ec 100644 --- a/server/src/services/explicit-native-continuation.ts +++ b/server/src/services/explicit-native-continuation.ts @@ -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)) diff --git a/server/src/services/native-local-process-stop.ts b/server/src/services/native-local-process-stop.ts index 81c0bc48469..3b862654348 100644 --- a/server/src/services/native-local-process-stop.ts +++ b/server/src/services/native-local-process-stop.ts @@ -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 { * 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 { * 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, 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): Promise { + "nativeSessionId" | "runnerInstanceId" | "nativeIssueId">): Promise { 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. diff --git a/server/src/services/native-runtime/native-cut-turn-start-retry.integration.test.ts b/server/src/services/native-runtime/native-cut-turn-start-retry.integration.test.ts new file mode 100644 index 00000000000..897af405700 --- /dev/null +++ b/server/src/services/native-runtime/native-cut-turn-start-retry.integration.test.ts @@ -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[] = []; + 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); +}); diff --git a/server/src/services/native-runtime/native-recovered-owner-followup.test.ts b/server/src/services/native-runtime/native-recovered-owner-followup.test.ts index 5e33e16a758..f27f169fdd5 100644 --- a/server/src/services/native-runtime/native-recovered-owner-followup.test.ts +++ b/server/src/services/native-runtime/native-recovered-owner-followup.test.ts @@ -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") { diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index e15be30eaad..aad0b18f0aa 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -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 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 { + 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;