fix(heartbeat): prevent zombie run coalescing and ensure startup reap completes before timer ticks (#1731)

## Thinking Path > - Paperclip orchestrates AI agents for zero-human companies > - Agents run in heartbeats — short execution windows triggered by the heartbeat service > - The heartbeat service coalesces overlapping wakeups: if a run for an agent is already active, a new wakeup merges into it rather than creating a duplicate > - But when the server restarts, in-progress runs are left in `"running"` status in the database — their child processes are gone, but the DB rows persist as orphans > - The startup `reapOrphanedRuns()` was fired as a `void` promise — the timer interval started immediately in parallel, so the first timer tick could coalesce a new wakeup into an orphaned "running" row before the reap had a chance to remove it > - Once coalesced, the orphan's `updatedAt` refreshed, making the reaper skip it as "not old enough" — a zombie run that prevents the agent from ever waking again > - This PR fixes both the coalescing guard (do not coalesce into a zombie) and the startup ordering (await reap before starting the timer), eliminating the death spiral ## What Changed - **`server/src/index.ts`** — `startServer` now `await`s `reapOrphanedRuns()` (with one retry) before calling `setInterval`. Timer ticks cannot start until orphaned runs are cleaned up. - **`server/src/services/heartbeat.ts`** — Added two exported pure functions: - `isZombieRun(run, tracked)` — returns `true` if a run is `"running"` in the DB but has no live entry in the in-memory `runningProcesses` Map - `filterZombieCoalesceTarget(target, tracked)` — returns `null` if the coalesce candidate is a zombie, letting the wakeup fall through to create a new queued run instead - Both coalescing call sites now use `filterZombieCoalesceTarget` before deciding to coalesce - **`server/src/__tests__/heartbeat-zombie-guard.test.ts`** — 8 new behavioral tests covering `isZombieRun` and `filterZombieCoalesceTarget`, including the critical zombie scenario, legitimate live runs, queued runs (must never be filtered), and null pass-through ## Verification ```bash # Run the new tests pnpm test:run ``` Manual reproduction (before fix): 1. Start an agent on a timer heartbeat 2. Kill the server mid-run (child process dies, DB row stays `"running"`) 3. Restart the server 4. Observe: agent never wakes again — subsequent wakeups coalesce into the dead run, refreshing `updatedAt`, keeping it alive forever After fix: startup reap clears the orphan before the timer starts; subsequent wakeups create fresh queued runs. The one pre-existing test failure (`worktree helpers > copies shared git hooks`) is unrelated — it fails on `upstream/master` as well due to a `pnpm install` failure in the test environment. ## Risks - **Startup latency**: `await reapOrphanedRuns()` adds a small delay before the timer starts. In practice this is a fast DB query. The retry adds at most one extra attempt on transient failure. - **Behavior change**: Wakeups that previously coalesced into zombie runs will now create new queued runs instead. This is the correct behavior — the zombie was preventing any forward progress. - **Queued runs unaffected**: `isZombieRun` only flags `"running"` status. Queued runs pass through `filterZombieCoalesceTarget` unchanged (covered by tests). ## Model Used - OpenAI GPT-5 via a Codex-style terminal coding agent with tool use and git/gh access. Exact hosted alias is not exposed in this environment. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] If this change affects the UI, I have included before/after screenshots - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] I will address all Greptile and reviewer comments before requesting merge ## Cross-references and status (maintainer) Refs #3168 Refs #4174 Refs #4697 Refs #6399 Related PRs checked: #4075, #4705, #5232, #6952 - [x] I have searched GitHub for duplicate or related PRs and linked them above --------- Co-authored-by: Devin Foley <devin@paperclip.ing>
2026-06-11 23:54:32 -05:00
parent 130219c0be
commit d782c4cd53
4 changed files with 312 additions and 108 deletions
@@ -2,7 +2,7 @@ import { randomUUID } from "node:crypto";
 import { createServer } from "node:http";
 import { and, asc, eq } from "drizzle-orm";
 import { WebSocketServer } from "ws";
-import { afterAll, beforeAll, describe, expect, it } from "vitest";
+import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest";
 import {
  agents,
  agentWakeupRequests,
@@ -12,6 +12,7 @@ import {
  issueComments,
  issues,
 } from "@paperclipai/db";
+import { runningProcesses } from "../adapters/index.js";
 import { heartbeatService } from "../services/heartbeat.ts";
 import { SUCCESSFUL_RUN_HANDOFF_REQUIRED_NOTICE_BODY } from "../services/recovery/index.ts";
 import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.ts";
@@ -161,6 +162,10 @@ describe("heartbeat comment wake batching", () => {
    await tempDb?.cleanup();
  });

+  afterEach(() => {
+    runningProcesses.clear();
+  });
+
  it("defers approval-approved wakes for a running issue so the assignee resumes after the run", async () => {
    const companyId = randomUUID();
    const agentId = randomUUID();
@@ -201,6 +206,11 @@ describe("heartbeat comment wake batching", () => {
        wakeReason: "issue_assigned",
      },
    });
+    runningProcesses.set(runId, {
+      child: {} as never,
+      graceSec: 0,
+      processGroupId: null,
+    });

    await db.insert(issues).values({
      id: issueId,
@@ -0,0 +1,125 @@
+import { describe, expect, it } from "vitest";
+import {
+  isZombieRun,
+  filterZombieCoalesceTarget,
+} from "../services/heartbeat.ts";
+
+// ---------------------------------------------------------------------------
+// isZombieRun — the core predicate
+// ---------------------------------------------------------------------------
+describe("isZombieRun", () => {
+  it("returns true for a running run not tracked in runningProcesses", () => {
+    const run = { status: "running", id: "run-1" };
+    const tracked = new Map<string, unknown>();
+
+    expect(isZombieRun(run, tracked)).toBe(true);
+  });
+
+  it("returns false for a queued run not tracked in runningProcesses", () => {
+    const run = { status: "queued", id: "run-2" };
+    const tracked = new Map<string, unknown>();
+
+    expect(isZombieRun(run, tracked)).toBe(false);
+  });
+
+  it("returns false for a running run that IS tracked in runningProcesses", () => {
+    const run = { status: "running", id: "run-3" };
+    const tracked = new Map<string, unknown>([["run-3", { pid: 12345 }]]);
+
+    expect(isZombieRun(run, tracked)).toBe(false);
+  });
+
+  it("returns false for a failed run not tracked in runningProcesses", () => {
+    const run = { status: "failed", id: "run-4" };
+    const tracked = new Map<string, unknown>();
+
+    expect(isZombieRun(run, tracked)).toBe(false);
+  });
+
+  it("returns false for a completed run not tracked in runningProcesses", () => {
+    const run = { status: "completed", id: "run-5" };
+    const tracked = new Map<string, unknown>();
+
+    expect(isZombieRun(run, tracked)).toBe(false);
+  });
+});
+
+// ---------------------------------------------------------------------------
+// filterZombieCoalesceTarget — the coalescing guard used in both paths
+//
+// These tests exercise the BEHAVIOR described in spec AC2 and AC3:
+// "Coalescing does not refresh updatedAt on zombie runs"
+// When the target is a zombie, the filter returns null so the wakeup
+// falls through to create a new queued run instead of merging into the dead one.
+// ---------------------------------------------------------------------------
+describe("filterZombieCoalesceTarget", () => {
+  // Bug 1 scenario: a "running" run with no live process is a zombie.
+  // Coalescing into it would refresh updatedAt, making it immortal.
+  it("returns null for a zombie running run (the critical bug fix)", () => {
+    const zombieRun = { status: "running", id: "zombie-1" };
+    const emptyTracked = new Map<string, unknown>();
+
+    expect(filterZombieCoalesceTarget(zombieRun, emptyTracked)).toBeNull();
+  });
+
+  // Legitimate running process — coalescing should proceed normally.
+  it("passes through a legitimate running run that IS tracked", () => {
+    const liveRun = { status: "running", id: "live-1" };
+    const tracked = new Map<string, unknown>([["live-1", { pid: 99 }]]);
+
+    expect(filterZombieCoalesceTarget(liveRun, tracked)).toBe(liveRun);
+  });
+
+  // Queued runs don't have processes yet — they must always pass through.
+  // isZombieRun only flags "running" status, so queued runs are safe.
+  it("passes through a queued run not tracked (queued runs are not zombies)", () => {
+    const queuedRun = { status: "queued", id: "queued-1" };
+    const emptyTracked = new Map<string, unknown>();
+
+    expect(filterZombieCoalesceTarget(queuedRun, emptyTracked)).toBe(queuedRun);
+  });
+
+  // null target means no candidate to coalesce into — pass through.
+  it("passes through null target unchanged", () => {
+    const tracked = new Map<string, unknown>();
+
+    expect(filterZombieCoalesceTarget(null, tracked)).toBeNull();
+  });
+
+  // Terminal states should never appear as coalesce targets, but if they do,
+  // they should pass through (they're not zombies — they're done).
+  it("passes through a failed run (terminal state, not a zombie)", () => {
+    const failedRun = { status: "failed", id: "failed-1" };
+    const emptyTracked = new Map<string, unknown>();
+
+    expect(filterZombieCoalesceTarget(failedRun, emptyTracked)).toBe(failedRun);
+  });
+
+  it("passes through a completed run (terminal state, not a zombie)", () => {
+    const completedRun = { status: "completed", id: "done-1" };
+    const emptyTracked = new Map<string, unknown>();
+
+    expect(filterZombieCoalesceTarget(completedRun, emptyTracked)).toBe(completedRun);
+  });
+
+  // Regression guard: after server restart, runningProcesses is empty.
+  // Multiple zombie runs should all be filtered to null.
+  it("filters multiple zombie runs independently (post-restart scenario)", () => {
+    const emptyTracked = new Map<string, unknown>();
+    const zombie1 = { status: "running", id: "z1" };
+    const zombie2 = { status: "running", id: "z2" };
+
+    expect(filterZombieCoalesceTarget(zombie1, emptyTracked)).toBeNull();
+    expect(filterZombieCoalesceTarget(zombie2, emptyTracked)).toBeNull();
+  });
+
+  // Mixed scenario: one zombie, one live. Only the zombie is filtered.
+  it("correctly distinguishes zombie from live when multiple runs exist", () => {
+    const tracked = new Map<string, unknown>([["live-1", { pid: 42 }]]);
+    const zombie = { status: "running", id: "zombie-1" };
+    const live = { status: "running", id: "live-1" };
+
+    expect(filterZombieCoalesceTarget(zombie, tracked)).toBeNull();
+    expect(filterZombieCoalesceTarget(live, tracked)).toBe(live);
+  });
+});
@@ -745,56 +745,73 @@ export async function startServer(): Promise<StartedServer> {
  if (config.heartbeatSchedulerEnabled) {
    const heartbeat = heartbeatService(db as any, { pluginWorkerManager });
    const routines = routineService(db as any, { pluginWorkerManager });
-  
-    // Reap orphaned running runs at startup while in-memory execution state is empty,
-    // then resume any persisted queued runs that were waiting on the previous process.
-    void heartbeat
-      .reapOrphanedRuns()
-      .then(() => heartbeat.promoteDueScheduledRetries())
-      .then(async (promotion) => {
-        await heartbeat.resumeQueuedRuns();
-        const reconciled = await heartbeat.reconcileStrandedAssignedIssues();
-        if (
-          promotion.promoted > 0 ||
-          reconciled.assignmentDispatched > 0 ||
-          reconciled.dispatchRequeued > 0 ||
-          reconciled.continuationRequeued > 0 ||
-          reconciled.successfulRunHandoffEscalated > 0 ||
-          reconciled.escalated > 0
-        ) {
-          logger.warn(
-            { promotedScheduledRetries: promotion.promoted, promotedScheduledRetryRunIds: promotion.runIds, ...reconciled },
-            "startup heartbeat recovery changed assigned issue state",
+
+    // Reap orphaned runs before timer ticks start so wakeups cannot coalesce
+    // into a dead "running" row during startup recovery.
+    await (async () => {
+      for (let attempt = 1; attempt <= 2; attempt++) {
+        try {
+          const result = await heartbeat.reapOrphanedRuns();
+          logger.info(
+            { reaped: result.reaped, runIds: result.runIds },
+            "startup reap of orphaned heartbeat runs complete",
          );
+          break;
+        } catch (err) {
+          if (attempt < 2) {
+            logger.warn({ err, attempt }, "startup reap failed, retrying");
+          } else {
+            logger.error(
+              { err },
+              "startup reap of orphaned heartbeat runs failed after retry — periodic reaper will serve as degraded backstop",
+            );
+          }
        }
-      })
-      .then(async () => {
-        const reconciled = await heartbeat.reconcileIssueGraphLiveness();
-        if (reconciled.escalationsCreated > 0) {
-          logger.warn({ ...reconciled }, "startup issue-graph liveness reconciliation created escalations");
-        }
-      })
-      .then(async () => {
-        const scanned = await heartbeat.scanSilentActiveRuns();
-        if (scanned.created > 0 || scanned.escalated > 0) {
-          logger.warn({ ...scanned }, "startup active-run output watchdog created review work");
-        }
-      })
-      .then(async () => {
-        const swept = await heartbeat.sweepStaleIssueLocks();
-        if (swept.cleared > 0) {
-          logger.warn({ ...swept }, "startup stale-lock sweeper cleared issue locks");
-        }
-      })
-      .then(async () => {
-        const reviewed = await heartbeat.reconcileProductivityReviews();
-        if (reviewed.created > 0 || reviewed.updated > 0 || reviewed.failed > 0) {
-          logger.warn({ ...reviewed }, "startup productivity reconciliation created or updated review work");
-        }
-      })
-      .catch((err) => {
-        logger.error({ err }, "startup heartbeat recovery failed");
-      });
+      }
+
+      const promotion = await heartbeat.promoteDueScheduledRetries();
+      await heartbeat.resumeQueuedRuns();
+      const reconciled = await heartbeat.reconcileStrandedAssignedIssues();
+      if (
+        promotion.promoted > 0 ||
+        reconciled.assignmentDispatched > 0 ||
+        reconciled.dispatchRequeued > 0 ||
+        reconciled.continuationRequeued > 0 ||
+        reconciled.successfulRunHandoffEscalated > 0 ||
+        reconciled.escalated > 0
+      ) {
+        logger.warn(
+          { promotedScheduledRetries: promotion.promoted, promotedScheduledRetryRunIds: promotion.runIds, ...reconciled },
+          "startup heartbeat recovery changed assigned issue state",
+        );
+      }
+
+      const issueGraphReconciled = await heartbeat.reconcileIssueGraphLiveness();
+      if (issueGraphReconciled.escalationsCreated > 0) {
+        logger.warn(
+          { ...issueGraphReconciled },
+          "startup issue-graph liveness reconciliation created escalations",
+        );
+      }
+
+      const scanned = await heartbeat.scanSilentActiveRuns();
+      if (scanned.created > 0 || scanned.escalated > 0) {
+        logger.warn({ ...scanned }, "startup active-run output watchdog created review work");
+      }
+
+      const swept = await heartbeat.sweepStaleIssueLocks();
+      if (swept.cleared > 0) {
+        logger.warn({ ...swept }, "startup stale-lock sweeper cleared issue locks");
+      }
+
+      const reviewed = await heartbeat.reconcileProductivityReviews();
+      if (reviewed.created > 0 || reviewed.updated > 0 || reviewed.failed > 0) {
+        logger.warn({ ...reviewed }, "startup productivity reconciliation created or updated review work");
+      }
+    })().catch((err) => {
+      logger.error({ err }, "startup heartbeat recovery failed");
+    });
+
    setInterval(() => {
      void heartbeat
        .tickTimers(new Date())
@@ -2107,6 +2107,37 @@ export function formatRuntimeWorkspaceWarningLog(warning: string) {
  };
 }

+/**
+ * A run is a "zombie" if it's marked as running in the DB but has no live
+ * execution tracked in memory. This happens when the server restarts and the
+ * execution is lost, or when the DB row outlives the in-memory run state.
+ *
+ * Queued runs are never zombies — they don't have processes yet.
+ */
+export function isZombieRun(
+  run: { status: string; id: string },
+  tracked: { has(id: string): boolean },
+): boolean {
+  return run.status === "running" && !tracked.has(run.id);
+}
+
+/**
+ * Filter a coalesce target — if it's a zombie run, return null so the
+ * wakeup falls through to create a new queued run instead of coalescing
+ * into the dead process (which would refresh updatedAt and make it immortal).
+ *
+ * Queued runs pass through unchanged (they have no process yet).
+ * Null targets pass through unchanged.
+ */
+export function filterZombieCoalesceTarget<
+  T extends { status: string; id: string },
+>(
+  target: T | null,
+  tracked: { has(id: string): boolean },
+): T | null {
+  return target && isZombieRun(target, tracked) ? null : target;
+}
+
 export function describeSessionResetReason(
  contextSnapshot: Record<string, unknown> | null | undefined,
 ) {
@@ -3088,6 +3119,11 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
  });
  const workspaceOperationsSvc = workspaceOperationService(db);
  const activeRunExecutions = new Set<string>();
+  const liveRunExecutions = {
+    has(id: string) {
+      return runningProcesses.has(id) || activeRunExecutions.has(id);
+    },
+  };
  const budgetHooks = {
    cancelWorkForScope: cancelBudgetScopeWork,
  };
@@ -10504,10 +10540,19 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
            shouldQueueFollowupForRunningIssueWake({ contextSnapshot: enrichedContextSnapshot, wakeCommentId }) &&
            activeExecutionRun.status === "running" &&
            isSameExecutionAgent;
+          const availableActiveExecutionRun = filterZombieCoalesceTarget(
+            activeExecutionRun,
+            liveRunExecutions,
+          );

-          if (isSameExecutionAgent && !shouldDeferFollowupWake && !shouldQueueFollowupForRunningWake) {
+          if (
+            isSameExecutionAgent
+            && !shouldDeferFollowupWake
+            && !shouldQueueFollowupForRunningWake
+            && availableActiveExecutionRun
+          ) {
            const mergedContextSnapshot = mergeCoalescedContextSnapshot(
-              activeExecutionRun.contextSnapshot,
+              availableActiveExecutionRun.contextSnapshot,
              enrichedContextSnapshot,
            );
            const mergedRun = await tx
@@ -10516,9 +10561,9 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
                contextSnapshot: mergedContextSnapshot,
                updatedAt: new Date(),
              })
-              .where(eq(heartbeatRuns.id, activeExecutionRun.id))
+              .where(eq(heartbeatRuns.id, availableActiveExecutionRun.id))
              .returning()
-              .then((rows) => rows[0] ?? activeExecutionRun);
+              .then((rows) => rows[0] ?? availableActiveExecutionRun);

            await tx.insert(agentWakeupRequests).values({
              companyId: agent.companyId,
@@ -10539,67 +10584,69 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
            return { kind: "coalesced" as const, run: mergedRun };
          }

-          const deferredPayload = {
-            ...(payload ?? {}),
-            issueId,
-            [DEFERRED_WAKE_CONTEXT_KEY]: enrichedContextSnapshot,
-          };
-
-          const existingDeferred = await tx
-            .select()
-            .from(agentWakeupRequests)
-            .where(
-              and(
-                eq(agentWakeupRequests.companyId, agent.companyId),
-                eq(agentWakeupRequests.agentId, agentId),
-                eq(agentWakeupRequests.status, "deferred_issue_execution"),
-                sql`${agentWakeupRequests.payload} ->> 'issueId' = ${issue.id}`,
-              ),
-            )
-            .orderBy(asc(agentWakeupRequests.requestedAt))
-            .limit(1)
-            .then((rows) => rows[0] ?? null);
-
-          if (existingDeferred) {
-            const existingDeferredPayload = parseObject(existingDeferred.payload);
-            const existingDeferredContext = parseObject(existingDeferredPayload[DEFERRED_WAKE_CONTEXT_KEY]);
-            const mergedDeferredContext = mergeCoalescedContextSnapshot(
-              existingDeferredContext,
-              enrichedContextSnapshot,
-            );
-            const mergedDeferredPayload = {
-              ...existingDeferredPayload,
+          if (availableActiveExecutionRun) {
+            const deferredPayload = {
              ...(payload ?? {}),
              issueId,
-              [DEFERRED_WAKE_CONTEXT_KEY]: mergedDeferredContext,
+              [DEFERRED_WAKE_CONTEXT_KEY]: enrichedContextSnapshot,
            };

-            await tx
-              .update(agentWakeupRequests)
-              .set({
-                payload: mergedDeferredPayload,
-                coalescedCount: (existingDeferred.coalescedCount ?? 0) + 1,
-                updatedAt: new Date(),
-              })
-              .where(eq(agentWakeupRequests.id, existingDeferred.id));
+            const existingDeferred = await tx
+              .select()
+              .from(agentWakeupRequests)
+              .where(
+                and(
+                  eq(agentWakeupRequests.companyId, agent.companyId),
+                  eq(agentWakeupRequests.agentId, agentId),
+                  eq(agentWakeupRequests.status, "deferred_issue_execution"),
+                  sql`${agentWakeupRequests.payload} ->> 'issueId' = ${issue.id}`,
+                ),
+              )
+              .orderBy(asc(agentWakeupRequests.requestedAt))
+              .limit(1)
+              .then((rows) => rows[0] ?? null);
+
+            if (existingDeferred) {
+              const existingDeferredPayload = parseObject(existingDeferred.payload);
+              const existingDeferredContext = parseObject(existingDeferredPayload[DEFERRED_WAKE_CONTEXT_KEY]);
+              const mergedDeferredContext = mergeCoalescedContextSnapshot(
+                existingDeferredContext,
+                enrichedContextSnapshot,
+              );
+              const mergedDeferredPayload = {
+                ...existingDeferredPayload,
+                ...(payload ?? {}),
+                issueId,
+                [DEFERRED_WAKE_CONTEXT_KEY]: mergedDeferredContext,
+              };
+
+              await tx
+                .update(agentWakeupRequests)
+                .set({
+                  payload: mergedDeferredPayload,
+                  coalescedCount: (existingDeferred.coalescedCount ?? 0) + 1,
+                  updatedAt: new Date(),
+                })
+                .where(eq(agentWakeupRequests.id, existingDeferred.id));
+
+              return { kind: "deferred" as const };
+            }
+
+            await tx.insert(agentWakeupRequests).values({
+              companyId: agent.companyId,
+              agentId,
+              source,
+              triggerDetail,
+              reason: "issue_execution_deferred",
+              payload: deferredPayload,
+              status: "deferred_issue_execution",
+              requestedByActorType: opts.requestedByActorType ?? null,
+              requestedByActorId: opts.requestedByActorId ?? null,
+              idempotencyKey: opts.idempotencyKey ?? null,
+            });

            return { kind: "deferred" as const };
          }
-
-          await tx.insert(agentWakeupRequests).values({
-            companyId: agent.companyId,
-            agentId,
-            source,
-            triggerDetail,
-            reason: "issue_execution_deferred",
-            payload: deferredPayload,
-            status: "deferred_issue_execution",
-            requestedByActorType: opts.requestedByActorType ?? null,
-            requestedByActorId: opts.requestedByActorId ?? null,
-            idempotencyKey: opts.idempotencyKey ?? null,
-          });
-
-          return { kind: "deferred" as const };
        }

        const wakeupRequest = await tx
@@ -10693,11 +10740,16 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
      !sameScopeQueuedRun &&
      shouldQueueFollowupForRunningIssueWake({ contextSnapshot: enrichedContextSnapshot, wakeCommentId });

-    const coalescedTargetRun =
+    const rawCoalescedTarget =
      sameScopeQueuedRun ??
      sameScopeScheduledRetryRun ??
      (shouldQueueFollowupForRunningWake ? null : sameScopeRunningRun ?? null);

+    const coalescedTargetRun = filterZombieCoalesceTarget(
+      rawCoalescedTarget,
+      liveRunExecutions,
+    );
+
    if (coalescedTargetRun) {
      const mergedContextSnapshot = mergeCoalescedContextSnapshot(
        coalescedTargetRun.contextSnapshot,