From 7b90eb1b358af5fa8d545ca25c47f6261e0717ef Mon Sep 17 00:00:00 2001 From: Dotta Date: Wed, 15 Apr 2026 08:54:49 -0500 Subject: [PATCH] Fix continuation recovery escalation gating --- .../heartbeat-process-recovery.test.ts | 48 ++++++++++++- server/src/services/heartbeat.ts | 70 ++++++++++++++++--- 2 files changed, 107 insertions(+), 11 deletions(-) diff --git a/server/src/__tests__/heartbeat-process-recovery.test.ts b/server/src/__tests__/heartbeat-process-recovery.test.ts index 338c702b0e3..61f89ba09c1 100644 --- a/server/src/__tests__/heartbeat-process-recovery.test.ts +++ b/server/src/__tests__/heartbeat-process-recovery.test.ts @@ -308,6 +308,9 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { runStatus: "failed" | "timed_out" | "cancelled" | "succeeded"; retryReason?: "assignment_recovery" | "issue_continuation_needed" | null; assignToUser?: boolean; + runErrorCode?: string | null; + runError?: string | null; + resultJson?: Record | null; }) { const companyId = randomUUID(); const agentId = randomUUID(); @@ -348,7 +351,10 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { runId, claimedAt: now, finishedAt: new Date("2026-03-19T00:05:00.000Z"), - error: input.runStatus === "succeeded" ? null : "run failed before issue advanced", + error: + input.runStatus === "succeeded" + ? null + : (input.runError ?? "run failed before issue advanced"), }); await db.insert(heartbeatRuns).values({ @@ -370,8 +376,15 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { startedAt: now, finishedAt: new Date("2026-03-19T00:05:00.000Z"), updatedAt: new Date("2026-03-19T00:05:00.000Z"), - errorCode: input.runStatus === "succeeded" ? null : "process_lost", - error: input.runStatus === "succeeded" ? null : "run failed before issue advanced", + errorCode: + input.runStatus === "succeeded" + ? null + : (input.runErrorCode ?? "process_lost"), + error: + input.runStatus === "succeeded" + ? null + : (input.runError ?? "run failed before issue advanced"), + resultJson: input.resultJson ?? null, }); await db.insert(issues).values({ @@ -632,6 +645,10 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { status: "in_progress", runStatus: "failed", retryReason: "issue_continuation_needed", + runErrorCode: "adapter_failed", + resultJson: { + summary: "OpenAI API request failed with 429 rate_limit_exceeded", + }, }); const heartbeat = heartbeatService(db); @@ -646,6 +663,31 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { const comments = await db.select().from(issueComments).where(eq(issueComments.issueId, issueId)); expect(comments).toHaveLength(1); expect(comments[0]?.body).toContain("retried continuation"); + expect(comments[0]?.body).toContain( + "Latest failure summary: adapter_failed: OpenAI API request failed with 429 rate_limit_exceeded", + ); + }); + + it("does not block stranded in-progress work after a successful continuation retry", async () => { + const { issueId } = await seedStrandedIssueFixture({ + status: "in_progress", + runStatus: "succeeded", + retryReason: "issue_continuation_needed", + }); + const heartbeat = heartbeatService(db); + + const result = await heartbeat.reconcileStrandedAssignedIssues(); + expect(result.dispatchRequeued).toBe(0); + expect(result.continuationRequeued).toBe(0); + expect(result.escalated).toBe(0); + expect(result.skipped).toBe(1); + expect(result.issueIds).toEqual([]); + + const issue = await db.select().from(issues).where(eq(issues.id, issueId)).then((rows) => rows[0] ?? null); + expect(issue?.status).toBe("in_progress"); + + const comments = await db.select().from(issueComments).where(eq(issueComments.issueId, issueId)); + expect(comments).toHaveLength(0); }); it("does not reconcile user-assigned work through the agent stranded-work recovery path", async () => { diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 34f7c8ba204..e626b910af3 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -92,6 +92,7 @@ const MAX_INLINE_WAKE_COMMENT_BODY_CHARS = 4_000; const MAX_INLINE_WAKE_COMMENT_BODY_TOTAL_CHARS = 12_000; const execFile = promisify(execFileCallback); const ACTIVE_HEARTBEAT_RUN_STATUSES = ["queued", "running"] as const; +const FAILED_CONTINUATION_RECOVERY_STATUSES = new Set(["failed", "timed_out", "cancelled"]); const SESSIONED_LOCAL_ADAPTERS = new Set([ "claude_local", "codex_local", @@ -519,6 +520,12 @@ function readNonEmptyString(value: unknown): string | null { return typeof value === "string" && value.trim().length > 0 ? value : null; } +function readFirstSummaryLine(value: unknown): string | null { + const text = readNonEmptyString(value); + if (!text) return null; + return text.split(/\r?\n/, 1)[0]?.trim() ?? null; +} + function normalizeLedgerBillingType(value: unknown): BillingType { const raw = readNonEmptyString(value); switch (raw) { @@ -2768,6 +2775,42 @@ export function heartbeatService(db: Db) { return queued; } + function didContinuationRecoveryFail( + latestRun: typeof heartbeatRuns.$inferSelect | null, + latestRetryReason: string | null, + ) { + return Boolean( + latestRun && + latestRetryReason === "issue_continuation_needed" && + FAILED_CONTINUATION_RECOVERY_STATUSES.has(latestRun.status), + ); + } + + function formatLatestRunFailureSummary(latestRun: typeof heartbeatRuns.$inferSelect | null) { + if (!latestRun) return null; + + const latestSummary = summarizeHeartbeatRunResultJson(latestRun.resultJson); + const summaryLine = + readFirstSummaryLine(latestSummary?.summary) ?? + readFirstSummaryLine(latestSummary?.result) ?? + readFirstSummaryLine(latestSummary?.message) ?? + readFirstSummaryLine(latestSummary?.error) ?? + readFirstSummaryLine(latestRun.error); + const errorCode = readNonEmptyString(latestRun.errorCode); + + if (errorCode && summaryLine) return `${errorCode}: ${summaryLine}`; + return summaryLine ?? errorCode; + } + + function buildStrandedAssignedIssueComment(input: { + baseComment: string; + latestRun: typeof heartbeatRuns.$inferSelect | null; + }) { + const failureSummary = formatLatestRunFailureSummary(input.latestRun); + if (!failureSummary) return input.baseComment; + return `${input.baseComment}\n\nLatest failure summary: ${failureSummary}`; + } + async function escalateStrandedAssignedIssue(input: { issue: typeof issues.$inferSelect; previousStatus: "todo" | "in_progress"; @@ -2861,9 +2904,12 @@ export function heartbeatService(db: Db) { issue, previousStatus: "todo", latestRun, - comment: - "Paperclip automatically retried dispatch for this assigned `todo` issue after a lost wake/run, " + - "but it still has no live execution path. Moving it to `blocked` so it is visible for intervention.", + comment: buildStrandedAssignedIssueComment({ + latestRun, + baseComment: + "Paperclip automatically retried dispatch for this assigned `todo` issue after a lost wake/run, " + + "but it still has no live execution path. Moving it to `blocked` so it is visible for intervention.", + }), }); if (updated) { result.escalated += 1; @@ -2891,15 +2937,18 @@ export function heartbeatService(db: Db) { continue; } - if (latestRetryReason === "issue_continuation_needed") { + if (didContinuationRecoveryFail(latestRun, latestRetryReason)) { const updated = await escalateStrandedAssignedIssue({ issue, previousStatus: "in_progress", latestRun, - comment: - "Paperclip automatically retried continuation for this assigned `in_progress` issue after its live " + - "execution disappeared, but it still has no live execution path. Moving it to `blocked` so it is " + - "visible for intervention.", + comment: buildStrandedAssignedIssueComment({ + latestRun, + baseComment: + "Paperclip automatically retried continuation for this assigned `in_progress` issue after its live " + + "execution disappeared, but it still has no live execution path. Moving it to `blocked` so it is " + + "visible for intervention.", + }), }); if (updated) { result.escalated += 1; @@ -2910,6 +2959,11 @@ export function heartbeatService(db: Db) { continue; } + if (latestRetryReason === "issue_continuation_needed" && latestRun?.status === "succeeded") { + result.skipped += 1; + continue; + } + const queued = await enqueueStrandedIssueRecovery({ issueId: issue.id, agentId,