diff --git a/server/src/__tests__/heartbeat-process-recovery.test.ts b/server/src/__tests__/heartbeat-process-recovery.test.ts index d54239851aa..64b373656fb 100644 --- a/server/src/__tests__/heartbeat-process-recovery.test.ts +++ b/server/src/__tests__/heartbeat-process-recovery.test.ts @@ -1,6 +1,6 @@ import { randomUUID } from "node:crypto"; import { spawn, type ChildProcess } from "node:child_process"; -import { eq, or, inArray } from "drizzle-orm"; +import { and, eq, or, inArray } from "drizzle-orm"; import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; import { activityLog, @@ -529,6 +529,87 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { return { companyId, agentId, runId, wakeupRequestId, issueId }; } + async function expectStrandedRecoveryArtifacts(input: { + companyId: string; + agentId: string; + issueId: string; + runId: string; + previousStatus: "todo" | "in_progress"; + retryReason: "assignment_recovery" | "issue_continuation_needed"; + }) { + const recovery = await waitForValue(async () => + db.select().from(issues).where( + and( + eq(issues.companyId, input.companyId), + eq(issues.originKind, "stranded_issue_recovery"), + eq(issues.originId, input.issueId), + ), + ).then((rows) => rows[0] ?? null), + ); + if (!recovery) throw new Error("Expected stranded issue recovery issue to be created"); + + expect(recovery).toMatchObject({ + companyId: input.companyId, + parentId: input.issueId, + assigneeAgentId: input.agentId, + originKind: "stranded_issue_recovery", + originId: input.issueId, + originRunId: input.runId, + priority: "medium", + }); + expect(recovery.title).toContain("Recover stalled issue"); + expect(recovery.description).toContain(`Previous source status: \`${input.previousStatus}\``); + expect(recovery.description).toContain(`Retry reason: \`${input.retryReason}\``); + expect(recovery.description).toContain("Fix the runtime/adapter problem"); + + const relation = await db + .select() + .from(issueRelations) + .where( + and( + eq(issueRelations.companyId, input.companyId), + eq(issueRelations.issueId, recovery.id), + eq(issueRelations.relatedIssueId, input.issueId), + eq(issueRelations.type, "blocks"), + ), + ) + .then((rows) => rows[0] ?? null); + expect(relation).toBeTruthy(); + + const wakeups = await db + .select() + .from(agentWakeupRequests) + .where(eq(agentWakeupRequests.agentId, input.agentId)); + const recoveryWakeup = wakeups.find((wakeup) => { + const payload = wakeup.payload as Record | null; + return payload?.issueId === recovery.id && + payload?.sourceIssueId === input.issueId && + payload?.strandedRunId === input.runId; + }); + expect(recoveryWakeup).toMatchObject({ + companyId: input.companyId, + reason: "issue_assigned", + source: "assignment", + }); + + const recoveryRun = recoveryWakeup?.runId + ? await db + .select() + .from(heartbeatRuns) + .where(eq(heartbeatRuns.id, recoveryWakeup.runId)) + .then((rows) => rows[0] ?? null) + : null; + expect(recoveryRun?.contextSnapshot).toMatchObject({ + issueId: recovery.id, + taskId: recovery.id, + source: "stranded_issue_recovery", + sourceIssueId: input.issueId, + strandedRunId: input.runId, + }); + + return recovery; + } + async function seedQueuedIssueRunFixture() { const companyId = randomUUID(); const agentId = randomUUID(); @@ -892,7 +973,7 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { }); it("blocks assigned todo work after the one automatic dispatch recovery was already used", async () => { - const { issueId } = await seedStrandedIssueFixture({ + const { companyId, agentId, issueId, runId } = await seedStrandedIssueFixture({ status: "todo", runStatus: "failed", retryReason: "assignment_recovery", @@ -907,10 +988,20 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { const issue = await db.select().from(issues).where(eq(issues.id, issueId)).then((rows) => rows[0] ?? null); expect(issue?.status).toBe("blocked"); + const recovery = await expectStrandedRecoveryArtifacts({ + companyId, + agentId, + issueId, + runId, + previousStatus: "todo", + retryReason: "assignment_recovery", + }); + const comments = await db.select().from(issueComments).where(eq(issueComments.issueId, issueId)); expect(comments).toHaveLength(1); expect(comments[0]?.body).toContain("retried dispatch"); expect(comments[0]?.body).toContain("Latest retry failure: `process_lost` - run failed before issue advanced."); + expect(comments[0]?.body).toContain(`Recovery issue: [${recovery.identifier}]`); }); it("assigns open unassigned blockers back to their creator agent", async () => { @@ -1197,7 +1288,7 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { expect(wakes.some((row) => row.reason === "run_liveness_continuation")).toBe(false); }); it("blocks stranded in-progress work after the continuation retry was already used", async () => { - const { issueId } = await seedStrandedIssueFixture({ + const { companyId, agentId, issueId, runId } = await seedStrandedIssueFixture({ status: "in_progress", runStatus: "failed", retryReason: "issue_continuation_needed", @@ -1212,10 +1303,20 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { const issue = await db.select().from(issues).where(eq(issues.id, issueId)).then((rows) => rows[0] ?? null); expect(issue?.status).toBe("blocked"); + const recovery = await expectStrandedRecoveryArtifacts({ + companyId, + agentId, + issueId, + runId, + previousStatus: "in_progress", + retryReason: "issue_continuation_needed", + }); + 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 retry failure: `process_lost` - run failed before issue advanced."); + expect(comments[0]?.body).toContain(`Recovery issue: [${recovery.identifier}]`); }); it("re-enqueues continuation when the latest automatic continuation succeeded without closing the issue", async () => { diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 2f100a7da9a..24813eb077b 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -164,6 +164,7 @@ export const ACTIVE_RUN_OUTPUT_SUSPICION_THRESHOLD_MS = 60 * 60 * 1000; export const ACTIVE_RUN_OUTPUT_CRITICAL_THRESHOLD_MS = 4 * 60 * 60 * 1000; export const ACTIVE_RUN_OUTPUT_PROGRESS_FLUSH_INTERVAL_MS = 60 * 1000; const STALE_ACTIVE_RUN_EVALUATION_ORIGIN_KIND = "stale_active_run_evaluation"; +const STRANDED_ISSUE_RECOVERY_ORIGIN_KIND = "stranded_issue_recovery"; const ACTIVE_RUN_OUTPUT_EVIDENCE_TAIL_BYTES = 8 * 1024; export const BOUNDED_TRANSIENT_HEARTBEAT_RETRY_DELAYS_MS = [ 2 * 60 * 1000, @@ -4127,6 +4128,7 @@ export function heartbeatService(db: Db) { return db .select({ id: heartbeatRuns.id, + agentId: heartbeatRuns.agentId, status: heartbeatRuns.status, error: heartbeatRuns.error, errorCode: heartbeatRuns.errorCode, @@ -4359,21 +4361,204 @@ export function heartbeatService(db: Db) { return { assigned, skipped, issueIds }; } + async function getCompanyIssuePrefix(companyId: string) { + return db + .select({ issuePrefix: companies.issuePrefix }) + .from(companies) + .where(eq(companies.id, companyId)) + .then((rows) => rows[0]?.issuePrefix ?? "PAP"); + } + + async function findOpenStrandedIssueRecoveryIssue(companyId: string, sourceIssueId: string) { + return db + .select() + .from(issues) + .where( + and( + eq(issues.companyId, companyId), + eq(issues.originKind, STRANDED_ISSUE_RECOVERY_ORIGIN_KIND), + eq(issues.originId, sourceIssueId), + isNull(issues.hiddenAt), + notInArray(issues.status, ["done", "cancelled"]), + ), + ) + .orderBy(desc(issues.createdAt)) + .limit(1) + .then((rows) => rows[0] ?? null); + } + + async function resolveStrandedIssueRecoveryOwnerAgentId(issue: typeof issues.$inferSelect) { + const candidateIds: string[] = []; + if (issue.assigneeAgentId) { + const assignee = await getAgent(issue.assigneeAgentId); + if (assignee?.reportsTo) candidateIds.push(assignee.reportsTo); + } + if (issue.createdByAgentId) { + const creator = await getAgent(issue.createdByAgentId); + if (creator?.reportsTo) candidateIds.push(creator.reportsTo); + candidateIds.push(issue.createdByAgentId); + } + + const roleCandidates = await db + .select() + .from(agents) + .where(and(eq(agents.companyId, issue.companyId), inArray(agents.role, ["cto", "ceo"]))) + .orderBy(sql`case when ${agents.role} = 'cto' then 0 else 1 end`, asc(agents.createdAt)); + candidateIds.push(...roleCandidates.map((agent) => agent.id)); + if (issue.assigneeAgentId) candidateIds.push(issue.assigneeAgentId); + + const seen = new Set(); + for (const agentId of candidateIds) { + if (seen.has(agentId)) continue; + seen.add(agentId); + const candidate = await getAgent(agentId); + if (!candidate || candidate.companyId !== issue.companyId) continue; + const budgetBlock = await budgets.getInvocationBlock(issue.companyId, candidate.id, { + issueId: issue.id, + projectId: issue.projectId, + }); + if (isAgentInvokable(candidate) && !budgetBlock) return candidate.id; + } + + return null; + } + + function buildStrandedIssueRecoveryDescription(input: { + issue: typeof issues.$inferSelect; + latestRun: Pick< + typeof heartbeatRuns.$inferSelect, + "id" | "agentId" | "status" | "error" | "errorCode" | "contextSnapshot" + > | null; + previousStatus: "todo" | "in_progress"; + prefix: string; + }) { + const sourceIssue = issueUiLink({ identifier: input.issue.identifier, id: input.issue.id }, input.prefix); + const runLink = input.latestRun + ? `[\`${input.latestRun.id}\`](/${input.prefix}/agents/${input.latestRun.agentId}/runs/${input.latestRun.id})` + : "none"; + const retryReason = readNonEmptyString(parseObject(input.latestRun?.contextSnapshot)?.retryReason) ?? "unknown"; + const failureSummary = summarizeRunFailureForIssueComment(input.latestRun); + + return [ + "Paperclip exhausted automatic recovery for an assigned issue and created this explicit recovery task.", + "", + "## Source", + "", + `- Source issue: ${sourceIssue}`, + `- Previous source status: \`${input.previousStatus}\``, + `- Latest retry run: ${runLink}`, + `- Latest retry status: \`${input.latestRun?.status ?? "unknown"}\``, + `- Retry reason: \`${retryReason}\``, + failureSummary ? `- Failure: ${failureSummary.trim()}` : "- Failure: none recorded", + "", + "## Required Action", + "", + "- Inspect the latest run and source issue state.", + "- Fix the runtime/adapter problem, reassign the source issue, or convert the source issue into a clear manual-review state.", + "- When the source issue has a live execution path or has been intentionally resolved, mark this recovery issue done.", + ].join("\n"); + } + + async function ensureStrandedIssueRecoveryIssue(input: { + issue: typeof issues.$inferSelect; + latestRun: Pick< + typeof heartbeatRuns.$inferSelect, + "id" | "agentId" | "status" | "error" | "errorCode" | "contextSnapshot" + > | null; + previousStatus: "todo" | "in_progress"; + }) { + const existing = await findOpenStrandedIssueRecoveryIssue(input.issue.companyId, input.issue.id); + if (existing) return existing; + + const ownerAgentId = await resolveStrandedIssueRecoveryOwnerAgentId(input.issue); + 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, + }); + + await enqueueWakeup(ownerAgentId, { + source: "assignment", + triggerDetail: "system", + reason: "issue_assigned", + payload: { + issueId: recovery.id, + sourceIssueId: input.issue.id, + strandedRunId: input.latestRun?.id ?? null, + }, + requestedByActorType: "system", + requestedByActorId: null, + contextSnapshot: { + issueId: recovery.id, + taskId: recovery.id, + wakeReason: "issue_assigned", + source: STRANDED_ISSUE_RECOVERY_ORIGIN_KIND, + sourceIssueId: input.issue.id, + strandedRunId: input.latestRun?.id ?? null, + }, + }); + + return recovery; + } + async function escalateStrandedAssignedIssue(input: { issue: typeof issues.$inferSelect; previousStatus: "todo" | "in_progress"; latestRun: Pick< typeof heartbeatRuns.$inferSelect, - "id" | "status" | "error" | "errorCode" | "contextSnapshot" + "id" | "agentId" | "status" | "error" | "errorCode" | "contextSnapshot" > | null; comment: string; }) { + const recoveryIssue = await ensureStrandedIssueRecoveryIssue({ + issue: input.issue, + previousStatus: input.previousStatus, + latestRun: input.latestRun, + }); + const blockerIds = await existingBlockerIssueIds(input.issue.companyId, input.issue.id); + const nextBlockerIds = recoveryIssue + ? [...new Set([...blockerIds, recoveryIssue.id])] + : blockerIds; const updated = await issuesSvc.update(input.issue.id, { status: "blocked", + blockedByIssueIds: nextBlockerIds, }); if (!updated) return null; - await issuesSvc.addComment(input.issue.id, input.comment, {}); + const prefix = await getCompanyIssuePrefix(input.issue.companyId); + const recoveryLine = recoveryIssue + ? [ + "", + `- Recovery issue: ${issueUiLink({ identifier: recoveryIssue.identifier, id: recoveryIssue.id }, prefix)}`, + "- Next action: the recovery owner should either restore a live execution path or record the manual resolution, then mark the recovery issue done.", + ].join("\n") + : ""; + + await issuesSvc.addComment(input.issue.id, `${input.comment}${recoveryLine}`, {}); await logActivity(db, { companyId: input.issue.companyId, @@ -4392,6 +4577,8 @@ export function heartbeatService(db: Db) { latestRunId: input.latestRun?.id ?? null, latestRunStatus: input.latestRun?.status ?? null, latestRunErrorCode: input.latestRun?.errorCode ?? null, + recoveryIssueId: recoveryIssue?.id ?? null, + blockerIssueIds: nextBlockerIds, }, }); diff --git a/ui/src/components/ActivityCharts.test.tsx b/ui/src/components/ActivityCharts.test.tsx index bfb4b47026a..e9fe837161a 100644 --- a/ui/src/components/ActivityCharts.test.tsx +++ b/ui/src/components/ActivityCharts.test.tsx @@ -56,6 +56,10 @@ function createRun(overrides: Partial = {}): HeartbeatRun { logBytes: null, logSha256: null, logCompressed: false, + lastOutputAt: null, + lastOutputSeq: 0, + lastOutputStream: null, + lastOutputBytes: null, stdoutExcerpt: null, stderrExcerpt: null, errorCode: null, diff --git a/ui/src/lib/inbox.test.ts b/ui/src/lib/inbox.test.ts index 977187ecb3d..6fef7bfbb9a 100644 --- a/ui/src/lib/inbox.test.ts +++ b/ui/src/lib/inbox.test.ts @@ -147,6 +147,10 @@ function makeRun(id: string, status: HeartbeatRun["status"], createdAt: string, logBytes: null, logSha256: null, logCompressed: false, + lastOutputAt: null, + lastOutputSeq: 0, + lastOutputStream: null, + lastOutputBytes: null, errorCode: null, externalRunId: null, processPid: null, diff --git a/ui/src/pages/Inbox.test.tsx b/ui/src/pages/Inbox.test.tsx index 8f6c4c43d38..2d12b81fd3c 100644 --- a/ui/src/pages/Inbox.test.tsx +++ b/ui/src/pages/Inbox.test.tsx @@ -141,6 +141,10 @@ describe("FailedRunInboxRow", () => { logBytes: null, logSha256: null, logCompressed: false, + lastOutputAt: null, + lastOutputSeq: 0, + lastOutputStream: null, + lastOutputBytes: null, errorCode: null, externalRunId: null, processPid: null, diff --git a/ui/storybook/stories/data-viz-misc.stories.tsx b/ui/storybook/stories/data-viz-misc.stories.tsx index ace1d11d876..9505b5a7901 100644 --- a/ui/storybook/stories/data-viz-misc.stories.tsx +++ b/ui/storybook/stories/data-viz-misc.stories.tsx @@ -114,6 +114,10 @@ function makeHeartbeatRun(overrides: Partial): HeartbeatRun { logBytes: 0, logSha256: null, logCompressed: false, + lastOutputAt: null, + lastOutputSeq: 0, + lastOutputStream: null, + lastOutputBytes: null, stdoutExcerpt: null, stderrExcerpt: null, errorCode: null,