Recover productive stranded runs

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<string, unknown> | 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<string, unknown> | 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 () => {

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<typeof heartbeatRuns.$inferSelect, "id" | "companyId" | "agentId" | "continuationAttempt">;
 type IssueRow = Pick<
   typeof issues.$inferSelect,
   "id" | "companyId" | "identifier" | "title" | "status" | "assigneeAgentId" | "executionState" | "projectId"

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<LatestIssueRun>;
+  }) {
+    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({