diff --git a/server/src/__tests__/heartbeat-process-recovery.test.ts b/server/src/__tests__/heartbeat-process-recovery.test.ts index cb02a301008..1531e40cb89 100644 --- a/server/src/__tests__/heartbeat-process-recovery.test.ts +++ b/server/src/__tests__/heartbeat-process-recovery.test.ts @@ -819,13 +819,12 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { } it("does not reap active adapter executions started by another heartbeat service instance", async () => { - let releaseAdapter: (() => void) | null = null; + let releaseAdapter!: () => void; + const adapterRelease = new Promise(resolve => { releaseAdapter = resolve; }); const adapterStarted = new Promise((resolve) => { mockAdapterExecute.mockImplementationOnce(async () => { resolve(); - await new Promise((release) => { - releaseAdapter = release; - }); + await adapterRelease; return { exitCode: 0, signal: null, @@ -849,52 +848,55 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { const executorHeartbeat = heartbeatService(db); const reaperHeartbeat = heartbeatService(db); - await executorHeartbeat.resumeQueuedRuns(); - await Promise.race([ - adapterStarted, - new Promise((_, reject) => { - setTimeout( - () => - reject( - new Error("Timed out waiting for adapter execution to start"), - ), - 3_000, - ); - }), - ]); + try { + await executorHeartbeat.resumeQueuedRuns(); + await Promise.race([ + adapterStarted, + new Promise((_, reject) => { + setTimeout( + () => + reject( + new Error("Timed out waiting for adapter execution to start"), + ), + 3_000, + ); + }), + ]); - await db - .update(heartbeatRuns) - .set({ - updatedAt: new Date("2026-03-19T00:00:00.000Z"), - }) - .where(eq(heartbeatRuns.id, runId)); + await db + .update(heartbeatRuns) + .set({ + updatedAt: new Date("2026-03-19T00:00:00.000Z"), + }) + .where(eq(heartbeatRuns.id, runId)); - const result = await reaperHeartbeat.reapOrphanedRuns({ - staleThresholdMs: 1, - }); - expect(result).toEqual({ reaped: 0, runIds: [] }); + const result = await reaperHeartbeat.reapOrphanedRuns({ + staleThresholdMs: 1, + }); + expect(result).toEqual({ reaped: 0, runIds: [] }); - const activeRun = await reaperHeartbeat.getRun(runId); - expect(activeRun?.status).toBe("running"); - expect(activeRun?.errorCode).toBeNull(); + const activeRun = await reaperHeartbeat.getRun(runId); + expect(activeRun?.status).toBe("running"); + expect(activeRun?.errorCode).toBeNull(); - const wakeup = await db - .select() - .from(agentWakeupRequests) - .where(eq(agentWakeupRequests.id, wakeupRequestId)) - .then((rows) => rows[0] ?? null); - expect(wakeup?.status).toBe("claimed"); + const wakeup = await db + .select() + .from(agentWakeupRequests) + .where(eq(agentWakeupRequests.id, wakeupRequestId)) + .then((rows) => rows[0] ?? null); + expect(wakeup?.status).toBe("claimed"); - if (!releaseAdapter) - throw new Error("Adapter release handle was not captured"); - releaseAdapter(); - const settledRun = await waitForRunToSettle( - executorHeartbeat, - runId, - 5_000, - ); - expect(settledRun?.status).toBe("succeeded"); + releaseAdapter(); + const settledRun = await waitForRunToSettle( + executorHeartbeat, + runId, + 5_000, + ); + expect(settledRun?.status).toBe("succeeded"); + } finally { + releaseAdapter(); + await executorHeartbeat.drainActiveRunExecutions(); + } }); async function seedStrandedIssueFixture(input: { @@ -3407,7 +3409,8 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { }); it("persists codex_local spawn identity before hot restart and never loses the live run for missing metadata", async () => { - let releaseAdapter: (() => void) | null = null; + let releaseAdapter!: () => void; + const adapterRelease = new Promise(resolve => { releaseAdapter = resolve; }); let spawnedPid: number | null = null; const adapterStarted = new Promise((resolve) => { mockAdapterExecute.mockImplementationOnce(async (rawInput?: unknown) => { @@ -3429,9 +3432,7 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { startedAt: new Date("2026-07-30T07:00:00.000Z").toISOString(), }); resolve(); - await new Promise((release) => { - releaseAdapter = release; - }); + await adapterRelease; return { exitCode: 0, signal: null, @@ -3456,87 +3457,90 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { includeIssue: false, }); const heartbeat = heartbeatService(db); - await heartbeat.resumeQueuedRuns(); - await Promise.race([ - adapterStarted, - new Promise((_, reject) => { - setTimeout( - () => - reject( - new Error("Timed out waiting for codex_local spawn identity"), - ), - 3_000, - ); - }), - ]); - - const running = await waitForValue(async () => - db - .select() - .from(heartbeatRuns) - .where(eq(heartbeatRuns.id, runId)) - .then((rows) => { - const row = rows[0] ?? null; - return row?.status === "running" && row.processPid ? row : null; + try { + await heartbeat.resumeQueuedRuns(); + await Promise.race([ + adapterStarted, + new Promise((_, reject) => { + setTimeout( + () => + reject( + new Error("Timed out waiting for codex_local spawn identity"), + ), + 3_000, + ); }), - ); - const observedProcessStartedAt = await readProcessStartedAt(spawnedPid!); - expect(observedProcessStartedAt).not.toBeNull(); - expect(running).toMatchObject({ - id: runId, - status: "running", - processPid: spawnedPid, - processGroupId: null, - processStartedAt: new Date(observedProcessStartedAt!), - }); + ]); - await withTempPaperclipHome(async (home) => { - await writeHotRestartIntent({ - previousServerPid: process.pid, - previousServerVersion: "old-version", - requestedAt: new Date("2026-07-30T07:01:00.000Z"), - }); - await heartbeat.prepareHotRestartShutdown( - "SIGTERM", - new Date("2026-07-30T07:02:00.000Z"), - ); - - const adoption = await heartbeat.reconcileHotRestartAdoption( - new Date("2026-07-30T07:03:00.000Z"), - ); - expect(adoption).toMatchObject({ - mode: "reported", - adoptedRunIds: [runId], - finalizedWhileDownRunIds: [], - lostRunIds: [], - }); - const report = JSON.parse( - await fs.readFile(resolveHotRestartReportPath(home), "utf8"), - ) as { runs?: Array> }; - expect(report.runs).toEqual( - expect.arrayContaining([ - expect.objectContaining({ - runId, - classification: "adopted", - reason: "process_pid_alive", + const running = await waitForValue(async () => + db + .select() + .from(heartbeatRuns) + .where(eq(heartbeatRuns.id, runId)) + .then((rows) => { + const row = rows[0] ?? null; + return row?.status === "running" && row.processPid ? row : null; }), - ]), ); - expect(report.runs).not.toEqual( - expect.arrayContaining([ - expect.objectContaining({ - runId, - reason: "missing_process_metadata", - }), - ]), - ); - }); + const observedProcessStartedAt = await readProcessStartedAt(spawnedPid!); + expect(observedProcessStartedAt).not.toBeNull(); + expect(running).toMatchObject({ + id: runId, + status: "running", + processPid: spawnedPid, + processGroupId: null, + processStartedAt: new Date(observedProcessStartedAt!), + }); - if (!releaseAdapter) - throw new Error("Adapter release handle was not captured"); - releaseAdapter(); - const settled = await waitForRunToSettle(heartbeat, runId, 5_000); - expect(settled?.status).toBe("succeeded"); + await withTempPaperclipHome(async (home) => { + await writeHotRestartIntent({ + previousServerPid: process.pid, + previousServerVersion: "old-version", + requestedAt: new Date("2026-07-30T07:01:00.000Z"), + }); + await heartbeat.prepareHotRestartShutdown( + "SIGTERM", + new Date("2026-07-30T07:02:00.000Z"), + ); + + const adoption = await heartbeat.reconcileHotRestartAdoption( + new Date("2026-07-30T07:03:00.000Z"), + ); + expect(adoption).toMatchObject({ + mode: "reported", + adoptedRunIds: [runId], + finalizedWhileDownRunIds: [], + lostRunIds: [], + }); + const report = JSON.parse( + await fs.readFile(resolveHotRestartReportPath(home), "utf8"), + ) as { runs?: Array> }; + expect(report.runs).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + runId, + classification: "adopted", + reason: "process_pid_alive", + }), + ]), + ); + expect(report.runs).not.toEqual( + expect.arrayContaining([ + expect.objectContaining({ + runId, + reason: "missing_process_metadata", + }), + ]), + ); + }); + + releaseAdapter(); + const settled = await waitForRunToSettle(heartbeat, runId, 5_000); + expect(settled?.status).toBe("succeeded"); + } finally { + releaseAdapter(); + await heartbeat.drainActiveRunExecutions(); + } }); it("reports adopted hot-restart runs before startup reap can mark them process_lost", async () => {