fix(native): do not charge a restart resume that made progress before the next crash (#19720)

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work
> - Native runs have an execution recovery budget of three provider
attempts, so a run that keeps failing stops retrying
> - A control-plane crash (no graceful drain) makes the next server
resume the run, and that resume charges one attempt
> - #19625 stopped charging a resume after a graceful drain, but crash
restarts still charge, even when the previous resume did real work
> - On a deploy day with several restarts, a healthy run used up its
budget and the task needed a manual Retry
> - This pull request does not charge a crash resume when the previous
restart resume made recorded progress
> - The benefit is that restarts alone do not exhaust a run that keeps
making progress, and genuine loops stay bounded

## Linked Issues or Issue Description

Refs #19625

**What happened?**
In restart-chaos testing (storm wave, OpenCode run), a run crossed four
control-plane restarts: K1, K1, K2 (drain), K1. Each resume was clean:
the runner restored the session, recorded
`provider_turn_lost_on_restore`, and the #19699 continuation turn then
ran tools for minutes. The two K1 resumes each charged one attempt
(attempt 1 to 3). The drain resume did not charge (#19625). The fourth
restart found attempt 3 and blocked the run with
`native_restart_recovery_blocked` /
`execution_recovery_budget_exhausted`. Seven runs in the same wave
failed in the same way. In every case, each resume had accepted a new
turn and completed tool work before the next restart.

**Expected behavior**
A run must not fail because it survived several controller restarts that
each resumed cleanly and made progress. The budget must still stop a
resume that fails each time, a crash loop that makes no progress, and a
provider that fails repeatedly.

**Steps to reproduce**
Start a native local run. Kill the server (SIGKILL) while the run is
active and let it resume. Let the resumed turn run tools. Repeat three
times. The third crash resume is blocked with
`execution_recovery_budget_exhausted`.

**Paperclip version or commit**
`master` at `8a34c1b62c`.

## What Changed

- `claimNativeRestartRecoveries()` (`native-restart-recovery.ts`)
records `progressWatermarkSeq` (the run's highest
`heartbeat_run_events.seq`) in the history entry of each recovery claim.
- New `previousResumeMadeProgress()`: the next claim reads the watermark
of the most recent claim. It finds progress when, after that watermark,
a `turn.accepted` event names a turn, and a later
`tool.execution.completed` or `item.completed` event (not a `usage`
item) belongs to the same turn. The evidence is committed run events,
not timing.
- When that progress exists, the claim is not charged
(`chargesProviderAttempt: false`, and the `>= 3` exhausted gate is
skipped). This uses the same path as the clean-drain rule from #19625.
The history entry records `attemptKeptForProgress: true`.
- Restore activity does not count: lineage items without a turn,
`turn.failed`, work on the interrupted turn, `turn.submitted`, and an
accepted turn with only usage or a failure. So a resume that dies or
fails before it does work still charges.
- The credit stops at controller generation 12
(`NATIVE_PROGRESS_CREDIT_GENERATION_LIMIT`). A run that does some work
and then crashes the controller every time still reaches the budget.
- The first restart (which interrupts the original execution, not a
resume) still charges. A history entry from an older release has no
watermark, so it gets no credit.
- `doc/execution-semantics.md` states the rule next to the three-attempt
budget.

## Verification

New tests in
`server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts`
(embedded Postgres, real `claimNativeRestartRecoveries`; between claims
the test applies the executor's attempt charge and marks the controller
dead):

- `keeps recovering a run whose restart resumes each completed work
before the next crash`: five crash resumes, each followed by restore
noise and one completed tool in the continuation turn. Charges are
`[true, false, false, false, false]` and the attempt stays 2. Red on
master (attempt 2, 3, then blocked).
- `still exhausts the budget when restart resumes complete no work
between crashes`: resumes followed only by restore noise and a failed
continuation turn. Attempt 2, attempt 3, then
`execution_recovery_budget_exhausted`. Green on master and with the fix.
- `bounds a crash loop that makes progress before every crash`: credit
stops at the generation limit, the last attempt is spent, and the run
blocks. Red on master.
- `native-runner-restart-recovery.integration.test.ts`,
`heartbeat-native-restart-suspension.test.ts`,
`native-restart-recovery-isolation.test.ts` and
`native-remote-retry-admission.test.ts`: 218 passed, 14 skipped.
- I ran the progress query against the live database of the failing
wave. For all seven blocked runs, every resume had progress before the
next restart, so this rule would have kept each of them recovering.
- `tsc --noEmit` for `server` has no errors in the changed files.

## Risks

- A run that makes progress and then makes the controller crash each
time gets up to about ten extra resumes before the budget applies. The
generation limit bounds this.
- The rule depends on the PRP event shape (`payload.prpEvent.turnId` on
`turn.accepted` and on completed work). Events without that shape get no
credit, so the failure mode is the old behavior (charge), not an
unbounded loop.
- The query runs once per claim, on the `(run_id, seq)` unique index
range after the watermark.

> 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

- Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, via Claude Code with
tool use (shell, file edits, test execution). Extended reasoning
enabled.

## 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
- [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 Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
DottaandClaude Opus 5.5 authored and GitHub committed 2026-11-06 22:04:19 -06:00
1 parent d1ee25d0f7
commit 5cc89b1e4d
3 files changed
+223 -5

No files matched your search

+1 -1
View File
@@ -2285,7 +2285,7 @@ unknown operations or partial transcripts do not. Replacement uses a fresh
session and retires only the exact predecessor's obsolete recovery hold while
recording the proof and successor lineage. Retained provider files are not edited.
Bootstrap retries, exact-checkpoint resumes, and fresh replacement sessions share three total provider attempts, including the original attempt. Linked run IDs, controller restarts, and duplicate wakes do not reset this budget. A resume after a graceful shutdown drain suspended the run for reattachment does not consume an attempt, because the run did not cause that restart. A resume after a crash, which leaves no drain record, still consumes one. Exact-checkpoint resumes retain the 30-second delay. Replacement scheduling and predecessor lineage commit together, with one successor per predecessor and admission through the normal task locks, authorization, pause, approval, and budget gates.
Bootstrap retries, exact-checkpoint resumes, and fresh replacement sessions share three total provider attempts, including the original attempt. Linked run IDs, controller restarts, and duplicate wakes do not reset this budget. A resume after a graceful shutdown drain suspended the run for reattachment does not consume an attempt, because the run did not cause that restart. A resume after a crash, which leaves no drain record, still consumes one, unless the previous restart resume made progress before the crash. Each restart claim records the run's event-log position. Progress means that, after that position, the provider accepted a new turn and then completed work in that turn: a tool execution or a completed item other than a usage report. Restore activity, failed turns, and accepted turns without completed work are not progress, so a crash loop without progress and a resume that fails each time still use up the budget. This credit stops after 12 controller generations, so a run that makes progress and then crashes the controller each time is still bounded. Exact-checkpoint resumes retain the 30-second delay. Replacement scheduling and predecessor lineage commit together, with one successor per predecessor and admission through the normal task locks, authorization, pause, approval, and budget gates.
A due failed remote retry can retain its provider PID after heartbeat completion
releases the task execution lock. Both remote workspace reference formats and
@@ -159,6 +159,82 @@ export function suspendedByCleanDrain(coordinator: {
return false;
}
/**
* A controller restart (a crash, so no drain record) that interrupts a resume
* which had already made progress is also not the run's failure. Each claim
* records the run's event-log position; the next claim does not charge an
* attempt when, after that position, the provider accepted a new turn and then
* completed work in it (a tool execution or a non-usage item). A resume that
* dies before it completes any work in a new turn still charges, so a crash
* loop without progress, a resume that fails each time, and provider failures
* stay bounded. Credit stops after NATIVE_PROGRESS_CREDIT_GENERATION_LIMIT
* controller generations, so a run that makes progress and then crashes the
* controller every time is still bounded.
*/
export const NATIVE_PROGRESS_CREDIT_GENERATION_LIMIT = 12;
const RESUME_PROGRESS_EVENT_TYPES = ["tool.execution.completed", "item.completed"] as const;
type RecoveryClaimProgressEntry = { disposition?: unknown; progressWatermarkSeq?: unknown; progressAttempt?: unknown };
/**
* The watermark of the most recent claim, only while the run is still on the
* provider attempt that claim resumed. A normal retry in between starts a new
* attempt, so the earlier resume's work no longer counts for it.
*/
export function lastRecoveryClaimProgressWatermark(
recoveryHistory: ReadonlyArray<RecoveryClaimProgressEntry>,
currentAttempt: number,
): number | null {
for (let index = recoveryHistory.length - 1; index >= 0; index -= 1) {
const entry = recoveryHistory[index];
if (!RECOVERY_CLAIM_DISPOSITIONS.has(entry?.disposition)) continue;
if (entry?.progressAttempt !== currentAttempt) return null;
const watermark = entry?.progressWatermarkSeq;
return typeof watermark === "number" && Number.isSafeInteger(watermark) && watermark >= 0 ? watermark : null;
}
return null;
}
async function nativeRunEventWatermark(tx: Pick<Db, "select">, runId: string): Promise<number> {
const [row] = await tx
.select({ seq: sql<string | number | null>`max(${heartbeatRunEvents.seq})` })
.from(heartbeatRunEvents)
.where(eq(heartbeatRunEvents.runId, runId));
return Number(row?.seq ?? 0);
}
export async function previousResumeMadeProgress(tx: Pick<Db, "select">, input: {
runId: string;
coordinator: {
attempt: number;
controllerGeneration: number;
recoveryHistory: ReadonlyArray<RecoveryClaimProgressEntry>;
};
}): Promise<boolean> {
if (input.coordinator.attempt <= 0) return false;
if (input.coordinator.controllerGeneration >= NATIVE_PROGRESS_CREDIT_GENERATION_LIMIT) return false;
const watermark = lastRecoveryClaimProgressWatermark(input.coordinator.recoveryHistory, input.coordinator.attempt);
if (watermark === null) return false;
const workTurnId = sql`${heartbeatRunEvents.payload} -> 'prpEvent' ->> 'turnId'`;
const [work] = await tx
.select({ seq: heartbeatRunEvents.seq })
.from(heartbeatRunEvents)
.where(and(
eq(heartbeatRunEvents.runId, input.runId),
gt(heartbeatRunEvents.seq, watermark),
inArray(heartbeatRunEvents.eventType, [...RESUME_PROGRESS_EVENT_TYPES]),
sql`coalesce(${heartbeatRunEvents.payload} -> 'prpEvent' -> 'payload' ->> 'kind', '') <> 'usage'`,
// The inner alias hides the table name, so unqualified outer references
// below resolve to the outer completed-work row.
sql`exists (select 1 from ${heartbeatRunEvents} as accepted_turn where accepted_turn.run_id = ${input.runId}
and accepted_turn.seq > ${watermark} and accepted_turn.seq < ${heartbeatRunEvents.seq}
and accepted_turn.event_type = 'turn.accepted'
and accepted_turn.payload -> 'prpEvent' ->> 'turnId' = ${workTurnId})`,
))
.limit(1);
return work !== undefined;
}
const controllerBootId = randomUUID();
function processIsAlive(pid: number | null | undefined): boolean {
@@ -1184,6 +1260,9 @@ export async function claimNativeRestartRecoveries(input: {
// Read before this claim appends its own history entry.
const cleanDrainSuspension = suspendedByCleanDrain(row.coordinator);
const resumeKeepsAttempt = cleanDrainSuspension || await previousResumeMadeProgress(tx, {
runId: row.run.id, coordinator: row.coordinator,
});
const classification = classifyNativeRunnerRecoveryEvidence({
remoteSandbox: remoteSandbox || Boolean(pluginTask && durableAuthority),
unclaimedBootstrap,
@@ -1207,7 +1286,7 @@ export async function claimNativeRestartRecoveries(input: {
?.runTerminalState === "failed" &&
nativeRestartInterruptedTurnId(checkpointRecord as unknown as PersistedNativeSession) === null,
providerAttempt: row.coordinator.attempt,
resumeChargesAttempt: !cleanDrainSuspension,
resumeChargesAttempt: !resumeKeepsAttempt,
});
if (stopCleanup && classification.claimKind !== "reattach_remote_runner") {
return { kind: "blocked", runId: row.run.id, reason: "native_stop_requires_existing_remote_session" } as const;
@@ -1381,6 +1460,11 @@ export async function claimNativeRestartRecoveries(input: {
recycledProviderPids: providerProcesses.recycledPids,
stderrTail: row.run.stderrExcerpt,
});
// The next claim credits this resume only for work recorded after here.
const progressWatermarkSeq = await nativeRunEventWatermark(tx, row.run.id);
const attemptKeptForProgress = resumeKeepsAttempt && !cleanDrainSuspension;
// The attempt the executor runs this resume as (see its execution claim).
const progressAttempt = nextNativeProviderAttempt(providerAttempt, claimKind, !resumeKeepsAttempt);
await notifyLiveRunRecovery(tx, { companyId: row.run.companyId, runId: row.run.id, agentId: row.run.agentId });
const claimed = await tx
.update(nativeRunFinalizations)
@@ -1403,7 +1487,9 @@ export async function claimNativeRestartRecoveries(input: {
? "resuming_session"
: "bootstrap_incomplete",
recoveryRequestId: input.recoveryRequestId ?? null,
recoveryHistory: appendBoundedRecoveryHistory(event),
recoveryHistory: appendBoundedRecoveryHistory({
...event, progressWatermarkSeq, progressAttempt, ...(attemptKeptForProgress ? { attemptKeptForProgress } : {}),
}),
failureCode: null,
nextAttemptAt: null,
updatedAt: now,
@@ -1471,7 +1557,7 @@ export async function claimNativeRestartRecoveries(input: {
providerAttempt,
restartKind: input.restartKind,
recoveryRequestId: input.recoveryRequestId ?? null,
chargesProviderAttempt: !cleanDrainSuspension,
chargesProviderAttempt: !resumeKeepsAttempt,
};
if (claimKind === "reattach_remote_runner") {
return {
@@ -86,7 +86,7 @@ vi.mock("../../sentry.js", async () => {
captureRunFailure: mockCaptureRunFailure,
};
});
import { claimNativeRestartRecoveries, currentNativeControllerIdentity, NATIVE_RESTART_RECOVERY_FAILED, type NativeControllerIdentity } from "./native-restart-recovery.js";
import { claimNativeRestartRecoveries, currentNativeControllerIdentity, NATIVE_PROGRESS_CREDIT_GENERATION_LIMIT, NATIVE_RESTART_RECOVERY_FAILED, nextNativeProviderAttempt, type NativeControllerIdentity } from "./native-restart-recovery.js";
// Exact v5 wire prompt from before the task-monitor upgrade (#15446).
const historicalPrompt = {
@@ -2606,6 +2606,138 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", (
: { kind: "blocked", runId: fixture.runId, reason: "execution_recovery_budget_exhausted" })]);
});
// RC-22b: controller crashes (no drain record) that each interrupted a resume
// which had already done work exhausted the three-attempt budget.
async function seedInterruptedCheckpoint(label: string) {
const fixture = await seedRun(label);
await fixture.db.update(heartbeatRuns).set({ runnerProfileJson: { sessionCheckpoint: {
driverKind: "opencode_server",
sessionId: fixture.native.normalizedSessionId,
identity: { companyId, issueId: fixture.issueId, runId: fixture.runId, agentId, sessionId: fixture.native.normalizedSessionId },
providerSessionId: "interrupted-provider",
activeTurnId: null,
semanticResult: null,
terminal: { runTerminalState: "failed", turnTerminalState: "failed" },
terminalTurns: [{ turnId: "interrupted-turn", fingerprint: JSON.stringify({
terminalState: "failed", result: null,
error: { code: "provider_turn_lost_on_restore", recoverable: true },
}) }],
} } }).where(eq(heartbeatRuns.id, fixture.runId));
// The original execution claimed attempt 1.
await fixture.db.update(nativeRunFinalizations).set({ attempt: 1, controllerGeneration: 1 })
.where(eq(nativeRunFinalizations.runId, fixture.runId));
return fixture;
}
async function appendRunEvents(fixture: Awaited<ReturnType<typeof seedRun>>, events: Array<{
eventType: string; turnId?: string; kind?: string;
}>) {
const [last] = await fixture.db.select({ seq: sql<number>`coalesce(max(${heartbeatRunEvents.seq}), 0)::int` })
.from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, fixture.runId));
let seq = last?.seq ?? 0;
for (const event of events) {
seq += 1;
await fixture.db.insert(heartbeatRunEvents).values({ companyId, runId: fixture.runId, agentId, seq,
eventType: event.eventType, stream: "system", level: "info",
payload: { prpEvent: { runId: fixture.runId, eventType: event.eventType, sourceKind: "runner",
...(event.turnId ? { turnId: event.turnId } : {}),
payload: event.kind ? { kind: event.kind } : {} } } });
}
}
// The new controller resumes, the executor applies the claim's charge to the
// attempt, and then the controller dies (a crash leaves no drain record).
async function resumeThenCrash(fixture: Awaited<ReturnType<typeof seedRun>>) {
const [disposition] = await claimNativeRestartRecoveries({
db: fixture.db, controller: successor, restartKind: "hard", runIds: [fixture.runId],
});
if (disposition?.kind !== "resume_dead_runner") return { disposition };
const [coordinator] = await fixture.db.select().from(nativeRunFinalizations)
.where(eq(nativeRunFinalizations.runId, fixture.runId));
const attempt = nextNativeProviderAttempt(coordinator.attempt, disposition.kind, disposition.chargesProviderAttempt !== false);
await fixture.db.update(nativeRunFinalizations).set({
attempt, controllerBootId: randomUUID(), controllerPid: 2_000_000_001,
}).where(eq(nativeRunFinalizations.runId, fixture.runId));
return { disposition, attempt };
}
// What a resumed turn records on restore before it does any new work.
const restoreNoise = (round: number) => [
{ eventType: "item.completed", kind: "thread_lineage" },
{ eventType: "turn.failed", turnId: "interrupted-turn" },
{ eventType: "tool.execution.completed", turnId: "interrupted-turn" },
{ eventType: "turn.submitted" },
{ eventType: "turn.accepted", turnId: `continuation-${round}` },
{ eventType: "item.completed", turnId: `continuation-${round}`, kind: "usage" },
];
it("keeps recovering a run whose restart resumes each completed work before the next crash", async () => {
const fixture = await seedInterruptedCheckpoint("PROGRESS-BETWEEN-CRASHES");
const attempts: number[] = [];
const charges: boolean[] = [];
for (let round = 1; round <= 5; round += 1) {
const { disposition, attempt } = await resumeThenCrash(fixture);
expect(disposition).toMatchObject({ kind: "resume_dead_runner", runId: fixture.runId });
charges.push((disposition as { chargesProviderAttempt?: boolean }).chargesProviderAttempt !== false);
attempts.push(attempt!);
await appendRunEvents(fixture, [...restoreNoise(round),
{ eventType: "tool.execution.completed", turnId: `continuation-${round}` }]);
}
// The first restart interrupted the original execution, not a resume.
expect(charges).toEqual([true, false, false, false, false]);
expect(attempts).toEqual([2, 2, 2, 2, 2]);
const [coordinator] = await fixture.db.select().from(nativeRunFinalizations)
.where(eq(nativeRunFinalizations.runId, fixture.runId));
expect((coordinator.recoveryHistory as Array<Record<string, unknown>>).filter(entry => entry.attemptKeptForProgress))
.toHaveLength(4);
});
it("does not credit a resume's work to a later provider attempt started by a normal retry", async () => {
const fixture = await seedInterruptedCheckpoint("PROGRESS-THEN-RETRY");
await resumeThenCrash(fixture);
await appendRunEvents(fixture, [{ eventType: "turn.accepted", turnId: "resumed" },
{ eventType: "tool.execution.completed", turnId: "resumed" }]);
// The resume (attempt 2) did work. A transport failure then started a
// normal local retry at attempt 3, which crashed before doing any work.
await fixture.db.update(nativeRunFinalizations).set({ attempt: 3 })
.where(eq(nativeRunFinalizations.runId, fixture.runId));
const { disposition } = await resumeThenCrash(fixture);
expect(disposition).toMatchObject({ kind: "blocked", reason: "execution_recovery_budget_exhausted" });
});
it("still exhausts the budget when restart resumes complete no work between crashes", async () => {
const fixture = await seedInterruptedCheckpoint("NO-PROGRESS-BETWEEN-CRASHES");
const outcomes: Array<{ kind?: string; attempt?: number; reason?: string }> = [];
for (let round = 1; round <= 3; round += 1) {
const { disposition, attempt } = await resumeThenCrash(fixture);
outcomes.push({ kind: disposition?.kind, attempt, reason: (disposition as { reason?: string }).reason });
// Restore activity and an accepted turn that fails before any work.
await appendRunEvents(fixture, [...restoreNoise(round), { eventType: "turn.failed", turnId: `continuation-${round}` }]);
}
expect(outcomes).toEqual([
expect.objectContaining({ kind: "resume_dead_runner", attempt: 2 }),
expect.objectContaining({ kind: "resume_dead_runner", attempt: 3 }),
expect.objectContaining({ kind: "blocked", reason: "execution_recovery_budget_exhausted" }),
]);
});
it("bounds a crash loop that makes progress before every crash", async () => {
const fixture = await seedInterruptedCheckpoint("PROGRESS-CRASH-LOOP");
const outcomes: Array<{ kind?: string; attempt?: number }> = [];
for (let round = 1; round <= NATIVE_PROGRESS_CREDIT_GENERATION_LIMIT + 3; round += 1) {
const { disposition, attempt } = await resumeThenCrash(fixture);
outcomes.push({ kind: disposition?.kind, attempt });
if (disposition?.kind !== "resume_dead_runner") break;
await appendRunEvents(fixture, [{ eventType: "turn.accepted", turnId: `loop-${round}` },
{ eventType: "item.completed", turnId: `loop-${round}`, kind: "assistant_message" }]);
}
// Generations 2..12 are claimed; credit stops at the generation limit and
// the remaining attempt is spent before recovery blocks.
expect(outcomes.at(-1)).toEqual({ kind: "blocked", attempt: undefined });
expect(outcomes.slice(0, -1).map(outcome => outcome.attempt))
.toEqual([2, ...Array(NATIVE_PROGRESS_CREDIT_GENERATION_LIMIT - 2).fill(2), 3]);
});
it("terminalizes the recorded failed-checkpoint incident atomically instead of resuming it on upgrade", async () => {
const fixture = await seedRun("FAILED-CHECKPOINT");
await fixture.db.update(heartbeatRuns).set({ runnerProfileJson: { sessionCheckpoint: {