From 8fd917aedfe53c5a7e062e7ec440323aa5870102 Mon Sep 17 00:00:00 2001 From: Nicky Leach Date: Tue, 3 Nov 2026 16:37:39 -0800 Subject: [PATCH] fix(native): accept late provider notices in recovery (#18890) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Native Runner sessions keep a provider conversation across task turns. > - When a later run gets a new task lease, recovery must prove that the prior run settled. > - Recovery required `run.terminal` to be the last committed event. > - A provider can send a late notification for a settled turn. The Runner then records a `provider.notice.recorded` warning after `run.terminal`. > - This false denial stranded a clean session. The notice stays in the journal, so each later run on that session meets the same denial. > - This pull request lets recovery accept a recoverable notice after the terminal event. All other new events still block recovery. > - The benefit is that a late warning no longer ends session continuity. A denial now also names the check that failed. ## Linked Issues or Issue Description No issue exists. I searched open and closed pull requests and issues for the error code, the event name and the function name. I found no duplicate. **What happened?** A run failed in about nine seconds with `runner_state_identity_mismatch: plugin_checkpoint_recovery_unproven`. No provider work started. The prior run had finished and was committed as succeeded. Its journal had one `provider.notice.recorded` warning, "ignored a notification for a settled Codex turn", 1 ms after `run.terminal`. A second session had the same warning about 61 seconds after a normally completed turn. **Expected behavior** Recovery accepts the session and continues the provider conversation. A recoverable notice does not start work, so it must not block recovery. **Steps to reproduce** 1. Start a warm native session in a plugin sandbox environment. 2. Finish a run so that a provider notice follows `run.terminal` in the journal. 3. Wait until the idle timeout stops the Runner task and releases the lease. 4. Start the next run on that session. It gets a new task lease. 5. The run fails with the error above. The notice stays in the journal, so a retry on the same session meets the same denial. The `local-receipt-late-notice` test mode in `native-session-executor.test.ts` reproduces this failure on the unfixed code. **Paperclip version or commit** Observed on `d8a9b3eb0c`. The same check is on `master` before this change. **Deployment mode** Cloud deployment, with a native Runner in a plugin sandbox environment. ## What Changed - `isSettledRemoteWarmControlState` accepts a `provider.notice.recorded` event after the terminal event. The notice must be recoverable and have `info` or `warning` severity. It must come from the same runner, run, session and lease. - Error notices, unrecoverable notices, notices from another source and new work still block recovery. - Checkpoint recovery and local quiescent recovery now use this function. They no longer have separate checks that require `run.terminal` to be the last event. - Each checkpoint recovery denial adds a specific reason after the existing code, for example `plugin_checkpoint_recovery_unproven: journal_not_settled`. The `runner_state_identity_mismatch: ` prefix does not change. - I reworded the existing comment on `isSettledRemoteWarmControlState` so that it stays correct. I added no new code comments. ## Verification - **Failure before the fix:** I added the tests first and ran them on the unfixed code. 38 tests failed. In the 4 `local-receipt-late-notice` cases, checkpoint recovery rejected with `runner_state_identity_mismatch`. In the `late provider notice` case, quiescent recovery rejected with `quarantine_owner_unverified`. The new unit test for notices also failed. The other 32 failures were assertions for the new denial reasons. - **Targeted tests after the fix:** 134 tests passed, with 987 skipped by the filter. - `cd server && pnpm exec vitest run src/services/native-runtime/remote-warm-session-recovery.test.ts src/services/native-runtime/native-session-executor.test.ts -t "settled remote warm control evidence|recovers a blocked|automatically recovers only a proven settled local session"` - **Type error found in review:** Vitest does not check types, and my first version passed these tests. Review of the diff showed an incompatible `identity` parameter type. I changed that type to `Record`. - **After the type change:** `remote-warm-session-recovery.test.ts` passed (23 tests) and `pnpm run typecheck` in `server` passed. I did not run the 134 tests again, because this change affected only types. - **Real data, read-only:** I ran the new function on 285 session journals from a test deployment. The result changed for 2 journals. These are the 2 stranded sessions, and each has only provider notices after `run.terminal`. - **Not run locally:** I did not run the full test suite. I stopped a run of the wider native-runtime test files before it reported a result. CI provides the full regression coverage. - **Not verified:** no live run has reached `read_state` on a stranded session after this fix. The tests use a fixture for that call. ## Risks - **Stricter identity check:** checkpoint and quiescent recovery now also require the journal's saved identity to match. Recovery reads this identity from the same file, so only damaged state should fail this check. - **Wider goal snapshot rule:** these two paths now accept idle goal snapshots after the terminal event. Warm handoff already used this rule. - **Wider notice rule:** warm handoff and the settled projection check also accept recoverable notices now. A notice does not start work, and the checks for pending commands and unanswered tool input stay in place. - **Longer error text:** code that compares the full `plugin_checkpoint_recovery_unproven` message for equality would stop matching. I found no other reference to this code in the repository. - **Stranded sessions:** these sessions need no data repair. They can still fail with `checkpoint_not_settled` if the sandbox checkpoint is not clean. - **Migrations:** this change has no migration. > 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 - Anthropic Claude Opus 5.5 (`claude-opus-5-5`), through Claude Code, with tool use and code execution. A human directed the diagnosis and approved the change. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-authored-by: Claude --- .../native-session-executor.test.ts | 56 +++++++++++++++++-- .../native-runtime/native-session-executor.ts | 37 +++++------- .../remote-warm-session-recovery.test.ts | 26 +++++++++ .../remote-warm-session-recovery.ts | 19 ++++--- 4 files changed, 102 insertions(+), 36 deletions(-) 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; }); }