diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 6c536ced184..7ae7a155af9 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -1576,7 +1576,24 @@ unknown operations or partial transcripts do not. Replacement uses a fresh session and retires only the exact predecessor's obsolete recovery hold while recording the proof and successor lineage. Retained provider files are not edited. -Bootstrap retries, exact-checkpoint resumes, and fresh replacement sessions share three total provider attempts, including the original attempt. Linked run IDs, controller restarts, and duplicate wakes do not reset this budget. Automatic attempts retain the 30-second delay. Replacement scheduling and predecessor lineage commit together, with one successor per predecessor and admission through the normal task locks, authorization, pause, approval, and budget gates. +Bootstrap retries, exact-checkpoint resumes, and fresh replacement sessions share three total provider attempts, including the original attempt. Linked run IDs, controller restarts, and duplicate wakes do not reset this budget. Exact-checkpoint resumes retain the 30-second delay. Replacement scheduling and predecessor lineage commit together, with one successor per predecessor and admission through the normal task locks, authorization, pause, approval, and budget gates. + +A bootstrap retry requires durable evidence that no provider session, provider +event, or checkpoint exists. A runner report that `session.open` failed or was +rejected qualifies, because that command runs before any prompt. An +indeterminate `session.open` outcome does not qualify. Bootstrap retries back +off: about 30 seconds after the first attempt and about 60 seconds after the +second, with jitter that is persisted with the failure. Before it schedules a +bootstrap retry, the executor looks for a known provider limit for the agent. +Evidence is a `rate_limit` or `quota_exhausted` provider failure on the startup +failure itself, or on an agent run that started in the last day, or a recorded +usage-limit error code. The retry then waits until the provider's reported +reset time, or until one hour after the limit failure when no reset time is +known. A later successful run of the agent clears older evidence. A limit never +holds a retry for more than one day. Limit evidence only delays a retry; it does +not grant retry authority or change the budget. When the budget is exhausted, +the task is blocked and its recovery action states that provider startup failed +before any provider work began. A failed, unowned native sandbox retry can outlive an explicit Retry that already completed successfully. Startup recovery and ordinary retry admission retire that diff --git a/server/src/__tests__/chat-channels.integration.test.ts b/server/src/__tests__/chat-channels.integration.test.ts index 5364202d735..1f6172f7d65 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/server/src/__tests__/chat-channels.integration.test.ts @@ -66676,6 +66676,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { it.each([ "bootstrap", + "session_open", "checkpoint", "distinct_account", "null_account", @@ -66789,7 +66790,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { kind === "retryable" ? "retryable_failure" : "terminal_failure", nativeSessionId, runnerInstanceId, - errorCode: "provider_initialize_timeout", + errorCode: kind === "session_open" ? "provider_session_open_rejected" : "provider_initialize_timeout", processPid: kind === "live_pid" ? process.pid : null, runnerProfileJson: kind === "missing_input" @@ -66831,7 +66832,9 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { ? "native_event_replay_conflict" : kind === "cleanup" ? "native_session_cleanup_quarantined" - : "provider_initialize_timeout", + : kind === "session_open" + ? "provider_session_open_rejected" + : "provider_initialize_timeout", recoveryMode: kind === "ambiguous" ? "ambiguous_state" @@ -66938,7 +66941,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { ); // A null account without existing provider-session evidence remains // conservatively ineligible for the exhausted-run recovery lane. - if (["bootstrap", "checkpoint", "distinct_account", "notice_cleanup", "notice_replacement"].includes(kind)) { + if (["bootstrap", "session_open", "checkpoint", "distinct_account", "notice_cleanup", "notice_replacement"].includes(kind)) { const staged = await attempt; await expect( context.service.processFailedChatRunRetry(staged.actionId), diff --git a/server/src/services/chat-channels/failed-run-retry.ts b/server/src/services/chat-channels/failed-run-retry.ts index ecb300c3204..d49bbb6aafe 100644 --- a/server/src/services/chat-channels/failed-run-retry.ts +++ b/server/src/services/chat-channels/failed-run-retry.ts @@ -550,6 +550,7 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti "provider_request_protocol_error", "provider_transport_failed", "native_runner_process_exited", + "provider_session_open_rejected", ]; if ( !coordinator || diff --git a/server/src/services/native-runtime/native-session-executor.test.ts b/server/src/services/native-runtime/native-session-executor.test.ts index 94fa60242b5..86df326e85f 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -17317,6 +17317,22 @@ it("keeps an ACPX missing runtime package failure actionable and stops automatic expect(code).toBe("native_provider_runtime_missing"); expect(nativeSessionFailureDisposition(1, new Date(), code)).toEqual({ phase: "terminal_failure", failureCode: code, nextAttemptAt: null }); }); +it("classifies only a definite session.open outcome as a bounded pre-session failure", () => { + const detail = '{"result":{"code":"command_execution_failed","message":"failed to start ACPX provider: ACPX sidecar command session.open was rejected (retryable=false, classification=session_ensure_failed)"},"status":"failed"}'; + for (const status of ["failed", "rejected"]) { + const code = nativeSessionFailureSourceCode(new Error(`PRP command session.open ${status}: ${detail}`)); + expect(code).toBe("provider_session_open_rejected"); + expect(nativeSessionFailureDisposition(2, new Date(), code)).toMatchObject({ phase: "retryable_failure" }); + expect(nativeSessionFailureDisposition(3, new Date(), code)).toMatchObject({ + phase: "terminal_failure", failureCode: "native_session_retry_exhausted", nextAttemptAt: null, + }); + } + // The runner cannot say whether the command ran, so provider state stays unknown. + expect(nativeSessionFailureSourceCode(new Error(`PRP command session.open indeterminate: ${detail}`))) + .toBe("native_session_interrupted"); + expect(nativeSessionFailureSourceCode(new Error(`PRP command turn.start failed: ${detail}`))) + .toBe("native_session_interrupted"); +}); describe("HTTP Stop during remote runner staging", () => { it("does not dispatch after Stop and bounded bootstrap cleanup", async () => { const { createCapabilityRunnerdCodexTransport } = await import("@paperclipai/paperclip-runner/live"); diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index ada28ba032d..06a4f78b4a2 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -70,6 +70,7 @@ import { createAgentIdentityRedactor } from "../agent-identity-redaction.js"; import { parseCodexCliVersion } from "./codex-runtime-compatibility.js"; import { createNativeToolTrace, type NativeToolTrace } from "./native-tool-trace.js"; import { createNativeProviderFailureObservation } from "./native-provider-failure.js"; +import { nativeStartupRetryExhaustedAction, nativeStartupRetryWait, planNativeStartupRetry } from "./native-startup-retry.js"; import { createNativeDeliverableTextObservation } from "./native-deliverable-text.js"; import { createNativeGitHubAccess, type NativeGitHubAccess } from "./native-github-access.js"; import { resolveGitHubOperationCredentials } from "../github-operation-credentials.js"; @@ -7061,6 +7062,7 @@ export function nativeSessionFailureSourceCode( | "native_provider_model_rejected" | "native_current_wake_comments_unread" | "native_current_wake_comments_changed_after_read" + | "provider_session_open_rejected" | "native_session_interrupted" { if (error instanceof NativePermissionDeclinedError) return "native_permission_declined"; if (error instanceof NativeProviderTerminalFailure) { @@ -7144,6 +7146,12 @@ export function nativeSessionFailureSourceCode( if (/native_current_wake_comments_unread/i.test(message)) { return "native_current_wake_comments_unread"; } + // The runner reported a definite session.open outcome. That command runs + // before any prompt, so provider work cannot have started. An indeterminate + // outcome stays generic. + if (/\bPRP command session\.open (?:failed|rejected): /.test(message)) { + return "provider_session_open_rejected"; + } return "native_session_interrupted"; } @@ -7226,6 +7234,7 @@ export async function nativeProviderRecoveryEvidence(input: { "provider_request_timeout", "provider_request_protocol_error", "native_runner_process_exited", + "provider_session_open_rejected", ]).has(input.sourceFailureCode); if (!checkpointExists && !providerEventsExist && definitelyPreSession) { return { @@ -10382,10 +10391,31 @@ async function executePaperclipNativeSessionWithinScope( recoveryEvidence.recoveryMode === "ambiguous_state" ? sourceFailureCode : disposition.failureCode; + const providerFailure = providerFailureFromError(error); + // Pre-provider startup retries back off and wait out a known provider + // limit; their exhaustion explains the startup failure. + const startupRetryExhausted = + recoveryEvidence.recoveryMode === "bootstrap_retry" && + failureCode === "native_session_retry_exhausted"; + const startupRetry = + recoveryEvidence.recoveryMode === "bootstrap_retry" && + (phase === "retryable_failure" || startupRetryExhausted) + ? await planNativeStartupRetry({ + db: input.db, + binding: input.execution.binding, + attempt, + retryable: phase === "retryable_failure", + failure: providerFailure, + now, + }) + : null; + const providerCooldownAt = startupRetry?.providerCooldownAt ?? null; const nextAttemptAt = recoveryEvidence.recoveryMode === "ambiguous_state" ? null - : disposition.nextAttemptAt; + : startupRetry + ? startupRetry.nextAttemptAt + : disposition.nextAttemptAt; const recoveryProjection = nativeSessionRecoveryProjection({ phase, failureCode, @@ -10394,7 +10424,6 @@ async function executePaperclipNativeSessionWithinScope( const { exhausted } = recoveryProjection; const integrityFailure = sourceFailureCode === "native_event_replay_conflict"; - const providerFailure = providerFailureFromError(error); const message = redactIdentityText( error instanceof Error ? error.message : String(error), ).slice(0, 2_000); @@ -10449,6 +10478,7 @@ async function executePaperclipNativeSessionWithinScope( recoveryEvidence.providerSessionEstablished, providerEventsExist: recoveryEvidence.providerEventsExist, checkpointExists: recoveryEvidence.checkpointExists, + ...(providerCooldownAt ? { providerCooldownAt: providerCooldownAt.toISOString() } : {}), recoveryOwner: recoveryProjection.recoveryOwner, nextAction: sourceFailureCode === "native_permission_declined" @@ -10469,11 +10499,13 @@ async function executePaperclipNativeSessionWithinScope( ? "Restore model provider usage capacity, then explicitly retry the task. Automatic retries cannot resolve an exhausted provider allowance." : ownershipUnverified ? "Inspect the retained runner's executable and authenticated connection. Do not replace its provider until ownership is safely resolved." + : startupRetryExhausted + ? nativeStartupRetryExhaustedAction(attempt, providerCooldownAt) : exhausted ? "Inspect the persisted native session after its bounded resume budget was exhausted." : recoveryEvidence.recoveryMode === "bootstrap_retry" - ? "Retry provider bootstrap on this same run; durable evidence proves no provider session or provider event was created." + ? `Retry provider bootstrap on this same run${nativeStartupRetryWait(providerCooldownAt)}; durable evidence proves no provider session or provider event was created.` : "Resume this same run from its exact persisted native provider checkpoint after the retry delay.", }), nextAttemptAt, @@ -10640,6 +10672,7 @@ async function executePaperclipNativeSessionWithinScope( recoveryMode: recoveryEvidence.recoveryMode, providerSessionEstablished: recoveryEvidence.providerSessionEstablished, + ...(providerCooldownAt ? { providerCooldownAt: providerCooldownAt.toISOString() } : {}), }, nextAction: sourceFailureCode === "native_permission_declined" @@ -10658,10 +10691,12 @@ async function executePaperclipNativeSessionWithinScope( ? "Restore model provider usage capacity, then explicitly retry the task; automatic retries are stopped." : ownershipUnverified ? "Resolve the retained runner's authentication or executable compatibility before an explicit recovery; do not blindly restart, cancel, or replace its provider session." + : startupRetryExhausted + ? nativeStartupRetryExhaustedAction(attempt, providerCooldownAt) : exhausted ? "Inspect the provider trace and explicitly choose a replacement run or provider configuration; automatic provider work is stopped." : recoveryEvidence.recoveryMode === "bootstrap_retry" - ? "Retry bootstrap on the same run without manufacturing a provider checkpoint." + ? `Retry bootstrap on the same run${nativeStartupRetryWait(providerCooldownAt)} without manufacturing a provider checkpoint.` : "Resume the exact persisted native session on the same heartbeat run.", wakePolicy: nextAttemptAt ? { diff --git a/server/src/services/native-runtime/native-startup-retry.integration.test.ts b/server/src/services/native-runtime/native-startup-retry.integration.test.ts new file mode 100644 index 00000000000..e335d4e13a3 --- /dev/null +++ b/server/src/services/native-runtime/native-startup-retry.integration.test.ts @@ -0,0 +1,554 @@ +import { randomUUID } from "node:crypto"; +import { eq, sql } from "drizzle-orm"; +import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; +import { + agents, + companies, + completionContracts, + createDb, + heartbeatRunEvents, + heartbeatRuns, + issueRecoveryActions, + issues, + nativeRunFinalizations, +} from "@paperclipai/db"; +import type { NativeExecutionInputV1 } from "@paperclipai/paperclip-runner"; +import { startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; + +const provider = vi.hoisted(() => ({ execute: vi.fn() })); +vi.mock("../../vendor/paperclip-runner/index.js", async (importOriginal) => ({ + ...(await importOriginal< + typeof import("../../vendor/paperclip-runner/index.js") + >()), + executeNativeSession: provider.execute, +})); + +import { executePaperclipNativeSession } from "./native-session-executor.js"; +import { buildNativeCompletionContract } from "./completion-contracts.js"; +import { prepareNativeHeartbeatRun } from "./prepare-native-run.js"; +import { claimNativeSessionResumptions } from "./native-finalization-reconciler.js"; + +type SessionOpenStatus = "failed" | "rejected" | "indeterminate"; + +/** Mirrors the runner's failed-command error: provider detail stays off the message. */ +function sessionOpenFailure(options: { + status?: SessionOpenStatus; + providerFailure?: Record; +} = {}) { + const status = options.status ?? "failed"; + const result = { + result: { + code: "command_execution_failed", + message: options.providerFailure + ? "Provider request failed" + : "failed to start ACPX provider: ACPX sidecar command session.open was rejected (retryable=false, classification=session_ensure_failed)", + ...(options.providerFailure ? { providerFailure: options.providerFailure } : {}), + }, + status, + }; + const error = new Error( + `PRP command session.open ${status}: ${JSON.stringify(result, (key, value) => + key === "providerFailure" ? undefined : value)}`, + ); + Object.defineProperty(error, "runnerCommandResult", { + value: structuredClone(result), + enumerable: false, + }); + return error; +} + +function limitFailure(overrides: Record = {}) { + return { + schema: "paperclip.provider-failure.v1", + provider: "anthropic", + phase: "turn", + category: "rate_limit", + message: "Usage limit reached for this account.", + retryability: "unknown", + ...overrides, + }; +} + +const backoffWindow = (from: Date, attempt: number) => { + const base = 30_000 * 2 ** (attempt - 1); + return [from.getTime() + base * 0.8, from.getTime() + base * 1.2] as const; +}; + +describe("native provider startup retry", () => { + let temporary: Awaited>; + let db: ReturnType; + const companyId = randomUUID(); + + beforeAll(async () => { + temporary = await startEmbeddedPostgresTestDatabase("native-startup-retry-"); + db = createDb(temporary.connectionString); + await db.insert(companies).values({ + id: companyId, + name: "Native startup retry", + issuePrefix: "NSR", + }); + }); + + afterAll(async () => { + await temporary?.cleanup(); + }); + + async function createAgent() { + const agentId = randomUUID(); + await db.insert(agents).values({ + id: agentId, + companyId, + name: `Startup agent ${agentId.slice(0, 8)}`, + status: "active", + adapterType: "paperclip_runner", + adapterConfig: { provider: "acpx", agent: "claude" }, + }); + return agentId; + } + + async function seedRun(agentId: string) { + const issueId = randomUUID(); + const runId = randomUUID(); + const [issue] = await db + .insert(issues) + .values({ + id: issueId, + companyId, + title: "Start work after a provider startup failure", + status: "in_progress", + workMode: "standard", + assigneeAgentId: agentId, + }) + .returning(); + const [run] = await db + .insert(heartbeatRuns) + .values({ + id: runId, + companyId, + agentId, + status: "running", + runtimeMode: "native", + nativeIssueId: issueId, + invocationSource: "assignment", + triggerDetail: "system", + startedAt: new Date(), + contextSnapshot: { issueId }, + }) + .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 execution: NativeExecutionInputV1 = { + schema: "paperclip.native-execution-input.v1", + binding: { companyId, runId, issueId, agentId, executionWorkspaceId: runId }, + provider: { kind: "codex", model: "gpt-5.6-luna" }, + task: { + identifier: issueId, + title: issue.title, + description: null, + prompt: "Continue the task.", + 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: buildNativeCompletionContract(issue, { revision: contract.revision }), + }, + interactionResponses: [], + credentialBindings: [], + }; + // Dispatch persists this envelope before execution; same-run recovery needs it. + await db + .update(heartbeatRuns) + .set({ + runnerProfileJson: sql`coalesce(${heartbeatRuns.runnerProfileJson}, '{}'::jsonb) + || ${JSON.stringify({ nativeExecutionInput: execution })}::jsonb`, + }) + .where(eq(heartbeatRuns.id, runId)); + await db.insert(nativeRunFinalizations).values({ + runId, + companyId, + issueId, + phase: "observed", + }); + return { agentId, issueId, runId, execution, runnerInstanceId: native.runnerInstanceId }; + } + + type Seeded = Awaited>; + + async function failStartup( + seeded: Seeded, + failure: Error, + options: { leaseOwner?: string; beforeFailure?: () => Promise } = {}, + ) { + provider.execute.mockReset().mockImplementation(async () => { + await options.beforeFailure?.(); + throw failure; + }); + const startedAt = new Date(); + await expect( + executePaperclipNativeSession({ + db, + execution: seeded.execution, + runnerInstanceId: seeded.runnerInstanceId, + ...(options.leaseOwner ? { leaseOwner: options.leaseOwner } : {}), + backend: { + descriptor: async () => { + throw new Error("Unexpected provider descriptor call"); + }, + openSession: async () => { + throw new Error("Unexpected provider session call"); + }, + }, + }), + ).rejects.toThrow("PRP command session.open"); + expect(provider.execute).toHaveBeenCalledTimes(1); + return startedAt; + } + + async function read(seeded: Seeded) { + const [coordinator] = await db + .select() + .from(nativeRunFinalizations) + .where(eq(nativeRunFinalizations.runId, seeded.runId)); + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, seeded.runId)); + const [issue] = await db.select().from(issues).where(eq(issues.id, seeded.issueId)); + const actions = await db + .select() + .from(issueRecoveryActions) + .where(eq(issueRecoveryActions.sourceIssueId, seeded.issueId)); + const action = actions.find((entry) => ["active", "escalated"].includes(entry.status)) ?? null; + return { coordinator: coordinator!, run: run!, issue: issue!, action }; + } + + /** Claims the scheduled retry through the same path the recovery scheduler uses. */ + async function claimRetry(seeded: Seeded) { + const { coordinator } = await read(seeded); + expect(coordinator.nextAttemptAt).toBeInstanceOf(Date); + const early = await claimNativeSessionResumptions({ + db, + runnerInstanceId: "startup-retry-test", + runIds: [seeded.runId], + now: new Date(coordinator.nextAttemptAt!.getTime() - 1), + deferIneligible: false, + }); + expect(early).toEqual([]); + const claims = await claimNativeSessionResumptions({ + db, + runnerInstanceId: "startup-retry-test", + runIds: [seeded.runId], + now: new Date(coordinator.nextAttemptAt!.getTime() + 1), + deferIneligible: false, + }); + expect(claims).toHaveLength(1); + return claims[0]!.leaseOwner; + } + + async function recordPriorRun(agentId: string, values: Partial) { + await db.insert(heartbeatRuns).values({ + id: randomUUID(), + companyId, + agentId, + runtimeMode: "native", + invocationSource: "assignment", + triggerDetail: "system", + ...values, + }); + } + + it("retries a rejected session.open with growing backoff instead of blocking the task", async () => { + const seeded = await seedRun(await createAgent()); + + const firstFailedAt = await failStartup(seeded, sessionOpenFailure()); + let state = await read(seeded); + expect(state.coordinator).toMatchObject({ + phase: "retryable_failure", + recoveryState: "resuming_session", + attempt: 1, + leaseOwner: null, + failureDetail: expect.objectContaining({ + originalFailureCode: "provider_session_open_rejected", + recoveryMode: "bootstrap_retry", + providerSessionEstablished: false, + providerEventsExist: false, + checkpointExists: false, + }), + }); + const [firstLow, firstHigh] = backoffWindow(firstFailedAt, 1); + expect(state.coordinator.nextAttemptAt!.getTime()).toBeGreaterThanOrEqual(firstLow); + expect(state.coordinator.nextAttemptAt!.getTime()).toBeLessThanOrEqual(firstHigh + 5_000); + expect(state.coordinator.failureDetail).not.toHaveProperty("providerCooldownAt"); + expect(state.issue.status).toBe("in_progress"); + expect(state.run).toMatchObject({ + status: "failed", + nativePhase: "retryable_failure", + errorCode: "provider_session_open_rejected", + }); + expect(state.action).toMatchObject({ + ownerType: "agent", + ownerAgentId: seeded.agentId, + wakePolicy: { + kind: "resume_native_run", + runId: seeded.runId, + notBefore: state.coordinator.nextAttemptAt!.toISOString(), + }, + }); + + const secondFailedAt = await failStartup(seeded, sessionOpenFailure(), { + leaseOwner: await claimRetry(seeded), + }); + state = await read(seeded); + expect(state.coordinator).toMatchObject({ phase: "retryable_failure", attempt: 2 }); + const [secondLow, secondHigh] = backoffWindow(secondFailedAt, 2); + expect(state.coordinator.nextAttemptAt!.getTime()).toBeGreaterThanOrEqual(secondLow); + expect(state.coordinator.nextAttemptAt!.getTime()).toBeLessThanOrEqual(secondHigh + 5_000); + expect(state.issue.status).toBe("in_progress"); + }); + + it("blocks with a clear reason once the bounded startup budget is exhausted", async () => { + const seeded = await seedRun(await createAgent()); + await failStartup(seeded, sessionOpenFailure()); + await failStartup(seeded, sessionOpenFailure(), { leaseOwner: await claimRetry(seeded) }); + await failStartup(seeded, sessionOpenFailure(), { leaseOwner: await claimRetry(seeded) }); + + const state = await read(seeded); + expect(state.coordinator).toMatchObject({ + phase: "terminal_failure", + recoveryState: "blocked", + failureCode: "native_session_retry_exhausted", + attempt: 3, + nextAttemptAt: null, + failureDetail: expect.objectContaining({ + originalFailureCode: "provider_session_open_rejected", + recoveryMode: "bootstrap_retry", + providerSessionEstablished: false, + }), + }); + expect(state.issue.status).toBe("blocked"); + expect(state.issue.executionRunId).toBeNull(); + expect(state.action).toMatchObject({ + ownerType: "board", + cause: "native_session_retry_exhausted", + wakePolicy: null, + }); + expect(state.action!.nextAction).toContain("3 attempts"); + expect(state.action!.nextAction).toContain("before any provider work began"); + expect(String(state.coordinator.failureDetail?.nextAction)).toContain("before any provider work began"); + // The task cannot be claimed for another automatic attempt. + expect(await claimNativeSessionResumptions({ + db, + runnerInstanceId: "startup-retry-test", + runIds: [seeded.runId], + now: new Date(Date.now() + 60 * 60_000), + deferIneligible: false, + })).toEqual([]); + }); + + it("waits for the cooldown of a recent provider usage limit before retrying", async () => { + const agentId = await createAgent(); + const limitedAt = new Date(Date.now() - 5 * 60_000); + await recordPriorRun(agentId, { + status: "failed", + startedAt: new Date(limitedAt.getTime() - 60_000), + finishedAt: limitedAt, + errorCode: "native_provider_terminal_failed", + resultJson: { providerFailure: limitFailure() }, + }); + const seeded = await seedRun(agentId); + + const failedAt = await failStartup(seeded, sessionOpenFailure()); + const state = await read(seeded); + const cooldownAt = new Date(limitedAt.getTime() + 60 * 60_000); + expect(state.coordinator).toMatchObject({ + phase: "retryable_failure", + failureDetail: expect.objectContaining({ + recoveryMode: "bootstrap_retry", + providerCooldownAt: cooldownAt.toISOString(), + }), + }); + const [low, high] = backoffWindow(cooldownAt, 1); + expect(state.coordinator.nextAttemptAt!.getTime()).toBeGreaterThanOrEqual(low); + expect(state.coordinator.nextAttemptAt!.getTime()).toBeLessThanOrEqual(high); + expect(state.coordinator.nextAttemptAt!.getTime()).toBeGreaterThan(failedAt.getTime() + 50 * 60_000); + expect(String(state.coordinator.failureDetail?.nextAction)).toContain(cooldownAt.toISOString()); + expect(state.issue.status).toBe("in_progress"); + expect(state.action).toMatchObject({ + ownerType: "agent", + evidence: expect.objectContaining({ providerCooldownAt: cooldownAt.toISOString() }), + wakePolicy: expect.objectContaining({ + kind: "resume_native_run", + notBefore: state.coordinator.nextAttemptAt!.toISOString(), + }), + }); + expect(await claimNativeSessionResumptions({ + db, + runnerInstanceId: "startup-retry-test", + runIds: [seeded.runId], + now: new Date(cooldownAt.getTime() - 1), + deferIneligible: false, + })).toEqual([]); + }); + + it("uses a provider-reported reset time from earlier runs or from the startup failure itself", async () => { + const recordedAgent = await createAgent(); + const recordedResetAt = new Date(Date.now() + 3 * 60 * 60_000); + await recordPriorRun(recordedAgent, { + status: "failed", + startedAt: new Date(Date.now() - 20 * 60_000), + finishedAt: new Date(Date.now() - 10 * 60_000), + errorCode: "native_provider_terminal_failed", + resultJson: { + providerFailure: limitFailure({ + category: "quota_exhausted", + retryAt: recordedResetAt.toISOString(), + retryability: "requires_action", + }), + }, + }); + const recorded = await seedRun(recordedAgent); + await failStartup(recorded, sessionOpenFailure()); + let state = await read(recorded); + expect(state.coordinator.failureDetail).toMatchObject({ + providerCooldownAt: recordedResetAt.toISOString(), + }); + expect(state.coordinator.nextAttemptAt!.getTime()).toBeGreaterThanOrEqual( + backoffWindow(recordedResetAt, 1)[0], + ); + + // A runner that classifies the startup rejection itself supplies the evidence. + const startupResetAt = new Date(Date.now() + 2 * 60 * 60_000); + const classified = await seedRun(await createAgent()); + await failStartup(classified, sessionOpenFailure({ + providerFailure: limitFailure({ + phase: "startup", + retryAt: startupResetAt.toISOString(), + retryability: "transient", + }), + })); + state = await read(classified); + expect(state.coordinator).toMatchObject({ + phase: "retryable_failure", + failureDetail: expect.objectContaining({ + recoveryMode: "bootstrap_retry", + providerCooldownAt: startupResetAt.toISOString(), + }), + }); + expect(state.coordinator.nextAttemptAt!.getTime()).toBeGreaterThanOrEqual( + backoffWindow(startupResetAt, 1)[0], + ); + expect(state.issue.status).toBe("in_progress"); + }); + + it("holds a startup retry for at most one day before trying again", async () => { + const seeded = await seedRun(await createAgent()); + const failedAt = await failStartup(seeded, sessionOpenFailure({ + providerFailure: limitFailure({ + phase: "startup", + category: "quota_exhausted", + retryAt: new Date(Date.now() + 3 * 24 * 60 * 60_000).toISOString(), + retryability: "requires_action", + }), + })); + const state = await read(seeded); + const cooldownAt = Date.parse(String(state.coordinator.failureDetail?.providerCooldownAt)); + expect(cooldownAt).toBeGreaterThanOrEqual(failedAt.getTime() + 24 * 60 * 60_000); + expect(cooldownAt).toBeLessThanOrEqual(Date.now() + 24 * 60 * 60_000); + expect(state.coordinator.nextAttemptAt!.getTime()).toBeGreaterThanOrEqual(backoffWindow(new Date(cooldownAt), 1)[0]); + }); + + it("ignores limit evidence that a later success, another agent, or time has cleared", async () => { + const agentId = await createAgent(); + const otherAgentId = await createAgent(); + await recordPriorRun(agentId, { + status: "failed", + startedAt: new Date(Date.now() - 30 * 60_000), + finishedAt: new Date(Date.now() - 25 * 60_000), + errorCode: "native_provider_terminal_failed", + resultJson: { providerFailure: limitFailure() }, + }); + await recordPriorRun(agentId, { + status: "succeeded", + startedAt: new Date(Date.now() - 20 * 60_000), + finishedAt: new Date(Date.now() - 15 * 60_000), + }); + await recordPriorRun(agentId, { + status: "failed", + startedAt: new Date(Date.now() - 3 * 60 * 60_000), + finishedAt: new Date(Date.now() - 2 * 60 * 60_000), + errorCode: "native_provider_usage_limit", + }); + await recordPriorRun(otherAgentId, { + status: "failed", + startedAt: new Date(Date.now() - 2 * 60_000), + finishedAt: new Date(Date.now() - 60_000), + errorCode: "native_provider_terminal_failed", + resultJson: { providerFailure: limitFailure() }, + }); + const seeded = await seedRun(agentId); + + const failedAt = await failStartup(seeded, sessionOpenFailure()); + const state = await read(seeded); + expect(state.coordinator.phase).toBe("retryable_failure"); + expect(state.coordinator.failureDetail).not.toHaveProperty("providerCooldownAt"); + expect(state.coordinator.nextAttemptAt!.getTime()).toBeLessThanOrEqual( + backoffWindow(failedAt, 1)[1] + 5_000, + ); + }); + + it.each([ + { name: "a durable provider event exists", status: "failed" as const, providerEvent: true }, + { name: "the session.open outcome is indeterminate", status: "indeterminate" as const, providerEvent: false }, + ])("never retries automatically when $name", async ({ status, providerEvent }) => { + const seeded = await seedRun(await createAgent()); + await failStartup(seeded, sessionOpenFailure({ status }), { + beforeFailure: providerEvent + ? async () => { + await db.insert(heartbeatRunEvents).values({ + companyId, + runId: seeded.runId, + agentId: seeded.agentId, + seq: 900, + eventType: "turn.started", + stream: "system", + }); + } + : undefined, + }); + + const state = await read(seeded); + expect(state.coordinator).toMatchObject({ + phase: "terminal_failure", + recoveryState: "blocked", + nextAttemptAt: null, + failureDetail: expect.objectContaining({ + recoveryMode: "ambiguous_state", + providerSessionEstablished: providerEvent, + }), + }); + expect(state.issue.status).toBe("blocked"); + expect(state.action).toMatchObject({ ownerType: "board", wakePolicy: null }); + expect(await claimNativeSessionResumptions({ + db, + runnerInstanceId: "startup-retry-test", + runIds: [seeded.runId], + now: new Date(Date.now() + 60 * 60_000), + deferIneligible: false, + })).toEqual([]); + }); +}); diff --git a/server/src/services/native-runtime/native-startup-retry.ts b/server/src/services/native-runtime/native-startup-retry.ts new file mode 100644 index 00000000000..619859de44f --- /dev/null +++ b/server/src/services/native-runtime/native-startup-retry.ts @@ -0,0 +1,108 @@ +import { createHash } from "node:crypto"; +import { and, desc, eq, gte, inArray, ne, or, sql } from "drizzle-orm"; +import { heartbeatRuns, type Db } from "@paperclipai/db"; +import { readProviderFailure, type ProviderFailure } from "@paperclipai/shared"; +import { logger } from "../../middleware/logger.js"; +import { infrastructureRetryDelayMs, PROVIDER_LIMIT_DEFAULT_COOLDOWN_MS } from "../system-recovery-state.js"; + +/** Categories that report an exhausted account or request allowance. */ +const PROVIDER_LIMIT_CATEGORIES = ["rate_limit", "quota_exhausted"]; +/** Run errors that record an exhausted allowance without a typed provider failure. */ +const PROVIDER_LIMIT_ERROR_CODES = ["native_provider_usage_limit", "provider_quota"]; +/** Limit evidence older than this is ignored, and no limit holds a retry longer. */ +const NATIVE_PROVIDER_LIMIT_WINDOW_MS = 24 * 60 * 60_000; +const LIMIT_EVIDENCE_ROWS = 20; + +function limitCooldownAt(failure: ProviderFailure | null, observedAt: Date): number | null { + if (!failure || !PROVIDER_LIMIT_CATEGORIES.includes(failure.category)) return null; + const retryAt = failure.retryAt ? Date.parse(failure.retryAt) : Number.NaN; + return Number.isFinite(retryAt) ? retryAt : observedAt.getTime() + PROVIDER_LIMIT_DEFAULT_COOLDOWN_MS; +} + +/** + * The latest known provider limit for this agent, from the current failure or + * from the agent's recent runs. A later successful run clears older evidence. + */ +async function nativeProviderCooldownAt(input: { + db: Db; + binding: { companyId: string; agentId: string; runId: string }; + failure: ProviderFailure | null; + now: Date; +}): Promise { + const now = input.now.getTime(); + let cooldownAt = limitCooldownAt(input.failure, input.now) ?? 0; + const rows = await input.db + .select({ + status: heartbeatRuns.status, + errorCode: heartbeatRuns.errorCode, + observedAt: sql`coalesce(${heartbeatRuns.finishedAt}, ${heartbeatRuns.updatedAt})`.mapWith(heartbeatRuns.updatedAt), + providerFailure: sql`${heartbeatRuns.resultJson} -> 'providerFailure'`, + retryNotBefore: sql`${heartbeatRuns.resultJson} ->> 'retryNotBefore'`, + }) + .from(heartbeatRuns) + .where(and( + eq(heartbeatRuns.companyId, input.binding.companyId), + eq(heartbeatRuns.agentId, input.binding.agentId), + gte(heartbeatRuns.startedAt, new Date(now - NATIVE_PROVIDER_LIMIT_WINDOW_MS)), + ne(heartbeatRuns.id, input.binding.runId), + or( + eq(heartbeatRuns.status, "succeeded"), + inArray(heartbeatRuns.errorCode, PROVIDER_LIMIT_ERROR_CODES), + inArray(sql`${heartbeatRuns.resultJson} #>> '{providerFailure,category}'`, PROVIDER_LIMIT_CATEGORIES), + ), + )) + .orderBy(desc(sql`coalesce(${heartbeatRuns.finishedAt}, ${heartbeatRuns.updatedAt})`)) + .limit(LIMIT_EVIDENCE_ROWS); + for (const row of rows) { + // The provider accepted work after this point, so its allowance was restored. + if (row.status === "succeeded") break; + const failure = readProviderFailure(row.providerFailure); + let recorded = limitCooldownAt(failure, row.observedAt); + if (recorded === null && PROVIDER_LIMIT_ERROR_CODES.includes(row.errorCode ?? "")) { + const reset = row.retryNotBefore ? Date.parse(row.retryNotBefore) : Number.NaN; + recorded = Number.isFinite(reset) ? reset : row.observedAt.getTime() + PROVIDER_LIMIT_DEFAULT_COOLDOWN_MS; + } + if (recorded !== null) cooldownAt = Math.max(cooldownAt, recorded); + } + return cooldownAt > now ? new Date(Math.min(cooldownAt, now + NATIVE_PROVIDER_LIMIT_WINDOW_MS)) : null; +} + +/** + * Schedules a same-run bootstrap retry after a failure that provably came + * before provider work. The attempt budget stays with the coordinator; this + * adds exponential backoff and waits out a known provider limit. + */ +export async function planNativeStartupRetry(input: { + db: Db; + binding: { companyId: string; agentId: string; runId: string }; + attempt: number; + retryable: boolean; + failure: ProviderFailure | null; + now: Date; +}): Promise<{ nextAttemptAt: Date | null; providerCooldownAt: Date | null }> { + // Limit evidence only refines the schedule. A failed read must not stop the + // failure from settling, so the retry falls back to plain backoff. + const providerCooldownAt = await nativeProviderCooldownAt(input).catch((error: unknown) => { + logger.warn({ err: error, runId: input.binding.runId }, "native provider limit evidence unavailable"); + return null; + }); + if (!input.retryable) return { nextAttemptAt: null, providerCooldownAt }; + // Persisted with the failure, so a restart keeps the same jitter. + const sample = createHash("sha256").update(`${input.binding.runId}:${input.attempt}`) + .digest().readUInt32BE(0) / 0xffffffff; + const from = Math.max(input.now.getTime(), providerCooldownAt?.getTime() ?? 0); + return { + nextAttemptAt: new Date(from + infrastructureRetryDelayMs(input.attempt, sample)), + providerCooldownAt, + }; +} + +export function nativeStartupRetryExhaustedAction(attempts: number, providerCooldownAt: Date | null) { + return `Provider startup failed on ${attempts} attempts before any provider work began. Automatic retries are stopped.${ + providerCooldownAt ? ` The provider usage limit for this agent is expected to reset at ${providerCooldownAt.toISOString()}.` : "" + } Check the agent's AI connection and provider usage, then retry the task.`; +} + +export function nativeStartupRetryWait(providerCooldownAt: Date | null) { + return providerCooldownAt ? ` after the provider usage limit resets at ${providerCooldownAt.toISOString()}` : ""; +} diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index e020141baa4..67d6eeae20c 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -112,6 +112,7 @@ import { emitAgentTaskRun } from "../agent-task-run-telemetry.js"; import { budgetService } from "../budgets.js"; import { unadmittedChatWakeupCondition } from "../durable-chat-wakeup.js"; import { issueRecoveryActionService } from "../issue-recovery-actions.js"; +import { PROVIDER_LIMIT_DEFAULT_COOLDOWN_MS } from "../system-recovery-state.js"; import { hasUnverifiedNativeWorkspaceFinalizationOwner } from "../native-runtime/native-workspace-finalization-recovery-state.js"; import { legacyExecutionNeedsReconciliationWithEvidence, @@ -569,7 +570,7 @@ const INTERACTION_CONTINUATION_REQUEUE_MAX_ATTEMPTS = 3; const CONTINUATION_RECOVERY_TRANSIENT_MAX_ATTEMPTS = 3; const CONTINUATION_RECOVERY_DEFAULT_MAX_ATTEMPTS = 1; const CONTINUATION_RECOVERY_TRANSIENT_BASE_BACKOFF_MS = 60_000; -export const PROVIDER_QUOTA_RECOVERY_DEFAULT_BACKOFF_MS = 60 * 60 * 1000; +export const PROVIDER_QUOTA_RECOVERY_DEFAULT_BACKOFF_MS = PROVIDER_LIMIT_DEFAULT_COOLDOWN_MS; const PROVIDER_QUOTA_ERROR_RE = /(?:you(?:'|’)ve hit your (?:\w+ )?limit|usage limit(?: reached| exceeded)?|provider quota|quota (?:limit )?exceeded|model (?:is )?at capacity)/i; diff --git a/server/src/services/system-recovery-state.ts b/server/src/services/system-recovery-state.ts index d29e20f4150..c1777a1a8f6 100644 --- a/server/src/services/system-recovery-state.ts +++ b/server/src/services/system-recovery-state.ts @@ -9,6 +9,8 @@ export const SYSTEM_RECOVERY_WINDOW_MS = 30 * 60_000; export function infrastructureRetryDelayMs(retryNumber: number, sample: number) { return Math.round(30_000 * 2 ** (retryNumber - 1) * (0.8 + Math.min(1, Math.max(0, sample)) * 0.4)); } +/** Assumed provider limit cooldown when the provider reports no reset time. */ +export const PROVIDER_LIMIT_DEFAULT_COOLDOWN_MS = 60 * 60_000; export const recoveryFailureCodes = [ "historical_source_missing", "source_binding_invalid", "allocation_ownership_ambiguous",