diff --git a/server/src/services/native-runtime/native-session-executor.test.ts b/server/src/services/native-runtime/native-session-executor.test.ts index 4e76dd2b69c..ef43bf02e7f 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -14022,7 +14022,7 @@ describe("runnerd provider runtime wiring", () => { || request.args.includes("paperclip-runner-claim-unstarted-session"))).toBe(false); }); - it.each(["claude", "codex"].flatMap(provider => (["per_turn", "warm"] as const).flatMap(policy => ["canonical", "resumed", "resumed-timeout", "local-receipt", "local-receipt-missing-terminal", "local-receipt-pending-command", "quarantined", "empty-runner", "empty-runner-quarantined", "nonempty-runner", "wrong-identity", "active", "foreign-scope", "no-stop-proof", "not-suspended", "pending-outbox", "pending-command", "missing-terminal", "ambiguous", "contradictory"].filter(mode => policy === "per_turn" || !mode.startsWith("resumed")).map(mode => [provider, policy, mode] as const))))( + it.each(["claude", "codex"].flatMap(provider => (["per_turn", "warm"] as const).flatMap(policy => ["canonical", "resumed", "resumed-timeout", "local-receipt", "local-receipt-late-notice", "local-receipt-late-work", "local-receipt-missing-terminal", "local-receipt-pending-command", "quarantined", "empty-runner", "empty-runner-quarantined", "nonempty-runner", "wrong-identity", "active", "foreign-scope", "no-stop-proof", "not-suspended", "pending-outbox", "pending-command", "missing-terminal", "ambiguous", "contradictory"].filter(mode => policy === "per_turn" || !mode.startsWith("resumed")).map(mode => [provider, policy, mode] as const))))( "recovers a blocked %s %s plugin checkpoint on a replacement lease: %s", async (provider, policy, mode) => { const { permissionPolicy: _permissionPolicy, ...claudeProfile } = QUALIFIED_ACPX_PROFILES.claude; const first = { ...scopedNativeExecution(`plugin-${provider}-${policy}-${mode}`), @@ -14038,7 +14038,12 @@ describe("runnerd provider runtime wiring", () => { const scoped = await initializeScopedRunnerRoot(first, identity.runnerInstanceId); await mkdir(join(scoped, "control-plane"), { recursive: true }); const journal = JSON.stringify({ ...durableControlPlaneState(identity), commands: [{ status: mode.endsWith("pending-command") ? "pending" : "completed" }], - committedEvents: mode.endsWith("missing-terminal") ? [] : [{ eventType: "run.terminal", envelope: { ...identity, payload: { reportedWorkDisposition: "blocked" } } }] }); + committedEvents: mode.endsWith("missing-terminal") ? [] : [{ eventType: "run.terminal", envelope: { ...identity, payload: { reportedWorkDisposition: "blocked" } } }, + ...(mode === "local-receipt-late-notice" ? [{ eventType: "provider.notice.recorded", envelope: { ...identity, payload: { schema: "paperclip.prp.event.v1", eventType: "provider.notice.recorded", + runId: identity.runId, normalizedSessionId: identity.normalizedSessionId, sourceInstanceId: identity.runnerInstanceId, sourceKind: "runner", + payload: { schema: "paperclip.provider.notice.v1", noticeId: "codex-warning", severity: "warning", category: "warning", scope: "turn", recoverable: true, userActionable: true, + summary: "ignored a notification for a settled Codex turn" } } } }] : []), + ...(mode === "local-receipt-late-work" ? [{ eventType: "turn.started", envelope: { ...identity } }] : [])] }); await writeFile(join(scoped, "control-plane/control-plane-state.json"), journal); if (mode.startsWith("empty-runner") || mode === "nonempty-runner") { await mkdir(join(scoped, "runner")); @@ -14111,7 +14116,7 @@ describe("runnerd provider runtime wiring", () => { expect(await readFile(join(scoped, "control-plane/control-plane-state.json"), "utf8")).toBe(journal); } const backend = throughExecutor ? executePaperclipNativeSession({ ...launch, useRunnerd: true }) : createRunnerdBackend(launch); - if (["canonical", "resumed", "resumed-timeout", "local-receipt", "quarantined", "empty-runner", "empty-runner-quarantined"].includes(mode)) { + if (["canonical", "resumed", "resumed-timeout", "local-receipt", "local-receipt-late-notice", "quarantined", "empty-runner", "empty-runner-quarantined"].includes(mode)) { await expect(backend).resolves.toBeDefined(); expect(state.createBackend).toHaveBeenCalledOnce(); expect(owner).toHaveBeenCalledWith(expect.objectContaining({ priorIdentity: identity, @@ -14121,9 +14126,18 @@ describe("runnerd provider runtime wiring", () => { expect(JSON.parse(await readFile(join(scoped, "plugin-runner-state.json"), "utf8"))).toMatchObject({ ...identity, lifecycle: "suspended" }); expect(await readFile(join(scoped, "control-plane/control-plane-state.json"), "utf8")).toBe(journal); } else { - await expect(backend).rejects.toThrow(); + await expect(backend).rejects.toThrow({ + "local-receipt-late-work": "plugin_checkpoint_recovery_unproven: journal_not_settled", + "local-receipt-missing-terminal": "plugin_checkpoint_recovery_unproven: journal_not_settled", + "local-receipt-pending-command": "plugin_checkpoint_recovery_unproven: journal_not_settled", + "pending-command": "plugin_checkpoint_recovery_unproven: journal_not_settled", + "missing-terminal": "plugin_checkpoint_recovery_unproven: journal_not_settled", + "ambiguous": "plugin_checkpoint_recovery_unproven: quarantine_ambiguous", + "not-suspended": "plugin_checkpoint_recovery_unproven: checkpoint_not_settled", + "pending-outbox": "plugin_checkpoint_recovery_unproven: checkpoint_not_settled", + }[mode as string] ?? ""); expect(state.createBackend).not.toHaveBeenCalled(); - if (["local-receipt-missing-terminal", "local-receipt-pending-command", "active", "foreign-scope", "pending-command", "missing-terminal", "ambiguous", "contradictory", "nonempty-runner", "no-stop-proof"].includes(mode)) { + if (["local-receipt-late-work", "local-receipt-missing-terminal", "local-receipt-pending-command", "active", "foreign-scope", "pending-command", "missing-terminal", "ambiguous", "contradictory", "nonempty-runner", "no-stop-proof"].includes(mode)) { expect(task).not.toHaveBeenCalled(); } if (mode === "ambiguous") expect(await readFile(join(root, "control-plane/control-plane-state.json"), "utf8")).toBe(journal); @@ -15241,6 +15255,8 @@ describe("runnerd provider runtime wiring", () => { it.each([ "quarantined", "large control plane", + "late provider notice", + "post-terminal work", "empty retry shell", "unsuspended current", "live runner", @@ -15457,6 +15473,35 @@ describe("runnerd provider runtime wiring", () => { })) : []), { eventType: "run.terminal", envelope: identity }, + ...(scenario === "late provider notice" + ? [{ + eventType: "provider.notice.recorded", + envelope: { + ...identity, + payload: { + schema: "paperclip.prp.event.v1", + eventType: "provider.notice.recorded", + runId: identity.runId, + normalizedSessionId: identity.normalizedSessionId, + sourceInstanceId: identity.runnerInstanceId, + sourceKind: "runner", + payload: { + schema: "paperclip.provider.notice.v1", + noticeId: "codex-warning", + severity: "warning", + category: "warning", + scope: "turn", + recoverable: true, + userActionable: true, + summary: "ignored a notification for a settled Codex turn", + }, + }, + }, + }] + : []), + ...(scenario === "post-terminal work" + ? [{ eventType: "turn.started", envelope: identity }] + : []), ], }), ); @@ -15497,6 +15542,7 @@ describe("runnerd provider runtime wiring", () => { "quarantined", "released process identity", "large control plane", + "late provider notice", "empty retry shell", "unsuspended current", "active goal", diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index c152cb32e15..6bbff7593c7 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -5049,25 +5049,25 @@ async function recoverPluginRunnerCheckpoint(input: { target.sandboxLeaseAcquisition?.outcome === "resumed") return; let root = scoped; const quarantined = !existsSync(scoped); - const deny = (): never => { throw new Error("runner_state_identity_mismatch: plugin_checkpoint_recovery_unproven"); }; + const deny = (reason: string): never => { throw new Error(`runner_state_identity_mismatch: plugin_checkpoint_recovery_unproven: ${reason}`); }; if (quarantined) { const quarantine = resolve(runnerdStateBase(), "quarantine"); if (!existsSync(quarantine)) return; - if (!isSafeNativeStateDirectory(quarantine)) return deny(); + if (!isSafeNativeStateDirectory(quarantine)) return deny("quarantine_unsafe"); const candidates = readdirSync(quarantine).filter(name => name.startsWith(`${basename(scoped)}.identity_indeterminate.`)); if (candidates.length === 0) return; - if (candidates.length !== 1) return deny(); + if (candidates.length !== 1) return deny("quarantine_ambiguous"); root = resolve(quarantine, candidates[0]!); } if (!existsSync(root)) return; - if (!isSafeNativeStateDirectory(root)) return deny(); + if (!isSafeNativeStateDirectory(root)) return deny("root_unsafe"); if (hasRetainedWarmTransitionEvidence(root)) { - if (quarantined) return deny(); + if (quarantined) return deny("quarantined_warm_transition"); return; } const identity = readRunnerdDurableIdentity(root); if (!durableIdentityMatchesSession(identity, input.execution) || identity.runId === input.execution.binding.runId) { - if (quarantined) return deny(); + if (quarantined) return deny("quarantined_identity_mismatch"); return; } const direct = runnerdAuthorityLifecycle(root, identity); @@ -5075,32 +5075,29 @@ async function recoverPluginRunnerCheckpoint(input: { !existsSync(resolve(root, "runner", "runner-state.json")) && pluginRunnerdAuthorityLifecycle(root, identity) === "suspended"; // Never hide contradictory direct authority behind a remote checkpoint. if (direct !== "absent" && !(quarantined && direct === "suspended") && !verifiedPluginReceipt) { - if (quarantined) return deny(); + if (quarantined) return deny("quarantined_authority_conflict"); return; } await assertCleanupActivationCommitted(input.db, root, input.execution); const prior = await verifyPriorRunnerdStateForSessionScope({ ...input, root, identity, allowVerifiedBackup: true }); if (prior !== "terminal_state_indeterminate" && !((quarantined || verifiedPluginReceipt) && prior === "verified")) { - if (quarantined) return deny(); + if (quarantined) return deny("quarantined_prior_owner_unverified"); return; } const control = record(JSON.parse(readBoundedNativeFile(resolve(root, "control-plane", "control-plane-state.json"), NATIVE_CONTROL_PLANE_STATE_MAX_BYTES, "runner_durable_identity_too_large").toString("utf8"))); - const terminal = record(Array.isArray(control.committedEvents) ? control.committedEvents.at(-1) : null); - if (!Array.isArray(control.commands) || control.commands.some(command => record(command).status !== "completed") || - terminal.eventType !== "run.terminal" || record(terminal.envelope).runId !== identity.runId || - record(terminal.envelope).normalizedSessionId !== identity.normalizedSessionId) return deny(); + if (!isSettledRemoteWarmControlState(control, identity)) return deny("journal_not_settled"); const resumeFrom = await resolvePluginTaskResume(input.db, { companyId: input.execution.binding.companyId, leaseId: target.leaseId, priorIdentity: identity, inspectCheckpoint: true }); const checkpoint = environmentTaskResultSchema.parse(await target.runnerTask({ kind: "read_state", identity, resumeFrom })); if (checkpoint.kind !== "state" || checkpoint.state.lifecycle !== "suspended" || !Array.isArray(checkpoint.state.outbox) || checkpoint.state.outbox.length !== 0 || checkpoint.state.pendingTerminalDelivery != null || checkpoint.state.pendingProviderCleanup != null || - checkpoint.state.warmTransition != null) return deny(); + checkpoint.state.warmTransition != null) return deny("checkpoint_not_settled"); savePluginRunnerCheckpoint(root, checkpoint.state, identity); - if (await verifyPriorRunnerdStateForSessionScope({ ...input, root, identity, allowVerifiedBackup: true }) !== "verified") return deny(); + if (await verifyPriorRunnerdStateForSessionScope({ ...input, root, identity, allowVerifiedBackup: true }) !== "verified") return deny("checkpoint_unverified"); if (quarantined) { - if (existsSync(scoped)) return deny(); + if (existsSync(scoped)) return deny("quarantine_destination_exists"); renameSync(root, scoped); } } @@ -5517,11 +5514,7 @@ async function recoverQuiescentRunnerdState(input: { const runner = record(JSON.parse(runnerBytes)); const control = record(JSON.parse(controlBytes)); const provider = record(JSON.parse(providerBytes)); - const commands = control.commands; const events = control.committedEvents; - const terminalEnvelope = record( - record(Array.isArray(events) ? events.at(-1) : null).envelope, - ); if ( runner.schema !== RUNNERD_STATE_SCHEMA || !["ready", "suspended"].includes(String(runner.lifecycle)) || @@ -5537,12 +5530,8 @@ async function recoverQuiescentRunnerdState(input: { !Array.isArray(runner.outbox) || runner.outbox.length !== 0 || runner.pendingTerminalDelivery !== null || - !Array.isArray(commands) || - commands.some((command) => record(command).status !== "completed") || !Array.isArray(events) || - record(events.at(-1)).eventType !== "run.terminal" || - terminalEnvelope.runId !== identity.runId || - terminalEnvelope.normalizedSessionId !== identity.normalizedSessionId || + !isSettledRemoteWarmControlState(control, identity) || provider.completedTurnAuthoritative !== true ) continue; diff --git a/server/src/services/native-runtime/remote-warm-session-recovery.test.ts b/server/src/services/native-runtime/remote-warm-session-recovery.test.ts index 91317d2af4b..91231e07f14 100644 --- a/server/src/services/native-runtime/remote-warm-session-recovery.test.ts +++ b/server/src/services/native-runtime/remote-warm-session-recovery.test.ts @@ -7,6 +7,10 @@ const goal = { schema: "paperclip.session_goal.snapshot.v1", goal: null, working const notification = { schema: "paperclip.prp.event.v1", eventType: "session.goal.snapshot", runId: identity.runId, normalizedSessionId: identity.normalizedSessionId, sourceInstanceId: "runner", sourceKind: "runner", payload: goal }; const snapshot = { eventType: "session.goal.snapshot", envelope: { ...identity, payload: notification } }; +const notice = { schema: "paperclip.provider.notice.v1", noticeId: "codex-warning", severity: "warning", category: "warning", scope: "turn", + recoverable: true, userActionable: true, summary: "ignored a notification for a settled Codex turn" }; +const noticeNotification = { ...notification, eventType: "provider.notice.recorded", payload: notice }; +const lateNotice = { eventType: "provider.notice.recorded", envelope: { ...identity, payload: noticeNotification } }; const state = (events: unknown[]) => ({ identity, commands: [{ status: "completed" }], committedEvents: events }); describe("settled remote warm control evidence", () => { @@ -34,6 +38,28 @@ describe("settled remote warm control evidence", () => { expect(isSettledRemoteWarmControlState(state([terminal, suffix]), identity)).toBe(false); }); + it("accepts recoverable provider notices recorded after the terminal", () => { + const info = { ...lateNotice, envelope: { ...lateNotice.envelope, payload: { ...noticeNotification, payload: { ...notice, severity: "info" } } } }; + expect(isSettledRemoteWarmControlState(state([terminal, lateNotice]), identity)).toBe(true); + expect(isSettledRemoteWarmControlState(state([terminal, lateNotice, snapshot, info]), identity)).toBe(true); + }); + + it.each([ + ["error notice", { ...lateNotice, envelope: { ...lateNotice.envelope, payload: { ...noticeNotification, payload: { ...notice, severity: "error" } } } }], + ["unrecoverable notice", { ...lateNotice, envelope: { ...lateNotice.envelope, payload: { ...noticeNotification, payload: { ...notice, recoverable: false } } } }], + ["unknown notice schema", { ...lateNotice, envelope: { ...lateNotice.envelope, payload: { ...noticeNotification, payload: { ...notice, schema: "paperclip.provider.notice.v2" } } } }], + ["notification type mismatch", { ...lateNotice, envelope: { ...lateNotice.envelope, payload: { ...noticeNotification, eventType: "session.goal.snapshot" } } }], + ["foreign notice source", { ...lateNotice, envelope: { ...lateNotice.envelope, payload: { ...noticeNotification, sourceInstanceId: "other" } } }], + ["foreign notice lease", { ...lateNotice, envelope: { ...lateNotice.envelope, environmentLeaseId: "other" } }], + ["control plane notice source", { ...lateNotice, envelope: { ...lateNotice.envelope, payload: { ...noticeNotification, sourceKind: "control_plane" } } }], + ])("holds a %s suffix", (_name, suffix) => { + expect(isSettledRemoteWarmControlState(state([terminal, suffix]), identity)).toBe(false); + }); + + it("holds new work recorded after a late provider notice", () => { + expect(isSettledRemoteWarmControlState(state([terminal, lateNotice, { ...snapshot, eventType: "turn.started" }]), identity)).toBe(false); + }); + it("holds missing terminals, pending commands and mixed core identities", () => { expect(isSettledRemoteWarmControlState(state([snapshot]), identity)).toBe(false); expect(isSettledRemoteWarmControlState({ ...state([terminal, snapshot]), commands: [{ status: "dispatched" }] }, identity)).toBe(false); diff --git a/server/src/services/native-runtime/remote-warm-session-recovery.ts b/server/src/services/native-runtime/remote-warm-session-recovery.ts index ec63a72325e..09222910c96 100644 --- a/server/src/services/native-runtime/remote-warm-session-recovery.ts +++ b/server/src/services/native-runtime/remote-warm-session-recovery.ts @@ -14,9 +14,10 @@ export type RemoteWarmSessionRecovery = { revalidate: (leaseOwner: string) => Promise; }; -/** A goal read can publish an idle snapshot after the terminal. It carries no - * new work; every other suffix still requires reconciliation before handoff. */ -export function isSettledRemoteWarmControlState(control: Record, identity: Record): boolean { +/** A goal read can publish an idle snapshot and a provider can publish a + * recoverable notice after the terminal. Neither carries new work; every other + * suffix still requires reconciliation before handoff. */ +export function isSettledRemoteWarmControlState(control: Record, identity: Record): boolean { const record = (value: unknown): Record => value !== null && typeof value === "object" && !Array.isArray(value) ? value as Record : {}; const commands = control.commands; @@ -34,13 +35,17 @@ export function isSettledRemoteWarmControlState(control: Record const envelope = record(record(event).envelope); const notification = record(envelope.payload); const payload = record(notification.payload); - return record(event).eventType === "session.goal.snapshot" && matchesSession(event) && + const eventType = record(event).eventType; + const inert = eventType === "session.goal.snapshot" + ? payload.schema === "paperclip.session_goal.snapshot.v1" && payload.goal === null && payload.workingNow === false + : eventType === "provider.notice.recorded" && payload.schema === "paperclip.provider.notice.v1" && + payload.recoverable === true && (payload.severity === "info" || payload.severity === "warning"); + return inert && matchesSession(event) && Object.entries(identity).every(([key, value]) => envelope[key] === value) && (notification.schema === "paperclip.prp.event.v1" || notification.schema === "paperclip.prp.event.v2") && - notification.eventType === "session.goal.snapshot" && + notification.eventType === eventType && notification.sourceInstanceId === identity.runnerInstanceId && notification.sourceKind === "runner" && - notification.runId === identity.runId && notification.normalizedSessionId === identity.normalizedSessionId && - payload.schema === "paperclip.session_goal.snapshot.v1" && payload.goal === null && payload.workingNow === false; + notification.runId === identity.runId && notification.normalizedSessionId === identity.normalizedSessionId; }); }