diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index f83ff2811ee..3f0ff6d81ab 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -1903,7 +1903,13 @@ All of these must be true: The provider session then follows the normal resume-or-replace policy. A resumed sandbox, a warm session, a plugin task, a restart claim, or any missing proof keeps the existing `prior_owner_terminal_state_indeterminate` -rejection. +rejection. When admission attempted the retirement, the error names the first +proof that failed, for example +`prior_owner_terminal_state_indeterminate; retirement_refused=owner_observed` or +`retirement_refused=journal_commands_unsettled(runner.suspend.pending)`. The +reason uses only fixed lists of known names: command types, command statuses, +owner phases, recovery states, lifecycle states, and fence reasons. Any other +stored value appears as `other`. The reason never includes payloads or paths. Resume source selection skips a failed follow-up that never started provider work. Such a run holds only the provider checkpoint that it inherited. Without 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 0e5ff1b62e9..96c4bd9290e 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 @@ -44,6 +44,8 @@ import { createRunnerdBackend } from "./native-session-executor.js"; // authority proves admission passed; no provider or runner process starts. const ADMITTED = "follow-up admission reached tool authority"; const STRANDED = "runner_state_identity_mismatch: prior_owner_terminal_state_indeterminate"; +/** The held follow-up names the retirement proof that failed. */ +const stranded = (refused: string) => `${STRANDED}; retirement_refused=${refused}`; describe.skipIf(!recoveryPostgresSupport.supported)("follow-up after a recovered per-turn native owner", () => { let database: Awaited>; @@ -223,7 +225,7 @@ describe.skipIf(!recoveryPostgresSupport.supported)("follow-up after a recovered async function keepLocalRootWithoutPointer(prior: Awaited>, followUp: Awaited>) { await db.update(nativeRunFinalizations).set({ phase: "observed" }).where(eq(nativeRunFinalizations.runId, prior.runId)); - expect(await followUp.start()).toBe(STRANDED); + expect(await followUp.start()).toBe(stranded("owner_observed")); const [copy] = await quarantined(); await rename(join(stateDirectory, "quarantine", copy!), join(stateDirectory, copy!.split(".")[0]!)); await rm(join(stateDirectory, "quarantine"), { recursive: true, force: true }); @@ -264,7 +266,7 @@ describe.skipIf(!recoveryPostgresSupport.supported)("follow-up after a recovered const prior = await seedRecoveredPriorRun("settled", { controller }); const followUp = await admitFollowUp(prior); await keepLocalRootWithoutPointer(prior, followUp); - expect(await followUp.start()).toBe(STRANDED); + expect(await followUp.start()).toMatch(/; retirement_refused=controller_/); expect(followUp.logs.join("")).not.toContain("finished without a saved runner checkpoint"); } finally { child.kill("SIGKILL"); @@ -284,7 +286,7 @@ describe.skipIf(!recoveryPostgresSupport.supported)("follow-up after a recovered await db.update(nativeRunFinalizations).set({ phase: "committed" }).where(eq(nativeRunFinalizations.runId, prior.runId)); } })(); - expect(error).toBe(STRANDED); + expect(error).toBe(stranded("owner_observed")); await db.update(heartbeatRuns).set({ status: "failed", errorCode: "adapter_failed", error, finishedAt: new Date(), nativePhase: "observed", resultJson: { executionRecovery: { kind: "bootstrap", providerWorkStarted: false } } }) .where(eq(heartbeatRuns.id, followUp.runId)); @@ -371,7 +373,7 @@ describe.skipIf(!recoveryPostgresSupport.supported)("follow-up after a recovered const prior = await seedRecoveredPriorRun(); const followUp = await admitFollowUp(prior); await db.update(nativeRunFinalizations).set({ phase: "observed" }).where(eq(nativeRunFinalizations.runId, prior.runId)); - expect(await followUp.start()).toBe(STRANDED); + expect(await followUp.start()).toBe(stranded("owner_observed")); expect(await quarantined()).toHaveLength(1); await db.update(nativeRunFinalizations).set({ phase: "committed" }).where(eq(nativeRunFinalizations.runId, prior.runId)); expect(await followUp.start()).toBe(ADMITTED); @@ -384,7 +386,7 @@ describe.skipIf(!recoveryPostgresSupport.supported)("follow-up after a recovered // Learn the scope key from a held attempt, then give the active root the // local evidence of an owner whose Runner never suspended. await db.update(nativeRunFinalizations).set({ phase: "observed" }).where(eq(nativeRunFinalizations.runId, prior.runId)); - expect(await followUp.start()).toBe(STRANDED); + expect(await followUp.start()).toBe(stranded("owner_observed")); const [copy] = await quarantined(); const root = join(stateDirectory, copy!.split(".")[0]!); await rename(join(stateDirectory, "quarantine", copy!), root); @@ -392,16 +394,16 @@ describe.skipIf(!recoveryPostgresSupport.supported)("follow-up after a recovered await writeFile(join(root, "runner", "runner-state.json"), JSON.stringify({ schema: "paperclip.runner.durable.state.v1", lifecycle: "ready", ...prior.identity })); await db.update(nativeRunFinalizations).set({ phase: "committed" }).where(eq(nativeRunFinalizations.runId, prior.runId)); - expect(await followUp.start()).toBe(STRANDED); + expect(await followUp.start()).toBe(stranded("runner_lifecycle_not_suspended")); expect(await quarantined()).toHaveLength(1); // The retry restores only the database journal. The quarantined receipt // must still hold it. - expect(await followUp.start()).toBe(STRANDED); + expect(await followUp.start()).toBe(stranded("quarantine_runner_lifecycle_not_suspended")); expect(await quarantined()).toHaveLength(2); expect(followUp.logs.join("")).not.toContain("finished without a saved runner checkpoint"); }, 60_000); - it.each(["uncommitted result", "live controller lease", "unexpired controller lease", "recovery pending", "pending command", + it.each(["uncommitted result", "committed without result", "live controller lease", "unexpired controller lease", "recovery pending", "pending command", "unanswered tool input", "foreign quarantine"] as const)("keeps the prior owner held with %s", async scenario => { const prior = await seedRecoveredPriorRun(scenario === "pending command" ? "pending-command" : scenario === "unanswered tool input" ? "unanswered-tool" : "settled"); @@ -411,6 +413,9 @@ describe.skipIf(!recoveryPostgresSupport.supported)("follow-up after a recovered await db.update(nativeRunFinalizations).set({ phase: "retryable_failure", resultId: null }) .where(eq(nativeRunFinalizations.runId, prior.runId)); } + if (scenario === "committed without result") { + await db.update(nativeRunFinalizations).set({ resultId: null }).where(eq(nativeRunFinalizations.runId, prior.runId)); + } if (scenario === "live controller lease") { await db.update(nativeRunFinalizations).set({ leaseOwner: "another-controller", leaseExpiresAt: new Date(Date.now() + 60_000) }) .where(eq(nativeRunFinalizations.runId, prior.runId)); @@ -427,7 +432,7 @@ describe.skipIf(!recoveryPostgresSupport.supported)("follow-up after a recovered // A copy of this projection is harmless; a different owner saved under // the same scope is unresolved evidence that must not be overtaken. await db.update(nativeRunFinalizations).set({ phase: "observed" }).where(eq(nativeRunFinalizations.runId, prior.runId)); - expect(await followUp.start()).toBe(STRANDED); + expect(await followUp.start()).toBe(stranded("owner_observed")); const [copy] = await quarantined(); const foreign = join(stateDirectory, "quarantine", copy!.replace(/\.[0-9a-f-]{36}$/, `.${randomUUID()}`)); await cp(join(stateDirectory, "quarantine", copy!), foreign, { recursive: true }); @@ -438,8 +443,18 @@ describe.skipIf(!recoveryPostgresSupport.supported)("follow-up after a recovered expectedQuarantine = 3; } const outcome = await followUp.start(); - // A live owner is still waited for before any projection is rebuilt. - expect(outcome).toBe(scenario === "live controller lease" ? "native_controller_recovery_waiting_for_owner" : STRANDED); + // A live owner is still waited for before any projection is rebuilt. Every + // other hold names its failed proof instead of one indeterminate state. + expect(outcome).toBe({ + "uncommitted result": stranded("owner_retryable_failure"), + "committed without result": stranded("owner_result_missing"), + "live controller lease": "native_controller_recovery_waiting_for_owner", + "unexpired controller lease": stranded("controller_remote_controller_lease_live"), + "recovery pending": stranded("owner_recovery_resuming_session"), + "pending command": stranded("journal_commands_unsettled(runner.suspend.pending)"), + "unanswered tool input": stranded("journal_tool_input_unanswered"), + "foreign quarantine": stranded("quarantine_journal_identity_mismatch"), + }[scenario]); expect(await quarantined()).toHaveLength(scenario === "live controller lease" ? 0 : expectedQuarantine); expect(followUp.logs.join("")).not.toContain("finished without a saved runner checkpoint"); }, 60_000); diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 70d41a6cc0b..7b5335794a0 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -5198,30 +5198,67 @@ function moveNativeContinuityState(root: string, archiveRoot: string): void { * evidence, even an unsuspended receipt, and unknown entries are owner * claims that this proof never overrides. */ -function readSettledRunnerdProjection(root: string, identity: Record): string | null { +type SettledRunnerdProjection = { bytes: string; refused?: never } | { bytes?: never; refused: string }; + +function readSettledRunnerdProjection(root: string, identity: Record): SettledRunnerdProjection { + const refuse = (refused: string): SettledRunnerdProjection => ({ refused }); try { - if (!isSafeNativeStateDirectory(root) || - runnerdAuthorityLifecycle(root, identity as RunnerdDurableIdentity) !== "absent" || - readdirSync(root).some(name => name !== "continuity-breaks" && !NATIVE_CONTINUITY_STATE_ENTRIES.includes(name))) { - return null; - } + if (!isSafeNativeStateDirectory(root)) return refuse("root_unsafe"); + const lifecycle = runnerdAuthorityLifecycle(root, identity as RunnerdDurableIdentity); + if (lifecycle !== "absent") return refuse(`runner_lifecycle_${lifecycle}`); + const unexpected = readdirSync(root).find(name => name !== "continuity-breaks" && !NATIVE_CONTINUITY_STATE_ENTRIES.includes(name)); + if (unexpected !== undefined) return refuse("unexpected_root_entry"); const bytes = readBoundedNativeFile(resolve(root, "control-plane", "control-plane-state.json"), NATIVE_CONTROL_PLANE_STATE_MAX_BYTES, "runner_durable_identity_too_large").toString("utf8"); const control = record(JSON.parse(bytes)); - const answered = new Set((Array.isArray(control.commands) ? control.commands : []).map(record) + const commands = (Array.isArray(control.commands) ? control.commands : []).map(record); + const answered = new Set(commands .filter(command => command.type === "semantic_tool.result" && command.status === "completed") .map(command => record(command.payload).sourceEventId)); - if (!RUNNERD_CONTROL_PLANE_STATE_SCHEMAS.has(String(control.schema)) || control.warmTransition != null || - control.stagedRunAttachment != null || !isSettledRemoteWarmControlState(control, identity) || - (control.committedEvents as unknown[]).map(record).some(event => - (event.eventType === "semantic_tool.input" || event.eventType === "mcp_app.tool_input") && - !answered.has(event.sourceEventId))) return null; - return bytes; + if (!RUNNERD_CONTROL_PLANE_STATE_SCHEMAS.has(String(control.schema))) return refuse("journal_schema_unsupported"); + if (control.warmTransition != null) return refuse("journal_warm_transition"); + if (control.stagedRunAttachment != null) return refuse("journal_staged_attachment"); + if (Object.entries(identity).some(([key, value]) => record(control.identity)[key] !== value)) { + return refuse("journal_identity_mismatch"); + } + if (!isSettledRemoteWarmControlState(control, identity)) { + // Closed vocabulary only: known command types and statuses, never payloads. + const pending = commands.filter(command => command.status !== "completed") + .map(command => `${retirementReasonName("commandType", command.type)}.${retirementReasonName("commandStatus", command.status)}`); + return refuse(pending.length > 0 ? `journal_commands_unsettled(${[...new Set(pending)].slice(0, 4).join(",")})` + : "journal_terminal_unsettled"); + } + if ((control.committedEvents as unknown[]).map(record).some(event => + (event.eventType === "semantic_tool.input" || event.eventType === "mcp_app.tool_input") && + !answered.has(event.sourceEventId))) return refuse("journal_tool_input_unanswered"); + return { bytes }; } catch { - return null; + return refuse("journal_unreadable"); } } +// Refusal reasons reach run errors, so they draw only from these closed sets. +// Any other persisted value reports as "other". +const RETIREMENT_REASON_NAMES = { + // Mirrors the durable controller's stored-journal contract (commandTypes and + // the command status union in paperclip-runner's durable PRP control plane). + commandType: new Set(["run.prepare", "run.attach", "session.open", "turn.start", "turn.steer", "turn.interrupt", + "turn.stop", "request.resolve", "interaction.receipt", "semantic_tool.result", "external_provider.operation", + "session.snapshot", "session.goal.get", "session.goal.set", "session.goal.clear", "session.close", + "session.budget.increase", "session.destroy", "run.cancel", "runner.drain", "runner.suspend", "runner.shutdown"]), + commandStatus: new Set(["pending", "failed", "rejected", "indeterminate"]), + ownerPhase: new Set(["observed", "retryable_failure", "terminal_failure", "workspace_finalizing", + "recovery_needed"]), + recoveryState: new Set(["blocked", "resuming_session", "awaiting_evidence", "awaiting_runner_reattach"]), + controllerReason: new Set(["controller_still_alive", "coordinated_controller_still_alive", "remote_controller_lease_live", + "expired_lease_without_controller_identity", "live_legacy_lease_without_controller_identity"]), + sandboxOutcome: new Set(["created", "resumed", "replacement"]), +} as const; + +function retirementReasonName(kind: keyof typeof RETIREMENT_REASON_NAMES, value: unknown): string { + return typeof value === "string" && RETIREMENT_REASON_NAMES[kind].has(value) ? value : "other"; +} + /** * A per-turn sandbox turn can finish through recovery after its post-turn * checkpoint copy failed: the result is committed and the controller lease is @@ -5240,21 +5277,27 @@ async function retireFinishedPerTurnRunnerdOwner(input: { execution: NativeExecutionInput; runnerExecutionTarget?: AdapterExecutionTarget | null; onLog?: (stream: "stdout" | "stderr", chunk: string) => Promise; -}): Promise { +}): Promise<{ retired: true } | { retired: false; refused: string }> { const target = input.runnerExecutionTarget; // A resumed sandbox still holds the prior Runner's own lifecycle, which its // existing recovery path reads. Plugin tasks own a separate saved checkpoint. - if (input.execution.session.lifecyclePolicy.mode !== "per_turn" || target?.kind !== "remote" || - target.transport !== "sandbox" || !target.runner || target.runnerTask || - !["created", "replacement"].includes(target.sandboxLeaseAcquisition?.outcome ?? "")) return false; + const refuse = (refused: string) => ({ retired: false as const, refused }); + if (input.execution.session.lifecyclePolicy.mode !== "per_turn") return refuse("lifecycle_not_per_turn"); + if (target?.kind !== "remote" || target.transport !== "sandbox" || !target.runner || target.runnerTask) { + return refuse("target_not_sandbox_runner"); + } + if (!["created", "replacement"].includes(target.sandboxLeaseAcquisition?.outcome ?? "")) { + return refuse(`sandbox_${retirementReasonName("sandboxOutcome", target.sandboxLeaseAcquisition?.outcome)}`); + } const identity: Record = {}; for (const key of ["runId", "normalizedSessionId", "runnerInstanceId", "environmentLeaseId", "turnId", "itemId"]) { const value = input.identity[key]; - if (typeof value !== "string" || !value) return false; + if (typeof value !== "string" || !value) return refuse("durable_identity_incomplete"); identity[key] = value; } - const controlBytes = readSettledRunnerdProjection(input.root, identity); - if (controlBytes === null) return false; + const projection = readSettledRunnerdProjection(input.root, identity); + if (projection.refused !== undefined) return refuse(projection.refused); + const controlBytes = projection.bytes; const { binding } = input.execution; let prior: { run: typeof heartbeatRuns.$inferSelect; owner: typeof nativeRunFinalizations.$inferSelect } | undefined; try { @@ -5266,36 +5309,39 @@ async function retireFinishedPerTurnRunnerdOwner(input: { eq(nativeRunFinalizations.issueId, binding.issueId))) .limit(1); } catch { - return false; - } - if (!prior || prior.run.runtimeMode !== "native" || !TERMINAL_HEARTBEAT_RUN_STATUSES.has(prior.run.status) || - prior.owner.phase !== "committed" || !prior.owner.resultId || prior.owner.leaseOwner || prior.owner.recoveryState) { - return false; + return refuse("owner_unavailable"); } + if (!prior || prior.run.runtimeMode !== "native") return refuse("owner_missing"); + if (!TERMINAL_HEARTBEAT_RUN_STATUSES.has(prior.run.status)) return refuse("owner_run_not_terminal"); + if (prior.owner.phase !== "committed") return refuse(`owner_${retirementReasonName("ownerPhase", prior.owner.phase)}`); + if (!prior.owner.resultId) return refuse("owner_result_missing"); + if (prior.owner.leaseOwner) return refuse("owner_lease_held"); + if (prior.owner.recoveryState) return refuse(`owner_recovery_${retirementReasonName("recoveryState", prior.owner.recoveryState)}`); // A durable controller is fenced by its released database lease; a legacy // controller still needs process evidence, exactly as for warm recovery. const finishedHere = () => isCurrentNativeControllerProcess(prior!.owner) && !activeNativeSessions.has(identity.runId) && !nativeSessionStartups.has(identity.runId); const profile = record(prior.run.runnerProfileJson); - const stopped = finishedHere() ? { allowed: true } : await evaluateNativeControllerTakeover({ owner: prior.owner, + const stopped: { allowed: boolean; reason?: string } = finishedHere() ? { allowed: true } : await evaluateNativeControllerTakeover({ owner: prior.owner, now: new Date(), requireStoppedController: true, remoteLeaseFenced: readNativeControllerRecovery(profile.nativeControllerRecovery, profile.sessionCheckpoint) !== null }); - if (!stopped.allowed) return false; + if (!stopped.allowed) return refuse(`controller_${retirementReasonName("controllerReason", stopped.reason)}`); const provenFinishedHere = finishedHere(); // Earlier rejected admissions can quarantine the same projection. Each copy // must pass the same proof: a rebuilt root never carries the local Runner // evidence that held an earlier attempt, so that evidence stays decisive. const quarantine = resolve(runnerdStateBase(), "quarantine"); if (existsSync(quarantine)) { - if (!isSafeNativeStateDirectory(quarantine)) return false; + if (!isSafeNativeStateDirectory(quarantine)) return refuse("quarantine_unsafe"); for (const name of readdirSync(quarantine)) { - if (name.startsWith(`${basename(input.root)}.`) && - readSettledRunnerdProjection(resolve(quarantine, name), identity) === null) return false; + if (!name.startsWith(`${basename(input.root)}.`)) continue; + const copy = readSettledRunnerdProjection(resolve(quarantine, name), identity); + if (copy.refused !== undefined) return refuse(`quarantine_${copy.refused}`); } } // Database reads can await. Retire only the exact evidence proven above. - if (readSettledRunnerdProjection(input.root, identity) !== controlBytes || - (provenFinishedHere && !finishedHere())) return false; + if (readSettledRunnerdProjection(input.root, identity).bytes !== controlBytes) return refuse("journal_changed"); + if (provenFinishedHere && !finishedHere()) return refuse("controller_restarted_here"); const archivedAt = Date.now(); const archiveRoot = resolve(input.root, "continuity-breaks", `${archivedAt}-${randomUUID()}`); mkdirSync(archiveRoot, { recursive: true, mode: 0o700 }); @@ -5305,7 +5351,7 @@ async function retireFinishedPerTurnRunnerdOwner(input: { "stdout", `[paperclip-runner] Run ${identity.runId} finished without a saved runner checkpoint; starting a new runner for this conversation.\n`, ); - return true; + return { retired: true }; } async function migrateRunnerdStateRootForExecution(input: { @@ -5445,13 +5491,13 @@ async function migrateRunnerdStateRootForExecution(input: { allowVerifiedBackup: input.allowVerifiedBackup, retainedWarmOwnerRunId: input.retainedWarmOwnerRunId, }); - if ( - verification === "terminal_state_indeterminate" && - !input.restartRecovery && - !input.retainedWarmOwnerRunId && - (await retireFinishedPerTurnRunnerdOwner({ ...input, root: scoped, identity })) - ) { - return; + // Name the failed retirement proof: the shared error otherwise hides + // which of several independent fences held the follow-up. + let retirementRefusal = ""; + if (verification === "terminal_state_indeterminate" && !input.restartRecovery && !input.retainedWarmOwnerRunId) { + const retirement = await retireFinishedPerTurnRunnerdOwner({ ...input, root: scoped, identity }); + if (retirement.retired) return; + if (retirement.refused !== "lifecycle_not_per_turn") retirementRefusal = `; retirement_refused=${retirement.refused}`; } if ( verification !== "verified" && @@ -5465,7 +5511,7 @@ async function migrateRunnerdStateRootForExecution(input: { : "identity_indeterminate", ); } - throw new Error(`runner_state_identity_mismatch: prior_owner_${verification}`); + throw new Error(`runner_state_identity_mismatch: prior_owner_${verification}${retirementRefusal}`); } } return;