From c6e4ed1547457ab5095424efbb2bb75a75249c37 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Sat, 7 Nov 2026 21:18:22 -0600 Subject: [PATCH] feat(recovery): settle a claimed wake whose run finished without settling it (#19934) 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. > - A wake request is claimed by the run that handles it, and that run marks it finished. > - In a K4 chaos wave (#19927), native restart recovery blocked a run and ended it, but left its assignment wake `claimed`. Nothing ever released it. > - The durable-ownership plan settles such records on one shared path, by proof, never by age. > - This pull request registers a settle for `wake_request_claimed`. It is Phase 2's second kind and is stacked on the accepted-result lease PR. > - The benefit is that no run-ending path can strand a claimed wake. ## Linked Issues or Issue Description Refs #19831 (plan), #19861 (registry), #19927 (the data that found this). Stacked on #19933 (it adds the `settle` hook). **What happened?** Run `7c45860b…` failed with `native_restart_recovery_blocked` (`execution_recovery_budget_exhausted`) at 23:19:41. Its wake `6b3c4cce…` stayed `claimed` until the end of the wave (23:44 and later). Rule I2 flagged it. **What did you expect?** A claimed wake ends when its run ends, with the run's outcome. **Steps to reproduce** 1. `node cli/node_modules/tsx/dist/cli.mjs tests/restart-chaos/main.ts --env local --faults K4 --k4-hold-max-seconds 600 --tasks 6 --concurrency 3 --seed 9203 --keep` 2. Look for `agent_wakeup_requests` rows with status `claimed` whose run is terminal. ## What Changed - New `recovery/wake-request-claimed-settlement.ts`: when the owner run is terminal, and for a native run its coordinator is `committed` or `terminal_failure`, the claim gets the run's outcome. That is `completed` for a succeeded run and otherwise the run's status, as run completion and native terminal projection already do. It is compare-and-swap on the claim (id, run, status `claimed`). - `ownership-registry.ts`: `wake_request_claimed` registers that settle. - `native-finalization-reconciler.ts`: the pass log line includes the settled count per kind. ## Verification - `cd server && npx vitest run src/__tests__/ownership-reconciliation-report.test.ts`: 17 pass, including the new case. It covers a blocked native run, a native run whose projection is still pending (kept), a succeeded legacy run, a live run (kept), and a read-only pass (no change). - Red/green (test written after the code): the new test fails on master and passes here. - Server typecheck passes. ## Risks - No persisted shape changes, so there is no previous-release story beyond this: claims saved by an older release are settled the same way. - Race with the run's own settle: it waits for a native coordinator to be final. For a legacy run, both write the same status, so the race is idempotent. - No change adds work before `turn.start`. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, in Claude Code with tool use (shell, file edits). ## 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 - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [ ] I will address all Greptile and reviewer comments before requesting merge 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 --- .../ownership-reconciliation-report.test.ts | 59 ++++++++++++++- .../native-finalization-reconciler.ts | 3 +- .../services/recovery/ownership-registry.ts | 8 +- .../wake-request-claimed-settlement.ts | 75 +++++++++++++++++++ 4 files changed, 141 insertions(+), 4 deletions(-) create mode 100644 server/src/services/recovery/wake-request-claimed-settlement.ts diff --git a/server/src/__tests__/ownership-reconciliation-report.test.ts b/server/src/__tests__/ownership-reconciliation-report.test.ts index 3417f93a01a..3cb6f8f483a 100644 --- a/server/src/__tests__/ownership-reconciliation-report.test.ts +++ b/server/src/__tests__/ownership-reconciliation-report.test.ts @@ -134,7 +134,7 @@ describe("report-only ownership reconciliation", () => { const names = OWNABLE_KINDS.map(kind => kind.id); expect(new Set(names).size).toBe(names.length); expect(names.length).toBeGreaterThanOrEqual(40); - expect(OWNABLE_KINDS.filter(kind => kind.settle !== null).map(kind => kind.id)).toEqual(["native_accepted_result_lease"]); + expect(OWNABLE_KINDS.filter(kind => kind.settle !== null).map(kind => kind.id)).toEqual(["native_accepted_result_lease", "wake_request_claimed"]); expect(OWNABLE_KINDS.filter(kind => !kind.listInProgress).every(kind => kind.coverageGap)).toBe(true); }); @@ -221,6 +221,63 @@ describe("report-only ownership reconciliation", () => { } finally { vi.stubEnv("PAPERCLIP_EPHEMERAL_SERVER", ""); } }); + it("settles a claimed wake whose run finished without settling it, and only that one", async () => { + // K4 wave (2026-11-07): native restart recovery blocked a run and ended it + // without settling its assignment wake, which stayed `claimed` for good. + const issueIds = { blocked: randomUUID(), pending: randomUUID(), unprojected: randomUUID(), applied: randomUUID() }; + await db.insert(issues).values(Object.entries(issueIds).map(([title, id]) => ({ id, companyId, title }))); + const run = (id: string, status: string, extra: Record = {}) => ({ id, companyId, agentId, status, + invocationSource: "manual", ...(status === "running" ? {} : { finishedAt: minutesAgo(2) }), ...extra }); + const native = (issueId: string, nativePhase: string | null, extra: Record = {}) => + ({ runtimeMode: "native", nativeIssueId: issueId, issueId, nativePhase, ...extra }); + const runs = { blocked: randomUUID(), pending: randomUUID(), unprojected: randomUUID(), applied: randomUUID(), + legacyDone: randomUUID(), interrupted: randomUUID(), live: randomUUID() }; + await db.insert(heartbeatRuns).values([ + run(runs.blocked, "failed", native(issueIds.blocked, "terminal_failure", { error: "execution_recovery_budget_exhausted", + errorCode: "native_restart_recovery_blocked" })), + // The coordinator is not final yet. + run(runs.pending, "failed", native(issueIds.pending, "observed")), + // Final coordinator, run row not yet projected (its phase lags): + // projection may still change the run's outcome. + run(runs.unprojected, "failed", native(issueIds.unprojected, "workspace_finalizing")), + run(runs.applied, "succeeded", native(issueIds.applied, null)), + run(runs.legacyDone, "succeeded"), + run(runs.interrupted, "interrupted"), + run(runs.live, "running"), + ]); + await db.insert(nativeRunFinalizations).values([ + { runId: runs.blocked, companyId, issueId: issueIds.blocked, phase: "terminal_failure", failureCode: "native_restart_recovery_blocked" }, + { runId: runs.pending, companyId, issueId: issueIds.pending, phase: "observed" }, + { runId: runs.applied, companyId, issueId: issueIds.applied, phase: "applied" }, + ]); + await db.insert(nativeRunFinalizations).values({ runId: runs.unprojected, companyId, issueId: issueIds.unprojected, + phase: "terminal_failure", failureCode: "fixture" }); + const wakes = Object.fromEntries(Object.entries(runs).map(([name, runId]) => [name, { id: randomUUID(), runId }])); + await db.insert(agentWakeupRequests).values(Object.values(wakes).map(wake => ({ id: wake.id, companyId, agentId, + source: "assignment", status: "claimed", runId: wake.runId, claimedAt: minutesAgo(5) }))); + const statusOf = async () => Object.fromEntries((await db.select({ id: agentWakeupRequests.id, status: agentWakeupRequests.status }) + .from(agentWakeupRequests)).map(row => [Object.keys(wakes).find(name => wakes[name].id === row.id), row.status])); + const allClaimed = Object.fromEntries(Object.keys(wakes).map(name => [name, "claimed"])); + + // A read-only pass never settles. + await runOwnershipReconciliationPass(db, { state: createOwnershipReportState(), record: false }); + expect(await statusOf()).toEqual(allClaimed); + + const state = createOwnershipReportState(); + const first = await runOwnershipReconciliationPass(db, { state }); + expect(kindReport(first, "wake_request_claimed")).toMatchObject({ inProgress: 7, settled: 4 }); + expect(await statusOf()).toEqual({ blocked: "failed", pending: "claimed", unprojected: "claimed", applied: "completed", + legacyDone: "completed", interrupted: "cancelled", live: "claimed" }); + const [blocked] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, wakes.blocked.id)); + expect(blocked).toMatchObject({ error: "execution_recovery_budget_exhausted" }); + expect(blocked.finishedAt).not.toBeNull(); + // Settled claims are no longer in progress. The two waiting on projection + // are not settled, and rule I2 names them if projection never comes. + const second = await runOwnershipReconciliationPass(db, { state }); + expect(second.violations.filter(violation => violation.kind === "wake_request_claimed").map(violation => violation.recordId).sort()) + .toEqual([wakes.pending.id, wakes.unprojected.id].sort()); + }); + it("writes nothing when recording is off", async () => { await seed(); const report = await runOwnershipReconciliationPass(db, { state: createOwnershipReportState(), record: false }); diff --git a/server/src/services/native-runtime/native-finalization-reconciler.ts b/server/src/services/native-runtime/native-finalization-reconciler.ts index c71dd223ab3..b5408f2d476 100644 --- a/server/src/services/native-runtime/native-finalization-reconciler.ts +++ b/server/src/services/native-runtime/native-finalization-reconciler.ts @@ -897,7 +897,8 @@ export function registerOwnershipReconciliationWork( inProgress: report.inProgress, violations: report.violations.length, kinds: Object.fromEntries(report.kinds.filter(kind => kind.inProgress > 0 || kind.error) - .map(kind => [kind.kind, { ...kind.verdicts, ...(kind.truncated ? { truncated: true } : {}), + .map(kind => [kind.kind, { ...kind.verdicts, ...(kind.settled ? { settled: kind.settled } : {}), + ...(kind.truncated ? { truncated: true } : {}), ...(kind.error ? { error: kind.error } : {}) }])), }, "ownership reconciliation pass"); } diff --git a/server/src/services/recovery/ownership-registry.ts b/server/src/services/recovery/ownership-registry.ts index 601484bda57..988e568c767 100644 --- a/server/src/services/recovery/ownership-registry.ts +++ b/server/src/services/recovery/ownership-registry.ts @@ -31,6 +31,7 @@ import { requestNativeAcceptedResultLeaseOperatorAction, settleNativeAcceptedResultLease, } from "../native-runtime/native-accepted-result-lease-settlement.js"; +import { settleClaimedWakeRequest } from "./wake-request-claimed-settlement.js"; /** * Registry of ownable record kinds (plan section C). @@ -106,7 +107,8 @@ export type OwnableKind = { coverageGap?: string; }; -const TERMINAL_PHASES = ["committed", "terminal_failure", "applied"]; +/** Final `native_run_finalizations` phases (shared with settle paths). */ +export const TERMINAL_PHASES = ["committed", "terminal_failure", "applied"]; function page(rows: T[], limit: number): { rows: T[]; truncated: boolean } { return rows.length > limit ? { rows: rows.slice(0, limit), truncated: true } : { rows, truncated: false }; @@ -649,7 +651,9 @@ export const OWNABLE_KINDS: readonly OwnableKind[] = [ // Wake queue and recovery { id: "wake_request_claimed", title: "Wake request (claimed)", obligation: "ownership", storage: "database", - settle: null, invariantChecks: [2, 4], + // Shared path since Phase 2: a claim whose run finished is settled with + // that run's outcome, so a path that ends a run without it cannot strand it. + settle: settleClaimedWakeRequest, invariantChecks: [2, 4], listInProgress: async (db, limit, offset) => { const rows = await db.select({ id: agentWakeupRequests.id, companyId: agentWakeupRequests.companyId, runId: agentWakeupRequests.runId, updatedAt: agentWakeupRequests.updatedAt }).from(agentWakeupRequests) diff --git a/server/src/services/recovery/wake-request-claimed-settlement.ts b/server/src/services/recovery/wake-request-claimed-settlement.ts new file mode 100644 index 00000000000..68ac1146803 --- /dev/null +++ b/server/src/services/recovery/wake-request-claimed-settlement.ts @@ -0,0 +1,75 @@ +import { and, eq, inArray, isNotNull } from "drizzle-orm"; +import { agentWakeupRequests, heartbeatRuns, nativeRunFinalizations, type Db } from "@paperclipai/db"; +import { notifyExternalConversationSettlement } from "../external-conversation-settlement-notifications.js"; +import { notifyStrandedAgentStatus } from "../live-run-recovery-notifications.js"; +import type { OwnerProof } from "../owner-proof.js"; +import { TERMINAL_PHASES, type InProgressRecord, type SettleOutcome } from "./ownership-registry.js"; + +const TERMINAL_RUN_STATUSES = ["succeeded", "failed", "cancelled", "timed_out", "interrupted"] as const; + +/** The wake status other settlement paths write for a finished run: run + * completion writes `completed` or the run status, and a graceful shutdown + * writes `cancelled` for an interrupted run. */ +function wakeStatusFor(runStatus: string): string { + if (runStatus === "succeeded") return "completed"; + if (runStatus === "interrupted") return "cancelled"; + return runStatus; +} + +/** + * Settle of the `wake_request_claimed` ownership kind (plan Phase 2). + * + * A claimed wake is owned by the run that claimed it, and that run settles it + * when it finishes. A path that ends the run without that step leaves the wake + * `claimed` for good (seen in a K4 wave: native restart recovery blocked the + * run, and its assignment wake stayed claimed). The owner's terminal row is + * the proof; age never is. + * + * Settle copies the run's outcome onto exactly that claim, the way the other + * settlement paths do (`wakeStatusFor`). For a native run it waits until the + * run row carries the coordinator's final phase, and reads it under the run + * row lock, so it never records an outcome that projection later changes. The intent was + * delivered (the run claimed it), so this is the delivered terminal reason. + */ +export async function settleClaimedWakeRequest(db: Db, record: InProgressRecord, + context: { now: Date; proof: OwnerProof }): Promise { + if (record.owner.kind !== "run") return { settled: false, reason: "owner_not_a_run" }; + const runId = record.owner.runId; + return db.transaction(async transaction => { + const tx = transaction as unknown as Db; + // Lock the run row, as native terminal projection does, so its final + // outcome cannot change between this read and the wake write. + const [run] = await tx.select({ id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, + agentId: heartbeatRuns.agentId, status: heartbeatRuns.status, runtimeMode: heartbeatRuns.runtimeMode, + nativePhase: heartbeatRuns.nativePhase, finishedAt: heartbeatRuns.finishedAt, error: heartbeatRuns.error }) + .from(heartbeatRuns) + .where(and(eq(heartbeatRuns.id, runId), inArray(heartbeatRuns.status, [...TERMINAL_RUN_STATUSES]), + isNotNull(heartbeatRuns.finishedAt))) + .for("update"); + if (!run) return { settled: false, reason: "owner_run_not_finished" }; + if (run.runtimeMode === "native") { + const [coordinator] = await tx.select({ phase: nativeRunFinalizations.phase }).from(nativeRunFinalizations) + .where(eq(nativeRunFinalizations.runId, run.id)); + // The coordinator commits before the run row is projected; settle only + // once the projection carried that final phase onto the run. A legacy + // `applied` row has no later projection. + if (coordinator && (!TERMINAL_PHASES.includes(coordinator.phase) + || (coordinator.phase !== "applied" && run.nativePhase !== coordinator.phase))) { + return { settled: false, reason: "native_projection_pending" }; + } + } + const claim = and(eq(agentWakeupRequests.id, record.recordId), eq(agentWakeupRequests.companyId, run.companyId), + eq(agentWakeupRequests.runId, run.id), eq(agentWakeupRequests.status, "claimed")); + const [current] = await tx.select({ agentId: agentWakeupRequests.agentId }).from(agentWakeupRequests) + .where(claim).for("update"); + if (!current) return { settled: false, reason: "claim_no_longer_held" }; + // Work intents before the write they announce (sleep-aware background work). + await notifyExternalConversationSettlement(tx, { companyId: run.companyId }); + await notifyStrandedAgentStatus(tx, { companyId: run.companyId, agentId: current.agentId }); + const updated = await tx.update(agentWakeupRequests).set({ + status: wakeStatusFor(run.status), + finishedAt: run.finishedAt, error: run.error, updatedAt: context.now, + }).where(claim).returning({ id: agentWakeupRequests.id }); + return updated.length ? { settled: true, reason: context.proof.reason } : { settled: false, reason: "claim_no_longer_held" }; + }); +}