From db695bd2bc13fc2b1f8bf55cf3e1c00d8b1448e8 Mon Sep 17 00:00:00 2001 From: Dotta Date: Thu, 16 Jul 2026 13:55:25 +0000 Subject: [PATCH] fix(heartbeat): release dead-run finalize barriers Co-Authored-By: Paperclip --- .../heartbeat-process-recovery.test.ts | 195 ++++++++++++++++++ server/src/services/heartbeat.ts | 105 ++++++++++ 2 files changed, 300 insertions(+) diff --git a/server/src/__tests__/heartbeat-process-recovery.test.ts b/server/src/__tests__/heartbeat-process-recovery.test.ts index 9a5e85558c8..bde1c2f91fb 100644 --- a/server/src/__tests__/heartbeat-process-recovery.test.ts +++ b/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); diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 08d86a76726..56d6ffca5f2 100644 --- a/server/src/services/heartbeat.ts +++ b/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,