From 54946350504d7d63307f2b263825c2ebb6098ec3 Mon Sep 17 00:00:00 2001 From: Dotta Date: Mon, 27 Apr 2026 13:13:27 -0500 Subject: [PATCH] Handle stranded recovery creation races Co-Authored-By: Paperclip --- .../heartbeat-process-recovery.test.ts | 67 ++++++++++++++++- server/src/services/recovery/service.ts | 72 ++++++++++++------- 2 files changed, 109 insertions(+), 30 deletions(-) diff --git a/server/src/__tests__/heartbeat-process-recovery.test.ts b/server/src/__tests__/heartbeat-process-recovery.test.ts index 6814c8695d9..0723eb4e63a 100644 --- a/server/src/__tests__/heartbeat-process-recovery.test.ts +++ b/server/src/__tests__/heartbeat-process-recovery.test.ts @@ -511,7 +511,9 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { runId, claimedAt: now, finishedAt: new Date("2026-03-19T00:05:00.000Z"), - error: input.runStatus === "succeeded" ? null : (input.runError ?? "run failed before issue advanced"), + error: input.runStatus === "succeeded" + ? null + : ("runError" in input ? input.runError : "run failed before issue advanced"), }); await db.insert(heartbeatRuns).values({ @@ -533,8 +535,12 @@ 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 : (input.runErrorCode ?? "process_lost"), - error: input.runStatus === "succeeded" ? null : (input.runError ?? "run failed before issue advanced"), + errorCode: input.runStatus === "succeeded" + ? null + : ("runErrorCode" in input ? input.runErrorCode : "process_lost"), + error: input.runStatus === "succeeded" + ? null + : ("runError" in input ? input.runError : "run failed before issue advanced"), }); await db.insert(issues).values([ @@ -1544,6 +1550,61 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { expect(comments[0]?.body).toContain(`Recovery issue: [${recovery.identifier}]`); }); + it("redacts error-code-only stranded recovery failures in issue copy", async () => { + const { companyId, agentId, issueId, runId } = await seedStrandedIssueFixture({ + status: "in_progress", + runStatus: "failed", + retryReason: "issue_continuation_needed", + runErrorCode: "adapter_exit_code", + runError: null, + }); + const heartbeat = heartbeatService(db); + + const result = await heartbeat.reconcileStrandedAssignedIssues(); + expect(result.escalated).toBe(1); + + const recovery = await expectStrandedRecoveryArtifacts({ + companyId, + agentId, + issueId, + runId, + previousStatus: "in_progress", + retryReason: "issue_continuation_needed", + }); + expect(recovery.description).toContain("Latest retry failure details were withheld from the issue thread"); + expect(recovery.description).not.toContain("- Failure: none recorded"); + + const comments = await db.select().from(issueComments).where(eq(issueComments.issueId, issueId)); + expect(comments).toHaveLength(1); + expect(comments[0]?.body).toContain("Latest retry failure details were withheld from the issue thread"); + expect(comments[0]?.body).not.toContain("- Failure: none recorded"); + }); + + it("reuses the raced stranded recovery issue when duplicate active recovery creation conflicts", async () => { + const { companyId, issueId } = await seedStrandedIssueFixture({ + status: "in_progress", + runStatus: "failed", + retryReason: "issue_continuation_needed", + }); + const heartbeat = heartbeatService(db); + + const results = await Promise.allSettled( + Array.from({ length: 8 }, () => heartbeat.reconcileStrandedAssignedIssues()), + ); + expect(results.every((result) => result.status === "fulfilled")).toBe(true); + + const recoveries = await db + .select() + .from(issues) + .where(and( + eq(issues.companyId, companyId), + eq(issues.originKind, "stranded_issue_recovery"), + eq(issues.originId, issueId), + )); + expect(recoveries).toHaveLength(1); + await expect(sourceBlockerIssueIds(companyId, issueId)).resolves.toEqual([recoveries[0]?.id]); + }); + it("blocks stranded recovery issues in place instead of creating nested recovery issues", async () => { const sourceIssueId = randomUUID(); const { companyId, agentId, issueId, runId } = await seedStrandedIssueFixture({ diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index 3bffa3a0bd9..0f435d3bca5 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -102,7 +102,7 @@ function readNonEmptyString(value: unknown): string | null { function summarizeRunFailureForIssueComment(run: LatestIssueRun) { if (!run) return null; - if (readNonEmptyString(run.error)) { + if (readNonEmptyString(run.error) || readNonEmptyString(run.errorCode)) { return " Latest retry failure details were withheld from the issue thread; inspect the linked run for evidence."; } return null; @@ -814,6 +814,16 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) ); } + function isUniqueStrandedIssueRecoveryConflict(error: unknown) { + if (!error || typeof error !== "object") return false; + const maybe = error as { code?: string; constraint?: string; message?: string }; + return maybe.code === "23505" && + ( + maybe.constraint === "issues_active_stranded_issue_recovery_uq" || + typeof maybe.message === "string" && maybe.message.includes("issues_active_stranded_issue_recovery_uq") + ); + } + async function ensureSourceIssueBlockedByStaleEvaluation(input: { sourceIssue: typeof issues.$inferSelect | null; evaluationIssue: { id: string; identifier: string | null }; @@ -1267,32 +1277,40 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) if (!ownerAgentId) return null; const prefix = await getCompanyIssuePrefix(input.issue.companyId); - const recovery = await issuesSvc.create(input.issue.companyId, { - title: `Recover stalled issue ${input.issue.identifier ?? input.issue.title}`, - description: buildStrandedIssueRecoveryDescription({ - issue: input.issue, - latestRun: input.latestRun, - previousStatus: input.previousStatus, - prefix, - }), - status: "todo", - priority: input.issue.priority, - parentId: input.issue.id, - projectId: input.issue.projectId, - goalId: input.issue.goalId, - assigneeAgentId: ownerAgentId, - originKind: STRANDED_ISSUE_RECOVERY_ORIGIN_KIND, - originId: input.issue.id, - originRunId: input.latestRun?.id ?? null, - originFingerprint: [ - STRANDED_ISSUE_RECOVERY_ORIGIN_KIND, - input.issue.companyId, - input.issue.id, - input.latestRun?.id ?? "no-run", - ].join(":"), - billingCode: input.issue.billingCode, - inheritExecutionWorkspaceFromIssueId: input.issue.id, - }); + let recovery: Awaited>; + try { + recovery = await issuesSvc.create(input.issue.companyId, { + title: `Recover stalled issue ${input.issue.identifier ?? input.issue.title}`, + description: buildStrandedIssueRecoveryDescription({ + issue: input.issue, + latestRun: input.latestRun, + previousStatus: input.previousStatus, + prefix, + }), + status: "todo", + priority: input.issue.priority, + parentId: input.issue.id, + projectId: input.issue.projectId, + goalId: input.issue.goalId, + assigneeAgentId: ownerAgentId, + originKind: STRANDED_ISSUE_RECOVERY_ORIGIN_KIND, + originId: input.issue.id, + originRunId: input.latestRun?.id ?? null, + originFingerprint: [ + STRANDED_ISSUE_RECOVERY_ORIGIN_KIND, + input.issue.companyId, + input.issue.id, + input.latestRun?.id ?? "no-run", + ].join(":"), + billingCode: input.issue.billingCode, + inheritExecutionWorkspaceFromIssueId: input.issue.id, + }); + } catch (error) { + if (!isUniqueStrandedIssueRecoveryConflict(error)) throw error; + const raced = await findOpenStrandedIssueRecoveryIssue(input.issue.companyId, input.issue.id); + if (!raced) throw error; + return raced; + } await deps.enqueueWakeup(ownerAgentId, { source: "assignment",