diff --git a/server/src/__tests__/heartbeat-process-recovery.test.ts b/server/src/__tests__/heartbeat-process-recovery.test.ts index 6e1656483ca..4479bc1adb8 100644 --- a/server/src/__tests__/heartbeat-process-recovery.test.ts +++ b/server/src/__tests__/heartbeat-process-recovery.test.ts @@ -347,6 +347,16 @@ async function waitForValue( return latest ?? null; } +function createDeferred() { + let resolve!: (value: T | PromiseLike) => void; + let reject!: (reason?: unknown) => void; + const promise = new Promise((res, rej) => { + resolve = res; + reject = rej; + }); + return { promise, resolve, reject }; +} + async function waitForHeartbeatIdle( db: ReturnType, timeoutMs = 3_000, @@ -1479,6 +1489,57 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { expect(wakeup?.status).toBe("claimed"); }); + it("deduplicates concurrent process_lost reaps of the same local run", async () => { + const { runId, wakeupRequestId } = await seedRunFixture({ + processPid: 999_999_999, + includeIssue: false, + }); + const jobStatusGate = createDeferred(); + let jobStatusCalls = 0; + mockListAgentJobRunStatuses + .mockImplementationOnce(async () => { + jobStatusCalls += 1; + return jobStatusGate.promise; + }) + .mockImplementationOnce(async () => { + jobStatusCalls += 1; + return jobStatusGate.promise; + }); + + const firstReap = heartbeat.reapOrphanedRuns({ suppressDispatchAfterReap: true }); + const secondReap = heartbeat.reapOrphanedRuns({ suppressDispatchAfterReap: true }); + + expect(await waitForValue(async () => jobStatusCalls >= 2 ? jobStatusCalls : null)).toBe(2); + jobStatusGate.resolve(null); + + const results = await Promise.all([firstReap, secondReap]); + expect(results.reduce((sum, result) => sum + result.reaped, 0)).toBe(1); + expect(results.flatMap((result) => result.runIds)).toEqual([runId]); + + const run = await heartbeat.getRun(runId); + expect(run?.status).toBe("failed"); + expect(run?.errorCode).toBe("process_lost"); + + const wakeup = await db + .select() + .from(agentWakeupRequests) + .where(eq(agentWakeupRequests.id, wakeupRequestId)) + .then((rows) => rows[0] ?? null); + expect(wakeup?.status).toBe("failed"); + + const runEvents = await db + .select() + .from(heartbeatRunEvents) + .where(eq(heartbeatRunEvents.runId, runId)); + expect( + runEvents.filter((event) => + event.eventType === "lifecycle" && + event.level === "error" && + event.message.includes("Process lost"), + ), + ).toHaveLength(1); + }); + it("skips generic timer wakes without invoking an adapter when no assigned work is actionable", async () => { const { companyId, agentId } = await seedIdleTimerAgentFixture(); const heartbeat = createHeartbeat(); diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 5a8a02ef7a0..e16628c5e41 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -16871,7 +16871,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) // once the mint is committed (so a failed/aborted setRunStatus never // over-counts). Split by adapter + error-string bucket + the durable // classification, all bounded. - let finalizedRun = await setRunStatus(run.id, "failed", { + const finalizedRunWrite = await setRunStatusIfRunning(run.id, "failed", { error: shouldRetry ? `${baseMessage}; retrying once` : baseMessage, errorCode: "process_lost", finishedAt: now, @@ -16894,12 +16894,21 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) : result; })(), }); + if (!finalizedRunWrite.updated || !finalizedRunWrite.run) { + // Another reap invocation won the terminal transition for this run. + // Do not duplicate process_lost metrics, retries, wakeup finalization, + // run events, or issue-promotion side effects. + if (finalizedRunWrite.run?.status !== "running") { + runningProcesses.delete(run.id); + activeRunExecutions.delete(run.id); + } + continue; + } + let finalizedRun = finalizedRunWrite.run; await setWakeupStatus(run.wakeupRequestId, "failed", { finishedAt: now, error: shouldRetry ? `${baseMessage}; retrying once` : baseMessage, }); - if (!finalizedRun) finalizedRun = await getRun(run.id); - if (!finalizedRun) continue; // BLO-16184: the process_lost mint is now committed for this run -- count it // (bounded adapter + error-string bucket + durable classification). recordProcessLost({