From 4cb32c6c644d36cb3926555a9fa9cc26b7f16b15 Mon Sep 17 00:00:00 2001 From: Dotta Date: Thu, 30 Apr 2026 15:24:01 -0500 Subject: [PATCH] Isolate advanced run liveness continuation Co-Authored-By: Paperclip --- .../heartbeat-process-recovery.test.ts | 52 ++++++++++++------- .../src/__tests__/run-continuations.test.ts | 24 ++++++++- server/src/services/heartbeat.ts | 2 +- .../recovery/run-liveness-continuations.ts | 5 +- server/src/services/recovery/service.ts | 5 +- 5 files changed, 63 insertions(+), 25 deletions(-) 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__/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/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/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..9b930874468 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -189,13 +189,12 @@ function isUnsuccessfulTerminalIssueRun(latestRun: LatestIssueRun) { } function isSuccessfulInProgressContinuationRun(latestRun: LatestIssueRun) { - return latestRun?.status === "succeeded"; + return latestRun?.status === "succeeded" && latestRun.livenessState !== "advanced"; } function isProductiveContinuationRun(latestRun: LatestIssueRun) { return latestRun?.status === "succeeded" && - (latestRun.livenessState === "advanced" || - latestRun.livenessState === "completed" || + (latestRun.livenessState === "completed" || latestRun.livenessState === "blocked" || latestRun.livenessState === "needs_followup"); }