diff --git a/server/src/__tests__/heartbeat-comment-wake-batching.test.ts b/server/src/__tests__/heartbeat-comment-wake-batching.test.ts index 117ae4bb..5b9f9413 100644 --- a/server/src/__tests__/heartbeat-comment-wake-batching.test.ts +++ b/server/src/__tests__/heartbeat-comment-wake-batching.test.ts @@ -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, diff --git a/server/src/__tests__/heartbeat-zombie-guard.test.ts b/server/src/__tests__/heartbeat-zombie-guard.test.ts new file mode 100644 index 00000000..78311def --- /dev/null +++ b/server/src/__tests__/heartbeat-zombie-guard.test.ts @@ -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(); + + 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(); + + 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([["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(); + + 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(); + + 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(); + + 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([["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(); + + 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(); + + 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(); + + 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(); + + 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(); + 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([["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); + }); +}); diff --git a/server/src/index.ts b/server/src/index.ts index 6708c535..f18fb700 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -745,56 +745,73 @@ export async function startServer(): Promise { 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()) diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 39f6794a..41edce2e 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -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 | null | undefined, ) { @@ -3088,6 +3119,11 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) }); const workspaceOperationsSvc = workspaceOperationService(db); const activeRunExecutions = new Set(); + 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,