From 1686cd47ee27a6d0742890131302a8cf94283028 Mon Sep 17 00:00:00 2001 From: Dotta Date: Thu, 30 Apr 2026 13:38:46 -0500 Subject: [PATCH] Harden issue recovery reliability Co-Authored-By: Paperclip --- .../__tests__/agent-live-run-routes.test.ts | 65 ++++++++++++++++++- ...eartbeat-issue-liveness-escalation.test.ts | 17 ++++- .../heartbeat-process-recovery.test.ts | 52 +++++++++------ .../__tests__/issue-blocker-attention.test.ts | 52 +++++++++++++++ .../src/__tests__/run-continuations.test.ts | 24 ++++++- server/src/routes/agents.ts | 4 +- server/src/services/heartbeat.ts | 2 +- server/src/services/issues.ts | 47 ++++++++------ .../recovery/run-liveness-continuations.ts | 5 +- server/src/services/recovery/service.ts | 10 ++- 10 files changed, 227 insertions(+), 51 deletions(-) diff --git a/server/src/__tests__/agent-live-run-routes.test.ts b/server/src/__tests__/agent-live-run-routes.test.ts index 0bfa209caab..c8243370023 100644 --- a/server/src/__tests__/agent-live-run-routes.test.ts +++ b/server/src/__tests__/agent-live-run-routes.test.ts @@ -10,7 +10,6 @@ const mockHeartbeatService = vi.hoisted(() => ({ buildRunOutputSilence: vi.fn(), getRunIssueSummary: vi.fn(), getActiveRunIssueSummaryForAgent: vi.fn(), - buildRunOutputSilence: vi.fn(), getRunLogAccess: vi.fn(), readLog: vi.fn(), })); @@ -71,7 +70,7 @@ function registerModuleMocks() { })); } -async function createApp() { +async function createApp(db: Record = {}) { const [{ agentRoutes }, { errorHandler }] = await Promise.all([ vi.importActual("../routes/agents.js"), vi.importActual("../middleware/index.js"), @@ -88,11 +87,32 @@ async function createApp() { }; next(); }); - app.use("/api", agentRoutes({} as any)); + app.use("/api", agentRoutes(db as any)); app.use(errorHandler); return app; } +function createLiveRunsDbStub(rows: Array>) { + const limit = vi.fn(async (value: number) => rows.slice(0, value)); + const orderedQuery = { + limit, + then: (resolve: (value: Array>) => unknown) => Promise.resolve(rows).then(resolve), + }; + const query = { + from: vi.fn().mockReturnThis(), + innerJoin: vi.fn().mockReturnThis(), + where: vi.fn().mockReturnThis(), + orderBy: vi.fn().mockReturnValue(orderedQuery), + }; + + return { + db: { + select: vi.fn().mockReturnValue(query), + }, + limit, + }; +} + async function requestApp( app: express.Express, buildRequest: (baseUrl: string) => request.Test, @@ -284,4 +304,43 @@ describe("agent live run routes", () => { nextOffset: 5, }); }); + + it("caps company live run polling by default", async () => { + const rows = Array.from({ length: 75 }, (_, index) => ({ + id: `run-${index}`, + companyId: "company-1", + status: "running", + invocationSource: "on_demand", + triggerDetail: "manual", + startedAt: new Date("2026-04-10T09:30:00.000Z"), + finishedAt: null, + createdAt: new Date(`2026-04-10T09:${String(index % 60).padStart(2, "0")}:00.000Z`), + agentId: "agent-1", + agentName: "Builder", + adapterType: "codex_local", + logBytes: 0, + livenessState: "healthy", + livenessReason: null, + continuationAttempt: 0, + lastUsefulActionAt: null, + nextAction: null, + lastOutputAt: null, + lastOutputSeq: null, + lastOutputStream: null, + lastOutputBytes: 0, + processStartedAt: null, + issueId: "issue-1", + })); + const { db, limit } = createLiveRunsDbStub(rows); + + const res = await requestApp( + await createApp(db), + (baseUrl) => request(baseUrl).get("/api/companies/company-1/live-runs"), + ); + + expect(res.status, JSON.stringify(res.body)).toBe(200); + expect(limit).toHaveBeenCalledWith(50); + expect(res.body).toHaveLength(50); + expect(mockHeartbeatService.buildRunOutputSilence).toHaveBeenCalledTimes(50); + }); }); diff --git a/server/src/__tests__/heartbeat-issue-liveness-escalation.test.ts b/server/src/__tests__/heartbeat-issue-liveness-escalation.test.ts index 369bb05c2b7..10ad103cd44 100644 --- a/server/src/__tests__/heartbeat-issue-liveness-escalation.test.ts +++ b/server/src/__tests__/heartbeat-issue-liveness-escalation.test.ts @@ -289,10 +289,23 @@ describeEmbeddedPostgres("heartbeat issue graph liveness escalation", () => { const heartbeat = heartbeatService(db); const first = await heartbeat.reconcileIssueGraphLiveness(); - const second = await heartbeat.reconcileIssueGraphLiveness(); expect(first.escalationsCreated).toBe(1); + const [sourceAfterFirst] = await db + .select({ updatedAt: issues.updatedAt }) + .from(issues) + .where(eq(issues.id, blockedIssueId)); + const eventsAfterFirst = await db.select().from(activityLog).where(eq(activityLog.companyId, companyId)); + expect(eventsAfterFirst.filter((event) => event.action === "issue.blockers.updated")).toHaveLength(1); + + const second = await heartbeat.reconcileIssueGraphLiveness(); + expect(second.escalationsCreated).toBe(0); + const [sourceAfterSecond] = await db + .select({ updatedAt: issues.updatedAt }) + .from(issues) + .where(eq(issues.id, blockedIssueId)); + expect(sourceAfterSecond?.updatedAt.getTime()).toBe(sourceAfterFirst?.updatedAt.getTime()); const escalations = await db .select() @@ -345,7 +358,7 @@ describeEmbeddedPostgres("heartbeat issue graph liveness escalation", () => { projectWorkspaceSourceIssueId: blockerIssueId, }, }); - expect(events.some((event) => event.action === "issue.blockers.updated")).toBe(true); + expect(events.filter((event) => event.action === "issue.blockers.updated")).toHaveLength(1); }); it("skips budget-blocked direct owners and assigns recovery to the manager fallback", async () => { diff --git a/server/src/__tests__/heartbeat-process-recovery.test.ts b/server/src/__tests__/heartbeat-process-recovery.test.ts index d66c2617ea4..048ecb94202 100644 --- a/server/src/__tests__/heartbeat-process-recovery.test.ts +++ b/server/src/__tests__/heartbeat-process-recovery.test.ts @@ -2180,8 +2180,8 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { expect(wakeups).toHaveLength(1); }); - it("records productive continuation instead of recovery when the latest automatic continuation succeeded", async () => { - const { agentId, issueId, runId } = await seedStrandedIssueFixture({ + it("does not treat a productive terminal run as healthy when in-progress work has no live path", async () => { + const { companyId, agentId, issueId } = await seedStrandedIssueFixture({ status: "in_progress", runStatus: "succeeded", retryReason: "issue_continuation_needed", @@ -2189,27 +2189,41 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { }); const heartbeat = heartbeatService(db); - const result = await heartbeat.reconcileStrandedAssignedIssues(); - expect(result.continuationRequeued).toBe(0); - expect(result.productiveContinuationObserved).toBe(1); - expect(result.successfulContinuationObserved).toBe(0); - expect(result.escalated).toBe(0); - expect(result.issueIds).toEqual([]); + const sourceIssue = await db.select().from(issues).where(eq(issues.id, issueId)).then((rows) => rows[0] ?? null); + expect(sourceIssue).toMatchObject({ + status: "in_progress", + assigneeAgentId: agentId, + assigneeUserId: null, + executionRunId: null, + }); - 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); - - const runs = await db + const activeRuns = await db .select() .from(heartbeatRuns) - .where(eq(heartbeatRuns.agentId, agentId)); - expect(runs.map((row) => row.id)).toEqual([runId]); + .where(and(eq(heartbeatRuns.companyId, companyId), inArray(heartbeatRuns.status, ["queued", "running"]))); + expect(activeRuns).toHaveLength(0); - const wakeups = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.agentId, agentId)); - expect(wakeups).toHaveLength(1); + const liveWakeups = await db + .select() + .from(agentWakeupRequests) + .where(and(eq(agentWakeupRequests.companyId, companyId), inArray(agentWakeupRequests.status, ["queued", "deferred_issue_execution"]))); + expect(liveWakeups).toHaveLength(0); + + const result = await heartbeat.reconcileStrandedAssignedIssues(); + expect(result.productiveContinuationObserved).toBe(0); + expect(result.continuationRequeued + result.escalated).toBe(1); + expect(result.issueIds).toEqual([issueId]); + + const comments = await db.select().from(issueComments).where(eq(issueComments.issueId, issueId)); + const recoveryIssues = await db + .select() + .from(issues) + .where(and(eq(issues.companyId, companyId), eq(issues.originKind, "stranded_issue_recovery"))); + const followupWakeups = await db + .select() + .from(agentWakeupRequests) + .where(and(eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.reason, "issue_continuation_needed"))); + expect(comments.length + recoveryIssues.length + followupWakeups.length).toBeGreaterThan(0); }); it("does not reconcile user-assigned work through the agent stranded-work recovery path", async () => { diff --git a/server/src/__tests__/issue-blocker-attention.test.ts b/server/src/__tests__/issue-blocker-attention.test.ts index 611921911ad..66df6959376 100644 --- a/server/src/__tests__/issue-blocker-attention.test.ts +++ b/server/src/__tests__/issue-blocker-attention.test.ts @@ -76,6 +76,9 @@ describeEmbeddedPostgres("issue blocker attention", () => { status: string; parentId?: string | null; assigneeAgentId?: string | null; + originKind?: string | null; + originId?: string | null; + originFingerprint?: string | null; }) { const id = input.id ?? randomUUID(); await db.insert(issues).values({ @@ -87,6 +90,9 @@ describeEmbeddedPostgres("issue blocker attention", () => { priority: "medium", parentId: input.parentId ?? null, assigneeAgentId: input.assigneeAgentId ?? null, + originKind: input.originKind ?? "manual", + originId: input.originId ?? null, + originFingerprint: input.originFingerprint ?? "default", }); return id; } @@ -356,6 +362,52 @@ describeEmbeddedPostgres("issue blocker attention", () => { }); }); + it("treats open liveness escalation blockers as covered waiting paths", async () => { + const { companyId, agentId } = await createCompany("PBL"); + const parentId = await insertIssue({ companyId, identifier: "PBL-1", title: "Parent", status: "blocked" }); + const cancelledLeafId = await insertIssue({ + companyId, + identifier: "PBL-2", + title: "Cancelled blocker", + status: "cancelled", + assigneeAgentId: agentId, + }); + const incidentKey = [ + "harness_liveness", + companyId, + parentId, + "blocked_by_cancelled_issue", + cancelledLeafId, + ].join(":"); + const escalationId = await insertIssue({ + companyId, + identifier: "PBL-3", + title: "Liveness escalation", + status: "todo", + assigneeAgentId: agentId, + originKind: "harness_liveness_escalation", + originId: incidentKey, + originFingerprint: [ + "harness_liveness_leaf", + companyId, + "blocked_by_cancelled_issue", + cancelledLeafId, + ].join(":"), + }); + await block({ companyId, blockerIssueId: cancelledLeafId, blockedIssueId: parentId }); + await block({ companyId, blockerIssueId: escalationId, blockedIssueId: parentId }); + + const parent = (await svc.list(companyId, { status: "blocked,todo" })).find((issue) => issue.id === parentId); + + expect(parent?.blockerAttention).toMatchObject({ + state: "covered", + reason: "active_dependency", + unresolvedBlockerCount: 2, + coveredBlockerCount: 2, + attentionBlockerCount: 0, + }); + }); + it("does not treat a scheduled retry as actively covered work", async () => { const { companyId, agentId } = await createCompany("PBY"); const parentId = await insertIssue({ companyId, identifier: "PBY-1", title: "Parent", status: "blocked" }); diff --git a/server/src/__tests__/run-continuations.test.ts b/server/src/__tests__/run-continuations.test.ts index a9e1e7b0006..ce48001835b 100644 --- a/server/src/__tests__/run-continuations.test.ts +++ b/server/src/__tests__/run-continuations.test.ts @@ -106,6 +106,28 @@ describe("run liveness continuations", () => { expect(decision.nextAttempt).toBe(2); }); + it("enqueues advanced terminal runs so progress is not mistaken for a live path", () => { + const decision = decideRunLivenessContinuation({ + run: run(), + issue: issue(), + agent: agent(), + livenessState: "advanced", + livenessReason: "Run produced concrete action evidence: created an issue comment", + nextAction: "Resume the implementation from the remaining acceptance criteria.", + budgetBlocked: false, + idempotentWakeExists: false, + }); + + expect(decision.kind).toBe("enqueue"); + if (decision.kind !== "enqueue") return; + expect(decision.payload).toMatchObject({ + issueId, + sourceRunId: runId, + livenessState: "advanced", + instruction: "Resume the implementation from the remaining acceptance criteria.", + }); + }); + it("does not enqueue a third continuation and returns an exhaustion comment", () => { const decision = decideRunLivenessContinuation({ run: run({ continuationAttempt: 2 }), @@ -126,7 +148,7 @@ describe("run liveness continuations", () => { it("skips non-actionable and guarded issues", () => { const guardedCases = [ - { livenessState: "advanced" as const }, + { livenessState: "needs_followup" as const }, { issue: issue({ status: "done" }) }, { issue: issue({ assigneeAgentId: "other-agent" }) }, { issue: issue({ executionState: { status: "pending" } }) }, diff --git a/server/src/routes/agents.ts b/server/src/routes/agents.ts index b3c7701b853..47d586687f5 100644 --- a/server/src/routes/agents.ts +++ b/server/src/routes/agents.ts @@ -2703,8 +2703,8 @@ export function agentRoutes( const companyId = req.params.companyId as string; assertCompanyAccess(req, companyId); - const minCount = readLiveRunsQueryInt(req.query.minCount, 50); - const limit = readLiveRunsQueryInt(req.query.limit, 50); + const minCount = readLiveRunsQueryInt(req.query.minCount, 50, 50); + const limit = readLiveRunsQueryInt(req.query.limit, 50, 50); const columns = { id: heartbeatRuns.id, diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 25535ed63a1..89439657c0a 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -2805,7 +2805,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) async function handleRunLivenessContinuation(run: typeof heartbeatRuns.$inferSelect) { const livenessState = run.livenessState as RunLivenessState | null; - if (livenessState !== "plan_only" && livenessState !== "empty_response") return; + if (livenessState !== "plan_only" && livenessState !== "empty_response" && livenessState !== "advanced") return; const context = parseObject(run.contextSnapshot); const issueId = readNonEmptyString(context.issueId); diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index cce66d09c95..3f89c2c2472 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -52,6 +52,7 @@ import { issueTreeControlService, type ActiveIssueTreePauseHoldGate, } from "./issue-tree-control.js"; +import { parseIssueGraphLivenessIncidentKey } from "./recovery/origins.js"; const ALL_ISSUE_STATUSES = ["backlog", "todo", "in_progress", "in_review", "blocked", "done", "cancelled"]; const MAX_ISSUE_COMMENT_PAGE_LIMIT = 500; @@ -1174,12 +1175,12 @@ async function listIssueBlockerAttentionMap( } } - const reviewNodeIds = [...nodesById.values()] - .filter((node) => node.status === "in_review") + const explicitWaitCandidateIds = [...nodesById.values()] + .filter((node) => node.status !== "done") .map((node) => node.id); const explicitWaitingIssueIds = new Set(); - if (reviewNodeIds.length > 0) { - for (const chunk of chunkList(reviewNodeIds, ISSUE_LIST_RELATED_QUERY_CHUNK_SIZE)) { + if (explicitWaitCandidateIds.length > 0) { + for (const chunk of chunkList(explicitWaitCandidateIds, ISSUE_LIST_RELATED_QUERY_CHUNK_SIZE)) { const interactionRows: Array<{ issueId: string }> = await dbOrTx .select({ issueId: issueThreadInteractions.issueId }) .from(issueThreadInteractions) @@ -1204,22 +1205,25 @@ async function listIssueBlockerAttentionMap( ), ); for (const row of approvalRows) explicitWaitingIssueIds.add(row.issueId); + } - const recoveryRows: Array<{ originId: string | null }> = await dbOrTx - .select({ originId: issues.originId }) - .from(issues) - .where( - and( - eq(issues.companyId, companyId), - eq(issues.originKind, BLOCKER_ATTENTION_OPEN_RECOVERY_ORIGIN_KIND), - isNull(issues.hiddenAt), - inArray(issues.originId, chunk), - notInArray(issues.status, BLOCKER_ATTENTION_OPEN_RECOVERY_TERMINAL_STATUSES), - ), - ); - for (const row of recoveryRows) { - if (row.originId) explicitWaitingIssueIds.add(row.originId); - } + const recoveryRows: Array<{ id: string; originId: string | null }> = await dbOrTx + .select({ id: issues.id, originId: issues.originId }) + .from(issues) + .where( + and( + eq(issues.companyId, companyId), + eq(issues.originKind, BLOCKER_ATTENTION_OPEN_RECOVERY_ORIGIN_KIND), + isNull(issues.hiddenAt), + notInArray(issues.status, BLOCKER_ATTENTION_OPEN_RECOVERY_TERMINAL_STATUSES), + ), + ); + for (const row of recoveryRows) { + const parsed = parseIssueGraphLivenessIncidentKey(row.originId); + if (!parsed || parsed.companyId !== companyId) continue; + explicitWaitingIssueIds.add(row.id); + explicitWaitingIssueIds.add(parsed.issueId); + explicitWaitingIssueIds.add(parsed.leafIssueId); } } @@ -1257,8 +1261,11 @@ async function listIssueBlockerAttentionMap( if (node.status === "done") { return { covered: true, stalled: false, sampleBlockerIdentifier: nodeSample, sampleStalledBlockerIdentifier: null }; } + if (explicitWaitingIssueIds.has(node.id)) { + return { covered: true, stalled: false, sampleBlockerIdentifier: nodeSample, sampleStalledBlockerIdentifier: null }; + } if (node.status === "in_review") { - const hasWaitingPath = activeIssueIds.has(node.id) || Boolean(node.assigneeUserId) || explicitWaitingIssueIds.has(node.id); + const hasWaitingPath = activeIssueIds.has(node.id) || Boolean(node.assigneeUserId); if (hasWaitingPath) { return { covered: true, stalled: false, sampleBlockerIdentifier: nodeSample, sampleStalledBlockerIdentifier: null }; } diff --git a/server/src/services/recovery/run-liveness-continuations.ts b/server/src/services/recovery/run-liveness-continuations.ts index 1b4d2cf4122..f9d7f01ee53 100644 --- a/server/src/services/recovery/run-liveness-continuations.ts +++ b/server/src/services/recovery/run-liveness-continuations.ts @@ -7,7 +7,7 @@ import { RECOVERY_REASON_KINDS } from "./origins.js"; export const RUN_LIVENESS_CONTINUATION_REASON = RECOVERY_REASON_KINDS.runLivenessContinuation; export const DEFAULT_MAX_LIVENESS_CONTINUATION_ATTEMPTS = 2; -const ACTIONABLE_LIVENESS_STATES = new Set(["plan_only", "empty_response"]); +const ACTIONABLE_LIVENESS_STATES = new Set(["plan_only", "empty_response", "advanced"]); const CONTINUATION_ACTIVE_ISSUE_STATUSES = new Set(["todo", "in_progress"]); // A prior adapter error should not permanently suppress bounded liveness // continuations; the max-attempt/idempotency guards prevent unbounded retries. @@ -127,6 +127,9 @@ export function decideRunLivenessContinuation(input: { if (budgetBlocked) { return { kind: "skip", reason: "budget hard stop blocks continuation" }; } + if (livenessState === "advanced" && !nextAction) { + return { kind: "skip", reason: "advanced run has no concrete continuation instruction" }; + } const currentAttempt = readContinuationAttempt(run.continuationAttempt); if (currentAttempt >= maxAttempts) { diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index fc9fb7e1f64..74ba45e19a0 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -189,7 +189,7 @@ function isUnsuccessfulTerminalIssueRun(latestRun: LatestIssueRun) { } function isSuccessfulInProgressContinuationRun(latestRun: LatestIssueRun) { - return latestRun?.status === "succeeded"; + return latestRun?.status === "succeeded" && latestRun.livenessState !== "advanced"; } function isProductiveContinuationRun(latestRun: LatestIssueRun) { @@ -2250,10 +2250,16 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) }) { const blockerIds = await existingBlockerIssueIds(input.issue.companyId, input.issue.id); const nextBlockerIds = [...new Set([...blockerIds, input.escalationIssueId])]; + const isAlreadyBlockedByEscalation = blockerIds.includes(input.escalationIssueId); + const isAlreadyBlocked = input.issue.status === "blocked"; + if (isAlreadyBlockedByEscalation && isAlreadyBlocked) { + return input.issue; + } + const update: Partial & { blockedByIssueIds: string[] } = { blockedByIssueIds: nextBlockerIds, }; - if (input.issue.status !== "blocked") { + if (!isAlreadyBlocked) { update.status = "blocked"; }