fix(heartbeat): release dead-run finalize barriers

Co-Authored-By: Paperclip <noreply@paperclip.ing>

server/src/__tests__/heartbeat-process-recovery.test.ts

@@ -101,6 +101,7 @@ import {
   redactDetectedSuccessfulRunProgressSummaryForBoard,
 } from "../services/heartbeat.ts";
 import { secretService } from "../services/secrets.ts";
+import { issueService } from "../services/issues.ts";
 import {
   SUCCESSFUL_RUN_HANDOFF_EXHAUSTED_NOTICE_BODY,
   SUCCESSFUL_RUN_HANDOFF_REQUIRED_NOTICE_BODY,
@@ -583,6 +584,105 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
     return { environmentId, leaseId };
   }
 
+  async function seedDeadRunFinalizeBarrierFixture(input?: { runStatus?: "running" | "failed" }) {
+    const seeded = await seedRunFixture({
+      agentStatus: "running",
+      adapterType: "test",
+      runStatus: input?.runStatus ?? "running",
+      runErrorCode: input?.runStatus === "failed" ? "process_lost" : null,
+      runError: input?.runStatus === "failed" ? "Process lost during a prior server lifetime" : null,
+    });
+    const projectId = randomUUID();
+    const executionWorkspaceId = randomUUID();
+    const dependentAgentId = randomUUID();
+    const dependentIssueId = randomUUID();
+    const now = new Date("2026-03-19T00:01:00.000Z");
+
+    await db.insert(projects).values({
+      id: projectId,
+      companyId: seeded.companyId,
+      name: "Dead run finalize barrier project",
+      status: "in_progress",
+    });
+    await db.insert(executionWorkspaces).values({
+      id: executionWorkspaceId,
+      companyId: seeded.companyId,
+      projectId,
+      sourceIssueId: seeded.issueId,
+      mode: "shared_workspace",
+      strategyType: "project_primary",
+      name: "Dead run finalize barrier workspace",
+      status: "active",
+      cwd: "/workspace/dead-run-finalize",
+    });
+    await db
+      .update(issues)
+      .set({
+        status: "done",
+        projectId,
+        executionWorkspaceId,
+        completedAt: now,
+      })
+      .where(eq(issues.id, seeded.issueId));
+    await db
+      .update(heartbeatRuns)
+      .set({
+        contextSnapshot: {
+          issueId: seeded.issueId,
+          executionWorkspaceId,
+        },
+        ...(input?.runStatus === "failed" ? { finishedAt: now, updatedAt: now } : {}),
+      })
+      .where(eq(heartbeatRuns.id, seeded.runId));
+
+    await db.insert(agents).values({
+      id: dependentAgentId,
+      companyId: seeded.companyId,
+      name: "Dependent Agent",
+      role: "engineer",
+      status: "idle",
+      adapterType: "codex_local",
+      adapterConfig: {},
+      runtimeConfig: {},
+      permissions: {},
+    });
+    await db.insert(issues).values({
+      id: dependentIssueId,
+      companyId: seeded.companyId,
+      projectId,
+      title: "Dependent waiting for dead run finalize",
+      status: "blocked",
+      priority: "medium",
+      assigneeAgentId: dependentAgentId,
+      responsibleUserId: "responsible-user",
+      issueNumber: 2,
+      identifier: `T${seeded.companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}-2`,
+    });
+    await db.insert(issueRelations).values({
+      companyId: seeded.companyId,
+      issueId: seeded.issueId,
+      relatedIssueId: dependentIssueId,
+      type: "blocks",
+    });
+    await db.insert(workspaceOperations).values({
+      companyId: seeded.companyId,
+      executionWorkspaceId,
+      heartbeatRunId: seeded.runId,
+      issueId: seeded.issueId,
+      phase: "adapter_execute",
+      status: "succeeded",
+      startedAt: now,
+      finishedAt: now,
+    });
+
+    return {
+      ...seeded,
+      dependentAgentId,
+      dependentIssueId,
+      executionWorkspaceId,
+    };
+  }
+
   it("does not reap active adapter executions started by another heartbeat service instance", async () => {
     let releaseAdapter: (() => void) | null = null;
     const adapterStarted = new Promise<void>((resolve) => {
@@ -1303,6 +1403,126 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
     expect(checkoutReleasedIssue?.checkoutRunId).toBeNull();
   });
 
+  it("releases a done blocker finalize barrier when its run is reaped as process_lost", async () => {
+    const fixture = await seedDeadRunFinalizeBarrierFixture();
+    const heartbeat = heartbeatService(db);
+
+    const result = await heartbeat.reapOrphanedRuns();
+
+    expect(result).toMatchObject({ reaped: 1, runIds: [fixture.runId] });
+    const syntheticFinalizes = await db
+      .select()
+      .from(workspaceOperations)
+      .where(
+        and(
+          eq(workspaceOperations.heartbeatRunId, fixture.runId),
+          eq(workspaceOperations.phase, "workspace_finalize"),
+        ),
+      );
+    expect(syntheticFinalizes).toHaveLength(1);
+    expect(syntheticFinalizes[0]).toMatchObject({
+      companyId: fixture.companyId,
+      executionWorkspaceId: fixture.executionWorkspaceId,
+      issueId: fixture.issueId,
+      status: "succeeded",
+      metadata: expect.objectContaining({
+        synthetic: true,
+        reason: "process_lost",
+        source: "run.reap",
+        outcome: "terminal_without_finalize",
+      }),
+    });
+
+    await expect(issueService(db).getDependencyReadiness(fixture.dependentIssueId)).resolves.toMatchObject({
+      isDependencyReady: true,
+      pendingFinalizeBlockerIssueIds: [],
+    });
+    const wake = await db
+      .select()
+      .from(agentWakeupRequests)
+      .where(
+        and(
+          eq(agentWakeupRequests.agentId, fixture.dependentAgentId),
+          eq(agentWakeupRequests.reason, "issue_blockers_resolved"),
+        ),
+      )
+      .then((rows) => rows[0] ?? null);
+    expect(wake).toMatchObject({
+      reason: "issue_blockers_resolved",
+      idempotencyKey: `issue_blockers_resolved:${fixture.dependentIssueId}:${fixture.issueId}`,
+    });
+  });
+
+  it("sweeps a pre-existing terminal-run finalize barrier exactly once", async () => {
+    const fixture = await seedDeadRunFinalizeBarrierFixture({ runStatus: "failed" });
+    const heartbeat = heartbeatService(db);
+
+    const first = await heartbeat.reconcileTerminalRunWorkspaceFinalizes("startup_sweep");
+    const second = await heartbeat.reconcileTerminalRunWorkspaceFinalizes("interval_sweep");
+
+    expect(first).toMatchObject({
+      checked: 1,
+      created: 1,
+      wakesHealed: 1,
+      runIds: [fixture.runId],
+    });
+    expect(second).toMatchObject({ checked: 1, created: 0, wakesHealed: 0, runIds: [] });
+    const syntheticFinalizes = await db
+      .select()
+      .from(workspaceOperations)
+      .where(
+        and(
+          eq(workspaceOperations.heartbeatRunId, fixture.runId),
+          eq(workspaceOperations.phase, "workspace_finalize"),
+        ),
+      );
+    expect(syntheticFinalizes).toHaveLength(1);
+    expect(syntheticFinalizes[0]?.metadata).toMatchObject({
+      synthetic: true,
+      reason: "process_lost",
+      source: "startup_sweep",
+    });
+    await expect(issueService(db).getDependencyReadiness(fixture.dependentIssueId)).resolves.toMatchObject({
+      isDependencyReady: true,
+      pendingFinalizeBlockerIssueIds: [],
+    });
+  });
+
+  it("does not mask an explicit failed workspace finalize with a synthetic success", async () => {
+    const fixture = await seedDeadRunFinalizeBarrierFixture({ runStatus: "failed" });
+    const failedAt = new Date("2026-03-19T00:02:00.000Z");
+    await db.insert(workspaceOperations).values({
+      companyId: fixture.companyId,
+      executionWorkspaceId: fixture.executionWorkspaceId,
+      heartbeatRunId: fixture.runId,
+      issueId: fixture.issueId,
+      phase: "workspace_finalize",
+      status: "failed",
+      metadata: { errorMessage: "sync-back failed" },
+      startedAt: failedAt,
+      finishedAt: failedAt,
+    });
+
+    const result = await heartbeatService(db).reconcileTerminalRunWorkspaceFinalizes("startup_sweep");
+
+    expect(result).toMatchObject({ checked: 1, created: 0, wakesHealed: 0, runIds: [] });
+    const finalizes = await db
+      .select()
+      .from(workspaceOperations)
+      .where(
+        and(
+          eq(workspaceOperations.heartbeatRunId, fixture.runId),
+          eq(workspaceOperations.phase, "workspace_finalize"),
+        ),
+      );
+    expect(finalizes).toHaveLength(1);
+    expect(finalizes[0]).toMatchObject({ status: "failed", metadata: { errorMessage: "sync-back failed" } });
+    await expect(issueService(db).getDependencyReadiness(fixture.dependentIssueId)).resolves.toMatchObject({
+      isDependencyReady: false,
+      pendingFinalizeBlockerIssueIds: [fixture.issueId],
+    });
+  });
+
   it("interrupts running runs on graceful shutdown and queues restart recovery without recording a failure", async () => {
     const { agentId, runId, issueId, wakeupRequestId } = await seedRunFixture({
       agentStatus: "running",

server/src/index.ts

@@ -880,6 +880,14 @@ export async function startServer(): Promise<StartedServer> {
           }
         }
 
+        const finalized = await heartbeat.reconcileTerminalRunWorkspaceFinalizes("startup_sweep");
+        if (finalized.created > 0) {
+          logger.warn(
+            { ...finalized },
+            "startup terminal-run workspace-finalize reconciliation released stuck barriers",
+          );
+        }
+
         const promotion = await heartbeat.promoteDueScheduledRetries();
         await heartbeat.resumeQueuedRuns();
         const reconciled = await heartbeat.reconcileStrandedAssignedIssues();
@@ -995,6 +1003,15 @@ export async function startServer(): Promise<StartedServer> {
         // persisted queued work is still being driven forward.
         trackHeartbeatSchedulerWork(heartbeat
           .reapOrphanedRuns({ staleThresholdMs: 5 * 60 * 1000 })
+          .then(async () => {
+            const finalized = await heartbeat.reconcileTerminalRunWorkspaceFinalizes("interval_sweep");
+            if (finalized.created > 0) {
+              logger.warn(
+                { ...finalized },
+                "periodic terminal-run workspace-finalize reconciliation released stuck barriers",
+              );
+            }
+          })
           .then(() => heartbeat.promoteDueScheduledRetries())
           .then(async (promotion) => {
             await heartbeat.resumeQueuedRuns();

server/src/services/heartbeat.ts

@@ -5088,6 +5088,197 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
   };
   const budgets = budgetService(db, budgetHooks);
   const recovery = recoveryService(db, { enqueueWakeup });
+  const terminalRunFinalizeReconciliations = new Map<string, Promise<{
+    created: boolean;
+    wakeHealed: number;
+  }>>();
+
+  async function reconcileTerminalRunWorkspaceFinalize(
+    run: typeof heartbeatRuns.$inferSelect,
+    source: "run.failure" | "run.reap" | "startup_sweep" | "interval_sweep",
+  ) {
+    const existing = terminalRunFinalizeReconciliations.get(run.id);
+    if (existing) return existing;
+
+    const reconciliation = (async () => {
+      if (
+        !UNSUCCESSFUL_HEARTBEAT_RUN_TERMINAL_STATUSES.includes(
+          run.status as (typeof UNSUCCESSFUL_HEARTBEAT_RUN_TERMINAL_STATUSES)[number],
+        )
+      ) {
+        return { created: false, wakeHealed: 0 };
+      }
+
+      const context = parseObject(run.contextSnapshot);
+      const issueId = readNonEmptyString(context.issueId);
+      if (!issueId) return { created: false, wakeHealed: 0 };
+
+      const issue = await db
+        .select({
+          id: issues.id,
+          companyId: issues.companyId,
+          status: issues.status,
+          executionWorkspaceId: issues.executionWorkspaceId,
+        })
+        .from(issues)
+        .where(and(eq(issues.id, issueId), eq(issues.companyId, run.companyId)))
+        .then((rows) => rows[0] ?? null);
+      if (!issue || issue.status !== "done" || !issue.executionWorkspaceId) {
+        return { created: false, wakeHealed: 0 };
+      }
+
+      const latestOperation = await db
+        .select({
+          heartbeatRunId: workspaceOperations.heartbeatRunId,
+          phase: workspaceOperations.phase,
+        })
+        .from(workspaceOperations)
+        .where(
+          and(
+            eq(workspaceOperations.companyId, run.companyId),
+            eq(workspaceOperations.issueId, issue.id),
+            eq(workspaceOperations.executionWorkspaceId, issue.executionWorkspaceId),
+          ),
+        )
+        .orderBy(
+          desc(workspaceOperations.startedAt),
+          desc(workspaceOperations.createdAt),
+          desc(workspaceOperations.id),
+        )
+        .limit(1)
+        .then((rows) => rows[0] ?? null);
+      if (
+        !latestOperation ||
+        latestOperation.heartbeatRunId !== run.id ||
+        latestOperation.phase === "workspace_finalize"
+      ) {
+        return { created: false, wakeHealed: 0 };
+      }
+
+      const existingFinalize = await db
+        .select({ id: workspaceOperations.id })
+        .from(workspaceOperations)
+        .where(
+          and(
+            eq(workspaceOperations.companyId, run.companyId),
+            eq(workspaceOperations.heartbeatRunId, run.id),
+            eq(workspaceOperations.phase, "workspace_finalize"),
+          ),
+        )
+        .limit(1)
+        .then((rows) => rows[0] ?? null);
+      if (existingFinalize) return { created: false, wakeHealed: 0 };
+
+      const now = new Date();
+      const reason = readNonEmptyString(run.errorCode) ?? run.status;
+      await db.insert(workspaceOperations).values({
+        id: randomUUID(),
+        companyId: run.companyId,
+        executionWorkspaceId: issue.executionWorkspaceId,
+        heartbeatRunId: run.id,
+        issueId: issue.id,
+        phase: "workspace_finalize",
+        status: "succeeded",
+        metadata: {
+          synthetic: true,
+          reason,
+          source,
+          outcome: "terminal_without_finalize",
+          terminalRunStatus: run.status,
+          terminalRunErrorCode: run.errorCode ?? null,
+        },
+        startedAt: now,
+        finishedAt: now,
+        createdAt: now,
+        updatedAt: now,
+      });
+
+      const wakeResult = await recovery.reconcileResolvedDependencyWakeBackstop({
+        runId: run.id,
+        companyId: run.companyId,
+        blockerIssueId: issue.id,
+        source: "workspace.finalize",
+      });
+      return { created: true, wakeHealed: wakeResult.healed };
+    })().finally(() => {
+      terminalRunFinalizeReconciliations.delete(run.id);
+    });
+
+    terminalRunFinalizeReconciliations.set(run.id, reconciliation);
+    return reconciliation;
+  }
+
+  async function reconcileTerminalRunWorkspaceFinalizes(
+    source: "startup_sweep" | "interval_sweep" = "interval_sweep",
+  ) {
+    const blockerRows = await db
+      .select({
+        issueId: issues.id,
+        companyId: issues.companyId,
+        executionWorkspaceId: issues.executionWorkspaceId,
+      })
+      .from(issueRelations)
+      .innerJoin(issues, eq(issueRelations.issueId, issues.id))
+      .where(
+        and(
+          eq(issueRelations.type, "blocks"),
+          eq(issues.status, "done"),
+          sql`${issues.executionWorkspaceId} is not null`,
+        ),
+      );
+
+    const seenIssueIds = new Set<string>();
+    const reconciledRunIds: string[] = [];
+    let checked = 0;
+    let created = 0;
+    let wakesHealed = 0;
+
+    for (const blocker of blockerRows) {
+      if (!blocker.executionWorkspaceId || seenIssueIds.has(blocker.issueId)) continue;
+      seenIssueIds.add(blocker.issueId);
+      checked += 1;
+
+      const latestOperation = await db
+        .select({ heartbeatRunId: workspaceOperations.heartbeatRunId })
+        .from(workspaceOperations)
+        .where(
+          and(
+            eq(workspaceOperations.companyId, blocker.companyId),
+            eq(workspaceOperations.issueId, blocker.issueId),
+            eq(workspaceOperations.executionWorkspaceId, blocker.executionWorkspaceId),
+          ),
+        )
+        .orderBy(
+          desc(workspaceOperations.startedAt),
+          desc(workspaceOperations.createdAt),
+          desc(workspaceOperations.id),
+        )
+        .limit(1)
+        .then((rows) => rows[0] ?? null);
+      if (!latestOperation?.heartbeatRunId) continue;
+
+      const terminalRun = await db
+        .select()
+        .from(heartbeatRuns)
+        .where(
+          and(
+            eq(heartbeatRuns.id, latestOperation.heartbeatRunId),
+            eq(heartbeatRuns.companyId, blocker.companyId),
+            inArray(heartbeatRuns.status, [...UNSUCCESSFUL_HEARTBEAT_RUN_TERMINAL_STATUSES]),
+          ),
+        )
+        .then((rows) => rows[0] ?? null);
+      if (!terminalRun) continue;
+
+      const result = await reconcileTerminalRunWorkspaceFinalize(terminalRun, source);
+      if (!result.created) continue;
+      created += 1;
+      wakesHealed += result.wakeHealed;
+      reconciledRunIds.push(terminalRun.id);
+    }
+
+    return { checked, created, wakesHealed, runIds: reconciledRunIds };
+  }
 
   function isPlanApprovalConfirmationPayload(payload: unknown) {
     const target = parseObject(parseObject(payload).target);
@@ -10715,6 +10906,14 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
       if (!finalizedRun) finalizedRun = await getRun(run.id);
       if (!finalizedRun) continue;
       finalizedRun = await classifyAndPersistRunLiveness(finalizedRun, parseObject(finalizedRun.resultJson)) ?? finalizedRun;
+      try {
+        await reconcileTerminalRunWorkspaceFinalize(finalizedRun, "run.reap");
+      } catch (finalizeErr) {
+        logger.warn(
+          { err: finalizeErr, runId: finalizedRun.id },
+          "failed to reconcile workspace finalize after orphaned run reap",
+        );
+      }
       await releaseEnvironmentLeasesForRun({
         runId: finalizedRun.id,
         companyId: finalizedRun.companyId,
@@ -13236,6 +13435,14 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
           message,
         });
         const livenessRun = await classifyAndPersistRunLiveness(failedRun) ?? failedRun;
+        try {
+          await reconcileTerminalRunWorkspaceFinalize(livenessRun, "run.failure");
+        } catch (finalizeErr) {
+          logger.warn(
+            { err: finalizeErr, runId: livenessRun.id },
+            "failed to reconcile workspace finalize after terminal run failure",
+          );
+        }
         try {
           await completeSkillTestRunForHeartbeatOutcome({
             run: livenessRun,
@@ -15965,6 +16172,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
     reportRunActivity: clearDetachedRunWarning,
 
     reapOrphanedRuns,
+    reconcileTerminalRunWorkspaceFinalizes,
     // Override-aware scheduling-suppression check (honors the worktree
     // run-execution experimental setting). Callers outside the service that
     // gate on suppression should prefer this over the env-only resolver.