diff --git a/server/src/__tests__/heartbeat-process-recovery.test.ts b/server/src/__tests__/heartbeat-process-recovery.test.ts index 49e87b2fd5a..4b2912deb56 100644 --- a/server/src/__tests__/heartbeat-process-recovery.test.ts +++ b/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((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", diff --git a/server/src/index.ts b/server/src/index.ts index 8d81e21adaa..e96488ee39c 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -880,6 +880,14 @@ export async function startServer(): Promise { } } + 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 { // 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(); diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 7d6992f8530..7641e58c409 100644 --- a/server/src/services/heartbeat.ts +++ b/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>(); + + 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(); + 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.