diff --git a/server/src/__tests__/heartbeat-process-recovery.test.ts b/server/src/__tests__/heartbeat-process-recovery.test.ts index 55d1f107a76..cbc54e22943 100644 --- a/server/src/__tests__/heartbeat-process-recovery.test.ts +++ b/server/src/__tests__/heartbeat-process-recovery.test.ts @@ -2110,8 +2110,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("escalates a successful automatic continuation when no live post-run path remains", async () => { + const { companyId, agentId, issueId, runId } = await seedStrandedIssueFixture({ status: "in_progress", runStatus: "succeeded", retryReason: "issue_continuation_needed", @@ -2121,16 +2121,17 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { const result = await heartbeat.reconcileStrandedAssignedIssues(); expect(result.continuationRequeued).toBe(0); - expect(result.productiveContinuationObserved).toBe(1); + expect(result.productiveContinuationObserved).toBe(0); expect(result.successfulContinuationObserved).toBe(0); - expect(result.escalated).toBe(0); - expect(result.issueIds).toEqual([]); + expect(result.escalated).toBe(1); + expect(result.issueIds).toEqual([issueId]); const issue = await db.select().from(issues).where(eq(issues.id, issueId)).then((rows) => rows[0] ?? null); - expect(issue?.status).toBe("in_progress"); + expect(issue?.status).toBe("blocked"); const comments = await db.select().from(issueComments).where(eq(issueComments.issueId, issueId)); - expect(comments).toHaveLength(0); + expect(comments).toHaveLength(1); + expect(comments[0]?.body).toContain("successful automatic continuation"); const runs = await db .select() @@ -2140,9 +2141,15 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { const wakeups = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.agentId, agentId)); expect(wakeups).toHaveLength(1); + + const recoveryIssues = await db + .select() + .from(issues) + .where(and(eq(issues.companyId, companyId), eq(issues.originKind, "stranded_issue_recovery"))); + expect(recoveryIssues).toHaveLength(1); }); - it.fails("does not treat a productive terminal run as healthy when in-progress work has no live path", async () => { + 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", diff --git a/server/src/__tests__/run-continuations.test.ts b/server/src/__tests__/run-continuations.test.ts index 0f0f0380040..4524362c4bf 100644 --- a/server/src/__tests__/run-continuations.test.ts +++ b/server/src/__tests__/run-continuations.test.ts @@ -106,7 +106,7 @@ describe("run liveness continuations", () => { expect(decision.nextAttempt).toBe(2); }); - it.fails("treats an advanced terminal run as progress evidence, not a live continuation path", () => { + it("treats an advanced terminal run as progress evidence, not a live continuation path", () => { const decision = decideRunLivenessContinuation({ run: run(), issue: issue(), diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 15352a79283..963ad5bfb7a 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -2770,7 +2770,9 @@ 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) return; + const hasProgressFollowup = livenessState === "advanced" && Boolean(run.nextAction?.trim()); + if (livenessState !== "plan_only" && livenessState !== "empty_response" && !hasProgressFollowup) 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..18d81630ec0 100644 --- a/server/src/services/recovery/run-liveness-continuations.ts +++ b/server/src/services/recovery/run-liveness-continuations.ts @@ -8,6 +8,7 @@ export const RUN_LIVENESS_CONTINUATION_REASON = RECOVERY_REASON_KINDS.runLivenes export const DEFAULT_MAX_LIVENESS_CONTINUATION_ATTEMPTS = 2; const ACTIONABLE_LIVENESS_STATES = new Set(["plan_only", "empty_response"]); +const PROGRESS_WITH_FOLLOWUP_LIVENESS_STATES = new Set(["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. @@ -104,7 +105,15 @@ export function decideRunLivenessContinuation(input: { } = input; const maxAttempts = input.maxAttempts ?? DEFAULT_MAX_LIVENESS_CONTINUATION_ATTEMPTS; - if (!livenessState || !ACTIONABLE_LIVENESS_STATES.has(livenessState)) { + if (!livenessState) { + return { kind: "skip", reason: "liveness state is not actionable for continuation" }; + } + + const hasExplicitRunnableFollowup = Boolean(nextAction?.trim()); + const livenessCanContinue = + ACTIONABLE_LIVENESS_STATES.has(livenessState) || + PROGRESS_WITH_FOLLOWUP_LIVENESS_STATES.has(livenessState) && hasExplicitRunnableFollowup; + if (!livenessCanContinue) { return { kind: "skip", reason: "liveness state is not actionable for continuation" }; } if (!issue) return { kind: "skip", reason: "issue not found" }; diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index f544f4085fa..97b3e28df84 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -72,9 +72,11 @@ type RecoveryWakeup = ( type LatestIssueRun = Pick< typeof heartbeatRuns.$inferSelect, - "id" | "agentId" | "status" | "error" | "errorCode" | "contextSnapshot" | "livenessState" + "id" | "agentId" | "status" | "error" | "errorCode" | "contextSnapshot" | "livenessState" | "nextAction" > | null; +type PostRunIssueDisposition = "terminal" | "explicitly_waiting" | "explicitly_live" | "invalid"; + type WatchdogDecisionActor = | { type: "board"; userId?: string | null; runId?: string | null } | { type: "agent"; agentId?: string | null; runId?: string | null } @@ -188,16 +190,14 @@ function isUnsuccessfulTerminalIssueRun(latestRun: LatestIssueRun) { ); } -function isSuccessfulInProgressContinuationRun(latestRun: LatestIssueRun) { - return latestRun?.status === "succeeded"; -} - -function isProductiveContinuationRun(latestRun: LatestIssueRun) { +function isProductiveProgressRun(latestRun: LatestIssueRun) { return latestRun?.status === "succeeded" && - (latestRun.livenessState === "advanced" || + ( + latestRun.livenessState === "advanced" || latestRun.livenessState === "completed" || latestRun.livenessState === "blocked" || - latestRun.livenessState === "needs_followup"); + latestRun.livenessState === "needs_followup" + ); } function parseLivenessIncidentKey(incidentKey: string | null | undefined) { @@ -312,6 +312,7 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) errorCode: heartbeatRuns.errorCode, contextSnapshot: heartbeatRuns.contextSnapshot, livenessState: heartbeatRuns.livenessState, + nextAction: heartbeatRuns.nextAction, }) .from(heartbeatRuns) .where( @@ -371,6 +372,76 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) .then((rows) => Boolean(rows[0])); } + async function hasExplicitWaitingPath(issue: Pick) { + if (issue.status === "done" || issue.status === "cancelled") return true; + if (issue.status === "blocked" || issue.status === "in_review") return true; + if (issue.executionState) return true; + + const [interaction, approval, recoveryIssue] = await Promise.all([ + db + .select({ id: issueThreadInteractions.id }) + .from(issueThreadInteractions) + .where( + and( + eq(issueThreadInteractions.companyId, issue.companyId), + eq(issueThreadInteractions.issueId, issue.id), + eq(issueThreadInteractions.status, "pending"), + ), + ) + .limit(1) + .then((rows) => rows[0] ?? null), + db + .select({ id: issueApprovals.issueId }) + .from(issueApprovals) + .innerJoin(approvals, eq(issueApprovals.approvalId, approvals.id)) + .where( + and( + eq(issueApprovals.companyId, issue.companyId), + eq(issueApprovals.issueId, issue.id), + inArray(approvals.status, ["pending", "revision_requested"]), + ), + ) + .limit(1) + .then((rows) => rows[0] ?? null), + db + .select({ id: issues.id }) + .from(issues) + .where( + and( + eq(issues.companyId, issue.companyId), + eq(issues.originKind, STRANDED_ISSUE_RECOVERY_ORIGIN_KIND), + eq(issues.originId, issue.id), + notInArray(issues.status, ["done", "cancelled"]), + ), + ) + .limit(1) + .then((rows) => rows[0] ?? null), + ]); + + return Boolean(interaction || approval || recoveryIssue); + } + + async function classifyPostRunIssueDisposition( + issue: typeof issues.$inferSelect, + latestRun: LatestIssueRun, + ): Promise { + if (issue.status === "done" || issue.status === "cancelled") return "terminal"; + if (await hasActiveExecutionPath(issue.companyId, issue.id)) return "explicitly_live"; + if (await hasExplicitWaitingPath(issue)) return "explicitly_waiting"; + if (latestRun && EXECUTION_PATH_HEARTBEAT_RUN_STATUSES.includes( + latestRun.status as (typeof EXECUTION_PATH_HEARTBEAT_RUN_STATUSES)[number], + )) { + return "explicitly_live"; + } + return "invalid"; + } + + function didAutomaticContinuationFinishWithoutLivePath(latestRun: LatestIssueRun) { + if (latestRun?.status !== "succeeded") return false; + const latestContext = parseObject(latestRun.contextSnapshot); + return readNonEmptyString(latestContext.retryReason) === "issue_continuation_needed"; + } + async function enqueueStrandedIssueRecovery(input: { issueId: string; agentId: string; @@ -1705,15 +1776,39 @@ 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; + + const postRunDisposition = await classifyPostRunIssueDisposition(issue, latestRun); + if (postRunDisposition !== "invalid") { + if (postRunDisposition === "explicitly_live" && latestRun?.status === "succeeded") { + if (isProductiveProgressRun(latestRun)) { + result.productiveContinuationObserved += 1; + } else { + result.successfulContinuationObserved += 1; + } } result.skipped += 1; continue; } + + if (didAutomaticContinuationFinishWithoutLivePath(latestRun)) { + const updated = await escalateStrandedAssignedIssue({ + issue, + previousStatus: "in_progress", + latestRun, + comment: + "Paperclip observed a successful automatic continuation for this assigned `in_progress` issue, " + + "but the run has finished and no live execution, queued wake, waiting interaction, approval, or recovery owner now holds the next action. " + + "Moving it to `blocked` so it is visible for intervention.", + }); + if (updated) { + result.escalated += 1; + result.issueIds.push(issue.id); + } else { + result.skipped += 1; + } + continue; + } + if (didAutomaticRecoveryFail(latestRun, "issue_continuation_needed")) { const failureSummary = summarizeRunFailureForIssueComment(latestRun); const updated = await escalateStrandedAssignedIssue({