mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-12 14:19:00 +01:00
fix(recovery): charge a restart attempt only when the provider accepted a turn and made no progress (#19948)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - A run has an execution recovery budget: three provider attempts for native runs, and two failure retries for legacy runs > - In the restart chaos campaign, control-plane restarts used up that budget for runs that were healthy (RC-22, 22b, 22c, 40 and 60) > - #19720 and #19747 fixed two variants with watermark heuristics, and legacy interrupted preparation still spent the failure budget > - The durable-ownership plan (section F, invariant 5) asks for one rule: charge an attempt only when a recorded outcome proves that provider work started and no progress followed > - This pull request adds that rule and moves the native restart charge and the legacy preparation retry onto it > - The benefit is that a restart never spends budget for work the provider never started, while real failure loops and crash loops stay bounded ## Linked Issues or Issue Description Refs #19720, #19747, #19856 **What happened?** A control-plane restart charges a provider attempt by heuristics. The first crash restart always charges. A later restart charges unless the previous resume had no start evidence (#19747) or completed work (#19720). A runner that spawned but whose provider never accepted a turn counted as started, so a K6 crash loop could still spend the budget. Legacy runs whose preparation a restart interrupted (`providerWorkStarted: false`) spent `failureRetries`, the same counter as real failures (`execution-recovery-attempt.ts`). **Expected behavior** An attempt is charged only when the provider accepted a turn (`turn.accepted` or `turn.started`) and no accepted turn then completed work. A startup before spawn, a runner that spawned but was never admitted, and a `turn.start` the provider did not accept are not charged. The generation-12 cap still bounds uncharged restarts. **Steps to reproduce** Start a native run at its last attempt. Kill the controller after the runner spawns but before the provider accepts a turn. Before this change, the next restart claim blocks the run with `execution_recovery_budget_exhausted`. ## What Changed - New `server/src/services/native-runtime/provider-attempt-outcome.ts`: the one rule. `readProviderAttemptOutcome()` returns `provider_not_started`, `provider_started_without_progress` or `provider_progressed` from run events. `providerAttemptChargesBudget()` charges only the second. `providerAttemptWindowStart()` starts the attempt window at the latest `native.process_start_requested`, so earlier work never counts for a later attempt. `PROVIDER_ATTEMPT_CREDIT_GENERATION_LIMIT` (12) is the shared cap. - `restartResumeKeepsAttempt()` (`native-restart-recovery.ts`) now applies that rule to the attempt the restart interrupted, including the first restart. It keeps the unstarted-resume carry from #19747, the reattachment rule (an adopted provider counts as started), the drain exemption from #19625, and the generation-12 cap. A history entry from an older release without a recorded position still charges. - Legacy runs: `interrupted_preparation` is now an uncharged retry lane (`execution-recovery-attempt.ts`). It keeps its own counter, bounded at the same cap (12), and never changes `failureRetries`. `retries.ts` uses the longest transient delay after the second lane attempt. - `doc/execution-semantics.md` states the rule. - Accounting has its own "provider work started" readers (inventory Summary finding 5). Phase 3 owns those; this PR exposes the rule for them to adopt and does not change them. ## Verification - `native-runner-restart-recovery.integration.test.ts` (embedded Postgres, real `claimNativeRestartRecoveries`): 129 passed. New cases, red on master: - not charged at the last attempt for a claimed remote bootstrap, a remote bootstrap retry, and a claimed unlaunched remote run; - not charged when the runner spawned but was never admitted, and when the provider did not accept `turn.start`; - not charged on the first restart when the original attempt already made progress; - a normal retry that spawned but was never admitted is not charged, and earlier work still does not count for it. - Counter-cases, green on master and with the change: an accepted `turn.accepted` or `turn.started` with no progress is charged; the generation cap blocks at the last attempt; crash loops with and without progress stay bounded; the #19720 and #19747 cases still pass. - `execution-recovery-attempt.test.ts`: new lane tests (red on master). - `workspace-recovery-preparation-dispatch.test.ts` (25 passed), `heartbeat-process-recovery.test.ts` (interrupted-run cases), `run-dispatch/adapters/postgres.test.ts` (interrupted cases). - `server` typecheck (`scripts/typecheck-layers.mjs`, low-memory mode): no errors. - Not run: chaos waves (the coordinator schedules them on this host), and the full server suite (CI covers it). ## Risks - Behavior change: the first crash restart no longer charges when no accepted turn is recorded. If a crash lands before the controller projects the provider's `turn.accepted`, that attempt is not charged. This fails toward more retries, and the generation-12 cap bounds it. - A legacy preparation crash loop now gets up to 12 retries instead of 2. Each retry still passes the normal issue gates and keeps the transient delay. - The new queries use the `(run_id, seq)` index, once per restart claim. Nothing new runs before `turn.start`. > 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:
1 parent
c6e4ed1547
commit
a4a3e342ec
10 files changed
+405
-75
No files matched your search
@@ -7557,11 +7557,16 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
|
||||
.from(heartbeatRuns)
|
||||
.where(eq(heartbeatRuns.retryOfRunId, runId))
|
||||
.then((rows) => rows[0] ?? null);
|
||||
// The interrupted preparation spent no provider work: it starts its own
|
||||
// lane and keeps the one transient failure already spent (section F).
|
||||
expect(successor).toMatchObject({
|
||||
agentId,
|
||||
status: "scheduled_retry",
|
||||
scheduledRetryAttempt: 2,
|
||||
scheduledRetryAttempt: 1,
|
||||
scheduledRetryReason: "interrupted_preparation",
|
||||
contextSnapshot: expect.objectContaining({
|
||||
executionRetryAccounting: expect.objectContaining({ failureRetries: 1 }),
|
||||
}),
|
||||
});
|
||||
const sourceIssue = await db
|
||||
.select({ status: issues.status })
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
import { describe, expect, it } from "vitest";
|
||||
import { accountingForScheduledRetry, executionFailureRetryCount, executionRetryAttemptCount } from "./execution-recovery-attempt.js";
|
||||
import { accountingForScheduledRetry, executionFailureRetryCount, executionRetryAttemptCount, INTERRUPTED_PREPARATION_RETRY_MAX_ATTEMPTS, INTERRUPTED_PREPARATION_RETRY_REASON } from "./execution-recovery-attempt.js";
|
||||
import { RUNNER_REVISION_WAIT_RETRY_REASON } from "./runner-revision-mismatch.js";
|
||||
import { RUNNER_REVISION_WAIT_RETRY_REASON as DISPATCH_RUNNER_REVISION_WAIT_RETRY_REASON } from "../modules/run-dispatch/index.js";
|
||||
|
||||
@@ -67,3 +67,43 @@ describe("Runner update waits", () => {
|
||||
expect(executionRetryAttemptCount(waited, "transient_failure")).toBe(1);
|
||||
});
|
||||
});
|
||||
|
||||
// Section F: a preparation interrupted before provider work started spent no
|
||||
// provider work, so it never charges the failure budget. Its own lane counts
|
||||
// its retries, up to the same cap as native controller generations.
|
||||
describe("interrupted preparation retries", () => {
|
||||
const accounting = (failureRetries: number) => ({ version: 1, failureRetries, maxTurnContinuations: 0 });
|
||||
it("keeps the failure count when scheduling an interrupted-preparation retry", () => {
|
||||
const source = { scheduledRetryReason: "transient_failure", scheduledRetryAttempt: 1,
|
||||
contextSnapshot: { executionRetryAccounting: accounting(1) } };
|
||||
expect(executionRetryAttemptCount(source, INTERRUPTED_PREPARATION_RETRY_REASON)).toBe(0);
|
||||
expect(accountingForScheduledRetry(source, INTERRUPTED_PREPARATION_RETRY_REASON, 1))
|
||||
.toEqual({ ...accounting(1), interruptedPreparations: 1 });
|
||||
});
|
||||
it("counts consecutive interrupted preparations in their own lane, never as failures", () => {
|
||||
const retry = { scheduledRetryReason: INTERRUPTED_PREPARATION_RETRY_REASON, scheduledRetryAttempt: 5,
|
||||
contextSnapshot: { executionRetryAccounting: accounting(0) } };
|
||||
expect(executionRetryAttemptCount(retry, INTERRUPTED_PREPARATION_RETRY_REASON)).toBe(5);
|
||||
expect(executionFailureRetryCount(retry)).toBe(0);
|
||||
expect(accountingForScheduledRetry(retry, INTERRUPTED_PREPARATION_RETRY_REASON, 6))
|
||||
.toEqual({ ...accounting(0), interruptedPreparations: 6 });
|
||||
expect(INTERRUPTED_PREPARATION_RETRY_MAX_ATTEMPTS).toBe(12);
|
||||
});
|
||||
it("keeps the interrupted-preparation count across a wait retry in between", () => {
|
||||
const prepared = accountingForScheduledRetry({ scheduledRetryReason: INTERRUPTED_PREPARATION_RETRY_REASON,
|
||||
scheduledRetryAttempt: 4, contextSnapshot: { executionRetryAccounting: accounting(0) } },
|
||||
INTERRUPTED_PREPARATION_RETRY_REASON, 5);
|
||||
expect(prepared).toEqual({ ...accounting(0), interruptedPreparations: 5 });
|
||||
// A workspace wait in between carries the chain's accounting forward.
|
||||
const waited = { scheduledRetryReason: "workspace_busy", scheduledRetryAttempt: 1,
|
||||
contextSnapshot: { executionRetryAccounting: accountingForScheduledRetry(
|
||||
{ contextSnapshot: { executionRetryAccounting: prepared } }, "workspace_busy", 1) } };
|
||||
expect(executionRetryAttemptCount(waited, INTERRUPTED_PREPARATION_RETRY_REASON)).toBe(5);
|
||||
});
|
||||
it("still charges a real failure after interrupted preparations", () => {
|
||||
const retry = { scheduledRetryReason: INTERRUPTED_PREPARATION_RETRY_REASON, scheduledRetryAttempt: 3,
|
||||
contextSnapshot: { executionRetryAccounting: accounting(1) } };
|
||||
expect(executionRetryAttemptCount(retry, "transient_failure")).toBe(1);
|
||||
expect(accountingForScheduledRetry(retry, "transient_failure", 2).failureRetries).toBe(2);
|
||||
});
|
||||
});
|
||||
@@ -1,4 +1,21 @@
|
||||
import { RUNNER_REVISION_WAIT_RETRY_REASON } from "./runner-revision-mismatch.js";
|
||||
import { PROVIDER_ATTEMPT_CREDIT_GENERATION_LIMIT } from "./native-runtime/provider-attempt-outcome.js";
|
||||
|
||||
/**
|
||||
* A preparation that a restart interrupted before any provider work started
|
||||
* (its certificate records `providerWorkStarted: false`) spent no provider
|
||||
* work, so it never charges the failure budget (durable-ownership plan,
|
||||
* section F). Its own counter keeps a crash loop bounded, at the same limit
|
||||
* as the native controller-generation cap.
|
||||
*/
|
||||
export const INTERRUPTED_PREPARATION_RETRY_REASON = "interrupted_preparation";
|
||||
export const INTERRUPTED_PREPARATION_RETRY_MAX_ATTEMPTS = PROVIDER_ATTEMPT_CREDIT_GENERATION_LIMIT;
|
||||
|
||||
/** Retry lanes that keep their own counter and never spend failures. */
|
||||
const UNCHARGED_RETRY_LANES: ReadonlySet<string> = new Set([
|
||||
"workspace_busy", "ai_connection_busy", "ai_connection_pool_wait", RUNNER_REVISION_WAIT_RETRY_REASON,
|
||||
INTERRUPTED_PREPARATION_RETRY_REASON,
|
||||
]);
|
||||
|
||||
type RetryRun = {
|
||||
scheduledRetryAttempt?: number | null;
|
||||
@@ -12,6 +29,11 @@ export interface ExecutionRetryAccounting {
|
||||
version: 1;
|
||||
failureRetries: number;
|
||||
maxTurnContinuations: number;
|
||||
/**
|
||||
* Interrupted-preparation retries in this chain. Saved with the chain, so a
|
||||
* wait retry in between never resets it. Omitted while zero.
|
||||
*/
|
||||
interruptedPreparations?: number;
|
||||
}
|
||||
|
||||
function count(value: unknown): number | null {
|
||||
@@ -25,7 +47,9 @@ function savedAccounting(run: RetryRun): ExecutionRetryAccounting | null {
|
||||
const failureRetries = count(saved.failureRetries);
|
||||
const maxTurnContinuations = count(saved.maxTurnContinuations);
|
||||
if (saved.version !== 1 || failureRetries === null || maxTurnContinuations === null) return null;
|
||||
return { version: 1, failureRetries, maxTurnContinuations };
|
||||
const interruptedPreparations = count(saved.interruptedPreparations) ?? 0;
|
||||
return { version: 1, failureRetries, maxTurnContinuations,
|
||||
...(interruptedPreparations > 0 ? { interruptedPreparations } : {}) };
|
||||
}
|
||||
|
||||
function historicalFailureCount(run: RetryRun): number {
|
||||
@@ -44,12 +68,16 @@ function historicalFailureCount(run: RetryRun): number {
|
||||
|
||||
export function executionRetryAccounting(run: RetryRun): ExecutionRetryAccounting {
|
||||
const saved = savedAccounting(run);
|
||||
const nonFailureLane = ["max_turns_continuation", "issue_disposition_repair", "workspace_busy", "ai_connection_busy", "ai_connection_pool_wait", RUNNER_REVISION_WAIT_RETRY_REASON].includes(run.scheduledRetryReason ?? "");
|
||||
const reason = run.scheduledRetryReason ?? "";
|
||||
const nonFailureLane = reason === "max_turns_continuation" || reason === "issue_disposition_repair" || UNCHARGED_RETRY_LANES.has(reason);
|
||||
const interruptedPreparations = Math.max(saved?.interruptedPreparations ?? 0,
|
||||
reason === INTERRUPTED_PREPARATION_RETRY_REASON ? count(run.scheduledRetryAttempt) ?? 0 : 0);
|
||||
return {
|
||||
version: 1,
|
||||
failureRetries: Math.max(saved?.failureRetries ?? 0, saved && nonFailureLane ? 0 : historicalFailureCount(run)),
|
||||
maxTurnContinuations: Math.max(saved?.maxTurnContinuations ?? 0,
|
||||
run.scheduledRetryReason === "max_turns_continuation" ? count(run.scheduledRetryAttempt) ?? 0 : 0),
|
||||
...(interruptedPreparations > 0 ? { interruptedPreparations } : {}),
|
||||
};
|
||||
}
|
||||
|
||||
@@ -59,7 +87,8 @@ export function executionFailureRetryCount(run: RetryRun): number {
|
||||
}
|
||||
|
||||
export function executionRetryAttemptCount(run: RetryRun, reason: string): number {
|
||||
if (reason === "workspace_busy" || reason === "ai_connection_busy" || reason === "ai_connection_pool_wait" || reason === RUNNER_REVISION_WAIT_RETRY_REASON) {
|
||||
if (reason === INTERRUPTED_PREPARATION_RETRY_REASON) return executionRetryAccounting(run).interruptedPreparations ?? 0;
|
||||
if (UNCHARGED_RETRY_LANES.has(reason)) {
|
||||
return run.scheduledRetryReason === reason ? count(run.scheduledRetryAttempt) ?? 0 : 0;
|
||||
}
|
||||
const accounting = executionRetryAccounting(run);
|
||||
@@ -69,7 +98,7 @@ export function executionRetryAttemptCount(run: RetryRun, reason: string): numbe
|
||||
export function accountingForScheduledRetry(run: RetryRun, reason: string, attempt: number): ExecutionRetryAccounting {
|
||||
const accounting = executionRetryAccounting(run);
|
||||
if (reason === "max_turns_continuation") accounting.maxTurnContinuations = attempt;
|
||||
else if (reason !== "workspace_busy" && reason !== "ai_connection_busy" && reason !== "ai_connection_pool_wait"
|
||||
&& reason !== RUNNER_REVISION_WAIT_RETRY_REASON) accounting.failureRetries = attempt;
|
||||
else if (reason === INTERRUPTED_PREPARATION_RETRY_REASON) accounting.interruptedPreparations = attempt;
|
||||
else if (!UNCHARGED_RETRY_LANES.has(reason)) accounting.failureRetries = attempt;
|
||||
return accounting;
|
||||
}
|
||||
@@ -21,6 +21,8 @@ import {
|
||||
executionFailureRetryCount,
|
||||
executionRetryAttemptCount,
|
||||
accountingForScheduledRetry,
|
||||
INTERRUPTED_PREPARATION_RETRY_MAX_ATTEMPTS,
|
||||
INTERRUPTED_PREPARATION_RETRY_REASON,
|
||||
} from "../execution-recovery-attempt.js";
|
||||
import { WORKSPACE_GIT_SCAN_ERROR_CODES } from "../workspace-git-operation-scheduler.js";
|
||||
import { REMOTE_WORKSPACE_PREPARATION_TIMEOUT_CODE, REMOTE_WORKSPACE_RESTORE_FAILED_CODE } from "../remote-workspace-transport.js";
|
||||
@@ -478,19 +480,27 @@ export function createHeartbeatRetries(db: Db, dependencies: HeartbeatRetryDepen
|
||||
preparationRecovery.kind === "bootstrap" && preparationRecovery.providerWorkStarted === false
|
||||
&& (run.status === "interrupted" || await isInterruptedNativePreparation(db, run));
|
||||
const retryReason = opts?.retryReason ??
|
||||
(interruptedPreparation ? "interrupted_preparation" : BOUNDED_TRANSIENT_HEARTBEAT_RETRY_REASON);
|
||||
(interruptedPreparation ? INTERRUPTED_PREPARATION_RETRY_REASON : BOUNDED_TRANSIENT_HEARTBEAT_RETRY_REASON);
|
||||
const wakeReason =
|
||||
opts?.wakeReason ?? BOUNDED_TRANSIENT_HEARTBEAT_RETRY_WAKE_REASON;
|
||||
// Interrupted preparation spent no provider work: its own lane, bounded
|
||||
// like the native controller-generation cap (execution-recovery-attempt).
|
||||
const interruptedPreparationLane = retryReason === INTERRUPTED_PREPARATION_RETRY_REASON;
|
||||
const maxAttempts = Math.max(
|
||||
0,
|
||||
Math.floor(
|
||||
opts?.maxAttempts ?? BOUNDED_TRANSIENT_HEARTBEAT_RETRY_MAX_ATTEMPTS,
|
||||
opts?.maxAttempts ?? (interruptedPreparationLane
|
||||
? INTERRUPTED_PREPARATION_RETRY_MAX_ATTEMPTS : BOUNDED_TRANSIENT_HEARTBEAT_RETRY_MAX_ATTEMPTS),
|
||||
),
|
||||
);
|
||||
const consumedAttempts = executionRetryAttemptCount(run, retryReason);
|
||||
const nextAttempt = consumedAttempts + 1;
|
||||
const computedBaseSchedule =
|
||||
opts?.delayMs != null
|
||||
// An interrupted preparation never spends failures, but it never revives
|
||||
// a chain whose real failures already spent the failure budget either.
|
||||
const failureBudgetSpent = interruptedPreparationLane && opts?.maxAttempts == null
|
||||
&& executionFailureRetryCount(run) >= BOUNDED_TRANSIENT_HEARTBEAT_RETRY_MAX_ATTEMPTS;
|
||||
const computedBaseSchedule = failureBudgetSpent ? null
|
||||
: opts?.delayMs != null
|
||||
? nextAttempt <= maxAttempts
|
||||
? {
|
||||
attempt: nextAttempt,
|
||||
@@ -503,7 +513,14 @@ export function createHeartbeatRetries(db: Db, dependencies: HeartbeatRetryDepen
|
||||
}
|
||||
: null
|
||||
: nextAttempt <= maxAttempts
|
||||
? computeBoundedTransientHeartbeatRetrySchedule(
|
||||
? interruptedPreparationLane
|
||||
? (() => {
|
||||
// Later lane attempts keep the longest transient delay.
|
||||
const schedule = computeBoundedTransientHeartbeatRetrySchedule(
|
||||
Math.min(nextAttempt, BOUNDED_TRANSIENT_HEARTBEAT_RETRY_MAX_ATTEMPTS), now, opts?.random);
|
||||
return schedule ? { ...schedule, attempt: nextAttempt } : null;
|
||||
})()
|
||||
: computeBoundedTransientHeartbeatRetrySchedule(
|
||||
nextAttempt,
|
||||
now,
|
||||
opts?.random,
|
||||
|
||||
@@ -34,6 +34,7 @@ import {
|
||||
nativeRunFinalizations,
|
||||
} from "@paperclipai/db";
|
||||
import { REMOTE_CONTROLLER_LEASE_MS } from "./native-runner-recovery-contract.js";
|
||||
import { PROVIDER_ATTEMPT_CREDIT_GENERATION_LIMIT, providerAttemptChargesBudget, providerAttemptWindowStart, readProviderAttemptOutcome } from "./provider-attempt-outcome.js";
|
||||
import { readProcessStartedAt } from "../hot-restart.js";
|
||||
import { ephemeralServerEnabled } from "../ephemeral-server-mode.js";
|
||||
import { persistActivity } from "../activity-log.js";
|
||||
@@ -169,20 +170,18 @@ export function suspendedByCleanDrain(coordinator: {
|
||||
}
|
||||
|
||||
/**
|
||||
* 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. A resume the controller died before starting is not charged
|
||||
* either (restartResumeKeepsAttempt). 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.
|
||||
* A controller restart (a crash, so no drain record) charges the interrupted
|
||||
* attempt only by the work-based rule in provider-attempt-outcome.ts: the
|
||||
* provider accepted a turn in that attempt and no accepted turn completed work
|
||||
* before the crash. Each claim records the run's event-log position, so the
|
||||
* next claim judges only what happened after it. A resume the controller died
|
||||
* before starting, a runner that never admitted its provider, and a turn.start
|
||||
* the provider did not accept are not charged. Credit stops after
|
||||
* NATIVE_PROGRESS_CREDIT_GENERATION_LIMIT controller generations, so a crash
|
||||
* loop that never starts the provider, or 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;
|
||||
export const NATIVE_PROGRESS_CREDIT_GENERATION_LIMIT = PROVIDER_ATTEMPT_CREDIT_GENERATION_LIMIT;
|
||||
|
||||
type RecoveryClaimProgressEntry = { disposition?: unknown; progressWatermarkSeq?: unknown; progressAttempt?: unknown };
|
||||
|
||||
@@ -193,6 +192,13 @@ type RecoveryClaimProgressEntry = { disposition?: unknown; progressWatermarkSeq?
|
||||
export function lastRecoveryClaimResume(
|
||||
recoveryHistory: ReadonlyArray<RecoveryClaimProgressEntry>,
|
||||
): { watermark: number; attempt: number; launchesRunner: boolean } | null {
|
||||
const previous = previousRecoveryClaim(recoveryHistory);
|
||||
return previous === "none" || previous === "unrecorded" ? null : previous;
|
||||
}
|
||||
|
||||
function previousRecoveryClaim(
|
||||
recoveryHistory: ReadonlyArray<RecoveryClaimProgressEntry>,
|
||||
): { watermark: number; attempt: number; launchesRunner: boolean } | "none" | "unrecorded" {
|
||||
for (let index = recoveryHistory.length - 1; index >= 0; index -= 1) {
|
||||
const entry = recoveryHistory[index];
|
||||
if (!RECOVERY_CLAIM_DISPOSITIONS.has(entry?.disposition)) continue;
|
||||
@@ -202,9 +208,9 @@ export function lastRecoveryClaimResume(
|
||||
const launchesRunner = entry?.disposition === "resume_dead_runner" || entry?.disposition === "bootstrap_incomplete";
|
||||
return typeof watermark === "number" && Number.isSafeInteger(watermark) && watermark >= 0
|
||||
&& typeof attempt === "number" && Number.isSafeInteger(attempt)
|
||||
? { watermark, attempt, launchesRunner } : null;
|
||||
? { watermark, attempt, launchesRunner } : "unrecorded";
|
||||
}
|
||||
return null;
|
||||
return "none";
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -240,15 +246,18 @@ async function nativeRunEventWatermark(tx: Pick<Db, "select">, runId: string): P
|
||||
|
||||
/**
|
||||
* Whether this restart claim keeps the provider attempt instead of charging
|
||||
* one. Both cases are judged from the previous claim's recorded position:
|
||||
* - That resume launched a new runner (not a reattachment) and never started:
|
||||
* no runner was requested and no runner event was
|
||||
* recorded before the controller died (for example in a K6 crash loop). It
|
||||
* used no provider work, so this claim replaces it and carries its charge:
|
||||
* the attempt ends at the attempt that resume was to run as, never above it.
|
||||
* - That resume made progress on the current attempt (see above). A normal
|
||||
* retry in between starts a new attempt, and earlier work does not count.
|
||||
* Neither applies at or after NATIVE_PROGRESS_CREDIT_GENERATION_LIMIT.
|
||||
* one, judged by the work-based rule (provider-attempt-outcome.ts) over the
|
||||
* attempt the restart interrupted:
|
||||
* - The first restart judges the original execution's attempt.
|
||||
* - A later restart judges the attempt that ran after the previous claim's
|
||||
* recorded position. A normal retry in between starts a new attempt window,
|
||||
* so earlier work does not count for it.
|
||||
* - When the previous claim launched a new runner but that resume never
|
||||
* reached execution, it consumed nothing and this claim replaces it. Its
|
||||
* charge decision was never applied, so this claim carries it: the attempt
|
||||
* ends at the attempt that resume was to run as, never above it.
|
||||
* - A claim history written before claims recorded their position charges.
|
||||
* At or after NATIVE_PROGRESS_CREDIT_GENERATION_LIMIT every restart charges.
|
||||
*/
|
||||
export async function restartResumeKeepsAttempt(tx: Pick<Db, "select">, input: {
|
||||
runId: string;
|
||||
@@ -260,32 +269,27 @@ export async function restartResumeKeepsAttempt(tx: Pick<Db, "select">, input: {
|
||||
}): Promise<boolean> {
|
||||
if (input.coordinator.attempt <= 0) return false;
|
||||
if (input.coordinator.controllerGeneration >= NATIVE_PROGRESS_CREDIT_GENERATION_LIMIT) return false;
|
||||
const previous = lastRecoveryClaimResume(input.coordinator.recoveryHistory);
|
||||
if (previous === null) return false;
|
||||
// Only a claim that launches a new runner can be unstarted. A reattachment
|
||||
// adopted provider work that had already started, so only progress counts.
|
||||
if (previous.launchesRunner && !await nativeExecutionStartedSince(tx, input.runId, previous.watermark))
|
||||
const previous = previousRecoveryClaim(input.coordinator.recoveryHistory);
|
||||
if (previous === "unrecorded") return false;
|
||||
if (previous !== "none" && previous.launchesRunner
|
||||
&& !await nativeExecutionStartedSince(tx, input.runId, previous.watermark))
|
||||
return input.coordinator.attempt >= previous.attempt;
|
||||
if (previous.attempt !== input.coordinator.attempt) return false;
|
||||
const watermark = previous.watermark;
|
||||
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 claimedAt = previous === "none" ? 0 : previous.watermark;
|
||||
// A reattachment adopted a provider that had already started, and it keeps
|
||||
// the attempt; a normal retry after it starts a new attempt.
|
||||
const adoptedProviderWork = previous !== "none" && !previous.launchesRunner
|
||||
&& input.coordinator.attempt === previous.attempt;
|
||||
const window = adoptedProviderWork ? { seq: claimedAt }
|
||||
: await providerAttemptWindowStart(tx, {
|
||||
runId: input.runId, sinceSeq: claimedAt, providerAttempt: input.coordinator.attempt,
|
||||
});
|
||||
if (window === "unstarted") return true;
|
||||
const outcome = await readProviderAttemptOutcome(tx, {
|
||||
runId: input.runId,
|
||||
sinceSeq: window.seq,
|
||||
adoptedProviderWork,
|
||||
});
|
||||
return !providerAttemptChargesBudget(outcome);
|
||||
}
|
||||
|
||||
|
||||
|
||||
+117
-8
@@ -2546,7 +2546,6 @@ process.on("SIGTERM", () => child.kill("SIGTERM"));
|
||||
["the state was adopted", { mutate: core => { core.store.state.bootstrapHistory!.adopted = true; } },
|
||||
"ambiguous_remote_recovery_evidence"],
|
||||
["no checkpoint exists", { checkpoint: false }, "ambiguous_remote_recovery_evidence"],
|
||||
["the bootstrap budget is spent", { attempt: 3 }, "execution_recovery_budget_exhausted"],
|
||||
])("keeps a claimed remote bootstrap held when %s", async (label, options, reason) => {
|
||||
const fixture = await seedClaimedRemoteBootstrap(label.replace(/\W+/g, "-"), options);
|
||||
expect(await claimNativeRestartRecoveries({ db: fixture.db, controller: successor,
|
||||
@@ -2743,14 +2742,38 @@ process.on("SIGTERM", () => child.kill("SIGTERM"));
|
||||
{ kind: "blocked", reason: "ambiguous_remote_recovery_evidence" }],
|
||||
["a runner event was committed", { bootstrapRetry: true, event: { eventType: "item.updated", sourceEventId: "runner-event-1" } },
|
||||
{ kind: "blocked", reason: "ambiguous_remote_recovery_evidence" }],
|
||||
["the bootstrap budget is spent", { bootstrapRetry: true, attempt: 3 },
|
||||
{ kind: "blocked", reason: "execution_recovery_budget_exhausted" }],
|
||||
])("keeps a remote bootstrap retry held when %s", async (label, options, expected) => {
|
||||
const fixture = await seedUnlaunchedRemoteClaim(`RETRY-${label.replace(/\W+/g, "-")}`, options);
|
||||
expect(await claimNativeRestartRecoveries({ db: fixture.db, controller: successor,
|
||||
restartKind: "hard", runIds: [fixture.runId] })).toEqual([expect.objectContaining({ runId: fixture.runId, ...expected })]);
|
||||
});
|
||||
|
||||
// RC-22 / section F: a bootstrap whose provider never accepted a turn spent
|
||||
// no provider work, so a restart does not charge it, even at the last
|
||||
// attempt. The controller-generation cap still bounds a crash loop.
|
||||
it.each([
|
||||
["claimed remote bootstrap", () => seedClaimedRemoteBootstrap("UNCHARGED-CLAIMED", { attempt: 3 })],
|
||||
["remote bootstrap retry", () => seedUnlaunchedRemoteClaim("UNCHARGED-RETRY", { bootstrapRetry: true, attempt: 3 })],
|
||||
["claimed unlaunched remote run", () => seedUnlaunchedRemoteClaim("UNCHARGED-UNLAUNCHED", { attempt: 3 })],
|
||||
] as const)("does not charge a %s at its last attempt when no provider turn was accepted", async (_label, seed) => {
|
||||
const fixture = await seed();
|
||||
expect(await claimNativeRestartRecoveries({ db: fixture.db, controller: successor,
|
||||
restartKind: "hard", runIds: [fixture.runId] })).toEqual([expect.objectContaining({
|
||||
kind: "bootstrap_incomplete", runId: fixture.runId, providerAttempt: 3, chargesProviderAttempt: false })]);
|
||||
});
|
||||
|
||||
it.each([
|
||||
["claimed remote bootstrap", (label: string) => seedClaimedRemoteBootstrap(label, { attempt: 3 })],
|
||||
["remote bootstrap retry", (label: string) => seedUnlaunchedRemoteClaim(label, { bootstrapRetry: true, attempt: 3 })],
|
||||
] as const)("blocks a %s at its last attempt once the controller generation cap is reached", async (label, seed) => {
|
||||
const fixture = await seed(`GEN-CAP-${label.replace(/\W+/g, "-")}`);
|
||||
await fixture.db.update(nativeRunFinalizations).set({ controllerGeneration: NATIVE_PROGRESS_CREDIT_GENERATION_LIMIT })
|
||||
.where(eq(nativeRunFinalizations.runId, fixture.runId));
|
||||
expect(await claimNativeRestartRecoveries({ db: fixture.db, controller: successor,
|
||||
restartKind: "hard", runIds: [fixture.runId] })).toEqual([expect.objectContaining({
|
||||
kind: "blocked", runId: fixture.runId, reason: "execution_recovery_budget_exhausted" })]);
|
||||
});
|
||||
|
||||
it("keeps a paused agent's remote bootstrap retry waiting instead of starting it", async () => {
|
||||
const fixture = await seedUnlaunchedRemoteClaim("PAUSED-RETRY", { bootstrapRetry: true });
|
||||
await fixture.db.update(agents).set({ status: "paused" }).where(eq(agents.id, agentId));
|
||||
@@ -2807,8 +2830,6 @@ process.on("SIGTERM", () => child.kill("SIGTERM"));
|
||||
{ kind: "blocked", reason: "ambiguous_remote_recovery_evidence" }],
|
||||
["a provider session is recorded", { run: { sessionIdAfter: "provider-session" } },
|
||||
{ kind: "blocked", reason: "ambiguous_remote_recovery_evidence" }],
|
||||
["the bootstrap budget is spent", { attempt: 3 },
|
||||
{ kind: "blocked", reason: "execution_recovery_budget_exhausted" }],
|
||||
["the claiming controller is still alive", { liveController: true },
|
||||
{ kind: "awaiting_evidence", reason: "controller_still_alive" }],
|
||||
])("keeps a claimed unlaunched remote run held when %s", async (label, options, expected) => {
|
||||
@@ -2844,6 +2865,9 @@ process.on("SIGTERM", () => child.kill("SIGTERM"));
|
||||
dispositionOnlyRecoveryConsumed: true, dispositionOnlyRecoveryTurnId: "continuation-turn" } : {}),
|
||||
} } }).where(eq(heartbeatRuns.id, fixture.runId));
|
||||
await fixture.db.update(nativeRunFinalizations).set({ attempt }).where(eq(nativeRunFinalizations.runId, fixture.runId));
|
||||
// The interrupted attempt's provider accepted a turn and completed no work.
|
||||
await appendRunEvents(fixture, [{ eventType: "native.process_start_requested" },
|
||||
{ eventType: "turn.accepted", turnId: lostContinuation ? "continuation-turn" : "interrupted-turn" }]);
|
||||
const dispositions = await claimNativeRestartRecoveries({
|
||||
db: fixture.db, controller: successor, restartKind: "hot", runIds: [fixture.runId],
|
||||
});
|
||||
@@ -2869,14 +2893,17 @@ process.on("SIGTERM", () => child.kill("SIGTERM"));
|
||||
error: { code: "provider_turn_lost_on_restore", recoverable: true },
|
||||
}) }],
|
||||
} } }).where(eq(heartbeatRuns.id, fixture.runId));
|
||||
// The original execution claimed attempt 1.
|
||||
// The original execution claimed attempt 1, and its provider accepted the
|
||||
// turn that the restart then interrupted before it completed any work.
|
||||
await fixture.db.update(nativeRunFinalizations).set({ attempt: 1, controllerGeneration: 1 })
|
||||
.where(eq(nativeRunFinalizations.runId, fixture.runId));
|
||||
await appendRunEvents(fixture, [{ eventType: "native.process_start_requested" },
|
||||
{ eventType: "turn.accepted", turnId: "interrupted-turn" }]);
|
||||
return fixture;
|
||||
}
|
||||
|
||||
async function appendRunEvents(fixture: Awaited<ReturnType<typeof seedRun>>, events: Array<{
|
||||
eventType: string; turnId?: string; kind?: string;
|
||||
eventType: string; turnId?: string; kind?: string; payload?: Record<string, unknown>;
|
||||
}>) {
|
||||
const [last] = await fixture.db.select({ seq: sql<number>`coalesce(max(${heartbeatRunEvents.seq}), 0)::int` })
|
||||
.from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, fixture.runId));
|
||||
@@ -2885,7 +2912,7 @@ process.on("SIGTERM", () => child.kill("SIGTERM"));
|
||||
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",
|
||||
payload: event.payload ?? { prpEvent: { runId: fixture.runId, eventType: event.eventType, sourceKind: "runner",
|
||||
...(event.turnId ? { turnId: event.turnId } : {}),
|
||||
payload: event.kind ? { kind: event.kind } : {} } } });
|
||||
}
|
||||
@@ -2949,10 +2976,59 @@ process.on("SIGTERM", () => child.kill("SIGTERM"));
|
||||
await fixture.db.update(nativeRunFinalizations).set({ attempt: 3 })
|
||||
.where(eq(nativeRunFinalizations.runId, fixture.runId));
|
||||
await appendRunEvents(fixture, [{ eventType: "native.process_start_requested" }]);
|
||||
// That retry spawned but its provider never accepted a turn: not charged.
|
||||
expect(await resumeThenCrash(fixture)).toMatchObject({
|
||||
disposition: { kind: "resume_dead_runner", chargesProviderAttempt: false }, attempt: 3 });
|
||||
// The next resume's provider accepts a turn and does no work. The first
|
||||
// resume's work belongs to an earlier attempt, so this one is charged.
|
||||
await appendRunEvents(fixture, [{ eventType: "native.process_start_requested" },
|
||||
{ eventType: "turn.accepted", turnId: "retry" }]);
|
||||
const { disposition } = await resumeThenCrash(fixture);
|
||||
expect(disposition).toMatchObject({ kind: "blocked", reason: "execution_recovery_budget_exhausted" });
|
||||
});
|
||||
|
||||
// Section F: the not-charged outcomes, judged on the attempt a crash
|
||||
// interrupted. Each starts at the last attempt, so a charge would block.
|
||||
it.each([
|
||||
{ label: "the runner was never requested", events: [] },
|
||||
{ label: "the runner spawned but was never admitted", events: [{ eventType: "native.process_start_requested" },
|
||||
{ eventType: "native.process_identity_recorded" }, { eventType: "session.resumed" }] },
|
||||
{ label: "the provider did not accept turn.start", events: [{ eventType: "native.process_start_requested" },
|
||||
{ eventType: "turn.submitted" }, { eventType: "turn.failed", turnId: "never-accepted" }] },
|
||||
])("does not charge a resume whose provider never accepted a turn ($label)", async ({ label, events }) => {
|
||||
const fixture = await seedInterruptedCheckpoint(`NOT-CHARGED-${label.replace(/\W+/g, "-")}`);
|
||||
expect(await resumeThenCrash(fixture)).toMatchObject({ attempt: 2 });
|
||||
await fixture.db.update(nativeRunFinalizations).set({ attempt: 3 })
|
||||
.where(eq(nativeRunFinalizations.runId, fixture.runId));
|
||||
// The previous resume is recorded to run as attempt 3.
|
||||
await fixture.db.update(nativeRunFinalizations).set({
|
||||
recoveryHistory: sql`jsonb_set(${nativeRunFinalizations.recoveryHistory},
|
||||
array[(jsonb_array_length(${nativeRunFinalizations.recoveryHistory}) - 1)::text, 'progressAttempt'], '3'::jsonb)`,
|
||||
}).where(eq(nativeRunFinalizations.runId, fixture.runId));
|
||||
await appendRunEvents(fixture, events);
|
||||
expect(await resumeThenCrash(fixture)).toMatchObject({
|
||||
disposition: { kind: "resume_dead_runner", chargesProviderAttempt: false }, attempt: 3 });
|
||||
});
|
||||
|
||||
it.each([
|
||||
{ label: "turn.accepted", eventType: "turn.accepted" },
|
||||
{ label: "the provider's turn.started", eventType: "turn.started" },
|
||||
])("charges a resume whose provider accepted a turn and made no progress ($label)", async ({ eventType }) => {
|
||||
const fixture = await seedInterruptedCheckpoint(`CHARGED-${eventType}`);
|
||||
expect(await resumeThenCrash(fixture)).toMatchObject({ attempt: 2 });
|
||||
await appendRunEvents(fixture, [{ eventType: "native.process_start_requested" }, { eventType, turnId: "accepted" },
|
||||
{ eventType: "item.completed", turnId: "accepted", kind: "usage" }]);
|
||||
expect(await resumeThenCrash(fixture)).toMatchObject({
|
||||
disposition: { kind: "resume_dead_runner", chargesProviderAttempt: true }, attempt: 3 });
|
||||
});
|
||||
|
||||
it("does not charge the first restart when the original attempt already made progress", async () => {
|
||||
const fixture = await seedInterruptedCheckpoint("ORIGINAL-PROGRESS");
|
||||
await appendRunEvents(fixture, [{ eventType: "tool.execution.completed", turnId: "interrupted-turn" }]);
|
||||
expect(await resumeThenCrash(fixture)).toMatchObject({
|
||||
disposition: { kind: "resume_dead_runner", chargesProviderAttempt: false }, attempt: 1 });
|
||||
});
|
||||
|
||||
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 }> = [];
|
||||
@@ -3019,6 +3095,39 @@ process.on("SIGTERM", () => child.kill("SIGTERM"));
|
||||
expect(disposition).toMatchObject({ kind: "blocked", reason: "execution_recovery_budget_exhausted" });
|
||||
});
|
||||
|
||||
// The executor records each attempt's provider attempt on its start.
|
||||
const processStart = (providerAttempt: number) =>
|
||||
({ eventType: "native.process_start_requested", payload: { hostname: "host", providerAttempt } });
|
||||
|
||||
it("charges an adopted reattachment whose runner died without progress, even after it recorded a start", async () => {
|
||||
const fixture = await seedInterruptedCheckpoint("REATTACH-START-NO-PROGRESS");
|
||||
await resumeThenCrash(fixture);
|
||||
await appendRunEvents(fixture, [...restoreNoise(1), { eventType: "turn.failed", turnId: "continuation-1" }]);
|
||||
expect(await resumeThenCrash(fixture)).toMatchObject({ attempt: 3 });
|
||||
await fixture.db.update(nativeRunFinalizations).set({
|
||||
recoveryHistory: sql`jsonb_set(${nativeRunFinalizations.recoveryHistory},
|
||||
array[(jsonb_array_length(${nativeRunFinalizations.recoveryHistory}) - 1)::text, 'disposition'],
|
||||
'"reattach_existing_runner"'::jsonb)`,
|
||||
}).where(eq(nativeRunFinalizations.runId, fixture.runId));
|
||||
// The reattachment's own admission records a start for the same attempt.
|
||||
await appendRunEvents(fixture, [processStart(3)]);
|
||||
const { disposition } = await resumeThenCrash(fixture);
|
||||
expect(disposition).toMatchObject({ kind: "blocked", reason: "execution_recovery_budget_exhausted" });
|
||||
});
|
||||
|
||||
it("does not charge a normal retry whose executor died after saving its attempt but before its start", async () => {
|
||||
const fixture = await seedInterruptedCheckpoint("RETRY-SAVED-UNSTARTED");
|
||||
expect(await resumeThenCrash(fixture)).toMatchObject({ attempt: 2 });
|
||||
// The resume (attempt 2) starts, its provider accepts a turn, and it
|
||||
// makes no progress. A normal retry then saves attempt 3 and the
|
||||
// controller dies before that retry records its start.
|
||||
await appendRunEvents(fixture, [processStart(2), { eventType: "turn.accepted", turnId: "resumed" }]);
|
||||
await fixture.db.update(nativeRunFinalizations).set({ attempt: 3 })
|
||||
.where(eq(nativeRunFinalizations.runId, fixture.runId));
|
||||
expect(await resumeThenCrash(fixture)).toMatchObject({
|
||||
disposition: { kind: "resume_dead_runner", chargesProviderAttempt: false }, attempt: 3 });
|
||||
});
|
||||
|
||||
it("charges the owed attempt once when an unstarted first resume is replaced", async () => {
|
||||
const fixture = await seedInterruptedCheckpoint("UNSTARTED-FIRST-RESUME");
|
||||
// The first restart charges, but the controller dies before the executor
|
||||
|
||||
@@ -10929,7 +10929,9 @@ async function executePaperclipNativeSessionWithinScope(
|
||||
level: "info",
|
||||
message: "Native execution requested; prior local stop evidence no longer applies.",
|
||||
// Binds a later identity-less stop proof to the host that launched.
|
||||
payload: { hostname: hostname() },
|
||||
// The provider attempt opens that attempt's outcome window for
|
||||
// the work-based budget (provider-attempt-outcome.ts).
|
||||
payload: { hostname: hostname(), providerAttempt: attempt },
|
||||
});
|
||||
crashPoint("native.process_start.after_requested_before_spawn");
|
||||
},
|
||||
|
||||
@@ -0,0 +1,121 @@
|
||||
import { and, desc, eq, gt, inArray, sql } from "drizzle-orm";
|
||||
import type { Db } from "@paperclipai/db";
|
||||
import { heartbeatRunEvents } from "@paperclipai/db";
|
||||
import { PROCESS_START_REQUESTED } from "../native-local-process-stop.js";
|
||||
|
||||
/**
|
||||
* Work-based attempt budgets (durable-ownership plan, section F).
|
||||
*
|
||||
* An attempt spends execution budget only when a recorded outcome proves that
|
||||
* the provider accepted a turn, and that attempt then made no progress before
|
||||
* it ended. An intent alone is never that proof:
|
||||
*
|
||||
* - `provider_not_started`: no accepted turn is recorded. This covers a
|
||||
* startup still marked `beforeSpawn`, a runner that spawned but was never
|
||||
* admitted, and a `turn.start` the provider did not accept. Not charged.
|
||||
* - `provider_started_without_progress`: the provider accepted a turn
|
||||
* (`turn.accepted`, or the provider's own `turn.started`), and no accepted
|
||||
* turn completed work afterwards. Charged.
|
||||
* - `provider_progressed`: an accepted turn completed work: a tool execution or
|
||||
* a completed item other than a usage report (#19720). Not charged.
|
||||
*
|
||||
* Restore activity is not progress: lineage items carry no turn, and work on a
|
||||
* turn accepted before the window (the interrupted turn) does not count unless
|
||||
* the attempt adopted that live provider work.
|
||||
*/
|
||||
export type ProviderAttemptOutcome =
|
||||
| "provider_not_started"
|
||||
| "provider_started_without_progress"
|
||||
| "provider_progressed";
|
||||
|
||||
export const PROVIDER_TURN_ACCEPTED_EVENT_TYPES = ["turn.accepted", "turn.started"] as const;
|
||||
export const PROVIDER_PROGRESS_EVENT_TYPES = ["tool.execution.completed", "item.completed"] as const;
|
||||
|
||||
/**
|
||||
* Uncharged restarts are still bounded: from this controller generation on
|
||||
* (native runs), or after this many uncharged preparation retries (legacy
|
||||
* runs), every restart charges.
|
||||
*/
|
||||
export const PROVIDER_ATTEMPT_CREDIT_GENERATION_LIMIT = 12;
|
||||
|
||||
/** The one charge rule. Every attempt-budget reader should use it. */
|
||||
export function providerAttemptChargesBudget(outcome: ProviderAttemptOutcome): boolean {
|
||||
return outcome === "provider_started_without_progress";
|
||||
}
|
||||
|
||||
/**
|
||||
* Where the outcome window of `providerAttempt` starts, after `sinceSeq`. The
|
||||
* executor records `native.process_start_requested` with its provider attempt
|
||||
* before any backend may spawn a runner, so a later normal retry starts a new
|
||||
* window and earlier work never counts for it.
|
||||
* - `{ seq }`: the latest start of that attempt (a reattachment writes one too).
|
||||
* - `"unstarted"`: the latest start names an earlier attempt. The executor
|
||||
* saved the new attempt but died before its start, so nothing in the event
|
||||
* log belongs to it.
|
||||
* - Starts written before they carried an attempt fall back to the latest start.
|
||||
*/
|
||||
export async function providerAttemptWindowStart(
|
||||
tx: Pick<Db, "select">,
|
||||
input: { runId: string; sinceSeq: number; providerAttempt: number },
|
||||
): Promise<{ seq: number } | "unstarted"> {
|
||||
const [start] = await tx
|
||||
.select({
|
||||
seq: heartbeatRunEvents.seq,
|
||||
attempt: sql<unknown>`${heartbeatRunEvents.payload} -> 'providerAttempt'`,
|
||||
})
|
||||
.from(heartbeatRunEvents)
|
||||
.where(and(
|
||||
eq(heartbeatRunEvents.runId, input.runId),
|
||||
gt(heartbeatRunEvents.seq, input.sinceSeq),
|
||||
eq(heartbeatRunEvents.eventType, PROCESS_START_REQUESTED),
|
||||
))
|
||||
.orderBy(desc(heartbeatRunEvents.seq))
|
||||
.limit(1);
|
||||
if (!start) return { seq: input.sinceSeq };
|
||||
const attempt = typeof start.attempt === "number" ? start.attempt : Number(start.attempt);
|
||||
if (start.attempt !== null && start.attempt !== undefined && Number.isSafeInteger(attempt)
|
||||
&& attempt < input.providerAttempt) return "unstarted";
|
||||
return { seq: Number(start.seq) };
|
||||
}
|
||||
|
||||
/**
|
||||
* Reads the outcome of the attempt recorded after `sinceSeq`.
|
||||
* `adoptedProviderWork` is set when the attempt adopted a live runner whose
|
||||
* provider had already started (a reattachment): it counts as started, and
|
||||
* work on a turn accepted before the window counts as its progress.
|
||||
*/
|
||||
export async function readProviderAttemptOutcome(
|
||||
tx: Pick<Db, "select">,
|
||||
input: { runId: string; sinceSeq: number; adoptedProviderWork?: boolean },
|
||||
): Promise<ProviderAttemptOutcome> {
|
||||
const workTurnId = sql`${heartbeatRunEvents.payload} -> 'prpEvent' ->> 'turnId'`;
|
||||
const acceptedFrom = input.adoptedProviderWork ? 0 : input.sinceSeq;
|
||||
const [work] = await tx
|
||||
.select({ seq: heartbeatRunEvents.seq })
|
||||
.from(heartbeatRunEvents)
|
||||
.where(and(
|
||||
eq(heartbeatRunEvents.runId, input.runId),
|
||||
gt(heartbeatRunEvents.seq, input.sinceSeq),
|
||||
inArray(heartbeatRunEvents.eventType, [...PROVIDER_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 > ${acceptedFrom} and accepted_turn.seq < ${heartbeatRunEvents.seq}
|
||||
and accepted_turn.event_type in (${sql.join(PROVIDER_TURN_ACCEPTED_EVENT_TYPES.map(type => sql`${type}`), sql`, `)})
|
||||
and accepted_turn.payload -> 'prpEvent' ->> 'turnId' = ${workTurnId})`,
|
||||
))
|
||||
.limit(1);
|
||||
if (work) return "provider_progressed";
|
||||
if (input.adoptedProviderWork) return "provider_started_without_progress";
|
||||
const [accepted] = await tx
|
||||
.select({ seq: heartbeatRunEvents.seq })
|
||||
.from(heartbeatRunEvents)
|
||||
.where(and(
|
||||
eq(heartbeatRunEvents.runId, input.runId),
|
||||
gt(heartbeatRunEvents.seq, input.sinceSeq),
|
||||
inArray(heartbeatRunEvents.eventType, [...PROVIDER_TURN_ACCEPTED_EVENT_TYPES]),
|
||||
))
|
||||
.limit(1);
|
||||
return accepted ? "provider_started_without_progress" : "provider_not_started";
|
||||
}
|
||||
@@ -350,8 +350,10 @@ describe("automatic retained workspace recovery", () => {
|
||||
const successors = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId),
|
||||
eq(heartbeatRuns.status, "scheduled_retry")));
|
||||
expect(successors).toHaveLength(1);
|
||||
expect(successors[0]).toMatchObject({ retryOfRunId: withheld.id, scheduledRetryAttempt: 2,
|
||||
scheduledRetryReason: "interrupted_preparation", contextSnapshot: { executionRetryAccounting: { failureRetries: 2 } } });
|
||||
// Interrupted preparation spent no provider work: it keeps the withheld
|
||||
// retry's failure count and starts its own lane (section F).
|
||||
expect(successors[0]).toMatchObject({ retryOfRunId: withheld.id, scheduledRetryAttempt: 1,
|
||||
scheduledRetryReason: "interrupted_preparation", contextSnapshot: { executionRetryAccounting: { failureRetries: 1 } } });
|
||||
expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toBeNull();
|
||||
expect((await run(f.runId)).status).toBe("failed");
|
||||
expect(await run(withheld.id)).toMatchObject({ status: "cancelled", startedAt: null });
|
||||
@@ -461,7 +463,8 @@ describe("automatic retained workspace recovery", () => {
|
||||
}
|
||||
const successors = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, withheld.id));
|
||||
expect(successors).toHaveLength(1);
|
||||
expect(successors[0]).toMatchObject({ status: "scheduled_retry", scheduledRetryAttempt: 2 });
|
||||
expect(successors[0]).toMatchObject({ status: "scheduled_retry", scheduledRetryAttempt: 1,
|
||||
scheduledRetryReason: "interrupted_preparation" });
|
||||
expect((await action(f.actionId)).evidence).toMatchObject({ bootstrapPreparationContinuation: "delivered",
|
||||
bootstrapPreparationContinuationRunId: successors[0].id });
|
||||
expect(await nextWorkspaceRecoveryAt(restarted)).toBeNull();
|
||||
@@ -489,7 +492,7 @@ describe("automatic retained workspace recovery", () => {
|
||||
await Promise.all(await reconcileRetainedWorkspaces(db, f.environmentRuntime, onRecovered, true, undefined, passiveSweep) ?? []);
|
||||
expect(attempts).toBe(2);
|
||||
expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, withheld.id)))
|
||||
.toEqual([expect.objectContaining({ status: "scheduled_retry", scheduledRetryAttempt: 2 })]);
|
||||
.toEqual([expect.objectContaining({ status: "scheduled_retry", scheduledRetryAttempt: 1 })]);
|
||||
expect(await nextWorkspaceRecoveryAt(db)).toBeNull();
|
||||
}, 30_000);
|
||||
|
||||
|
||||
Reference in new issue
Block a user