fix(heartbeat): release dead-run finalize barriers

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

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

@@ -46,6 +46,7 @@ import {
   startEmbeddedPostgresTestDatabase,
 } from "./helpers/embedded-postgres.js";
 import { runningProcesses } from "../adapters/index.ts";
+import { issueService } from "../services/issues.js";
 const mockTelemetryClient = vi.hoisted(() => ({ track: vi.fn() }));
 const mockTrackAgentFirstHeartbeat = vi.hoisted(() => vi.fn());
 const mockTerminateLocalService = vi.hoisted(() => vi.fn());
@@ -1953,6 +1954,200 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
     expect(lease?.releasedAt).toBeTruthy();
   });
 
+  it("synthesizes one finalize barrier for a dead done blocker and wakes its dependent", async () => {
+    const { companyId, agentId, runId, issueId: blockerIssueId } = await seedRunFixture({
+      agentStatus: "idle",
+      processPid: null,
+      processGroupId: null,
+    });
+    const projectId = randomUUID();
+    const executionWorkspaceId = randomUUID();
+    const dependentIssueId = randomUUID();
+
+    await db.insert(projects).values({
+      id: projectId,
+      companyId,
+      name: "Dead-run finalize project",
+      status: "in_progress",
+    });
+    await db.insert(executionWorkspaces).values({
+      id: executionWorkspaceId,
+      companyId,
+      projectId,
+      mode: "shared_workspace",
+      strategyType: "local_path",
+      name: "Dead-run finalize workspace",
+      status: "active",
+      providerType: "local_fs",
+    });
+    await db
+      .update(issues)
+      .set({
+        projectId,
+        status: "done",
+        executionWorkspaceId,
+        completedAt: new Date("2026-03-19T00:01:00.000Z"),
+      })
+      .where(eq(issues.id, blockerIssueId));
+    await db.insert(issues).values({
+      id: dependentIssueId,
+      companyId,
+      projectId,
+      title: "Dependent waiting on dead-run finalize",
+      status: "blocked",
+      priority: "medium",
+      assigneeAgentId: agentId,
+      responsibleUserId: "responsible-user",
+      issueNumber: 2,
+      identifier: `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}-2`,
+    });
+    await db.insert(issueRelations).values({
+      companyId,
+      issueId: blockerIssueId,
+      relatedIssueId: dependentIssueId,
+      type: "blocks",
+    });
+    await db.insert(workspaceOperations).values({
+      companyId,
+      executionWorkspaceId,
+      heartbeatRunId: runId,
+      issueId: blockerIssueId,
+      phase: "adapter_execute",
+      status: "succeeded",
+      metadata: { qaFixture: "dead-run-finalize", preFinalize: true },
+      startedAt: new Date("2026-03-19T00:00:30.000Z"),
+      finishedAt: new Date("2026-03-19T00:00:45.000Z"),
+    });
+
+    const before = await issueService(db).getDependencyReadiness(dependentIssueId);
+    expect(before).toMatchObject({
+      allBlockersDone: false,
+      isDependencyReady: false,
+      unresolvedBlockerCount: 1,
+      pendingFinalizeBlockerIssueIds: [blockerIssueId],
+    });
+
+    await Promise.all([
+      heartbeatService(db).reapOrphanedRuns(),
+      heartbeatService(db).reapOrphanedRuns(),
+    ]);
+
+    const finalizeRows = await db
+      .select({
+        phase: workspaceOperations.phase,
+        status: workspaceOperations.status,
+        metadata: workspaceOperations.metadata,
+      })
+      .from(workspaceOperations)
+      .where(and(
+        eq(workspaceOperations.heartbeatRunId, runId),
+        eq(workspaceOperations.phase, "workspace_finalize"),
+      ));
+    expect(finalizeRows).toEqual([{
+      phase: "workspace_finalize",
+      status: "succeeded",
+      metadata: expect.objectContaining({ synthetic: true }),
+    }]);
+
+    const after = await issueService(db).getDependencyReadiness(dependentIssueId);
+    expect(after).toMatchObject({
+      allBlockersDone: true,
+      isDependencyReady: true,
+      unresolvedBlockerCount: 0,
+      pendingFinalizeBlockerIssueIds: [],
+    });
+
+    const reconciled = await heartbeatService(db).reconcileIssueGraphLiveness();
+    expect(reconciled.dependencyWakesHealed).toBe(1);
+    expect(reconciled.dependencyWakeIssueIds).toEqual([dependentIssueId]);
+
+    const wake = await db
+      .select({
+        reason: agentWakeupRequests.reason,
+        idempotencyKey: agentWakeupRequests.idempotencyKey,
+      })
+      .from(agentWakeupRequests)
+      .where(and(
+        eq(agentWakeupRequests.agentId, agentId),
+        eq(agentWakeupRequests.reason, "issue_blockers_resolved"),
+      ))
+      .then((rows) => rows[0] ?? null);
+    expect(wake).toEqual({
+      reason: "issue_blockers_resolved",
+      idempotencyKey: `issue_blockers_resolved:${dependentIssueId}:${blockerIssueId}`,
+    });
+  });
+
+  it("leaves an existing successful live workspace finalize unchanged when reaping a dead done run", async () => {
+    const { companyId, runId, issueId } = await seedRunFixture({
+      agentStatus: "idle",
+      processPid: null,
+      processGroupId: null,
+    });
+    const projectId = randomUUID();
+    const executionWorkspaceId = randomUUID();
+
+    await db.insert(projects).values({
+      id: projectId,
+      companyId,
+      name: "Live finalize project",
+      status: "in_progress",
+    });
+    await db.insert(executionWorkspaces).values({
+      id: executionWorkspaceId,
+      companyId,
+      projectId,
+      mode: "shared_workspace",
+      strategyType: "local_path",
+      name: "Live finalize workspace",
+      status: "active",
+      providerType: "local_fs",
+    });
+    await db
+      .update(issues)
+      .set({
+        projectId,
+        status: "done",
+        executionWorkspaceId,
+        completedAt: new Date("2026-03-19T00:01:00.000Z"),
+      })
+      .where(eq(issues.id, issueId));
+    await db.insert(workspaceOperations).values([
+      {
+        companyId,
+        executionWorkspaceId,
+        heartbeatRunId: runId,
+        issueId,
+        phase: "adapter_execute",
+        status: "succeeded",
+        startedAt: new Date("2026-03-19T00:00:30.000Z"),
+        finishedAt: new Date("2026-03-19T00:00:45.000Z"),
+      },
+      {
+        companyId,
+        executionWorkspaceId,
+        heartbeatRunId: runId,
+        issueId,
+        phase: "workspace_finalize",
+        status: "succeeded",
+        metadata: { adapterType: "codex_local" },
+        startedAt: new Date("2026-03-19T00:00:50.000Z"),
+        finishedAt: new Date("2026-03-19T00:00:55.000Z"),
+      },
+    ]);
+
+    await heartbeatService(db).reapOrphanedRuns();
+
+    const finalizeRows = await db
+      .select({ metadata: workspaceOperations.metadata })
+      .from(workspaceOperations)
+      .where(and(
+        eq(workspaceOperations.heartbeatRunId, runId),
+        eq(workspaceOperations.phase, "workspace_finalize"),
+      ));
+    expect(finalizeRows).toEqual([{ metadata: { adapterType: "codex_local" } }]);
+  });
+
   it.skipIf(process.platform === "win32")("reaps orphaned descendant process groups when the parent pid is already gone", async () => {
     const orphan = await spawnOrphanedProcessGroup();
     cleanupPids.add(orphan.descendantPid);

server/src/services/heartbeat.ts

@@ -12191,6 +12191,110 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
       .then((rows) => rows[0] ?? null);
   }
 
+  async function reconcileSyntheticWorkspaceFinalizeForDeadRun(
+    run: typeof heartbeatRuns.$inferSelect,
+  ) {
+    const issueId = readNonEmptyString(parseObject(run.contextSnapshot).issueId);
+    if (!issueId) return false;
+
+    const inserted = await db.transaction(async (tx) => {
+      await tx.execute(
+        sql`SELECT pg_advisory_xact_lock(hashtextextended(${`dead_run_workspace_finalize:${run.companyId}:${run.id}`}, 0))`,
+      );
+
+      const issue = await tx
+        .select({
+          id: issues.id,
+          status: issues.status,
+          executionWorkspaceId: issues.executionWorkspaceId,
+        })
+        .from(issues)
+        .where(and(eq(issues.companyId, run.companyId), eq(issues.id, issueId)))
+        .then((rows) => rows[0] ?? null);
+      if (issue?.status !== "done" || !issue.executionWorkspaceId) return false;
+
+      const existingSyntheticFinalize = await tx
+        .select({ id: workspaceOperations.id })
+        .from(workspaceOperations)
+        .where(and(
+          eq(workspaceOperations.companyId, run.companyId),
+          eq(workspaceOperations.heartbeatRunId, run.id),
+          eq(workspaceOperations.phase, "workspace_finalize"),
+          eq(workspaceOperations.status, "succeeded"),
+          sql`${workspaceOperations.metadata} ->> 'synthetic' = 'true'`,
+        ))
+        .limit(1)
+        .then((rows) => rows[0] ?? null);
+      if (existingSyntheticFinalize) return false;
+
+      const latestOperation = await tx
+        .select({
+          heartbeatRunId: workspaceOperations.heartbeatRunId,
+          phase: workspaceOperations.phase,
+          status: workspaceOperations.status,
+          cwd: workspaceOperations.cwd,
+        })
+        .from(workspaceOperations)
+        .where(and(
+          eq(workspaceOperations.companyId, run.companyId),
+          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" && latestOperation.status === "succeeded")
+      ) {
+        return false;
+      }
+
+      const finalizedAt = new Date();
+      await tx.insert(workspaceOperations).values({
+        companyId: run.companyId,
+        executionWorkspaceId: issue.executionWorkspaceId,
+        heartbeatRunId: run.id,
+        issueId: issue.id,
+        phase: "workspace_finalize",
+        cwd: latestOperation.cwd,
+        status: "succeeded",
+        metadata: {
+          synthetic: true,
+          reason: "terminal_process_lost_reconciliation",
+          errorCode: run.errorCode ?? "process_lost",
+        },
+        startedAt: finalizedAt,
+        finishedAt: finalizedAt,
+        updatedAt: finalizedAt,
+      });
+      return true;
+    });
+
+    if (inserted) {
+      await logActivity(db, {
+        companyId: run.companyId,
+        actorType: "system",
+        actorId: "heartbeat_process_lost_reaper",
+        agentId: run.agentId,
+        runId: run.id,
+        action: "workspace.finalize_synthesized",
+        entityType: "issue",
+        entityId: issueId,
+        details: {
+          source: "heartbeat.reap_orphaned_runs",
+          errorCode: run.errorCode ?? "process_lost",
+        },
+      });
+    }
+
+    return inserted;
+  }
+
   async function reapOrphanedRuns(opts?: { staleThresholdMs?: number }) {
     const staleThresholdMs = opts?.staleThresholdMs ?? 0;
     const now = new Date();
@@ -12335,6 +12439,7 @@ 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;
+      await reconcileSyntheticWorkspaceFinalizeForDeadRun(finalizedRun);
       await releaseEnvironmentLeasesForRun({
         runId: finalizedRun.id,
         companyId: finalizedRun.companyId,