From f2715e02bb399bfbb97dc2fbb448cab2bb50686d Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Tue, 6 Oct 2026 10:59:18 -0500 Subject: [PATCH] fix(native): surface model capacity errors and retry automatically (#15347) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Native runs preserve provider events and results, then decide task state. > - Codex can fail a turn because the selected model is temporarily at capacity. > - Paperclip displayed a generic native-session failure and left the task in recovery. > - A known capacity failure needs a clear message and a durable delayed retry. > - This pull request uses committed terminal evidence, the status-effect ledger, and existing dispatch gates. > - The task can continue automatically without an unbounded retry loop or a model switch. ## Linked Issues or Issue Description Related: #13993 adds launcher capacity deferrals before provider startup. This change handles a committed native Codex `serverOverloaded` terminal after provider work has started. **What happened?** A native Codex turn ended with `codexErrorInfo: serverOverloaded` and “Selected model is at capacity. Please try a different model.” Paperclip saved the result, showed a generic failure, and required recovery instead of waiting and retrying. **Expected behavior** Show the capacity error directly. Schedule bounded retries after a short delay. Preserve saved work and honor current execution gates. **Steps to reproduce** 1. Run a native Codex task. 2. Commit a `turn.failed` event with `serverOverloaded`, then accept a failed result for the same turn. 3. Inspect the task error and its recovery state. The regression test reproduces this without a paid provider call. **Paperclip version or commit** Reproduced against master `16b7db35ffa0f9a95913c8cbdeea3d595435691f`. ## What Changed - Classify capacity failures from committed runner events and the pinned execution identity. Preserve accepted results and display the specific capacity error. - Atomically persist one scheduled successor with the status decision. Retry after one minute, then two minutes. Share the existing failure budget and stop after two automatic retries. - Reuse dispatch gates for ownership, task holds, dependencies, budget, and locks. Wait for predecessor execution, finalization, and cleanup before claiming a retry. - Preserve pending reviewer authority and suppress retries after reassignment or a successor claim. - Consume the failed run's resume receipt and delivered wake input. Rebuild ordinary continuation from the failed run so explicitly resumed tasks can retry without borrowing one-run authorization. - Label scheduled retries “Model at capacity.” Add a Storybook example and avoid duplicate punctuation in the existing retry card. - Add regression coverage and document the runtime contract. ## Verification - Targeted recovery and UI suites: 125 tests passed. Additional final cleanup and lock regressions passed. - Latest-head continuation and authorization regressions: 57 tests passed, including all four capacity integration tests against an isolated PostgreSQL database. - Rechecked all four capacity integration tests with the full runner's isolated `PAPERCLIP_HOME`, config, temporary directory, and serial fork settings: passed. - `pnpm -r typecheck`: passed. Final server typecheck passed. - `pnpm build`: passed. - `pnpm check:token-gates`: passed. - Browser: checked the real Storybook card. It shows the capacity message, automatic retry time, and existing Retry now action. - `pnpm test:run`: attempted and restarted after an interruption. The resumed run started before the final review correction and was stopped after recorded workspace/native test failures and the five-minute Git streaming timeout already documented on master. Exit 130; no complete local full-suite pass is claimed. Current-head isolated capacity tests and the complete CI suite pass. - A combined local recovery-suite attempt also hit PostgreSQL initialization failures at this macOS host's global shared-memory limit (32 slots). The focused recovery suites passed separately; final-head capacity tests also pass with the full runner's isolated environment settings. - Latest-head GitHub checks: all 56 checks green, including general and serialized test shards, browser E2E, typecheck, build, Runner verification, and canary dry run. No merge conflicts. - Greptile: 5/5 on `813a470b8c18c05aeb7e31e63e573c3a3a2f5cac`, with zero unresolved findings after fixing the resumed-task continuation issue. ## Risks - Capacity retries can repeat a task turn after partial work. They start a fresh provider session with task history and wait for predecessor cleanup. They do not resume the failed turn. - Retries retain the configured model unless an operator changes configuration. Persistent overload consumes the existing failure budget and then needs an explicit retry or model change. - Only run-bound native Codex `serverOverloaded` failures qualify. Usage-limit exhaustion, model/account incompatibility, unbound text, and unknown provider failures retain their current recovery behavior. - No schema migration is required. ## Model Used - OpenAI GPT-6 through Codex. The exact model ID and context-window size are not exposed to this session. Used reasoning, repository tools, code execution, and browser verification. No subagents. ## 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 - [ ] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip --- doc/DEVELOPING.md | 4 + doc/SPEC-implementation.md | 10 + doc/execution-semantics.md | 22 +++ .../heartbeat-process-recovery.test.ts | 4 +- .../run-dispatch/adapters/postgres.test.ts | 15 +- .../modules/run-dispatch/adapters/postgres.ts | 2 +- server/src/services/heartbeat.ts | 26 ++- .../native-provider-capacity-retry.ts | 69 +++++++ .../native-provider-failure-evidence.ts | 11 +- .../native-provider-failure.test.ts | 15 +- .../native-runtime/native-provider-failure.ts | 27 ++- ...tive-provider-overload.integration.test.ts | 178 ++++++++++++++++++ .../native-runtime/native-run-finalizer.ts | 26 ++- .../native-runtime/status-arbiter.test.ts | 21 ++- .../services/native-runtime/status-arbiter.ts | 41 +++- .../status-decision-committer.ts | 18 +- .../IssueScheduledRetryCard.test.tsx | 9 + ui/src/components/IssueScheduledRetryCard.tsx | 2 +- ui/src/lib/runRetryState.test.ts | 1 + ui/src/lib/runRetryState.ts | 1 + .../stories/model-capacity-retry.stories.tsx | 19 ++ 21 files changed, 481 insertions(+), 40 deletions(-) create mode 100644 server/src/services/native-runtime/native-provider-capacity-retry.ts create mode 100644 server/src/services/native-runtime/native-provider-overload.integration.test.ts create mode 100644 ui/storybook/stories/model-capacity-retry.stories.tsx diff --git a/doc/DEVELOPING.md b/doc/DEVELOPING.md index e1927b4296..f6d8a0edaa 100644 --- a/doc/DEVELOPING.md +++ b/doc/DEVELOPING.md @@ -197,6 +197,10 @@ the task shows **Model unavailable**, the provider's account restriction, and guidance to choose a supported model or clear the task's model override before retrying. The run retains this reason even when the runner saves a generic failure result. +When a committed Codex terminal reports `serverOverloaded`, the task displays +the model capacity error and **Model at capacity** on its scheduled retry card. +Automatic retries wait one and two minutes, then stop. Use +**Tasks → Model capacity retry** in Storybook to inspect the waiting state. Claude Code uses model-specific effort levels; Haiku has no effort slider. Grok uses its adapter's reasoning levels, including for its default model. Kimi shows effort only when its agent uses the CLI engine, including with its diff --git a/doc/SPEC-implementation.md b/doc/SPEC-implementation.md index 7e0a3b8ea2..7c0101bea4 100644 --- a/doc/SPEC-implementation.md +++ b/doc/SPEC-implementation.md @@ -1934,3 +1934,13 @@ per-turn snapshot participates in session compatibility, so subsequent turns remove stale instructions after edits or access revocation. See [Connection instructions](connections/CONNECTION-INSTRUCTIONS.md) for contracts, UI conventions, custom adapter integration, and initial memory templates. + +### Native provider capacity retry + +Committed, run-bound Codex `serverOverloaded` terminal failures display the model +capacity error directly and schedule at most two automatic retries, after one +and two minutes. Retries share the execution failure budget, retain task history, +and honor current ownership, review, governance, pause, dependency, budget, and +cleanup gates. Restart or duplicate finalization must not create another +successor. Permanent model/auth incompatibility and usage-limit exhaustion retain +their existing operator recovery requirements. diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 701852da70..6377a142b8 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -469,6 +469,28 @@ Warm sandbox reuse must match the current host Git tip and branch as well as the file snapshot and saved stamp, including managed nested repositories. A history or branch mismatch restages the host before the next run begins. +### Native provider model capacity + +A committed Codex `turn.failed` event with `codexErrorInfo: serverOverloaded`, +bound to the failed terminal turn and pinned execution identity, surfaces +“Selected model is at capacity. Please try a different model.” directly. +Paperclip preserves the accepted result and task history, then atomically records +a durable automatic retry with its status decision. The first retry waits one +minute; the second waits two minutes. Both spend the existing failure-retry +budget. Exhaustion requires an explicit retry or a model change. + +Retries use a fresh provider session and the normal task context, without an +automatic model switch. Consumed wake input and continuation receipts stay on +the failed run; `retryOfRunId` supplies task history without lending the new run +its predecessor's resume authority. Finalization replay and restart reuse the +same successor. +Scheduling preserves pending review authority and respects task holds; promotion, +claim, and dispatch recheck ownership, dependencies, governance, pause, budget, +and execution locks. Claim also waits for the predecessor's provider execution, +workspace finalization, and environment cleanup to settle. Model incompatibility, +usage-limit exhaustion, unknown failures, and unbound diagnostic text do not +qualify as capacity failures. + ### Workspace scan failures before provider startup Repository discovery distinguishes an ordinary folder from a failed Git read. diff --git a/server/src/__tests__/heartbeat-process-recovery.test.ts b/server/src/__tests__/heartbeat-process-recovery.test.ts index 88c1a64cfe..e936e13c57 100644 --- a/server/src/__tests__/heartbeat-process-recovery.test.ts +++ b/server/src/__tests__/heartbeat-process-recovery.test.ts @@ -14310,7 +14310,7 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { await commitNativeStatusDecision({ db, companyId: fixture.companyId, issueId: fixture.issueId, runId: fixture.runId, assessmentId: coordinator!.assessmentId!, priorStatus: issue!.status, priorStatusVersion: issue!.statusVersion, priorDecisionId: issue!.lastStatusDecisionId, - decision: { policyVersion: "phase6-v8", statusAction: "in_progress", toStatus: "in_progress", + decision: { policyVersion: "phase6-v9", statusAction: "in_progress", toStatus: "in_progress", reasonCode: "board_response_waiting", unblockDescriptor: null, effects: [] }, requireBoardResponseWaitSource: source!.source, requireBoardResponseWaitOrigin: origin!, }); @@ -14934,7 +14934,7 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { priorStatusVersion: issue!.statusVersion, priorDecisionId: issue!.lastStatusDecisionId, decision: { - policyVersion: "phase6-v8", + policyVersion: "phase6-v9", statusAction: reasonCode === "prior_status_preserved_no_live_path" ? "preserve" : "in_progress", toStatus: "in_progress", reasonCode, diff --git a/server/src/modules/run-dispatch/adapters/postgres.test.ts b/server/src/modules/run-dispatch/adapters/postgres.test.ts index 227cb78dcc..812eeb4352 100644 --- a/server/src/modules/run-dispatch/adapters/postgres.test.ts +++ b/server/src/modules/run-dispatch/adapters/postgres.test.ts @@ -191,16 +191,19 @@ describeEmbeddedPostgres("run-dispatch postgres adapter", () => { }); } - it.each(["executionRunId", "checkoutRunId"] as const)("suppresses delayed native replacement after another run acquires %s", async (lock) => { + it.each([ + ["executionRunId", "native_safe_replacement"], ["checkoutRunId", "native_safe_replacement"], + ["executionRunId", "native_provider_overloaded"], ["checkoutRunId", "native_provider_overloaded"], + ] as const)("suppresses delayed retry after another run acquires %s (%s)", async (lock, retryReason) => { const { companyId, agentId } = await seedCompanyAndAgent(); const issueId = randomUUID(); await seedIssue({ companyId, issueId, assigneeAgentId: agentId, status: "in_progress" }); - const contextSnapshot = { issueId, wakeReason: "native_safe_replacement", retryReason: "native_safe_replacement", forceFreshSession: true }; + const contextSnapshot = { issueId, wakeReason: retryReason, retryReason, forceFreshSession: true }; const replacementId = await seedRun({ companyId, agentId, status: "scheduled_retry", contextSnapshot }); const competingId = await seedRun({ companyId, agentId, status: "running", contextSnapshot: { issueId } }); await db.update(issues).set({ [lock]: competingId }).where(eq(issues.id, issueId)); const adapter = createPostgresRunDispatchAdapter(db); - expect(await adapter.evaluateScheduledRetryGate({ companyId, runId: replacementId, retryReasonOverride: "native_safe_replacement", now: new Date() })) + expect(await adapter.evaluateScheduledRetryGate({ companyId, runId: replacementId, retryReasonOverride: retryReason, now: new Date() })) .toMatchObject({ allowed: false, errorCode: "issue_execution_lock_changed" }); await db.update(heartbeatRuns).set({ status: "queued" }).where(eq(heartbeatRuns.id, replacementId)); expect(await adapter.cancelStaleQueuedRun({ companyId, runId: replacementId, expectedStatus: "queued", now: new Date() })) @@ -218,15 +221,15 @@ describeEmbeddedPostgres("run-dispatch postgres adapter", () => { expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, competingId)))[0]?.status).toBe("running"); }); - it("does not dispatch a replacement when the task becomes blocked after scheduling", async () => { + it.each(["native_safe_replacement", "native_provider_overloaded"])("does not dispatch %s when the task becomes blocked after scheduling", async retryReason => { const { companyId, agentId } = await seedCompanyAndAgent(); const issueId = randomUUID(); await seedIssue({ companyId, issueId, assigneeAgentId: agentId, status: "in_progress" }); - const contextSnapshot = { issueId, wakeReason: "native_safe_replacement", retryReason: "native_safe_replacement", forceFreshSession: true }; + const contextSnapshot = { issueId, wakeReason: retryReason, retryReason: retryReason, forceFreshSession: true }; const replacementId = await seedRun({ companyId, agentId, status: "scheduled_retry", contextSnapshot }); await db.update(issues).set({ status: "blocked" }).where(eq(issues.id, issueId)); const adapter = createPostgresRunDispatchAdapter(db); - expect(await adapter.evaluateScheduledRetryGate({ companyId, runId: replacementId, retryReasonOverride: "native_safe_replacement", now: new Date() })) + expect(await adapter.evaluateScheduledRetryGate({ companyId, runId: replacementId, retryReasonOverride: retryReason, now: new Date() })) .toMatchObject({ allowed: false, errorCode: "issue_blocked" }); await db.update(heartbeatRuns).set({ status: "queued" }).where(eq(heartbeatRuns.id, replacementId)); expect(await adapter.cancelStaleQueuedRun({ companyId, runId: replacementId, expectedStatus: "queued", now: new Date() })) diff --git a/server/src/modules/run-dispatch/adapters/postgres.ts b/server/src/modules/run-dispatch/adapters/postgres.ts index adaf1a97ce..acd38a9819 100644 --- a/server/src/modules/run-dispatch/adapters/postgres.ts +++ b/server/src/modules/run-dispatch/adapters/postgres.ts @@ -128,7 +128,7 @@ function classifyRetryReasonKind(retryReason: string | null): RetryReasonKind { if (retryReason === MAX_TURN_CONTINUATION_RETRY_REASON) return "max_turn_continuation"; if (retryReason === ISSUE_DISPOSITION_REPAIR_RETRY_REASON) return "disposition_repair"; if ((retryReason === "ai_connection_busy" || retryReason === "ai_connection_pool_wait")) return "ai_connection_wait"; - if (retryReason === "native_safe_replacement") return "native_safe_replacement"; + if (retryReason === "native_safe_replacement" || retryReason === "native_provider_overloaded") return "native_safe_replacement"; return "other"; } diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 9e8617e2fc..8d51256000 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -17578,26 +17578,32 @@ export function heartbeatService( )).for("update") : []; const ownsIssue = owner?.assigneeAgentId === run.agentId && context.wakeReason !== "source_scoped_recovery_action"; - if (ownsIssue && run.scheduledRetryReason === "native_safe_replacement" && + if (ownsIssue && ["native_safe_replacement", "native_provider_overloaded"].includes(run.scheduledRetryReason ?? "") && owner.checkoutRunId && owner.checkoutRunId !== run.id) { return { ownsIssue, blocked: true }; } - if (ownsIssue && owner.executionRunId && owner.executionRunId !== run.id) { + if (run.scheduledRetryReason === "native_provider_overloaded" && owner?.executionRunId && + owner.executionRunId !== run.id && owner.executionRunId !== run.retryOfRunId) { + return { ownsIssue, blocked: true }; + } + const previousRunId = run.scheduledRetryReason === "native_provider_overloaded" + ? run.retryOfRunId : ownsIssue ? owner?.executionRunId : null; + if (previousRunId && previousRunId !== run.id) { const [previous] = await tx.select({ status: heartbeatRuns.status }) .from(heartbeatRuns).where(and( - eq(heartbeatRuns.id, owner.executionRunId), + eq(heartbeatRuns.id, previousRunId), eq(heartbeatRuns.companyId, run.companyId), )); // A terminal result can precede workspace/lease cleanup on this or // another controller. Local absence alone is not a release receipt. if (!isHeartbeatRunTerminalStatus(previous?.status) || - liveRunExecutions.has(owner.executionRunId)) { + liveRunExecutions.has(previousRunId)) { return { ownsIssue, blocked: true }; } const [pendingLease] = await tx.select({ id: environmentLeases.id }) .from(environmentLeases).where(and( eq(environmentLeases.companyId, run.companyId), - eq(environmentLeases.heartbeatRunId, owner.executionRunId), + eq(environmentLeases.heartbeatRunId, previousRunId), or(and(isNull(environmentLeases.releasedAt), // Warm release deliberately retains the sandbox. Its successful // receipt settles the old run without destroying the resource. @@ -17611,7 +17617,7 @@ export function heartbeatService( phase: nativeRunFinalizations.phase, leaseOwner: nativeRunFinalizations.leaseOwner, }).from(nativeRunFinalizations).where(and( eq(nativeRunFinalizations.companyId, run.companyId), - eq(nativeRunFinalizations.runId, owner.executionRunId), + eq(nativeRunFinalizations.runId, previousRunId), )); if (pendingLease || (finalization && (finalization.leaseOwner || !["committed", "applied", "terminal_failure"].includes(finalization.phase)))) { @@ -17961,6 +17967,10 @@ export function heartbeatService( updatedAt: claimedAt, }; if (nativeReviewContext) { + if (run.scheduledRetryReason === "native_provider_overloaded") { + const predecessor = await lockIssueExecutionClaim(tx); + if (predecessor.blocked) return null; + } return claimQueuedNativeReviewRun(tx, { run, claimedAt, claimValues, agentNameKey: normalizeAgentNameKey(agent.name), @@ -18028,7 +18038,7 @@ export function heartbeatService( // Mention/context runs can touch an issue, but only the current assignee // owns the issue execution lock shown as the active run. eq(issues.assigneeAgentId, claimed.agentId), - claimed.scheduledRetryReason === "native_safe_replacement" + ["native_safe_replacement", "native_provider_overloaded"].includes(claimed.scheduledRetryReason ?? "") ? or( isNull(issues.checkoutRunId), eq(issues.checkoutRunId, claimed.id), @@ -22896,7 +22906,7 @@ export function heartbeatService( if ( !issueId || (!isResolvedInteractionContinuationWakeContext(context) && - run.scheduledRetryReason !== "native_safe_replacement") + !["native_safe_replacement", "native_provider_overloaded"].includes(run.scheduledRetryReason ?? "")) ) { return { dispatched: true, resultPromise: dispatch(() => {}) }; } diff --git a/server/src/services/native-runtime/native-provider-capacity-retry.ts b/server/src/services/native-runtime/native-provider-capacity-retry.ts new file mode 100644 index 0000000000..5b54bd0cdf --- /dev/null +++ b/server/src/services/native-runtime/native-provider-capacity-retry.ts @@ -0,0 +1,69 @@ +import { randomUUID } from "node:crypto"; +import { and, eq } from "drizzle-orm"; +import { agentWakeupRequests, heartbeatRuns, type Db } from "@paperclipai/db"; +import { accountingForScheduledRetry, executionFailureRetryCount } from "../execution-recovery-attempt.js"; +import { NATIVE_PROVIDER_CAPACITY_MAX_RETRIES, NATIVE_PROVIDER_OVERLOADED_CODE, NATIVE_PROVIDER_OVERLOADED_MESSAGE } from "./native-provider-failure.js"; + +/** Called under the finalizer's issue/coordinator locks, in the status-effect transaction. */ +export async function materializeNativeProviderCapacityRetry(input: { + tx: Db; + companyId: string; + issueId: string; + runId: string; + agentId: string; +}) { + const [run] = await input.tx.select().from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, input.companyId), eq(heartbeatRuns.id, input.runId), + eq(heartbeatRuns.agentId, input.agentId), eq(heartbeatRuns.nativeIssueId, input.issueId), + )).for("update"); + if (!run || run.runtimeMode !== "native") throw new Error("native_capacity_retry_run_binding_mismatch"); + // Share the predecessor claim with every other failure-retry lane. + const [existing] = await input.tx.select().from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, input.companyId), eq(heartbeatRuns.retryOfRunId, run.id), + )).limit(1); + if (existing) return existing.id; + const attempt = executionFailureRetryCount(run) + 1; + if (attempt > NATIVE_PROVIDER_CAPACITY_MAX_RETRIES) throw new Error("native_capacity_retry_exhausted"); + const now = new Date(); + const dueAt = new Date(now.getTime() + 60_000 * 2 ** (attempt - 1)); + const retryId = randomUUID(); + const predecessorContext = { ...run.contextSnapshot }; + // Resume receipts and delivered wake input belong to the failed run. The + // new turn has ordinary task authority; history comes from retryOfRunId. + // Match the consumed-input boundary used by native safe replacements. + for (const key of [ + "explicitUserContinuation", "wakeCommentId", "wakeCommentIds", "commentId", + "commentIds", "latestCommentId", "resumeIntent", "followUpRequested", + "paperclipWake", "paperclipWakeComment", "paperclipTaskMarkdown", "paperclipTaskMarkdownCompact", + "paperclipTaskMarkdownAssignment", "paperclipTaskMarkdownAssignmentCompact", "paperclipTurnContext", + ]) delete predecessorContext[key]; + const contextSnapshot = { + ...predecessorContext, + issueId: input.issueId, + taskId: input.issueId, + forceFreshSession: true, + retryOfRunId: run.id, + retryReason: NATIVE_PROVIDER_OVERLOADED_CODE, + wakeReason: NATIVE_PROVIDER_OVERLOADED_CODE, + scheduledRetryAttempt: attempt, + scheduledRetryAt: dueAt.toISOString(), + executionRetryAccounting: accountingForScheduledRetry(run, NATIVE_PROVIDER_OVERLOADED_CODE, attempt), + }; + const [wake] = await input.tx.insert(agentWakeupRequests).values({ + companyId: run.companyId, agentId: run.agentId, source: "automation", triggerDetail: "system", + reason: NATIVE_PROVIDER_OVERLOADED_CODE, status: "queued", runId: retryId, + idempotencyKey: `native-capacity:${run.companyId}:${run.id}`, + requestedByActorType: "system", requestedByActorId: "native_finalizer", + payload: contextSnapshot, + }).returning({ id: agentWakeupRequests.id }); + await input.tx.insert(heartbeatRuns).values({ + id: retryId, companyId: run.companyId, agentId: run.agentId, + invocationSource: "automation", triggerDetail: "system", status: "scheduled_retry", + responsibleUserId: run.responsibleUserId, wakeupRequestId: wake.id, + retryOfRunId: run.id, scheduledRetryReason: NATIVE_PROVIDER_OVERLOADED_CODE, + scheduledRetryAttempt: attempt, scheduledRetryAt: dueAt, + error: NATIVE_PROVIDER_OVERLOADED_MESSAGE, errorCode: NATIVE_PROVIDER_OVERLOADED_CODE, + contextSnapshot, + }); + return retryId; +} diff --git a/server/src/services/native-runtime/native-provider-failure-evidence.ts b/server/src/services/native-runtime/native-provider-failure-evidence.ts index e97143aca9..f2c53356d8 100644 --- a/server/src/services/native-runtime/native-provider-failure-evidence.ts +++ b/server/src/services/native-runtime/native-provider-failure-evidence.ts @@ -10,7 +10,7 @@ function record(value: unknown): Record { } /** Re-derive authority from immutable run-bound evidence, including after restart. */ -export async function readPersistedNativeModelRejection( +export async function readPersistedNativeProviderFailure( db: Db, run: typeof heartbeatRuns.$inferSelect, turnId: string | null, @@ -20,7 +20,7 @@ export async function readPersistedNativeModelRejection( terminal.turnTerminalState !== "failed" || terminal.runTerminalState !== "failed") return null; let execution; try { execution = parseNativeExecutionInput(record(run.runnerProfileJson).nativeExecutionInput); } catch { return null; } - if (execution.provider.kind !== "codex" || !execution.provider.model || + if (execution.provider.kind !== "codex" || execution.binding.runId !== run.id || execution.binding.companyId !== run.companyId || execution.binding.agentId !== run.agentId || execution.binding.issueId !== run.nativeIssueId || execution.session.normalizedSessionId !== run.nativeSessionId || @@ -43,3 +43,10 @@ export async function readPersistedNativeModelRejection( observed.observe(parsed.event); return observed.forTerminal(turnId, terminal); } + +export async function readPersistedNativeModelRejection( + ...args: Parameters +) { + const failure = await readPersistedNativeProviderFailure(...args); + return failure?.errorCode === "native_provider_model_rejected" ? failure : null; +} diff --git a/server/src/services/native-runtime/native-provider-failure.test.ts b/server/src/services/native-runtime/native-provider-failure.test.ts index 5db2942e4c..bb4804a985 100644 --- a/server/src/services/native-runtime/native-provider-failure.test.ts +++ b/server/src/services/native-runtime/native-provider-failure.test.ts @@ -1,6 +1,6 @@ import { describe, expect, it } from "vitest"; import type { PrpEvent, PrpTerminalState } from "../../vendor/paperclip-runner/index.js"; -import { createNativeProviderFailureObservation, NATIVE_MODEL_REJECTION_DIAGNOSTIC, NATIVE_MODEL_REJECTION_MESSAGE } from "./native-provider-failure.js"; +import { createNativeProviderFailureObservation, NATIVE_MODEL_REJECTION_DIAGNOSTIC, NATIVE_MODEL_REJECTION_MESSAGE, NATIVE_PROVIDER_OVERLOADED_MESSAGE } from "./native-provider-failure.js"; const model = "test-model"; const response = { @@ -19,6 +19,19 @@ const terminal: PrpTerminalState = { const observation = () => createNativeProviderFailureObservation({ kind: "codex", model }); describe("native provider failure observation", () => { + it("surfaces a structured capacity failure without copying arbitrary provider text", () => { + const observed = createNativeProviderFailureObservation({ kind: "codex" }); + observed.observe({ ...event, payload: { status: "failed", error: { codexErrorInfo: "serverOverloaded", message: "Private arbitrary response" } } }); + expect(observed.forTerminal("turn", terminal)).toEqual({ errorCode: "native_provider_overloaded", errorMessage: NATIVE_PROVIDER_OVERLOADED_MESSAGE, diagnostic: { provider: "codex", category: "server_overloaded" } }); + expect(observed.forTerminal("different-turn", terminal)).toBeNull(); + expect(observed.forTerminal("turn", { ...terminal, runTerminalState: "succeeded" })).toBeNull(); + }); + + it.each(["usageLimitExceeded", "other", "unauthorized", undefined])("does not retry unrelated provider errors (%s)", (codexErrorInfo) => { + const observed = observation(); + observed.observe({ ...event, payload: { status: "failed", error: { codexErrorInfo, message: NATIVE_PROVIDER_OVERLOADED_MESSAGE } } }); + expect(observed.forTerminal("turn", terminal)).toBeNull(); + }); it("binds the structured rejection to the failed turn and emits no response or model text", () => { const observed = observation(); observed.observe(event); diff --git a/server/src/services/native-runtime/native-provider-failure.ts b/server/src/services/native-runtime/native-provider-failure.ts index ec7380ffc8..3fa2e5bbf7 100644 --- a/server/src/services/native-runtime/native-provider-failure.ts +++ b/server/src/services/native-runtime/native-provider-failure.ts @@ -3,6 +3,15 @@ import type { PrpEvent, PrpTerminalState } from "../../vendor/paperclip-runner/i export const NATIVE_MODEL_REJECTION_MESSAGE = "The selected model is not supported by the current ChatGPT connection. Choose a supported model or a compatible AI connection."; +export const NATIVE_PROVIDER_OVERLOADED_MESSAGE = + "Selected model is at capacity. Please try a different model."; +export const NATIVE_PROVIDER_OVERLOADED_CODE = "native_provider_overloaded"; +export const NATIVE_PROVIDER_CAPACITY_MAX_RETRIES = 2; +export const NATIVE_PROVIDER_OVERLOADED_DIAGNOSTIC = { + provider: "codex", + category: "server_overloaded", +} as const; + export const NATIVE_MODEL_REJECTION_DIAGNOSTIC = { provider: "codex", category: "model_auth_incompatible", @@ -29,13 +38,18 @@ export function readNativeModelRejectionDiagnostic(value: unknown) { /** Observe only events accepted by the bound control-plane persistence port. */ export function createNativeProviderFailureObservation(provider: { kind: string; model?: string | null }) { let rejectedTurnId: string | null = null; + let overloadedTurnId: string | null = null; return { observe(event: PrpEvent) { - if (provider.kind !== "codex" || !provider.model || + if (provider.kind !== "codex" || event.sourceKind !== "runner" || event.eventType !== "turn.failed" || !event.turnId) return; const payload = record(event.payload); const failure = record(payload.error); - if (payload.status !== "failed" || payload.recoverable === true || failure.recoverable === true || + if (payload.status === "failed" && failure.codexErrorInfo === "serverOverloaded") { + overloadedTurnId = event.turnId; + return; + } + if (!provider.model || payload.status !== "failed" || payload.recoverable === true || failure.recoverable === true || typeof failure.message !== "string" || failure.message.length > 4_096) return; let response: Record; try { response = record(JSON.parse(failure.message)); } catch { return; } @@ -48,8 +62,13 @@ export function createNativeProviderFailureObservation(provider: { kind: string; rejectedTurnId = event.turnId; }, forTerminal(turnId: string | null | undefined, terminal: PrpTerminalState) { - if (!rejectedTurnId || turnId !== rejectedTurnId || - terminal.turnTerminalState !== "failed" || terminal.runTerminalState !== "failed") return null; + if (!turnId || terminal.turnTerminalState !== "failed" || terminal.runTerminalState !== "failed") return null; + if (turnId === overloadedTurnId) return { + errorCode: NATIVE_PROVIDER_OVERLOADED_CODE, + errorMessage: NATIVE_PROVIDER_OVERLOADED_MESSAGE, + diagnostic: { ...NATIVE_PROVIDER_OVERLOADED_DIAGNOSTIC }, + }; + if (turnId !== rejectedTurnId) return null; return { errorCode: "native_provider_model_rejected" as const, errorMessage: NATIVE_MODEL_REJECTION_MESSAGE, diff --git a/server/src/services/native-runtime/native-provider-overload.integration.test.ts b/server/src/services/native-runtime/native-provider-overload.integration.test.ts new file mode 100644 index 0000000000..862a7389bd --- /dev/null +++ b/server/src/services/native-runtime/native-provider-overload.integration.test.ts @@ -0,0 +1,178 @@ +import { randomUUID } from "node:crypto"; +import { and, eq } from "drizzle-orm"; +import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; +import { agents, agentWakeupRequests, companies, completionContracts, createDb, environmentLeases, heartbeatRuns, issueComments, issueRecoveryActions, issues, nativeRunFinalizations, statusDecisionEffects, statusDecisions } from "@paperclipai/db"; +import type { ControlPlanePort, NativeExecutionInputV1, PrpEvent, PrpTerminalState } from "@paperclipai/paperclip-runner"; +import { startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; +import { CONTROL_PLANE_CONFORMANCE_RESULT } from "../../vendor/paperclip-runner/testing.js"; + +const provider = vi.hoisted(() => ({ execute: vi.fn() })); +vi.mock("../../vendor/paperclip-runner/index.js", async (importOriginal) => ({ + ...(await importOriginal()), executeNativeSession: provider.execute, +})); +import { executePaperclipNativeSession } from "./native-session-executor.js"; +import { buildNativeCompletionContract } from "./completion-contracts.js"; +import { prepareNativeHeartbeatRun } from "./prepare-native-run.js"; +import { finalizeNativeRun } from "./native-run-finalizer.js"; +import { readPersistedNativeProviderFailure } from "./native-provider-failure-evidence.js"; +import { NATIVE_PROVIDER_OVERLOADED_MESSAGE } from "./native-provider-failure.js"; +import { heartbeatService } from "../heartbeat.js"; +import { createPostgresWakeQueueAdapter } from "../../modules/wake-queue/adapters/postgres.js"; +import { createReleaseIssueExecution } from "../../modules/wake-queue/application/use-cases.js"; +import { buildExecutionContinuation } from "../execution-continuation.js"; + +describe("durable native provider capacity retry", () => { + let temporary: Awaited>; + let db: ReturnType; + const companyId = randomUUID(), agentId = randomUUID(), otherAgentId = randomUUID(); + beforeAll(async () => { + temporary = await startEmbeddedPostgresTestDatabase("native-capacity-retry-"); + db = createDb(temporary.connectionString); + await db.insert(companies).values({ id: companyId, name: "Capacity regression", issuePrefix: "CAP" }); + await db.insert(agents).values([ + { id: agentId, companyId, name: "Capacity worker", status: "active", adapterType: "paperclip_runner", adapterConfig: { provider: "codex", model: "test-model" } }, + { id: otherAgentId, companyId, name: "New owner", status: "active", adapterType: "paperclip_runner" }, + ]); + }); + afterAll(async () => { await temporary?.cleanup(); }); + + async function acceptCapacityFailure(runId: string, issueId: string, mutateOwner = false) { + const [issue] = await db.select().from(issues).where(eq(issues.id, issueId)); + const [run] = await db.update(heartbeatRuns).set({ status: "running", runtimeMode: "native", nativeIssueId: issueId, error: null, errorCode: null }).where(eq(heartbeatRuns.id, runId)).returning(); + await db.update(issues).set({ executionRunId: runId }).where(eq(issues.id, issueId)); + const native = await prepareNativeHeartbeatRun({ db, run, issue, environmentLeaseId: randomUUID() }); + const [contract] = await db.select().from(completionContracts).where(eq(completionContracts.issueId, issueId)); + const contractInput = buildNativeCompletionContract(issue, { revision: contract.revision }); + await db.insert(nativeRunFinalizations).values({ runId, companyId, issueId, phase: "observed" }); + const execution: NativeExecutionInputV1 = { + schema: "paperclip.native-execution-input.v1", binding: { companyId, runId, issueId, agentId, executionWorkspaceId: runId }, + provider: { kind: "codex", model: "test-model" }, + task: { identifier: issueId, title: issue.title, description: null, prompt: "Continue the task using its history.", workMode: "standard" }, + workspace: { cwd: "/tmp", repoUrl: null, repoRef: null, branchName: null }, + session: { normalizedSessionId: native.normalizedSessionId, driverKind: "codex_app_server", protocolVersion: 1, lifecyclePolicy: { mode: "per_turn", idleTimeoutMs: null } }, + completionContract: { id: contract.id, sha256: contract.canonicalSha256, schemaVersion: "paperclip.completion-contract.v1", contract: contractInput }, + interactionResponses: [], credentialBindings: [], + }; + await db.update(heartbeatRuns).set({ runnerProfileJson: { nativeExecutionInput: execution } }).where(eq(heartbeatRuns.id, runId)); + const event: PrpEvent = { + schema: "paperclip.prp.event.v1", runId, normalizedSessionId: native.normalizedSessionId, turnId: native.turnId, + sourceInstanceId: native.runnerInstanceId, sourceEventId: "overload", sourceSeq: 1, sourceKind: "runner", eventType: "turn.failed", schemaVersion: 1, priority: 0, emittedAt: new Date().toISOString(), + payload: { status: "failed", error: { codexErrorInfo: "serverOverloaded", message: NATIVE_PROVIDER_OVERLOADED_MESSAGE } }, + }; + const terminal: PrpTerminalState = { schema: "paperclip.prp.terminal.v1", turnTerminalState: "failed", runTerminalState: "failed", reportedWorkDisposition: "needs_review" }; + const result = { ...CONTROL_PLANE_CONFORMANCE_RESULT, reportedWorkDisposition: "needs_review" as const, + summary: "I started investigating the task.", completionClaim: { ...CONTROL_PLANE_CONFORMANCE_RESULT.completionClaim, contractRevision: contractInput.revision } }; + provider.execute.mockReset().mockImplementation(async (options: { controlPlane: ControlPlanePort }) => { + await options.controlPlane.openRun({ identity: { companyId, issueId, runId, agentId, sessionId: native.normalizedSessionId }, backendKind: "mock", sourceInstanceId: native.runnerInstanceId }); + await options.controlPlane.appendEvent(event); + await options.controlPlane.completeRun({ result, terminal, turnId: native.turnId }); + return { result, terminal, turnId: native.turnId, normalizedSessionId: native.normalizedSessionId, providerSessionId: "fixture-session", driverKind: "codex_app_server", driverVersion: "fixture", nativeEventCount: 1, highestContiguousSourceSeq: 1, usage: null }; + }); + const unexpected = async () => { throw new Error("Unexpected provider call"); }; + const adapter = await executePaperclipNativeSession({ db, execution, runnerInstanceId: native.runnerInstanceId, backend: { descriptor: unexpected, openSession: unexpected } }); + expect(adapter).toMatchObject({ errorCode: "native_provider_overloaded", errorMessage: NATIVE_PROVIDER_OVERLOADED_MESSAGE }); + const [persisted] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)); + expect(persisted.errorCode).toBeNull(); // Restart before adapter metadata is saved. + expect(await readPersistedNativeProviderFailure(db, persisted, native.turnId, terminal)).toMatchObject({ errorCode: "native_provider_overloaded" }); + expect(await readPersistedNativeProviderFailure(db, { ...persisted, runnerProfileJson: null }, native.turnId, terminal)).toBeNull(); + expect(await readPersistedNativeProviderFailure(db, persisted, "different-turn", terminal)).toBeNull(); + if (mutateOwner) await db.update(issues).set({ assigneeAgentId: otherAgentId, executionRunId: null }).where(eq(issues.id, issueId)); + for (let replay = 0; replay < 2; replay++) await finalizeNativeRun({ db, runId, workspaceFinalizeStatus: "succeeded", projectRunStatus: true }); + expect(provider.execute).toHaveBeenCalledTimes(1); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)))[0]).toMatchObject({ status: "failed", error: NATIVE_PROVIDER_OVERLOADED_MESSAGE, errorCode: "native_provider_overloaded" }); + expect(await db.select().from(statusDecisions).where(eq(statusDecisions.runId, runId))).toHaveLength(1); + } + + it("retries after one and two minutes across restart/replay, then stops", async () => { + const issueId = randomUUID(), originalRunId = randomUUID(); + await db.insert(issues).values({ id: issueId, companyId, title: "Capacity task", status: "in_progress", assigneeAgentId: agentId, responsibleUserId: "capacity-owner" }); + await db.insert(heartbeatRuns).values({ id: originalRunId, companyId, agentId, responsibleUserId: "capacity-owner", contextSnapshot: { issueId, wakeCommentId: "original-message" } }); + let runId: string = originalRunId; + const unexpected = vi.fn(async () => { throw new Error("Unexpected recovery"); }); + const release = createReleaseIssueExecution({ + issueLock: createPostgresWakeQueueAdapter(db, { resolveResponsibleUserId: unexpected, getRoutineEnv: unexpected, resolveSessionBeforeForWakeup: unexpected }), + recovery: { escalateStrandedAssignedIssue: unexpected, escalateStrandedRecoveryIssueInPlace: unexpected }, + }); + for (let failure = 0; failure <= 2; failure++) { + const before = Date.now(); + await acceptCapacityFailure(runId, issueId); + const successors = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.retryOfRunId, runId))); + if (failure === 2) { + expect(successors).toHaveLength(0); + expect((await db.select().from(issues).where(eq(issues.id, issueId)))[0]).toMatchObject({ status: "blocked", unblockDescriptor: { action: expect.stringContaining("Automatic retries exhausted") } }); + break; + } + expect(successors).toHaveLength(1); + const retry = successors[0]; + expect(retry).toMatchObject({ status: "scheduled_retry", scheduledRetryAttempt: failure + 1, scheduledRetryReason: "native_provider_overloaded", error: NATIVE_PROVIDER_OVERLOADED_MESSAGE, + contextSnapshot: { issueId, forceFreshSession: true, executionRetryAccounting: { version: 1, failureRetries: failure + 1, maxTurnContinuations: 0 } } }); + expect(retry.contextSnapshot).not.toHaveProperty("wakeCommentId"); + expect(retry.scheduledRetryAt!.getTime() - before).toBeGreaterThanOrEqual(60_000 * 2 ** failure); + expect(retry.scheduledRetryAt!.getTime() - Date.now()).toBeLessThanOrEqual(60_000 * 2 ** failure); + const wakes = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.runId, retry.id)); + expect(wakes).toHaveLength(1); + const [decision] = await db.select().from(statusDecisions).where(eq(statusDecisions.runId, runId)); + expect(await db.select().from(statusDecisionEffects).where(eq(statusDecisionEffects.decisionId, decision.id))).toMatchObject([{ effectKind: "schedule_retry", targetType: "heartbeat_run", targetId: retry.id }]); + expect((await release({ companyId, runId, now: new Date() })).outcome.kind).toBe("released"); + expect(await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, issueId))).toHaveLength(0); + const schedulerAfterRestart = heartbeatService(db, { runtimeEnv: {} }); + expect(await schedulerAfterRestart.promoteDueScheduledRetries(new Date())).toMatchObject({ promoted: 0 }); + expect(await schedulerAfterRestart.promoteDueScheduledRetries(retry.scheduledRetryAt!)).toMatchObject({ promoted: 1, runIds: [retry.id] }); + runId = retry.id; + } + expect(unexpected).not.toHaveBeenCalled(); + }); + + it("uses failed-run history without lending a resumed task's consumed continuation receipt", async () => { + const issueId = randomUUID(), runId = randomUUID(), olderRunId = randomUUID(), commentId = randomUUID(); + await db.insert(issues).values({ id: issueId, companyId, title: "Resumed capacity task", status: "in_progress", assigneeAgentId: agentId, responsibleUserId: "capacity-owner" }); + await db.insert(issueComments).values({ id: commentId, companyId, issueId, authorUserId: "capacity-owner", authorType: "user", body: "Continue the investigation." }); + const consumedContext = { issueId, wakeCommentId: commentId, wakeCommentIds: [commentId], + explicitUserContinuation: { previousRunId: olderRunId, commentId }, + previousRunId: olderRunId, resumeIntent: "continue", paperclipTurnContext: "Previously delivered context" }; + await db.insert(heartbeatRuns).values([ + { id: olderRunId, companyId, agentId, status: "failed", contextSnapshot: { issueId } }, + { id: runId, companyId, agentId, responsibleUserId: "capacity-owner", contextSnapshot: consumedContext }, + ]); + await acceptCapacityFailure(runId, issueId); + const [retry] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, runId)); + expect(retry.contextSnapshot).not.toHaveProperty("explicitUserContinuation"); + expect(retry.contextSnapshot).not.toHaveProperty("wakeCommentId"); + expect(retry.contextSnapshot).not.toHaveProperty("resumeIntent"); + expect(retry.contextSnapshot).not.toHaveProperty("paperclipTurnContext"); + const continuation = await buildExecutionContinuation({ db, companyId, issueId, agentId, + runId: retry.id, context: retry.contextSnapshot!, summary: null, exposeLowTrustRaw: false }); + expect(continuation).toMatchObject({ trigger: { sourceRunId: runId }, objective: "Continue the investigation.", + messages: [expect.objectContaining({ id: commentId, body: "Continue the investigation." })] }); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)))[0].contextSnapshot) + .toEqual(consumedContext); + await db.update(heartbeatRuns).set({ status: "cancelled" }).where(eq(heartbeatRuns.id, retry.id)); + }); + + it("does not claim a capacity retry before predecessor cleanup settles", async () => { + const issueId = randomUUID(), runId = randomUUID(); + await db.insert(issues).values({ id: issueId, companyId, title: "Cleanup capacity task", status: "in_progress", assigneeAgentId: agentId, responsibleUserId: "capacity-owner" }); + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, responsibleUserId: "capacity-owner", contextSnapshot: { issueId } }); + await acceptCapacityFailure(runId, issueId); + const [retry] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, runId)); + await db.update(issues).set({ executionRunId: null, checkoutRunId: null }).where(eq(issues.id, issueId)); + const scheduler = heartbeatService(db, { runtimeEnv: {} }); + expect(await scheduler.promoteDueScheduledRetries(retry.scheduledRetryAt!)).toMatchObject({ promoted: 1 }); + await db.insert(environmentLeases).values({ companyId, heartbeatRunId: runId, provider: "local", status: "pending_cleanup", cleanupStatus: "failed" }); + await scheduler.resumeQueuedRuns(); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, retry.id)))[0].status).toBe("queued"); + expect(provider.execute).toHaveBeenCalledTimes(1); + // Do not leave a runnable fixture behind for the next case. + await db.update(heartbeatRuns).set({ status: "cancelled" }).where(eq(heartbeatRuns.id, retry.id)); + }); + + it("does not retry or change the task after reassignment", async () => { + const issueId = randomUUID(), runId = randomUUID(); + await db.insert(issues).values({ id: issueId, companyId, title: "Reassigned capacity task", status: "in_progress", assigneeAgentId: agentId, responsibleUserId: "capacity-owner" }); + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, responsibleUserId: "capacity-owner", contextSnapshot: { issueId } }); + await acceptCapacityFailure(runId, issueId, true); + expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, runId))).toHaveLength(0); + expect((await db.select().from(issues).where(eq(issues.id, issueId)))[0]).toMatchObject({ status: "in_progress", assigneeAgentId: otherAgentId }); + expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, companyId))).filter(wake => wake.payload?.issueId === issueId)).toHaveLength(0); + }); +}); diff --git a/server/src/services/native-runtime/native-run-finalizer.ts b/server/src/services/native-runtime/native-run-finalizer.ts index fb248551b9..3399b86bef 100644 --- a/server/src/services/native-runtime/native-run-finalizer.ts +++ b/server/src/services/native-runtime/native-run-finalizer.ts @@ -1,5 +1,6 @@ import { settleSlackConversation } from "../slack-conversation-lifecycle.js"; -import { readPersistedNativeModelRejection } from "./native-provider-failure-evidence.js"; +import { executionFailureRetryCount } from "../execution-recovery-attempt.js"; +import { readPersistedNativeProviderFailure } from "./native-provider-failure-evidence.js"; import { dismissAutomaticCompletionReviews } from "./automatic-completion-reviews.js"; import { getNativeReviewAssignment, readNativeReviewAssignmentContext } from "./native-review-participant.js"; import { conversationNativeDecision, isConversation } from "../agent-conversations.js"; @@ -1263,9 +1264,9 @@ export async function finalizeNativeRun(input: { issueId: authoritativeIssue.id, runId: run.id, }), - assessment.reportedDisposition === "yielded" && + (terminalState === "failed" || (assessment.reportedDisposition === "yielded" && assessment.continuation?.kind === "response_wake" && - assessment.hasBlockingRemainingWork + assessment.hasBlockingRemainingWork)) ? issueTreeControlService(input.db).getActivePauseHoldGate( run.companyId, authoritativeIssue.id, ) @@ -1276,15 +1277,18 @@ export async function finalizeNativeRun(input: { companyId: run.companyId, issueId: authoritativeIssue.id, agentId: run.agentId, contextSnapshot: reviewContext, allowResolvedByRunId: run.id, }) : null; - const providerModelRejection = await readPersistedNativeModelRejection( + const providerFailure = await readPersistedNativeProviderFailure( input.db, run, resultRow.turnId, envelope.terminal as PrpTerminalState, ); - const ownsModelRejectionDecision = + const ownsProviderFailureDecision = (!authoritativeIssue.executionRunId || authoritativeIssue.executionRunId === run.id) && (reviewContext ? nativeReview?.interaction.status === "pending" : authoritativeIssue.assigneeAgentId === run.agentId && !authoritativeIssue.assigneeUserId); const proposedDecision = resolveNativeFinalizerStatus({ - providerModelRejected: providerModelRejection !== null && ownsModelRejectionDecision, + providerModelRejected: providerFailure?.errorCode === "native_provider_model_rejected" && ownsProviderFailureDecision, + providerOverloaded: providerFailure?.errorCode === "native_provider_overloaded" && ownsProviderFailureDecision, + providerFailureSuperseded: providerFailure?.errorCode === "native_provider_overloaded" && !ownsProviderFailureDecision, + failureRetryCount: executionFailureRetryCount(run), ...(reviewContext ? { nativeReviewOutcome: nativeReview ? nativeReview.interaction.status === "pending" ? "pending" as const : "resolved" as const : "stale" as const } : {}), @@ -1370,6 +1374,8 @@ export async function finalizeNativeRun(input: { requireModelRejectionOwner: decision.reasonCode === "native_provider_model_rejected" ? { agentId: run.agentId, reviewContext } : undefined, + requireProviderFailureOwner: decision.reasonCode?.startsWith("native_provider_overloaded") + ? { agentId: run.agentId, reviewContext } : undefined, requireBoardResponseWaitSource: decision.reasonCode === "board_response_waiting" || repairBoardResponseWait ? boardResponseWait?.source @@ -1412,7 +1418,7 @@ export async function finalizeNativeRun(input: { ); const clearExecutionFailure = input.projectRunStatus && finalizationPhase === "committed" && terminalState === "succeeded"; const finalizationMetadata = { - ...(providerModelRejection ? { nativeProviderFailure: providerModelRejection.diagnostic } : {}), + ...(providerFailure ? { nativeProviderFailure: providerFailure.diagnostic } : {}), finalizationPhase, ...(finalizationPhase === "committed" ? { failureCode: null, originalFailureCode: null, nextAttemptAt: null } : {}), assessmentId: assessmentRow.id, @@ -1463,11 +1469,11 @@ export async function finalizeNativeRun(input: { nativePhase: finalizationPhase, nativePhaseUpdatedAt: now, ...(clearExecutionFailure ? { error: null, errorCode: null } : {}), - ...(providerModelRejection ? { + ...(providerFailure ? { // Recovery can finalize before heartbeat saves the adapter result. // Retain any independently recorded execution/cleanup failure. - error: sql`coalesce(${heartbeatRuns.error}, ${providerModelRejection.errorMessage})`, - errorCode: sql`coalesce(${heartbeatRuns.errorCode}, ${providerModelRejection.errorCode})`, + error: sql`coalesce(${heartbeatRuns.error}, ${providerFailure.errorMessage})`, + errorCode: sql`coalesce(${heartbeatRuns.errorCode}, ${providerFailure.errorCode})`, } : {}), resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) || ${JSON.stringify(finalizationMetadata)}::jsonb diff --git a/server/src/services/native-runtime/status-arbiter.test.ts b/server/src/services/native-runtime/status-arbiter.test.ts index 4162548217..b81270bd62 100644 --- a/server/src/services/native-runtime/status-arbiter.test.ts +++ b/server/src/services/native-runtime/status-arbiter.test.ts @@ -44,6 +44,25 @@ function arbitrate( } describe("native status authority", () => { + it("schedules a capacity retry while preserving partial work and review authority", () => { + for (const nativeReviewOutcome of [undefined, "pending"] as const) { + const decision = arbitrate({ terminalState: "failed", providerOverloaded: true, nativeReviewOutcome }); + expect(decision).toMatchObject({ statusAction: "preserve", reasonCode: "native_provider_overloaded" }); + expect(decision.effects).toContainEqual(expect.objectContaining({ kind: "schedule_retry", cause: "native_provider_overloaded" })); + } + expect(arbitrate({ terminalState: "failed", providerOverloaded: true, failureRetryCount: 2 })) + .toMatchObject({ statusAction: "blocked", reasonCode: "native_provider_overloaded_exhausted", effects: [{ kind: "bind_blocker", owner: "board", action: expect.stringContaining("Automatic retries exhausted") }] }); + }); + it.each([ + { hasActivePauseHold: true }, { hasUnresolvedIssueBlockers: true }, + { governanceGate: { kind: "approval" as const, id: "approval" } }, + { priorIssueStatus: "blocked" as const }, { priorIssueStatus: "done" as const }, + { workspaceFinalizeStatus: "failed" as const }, { nativeReviewOutcome: "stale" as const }, + { nativeReviewOutcome: "resolved" as const }, + ])("does not schedule capacity retries through existing authority or cleanup gates (%j)", (gate) => { + const decision = arbitrate({ terminalState: "failed", providerOverloaded: true, ...gate }); + expect(decision.effects.some(effect => effect.kind === "schedule_retry")).toBe(false); + }); it.each(["in_progress", "in_review"] as const)("blocks a current worker's proven model rejection without a retry (%s)", (priorIssueStatus) => { const decision = arbitrate({ priorIssueStatus, terminalState: "failed", providerModelRejected: true }); expect(decision).toMatchObject({ statusAction: "blocked", toStatus: "blocked", reasonCode: "native_provider_model_rejected", unblockDescriptor: { owner: "board" } }); @@ -528,7 +547,7 @@ describe("native status authority", () => { expect.objectContaining({ statusAction: "blocked", toStatus: "blocked", - policyVersion: "phase6-v8", + policyVersion: "phase6-v9", reasonCode: "current_track_blocker_waiting", unblockDescriptor: { owner: "board", diff --git a/server/src/services/native-runtime/status-arbiter.ts b/server/src/services/native-runtime/status-arbiter.ts index c4235536a6..d7525e401f 100644 --- a/server/src/services/native-runtime/status-arbiter.ts +++ b/server/src/services/native-runtime/status-arbiter.ts @@ -1,7 +1,7 @@ import type { NativeEvidenceAssessment } from "./evidence-classifier.js"; -import { NATIVE_MODEL_REJECTION_MESSAGE } from "./native-provider-failure.js"; +import { NATIVE_MODEL_REJECTION_MESSAGE, NATIVE_PROVIDER_CAPACITY_MAX_RETRIES, NATIVE_PROVIDER_OVERLOADED_CODE, NATIVE_PROVIDER_OVERLOADED_MESSAGE } from "./native-provider-failure.js"; -export const NATIVE_STATUS_ARBITER_POLICY_VERSION = "phase6-v8"; +export const NATIVE_STATUS_ARBITER_POLICY_VERSION = "phase6-v9"; export type NativeAuthoritativeIssueStatus = | "backlog" @@ -118,6 +118,9 @@ export function arbitrateNativeStatus(input: { nativeReviewOutcome?: "resolved" | "pending" | "stale"; /** Re-derived from the committed runner terminal and pinned execution identity. */ providerModelRejected?: boolean; + providerOverloaded?: boolean; + providerFailureSuperseded?: boolean; + failureRetryCount?: number; agentId: string; priorIssueStatus: NativeAuthoritativeIssueStatus; }): NativeStatusDecision { @@ -149,6 +152,11 @@ export function arbitrateNativeStatus(input: { ], }; } + if (input.providerFailureSuperseded) return { + policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, + statusAction: "preserve", toStatus: input.priorIssueStatus, + reasonCode: "provider_failure_authority_superseded", unblockDescriptor: null, effects: [], + }; if (input.providerModelRejected && input.terminalState === "failed" && input.nativeReviewOutcome !== "stale" && input.nativeReviewOutcome !== "resolved") { if (input.nativeReviewOutcome === "pending") { @@ -174,6 +182,35 @@ export function arbitrateNativeStatus(input: { effects: [{ kind: "bind_blocker", owner: "board", action: NATIVE_MODEL_REJECTION_MESSAGE }], }; } + if (input.providerOverloaded && input.terminalState === "failed" && + input.nativeReviewOutcome !== "stale" && input.nativeReviewOutcome !== "resolved") { + const held = input.governanceGate || input.hasUnresolvedIssueBlockers || input.hasActivePauseHold || input.priorIssueStatus === "blocked"; + const exhausted = (input.failureRetryCount ?? 0) >= NATIVE_PROVIDER_CAPACITY_MAX_RETRIES; + if (held || (exhausted && input.nativeReviewOutcome === "pending")) return { + policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, + statusAction: "preserve", toStatus: input.priorIssueStatus, + reasonCode: held ? "native_provider_overloaded_held" : "native_provider_overloaded_exhausted", + unblockDescriptor: null, effects: [{ kind: "release_checkout" }], + }; + if (exhausted) { + const action = `${NATIVE_PROVIDER_OVERLOADED_MESSAGE} Automatic retries exhausted; retry later or choose a different model.`; + return { + policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, + statusAction: "blocked", toStatus: "blocked", reasonCode: "native_provider_overloaded_exhausted", + unblockDescriptor: { owner: "board", action }, + effects: [{ kind: "bind_blocker", owner: "board", action }], + }; + } + return { + policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, + statusAction: "preserve", toStatus: input.priorIssueStatus, + reasonCode: NATIVE_PROVIDER_OVERLOADED_CODE, unblockDescriptor: null, + effects: [ + { kind: "schedule_retry", cause: NATIVE_PROVIDER_OVERLOADED_CODE, summary: "Retry after model capacity becomes available.", agentId: input.agentId }, + ...(input.nativeReviewOutcome === "pending" ? [{ kind: "release_checkout" as const }] : []), + ], + }; + } if (input.nativeReviewOutcome) { return { policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, diff --git a/server/src/services/native-runtime/status-decision-committer.ts b/server/src/services/native-runtime/status-decision-committer.ts index 21c14f5e93..3782958a47 100644 --- a/server/src/services/native-runtime/status-decision-committer.ts +++ b/server/src/services/native-runtime/status-decision-committer.ts @@ -49,6 +49,8 @@ import { type ActivityPublication, } from "../activity-log.js"; import { emitAgentTaskRun } from "../agent-task-run-telemetry.js"; +import { materializeNativeProviderCapacityRetry } from "./native-provider-capacity-retry.js"; +import { NATIVE_PROVIDER_OVERLOADED_CODE } from "./native-provider-failure.js"; export class NativeStatusRaceError extends Error { readonly code = "native_status_race" as const; @@ -788,6 +790,14 @@ async function materializeDecisionEffect(input: { } if (effect.kind === "schedule_retry") { failAt("continuation_materialization", input.failpoint); + if (effect.cause === NATIVE_PROVIDER_OVERLOADED_CODE) { + const retryId = await materializeNativeProviderCapacityRetry({ + tx: input.tx as unknown as Db, companyId: input.companyId, + issueId: input.issue.id, runId: input.runId, agentId: effect.agentId, + }); + return { effectKind: effect.kind, targetType: "heartbeat_run", targetId: retryId, + payload: { cause: effect.cause, summary: effect.summary } }; + } const wakeId = await enqueueWake({ tx: input.tx, companyId: input.companyId, @@ -1537,6 +1547,7 @@ export async function commitNativeStatusDecision(input: { supersedesCommittedDecisionId?: string; requireExternalChatResponseWaitAuthorization?: { agentId: string }; requireModelRejectionOwner?: { agentId: string; reviewContext: NativeReviewAssignmentContext | null }; + requireProviderFailureOwner?: { agentId: string; reviewContext: NativeReviewAssignmentContext | null }; requireBoardResponseWaitSource?: NativeBoardResponseWaitSource; requireBoardResponseWaitOrigin?: NativeBoardResponseWaitOrigin; reviewResponsePresentation?: { @@ -1552,6 +1563,9 @@ export async function commitNativeStatusDecision(input: { if (reasonCode === "native_provider_model_rejected" && !input.requireModelRejectionOwner) { throw new Error("native_model_rejection_owner_required"); } + if (reasonCode.startsWith("native_provider_overloaded") && !input.requireProviderFailureOwner) { + throw new Error("native_provider_failure_owner_required"); + } const publications: ActivityPublication[] = []; const terminalRunsToEmit: (typeof heartbeatRuns.$inferSelect)[] = []; const committed = await input.db.transaction(async (tx) => { @@ -1621,8 +1635,8 @@ export async function commitNativeStatusDecision(input: { ) { throw new NativeStatusRaceError(); } - if (input.requireModelRejectionOwner) { - const owner = input.requireModelRejectionOwner; + if (input.requireModelRejectionOwner || input.requireProviderFailureOwner) { + const owner = (input.requireProviderFailureOwner ?? input.requireModelRejectionOwner)!; // A successor can claim execution without changing statusVersion. Keep // this new blocking authority behind the same locked issue snapshot. if (issue.executionRunId && issue.executionRunId !== input.runId) throw new NativeStatusRaceError(); diff --git a/ui/src/components/IssueScheduledRetryCard.test.tsx b/ui/src/components/IssueScheduledRetryCard.test.tsx index a3b441bfb4..c3c7b5fc70 100644 --- a/ui/src/components/IssueScheduledRetryCard.test.tsx +++ b/ui/src/components/IssueScheduledRetryCard.test.tsx @@ -128,6 +128,15 @@ describe("IssueScheduledRetryCard", () => { expect(container.textContent).not.toMatch(/failed|Retry|Attempt|Replaces run/); expect(getRetryNowButton()).toBeNull(); }); + it("surfaces model capacity and the automatic retry without a generic failure", () => { + renderWithProviders(); + expect(container.textContent).toContain("Model at capacity"); + expect(container.textContent).toContain("Selected model is at capacity"); + expect(container.textContent).toContain("Paperclip will retry automatically"); + expect(container.textContent).not.toContain("Native session failed"); + }); + it("renders nothing when there is no scheduled retry", () => { renderWithProviders(); expect(getCard()).toBeNull(); diff --git a/ui/src/components/IssueScheduledRetryCard.tsx b/ui/src/components/IssueScheduledRetryCard.tsx index 8f792d7585..f9453fc291 100644 --- a/ui/src/components/IssueScheduledRetryCard.tsx +++ b/ui/src/components/IssueScheduledRetryCard.tsx @@ -115,7 +115,7 @@ export function IssueScheduledRetryCard({ ) : null} {scheduledRetry.error ? (
- Last attempt failed: {scheduledRetry.error}. Paperclip will retry automatically. + Last attempt failed: {scheduledRetry.error}{/[.!?]$/.test(scheduledRetry.error.trim()) ? "" : "."} Paperclip will retry automatically.
) : null} {isError ? ( diff --git a/ui/src/lib/runRetryState.test.ts b/ui/src/lib/runRetryState.test.ts index 9549cc3089..1152dba8eb 100644 --- a/ui/src/lib/runRetryState.test.ts +++ b/ui/src/lib/runRetryState.test.ts @@ -4,6 +4,7 @@ import { describeRunRetryState, formatRetryReason } from "./runRetryState"; describe("runRetryState", () => { it("formats internal retry reasons for operators", () => { expect(formatRetryReason("transient_failure")).toBe("Transient failure"); + expect(formatRetryReason("native_provider_overloaded")).toBe("Model at capacity"); expect(formatRetryReason("issue_continuation_needed")).toBe("Continuation needed"); expect(formatRetryReason("max_turns_continuation")).toBe("Max-turn continuation"); expect(formatRetryReason("custom_reason")).toBe("custom reason"); diff --git a/ui/src/lib/runRetryState.ts b/ui/src/lib/runRetryState.ts index ce332b48e9..2e4a47e8f4 100644 --- a/ui/src/lib/runRetryState.ts +++ b/ui/src/lib/runRetryState.ts @@ -19,6 +19,7 @@ export type RunRetryStateSummary = { }; const RETRY_REASON_LABELS: Record = { + native_provider_overloaded: "Model at capacity", ai_connection_pool_wait: "Waiting for pool usage to recover", transient_failure: "Transient failure", missing_issue_comment: "Missing task comment", diff --git a/ui/storybook/stories/model-capacity-retry.stories.tsx b/ui/storybook/stories/model-capacity-retry.stories.tsx new file mode 100644 index 0000000000..4ef94229a7 --- /dev/null +++ b/ui/storybook/stories/model-capacity-retry.stories.tsx @@ -0,0 +1,19 @@ +import type { Meta, StoryObj } from "@storybook/react-vite"; +import { IssueScheduledRetryCard } from "@/components/IssueScheduledRetryCard"; + +const meta = { + title: "Tasks/Model capacity retry", + component: IssueScheduledRetryCard, + decorators: [Story =>
], + args: { + issueId: "capacity-task", + scheduledRetry: { + runId: "capacity-retry", status: "scheduled_retry", agentId: "capacity-agent", agentName: "Codex", + retryOfRunId: "failed-capacity-run", scheduledRetryAt: new Date(Date.now() + 60_000).toISOString(), + scheduledRetryAttempt: 1, scheduledRetryReason: "native_provider_overloaded", retryExhaustedReason: null, + error: "Selected model is at capacity. Please try a different model.", errorCode: "native_provider_overloaded", + }, + }, +} satisfies Meta; +export default meta; +export const Waiting: StoryObj = {};