mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-10 12:37:55 +01:00
feat(recovery): settle a claimed wake whose run finished without settling it (#19934)
## 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 <noreply@anthropic.com>
This commit is contained in:
1 parent
83dec71a8f
commit
c6e4ed1547
4 files changed
+141
-4
No files matched your search
@@ -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<string, unknown> = {}) => ({ id, companyId, agentId, status,
|
||||
invocationSource: "manual", ...(status === "running" ? {} : { finishedAt: minutesAgo(2) }), ...extra });
|
||||
const native = (issueId: string, nativePhase: string | null, extra: Record<string, unknown> = {}) =>
|
||||
({ 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 });
|
||||
|
||||
@@ -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");
|
||||
}
|
||||
|
||||
@@ -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<T>(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)
|
||||
|
||||
@@ -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<SettleOutcome> {
|
||||
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" };
|
||||
});
|
||||
}
|
||||
Reference in new issue
Block a user