diff --git a/server/src/__tests__/heartbeat-process-recovery.test.ts b/server/src/__tests__/heartbeat-process-recovery.test.ts index a2faad9fd81..318dbd11e90 100644 --- a/server/src/__tests__/heartbeat-process-recovery.test.ts +++ b/server/src/__tests__/heartbeat-process-recovery.test.ts @@ -472,6 +472,7 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { retryReason?: "assignment_recovery" | "issue_continuation_needed" | null; assignToUser?: boolean; activePauseHold?: boolean; + livenessState?: "completed" | "advanced" | "plan_only" | "empty_response" | "blocked" | "failed" | "needs_followup" | null; runErrorCode?: string | null; runError?: string | null; }) { @@ -545,6 +546,7 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { error: input.runStatus === "succeeded" ? null : ("runError" in input ? input.runError : "run failed before issue advanced"), + livenessState: input.livenessState ?? null, }); await db.insert(issues).values([ @@ -2019,18 +2021,21 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { expect(wakeups).toHaveLength(1); }); - it("re-enqueues continuation when the latest automatic continuation succeeded without closing the issue", async () => { + it("records productive continuation instead of recovery when the latest automatic continuation succeeded", async () => { const { agentId, issueId, runId } = await seedStrandedIssueFixture({ status: "in_progress", runStatus: "succeeded", retryReason: "issue_continuation_needed", + livenessState: "advanced", }); const heartbeat = heartbeatService(db); const result = await heartbeat.reconcileStrandedAssignedIssues(); - expect(result.continuationRequeued).toBe(1); + expect(result.continuationRequeued).toBe(0); + expect(result.productiveContinuationObserved).toBe(1); + expect(result.successfulContinuationObserved).toBe(0); expect(result.escalated).toBe(0); - expect(result.issueIds).toEqual([issueId]); + 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"); @@ -2042,14 +2047,10 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { .select() .from(heartbeatRuns) .where(eq(heartbeatRuns.agentId, agentId)); - expect(runs).toHaveLength(2); + expect(runs.map((row) => row.id)).toEqual([runId]); - const retryRun = runs.find((row) => row.id !== runId); - expect(retryRun?.id).toBeTruthy(); - expect((retryRun?.contextSnapshot as Record)?.retryReason).toBe("issue_continuation_needed"); - if (retryRun) { - await waitForRunToSettle(heartbeat, retryRun.id); - } + const wakeups = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.agentId, agentId)); + expect(wakeups).toHaveLength(1); }); it("does not reconcile user-assigned work through the agent stranded-work recovery path", async () => { diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index f239195058e..56f354d5288 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -72,7 +72,7 @@ type RecoveryWakeup = ( type LatestIssueRun = Pick< typeof heartbeatRuns.$inferSelect, - "id" | "agentId" | "status" | "error" | "errorCode" | "contextSnapshot" + "id" | "agentId" | "status" | "error" | "errorCode" | "contextSnapshot" | "livenessState" > | null; type WatchdogDecisionActor = @@ -188,6 +188,18 @@ function isUnsuccessfulTerminalIssueRun(latestRun: LatestIssueRun) { ); } +function isSuccessfulInProgressContinuationRun(latestRun: LatestIssueRun) { + return latestRun?.status === "succeeded"; +} + +function isProductiveContinuationRun(latestRun: LatestIssueRun) { + return latestRun?.status === "succeeded" && + (latestRun.livenessState === "advanced" || + latestRun.livenessState === "completed" || + latestRun.livenessState === "blocked" || + latestRun.livenessState === "needs_followup"); +} + function parseLivenessIncidentKey(incidentKey: string | null | undefined) { if (!incidentKey) return null; return parseIssueGraphLivenessIncidentKey(incidentKey); @@ -299,6 +311,7 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) error: heartbeatRuns.error, errorCode: heartbeatRuns.errorCode, contextSnapshot: heartbeatRuns.contextSnapshot, + livenessState: heartbeatRuns.livenessState, }) .from(heartbeatRuns) .where( @@ -1557,6 +1570,8 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) assignmentDispatched: 0, dispatchRequeued: 0, continuationRequeued: 0, + productiveContinuationObserved: 0, + successfulContinuationObserved: 0, orphanBlockersAssigned: 0, escalated: 0, skipped: 0, @@ -1670,6 +1685,15 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) result.skipped += 1; continue; } + if (isSuccessfulInProgressContinuationRun(latestRun)) { + if (isProductiveContinuationRun(latestRun)) { + result.productiveContinuationObserved += 1; + } else { + result.successfulContinuationObserved += 1; + } + result.skipped += 1; + continue; + } if (didAutomaticRecoveryFail(latestRun, "issue_continuation_needed")) { const failureSummary = summarizeRunFailureForIssueComment(latestRun); const updated = await escalateStrandedAssignedIssue({