diff --git a/server/src/__tests__/heartbeat-process-recovery.test.ts b/server/src/__tests__/heartbeat-process-recovery.test.ts index 8f870ac602c..6ed72ee9977 100644 --- a/server/src/__tests__/heartbeat-process-recovery.test.ts +++ b/server/src/__tests__/heartbeat-process-recovery.test.ts @@ -921,6 +921,61 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { expect(comments[0]?.body).toContain(`Recovery issue: [${recovery.identifier}]`); }); + it("blocks failed recovery work in place during immediate terminal-run cleanup", async () => { + const sourceIssueId = randomUUID(); + const { companyId, agentId, runId, issueId } = await seedRunFixture({ + agentStatus: "idle", + processPid: 999_999_999, + processLossRetryCount: 1, + }); + await db + .update(issues) + .set({ + title: "Recover stalled issue PAP-1", + originKind: "stranded_issue_recovery", + originId: sourceIssueId, + }) + .where(eq(issues.id, issueId)); + const heartbeat = heartbeatService(db); + + const result = await heartbeat.reapOrphanedRuns(); + expect(result.reaped).toBe(1); + expect(result.runIds).toEqual([runId]); + + const runs = await db + .select() + .from(heartbeatRuns) + .where(eq(heartbeatRuns.agentId, agentId)); + expect(runs).toHaveLength(1); + expect(runs[0]?.status).toBe("failed"); + + const recoveryIssue = await waitForValue(async () => + db.select().from(issues).where(eq(issues.id, issueId)).then((rows) => { + const issue = rows[0] ?? null; + return issue?.status === "blocked" ? issue : null; + }) + ); + expect(recoveryIssue?.assigneeAgentId).toBe(agentId); + expect(recoveryIssue?.originKind).toBe("stranded_issue_recovery"); + expect(recoveryIssue?.originId).toBe(sourceIssueId); + expect(recoveryIssue?.executionRunId).toBeNull(); + + const nestedRecoveries = await db + .select() + .from(issues) + .where(and(eq(issues.companyId, companyId), eq(issues.originKind, "stranded_issue_recovery"), eq(issues.originId, issueId))); + expect(nestedRecoveries).toHaveLength(0); + + const comments = await waitForValue(async () => { + const rows = await db.select().from(issueComments).where(eq(issueComments.issueId, issueId)); + return rows.length > 0 ? rows : null; + }); + expect(comments).toHaveLength(1); + expect(comments[0]?.body).toContain("stopped automatic stranded-work recovery"); + expect(comments[0]?.body).toContain("recovery issues do not create nested `stranded_issue_recovery` issues"); + expect(comments[0]?.body).toContain("Latest retry failure: `process_lost`"); + }); + it("does not block paused-tree work when immediate continuation recovery is suppressed by the hold", async () => { const { companyId, agentId, runId, issueId } = await seedRunFixture({ agentStatus: "idle", @@ -1441,6 +1496,51 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { expect(comments[0]?.body).toContain(`Recovery issue: [${recovery.identifier}]`); }); + it("blocks stranded recovery issues in place instead of creating nested recovery issues", async () => { + const sourceIssueId = randomUUID(); + const { companyId, agentId, issueId, runId } = await seedStrandedIssueFixture({ + status: "in_progress", + runStatus: "failed", + }); + await db + .update(issues) + .set({ + title: "Recover stalled issue PAP-1", + originKind: "stranded_issue_recovery", + originId: sourceIssueId, + }) + .where(eq(issues.id, issueId)); + const heartbeat = heartbeatService(db); + + const result = await heartbeat.reconcileStrandedAssignedIssues(); + expect(result.dispatchRequeued).toBe(0); + expect(result.continuationRequeued).toBe(0); + expect(result.escalated).toBe(1); + expect(result.issueIds).toEqual([issueId]); + + const recoveryIssue = await db.select().from(issues).where(eq(issues.id, issueId)).then((rows) => rows[0] ?? null); + expect(recoveryIssue?.status).toBe("blocked"); + expect(recoveryIssue?.assigneeAgentId).toBe(agentId); + expect(recoveryIssue?.originKind).toBe("stranded_issue_recovery"); + expect(recoveryIssue?.originId).toBe(sourceIssueId); + + const nestedRecoveries = await db + .select() + .from(issues) + .where(and(eq(issues.companyId, companyId), eq(issues.originKind, "stranded_issue_recovery"), eq(issues.originId, issueId))); + expect(nestedRecoveries).toHaveLength(0); + + const runs = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.agentId, agentId)); + expect(runs).toHaveLength(1); + expect(runs[0]?.id).toBe(runId); + + const comments = await db.select().from(issueComments).where(eq(issueComments.issueId, issueId)); + expect(comments).toHaveLength(1); + expect(comments[0]?.body).toContain("stopped automatic stranded-work recovery"); + expect(comments[0]?.body).toContain("Latest retry failure: `process_lost` - run failed before issue advanced."); + expect(comments[0]?.body).toContain("recovery issues do not create nested `stranded_issue_recovery` issues"); + }); + it("does not escalate paused-tree recovery when the automatic continuation retry was cancelled by the hold", async () => { const { companyId, agentId, issueId } = await seedStrandedIssueFixture({ status: "in_progress", diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 20384a262ed..81433de744c 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -100,6 +100,7 @@ import { } from "./execution-workspace-policy.js"; import { instanceSettingsService } from "./instance-settings.js"; import { + RECOVERY_ORIGIN_KINDS, RUN_LIVENESS_CONTINUATION_REASON, buildRunLivenessContinuationIdempotencyKey, decideRunLivenessContinuation, @@ -6030,6 +6031,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) } const shouldBlockImmediately = + issue.originKind === RECOVERY_ORIGIN_KINDS.strandedIssueRecovery || !recoveryAgentInvokable || !recoveryAgent || didAutomaticRecoveryFail(run, issue.status === "todo" ? "assignment_recovery" : "issue_continuation_needed"); diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index f20dd05915f..213dbd93006 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -178,6 +178,19 @@ function isAgentInvokable(agent: typeof agents.$inferSelect | null | undefined) return Boolean(agent && !["paused", "terminated", "pending_approval"].includes(agent.status)); } +function isStrandedIssueRecoveryIssue(issue: Pick) { + return issue.originKind === STRANDED_ISSUE_RECOVERY_ORIGIN_KIND; +} + +function isUnsuccessfulTerminalIssueRun(latestRun: LatestIssueRun) { + return Boolean( + latestRun && + UNSUCCESSFUL_HEARTBEAT_RUN_TERMINAL_STATUSES.includes( + latestRun.status as (typeof UNSUCCESSFUL_HEARTBEAT_RUN_TERMINAL_STATUSES)[number], + ), + ); +} + function parseLivenessIncidentKey(incidentKey: string | null | undefined) { if (!incidentKey) return null; return parseIssueGraphLivenessIncidentKey(incidentKey); @@ -1248,6 +1261,8 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) latestRun: LatestIssueRun; previousStatus: "todo" | "in_progress"; }) { + if (isStrandedIssueRecoveryIssue(input.issue)) return null; + const existing = await findOpenStrandedIssueRecoveryIssue(input.issue.companyId, input.issue.id); if (existing) return existing; @@ -1306,6 +1321,78 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) return recovery; } + function buildRecoveryIssueInPlaceEscalationComment(input: { + issue: typeof issues.$inferSelect; + previousStatus: "todo" | "in_progress"; + latestRun: LatestIssueRun; + prefix: string; + }) { + const runLink = input.latestRun + ? runUiLink({ id: input.latestRun.id, agentId: input.latestRun.agentId }, input.prefix) + : "none"; + const retryReason = readNonEmptyString(parseObject(input.latestRun?.contextSnapshot)?.retryReason) ?? "none"; + const failureSummary = summarizeRunFailureForIssueComment(input.latestRun); + + return [ + "Paperclip stopped automatic stranded-work recovery for this recovery issue.", + "", + `- Recovery issue: ${issueUiLink({ identifier: input.issue.identifier, id: input.issue.id }, input.prefix)}`, + `- Previous status: \`${input.previousStatus}\``, + `- Latest run: ${runLink}`, + `- Latest run status: \`${input.latestRun?.status ?? "unknown"}\``, + `- Retry reason: \`${retryReason}\``, + failureSummary ? `- Failure: ${failureSummary.trim()}` : "- Failure: none recorded", + "- Guard: recovery issues do not create nested `stranded_issue_recovery` issues.", + "", + "Next action: the current recovery owner should inspect the failed run evidence, restore a live execution path or record the manual resolution, then move this recovery issue out of `blocked`.", + ].join("\n"); + } + + async function escalateStrandedRecoveryIssueInPlace(input: { + issue: typeof issues.$inferSelect; + previousStatus: "todo" | "in_progress"; + latestRun: LatestIssueRun; + }) { + const updated = await issuesSvc.update(input.issue.id, { status: "blocked" }); + if (!updated) return null; + + const prefix = await getCompanyIssuePrefix(input.issue.companyId); + await issuesSvc.addComment( + input.issue.id, + buildRecoveryIssueInPlaceEscalationComment({ + issue: input.issue, + previousStatus: input.previousStatus, + latestRun: input.latestRun, + prefix, + }), + {}, + ); + + await logActivity(db, { + companyId: input.issue.companyId, + actorType: "system", + actorId: "system", + agentId: null, + runId: null, + action: "issue.updated", + entityType: "issue", + entityId: input.issue.id, + details: { + identifier: input.issue.identifier, + status: "blocked", + previousStatus: input.previousStatus, + source: "recovery.reconcile_stranded_recovery_issue", + latestRunId: input.latestRun?.id ?? null, + latestRunStatus: input.latestRun?.status ?? null, + latestRunErrorCode: input.latestRun?.errorCode ?? null, + originKind: input.issue.originKind, + originId: input.issue.originId, + }, + }); + + return updated; + } + async function existingBlockerIssueIds(companyId: string, issueId: string) { return db .select({ blockerIssueId: issueRelations.issueId }) @@ -1348,6 +1435,14 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) latestRun: LatestIssueRun; comment: string; }) { + if (isStrandedIssueRecoveryIssue(input.issue)) { + return escalateStrandedRecoveryIssueInPlace({ + issue: input.issue, + previousStatus: input.previousStatus, + latestRun: input.latestRun, + }); + } + const recoveryIssue = await ensureStrandedIssueRecoveryIssue({ issue: input.issue, previousStatus: input.previousStatus, @@ -1448,6 +1543,21 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) } const latestRun = await getLatestIssueRun(issue.companyId, issue.id); + if (isStrandedIssueRecoveryIssue(issue) && isUnsuccessfulTerminalIssueRun(latestRun)) { + const updated = await escalateStrandedRecoveryIssueInPlace({ + issue, + previousStatus: issue.status as "todo" | "in_progress", + latestRun, + }); + if (updated) { + result.escalated += 1; + result.issueIds.push(issue.id); + } else { + result.skipped += 1; + } + continue; + } + if (issue.status === "todo") { if (!latestRun || latestRun.status === "succeeded") { result.skipped += 1;