diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 4e604ccd82b..6c2c318e36b 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -3062,9 +3062,28 @@ account at once. ACP harnesses report an exhausted allowance as a generic terminal limit failure; Runner adds the provider's typed session-failure title, and only a rate-limit or quota title becomes `rate_limit` or `quota_exhausted`. Context, turn and configured budget limits stay ordinary terminal failures. A -reset clock in the provider's wording, such as "resets 2am (UTC)", is used as +reset clock in the provider's wording, such as "resets 2am (UTC)", or a +calendar reset such as a weekly limit's "resets Nov 12, 6pm (UTC)", is used as the reset time. +A native run whose provider ends a turn with an exhausted allowance +(`native_provider_usage_limit`) still blocks its task with a recovery action. +When the run's last turn event is a provider-observed `turn.failed` that reports +a reset within eight days, the action says when the limit resets, and at that +time Paperclip requests the task's Retry itself, on behalf of the user who +requested the run or, for automatic wakes, the run's accountable user. That is +the same Retry the user would click: admission keeps every gate (stop proof, +saved-work recovery and action reconciliation), so it never starts work a +user's Retry could not. The request is idempotent per run, and at most three +automatic Retries follow one another without a successful run between them. +A hold whose reset is more than eight days out (for example a monthly limit) +still names its reset but leaves the Retry to the user. A hold whose reset is +unknown, more than a day in the past, or that has no accountable user keeps its +visible Retry, and its message says why. A connected chat conversation's Retry +is prepared through the chat service, as a user's is. The +reset deadline is rebuilt from the active holds at startup, so holds saved by an +earlier release are covered too. + An ingress endpoint failure can use that same bounded retry only when the environment provider returns a structured HTTP 429 or 5xx failure and the complete, encrypted controller checkpoint proves that this exact run never diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index 944eb1dcc49..b7067d220fc 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -159,6 +159,7 @@ const { registerIssueStateRecoveryWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve(), watchdogs: { reconciliationSnapshot: () => ({ revision: "startup-watchdogs:1", reconciled: true }), stop: vi.fn(async () => {}) } })), registerDeferredReceiptWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })), + registerProviderLimitResumeWork: vi.fn((_coordinator: unknown, _canRun: () => boolean, _prepareChatRetry?: () => unknown) => ({ ready: Promise.resolve() })), registerRunDispatchWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve(), retries: { reconciliationSnapshot: () => ({ revision: "startup-retries:1", reconciled: true }), stop: vi.fn(async () => {}) } })), registerScheduledRetryCancellationWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })), @@ -1048,7 +1049,14 @@ describe("startServer feedback export wiring", () => { expect(heartbeatServiceMock.registerActiveRunSilenceWork).toHaveBeenCalledTimes(enabled ? 1 : 0); expect(heartbeatServiceMock.registerRunDispatchWork).toHaveBeenCalledTimes(enabled ? 1 : 0); expect(heartbeatServiceMock.registerDeferredReceiptWork).toHaveBeenCalledTimes(enabled ? 1 : 0); + expect(heartbeatServiceMock.registerProviderLimitResumeWork).toHaveBeenCalledTimes(enabled ? 1 : 0); expect(heartbeatServiceMock.resumeQueuedRuns).not.toHaveBeenCalled(); + if (enabled) { + const [limitCoordinator, limitAdmitted, chatRetry] = heartbeatServiceMock.registerProviderLimitResumeWork.mock.calls[0]; + expect(limitCoordinator).toBe((await createAppMock.mock.results.at(-1)!.value).locals.deliveryWork); + expect(limitAdmitted()).toBe(true); + expect(typeof chatRetry).toBe("function"); + } if (enabled) { const [coordinator, admitted] = heartbeatServiceMock.registerRunDispatchWork.mock.calls[0]; const app = await createAppMock.mock.results.at(-1)!.value; diff --git a/server/src/index.ts b/server/src/index.ts index 8e785370055..dfa92d5a9ad 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -951,7 +951,8 @@ async function startServerWithDatabaseTeardown( logger.warn({ scope: cloudQualification.scope, blockerCodes: cloudQualification.blockers.map((blocker) => blocker.code) }, "Cloud configuration and package qualification has blockers"); } const idleDeadlineOwners: IdleDeadlineOwners = {}; - // createApp fills the chat Retry callback; the continuation worker reads it. + // createApp fills the chat Retry callback; the continuation and + // provider-limit Retry workers read it. const budgetResumeDelivery: BudgetResumeDelivery = {}; const app = await createApp(db as any, { idleDeadlineOwners, @@ -1771,6 +1772,10 @@ async function startServerWithDatabaseTeardown( } if (config.heartbeatSchedulerEnabled) await heartbeat.registerDeferredReceiptWork(deliveryWork, () => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped).ready; + // A connected conversation's Retry needs the chat callback createApp registered. + if (config.heartbeatSchedulerEnabled) await heartbeat.registerProviderLimitResumeWork(deliveryWork, + () => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped, + () => budgetResumeDelivery.prepareChatRetry).ready; if (config.heartbeatSchedulerEnabled) { const work = heartbeat.registerRunDispatchWork(deliveryWork, () => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped); diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index c8ecc5a9ba6..cd0da16ca72 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -58,6 +58,7 @@ export const DELIVERY_QUEUES = { scheduledRetryCancellation: "scheduled-retry-cancellation", executionFinalization: "execution-finalization", workspaceRecovery: "workspace-recovery", + providerLimitResume: "provider-limit-resume", executionReplacement: "execution-replacement", executionContinuation: "execution-continuation", executionStatus: "execution-status", diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 7724e412126..baaaace2ea1 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -9,6 +9,7 @@ import { isContextCompactionRun } from "./context-compaction.js"; import { isUnstartedTaskQuestionAnswer } from "./unstarted-task-question-answer.js"; import { isVerifiedDependencyInteractionResponse } from "./queued-interaction-response.js"; import { registerDeferredReceiptWork } from "./heartbeat/deferred-receipt-work.js"; +import { registerProviderLimitResumeWork, type PrepareChatRetry } from "./native-runtime/provider-limit-resume.js"; import { reconcileTaskChatRetryReceipts } from "./task-retry.js"; import { registerIssueRecoveryWork } from "./recovery/work.js"; import { registerRunDispatchWork } from "./heartbeat/dispatch-work.js"; @@ -4522,6 +4523,13 @@ export function heartbeatService( }, }, () => canRun() && readTaskDrain(new Date()) === null), readIdleWatchdogSchedule: taskWatchdogs.readIdleSchedule, + registerProviderLimitResumeWork: (coordinator: Parameters[1], canRun: () => boolean, + prepareChatRetry?: () => PrepareChatRetry | undefined) => + registerProviderLimitResumeWork(db, coordinator, { + resume: (companyId, requestId) => runControl.resumeTaskRetry(companyId, requestId), + schedulingSuppressed: async () => (await getSchedulingSuppression(true)).suppressed, + prepareChatRetry, + }, () => canRun() && readTaskDrain(new Date()) === null), registerIssueStateRecoveryWork: (coordinator: Parameters[1], canRun: () => boolean) => registerIssueRecoveryWork(db, coordinator, { stranded: scope => reconcileStrandedAssignedIssues({ ...scope, skipExternalConversationSettlement: true, skipRelatedRecovery: true }), diff --git a/server/src/services/heartbeat/restart-interrupted-turn.ts b/server/src/services/heartbeat/restart-interrupted-turn.ts index df89505a90e..a1458a27826 100644 --- a/server/src/services/heartbeat/restart-interrupted-turn.ts +++ b/server/src/services/heartbeat/restart-interrupted-turn.ts @@ -19,8 +19,8 @@ function isReattachmentLoss(error: string | null): boolean { return error !== null && REATTACHMENT_LOST_ERRORS.some(code => error === code || error.startsWith(`${code}: `)); } -const TURN_STARTED_EVENTS = ["turn.submitted", "turn.started", "turn.accepted"]; -const TURN_SETTLED_EVENTS = ["turn.completed", "turn.failed", "turn.cancelled", "turn.interrupted"]; +export const TURN_STARTED_EVENTS = ["turn.submitted", "turn.started", "turn.accepted"]; +export const TURN_SETTLED_EVENTS = ["turn.completed", "turn.failed", "turn.cancelled", "turn.interrupted"]; const asRecord = (value: unknown): Record => value && typeof value === "object" && !Array.isArray(value) ? value as Record : {}; @@ -29,7 +29,7 @@ const asRecord = (value: unknown): Record => * A driver that lost its event stream records `providerTerminalObserved: false` * (on the event or its stored PRP source): the real outcome is unknown, and * another turn could repeat work that already happened. */ -function isProvenTurnSettlement(event: { eventType: string; payload: unknown }) { +export function isProvenTurnSettlement(event: { eventType: string; payload: unknown }) { if (!TURN_SETTLED_EVENTS.includes(event.eventType)) return false; const payload = asRecord(event.payload); return payload.providerTerminalObserved !== false && diff --git a/server/src/services/native-runtime/native-provider-capacity.integration.test.ts b/server/src/services/native-runtime/native-provider-capacity.integration.test.ts index ee32b33e81a..d3a9a3b4167 100644 --- a/server/src/services/native-runtime/native-provider-capacity.integration.test.ts +++ b/server/src/services/native-runtime/native-provider-capacity.integration.test.ts @@ -1,8 +1,11 @@ import { randomUUID } from "node:crypto"; -import { eq } from "drizzle-orm"; +import { and, eq } from "drizzle-orm"; import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; import { agents, + agentWakeupRequests, + chatConversations, + chatEndpoints, companies, completionContracts, createDb, @@ -11,6 +14,8 @@ import { issueRecoveryActions, issues, nativeRunFinalizations, + toolApplications, + toolConnections, } from "@paperclipai/db"; import type { ControlPlanePort, @@ -31,6 +36,8 @@ import { executePaperclipNativeSession } from "./native-session-executor.js"; import { buildNativeCompletionContract } from "./completion-contracts.js"; import { PaperclipControlPlanePort } from "./paperclip-control-plane-port.js"; import { prepareNativeHeartbeatRun } from "./prepare-native-run.js"; +import { PROVIDER_LIMIT_RESUME_KEY_PREFIX, registerProviderLimitResumeWork, requestProviderLimitResumeRetry } from "./provider-limit-resume.js"; +import { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js"; describe("native provider capacity failure persistence", () => { let temporary: Awaited>; @@ -62,125 +69,211 @@ describe("native provider capacity failure persistence", () => { await temporary?.cleanup(); }); + /** Runs one native session whose provider ends its turn with `payload`. */ + async function failTurn(options: { + payload: Record; + replayed?: boolean; + responsibleUserId?: string | null; + }) { + const issueId = randomUUID(); + const runId = randomUUID(); + const [issue] = await db + .insert(issues) + .values({ + id: issueId, + companyId, + title: "Reply after a capacity 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", + contextSnapshot: { issueId }, + responsibleUserId: options.responsibleUserId ?? null, + }) + .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 completionInput = buildNativeCompletionContract(issue, { + revision: contract.revision, + }); + expect(completionInput).toEqual(contract.contractJson); + await db.insert(nativeRunFinalizations).values({ + runId, + companyId, + issueId, + phase: "observed", + }); + // A real terminal follows provider session establishment. Keep the + // recovery evidence explicit; this must not be a bootstrap retry. + await db + .update(heartbeatRuns) + .set({ + runnerProfileJson: { + sessionCheckpoint: { + providerSessionId: "capacity-test-provider-session", + providerIdentity: { threadId: "capacity-test-thread" }, + }, + }, + }) + .where(eq(heartbeatRuns.id, runId)); + + 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: "Reply to the current message.", + 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: completionInput, + }, + interactionResponses: [], + credentialBindings: [], + }; + const event: PrpEvent = { + schema: "paperclip.prp.event.v1", + runId, + normalizedSessionId: native.normalizedSessionId, + turnId: native.turnId, + sourceInstanceId: native.runnerInstanceId, + sourceEventId: "capacity-terminal", + sourceSeq: 1, + sourceKind: "runner", + eventType: "turn.failed", + schemaVersion: 1, + priority: 0, + emittedAt: new Date().toISOString(), + payload: options.payload, + }; + const openInput = { + identity: { + companyId, + issueId, + runId, + agentId, + sessionId: native.normalizedSessionId, + }, + backendKind: "mock" as const, + sourceInstanceId: native.runnerInstanceId, + }; + if (options.replayed) { + // Model a controller crash after the event commit but before its + // observational callback or failure coordinator could run. + const priorPort = new PaperclipControlPlanePort(db, { + companyId, + issueId, + runId, + agentId, + sessionId: native.normalizedSessionId, + sourceInstanceId: native.runnerInstanceId, + controlPlaneSourceInstanceId: "prior-controller", + completionContractId: contract.id, + completionContractSha256: contract.canonicalSha256, + }); + await priorPort.openRun(openInput); + expect((await priorPort.appendEvent(event)).disposition).toBe( + "committed", + ); + } + provider.execute + .mockReset() + .mockImplementation( + async (executeOptions: { controlPlane: ControlPlanePort }) => { + await executeOptions.controlPlane.openRun(openInput); + const appended = await executeOptions.controlPlane.appendEvent(event); + expect(appended.disposition).toBe( + options.replayed ? "duplicate" : "committed", + ); + // Production reports the provider terminal through the event, not by + // pattern-matching this exception's arbitrary message. + throw new Error("provider turn stopped"); + }, + ); + await expect( + executePaperclipNativeSession({ + db, + execution, + runnerInstanceId: native.runnerInstanceId, + backend: { + descriptor: async () => { + throw new Error( + "Unexpected provider descriptor call in capacity test", + ); + }, + openSession: async () => { + throw new Error( + "Unexpected provider session call in capacity test", + ); + }, + }, + }), + ).rejects.toThrow("provider turn stopped"); + expect(provider.execute).toHaveBeenCalledTimes(1); + return { runId, issueId, sourceEventId: event.sourceEventId }; + } + + async function holdFor(issueId: string) { + const [recovery] = await db + .select() + .from(issueRecoveryActions) + .where(eq(issueRecoveryActions.sourceIssueId, issueId)); + return recovery; + } + it.each([false, true])( "stops automatic retries for a durable usage terminal (replayed=%s)", async (replayed) => { - const issueId = randomUUID(); - const runId = randomUUID(); - const [issue] = await db - .insert(issues) - .values({ - id: issueId, - companyId, - title: "Reply after a capacity 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", - 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 completionInput = buildNativeCompletionContract(issue, { - revision: contract.revision, - }); - expect(completionInput).toEqual(contract.contractJson); - await db.insert(nativeRunFinalizations).values({ - runId, - companyId, - issueId, - phase: "observed", - }); - // A real terminal follows provider session establishment. Keep the - // recovery evidence explicit; this must not be a bootstrap retry. - await db - .update(heartbeatRuns) - .set({ - runnerProfileJson: { - sessionCheckpoint: { - providerSessionId: "capacity-test-provider-session", - providerIdentity: { threadId: "capacity-test-thread" }, - }, - }, - }) - .where(eq(heartbeatRuns.id, runId)); - - 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: "Reply to the current message.", - 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: completionInput, - }, - interactionResponses: [], - credentialBindings: [], - }; - const event: PrpEvent = { - schema: "paperclip.prp.event.v1", - runId, - normalizedSessionId: native.normalizedSessionId, - turnId: native.turnId, - sourceInstanceId: native.runnerInstanceId, - sourceEventId: "capacity-terminal", - sourceSeq: 1, - sourceKind: "runner", - eventType: "turn.failed", - schemaVersion: 1, - priority: 0, - emittedAt: new Date().toISOString(), + const { runId, issueId, sourceEventId } = await failTurn({ + replayed, payload: { status: "failed", error: { @@ -188,72 +281,7 @@ describe("native provider capacity failure persistence", () => { message: "Private account metadata", }, }, - }; - const openInput = { - identity: { - companyId, - issueId, - runId, - agentId, - sessionId: native.normalizedSessionId, - }, - backendKind: "mock" as const, - sourceInstanceId: native.runnerInstanceId, - }; - if (replayed) { - // Model a controller crash after the event commit but before its - // observational callback or failure coordinator could run. - const priorPort = new PaperclipControlPlanePort(db, { - companyId, - issueId, - runId, - agentId, - sessionId: native.normalizedSessionId, - sourceInstanceId: native.runnerInstanceId, - controlPlaneSourceInstanceId: "prior-controller", - completionContractId: contract.id, - completionContractSha256: contract.canonicalSha256, - }); - await priorPort.openRun(openInput); - expect((await priorPort.appendEvent(event)).disposition).toBe( - "committed", - ); - } - provider.execute - .mockReset() - .mockImplementation( - async (options: { controlPlane: ControlPlanePort }) => { - await options.controlPlane.openRun(openInput); - const appended = await options.controlPlane.appendEvent(event); - expect(appended.disposition).toBe( - replayed ? "duplicate" : "committed", - ); - // Production reports the provider terminal through the event, not by - // pattern-matching this exception's arbitrary message. - throw new Error("provider turn stopped"); - }, - ); - await expect( - executePaperclipNativeSession({ - db, - execution, - runnerInstanceId: native.runnerInstanceId, - backend: { - descriptor: async () => { - throw new Error( - "Unexpected provider descriptor call in capacity test", - ); - }, - openSession: async () => { - throw new Error( - "Unexpected provider session call in capacity test", - ); - }, - }, - }), - ).rejects.toThrow("provider turn stopped"); - - expect(provider.execute).toHaveBeenCalledTimes(1); + }); const [coordinator] = await db .select() .from(nativeRunFinalizations) @@ -271,10 +299,7 @@ describe("native provider capacity failure persistence", () => { .from(heartbeatRuns) .where(eq(heartbeatRuns.id, runId)); expect(failedRun.errorCode).toBe("native_provider_usage_limit"); - const [recovery] = await db - .select() - .from(issueRecoveryActions) - .where(eq(issueRecoveryActions.sourceIssueId, issueId)); + const recovery = await holdFor(issueId); expect(recovery).toMatchObject({ ownerType: "board", cause: "native_provider_usage_limit", @@ -294,8 +319,269 @@ describe("native provider capacity failure persistence", () => { .from(heartbeatRunEvents) .where(eq(heartbeatRunEvents.runId, runId)); expect( - events.filter((row) => row.sourceEventId === event.sourceEventId), + events.filter((row) => row.sourceEventId === sourceEventId), ).toHaveLength(1); + // No reset was reported, so nothing retries the task by itself. + await expect(requestProviderLimitResumeRetry(db, { companyId, actionId: recovery.id })) + .resolves.toEqual({ requested: false, reason: "reset_unknown" }); }, ); + + /** The ACP harness's weekly-limit terminal, as staging recorded it (PAP-1057). */ + function weeklyLimit(resetAt: Date) { + const month = resetAt.toLocaleString("en-US", { month: "short", timeZone: "UTC" }); + const hour = resetAt.getUTCHours(); + const clock = `${hour % 12 || 12}${hour < 12 ? "am" : "pm"}`; + return { + status: "failed", + error: { + code: "RUNTIME", + message: "ACP agent reported a terminal limit failure.", + providerFailure: { + schema: "paperclip.provider-failure.v1", + provider: "anthropic", + phase: "turn", + category: "quota_exhausted", + retryability: "requires_action", + message: `ACP agent reported a terminal limit failure.\nYou've hit your weekly limit · resets ${month} ${resetAt.getUTCDate()}, ${clock} (UTC)`, + }, + }, + }; + } + + it("waits for a reported usage-limit reset and then requests the task's Retry once", async () => { + const resetAt = new Date(Date.now() + 2 * 24 * 60 * 60_000); + resetAt.setUTCMinutes(0, 0, 0); + const userId = `user-${randomUUID()}`; + const { runId, issueId } = await failTurn({ payload: weeklyLimit(resetAt), responsibleUserId: userId }); + + const hold = await holdFor(issueId); + expect(hold).toMatchObject({ cause: "native_provider_usage_limit", + evidence: expect.objectContaining({ providerLimitResetAt: resetAt.toISOString() }) }); + // The hold says when work resumes, not that its session state is ambiguous. + expect(hold.nextAction).toBe(`The provider usage limit resets at ${resetAt.toISOString()}. ` + + "Paperclip will retry this task then; Retry starts it sooner if usage capacity is already restored."); + + // A hold saved by the previous release names no reset and calls the state ambiguous. + const { providerLimitResetAt: _reset, ...oldEvidence } = hold.evidence; + await db.update(issueRecoveryActions).set({ evidence: oldEvidence, nextAction: "Inspect the original provider failure " + + "and explicitly resolve the ambiguous session state; do not open a replacement provider session." }) + .where(eq(issueRecoveryActions.id, hold.id)); + + // Before the reset nothing starts; the owner keeps the reset as its deadline + // and tells the reader when work resumes. + await expect(requestProviderLimitResumeRetry(db, { companyId, actionId: hold.id })) + .resolves.toEqual({ requested: false, waitUntil: resetAt }); + expect(await holdFor(issueId)).toMatchObject({ nextAction: hold.nextAction, + evidence: expect.objectContaining({ providerLimitResetAt: resetAt.toISOString() }) }); + + // At the reset Paperclip asks for the same Retry the accountable user would. + const afterReset = new Date(resetAt.getTime() + 1000); + const outcome = await requestProviderLimitResumeRetry(db, { companyId, actionId: hold.id }, afterReset); + expect(outcome).toMatchObject({ requested: true }); + const [receipt] = await db.select().from(agentWakeupRequests).where(and( + eq(agentWakeupRequests.companyId, companyId), + eq(agentWakeupRequests.idempotencyKey, `task-retry:provider-limit-reset:${runId}`))); + expect(receipt).toMatchObject({ reason: "task_retry", requestedByActorType: "user", requestedByActorId: userId, + status: "deferred_issue_execution", payload: expect.objectContaining({ issueId, + taskRetry: expect.objectContaining({ sourceRunId: runId, actorId: userId }) }) }); + expect((await holdFor(issueId)).evidence.providerLimitResume).toMatchObject({ state: "requested", + requestId: receipt.id, resetAt: resetAt.toISOString() }); + + // A second pass never requests another Retry for the same hold. + await expect(requestProviderLimitResumeRetry(db, { companyId, actionId: hold.id }, afterReset)) + .resolves.toEqual({ requested: false, reason: "not_waiting" }); + const receipts = await db.select().from(agentWakeupRequests).where(and( + eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.reason, "task_retry"), + eq(agentWakeupRequests.agentId, agentId))); + expect(receipts.filter(row => (row.payload as Record).issueId === issueId)).toHaveLength(1); + }); + + it("starts the Retry from the hold's own deadline after a restart rebuilds it", async () => { + // A typed reset a few seconds out; the owner learns it only from durable rows. + const resetAt = new Date(Date.now() + 3000); + const payload = weeklyLimit(resetAt); + (payload.error.providerFailure as Record).retryAt = resetAt.toISOString(); + const userId = `user-${randomUUID()}`; + const { issueId } = await failTurn({ payload, responsibleUserId: userId }); + const hold = await holdFor(issueId); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, + onError: error => { errors.push(error); } }); + const resume = vi.fn(async () => undefined); + try { + const worker = registerProviderLimitResumeWork(db, coordinator, + { resume, schedulingSuppressed: async () => false }, () => true); + await worker.ready; + // Nothing starts before the reset. + expect(resume).not.toHaveBeenCalled(); + await vi.waitFor(() => expect(resume).toHaveBeenCalledTimes(1), { timeout: 20_000, interval: 200 }); + const [receipt] = await db.select().from(agentWakeupRequests).where(and( + eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.reason, "task_retry"), + eq(agentWakeupRequests.requestedByActorId, userId))); + expect(resume).toHaveBeenCalledWith(companyId, receipt.id); + expect(Date.now()).toBeGreaterThanOrEqual(resetAt.getTime()); + expect((await holdFor(issueId)).evidence.providerLimitResume).toMatchObject({ state: "requested", requestId: receipt.id }); + expect(hold.id).toBe((await holdFor(issueId)).id); + } finally { + await coordinator.stop(); + } + expect(errors).toEqual([]); + }, 30_000); + + it("leaves a visible Retry when no user is accountable for the run", async () => { + const resetAt = new Date(Date.now() + 3 * 60 * 60_000); + resetAt.setUTCMinutes(0, 0, 0); + const { issueId } = await failTurn({ payload: weeklyLimit(resetAt), responsibleUserId: null }); + const hold = await holdFor(issueId); + expect(hold.nextAction).toBe(`The provider usage limit resets at ${resetAt.toISOString()}. ` + + "Retry the task after that time, or sooner if usage capacity is restored."); + const outcome = await requestProviderLimitResumeRetry(db, { companyId, actionId: hold.id }, + new Date(resetAt.getTime() + 1000)); + expect(outcome).toEqual({ requested: false, reason: "no_accountable_user" }); + expect((await holdFor(issueId)).nextAction).toBe("Paperclip did not retry this task after the provider usage limit " + + "reset because no user is accountable for the run. Retry the task when usage capacity is available."); + }); + + it("prepares a connected chat conversation's Retry through the chat service", async () => { + const resetAt = new Date(Date.now() + 2 * 60 * 60_000); + resetAt.setUTCMinutes(0, 0, 0); + const userId = `user-${randomUUID()}`; + const { runId, issueId } = await failTurn({ payload: weeklyLimit(resetAt), responsibleUserId: userId }); + // The task belongs to a restricted external chat conversation. + const [application] = await db.insert(toolApplications).values({ companyId, name: "Slack", type: "chat" }).returning(); + const connectionId = randomUUID(); + await db.insert(toolConnections).values({ id: connectionId, companyId, applicationId: application!.id, uid: connectionId, + name: "Slack", transport: "chat_sdk", connectionPurpose: "channel" }); + const endpointId = randomUUID(); + await db.insert(chatEndpoints).values({ id: endpointId, companyId, connectionId, provider: "slack", publicId: endpointId, + assignedAgentId: agentId, status: "active", externalExecutionPolicy: "restricted" }); + await db.insert(chatConversations).values({ companyId, endpointId, issueId, externalConversationId: "channel", + externalThreadId: issueId, externalLabel: "#limits", state: "waiting" }); + const hold = await holdFor(issueId); + const chatActionId = randomUUID(); + const prepareChatRetry = vi.fn(async () => ({ actionId: chatActionId })); + const outcome = await requestProviderLimitResumeRetry(db, { companyId, actionId: hold.id, prepareChatRetry }, + new Date(resetAt.getTime() + 1000)); + expect(outcome).toMatchObject({ requested: true }); + expect(prepareChatRetry).toHaveBeenCalledWith(expect.anything(), expect.objectContaining({ + companyId, issueId, agentId, failedRunId: runId, initiatedByUserId: userId })); + const [receipt] = await db.select().from(agentWakeupRequests).where(and(eq(agentWakeupRequests.companyId, companyId), + eq(agentWakeupRequests.idempotencyKey, `task-retry:${PROVIDER_LIMIT_RESUME_KEY_PREFIX}${runId}`))); + expect(receipt.payload).toMatchObject({ taskRetry: expect.objectContaining({ chatActionId }) }); + }); + + it("replaces the retry promise on every give-up path", async () => { + const resetAt = new Date(Date.now() + 2 * 60 * 60_000); + resetAt.setUTCMinutes(0, 0, 0); + const afterReset = new Date(resetAt.getTime() + 1000); + // The failed run's terminal can no longer be read as a provider-observed reset. + const unreadable = await failTurn({ payload: weeklyLimit(resetAt), responsibleUserId: `user-${randomUUID()}` }); + await db.delete(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, unreadable.runId)); + const unreadableHold = await holdFor(unreadable.issueId); + expect(unreadableHold.nextAction).toContain("Paperclip will retry this task then"); + await expect(requestProviderLimitResumeRetry(db, { companyId, actionId: unreadableHold.id }, afterReset)) + .resolves.toEqual({ requested: false, reason: "reset_unknown" }); + expect((await holdFor(unreadable.issueId)).nextAction).toBe("Paperclip did not retry this task after the provider " + + "usage limit reset because the provider's reset could no longer be read. Retry the task when usage capacity is available."); + // The hold's run no longer failed for the usage limit. + const changed = await failTurn({ payload: weeklyLimit(resetAt), responsibleUserId: `user-${randomUUID()}` }); + await db.update(heartbeatRuns).set({ errorCode: "adapter_failed" }).where(eq(heartbeatRuns.id, changed.runId)); + const changedHold = await holdFor(changed.issueId); + await expect(requestProviderLimitResumeRetry(db, { companyId, actionId: changedHold.id }, afterReset)) + .resolves.toEqual({ requested: false, reason: "source_run_mismatch" }); + expect((await holdFor(changed.issueId)).nextAction).toBe("Paperclip did not retry this task after the provider " + + "usage limit reset because the failed run no longer matches this hold. Retry the task when usage capacity is available."); + }); + + it("stops after three automatic Retries in a row and counts again after a successful run", async () => { + const resetAt = new Date(Date.now() + 2 * 60 * 60_000); + resetAt.setUTCMinutes(0, 0, 0); + const afterReset = new Date(resetAt.getTime() + 1000); + const userId = `user-${randomUUID()}`; + const { issueId } = await failTurn({ payload: weeklyLimit(resetAt), responsibleUserId: userId }); + const hold = await holdFor(issueId); + // Earlier usage-limit Retries on this task, none followed by a success. + const earlierRetry = (task: string) => ({ companyId, agentId, source: "on_demand", triggerDetail: "manual", + reason: "task_retry", status: "completed", idempotencyKey: `task-retry:${PROVIDER_LIMIT_RESUME_KEY_PREFIX}${randomUUID()}`, + requestedByActorType: "user", requestedByActorId: userId, payload: { issueId: task }, + requestedAt: new Date(Date.now() - 60_000) }); + // Two in a row still leave this one. + await db.insert(agentWakeupRequests).values([earlierRetry(issueId), earlierRetry(issueId)]); + const [third] = await db.insert(agentWakeupRequests).values(earlierRetry(issueId)).returning(); + await expect(requestProviderLimitResumeRetry(db, { companyId, actionId: hold.id }, afterReset)) + .resolves.toEqual({ requested: false, reason: "automatic_retry_budget_used" }); + expect((await holdFor(issueId)).nextAction).toBe("Paperclip did not retry this task after the provider usage limit " + + "reset because three automatic Retries in a row did not restore usage. Retry the task when usage capacity is available."); + expect(third.id).toBeTruthy(); + + // A successful run on the task after earlier Retries restores the budget. + const second = await failTurn({ payload: weeklyLimit(resetAt), responsibleUserId: userId }); + await db.insert(agentWakeupRequests).values([earlierRetry(second.issueId), earlierRetry(second.issueId), + earlierRetry(second.issueId)]); + await db.insert(heartbeatRuns).values({ companyId, agentId, status: "succeeded", runtimeMode: "native", + nativeIssueId: second.issueId, invocationSource: "assignment", triggerDetail: "system", + contextSnapshot: { issueId: second.issueId }, finishedAt: new Date(Date.now() - 1000), + createdAt: new Date(Date.now() - 10 * 60_000) }); + const secondHold = await holdFor(second.issueId); + await expect(requestProviderLimitResumeRetry(db, { companyId, actionId: secondHold.id }, afterReset)) + .resolves.toMatchObject({ requested: true }); + }); + + it("rebuilds every waiting hold after a restart, past one page", async () => { + const resetAt = new Date(Date.now() + 2 * 60 * 60_000); + resetAt.setUTCMinutes(0, 0, 0); + const { issueId } = await failTurn({ payload: weeklyLimit(resetAt), responsibleUserId: `user-${randomUUID()}` }); + const hold = await holdFor(issueId); + // A hold saved before holds named their reset; it sorts after 600 others. + const { providerLimitResetAt: _reset, ...oldEvidence } = hold.evidence; + await db.delete(issueRecoveryActions).where(eq(issueRecoveryActions.id, hold.id)); + const otherCompanyId = randomUUID(); + await db.insert(companies).values({ id: otherCompanyId, name: "Other holds", issuePrefix: "OTH" }); + const otherIssues = Array.from({ length: 600 }, () => randomUUID()); + await db.insert(issues).values(otherIssues.map(id => ({ id, companyId: otherCompanyId, title: "Other", status: "in_progress" }))); + const { id: _id, companyId: _company, sourceIssueId: _issue, createdAt: _c, updatedAt: _u, ...template } = hold; + for (let page = 0; page < 6; page++) { + await db.insert(issueRecoveryActions).values(Array.from({ length: 100 }, (_, index) => ({ ...template, + companyId: otherCompanyId, sourceIssueId: otherIssues[page * 100 + index]!, + id: `00000000-0000-4000-8000-${String(page * 100 + index).padStart(12, "0")}`, + evidence: { ...oldEvidence, runId: randomUUID() } }))); + } + await db.insert(issueRecoveryActions).values({ ...hold, id: `ffffffff-ffff-4fff-bfff-${randomUUID().slice(-12)}`, + evidence: oldEvidence }); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, + onError: error => { errors.push(error); } }); + try { + const worker = registerProviderLimitResumeWork(db, coordinator, + { resume: vi.fn(async () => undefined), schedulingSuppressed: async () => false }, () => true); + await worker.ready; + // The hold past the first page reads its reset and records it. + await vi.waitFor(async () => expect((await holdFor(issueId)).evidence.providerLimitResetAt).toBe(resetAt.toISOString()), + { timeout: 25_000, interval: 250 }); + } finally { + await coordinator.stop(); + } + expect(errors).toEqual([]); + }, 40_000); + + it("reports a reset beyond the automatic wait without retrying by itself", async () => { + // A monthly limit, typed by the provider, three weeks out. + const resetAt = new Date(Date.now() + 21 * 24 * 60 * 60_000); + resetAt.setUTCMinutes(0, 0, 0); + const payload = weeklyLimit(resetAt); + (payload.error.providerFailure as Record).retryAt = resetAt.toISOString(); + const userId = `user-${randomUUID()}`; + const { issueId } = await failTurn({ payload, responsibleUserId: userId }); + const hold = await holdFor(issueId); + expect(hold.nextAction).toBe(`The provider usage limit resets at ${resetAt.toISOString()}. ` + + "Retry the task after that time, or sooner if usage capacity is restored."); + expect(hold.evidence).toMatchObject({ providerLimitResetAt: resetAt.toISOString() }); + await expect(requestProviderLimitResumeRetry(db, { companyId, actionId: hold.id }, + new Date(resetAt.getTime() + 1000))).resolves.toEqual({ requested: false, reason: "reset_beyond_wait" }); + expect((await holdFor(issueId)).nextAction).toBe(hold.nextAction); + const receipts = await db.select().from(agentWakeupRequests).where(and(eq(agentWakeupRequests.companyId, companyId), + eq(agentWakeupRequests.requestedByActorId, userId))); + expect(receipts).toEqual([]); + }); }); diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index cc2d29953b0..7ed7a9692c6 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -12155,6 +12155,18 @@ async function executePaperclipNativeSessionAttempt( .from(heartbeatRuns).where(eq(heartbeatRuns.id, input.execution.binding.runId)) .then(rows => rows.some(row => nativeCheckpointRecordsCompletedTurn(row.runnerProfileJson, input.execution.binding.runId))) .catch(() => false); + // An exhausted provider allowance is not ambiguous session state: the + // provider ended its turn. Say when it resets and whether Paperclip will + // retry then. A failed read only loses that detail. + const providerLimit = sourceFailureCode === "native_provider_usage_limit" && phase === "terminal_failure" + ? await import("./provider-limit-resume.js").then(async ({ planProviderLimitResume, providerLimitNextAction }) => { + const plan = await planProviderLimitResume(input.db, input.execution.binding.runId); + return { plan, nextAction: providerLimitNextAction(plan) }; + }).catch((failure: unknown) => { + logger.warn({ err: failure, runId: input.execution.binding.runId }, "provider limit reset unavailable"); + return null; + }) + : null; const recoveryProjection = nativeSessionRecoveryProjection({ phase, failureCode, @@ -12222,9 +12234,12 @@ async function executePaperclipNativeSessionAttempt( providerEventsExist: recoveryEvidence.providerEventsExist, checkpointExists: recoveryEvidence.checkpointExists, ...(providerCooldownAt ? { providerCooldownAt: providerCooldownAt.toISOString() } : {}), + ...(providerLimit?.plan.resetAt ? { providerLimitResetAt: providerLimit.plan.resetAt.toISOString() } : {}), recoveryOwner: recoveryProjection.recoveryOwner, nextAction: - sourceFailureCode === "native_permission_declined" + providerLimit + ? providerLimit.nextAction + : sourceFailureCode === "native_permission_declined" ? "Provider permission was declined. Inspect the task and give explicit direction before starting further provider work. Automatic recovery is stopped." : sourceFailureCode === "native_provider_approval_required" ? "Approval required. Review the operation and update the agent's permission setting before retrying. This runner has no interactive approval handler." @@ -12458,6 +12473,9 @@ async function executePaperclipNativeSessionAttempt( .where(eq(issues.id, input.execution.binding.issueId)); } if (!stillOwnsTask) return; + // The hold's reset deadline belongs to the provider-limit owner. + if (providerLimit?.plan.automatic) await notifyDeliveryWork(tx, DELIVERY_QUEUES.providerLimitResume, + JSON.stringify({ companyId: input.execution.binding.companyId })); await issueRecoveryActionService( tx as unknown as Db, ).upsertSourceScoped({ @@ -12485,9 +12503,12 @@ async function executePaperclipNativeSessionAttempt( providerSessionEstablished: recoveryEvidence.providerSessionEstablished, ...(providerCooldownAt ? { providerCooldownAt: providerCooldownAt.toISOString() } : {}), + ...(providerLimit?.plan.resetAt ? { providerLimitResetAt: providerLimit.plan.resetAt.toISOString() } : {}), }, nextAction: - sourceFailureCode === "native_permission_declined" + providerLimit + ? providerLimit.nextAction + : sourceFailureCode === "native_permission_declined" ? "Provider permission was declined. Inspect the task and give explicit direction before starting further provider work. Automatic recovery is stopped." : sourceFailureCode === "native_provider_approval_required" ? "Approval required. Review the operation and update the agent's permission setting before retrying. This runner has no interactive approval handler." diff --git a/server/src/services/native-runtime/provider-limit-resume.ts b/server/src/services/native-runtime/provider-limit-resume.ts new file mode 100644 index 00000000000..82497b620ba --- /dev/null +++ b/server/src/services/native-runtime/provider-limit-resume.ts @@ -0,0 +1,284 @@ +import { and, asc, desc, eq, gt, inArray, isNull, like, sql } from "drizzle-orm"; +import { + agentWakeupRequests, + heartbeatRunEvents, + heartbeatRuns, + issueRecoveryActions, + issues, + type Db, +} from "@paperclipai/db"; +import { readProviderFailure } from "@paperclipai/shared"; +import { HttpError } from "../../errors.js"; +import { logger } from "../../middleware/logger.js"; +import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js"; +import { DELIVERY_QUEUES } from "../delivery-work-notifications.js"; +import { isProvenTurnSettlement, TURN_SETTLED_EVENTS, TURN_STARTED_EVENTS } from "../heartbeat/restart-interrupted-turn.js"; +import { parseProviderQuotaClockReset } from "../provider-quota-reset.js"; +import { registerScopedDeadlineWork } from "../scoped-deadline-work.js"; +import { requestTaskRetry } from "../task-retry.js"; + +/** The chat service's Retry callback, which a connected conversation's Retry needs. */ +export type PrepareChatRetry = NonNullable[0]["prepareChatRetry"]>; + +/** + * A native run whose provider ended its turn because the account's usage limit + * ran out. When the provider reported when the limit resets, Paperclip asks for + * the task's Retry itself at that time, as the accountable user would. Retry + * admission keeps every gate (stop proof, saved work, action reconciliation), + * so this never starts work a user's own Retry could not. + */ +export const NATIVE_PROVIDER_USAGE_LIMIT = "native_provider_usage_limit"; +/** Longer than a weekly limit's reset, so the reset is the only wait. */ +export const PROVIDER_LIMIT_RESUME_MAX_WAIT_MS = 8 * 24 * 60 * 60_000; +/** A reset older than this belongs to an incident nobody is waiting on. */ +const STALE_RESET_MS = 24 * 60 * 60_000; +/** Automatic Retries in a row without a successful run between them. */ +const MAX_AUTOMATIC_RESUMES = 3; +export const PROVIDER_LIMIT_RESUME_KEY_PREFIX = "provider-limit-reset:"; +const LIMIT_CATEGORIES = ["rate_limit", "quota_exhausted"]; + +const asRecord = (value: unknown): Record => + value && typeof value === "object" && !Array.isArray(value) ? value as Record : {}; + +/** The provider-reported failure on a settled turn event, raw or PRP-wrapped. */ +function providerFailureOf(payload: unknown) { + const direct = asRecord(asRecord(payload).error).providerFailure; + const wrapped = asRecord(asRecord(asRecord(asRecord(payload).prpEvent).payload).error).providerFailure; + return readProviderFailure(direct) ?? readProviderFailure(wrapped); +} + +export type ProviderLimitReset = { + resetAt: Date; observedAt: Date; + /** False for a reset further out than Paperclip waits by itself (a monthly limit). */ + withinAutomaticWait: boolean; +}; + +/** + * When the provider said the usage limit resets, from the run's own durable + * events. Null unless the last turn event is a provider-observed settlement + * reporting an exhausted allowance with a reset after it. + */ +export async function readNativeUsageLimitReset( + db: Pick, + run: { id: string; companyId: string }, +): Promise { + const [lastTurn] = await db.select({ eventType: heartbeatRunEvents.eventType, payload: heartbeatRunEvents.payload, + createdAt: heartbeatRunEvents.createdAt }) + .from(heartbeatRunEvents).where(and( + eq(heartbeatRunEvents.companyId, run.companyId), eq(heartbeatRunEvents.runId, run.id), + inArray(heartbeatRunEvents.eventType, [...TURN_STARTED_EVENTS, ...TURN_SETTLED_EVENTS]), + )).orderBy(desc(heartbeatRunEvents.seq)).limit(1); + // An unsettled or unobserved turn may still be working: its outcome is unknown. + if (!lastTurn || lastTurn.eventType !== "turn.failed" || !isProvenTurnSettlement(lastTurn)) return null; + const failure = providerFailureOf(lastTurn.payload); + if (!failure || !LIMIT_CATEGORIES.includes(failure.category)) return null; + const observedAt = lastTurn.createdAt; + const typed = failure.retryAt ? new Date(failure.retryAt) : null; + const resetAt = typed && Number.isFinite(typed.getTime()) ? typed : parseProviderQuotaClockReset(failure.message, observedAt); + if (!resetAt || resetAt.getTime() <= observedAt.getTime()) return null; + return { resetAt, observedAt, + withinAutomaticWait: resetAt.getTime() - observedAt.getTime() <= PROVIDER_LIMIT_RESUME_MAX_WAIT_MS }; +} + +/** The user whose authority Retry uses: the requester, else the run's accountable user. */ +async function accountableUser(db: Pick, run: typeof heartbeatRuns.$inferSelect) { + const [wake] = run.wakeupRequestId ? await db.select({ type: agentWakeupRequests.requestedByActorType, + id: agentWakeupRequests.requestedByActorId }).from(agentWakeupRequests).where(and( + eq(agentWakeupRequests.companyId, run.companyId), eq(agentWakeupRequests.id, run.wakeupRequestId))) : []; + if (wake?.type === "user" && wake.id) return wake.id; + return run.responsibleUserId ?? null; +} + +export type ProviderLimitResumePlan = + | { automatic: true; resetAt: Date; actorId: string } + | { automatic: false; resetAt: Date | null }; + +/** Whether this failed run's task continues by itself once the limit resets. */ +export async function planProviderLimitResume(db: Pick, runId: string): Promise { + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)); + if (!run) return { automatic: false, resetAt: null }; + const reset = await readNativeUsageLimitReset(db, run); + if (!reset) return { automatic: false, resetAt: null }; + // A reset beyond the bounded wait is still reported; only the automatic Retry is bounded. + const actorId = reset.withinAutomaticWait ? await accountableUser(db, run) : null; + return actorId ? { automatic: true, resetAt: reset.resetAt, actorId } : { automatic: false, resetAt: reset.resetAt }; +} + +export function providerLimitNextAction(plan: ProviderLimitResumePlan) { + if (plan.automatic) { + return `The provider usage limit resets at ${plan.resetAt.toISOString()}. Paperclip will retry this task then; Retry starts it sooner if usage capacity is already restored.`; + } + return plan.resetAt + ? `The provider usage limit resets at ${plan.resetAt.toISOString()}. Retry the task after that time, or sooner if usage capacity is restored.` + : "Restore model provider usage capacity, then retry the task. The provider did not report when its usage limit resets."; +} + +type Outcome = + | { requested: true; requestId: string } + | { requested: false; waitUntil: Date } + | { requested: false; reason: string }; + +async function settle(db: Db, actionId: string, state: Record, nextAction?: string) { + await db.update(issueRecoveryActions).set({ + evidence: sql`coalesce(${issueRecoveryActions.evidence}, '{}'::jsonb) || jsonb_build_object('providerLimitResume', ${JSON.stringify(state)}::jsonb)`, + ...(nextAction ? { nextAction } : {}), + updatedAt: new Date(), + }).where(and(eq(issueRecoveryActions.id, actionId), sql`${issueRecoveryActions.evidence}->'providerLimitResume' is null`)); +} + +/** Why Paperclip did not retry after a reset it announced; the Retry stays. + * Every give-up reason is listed, so no give-up leaves the promise standing. */ +const NOT_RETRIED = { + source_run_mismatch: "the failed run no longer matches this hold", + reset_unknown: "the provider's reset could no longer be read", + task_changed: "the task changed", + no_accountable_user: "no user is accountable for the run", + automatic_retry_budget_used: "three automatic Retries in a row did not restore usage", + retry_stale: "a newer run exists", + retry_refused: "Retry was not allowed for the accountable user", + reset_stale: "the reset passed more than a day ago", +} as const; +type NotRetriedReason = keyof typeof NOT_RETRIED; + +/** + * Requests the task's Retry for one usage-limit hold once its reset has + * passed. Idempotent per run: the Retry key and the recorded outcome both + * prevent a second request. + */ +export async function requestProviderLimitResumeRetry(db: Db, input: { + companyId: string; actionId: string; prepareChatRetry?: PrepareChatRetry; +}, now = new Date()): Promise { + const [action] = await db.select().from(issueRecoveryActions).where(and( + eq(issueRecoveryActions.companyId, input.companyId), eq(issueRecoveryActions.id, input.actionId))); + if (!action || action.status !== "active" || action.kind !== "active_run_watchdog" || action.cause !== NATIVE_PROVIDER_USAGE_LIMIT + || action.evidence?.providerLimitResume) return { requested: false, reason: "not_waiting" }; + const runId = typeof action.evidence?.runId === "string" ? action.evidence.runId : null; + const [run] = runId ? await db.select().from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, action.companyId), sql`${heartbeatRuns.id}::text = ${runId}`)) : []; + const done = async (reason: NotRetriedReason, extra: Record = {}) => { + // Replace a promise to retry with what the user can do now. + const nextAction = action.evidence?.providerLimitResetAt + ? `Paperclip did not retry this task after the provider usage limit reset because ${NOT_RETRIED[reason]}. Retry the task when usage capacity is available.` + : undefined; + await settle(db, action.id, { state: "not_requested", reason, decidedAt: now.toISOString(), ...extra }, nextAction); + return { requested: false as const, reason }; + }; + // The failure that created the hold may still be settling its run. + if (run && ["queued", "running"].includes(run.status)) return { requested: false, waitUntil: new Date(now.getTime() + 60_000) }; + if (!run || run.runtimeMode !== "native" || run.status !== "failed" || run.errorCode !== NATIVE_PROVIDER_USAGE_LIMIT) + return done("source_run_mismatch"); + const reset = await readNativeUsageLimitReset(db, run); + if (!reset) return done("reset_unknown"); + if (!reset.withinAutomaticWait) { + // Too far out to wait by itself: the hold names the reset and leaves the + // Retry to the user; it never promised an automatic one. + await db.update(issueRecoveryActions).set({ + nextAction: providerLimitNextAction({ automatic: false, resetAt: reset.resetAt }), + evidence: sql`coalesce(${issueRecoveryActions.evidence}, '{}'::jsonb) || jsonb_build_object('providerLimitResetAt', ${reset.resetAt.toISOString()}::text)`, + updatedAt: now, + }).where(and(eq(issueRecoveryActions.id, action.id), eq(issueRecoveryActions.status, "active"))); + await settle(db, action.id, { state: "not_requested", reason: "reset_beyond_wait", resetAt: reset.resetAt.toISOString(), + decidedAt: now.toISOString() }); + return { requested: false, reason: "reset_beyond_wait" }; + } + if (reset.resetAt.getTime() > now.getTime()) { + // A hold saved by an earlier release names no reset; tell its reader when work resumes. + if (!action.evidence?.providerLimitResetAt) { + const actorId = await accountableUser(db, run); + await db.update(issueRecoveryActions).set({ + nextAction: providerLimitNextAction(actorId + ? { automatic: true, resetAt: reset.resetAt, actorId } : { automatic: false, resetAt: reset.resetAt }), + evidence: sql`coalesce(${issueRecoveryActions.evidence}, '{}'::jsonb) || jsonb_build_object('providerLimitResetAt', ${reset.resetAt.toISOString()}::text)`, + updatedAt: now, + }).where(and(eq(issueRecoveryActions.id, action.id), eq(issueRecoveryActions.status, "active"))); + } + return { requested: false, waitUntil: reset.resetAt }; + } + if (now.getTime() - reset.resetAt.getTime() > STALE_RESET_MS) return done("reset_stale", { resetAt: reset.resetAt.toISOString() }); + const issueId = action.sourceIssueId; + const [issue] = await db.select({ status: issues.status, assigneeAgentId: issues.assigneeAgentId }).from(issues) + .where(and(eq(issues.companyId, run.companyId), eq(issues.id, issueId))); + if (!issue || issue.assigneeAgentId !== run.agentId || ["done", "cancelled"].includes(issue.status)) return done("task_changed"); + const actorId = await accountableUser(db, run); + if (!actorId) return done("no_accountable_user"); + // Bound a provider that keeps reporting resets that do not restore capacity. + const [lastSuccess] = await db.select({ finishedAt: heartbeatRuns.finishedAt }).from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, run.companyId), eq(heartbeatRuns.agentId, run.agentId), eq(heartbeatRuns.status, "succeeded"), + sql`coalesce(${heartbeatRuns.nativeIssueId}::text, ${heartbeatRuns.issueId}::text, ${heartbeatRuns.contextSnapshot}->>'issueId') = ${issueId}`, + )).orderBy(desc(heartbeatRuns.finishedAt)).limit(1); + const earlier = await db.select({ id: agentWakeupRequests.id }).from(agentWakeupRequests).where(and( + eq(agentWakeupRequests.companyId, run.companyId), eq(agentWakeupRequests.agentId, run.agentId), + like(agentWakeupRequests.idempotencyKey, `task-retry:${PROVIDER_LIMIT_RESUME_KEY_PREFIX}%`), + sql`${agentWakeupRequests.payload}->>'issueId' = ${issueId}`, + lastSuccess?.finishedAt ? gt(agentWakeupRequests.requestedAt, lastSuccess.finishedAt) : undefined, + )).limit(MAX_AUTOMATIC_RESUMES); + if (earlier.length >= MAX_AUTOMATIC_RESUMES) return done("automatic_retry_budget_used"); + try { + const receipt = await requestTaskRetry({ db, companyId: run.companyId, issueId, actorId, + actorSource: actorId === "local-board" ? "local_implicit" : "session", + idempotencyKey: `${PROVIDER_LIMIT_RESUME_KEY_PREFIX}${run.id}`, observedRunId: run.id, + initiatedBy: "provider_limit_reset", prepareChatRetry: input.prepareChatRetry }); + if (!receipt.accepted || !receipt.requestId) return done("retry_stale"); + await settle(db, action.id, { state: "requested", requestId: receipt.requestId, resetAt: reset.resetAt.toISOString(), + requestedAt: now.toISOString() }); + return { requested: true, requestId: receipt.requestId }; + } catch (error) { + // Authority or conversation checks refused it; the visible Retry remains. + if (!(error instanceof HttpError)) throw error; + logger.info({ runId: run.id, status: error.status }, "provider limit reset left for an explicit Retry"); + return done("retry_refused", { status: error.status }); + } +} + +type Scope = { companyId: string }; +const LOAD_PAGE_SIZE = 500; +type Entry = { id: string; companyId: string; dueAt: number }; + +/** + * Deadline owner for usage-limit holds: one entry per hold, due at its + * provider-reported reset. Holds are the durable truth; a new hold notifies + * this queue, and startup reconstructs every waiting hold. + */ +export function registerProviderLimitResumeWork( + db: Db, + coordinator: ReturnType, + owner: { resume(companyId: string, requestId: string): Promise; schedulingSuppressed(): Promise; + /** Read when a Retry is requested: the chat service registers it after startup. */ + prepareChatRetry?: () => PrepareChatRetry | undefined }, + canRun: () => boolean, +) { + const cursors = new Map(); + return registerScopedDeadlineWork(coordinator, { + queue: DELIVERY_QUEUES.providerLimitResume, + decodeScope: hint => hint === "*" ? undefined : JSON.parse(hint) as Scope, + matches: (entry, scope) => !scope || entry.companyId === scope.companyId, + load: async scope => { + // One bounded page per pass, resumed from a stable cursor until every + // waiting hold is loaded: a hold left out would never retry. + const key = scope ? JSON.stringify(scope) : "*"; + const cursor = cursors.get(key); + const rows = await db.select({ id: issueRecoveryActions.id, companyId: issueRecoveryActions.companyId }) + .from(issueRecoveryActions).where(and( + scope ? eq(issueRecoveryActions.companyId, scope.companyId) : undefined, + eq(issueRecoveryActions.status, "active"), eq(issueRecoveryActions.kind, "active_run_watchdog"), + eq(issueRecoveryActions.cause, NATIVE_PROVIDER_USAGE_LIMIT), + sql`${issueRecoveryActions.evidence}->'providerLimitResume' is null`, + isNull(issueRecoveryActions.resolvedAt), + cursor ? gt(issueRecoveryActions.id, cursor) : undefined, + )).orderBy(asc(issueRecoveryActions.id)).limit(LOAD_PAGE_SIZE); + const complete = rows.length < LOAD_PAGE_SIZE; + if (complete) cursors.delete(key); else cursors.set(key, rows.at(-1)!.id); + // check() reads each hold's reset and returns its real deadline. + return { rows: rows.map(row => ({ ...row, dueAt: 0 })), complete }; + }, + check: async entry => { + if (await owner.schedulingSuppressed()) return { ...entry, dueAt: Date.now() + 5 * 60_000 }; + const outcome = await requestProviderLimitResumeRetry(db, { companyId: entry.companyId, actionId: entry.id, + prepareChatRetry: owner.prepareChatRetry?.() }); + if ("waitUntil" in outcome) return { ...entry, dueAt: outcome.waitUntil.getTime() }; + if (outcome.requested) await owner.resume(entry.companyId, outcome.requestId).catch((error: unknown) => + logger.warn({ err: error, actionId: entry.id }, "provider limit Retry remains queued")); + return null; + }, + }, canRun); +} diff --git a/server/src/services/provider-quota-reset.test.ts b/server/src/services/provider-quota-reset.test.ts new file mode 100644 index 00000000000..f3ba512145b --- /dev/null +++ b/server/src/services/provider-quota-reset.test.ts @@ -0,0 +1,43 @@ +import { describe, expect, it } from "vitest"; +import { parseProviderQuotaClockReset } from "./provider-quota-reset.js"; + +describe("parseProviderQuotaClockReset", () => { + it("reads a weekly limit's calendar reset, as Claude's ACP agent reports it", () => { + // Staging run c086fa26 (PAP-1057), 2026-11-10: the reset is two days out. + const now = new Date("2026-11-10T20:06:33.000Z"); + const message = "ACP agent reported a terminal limit failure.\nYou've hit your weekly limit · resets Nov 12, 6pm (UTC)"; + expect(parseProviderQuotaClockReset(message, now)).toEqual(new Date("2026-11-12T18:00:00.000Z")); + }); + + it("reads a calendar reset in a named time zone, with minutes and an ordinal", () => { + const now = new Date("2026-07-15T20:00:00.000Z"); + expect(parseProviderQuotaClockReset("resets Jul 17th at 4:30 PM (America/Chicago)", now)) + .toEqual(new Date("2026-07-17T21:30:00.000Z")); + expect(parseProviderQuotaClockReset("Resets on August 2, 9am (UTC)", now)) + .toEqual(new Date("2026-08-02T09:00:00.000Z")); + }); + + it("rolls a calendar reset that already passed into next year, and rejects impossible dates", () => { + expect(parseProviderQuotaClockReset("resets Jan 2, 6pm (UTC)", new Date("2026-12-30T10:00:00.000Z"))) + .toEqual(new Date("2027-01-02T18:00:00.000Z")); + expect(parseProviderQuotaClockReset("resets Nov 31, 6pm (UTC)", new Date("2026-11-10T10:00:00.000Z"))).toBeNull(); + }); + + it("reads a calendar reset that already passed as expired, never as a year away", () => { + // Recovery rereads an unchanged failure days after its reset. + expect(parseProviderQuotaClockReset("resets Nov 12, 6pm (UTC)", new Date("2026-11-14T10:00:00.000Z"))) + .toEqual(new Date("2026-11-12T18:00:00.000Z")); + expect(parseProviderQuotaClockReset("resets Dec 30, 6pm (UTC)", new Date("2027-01-03T10:00:00.000Z"))) + .toEqual(new Date("2026-12-30T18:00:00.000Z")); + }); + + it("keeps reading a clock-only reset as the next matching time", () => { + const now = new Date("2026-07-15T20:00:00.000Z"); + expect(parseProviderQuotaClockReset("You've hit your session limit · resets 2am (UTC)", now)) + .toEqual(new Date("2026-07-16T02:00:00.000Z")); + expect(parseProviderQuotaClockReset("Try again at 4:30 PM (America/Chicago).", now)) + .toEqual(new Date("2026-07-15T21:30:00.000Z")); + expect(parseProviderQuotaClockReset("resets 9pm", now)).toEqual(new Date("2026-07-15T21:00:00.000Z")); + expect(parseProviderQuotaClockReset("Provider quota exceeded for this model.", now)).toBeNull(); + }); +}); diff --git a/server/src/services/provider-quota-reset.ts b/server/src/services/provider-quota-reset.ts index b2d62613b93..25e275ce9e0 100644 --- a/server/src/services/provider-quota-reset.ts +++ b/server/src/services/provider-quota-reset.ts @@ -1,17 +1,29 @@ +const MONTHS = ["jan", "feb", "mar", "apr", "may", "jun", "jul", "aug", "sep", "oct", "nov", "dec"]; +const RESET_PATTERN = new RegExp( + "(?:try again at|resets?(?:\\s+(?:at|on))?)\\s+" + + // Optional calendar date, as in "resets Nov 12, 6pm (UTC)" for a weekly limit. + `(?:(${MONTHS.join("|")})[a-z]*\\.?\\s+(\\d{1,2})(?:st|nd|rd|th)?,?\\s+(?:at\\s+)?)?` + + "(\\d{1,2})(?::(\\d{2}))?\\s*(?:([ap])\\.?\\s*m\\.?)?(?:\\s*\\(([^)]+)\\)|\\s+([A-Z]{2,5}))?", + "i", +); + /** - * Reads a provider's "try again at 4:30 PM (America/Chicago)" or - * "resets 2am (UTC)" wording as the next matching wall-clock instant after - * `now`. A clock without a time zone is read as UTC. + * Reads a provider's "try again at 4:30 PM (America/Chicago)", "resets 2am + * (UTC)" or "resets Nov 12, 6pm (UTC)" wording. A clock alone is the next + * matching wall-clock instant after `now`; a calendar date is its occurrence + * nearest `now`, which is in the past when the reset already passed. A clock + * without a time zone is read as UTC. */ export function parseProviderQuotaClockReset(error: string, now: Date) { - const match = error.match( - /(?:try again at|resets?(?:\s+at)?)\s+(\d{1,2})(?::(\d{2}))?\s*(?:([ap])\.?\s*m\.?)?(?:\s*\(([^)]+)\)|\s+([A-Z]{2,5}))?/i, - ); + const match = error.match(RESET_PATTERN); if (!match) return null; - const hourValue = Number.parseInt(match[1] ?? "", 10); - const minute = Number.parseInt(match[2] ?? "0", 10); - const meridiem = (match[3] ?? "").toLowerCase(); + const monthIndex = match[1] ? MONTHS.indexOf(match[1].slice(0, 3).toLowerCase()) : null; + const dayOfMonth = match[2] ? Number.parseInt(match[2], 10) : null; + if (monthIndex !== null && (monthIndex < 0 || !dayOfMonth || dayOfMonth < 1 || dayOfMonth > 31)) return null; + const hourValue = Number.parseInt(match[3] ?? "", 10); + const minute = Number.parseInt(match[4] ?? "0", 10); + const meridiem = (match[5] ?? "").toLowerCase(); if (!Number.isInteger(hourValue)) return null; if ( meridiem ? hourValue < 1 || hourValue > 12 : hourValue < 0 || hourValue > 23 @@ -21,14 +33,7 @@ export function parseProviderQuotaClockReset(error: string, now: Date) { let hour = meridiem ? hourValue % 12 : hourValue; if (meridiem === "p") hour += 12; - const timeZone = (match[4] ?? match[5])?.trim(); - if (!timeZone) { - const retryAt = new Date(now); - retryAt.setUTCHours(hour, minute, 0, 0); - if (retryAt.getTime() <= now.getTime()) - retryAt.setUTCDate(retryAt.getUTCDate() + 1); - return retryAt; - } + const timeZone = (match[6] ?? match[7])?.trim() || "UTC"; try { const wallClock = (date: Date) => @@ -46,18 +51,13 @@ export function parseProviderQuotaClockReset(error: string, now: Date) { .map((part) => [part.type, part.value]), ); const nowParts = wallClock(now); - const buildRetryAt = (dayOffset: number) => { - const targetDay = new Date( - Date.UTC( - Number(nowParts.year), - Number(nowParts.month) - 1, - Number(nowParts.day) + dayOffset, - hour, - minute, - ), - ); - let candidate = targetDay; - const targetMs = targetDay.getTime(); + // The instant whose wall clock in `timeZone` reads this date and time. + const instant = (year: number, month: number, day: number) => { + const target = new Date(Date.UTC(year, month, day, hour, minute)); + // Reject dates that overflow, such as Nov 31. + if (monthIndex !== null && target.getUTCDate() !== day) return null; + let candidate = target; + const targetMs = target.getTime(); for (let attempt = 0; attempt < 4; attempt += 1) { const actual = wallClock(candidate); const actualMs = Date.UTC( @@ -73,8 +73,23 @@ export function parseProviderQuotaClockReset(error: string, now: Date) { } return candidate; }; - const sameDay = buildRetryAt(0); - return sameDay.getTime() > now.getTime() ? sameDay : buildRetryAt(1); + const year = Number(nowParts.year); + if (monthIndex !== null && dayOfMonth !== null) { + // The occurrence of that date nearest `now`: "resets Jan 2" read on + // Dec 30 is next year's. A date read after it passed stays in the past, + // so a caller reading an old message sees an expired reset rather than + // one a year away. + const candidates = [year - 1, year, year + 1] + .map((candidateYear) => instant(candidateYear, monthIndex, dayOfMonth)) + .filter((candidate): candidate is Date => candidate !== null); + if (!candidates.length) return null; + return candidates.reduce((nearest, candidate) => + Math.abs(candidate.getTime() - now.getTime()) < Math.abs(nearest.getTime() - now.getTime()) ? candidate : nearest); + } + const month = Number(nowParts.month) - 1; + const day = Number(nowParts.day); + const sameDay = instant(year, month, day)!; + return sameDay.getTime() > now.getTime() ? sameDay : instant(year, month, day + 1); } catch { return null; } diff --git a/server/src/services/recovery/provider-failure-classification.test.ts b/server/src/services/recovery/provider-failure-classification.test.ts index ce7da72da9f..d072b3d93a5 100644 --- a/server/src/services/recovery/provider-failure-classification.test.ts +++ b/server/src/services/recovery/provider-failure-classification.test.ts @@ -111,6 +111,17 @@ describe("classifyAdapterFailureForRecovery", () => { }); }); + it("releases the wait when an unchanged weekly-limit failure is reread after its calendar reset", () => { + // Read two days after "resets Nov 12": the wait is over, not a year away. + const now = new Date("2026-11-14T10:00:00.000Z"); + const classification = classifyAdapterFailureForRecovery({ + errorCode: "adapter_failed", + error: "You've hit your weekly limit · resets Nov 12, 6pm (UTC)", + resultJson: null, + }, now); + expect(classification).toEqual({ kind: "provider_quota", retryAt: now, parsedResetTime: true }); + }); + it("uses the default recovery backoff when quota reset time is absent", () => { const now = new Date("2026-07-15T20:00:00.000Z"); const classification = classifyAdapterFailureForRecovery({ diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index 4cadaea1b47..4456079b96f 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -676,7 +676,8 @@ export function classifyAdapterFailureForRecovery( if (parsedClockReset) { return { kind: "provider_quota", - retryAt: parsedClockReset, + // A calendar reset read after it passed has released the wait. + retryAt: parsedClockReset > now ? parsedClockReset : now, parsedResetTime: true, }; } diff --git a/server/src/services/task-retry.ts b/server/src/services/task-retry.ts index 72050465f52..4525c5986c7 100644 --- a/server/src/services/task-retry.ts +++ b/server/src/services/task-retry.ts @@ -224,12 +224,20 @@ export async function retireSupersededTaskRetry(db: Db, input: { type TaskRetryRequest = { db: Db; companyId: string; issueId: string; actorId: string; idempotencyKey: string; observedRunId: string | null; actorSource?: string; - /** Paperclip requests the user's Retry itself after a restart lost an unstarted turn. */ - initiatedBy?: "restart_recovery"; + /** Paperclip requests the user's Retry itself: after a restart lost an + * unstarted turn, or once a provider usage limit it waited on has reset. */ + initiatedBy?: "restart_recovery" | "provider_limit_reset"; prepareChatRetry?: (tx: Db, scope: { companyId: string; issueId: string; agentId: string; failedRunId: string; initiatedByUserId: string; retryIntentId?: string }) => Promise<{ actionId: string }>; }; +/** Activity actor for a Retry: the user, or the Paperclip hook acting for them. */ +function automaticRetryActor(input: Pick) { + if (!input.initiatedBy) return { actorType: "user" as const, actorId: input.actorId }; + return { actorType: "system" as const, + actorId: input.initiatedBy === "restart_recovery" ? "restart-recovery" : "provider-limit-reset" }; +} + const TASK_RETRY_CONTENTION_ATTEMPTS = 10; const TASK_RETRY_CONTENTION_DELAY_MS = 100; @@ -324,13 +332,11 @@ function admitTaskRetry(input: TaskRetryRequest) { source: "on_demand", triggerDetail: "manual", reason: "task_retry", idempotencyKey: key, requestedByActorType: "user", requestedByActorId: input.actorId, ...settled, payload: { issueId: issue.id, taskRetry: finalizeIntent } }); - const restartRecovery = input.initiatedBy === "restart_recovery"; - await persistActivity(db, { companyId: issue.companyId, actorType: restartRecovery ? "system" : "user", - actorId: restartRecovery ? "restart-recovery" : input.actorId, + await persistActivity(db, { companyId: issue.companyId, ...automaticRetryActor(input), action: "issue.retry_requested", entityType: "issue", entityId: issue.id, runId: latest.id, details: { requestId: canonicalId ?? finalizeId, sourceRunId: latest.id, attachedToExistingWork: Boolean(canonicalId), strategy: "finalize_result", - ...(restartRecovery ? { initiatedBy: "restart_recovery", onBehalfOfUserId: input.actorId } : {}) } }); + ...(input.initiatedBy ? { initiatedBy: input.initiatedBy, onBehalfOfUserId: input.actorId } : {}) } }); const finalizing = savedResult.kind === "rearmed" || savedResult.finalizing; return { requestId: canonicalId ?? finalizeId, accepted: true, stale: false, observedRunId: latest.id, chatActionId: undefined, ...(finalizing ? { finalizeRunId: latest.id } : {}) }; @@ -371,10 +377,10 @@ function admitTaskRetry(input: TaskRetryRequest) { // A new deliberate click (new key) replaces it with a fresh attempt under the // clicker's current authority instead of joining it. Waits on shared state // (cleanup, recovery, an active run) keep coalescing: that intent starts by - // itself once the gate clears. The restart hook never replaces a user's click. + // itself once the gate clears. An automatic hook never replaces a user's click. let replacedRequestId: string | undefined; if (active && !busy && !active.runId && active.status === "deferred_issue_execution" && - input.initiatedBy !== "restart_recovery" && activeIntent && !activeIntent.chatActionId && + !input.initiatedBy && activeIntent && !activeIntent.chatActionId && !activeIntent.queuedChatRequestId && isDeclinedTaskRetry(active.payload)) { await tx.update(agentWakeupRequests).set({ status: "skipped", error: "task_retry_replaced", finishedAt: new Date(), updatedAt: new Date() }).where(and(eq(agentWakeupRequests.id, active.id), @@ -462,13 +468,11 @@ function admitTaskRetry(input: TaskRetryRequest) { actionId: action.id }, previous, next); } } - const restartRecovery = input.initiatedBy === "restart_recovery"; - await persistActivity(db, { companyId: issue.companyId, actorType: restartRecovery ? "system" : "user", - actorId: restartRecovery ? "restart-recovery" : input.actorId, + await persistActivity(db, { companyId: issue.companyId, ...automaticRetryActor(input), action: "issue.retry_requested", entityType: "issue", entityId: issue.id, runId: latest.id, details: { requestId: active?.id ?? id, sourceRunId: latest.id, attachedToExistingWork: Boolean(active || busy), ...(replacedRequestId ? { replacedRequestId } : {}), - ...(restartRecovery ? { initiatedBy: "restart_recovery", onBehalfOfUserId: input.actorId } : {}) } }); + ...(input.initiatedBy ? { initiatedBy: input.initiatedBy, onBehalfOfUserId: input.actorId } : {}) } }); return { requestId: active?.id ?? id, accepted: true, stale: false, observedRunId: latest.id, chatActionId: intent.chatActionId, ...(!active && latest.status === "scheduled_retry" ? { scheduledRetryRunId: latest.id } : {}) };