mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-04 19:28:46 +01:00
fix(native): accept late provider notices in recovery (#18890)
## 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<string, unknown>`. - **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 <noreply@anthropic.com>
This commit is contained in:
Nicky LeachandClaude authored and GitHub committed
1 parent
82dd3bb702
commit
8fd917aedf
4 files changed
+102
-36
No files matched your search
@@ -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",
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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);
|
||||
|
||||
@@ -14,9 +14,10 @@ export type RemoteWarmSessionRecovery = {
|
||||
revalidate: (leaseOwner: string) => Promise<void>;
|
||||
};
|
||||
|
||||
/** 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<string, unknown>, identity: Record<string, string>): 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<string, unknown>, identity: Record<string, unknown>): boolean {
|
||||
const record = (value: unknown): Record<string, unknown> =>
|
||||
value !== null && typeof value === "object" && !Array.isArray(value) ? value as Record<string, unknown> : {};
|
||||
const commands = control.commands;
|
||||
@@ -34,13 +35,17 @@ export function isSettledRemoteWarmControlState(control: Record<string, unknown>
|
||||
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;
|
||||
});
|
||||
}
|
||||
|
||||
|
||||
Reference in new issue
Block a user