From c65ab09d9faf8c4675bc16957ccfd8a8b0bcfdf5 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Thu, 16 Jul 2026 09:24:38 -0500 Subject: [PATCH] fix(recovery): wait for provider quota resets (#9635) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source control plane people use to manage AI-agent companies and keep assigned work moving safely. > - The recovery subsystem decides whether a failed agent run should retry, wait, block for configuration, or escalate to another owner. > - Provider usage-limit failures currently arrive as generic `adapter_failed` results, so stranded-work reconciliation can create takeover recovery even when the provider states that capacity will reset later. > - Credential and model lookup failures are also configuration problems, not evidence that another agent should take over the task. > - This pull request classifies those failure families at recovery time and persists the classification on the run. > - Quota failures now schedule a monitor for the original assignee at the parsed reset time, or after a bounded default backoff when no reset time is available. > - The benefit is that transient provider capacity waits no longer wake recovery owners, while configuration failures stop with an actionable classification. ## Linked Issues or Issue Description No public GitHub issue exists for this exact change. **What happened?** When an assigned issue's latest run failed with a provider usage-limit message such as "try again at 12:00 AM (UTC)," recovery treated the run as generic `adapter_failed` work and could create a takeover action. Missing credentials and `model_not_found` failures followed the same generic path. **Expected behavior:** Provider quota failures should keep the original assignee and schedule a monitor for the reset time, without creating recovery work or immediately waking another owner. Missing credentials and model lookup failures should be classified as `configuration_incomplete` and blocked with the configuration fix recorded. **Steps to reproduce:** 1. Assign and start an issue for an agent. 2. Record a failed heartbeat run with `errorCode: adapter_failed` and a provider quota/reset message. 3. Run stranded assigned-issue reconciliation. 4. Observe that the old behavior routes the issue through generic recovery instead of waiting for provider capacity. Reproduced on `master` at `9af96461d`. This is a core recovery bug, not adapter-specific, and applies to built-from-source deployments with either embedded PGlite or Postgres. Related work checked: #9288 adds adapter-side Claude provider-limit classification; #5392 suppresses some recovery creation for quota-class errors; #9634 is a broader recovery-routing change with overlapping provider-quota behavior. This PR is the narrow recovery-service fix with focused parsed-reset, fallback-backoff, zero-takeover, and configuration-failure coverage. ## What Changed - Added conservative recovery-time classification for provider quota, missing-credential, and model-not-found adapter failures. - Parsed provider reset timestamps with a default one-hour backoff when no usable reset time is present. - Persisted `provider_quota` or `configuration_incomplete` metadata on the failed heartbeat run. - Scheduled quota monitors for the active issue owner, including the current review participant, without creating recovery actions or enqueueing takeover wakes. - Routed configuration failures to blocked recovery with actionable evidence instead of a takeover. - Added unit and embedded-database regression coverage for parsed/fallback quota timing, zero CTO/recovery wake behavior, and configuration classification. ## Verification - `pnpm exec vitest run server/src/services/recovery/provider-failure-classification.test.ts server/src/__tests__/issue-recovery-actions.test.ts server/src/__tests__/issue-monitor-scheduler.test.ts server/src/__tests__/heartbeat-stale-queue-invalidation.test.ts` — 4 files passed, 66 tests passed. - `pnpm --filter @paperclipai/shared typecheck` — passed. - `pnpm --filter @paperclipai/server typecheck` — passed. ## Risks - Recovery behavior changes for text-matched adapter failures; matching is intentionally conservative, and unmatched failures retain the existing generic recovery path. - Provider reset strings do not always include a date or timezone; parsing chooses the next future matching time and falls back to a one-hour wait when the timestamp is unusable. - This overlaps the provider-quota portion of broader recovery-routing PR #9634, so only one implementation should land if both remain open. - No schema, migration, API contract, or UI changes are included. No documentation update is needed because this corrects internal recovery behavior without changing operator commands or configuration. > For core feature work, check [`ROADMAP.md`](ROADMAP.md) first and discuss it in `#dev` before opening the PR. Feature PRs that overlap with planned core work may need to be redirected — check the roadmap first. See `CONTRIBUTING.md`. ## Model Used - OpenAI Codex with model `gpt-5.4`, medium reasoning, tool use, and code execution. The runtime does not expose its configured context-window size. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip --- packages/shared/src/constants.ts | 2 + packages/shared/src/index.ts | 1 + .../__tests__/issue-monitor-scheduler.test.ts | 68 +++ .../__tests__/issue-recovery-actions.test.ts | 572 ++++++++++++++++-- server/src/services/heartbeat.ts | 54 +- server/src/services/issue-execution-policy.ts | 4 + .../provider-failure-classification.test.ts | 95 +++ server/src/services/recovery/service.ts | 369 ++++++++++- 8 files changed, 1095 insertions(+), 70 deletions(-) create mode 100644 server/src/services/recovery/provider-failure-classification.test.ts diff --git a/packages/shared/src/constants.ts b/packages/shared/src/constants.ts index fc3db2e9ec..32ee7d609a 100644 --- a/packages/shared/src/constants.ts +++ b/packages/shared/src/constants.ts @@ -444,6 +444,8 @@ export type IssueMonitorScheduledBy = (typeof ISSUE_MONITOR_SCHEDULED_BY)[number export const ISSUE_EXECUTION_MONITOR_KINDS = ["external_service"] as const; export type IssueExecutionMonitorKind = (typeof ISSUE_EXECUTION_MONITOR_KINDS)[number]; +export const PROVIDER_QUOTA_MONITOR_SERVICE_NAME = "AI provider quota"; + export const ISSUE_EXECUTION_MONITOR_RECOVERY_POLICIES = [ "wake_owner", "create_recovery_issue", diff --git a/packages/shared/src/index.ts b/packages/shared/src/index.ts index 7184a011ba..cf991c13f3 100644 --- a/packages/shared/src/index.ts +++ b/packages/shared/src/index.ts @@ -212,6 +212,7 @@ export { ISSUE_EXECUTION_STAGE_TYPES, ISSUE_MONITOR_SCHEDULED_BY, ISSUE_EXECUTION_MONITOR_KINDS, + PROVIDER_QUOTA_MONITOR_SERVICE_NAME, ISSUE_EXECUTION_MONITOR_RECOVERY_POLICIES, ISSUE_EXECUTION_STATE_STATUSES, ISSUE_EXECUTION_MONITOR_STATE_STATUSES, diff --git a/server/src/__tests__/issue-monitor-scheduler.test.ts b/server/src/__tests__/issue-monitor-scheduler.test.ts index f682f682f5..87e44201da 100644 --- a/server/src/__tests__/issue-monitor-scheduler.test.ts +++ b/server/src/__tests__/issue-monitor-scheduler.test.ts @@ -1,6 +1,7 @@ import { randomUUID } from "node:crypto"; import { eq, sql } from "drizzle-orm"; import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest"; +import { PROVIDER_QUOTA_MONITOR_SERVICE_NAME } from "@paperclipai/shared"; import { activityLog, agentRuntimeState, @@ -270,6 +271,73 @@ describeEmbeddedPostgres("issue monitor scheduler", () => { expect(activity).toContain("issue.monitor_triggered"); }); + it("wakes a cross-agent review participant for provider quota monitors", async () => { + const { companyId, issueId, agentId: assigneeAgentId } = await seedFixture({ + issueStatus: "in_review", + monitor: { serviceName: PROVIDER_QUOTA_MONITOR_SERVICE_NAME }, + }); + const participantAgentId = randomUUID(); + await db.insert(agents).values({ + id: participantAgentId, + companyId, + name: "Quota-limited reviewer", + role: "engineer", + status: "active", + adapterType: "process", + adapterConfig: { + command: process.execPath, + args: ["-e", ""], + cwd: process.cwd(), + }, + runtimeConfig: { + heartbeat: { + enabled: false, + wakeOnDemand: true, + }, + }, + permissions: {}, + }); + seededAgentIds.add(participantAgentId); + const monitorState = await db + .select({ executionState: issues.executionState }) + .from(issues) + .where(eq(issues.id, issueId)) + .then((rows) => parseIssueExecutionState(rows[0]?.executionState ?? null)?.monitor ?? null); + await db.update(issues).set({ + executionState: { + status: "pending", + currentStageId: randomUUID(), + currentStageIndex: 0, + currentStageType: "review", + currentParticipant: { type: "agent", agentId: participantAgentId, userId: null }, + returnAssignee: { type: "agent", agentId: assigneeAgentId, userId: null }, + reviewRequest: null, + completedStageIds: [], + lastDecisionId: null, + lastDecisionOutcome: null, + monitor: monitorState, + }, + }).where(eq(issues.id, issueId)); + const heartbeat = heartbeatService(db); + + const result = await heartbeat.tickTimers(new Date("2026-04-11T12:31:00.000Z")); + + expect(result.enqueued).toBe(1); + const wakeups = await db.select().from(agentWakeupRequests); + expect(wakeups).toHaveLength(1); + expect(wakeups[0]).toMatchObject({ + agentId: participantAgentId, + reason: "execution_review_participant_recovery", + }); + await waitForHeartbeatIdle(); + const participantRuns = await db + .select() + .from(heartbeatRuns) + .where(eq(heartbeatRuns.agentId, participantAgentId)); + expect(participantRuns).toHaveLength(1); + expect(participantRuns[0]?.errorCode).not.toBe("issue_assignee_changed"); + }); + it("lets the board trigger a scheduled issue monitor immediately", async () => { const { issueId, agentId, nextCheckAt } = await seedFixture(); const heartbeat = heartbeatService(db); diff --git a/server/src/__tests__/issue-recovery-actions.test.ts b/server/src/__tests__/issue-recovery-actions.test.ts index 94948961aa..3228d5e55b 100644 --- a/server/src/__tests__/issue-recovery-actions.test.ts +++ b/server/src/__tests__/issue-recovery-actions.test.ts @@ -380,72 +380,538 @@ describeEmbeddedPostgres("issue recovery actions", () => { }, ); - it("creates a quota wait-recovery monitor without enqueueing a takeover wake", async () => { - const { companyId, coderId, sourceIssue } = await seedCompany(); + it("schedules a provider-quota monitor for the original assignee without creating recovery work", async () => { + const { companyId, coderId, sourceIssueId } = await seedCompany(); const runId = randomUUID(); - await seedHeartbeatRun({ companyId, agentId: coderId, runId, issueId: sourceIssue.id, status: "failed" }); + await db.insert(heartbeatRuns).values({ + id: runId, + companyId, + agentId: coderId, + invocationSource: "manual", + status: "failed", + error: "You've hit your usage limit for GPT-5. Try again at 12:00 AM (UTC).", + errorCode: "adapter_failed", + startedAt: new Date("2026-07-15T20:00:00.000Z"), + finishedAt: new Date("2026-07-15T20:01:00.000Z"), + contextSnapshot: { issueId: sourceIssueId }, + }); const enqueueWakeup = vi.fn(async () => null); const recovery = recoveryService(db, { enqueueWakeup }); - const retryAt = new Date(Date.now() + 60 * 60 * 1000).toISOString(); - await recovery.escalateStrandedAssignedIssue({ - issue: sourceIssue, - previousStatus: "in_progress", - latestRun: { - id: runId, - agentId: coderId, - status: "failed", - error: "provider usage limit exceeded", - errorCode: "adapter_failed", - contextSnapshot: { retryReason: "issue_continuation_needed" }, - livenessState: "needs_followup", - resultJson: { errorFamily: "provider_quota", providerQuotaRetryNotBefore: retryAt }, + const result = await recovery.reconcileStrandedAssignedIssues(); + + expect(result.providerQuotaMonitored).toBe(1); + const [updatedIssue] = await db.select().from(issues).where(eq(issues.id, sourceIssueId)); + expect(updatedIssue).toMatchObject({ + status: "in_progress", + assigneeAgentId: coderId, + monitorScheduledBy: "assignee", + monitorNotes: "Provider usage quota reached; retry the original assignee at the provider reset time.", + }); + expect(updatedIssue?.monitorNextCheckAt).toBeInstanceOf(Date); + expect(updatedIssue?.executionPolicy).toMatchObject({ + monitor: { + serviceName: "AI provider quota", + externalRef: runId, + maxAttempts: null, + recoveryPolicy: "wake_owner", + }, + }); + const [updatedRun] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)); + expect(updatedRun).toMatchObject({ errorCode: "provider_quota" }); + expect(updatedRun?.resultJson).toMatchObject({ errorFamily: "provider_quota" }); + expect(await db.select().from(issueRecoveryActions)).toHaveLength(0); + expect(enqueueWakeup).not.toHaveBeenCalled(); + + const secondResult = await recovery.reconcileStrandedAssignedIssues(); + expect(secondResult).toMatchObject({ providerQuotaMonitored: 0, skipped: 1 }); + expect(await db.select().from(issueRecoveryActions)).toHaveLength(0); + }); + + it("schedules another provider-quota monitor after a prior quota monitor fired", async () => { + const { companyId, coderId, sourceIssueId } = await seedCompany(); + await db.update(issues).set({ monitorAttemptCount: 1 }).where(eq(issues.id, sourceIssueId)); + const runId = randomUUID(); + await db.insert(heartbeatRuns).values({ + id: runId, + companyId, + agentId: coderId, + invocationSource: "manual", + status: "failed", + error: "Provider quota exceeded for this model.", + errorCode: "adapter_failed", + startedAt: new Date("2026-07-15T21:00:00.000Z"), + finishedAt: new Date("2026-07-15T21:01:00.000Z"), + contextSnapshot: { issueId: sourceIssueId }, + }); + const recovery = recoveryService(db, { enqueueWakeup: vi.fn(async () => null) }); + + const result = await recovery.reconcileStrandedAssignedIssues(); + + expect(result.providerQuotaMonitored).toBe(1); + const [updatedIssue] = await db.select().from(issues).where(eq(issues.id, sourceIssueId)); + expect(updatedIssue?.executionPolicy).toMatchObject({ + monitor: { + maxAttempts: null, + externalRef: runId, + }, + }); + }); + + it("skips provider-quota monitor scheduling for todo issues without aborting reconciliation", async () => { + const { companyId, coderId, sourceIssueId } = await seedCompany(); + await db.update(issues).set({ status: "todo" }).where(eq(issues.id, sourceIssueId)); + const runId = randomUUID(); + await db.insert(heartbeatRuns).values({ + id: runId, + companyId, + agentId: coderId, + invocationSource: "manual", + status: "failed", + error: "Provider quota exceeded for this model.", + errorCode: "adapter_failed", + startedAt: new Date("2026-07-15T20:00:00.000Z"), + finishedAt: new Date("2026-07-15T20:01:00.000Z"), + contextSnapshot: { issueId: sourceIssueId }, + }); + const enqueueWakeup = vi.fn(async () => null); + const recovery = recoveryService(db, { enqueueWakeup }); + + const result = await recovery.reconcileStrandedAssignedIssues(); + + expect(result).toMatchObject({ providerQuotaMonitored: 0, skipped: 1 }); + const [updatedIssue] = await db.select().from(issues).where(eq(issues.id, sourceIssueId)); + expect(updatedIssue).toMatchObject({ + status: "todo", + assigneeAgentId: coderId, + monitorNextCheckAt: null, + }); + const [updatedRun] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)); + expect(updatedRun?.errorCode).toBe("adapter_failed"); + expect(await db.select().from(issueRecoveryActions)).toHaveLength(0); + expect(enqueueWakeup).not.toHaveBeenCalled(); + }); + + it("does not create takeover recovery when a quota monitor cannot be scheduled", async () => { + const { companyId, coderId, sourceIssueId } = await seedCompany(); + await db.update(issues).set({ status: "in_review" }).where(eq(issues.id, sourceIssueId)); + const runId = randomUUID(); + await db.insert(heartbeatRuns).values({ + id: runId, + companyId, + agentId: coderId, + invocationSource: "manual", + status: "failed", + error: "Provider quota exceeded for this model.", + errorCode: "adapter_failed", + startedAt: new Date("2026-07-15T20:00:00.000Z"), + finishedAt: new Date("2026-07-15T20:01:00.000Z"), + contextSnapshot: { issueId: sourceIssueId }, + }); + const enqueueWakeup = vi.fn(async () => null); + const recovery = recoveryService(db, { enqueueWakeup }); + + const result = await recovery.reconcileStrandedAssignedIssues(); + + expect(result).toMatchObject({ providerQuotaMonitored: 0, skipped: 1 }); + const [updatedIssue] = await db.select().from(issues).where(eq(issues.id, sourceIssueId)); + expect(updatedIssue).toMatchObject({ + status: "in_review", + assigneeAgentId: coderId, + monitorNextCheckAt: null, + }); + const [updatedRun] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)); + expect(updatedRun?.errorCode).toBe("adapter_failed"); + expect(await db.select().from(issueRecoveryActions)).toHaveLength(0); + expect(enqueueWakeup).not.toHaveBeenCalled(); + }); + + it("schedules a quota monitor for a cross-agent active review participant", async () => { + const { companyId, managerId, coderId, sourceIssueId } = await seedCompany(); + const stageId = randomUUID(); + await db.update(issues).set({ + status: "in_review", + assigneeAgentId: coderId, + executionPolicy: { + mode: "normal", + commentRequired: true, + stages: [{ + id: stageId, + type: "review", + approvalsNeeded: 1, + participants: [{ id: randomUUID(), type: "agent", agentId: managerId, userId: null }], + }], + }, + executionState: { + status: "pending", + currentStageId: stageId, + currentStageIndex: 0, + currentStageType: "review", + currentParticipant: { type: "agent", agentId: managerId, userId: null }, + returnAssignee: { type: "agent", agentId: coderId, userId: null }, + reviewRequest: null, + completedStageIds: [], + lastDecisionId: null, + lastDecisionOutcome: null, + }, + }).where(eq(issues.id, sourceIssueId)); + const [reviewIssueBeforeRecovery] = await db.select().from(issues).where(eq(issues.id, sourceIssueId)); + expect(reviewIssueBeforeRecovery).toMatchObject({ + assigneeAgentId: coderId, + executionState: { + currentParticipant: { type: "agent", agentId: managerId }, + returnAssignee: { type: "agent", agentId: coderId }, + }, + }); + const runId = randomUUID(); + await db.insert(heartbeatRuns).values({ + id: runId, + companyId, + agentId: managerId, + invocationSource: "automation", + status: "failed", + error: "Provider quota exceeded for this model.", + errorCode: "adapter_failed", + startedAt: new Date("2026-07-15T20:00:00.000Z"), + finishedAt: new Date("2026-07-15T20:01:00.000Z"), + contextSnapshot: { issueId: sourceIssueId }, + }); + const enqueueWakeup = vi.fn(async () => null); + const recovery = recoveryService(db, { enqueueWakeup }); + + const result = await recovery.reconcileStrandedAssignedIssues(); + + expect(result).toMatchObject({ providerQuotaMonitored: 1, reviewParticipantRequeued: 0 }); + const [updatedIssue] = await db.select().from(issues).where(eq(issues.id, sourceIssueId)); + expect(updatedIssue).toMatchObject({ + status: "in_review", + assigneeAgentId: coderId, + monitorNextCheckAt: expect.any(Date), + monitorNotes: "Provider usage quota reached; retry the active review participant after the default recovery backoff.", + }); + const [updatedRun] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)); + expect(updatedRun?.errorCode).toBe("provider_quota"); + expect(await db.select().from(issueRecoveryActions)).toHaveLength(0); + expect(enqueueWakeup).not.toHaveBeenCalled(); + }); + + it("does not restamp an in_review quota monitor when the assignee has a newer terminal run", async () => { + const { companyId, managerId, coderId, sourceIssueId } = await seedCompany(); + const stageId = randomUUID(); + await db.update(issues).set({ + status: "in_review", + assigneeAgentId: coderId, + executionPolicy: { + mode: "normal", + commentRequired: true, + stages: [{ + id: stageId, + type: "review", + approvalsNeeded: 1, + participants: [{ id: randomUUID(), type: "agent", agentId: managerId, userId: null }], + }], + }, + executionState: { + status: "pending", + currentStageId: stageId, + currentStageIndex: 0, + currentStageType: "review", + currentParticipant: { type: "agent", agentId: managerId, userId: null }, + returnAssignee: { type: "agent", agentId: coderId, userId: null }, + reviewRequest: null, + completedStageIds: [], + lastDecisionId: null, + lastDecisionOutcome: null, + }, + }).where(eq(issues.id, sourceIssueId)); + const participantRunId = randomUUID(); + await db.insert(heartbeatRuns).values({ + id: participantRunId, + companyId, + agentId: managerId, + invocationSource: "automation", + status: "failed", + error: "Provider quota exceeded for this model.", + errorCode: "adapter_failed", + startedAt: new Date("2026-07-15T20:00:00.000Z"), + finishedAt: new Date("2026-07-15T20:01:00.000Z"), + contextSnapshot: { issueId: sourceIssueId }, + }); + const enqueueWakeup = vi.fn(async () => null); + const recovery = recoveryService(db, { enqueueWakeup }); + + const firstResult = await recovery.reconcileStrandedAssignedIssues(); + + expect(firstResult).toMatchObject({ providerQuotaMonitored: 1 }); + const [monitoredIssue] = await db.select().from(issues).where(eq(issues.id, sourceIssueId)); + const firstNextCheckAt = monitoredIssue?.monitorNextCheckAt; + expect(firstNextCheckAt).toBeInstanceOf(Date); + expect(monitoredIssue?.executionPolicy).toMatchObject({ + monitor: { + serviceName: "AI provider quota", + externalRef: participantRunId, }, }); - expect(enqueueWakeup).not.toHaveBeenCalled(); - const [action] = await db - .select() - .from(issueRecoveryActions) - .where(eq(issueRecoveryActions.sourceIssueId, sourceIssue.id)); - expect(action).toMatchObject({ - cause: "provider_quota", - ownerType: "system", - ownerAgentId: null, - returnOwnerAgentId: coderId, - wakePolicy: expect.objectContaining({ type: "monitor_only" }), - monitorPolicy: expect.objectContaining({ type: "wait_recovery", retryAgentId: coderId }), - }); - const scheduled = await db - .select() - .from(heartbeatRuns) - .where(eq(heartbeatRuns.status, "scheduled_retry")); - expect(scheduled).toHaveLength(1); - expect(scheduled[0]).toMatchObject({ - agentId: coderId, - scheduledRetryReason: "provider_quota_recovery", - }); - expect(scheduled[0]?.contextSnapshot).toMatchObject({ - issueId: sourceIssue.id, - wakeReason: "provider_quota_recovery", - }); - expect(scheduled[0]?.contextSnapshot).not.toHaveProperty("recoveryActionId"); - expect(scheduled[0]?.contextSnapshot).not.toHaveProperty("recoveryCause"); - const wakePayload = await buildPaperclipWakePayload({ - db, + await db.insert(heartbeatRuns).values({ + id: randomUUID(), companyId, - contextSnapshot: scheduled[0]?.contextSnapshot as Record, + agentId: coderId, + invocationSource: "automation", + status: "failed", + error: "Stale assignee wake fired after the issue entered review.", + errorCode: "issue_assignee_changed", + startedAt: new Date("2026-07-15T20:02:00.000Z"), + finishedAt: new Date("2026-07-15T20:03:00.000Z"), + contextSnapshot: { issueId: sourceIssueId }, }); - expect(wakePayload?.reason).toBe("provider_quota_recovery"); - expect(wakePayload?.recovery).toBeNull(); - const [updatedIssue] = await db - .select() - .from(issues) - .where(eq(issues.id, sourceIssue.id)); + + const secondResult = await recovery.reconcileStrandedAssignedIssues(); + + expect(secondResult).toMatchObject({ providerQuotaMonitored: 0, skipped: 1 }); + const [unchangedIssue] = await db.select().from(issues).where(eq(issues.id, sourceIssueId)); + expect(unchangedIssue?.monitorNextCheckAt?.getTime()).toBe(firstNextCheckAt?.getTime()); + expect(unchangedIssue?.executionPolicy).toMatchObject({ + monitor: { + serviceName: "AI provider quota", + externalRef: participantRunId, + }, + }); + expect(await db.select().from(issueRecoveryActions)).toHaveLength(0); + expect(enqueueWakeup).not.toHaveBeenCalled(); + }); + + it("classifies review recovery from the active participant run instead of a newer assignee run", async () => { + const { companyId, managerId, coderId, sourceIssueId } = await seedCompany(); + const stageId = randomUUID(); + await db.update(issues).set({ + status: "in_review", + assigneeAgentId: coderId, + executionPolicy: { + mode: "normal", + commentRequired: true, + stages: [{ + id: stageId, + type: "review", + approvalsNeeded: 1, + participants: [{ id: randomUUID(), type: "agent", agentId: managerId, userId: null }], + }], + }, + executionState: { + status: "pending", + currentStageId: stageId, + currentStageIndex: 0, + currentStageType: "review", + currentParticipant: { type: "agent", agentId: managerId, userId: null }, + returnAssignee: { type: "agent", agentId: coderId, userId: null }, + reviewRequest: null, + completedStageIds: [], + lastDecisionId: null, + lastDecisionOutcome: null, + }, + }).where(eq(issues.id, sourceIssueId)); + const participantRunId = randomUUID(); + const assigneeRunId = randomUUID(); + await db.insert(heartbeatRuns).values([{ + id: participantRunId, + companyId, + agentId: managerId, + invocationSource: "automation", + status: "failed", + error: "review process exited unexpectedly", + errorCode: "adapter_failed", + startedAt: new Date("2026-07-15T20:00:00.000Z"), + finishedAt: new Date("2026-07-15T20:01:00.000Z"), + contextSnapshot: { issueId: sourceIssueId }, + }, { + id: assigneeRunId, + companyId, + agentId: coderId, + invocationSource: "automation", + status: "failed", + error: "You've hit your usage limit. Try again at 11:00 PM (UTC)", + errorCode: "adapter_failed", + startedAt: new Date("2026-07-15T20:02:00.000Z"), + finishedAt: new Date("2026-07-15T20:03:00.000Z"), + contextSnapshot: { issueId: sourceIssueId }, + }]); + const enqueueWakeup = vi.fn(async () => ({ id: randomUUID() } as never)); + const recovery = recoveryService(db, { enqueueWakeup }); + + const result = await recovery.reconcileStrandedAssignedIssues(); + + expect(result).toMatchObject({ providerQuotaMonitored: 0, reviewParticipantRequeued: 1 }); + const [updatedIssue] = await db.select().from(issues).where(eq(issues.id, sourceIssueId)); + expect(updatedIssue).toMatchObject({ + status: "in_review", + assigneeAgentId: coderId, + monitorNextCheckAt: null, + }); + const [assigneeRun] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, assigneeRunId)); + expect(assigneeRun?.errorCode).toBe("adapter_failed"); + expect(enqueueWakeup).toHaveBeenCalledWith(managerId, expect.objectContaining({ + reason: "execution_review_participant_recovery", + payload: expect.objectContaining({ issueId: sourceIssueId, retryOfRunId: participantRunId }), + })); + }); + + it("blocks a cross-agent review participant with incomplete configuration", async () => { + const { companyId, managerId, coderId, sourceIssueId } = await seedCompany(); + const stageId = randomUUID(); + await db.update(issues).set({ + status: "in_review", + executionPolicy: { + mode: "normal", + commentRequired: true, + stages: [{ + id: stageId, + type: "review", + approvalsNeeded: 1, + participants: [{ id: randomUUID(), type: "agent", agentId: managerId, userId: null }], + }], + }, + executionState: { + status: "pending", + currentStageId: stageId, + currentStageIndex: 0, + currentStageType: "review", + currentParticipant: { type: "agent", agentId: managerId, userId: null }, + returnAssignee: { type: "agent", agentId: coderId, userId: null }, + reviewRequest: null, + completedStageIds: [], + lastDecisionId: null, + lastDecisionOutcome: null, + }, + }).where(eq(issues.id, sourceIssueId)); + const runId = randomUUID(); + await db.insert(heartbeatRuns).values({ + id: runId, + companyId, + agentId: managerId, + invocationSource: "automation", + status: "failed", + error: "model_not_found: requested review model does not exist", + errorCode: "adapter_failed", + startedAt: new Date("2026-07-15T20:00:00.000Z"), + finishedAt: new Date("2026-07-15T20:01:00.000Z"), + contextSnapshot: { issueId: sourceIssueId }, + }); + const enqueueWakeup = vi.fn(async () => ({ id: randomUUID() } as never)); + const recovery = recoveryService(db, { enqueueWakeup }); + + const result = await recovery.reconcileStrandedAssignedIssues(); + + expect(result).toMatchObject({ escalated: 1, reviewParticipantRequeued: 0 }); + const [updatedIssue] = await db.select().from(issues).where(eq(issues.id, sourceIssueId)); expect(updatedIssue).toMatchObject({ status: "blocked", + assigneeAgentId: managerId, + }); + const [updatedRun] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)); + expect(updatedRun?.errorCode).toBe("configuration_incomplete"); + const [action] = await db.select().from(issueRecoveryActions); + expect(action).toMatchObject({ + sourceIssueId, + ownerAgentId: managerId, + previousOwnerAgentId: coderId, + cause: "configuration_incomplete", + recoveryIssueId: null, + }); + expect(enqueueWakeup).not.toHaveBeenCalled(); + }); + + it("uses the default quota backoff when the provider does not state a reset time", async () => { + const { companyId, coderId, sourceIssueId } = await seedCompany(); + await db.insert(heartbeatRuns).values({ + id: randomUUID(), + companyId, + agentId: coderId, + invocationSource: "manual", + status: "failed", + error: "Provider quota exceeded for this model.", + errorCode: "adapter_failed", + startedAt: new Date("2026-07-15T20:00:00.000Z"), + finishedAt: new Date("2026-07-15T20:01:00.000Z"), + contextSnapshot: { issueId: sourceIssueId }, + }); + const recovery = recoveryService(db, { enqueueWakeup: vi.fn(async () => null) }); + + const result = await recovery.reconcileStrandedAssignedIssues(); + + expect(result.providerQuotaMonitored).toBe(1); + const [updatedIssue] = await db.select().from(issues).where(eq(issues.id, sourceIssueId)); + expect(updatedIssue).toMatchObject({ + status: "in_progress", + assigneeAgentId: coderId, + monitorNotes: "Provider usage quota reached; retry the original assignee after the default recovery backoff.", + }); + expect(await db.select().from(issueRecoveryActions)).toHaveLength(0); + }); + + it("classifies model lookup failures as configuration incomplete without waking a recovery owner", async () => { + const { companyId, coderId, sourceIssueId } = await seedCompany(); + const runId = randomUUID(); + await db.insert(heartbeatRuns).values({ + id: runId, + companyId, + agentId: coderId, + invocationSource: "manual", + status: "failed", + error: "model_not_found: requested model does not exist", + errorCode: "adapter_failed", + startedAt: new Date("2026-07-15T20:00:00.000Z"), + finishedAt: new Date("2026-07-15T20:01:00.000Z"), + contextSnapshot: { issueId: sourceIssueId }, + }); + const enqueueWakeup = vi.fn(async () => null); + const recovery = recoveryService(db, { enqueueWakeup }); + + const result = await recovery.reconcileStrandedAssignedIssues(); + + expect(result).toMatchObject({ escalated: 1, skipped: 0 }); + const [updatedIssue] = await db.select().from(issues).where(eq(issues.id, sourceIssueId)); + expect(updatedIssue?.status).toBe("blocked"); + const [updatedRun] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)); + expect(updatedRun?.errorCode).toBe("configuration_incomplete"); + const [action] = await db.select().from(issueRecoveryActions); + expect(action).toMatchObject({ + sourceIssueId, + cause: "configuration_incomplete", + recoveryIssueId: null, + }); + expect(enqueueWakeup).not.toHaveBeenCalled(); + }); + + it("does not classify stale configuration failures from a non-assignee run", async () => { + const { companyId, managerId, coderId, sourceIssueId } = await seedCompany(); + const runId = randomUUID(); + await db.insert(heartbeatRuns).values({ + id: runId, + companyId, + agentId: managerId, + invocationSource: "manual", + status: "failed", + error: "model_not_found: previous assignee model does not exist", + errorCode: "adapter_failed", + startedAt: new Date("2026-07-15T20:00:00.000Z"), + finishedAt: new Date("2026-07-15T20:01:00.000Z"), + contextSnapshot: { issueId: sourceIssueId }, + }); + const enqueueWakeup = vi.fn(async () => null); + const recovery = recoveryService(db, { enqueueWakeup }); + + const result = await recovery.reconcileStrandedAssignedIssues(); + + expect(result).toMatchObject({ escalated: 0, skipped: 1 }); + const [updatedIssue] = await db.select().from(issues).where(eq(issues.id, sourceIssueId)); + expect(updatedIssue).toMatchObject({ + status: "in_progress", assigneeAgentId: coderId, }); + const [updatedRun] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)); + expect(updatedRun?.errorCode).toBe("adapter_failed"); + expect(await db.select().from(issueRecoveryActions)).toHaveLength(0); + expect(enqueueWakeup).not.toHaveBeenCalled(); }); it("reuses the same source-scoped action when latest run IDs change while the cause stays the same", async () => { diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index e7ca5cd2e5..47ee92625a 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -9,6 +9,7 @@ import { AGENT_DEFAULT_MAX_CONCURRENT_RUNS, ISSUE_CONTINUATION_SUMMARY_DOCUMENT_KEY, MODEL_PROFILE_KEYS, + PROVIDER_QUOTA_MONITOR_SERVICE_NAME, envBindingSchema, isEnvironmentDriverSupportedForAdapter, type BillingType, @@ -6617,6 +6618,35 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) maxAttempts: monitor?.maxAttempts ?? null, recoveryPolicy: monitor?.recoveryPolicy ?? null, }; + const executionState = claimed.status === "in_review" + ? parseIssueExecutionState(claimed.executionState) + : null; + const currentParticipant = executionState?.status === "pending" + ? executionState.currentParticipant + : null; + const reviewParticipantAgentId = currentParticipant?.type === "agent" + ? currentParticipant.agentId + : null; + const isProviderQuotaReviewMonitor = monitor?.serviceName === PROVIDER_QUOTA_MONITOR_SERVICE_NAME && + Boolean(reviewParticipantAgentId); + const targetAgentId = isProviderQuotaReviewMonitor + ? reviewParticipantAgentId + : claimed.assigneeAgentId; + if (!targetAgentId) { + throw conflict("Issue monitor has no agent target"); + } + const wakeReason = isProviderQuotaReviewMonitor + ? EXECUTION_REVIEW_PARTICIPANT_RECOVERY_WAKE_REASON + : input.wakeReason; + const reviewRecoveryContext = isProviderQuotaReviewMonitor + ? { + retryReason: EXECUTION_REVIEW_PARTICIPANT_RECOVERY_RETRY_REASON, + currentStageId: executionState?.currentStageId ?? null, + currentStageType: executionState?.currentStageType ?? null, + reviewRecoveryInstruction: + "The previous reviewer run reached provider quota. Resume this execution-review stage now that the quota wait has elapsed.", + } + : {}; if (clearReason) { return clearIssueMonitorAndRecover({ @@ -6637,10 +6667,10 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) } try { - await enqueueWakeup(claimed.assigneeAgentId, { + await enqueueWakeup(targetAgentId, { source: input.source, triggerDetail: input.triggerDetail, - reason: input.wakeReason, + reason: wakeReason, idempotencyKey: `issue-monitor:${claimed.id}:${scheduledAtIso}`, payload: { issueId: claimed.id, @@ -6648,18 +6678,20 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) monitorAttemptCount: nextAttemptCount, monitorNotes: claimed.monitorNotes ?? null, ...monitorMetadata, + ...reviewRecoveryContext, source: input.activitySource, }, requestedByActorType: input.actorType, requestedByActorId: input.actorId, contextSnapshot: { issueId: claimed.id, - source: "issue.monitor", - wakeReason: input.wakeReason, + source: isProviderQuotaReviewMonitor ? "issue.execution_review_recovery" : "issue.monitor", + wakeReason, nextCheckAt: scheduledAtIso, monitorAttemptCount: nextAttemptCount, monitorNotes: claimed.monitorNotes ?? null, ...monitorMetadata, + ...reviewRecoveryContext, manualTrigger: input.activitySource === "manual", }, }); @@ -10866,7 +10898,14 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) } } - if (issue.assigneeAgentId !== run.agentId && !isInteractionWake) { + const reviewExecutionState = issue.status === "in_review" + ? parseIssueExecutionState(issue.executionState) + : null; + const reviewParticipant = reviewExecutionState?.currentParticipant ?? null; + const isCurrentReviewParticipant = reviewParticipant?.type === "agent" && + reviewParticipant.agentId === run.agentId; + + if (issue.assigneeAgentId !== run.agentId && !isInteractionWake && !isCurrentReviewParticipant) { return { stale: true, errorCode: "issue_assignee_changed", @@ -10915,8 +10954,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) } if (issue.status === "in_review") { - const executionState = parseIssueExecutionState(issue.executionState); - const currentParticipant = executionState?.currentParticipant ?? null; + const currentParticipant = reviewExecutionState?.currentParticipant ?? null; if (currentParticipant) { const participantMatches = currentParticipant.type === "agent" && currentParticipant.agentId === run.agentId; @@ -10928,7 +10966,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) "Cancelled because the in-review participant changed before the queued run could start; the current participant will be woken instead", details: { issueId, - currentStageType: executionState?.currentStageType ?? null, + currentStageType: reviewExecutionState?.currentStageType ?? null, currentParticipant, }, }; diff --git a/server/src/services/issue-execution-policy.ts b/server/src/services/issue-execution-policy.ts index 583b52c568..f09f47020e 100644 --- a/server/src/services/issue-execution-policy.ts +++ b/server/src/services/issue-execution-policy.ts @@ -1053,3 +1053,7 @@ export function applyIssueExecutionPolicyTransition(input: TransitionInput): Tra Object.assign(stageResult.patch, monitorPatch); return stageResult; } + +export function applyIssueMonitorPolicyTransition(input: TransitionInput): TransitionResult { + return { patch: applyMonitorTransition(input, {}) }; +} diff --git a/server/src/services/recovery/provider-failure-classification.test.ts b/server/src/services/recovery/provider-failure-classification.test.ts new file mode 100644 index 0000000000..20f5f43e4f --- /dev/null +++ b/server/src/services/recovery/provider-failure-classification.test.ts @@ -0,0 +1,95 @@ +import { describe, expect, it } from "vitest"; +import { + PROVIDER_QUOTA_RECOVERY_DEFAULT_BACKOFF_MS, + classifyAdapterFailureForRecovery, +} from "./service.js"; + +describe("classifyAdapterFailureForRecovery", () => { + it("classifies usage-limit messages and parses the provider reset time", () => { + const now = new Date("2026-07-15T20:00:00.000Z"); + const classification = classifyAdapterFailureForRecovery({ + errorCode: "adapter_failed", + error: "You've hit your usage limit for GPT-5. Try again at 4:30 PM (America/Chicago).", + resultJson: null, + }, now); + + expect(classification).toEqual({ + kind: "provider_quota", + retryAt: new Date("2026-07-15T21:30:00.000Z"), + 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({ + errorCode: "adapter_failed", + error: "Provider quota exceeded for this model.", + resultJson: null, + }, now); + + expect(classification).toEqual({ + kind: "provider_quota", + retryAt: new Date(now.getTime() + PROVIDER_QUOTA_RECOVERY_DEFAULT_BACKOFF_MS), + parsedResetTime: false, + }); + }); + + it("treats timezone-less provider reset clocks as UTC", () => { + const now = new Date("2026-07-15T20:00:00.000Z"); + const classification = classifyAdapterFailureForRecovery({ + errorCode: "adapter_failed", + error: "You've hit your usage limit. Try again at 4:30 PM.", + resultJson: null, + }, now); + + expect(classification).toEqual({ + kind: "provider_quota", + retryAt: new Date("2026-07-16T16:30:00.000Z"), + parsedResetTime: true, + }); + }); + + it("parses provider reset clocks in 24-hour format", () => { + const now = new Date("2026-07-15T20:00:00.000Z"); + const classification = classifyAdapterFailureForRecovery({ + errorCode: "adapter_failed", + error: "You've hit your usage limit. Try again at 21:30 (UTC).", + resultJson: null, + }, now); + + expect(classification).toEqual({ + kind: "provider_quota", + retryAt: new Date("2026-07-15T21:30:00.000Z"), + parsedResetTime: true, + }); + }); + + it.each([ + "model_not_found: requested model does not exist", + "No API credentials were found for this provider", + "API key is not set", + ])("classifies configuration failures: %s", (error) => { + expect(classifyAdapterFailureForRecovery({ + errorCode: "adapter_failed", + error, + resultJson: null, + })).toEqual({ kind: "configuration_incomplete" }); + }); + + it("ignores quota-like text from non-adapter failures", () => { + expect(classifyAdapterFailureForRecovery({ + errorCode: "timeout", + error: "Provider quota exceeded while waiting for a downstream service.", + resultJson: null, + })).toBeNull(); + }); + + it("does not treat a generic capacity limit as provider quota", () => { + expect(classifyAdapterFailureForRecovery({ + errorCode: "adapter_failed", + error: "Workspace storage capacity limit reached.", + resultJson: null, + })).toBeNull(); + }); +}); diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index 6d5b0d7da0..b370f7b711 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -4,6 +4,7 @@ import { DEFAULT_ISSUE_GRAPH_LIVENESS_AUTO_RECOVERY_LOOKBACK_HOURS, MAX_ISSUE_GRAPH_LIVENESS_AUTO_RECOVERY_LOOKBACK_HOURS, MIN_ISSUE_GRAPH_LIVENESS_AUTO_RECOVERY_LOOKBACK_HOURS, + PROVIDER_QUOTA_MONITOR_SERVICE_NAME, type IssueGraphLivenessAutoRecoveryPreview, type IssueGraphLivenessAutoRecoveryPreviewItem, } from "@paperclipai/shared"; @@ -38,12 +39,16 @@ import { instanceSettingsService } from "../instance-settings.js"; import { issueRecoveryActionService } from "../issue-recovery-actions.js"; import { issueTreeControlService } from "../issue-tree-control.js"; import { TERMINAL_HEARTBEAT_RUN_STATUSES, issueService } from "../issues.js"; +import { + applyIssueMonitorPolicyTransition, + normalizeIssueExecutionPolicy, + parseIssueExecutionState, +} from "../issue-execution-policy.js"; import { ISSUE_BLOCKERS_RESOLVED_WAKE_REASON, buildIssueBlockersResolvedWakeIdempotencyKey, findExistingIssueBlockersResolvedWakeForAnyKey, } from "../issue-dependency-wakeups.js"; -import { parseIssueExecutionState } from "../issue-execution-policy.js"; import { evaluateAgentInvokabilityFromDb } from "../agent-invokability.js"; import { getRunLogStore } from "../run-log-store.js"; import { @@ -159,8 +164,6 @@ type SuccessfulRunHandoffRecoveryEvidence = { maxHandoffAttempts: number; }; -const PROVIDER_QUOTA_RECOVERY_DEFAULT_BACKOFF_MS = 15 * 60 * 1000; - function readRecoveryRunErrorFamily(latestRun: LatestIssueRun) { const result = parseObject(latestRun?.resultJson); return readNonEmptyString(result.errorFamily); @@ -295,6 +298,122 @@ 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; + +const PROVIDER_QUOTA_ERROR_RE = + /(?:you(?:'|’)ve hit your usage limit|usage limit(?: reached| exceeded)?|provider quota|quota (?:limit )?exceeded|model (?:is )?at capacity)/i; +const CONFIGURATION_INCOMPLETE_ERROR_RE = + /(?:model_not_found|model [^\n]{0,120} not found|missing (?:api )?(?:key|credentials?)|credentials? (?:are |is )?missing|no (?:api )?(?:key|credentials?) (?:was |were )?(?:found|configured|provided)|api key (?:is )?(?:not set|unavailable))/i; + +export type AdapterFailureRecoveryClassification = + | { kind: "provider_quota"; retryAt: Date; parsedResetTime: boolean } + | { kind: "configuration_incomplete" } + | null; + +function parseProviderQuotaClockReset(error: string, now: Date) { + const match = error.match( + /try again at\s+(\d{1,2})(?::(\d{2}))?\s*(?:([ap])\.?\s*m\.?)?(?:\s*\(([^)]+)\)|\s+([A-Z]{2,5}))?/i, + ); + if (!match) return null; + + const hourValue = Number.parseInt(match[1] ?? "", 10); + const minute = Number.parseInt(match[2] ?? "0", 10); + const meridiem = (match[3] ?? "").toLowerCase(); + if (!Number.isInteger(hourValue)) return null; + if (meridiem ? hourValue < 1 || hourValue > 12 : hourValue < 0 || hourValue > 23) return null; + if (!Number.isInteger(minute) || minute < 0 || minute > 59) return null; + + 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; + } + + try { + const wallClock = (date: Date) => Object.fromEntries( + new Intl.DateTimeFormat("en-US", { + timeZone, + hourCycle: "h23", + year: "numeric", + month: "2-digit", + day: "2-digit", + hour: "2-digit", + minute: "2-digit", + }).formatToParts(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(); + for (let attempt = 0; attempt < 4; attempt += 1) { + const actual = wallClock(candidate); + const actualMs = Date.UTC( + Number(actual.year), + Number(actual.month) - 1, + Number(actual.day), + Number(actual.hour), + Number(actual.minute), + ); + const adjustment = targetMs - actualMs; + if (adjustment === 0) break; + candidate = new Date(candidate.getTime() + adjustment); + } + return candidate; + }; + const sameDay = buildRetryAt(0); + return sameDay.getTime() > now.getTime() ? sameDay : buildRetryAt(1); + } catch { + return null; + } +} + +export function classifyAdapterFailureForRecovery( + latestRun: Pick, "error" | "errorCode" | "resultJson">, + now = new Date(), +): AdapterFailureRecoveryClassification { + if ( + latestRun.errorCode !== "adapter_failed" && + latestRun.errorCode !== "provider_quota" && + latestRun.errorCode !== "configuration_incomplete" + ) { + return null; + } + const resultJson = parseObject(latestRun.resultJson); + const error = [latestRun.errorCode ?? "", latestRun.error ?? "", JSON.stringify(resultJson)].join("\n"); + if (latestRun.errorCode === "configuration_incomplete" || CONFIGURATION_INCOMPLETE_ERROR_RE.test(error)) { + return { kind: "configuration_incomplete" }; + } + if (latestRun.errorCode !== "provider_quota" && !PROVIDER_QUOTA_ERROR_RE.test(error)) return null; + + const persistedRetryAt = readNonEmptyString(resultJson.retryNotBefore) ?? + readNonEmptyString(resultJson.transientRetryNotBefore) ?? + readNonEmptyString(resultJson.providerQuotaRetryNotBefore); + const parsedPersistedRetryAt = persistedRetryAt ? new Date(persistedRetryAt) : null; + if (parsedPersistedRetryAt && !Number.isNaN(parsedPersistedRetryAt.getTime()) && parsedPersistedRetryAt > now) { + return { kind: "provider_quota", retryAt: parsedPersistedRetryAt, parsedResetTime: true }; + } + + const parsedClockReset = parseProviderQuotaClockReset(error, now); + if (parsedClockReset) { + return { kind: "provider_quota", retryAt: parsedClockReset, parsedResetTime: true }; + } + return { + kind: "provider_quota", + retryAt: new Date(now.getTime() + PROVIDER_QUOTA_RECOVERY_DEFAULT_BACKOFF_MS), + parsedResetTime: false, + }; +} type ContinuationRetryClassification = { kind: "transient_infra" | "non_retryable" | "default"; @@ -3210,6 +3329,139 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) return updated; } + async function persistAdapterFailureRecoveryClassification( + latestRun: NonNullable, + classification: NonNullable, + ): Promise> { + const classifiedRun = withAdapterFailureRecoveryClassification(latestRun, classification); + + await db + .update(heartbeatRuns) + .set({ + errorCode: classifiedRun.errorCode, + resultJson: parseObject(classifiedRun.resultJson), + updatedAt: new Date(), + }) + .where(eq(heartbeatRuns.id, latestRun.id)); + + return classifiedRun; + } + + function withAdapterFailureRecoveryClassification( + latestRun: NonNullable, + classification: NonNullable, + ): NonNullable { + const resultJson = parseObject(latestRun.resultJson); + const providerQuotaMetadata = classification.kind === "provider_quota" + ? { + errorFamily: "provider_quota", + retryNotBefore: classification.retryAt.toISOString(), + transientRetryNotBefore: classification.retryAt.toISOString(), + providerQuotaRetryNotBefore: classification.retryAt.toISOString(), + } + : { errorFamily: "configuration_incomplete" }; + const errorCode = classification.kind; + + return { + ...latestRun, + errorCode, + resultJson: { + ...resultJson, + ...providerQuotaMetadata, + recoveryClassification: errorCode, + }, + }; + } + + async function scheduleProviderQuotaRecoveryMonitor(input: { + issue: typeof issues.$inferSelect; + latestRun: NonNullable; + classification: Extract, { kind: "provider_quota" }>; + }) { + if (input.issue.status !== "in_progress" && input.issue.status !== "in_review") return null; + + const targetAgentId = getAdapterFailureRecoveryTargetAgentId(input.issue); + if (!targetAgentId || input.latestRun.agentId !== targetAgentId) return null; + + const previousPolicy = normalizeIssueExecutionPolicy(input.issue.executionPolicy ?? null); + const retryTargetDescription = input.issue.status === "in_review" + ? "the active review participant" + : "the original assignee"; + const policy = { + ...(previousPolicy ?? { mode: "normal" as const, commentRequired: true, stages: [] }), + monitor: { + nextCheckAt: input.classification.retryAt.toISOString(), + notes: input.classification.parsedResetTime + ? `Provider usage quota reached; retry ${retryTargetDescription} at the provider reset time.` + : `Provider usage quota reached; retry ${retryTargetDescription} after the default recovery backoff.`, + scheduledBy: "assignee" as const, + kind: "external_service" as const, + serviceName: PROVIDER_QUOTA_MONITOR_SERVICE_NAME, + externalRef: input.latestRun.id, + timeoutAt: null, + maxAttempts: null, + recoveryPolicy: "wake_owner" as const, + }, + }; + const transition = applyIssueMonitorPolicyTransition({ + issue: input.issue, + policy, + previousPolicy, + requestedStatus: input.issue.status, + requestedAssigneePatch: {}, + actor: { agentId: null, userId: null }, + monitorExplicitlyUpdated: true, + }); + const updated = await issuesSvc.update(input.issue.id, { + ...transition.patch, + executionPolicy: policy, + }); + if (!updated) return null; + + await logActivity(db, { + companyId: input.issue.companyId, + actorType: "system", + actorId: "recovery", + agentId: null, + runId: input.latestRun.id, + action: "issue.monitor_scheduled", + entityType: "issue", + entityId: input.issue.id, + details: { + identifier: input.issue.identifier, + source: "recovery.provider_quota", + latestRunId: input.latestRun.id, + errorCode: "provider_quota", + nextCheckAt: input.classification.retryAt.toISOString(), + parsedResetTime: input.classification.parsedResetTime, + targetAgentId, + }, + }); + + return updated; + } + + function getAdapterFailureRecoveryTargetAgentId(issue: typeof issues.$inferSelect) { + if (issue.status !== "in_review") return issue.assigneeAgentId; + + const pendingExecutionState = parseIssueExecutionState(issue.executionState); + const participant = pendingExecutionState?.status === "pending" + ? pendingExecutionState.currentParticipant + : null; + return participant?.type === "agent" ? participant.agentId : null; + } + + function hasPendingProviderQuotaRecoveryMonitor( + issue: typeof issues.$inferSelect, + latestRun: LatestIssueRun, + now: Date, + ) { + if (!latestRun || !issue.monitorNextCheckAt || issue.monitorNextCheckAt.getTime() <= now.getTime()) return false; + const monitor = parseObject(parseObject(issue.executionPolicy).monitor); + return readNonEmptyString(monitor.serviceName) === PROVIDER_QUOTA_MONITOR_SERVICE_NAME && + readNonEmptyString(monitor.externalRef) === latestRun.id; + } + async function reconcileStrandedAssignedIssues(opts?: { issueCreatedAtGte?: Date | null }) { const candidates = await db .select() @@ -3237,6 +3489,7 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) reviewParticipantRequeued: 0, escalated: 0, waitingOnReviewResolved: 0, + providerQuotaMonitored: 0, recentProgressExempted: 0, skipped: 0, issueIds: [] as string[], @@ -3287,11 +3540,22 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) continue; } - const latestRun = await getLatestIssueRun(issue.companyId, issue.id); + let latestRun = await getLatestIssueRun(issue.companyId, issue.id); if (latestRun?.status === "succeeded" && await hasPersistedDurableWaitPath(issue)) { result.skipped += 1; continue; } + const recoveryNow = new Date(); + const participantLatestRunForRecovery = issue.status === "in_review" && participantAgentId + ? await getLatestIssueRunForAgent(issue.companyId, issue.id, participantAgentId) + : null; + const providerQuotaMonitorRun = issue.status === "in_review" + ? participantLatestRunForRecovery + : latestRun; + if (hasPendingProviderQuotaRecoveryMonitor(issue, providerQuotaMonitorRun, recoveryNow)) { + result.skipped += 1; + continue; + } if (isStrandedIssueRecoveryIssue(issue) && isUnsuccessfulTerminalIssueRun(latestRun)) { const updated = await escalateStrandedRecoveryIssueInPlace({ issue, @@ -3307,6 +3571,51 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) continue; } + const adapterFailureClassification = issue.status !== "in_review" && latestRun && isUnsuccessfulTerminalIssueRun(latestRun) + ? classifyAdapterFailureForRecovery(latestRun, recoveryNow) + : null; + if (latestRun && adapterFailureClassification) { + const targetAgentId = getAdapterFailureRecoveryTargetAgentId(issue); + if (!targetAgentId || latestRun.agentId !== targetAgentId) { + result.skipped += 1; + continue; + } + + if (adapterFailureClassification.kind === "provider_quota") { + const monitored = await scheduleProviderQuotaRecoveryMonitor({ + issue, + latestRun, + classification: adapterFailureClassification, + }); + if (monitored) { + latestRun = await persistAdapterFailureRecoveryClassification(latestRun, adapterFailureClassification); + result.providerQuotaMonitored += 1; + result.issueIds.push(issue.id); + continue; + } + result.skipped += 1; + continue; + } else { + const updated = await escalateStrandedAssignedIssue({ + issue, + previousStatus: issue.status as StrandedPreviousStatus, + latestRun, + recoveryCause: "configuration_incomplete", + comment: + "Paperclip classified the latest adapter failure as `configuration_incomplete`. " + + "Moving the issue to `blocked` with the configuration fix recorded instead of creating a recovery takeover.", + }); + if (updated) { + latestRun = await persistAdapterFailureRecoveryClassification(latestRun, adapterFailureClassification); + result.escalated += 1; + result.issueIds.push(issue.id); + } else { + result.skipped += 1; + } + continue; + } + } + const acceptedContinuationInteraction = await getLatestAcceptedContinuationInteraction(issue.companyId, issue.id); const acceptedInteractionResolvedAt = acceptedContinuationInteraction ? acceptedContinuationInteraction.resolvedAt ?? acceptedContinuationInteraction.updatedAt @@ -3406,11 +3715,7 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) result.skipped += 1; continue; } - const participantLatestRun = await getLatestIssueRunForAgent( - issue.companyId, - issue.id, - participantAgentId, - ); + const participantLatestRun = participantLatestRunForRecovery; if (!participantLatestRun || !isTerminalIssueRun(participantLatestRun)) { if (!agentInvokable) { @@ -3434,6 +3739,52 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) continue; } + const participantAdapterFailureClassification = isUnsuccessfulTerminalIssueRun(participantLatestRun) + ? classifyAdapterFailureForRecovery(participantLatestRun, recoveryNow) + : null; + if (participantAdapterFailureClassification?.kind === "provider_quota") { + const monitored = await scheduleProviderQuotaRecoveryMonitor({ + issue, + latestRun: participantLatestRun, + classification: participantAdapterFailureClassification, + }); + if (monitored) { + latestRun = await persistAdapterFailureRecoveryClassification( + participantLatestRun, + participantAdapterFailureClassification, + ); + result.providerQuotaMonitored += 1; + result.issueIds.push(issue.id); + } else { + result.skipped += 1; + } + continue; + } + if (participantAdapterFailureClassification?.kind === "configuration_incomplete") { + const updated = await escalateStrandedAssignedIssue({ + issue, + previousStatus: "in_review", + latestRun: participantLatestRun, + recoveryCause: "configuration_incomplete", + recoveryOwnerAgentId: participantAgentId, + comment: + "Paperclip classified the active review participant's latest adapter failure as " + + "`configuration_incomplete`. Moving the issue to `blocked` with the configuration fix " + + "recorded instead of repeatedly requeueing the reviewer.", + }); + if (updated) { + latestRun = await persistAdapterFailureRecoveryClassification( + participantLatestRun, + participantAdapterFailureClassification, + ); + result.escalated += 1; + result.issueIds.push(issue.id); + } else { + result.skipped += 1; + } + continue; + } + if (!agentInvokable) { const updated = await escalateStrandedAssignedIssue({ issue,