diff --git a/server/src/__tests__/heartbeat-process-recovery.test.ts b/server/src/__tests__/heartbeat-process-recovery.test.ts index cbc54e22943..3d85828fbc4 100644 --- a/server/src/__tests__/heartbeat-process-recovery.test.ts +++ b/server/src/__tests__/heartbeat-process-recovery.test.ts @@ -473,6 +473,9 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { assignToUser?: boolean; activePauseHold?: boolean; livenessState?: "completed" | "advanced" | "plan_only" | "empty_response" | "blocked" | "failed" | "needs_followup" | null; + livenessReason?: string | null; + nextAction?: string | null; + continuationAttempt?: number | null; runErrorCode?: string | null; runError?: string | null; }) { @@ -547,6 +550,9 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { ? null : ("runError" in input ? input.runError : "run failed before issue advanced"), livenessState: input.livenessState ?? null, + livenessReason: input.livenessReason ?? null, + nextAction: input.nextAction ?? null, + continuationAttempt: input.continuationAttempt ?? 0, }); await db.insert(issues).values([ @@ -2110,6 +2116,124 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { expect(wakeups).toHaveLength(1); }); + it("queues a bounded liveness continuation for productive successful stranded in-progress work", async () => { + const { companyId, agentId, issueId, runId } = await seedStrandedIssueFixture({ + status: "in_progress", + runStatus: "succeeded", + livenessState: "advanced", + livenessReason: "Run made progress and left follow-up work", + nextAction: "Implement the remaining recovery branch.", + }); + const heartbeat = heartbeatService(db); + + const result = await heartbeat.reconcileStrandedAssignedIssues(); + expect(result.continuationRequeued).toBe(1); + expect(result.escalated).toBe(0); + expect(result.issueIds).toEqual([issueId]); + + const wakeups = await db + .select() + .from(agentWakeupRequests) + .where(and(eq(agentWakeupRequests.agentId, agentId), eq(agentWakeupRequests.reason, "run_liveness_continuation"))); + expect(wakeups).toHaveLength(1); + expect(wakeups[0]?.payload).toMatchObject({ + issueId, + sourceRunId: runId, + livenessState: "advanced", + continuationAttempt: 1, + instruction: "Implement the remaining recovery branch.", + }); + expect(wakeups[0]?.idempotencyKey).toBe(`run_liveness_continuation:${issueId}:${runId}:advanced:1`); + + const sourceRun = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)).then((rows) => rows[0] ?? null); + expect(sourceRun?.continuationAttempt).toBe(1); + + const issue = await db.select().from(issues).where(eq(issues.id, issueId)).then((rows) => rows[0] ?? null); + expect(issue?.status).toBe("in_progress"); + + const recoveryIssues = await db + .select() + .from(issues) + .where(and(eq(issues.companyId, companyId), eq(issues.originKind, "stranded_issue_recovery"))); + expect(recoveryIssues).toHaveLength(0); + }); + + it("blocks productive successful stranded work when bounded liveness continuation is exhausted", async () => { + const { companyId, agentId, issueId, runId } = await seedStrandedIssueFixture({ + status: "in_progress", + runStatus: "succeeded", + livenessState: "advanced", + livenessReason: "Run still has follow-up work", + nextAction: "Try the remaining implementation step.", + continuationAttempt: 2, + }); + const heartbeat = heartbeatService(db); + + const result = await heartbeat.reconcileStrandedAssignedIssues(); + expect(result.continuationRequeued).toBe(0); + 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("blocked"); + + const recovery = await db + .select() + .from(issues) + .where(and( + eq(issues.companyId, companyId), + eq(issues.originKind, "stranded_issue_recovery"), + eq(issues.originId, issueId), + )) + .then((rows) => rows[0] ?? null); + if (!recovery) throw new Error("Expected stranded recovery issue"); + expect(recovery).toMatchObject({ + companyId, + parentId: issueId, + assigneeAgentId: agentId, + originKind: "stranded_issue_recovery", + originId: issueId, + originRunId: runId, + }); + await expect(sourceBlockerIssueIds(companyId, issueId)).resolves.toEqual([recovery.id]); + + const comments = await db.select().from(issueComments).where(eq(issueComments.issueId, issueId)); + expect(comments).toHaveLength(1); + expect(comments[0]?.body).toContain("Bounded liveness continuation exhausted"); + expect(comments[0]?.body).toContain(`Recovery issue: [${recovery.identifier}]`); + }); + + it("does not duplicate an existing bounded liveness continuation wake for productive successful work", async () => { + const { companyId, agentId, issueId, runId } = await seedStrandedIssueFixture({ + status: "in_progress", + runStatus: "succeeded", + livenessState: "advanced", + nextAction: "Resume implementation from the recorded next action.", + }); + await db.insert(agentWakeupRequests).values({ + companyId, + agentId, + source: "automation", + triggerDetail: "system", + reason: "run_liveness_continuation", + payload: { issueId, sourceRunId: runId, livenessState: "advanced", continuationAttempt: 1 }, + status: "queued", + idempotencyKey: `run_liveness_continuation:${issueId}:${runId}:advanced:1`, + }); + const heartbeat = heartbeatService(db); + + const result = await heartbeat.reconcileStrandedAssignedIssues(); + expect(result.continuationRequeued).toBe(0); + expect(result.escalated).toBe(0); + expect(result.skipped).toBe(1); + + const wakeups = await db + .select() + .from(agentWakeupRequests) + .where(and(eq(agentWakeupRequests.agentId, agentId), eq(agentWakeupRequests.reason, "run_liveness_continuation"))); + expect(wakeups).toHaveLength(1); + }); + it("escalates a successful automatic continuation when no live post-run path remains", async () => { const { companyId, agentId, issueId, runId } = await seedStrandedIssueFixture({ status: "in_progress", @@ -2137,16 +2261,17 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { .select() .from(heartbeatRuns) .where(eq(heartbeatRuns.agentId, agentId)); - expect(runs.map((row) => row.id)).toEqual([runId]); - - const wakeups = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.agentId, agentId)); - expect(wakeups).toHaveLength(1); + const sourceIssueRuns = runs.filter((row) => (row.contextSnapshot as Record | null)?.issueId === issueId); + expect(sourceIssueRuns.map((row) => row.id)).toEqual([runId]); const recoveryIssues = await db .select() .from(issues) .where(and(eq(issues.companyId, companyId), eq(issues.originKind, "stranded_issue_recovery"))); expect(recoveryIssues).toHaveLength(1); + + const wakeups = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.agentId, agentId)); + expect(wakeups.some((row) => (row.payload as Record | null)?.issueId === recoveryIssues[0]?.id)).toBe(true); }); it("does not treat a productive terminal run as healthy when in-progress work has no live path", async () => { diff --git a/server/src/services/recovery/run-liveness-continuations.ts b/server/src/services/recovery/run-liveness-continuations.ts index 18d81630ec0..afa4c7a96c7 100644 --- a/server/src/services/recovery/run-liveness-continuations.ts +++ b/server/src/services/recovery/run-liveness-continuations.ts @@ -15,7 +15,7 @@ const CONTINUATION_ACTIVE_ISSUE_STATUSES = new Set(["todo", "in_progress"]); const CONTINUATION_AGENT_STATUSES = new Set(["active", "idle", "running", "error"]); const IDEMPOTENT_WAKE_STATUSES = ["queued", "deferred_issue_execution", "completed"]; -type HeartbeatRunRow = typeof heartbeatRuns.$inferSelect; +type HeartbeatRunRow = Pick; type IssueRow = Pick< typeof issues.$inferSelect, "id" | "companyId" | "identifier" | "title" | "status" | "assigneeAgentId" | "executionState" | "projectId" diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index 97b3e28df84..38cf03eff12 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -6,6 +6,7 @@ import { MIN_ISSUE_GRAPH_LIVENESS_AUTO_RECOVERY_LOOKBACK_HOURS, type IssueGraphLivenessAutoRecoveryPreview, type IssueGraphLivenessAutoRecoveryPreviewItem, + type RunLivenessState, } from "@paperclipai/shared"; import { agents, @@ -43,6 +44,13 @@ import { type IssueLivenessFinding, } from "./issue-graph-liveness.js"; import { isAutomaticRecoverySuppressedByPauseHold } from "./pause-hold-guard.js"; +import { + RUN_LIVENESS_CONTINUATION_REASON, + buildRunLivenessContinuationIdempotencyKey, + decideRunLivenessContinuation, + findExistingRunLivenessContinuationWake, + readContinuationAttempt, +} from "./run-liveness-continuations.js"; const EXECUTION_PATH_HEARTBEAT_RUN_STATUSES = ["queued", "running", "scheduled_retry"] as const; const UNSUCCESSFUL_HEARTBEAT_RUN_TERMINAL_STATUSES = ["failed", "cancelled", "timed_out"] as const; @@ -72,7 +80,17 @@ type RecoveryWakeup = ( type LatestIssueRun = Pick< typeof heartbeatRuns.$inferSelect, - "id" | "agentId" | "status" | "error" | "errorCode" | "contextSnapshot" | "livenessState" | "nextAction" + | "id" + | "companyId" + | "agentId" + | "status" + | "error" + | "errorCode" + | "contextSnapshot" + | "livenessState" + | "livenessReason" + | "nextAction" + | "continuationAttempt" > | null; type PostRunIssueDisposition = "terminal" | "explicitly_waiting" | "explicitly_live" | "invalid"; @@ -306,13 +324,16 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) return db .select({ id: heartbeatRuns.id, + companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId, status: heartbeatRuns.status, error: heartbeatRuns.error, errorCode: heartbeatRuns.errorCode, contextSnapshot: heartbeatRuns.contextSnapshot, livenessState: heartbeatRuns.livenessState, + livenessReason: heartbeatRuns.livenessReason, nextAction: heartbeatRuns.nextAction, + continuationAttempt: heartbeatRuns.continuationAttempt, }) .from(heartbeatRuns) .where( @@ -505,6 +526,89 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) }); } + async function continueOrEscalateSuccessfulStrandedRun(input: { + issue: typeof issues.$inferSelect; + agent: typeof agents.$inferSelect; + latestRun: NonNullable; + }) { + const { issue, agent, latestRun } = input; + const livenessState = latestRun.livenessState as RunLivenessState | null; + const nextAttempt = readContinuationAttempt(latestRun.continuationAttempt) + 1; + const idempotencyKey = buildRunLivenessContinuationIdempotencyKey({ + issueId: issue.id, + sourceRunId: latestRun.id, + livenessState: livenessState ?? "failed", + nextAttempt, + }); + const existingWake = await findExistingRunLivenessContinuationWake(db, { + companyId: issue.companyId, + idempotencyKey, + }); + const budgetBlocked = await isInvocationBudgetBlocked(issue, agent.id); + if (budgetBlocked) return "skipped" as const; + + const decision = decideRunLivenessContinuation({ + run: latestRun, + issue, + agent, + livenessState, + livenessReason: latestRun.livenessReason, + nextAction: latestRun.nextAction, + budgetBlocked, + idempotentWakeExists: Boolean(existingWake), + }); + + if (decision.kind === "enqueue") { + const queued = await deps.enqueueWakeup(agent.id, { + source: "automation", + triggerDetail: "system", + reason: RUN_LIVENESS_CONTINUATION_REASON, + payload: decision.payload, + contextSnapshot: decision.contextSnapshot, + idempotencyKey: decision.idempotencyKey, + requestedByActorType: "system", + requestedByActorId: "recovery.reconcile_stranded_assigned_issue", + }); + if (!queued) return "skipped" as const; + + await db + .update(heartbeatRuns) + .set({ + continuationAttempt: decision.nextAttempt, + updatedAt: new Date(), + }) + .where(eq(heartbeatRuns.id, latestRun.id)); + return "queued" as const; + } + + if (decision.kind === "skip" && decision.reason === "continuation wake already exists for this source run and attempt") { + return "skipped" as const; + } + + const continuationDetail = decision.kind === "exhausted" + ? `\n\n${decision.comment}` + : [ + "", + "", + "Paperclip did not queue an automatic continuation because the latest successful run did not leave a safe runnable action path.", + "", + `- Liveness state: \`${livenessState ?? "unknown"}\``, + `- Reason: ${decision.reason}`, + ].join("\n"); + + const updated = await escalateStrandedAssignedIssue({ + issue, + previousStatus: "in_progress", + latestRun, + comment: + "Paperclip observed a successful terminal run for this assigned `in_progress` issue, " + + "but no live execution, queued wake, waiting interaction, approval, or recovery owner now holds the next action." + + continuationDetail + + "\n\nMoving it to `blocked` so it is visible for intervention.", + }); + return updated ? "escalated" as const : "skipped" as const; + } + async function isInvocationBudgetBlocked(issue: typeof issues.$inferSelect, agentId: string) { const budgetBlock = await budgets.getInvocationBlock(issue.companyId, agentId, { issueId: issue.id, @@ -1809,6 +1913,24 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) continue; } + if (latestRun?.status === "succeeded") { + const successfulRunRecovery = await continueOrEscalateSuccessfulStrandedRun({ + issue, + agent, + latestRun, + }); + if (successfulRunRecovery === "queued") { + result.continuationRequeued += 1; + result.issueIds.push(issue.id); + } else if (successfulRunRecovery === "escalated") { + 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({