import { buildExecutionDiagnostics, buildSchedulerHeartbeat, schedulerHeartbeatStaleMs, staleRecoveryCandidate, taskHasTraceGapButFreshHeartbeat } from "../../src/components/microservices/code-queue/src/execution-diagnostics"; import type { ActiveRun } from "../../src/components/microservices/code-queue/src/code-agent/common"; import type { CodeQueueExecutionDiagnostics, QueueTask, SchedulerActiveRunHeartbeat, TaskStatus } from "../../src/components/microservices/code-queue/src/types"; export const CODE_QUEUE_LIVENESS_CHECK_NAMES = [ "code-queue:active-run-heartbeat-visible", "code-queue:trace-gap-not-stale", "code-queue:heartbeat-threshold-normalized", "code-queue:stale-active-owner-expired", "code-queue:stale-active-recovery-transition", "code-queue:control-plane-split-brain-diagnostics", "code-queue:oa-publisher-degraded-visible", ] as const; type CodeQueueLivenessCheckName = typeof CODE_QUEUE_LIVENESS_CHECK_NAMES[number]; interface FixtureCheck { name: CodeQueueLivenessCheckName; ok: boolean; detail: Record; } const now = "2026-05-19T00:10:00.000Z"; const freshAt = "2026-05-19T00:09:50.000Z"; const oldTraceAt = "2026-05-18T23:40:00.000Z"; const expiredAt = "2026-05-18T23:50:00.000Z"; function assertCondition(condition: unknown, message: string, detail: Record = {}): void { if (!condition) throw new Error(`${message}: ${JSON.stringify(detail)}`); } function fixtureTask(id: string, status: TaskStatus, heartbeat: SchedulerActiveRunHeartbeat | null = null): QueueTask { return { id, queueId: "default", queueEnteredAt: "2026-05-19T00:00:00.000Z", prompt: `${id} prompt`, basePrompt: `${id} prompt`, referenceTaskIds: [], referenceInjection: null, providerId: "D601", cwd: "/workspace", model: "gpt-5.5", reasoningEffort: null, executionMode: "default", maxAttempts: 99, status, createdAt: "2026-05-19T00:00:00.000Z", updatedAt: "2026-05-19T00:00:00.000Z", startedAt: status === "running" || status === "judging" ? "2026-05-19T00:00:00.000Z" : null, finishedAt: null, readAt: null, currentAttempt: status === "queued" ? 0 : 1, currentMode: status === "queued" ? null : "initial", codexThreadId: heartbeat?.codexThreadId ?? null, activeTurnId: heartbeat?.activeTurnId ?? null, schedulerHeartbeat: heartbeat, finalResponse: "", outputMaxSeq: heartbeat?.outputMaxSeq ?? 0, lastError: null, lastJudge: null, judgeFailCount: 0, promptHistory: [], output: [], events: [], attempts: [], cancelRequested: false, nextPrompt: null, nextMode: null, }; } function heartbeat(taskId: string, at: string, overrides: Partial = {}): SchedulerActiveRunHeartbeat { return { taskId, queueId: "default", attempt: 1, activeTurnId: "turn_fixture", codexThreadId: "thread_fixture", owner: "D601", schedulerInstance: "code-queue-scheduler-fixture", executionPlane: "scheduler-execution-plane", agentPort: "codex", status: "running", lastLocalHeartbeatAt: at, lastObservedAgentEventAt: at, lastPersistedTraceAt: at, outputMaxSeq: 10, source: "scheduler", ...overrides, }; } function activeRun(taskId: string, queueId = "default"): ActiveRun { return { taskId, queueId, app: { stop: () => undefined }, port: "codex", threadId: "thread_fixture", turnId: "turn_fixture", startedAt: "2026-05-19T00:00:00.000Z", lastLocalHeartbeatAt: freshAt, lastObservedAgentEventAt: freshAt, lastPersistedTraceAt: freshAt, }; } function schedulerDiagnostics(tasks: QueueTask[], activeRuns: ActiveRun[] = [], oaPublisher: unknown = null, heartbeatStaleMs?: number): CodeQueueExecutionDiagnostics { return buildExecutionDiagnostics({ now, heartbeatStaleMs, controlPlane: "D601-code-queue-scheduler", executionStateSource: "scheduler-execution-plane", tasks, activeRuns, activeRunSlotCount: activeRuns.length, activeQueueIds: activeRuns.map((run) => run.queueId), processingQueueIds: [], orphanedActiveTaskIds: tasks.filter((task) => (task.status === "running" || task.status === "judging") && !activeRuns.some((run) => run.taskId === task.id)).map((task) => task.id), oaPublisher: oaPublisher as never, }); } function checkHeartbeatThresholdNormalized(): FixtureCheck { const task = fixtureTask("codex_fixture_threshold_fresh_1", "running", heartbeat("codex_fixture_threshold_fresh_1", freshAt)); const run = activeRun(task.id); const zeroDecision = staleRecoveryCandidate({ task, localActive: false, now, heartbeatStaleMs: 0 }); const tooLowDecision = staleRecoveryCandidate({ task, localActive: false, now, heartbeatStaleMs: 1 }); const zeroDiagnostics = schedulerDiagnostics([task], [run], null, 0); const tooLowDiagnostics = schedulerDiagnostics([task], [run], null, 1); assertCondition(zeroDecision.allowed === false && zeroDecision.reason === "owner-heartbeat-fresh", "zero heartbeat threshold must normalize before stale recovery", { zeroDecision }); assertCondition(tooLowDecision.allowed === false && tooLowDecision.reason === "owner-heartbeat-fresh", "too-low heartbeat threshold must normalize before stale recovery", { tooLowDecision }); assertCondition(zeroDiagnostics.schedulerHeartbeatStaleMs === schedulerHeartbeatStaleMs, "zero diagnostics threshold should normalize to default stale window", zeroDiagnostics as unknown as Record); assertCondition(tooLowDiagnostics.schedulerHeartbeatStaleMs === schedulerHeartbeatStaleMs, "too-low diagnostics threshold should normalize to default stale window", tooLowDiagnostics as unknown as Record); assertCondition(zeroDiagnostics.state === "healthy" && zeroDiagnostics.effectiveLiveness === "healthy", "fresh heartbeat with zero requested threshold must stay healthy", zeroDiagnostics as unknown as Record); assertCondition(tooLowDiagnostics.state === "healthy" && tooLowDiagnostics.effectiveLiveness === "healthy", "fresh heartbeat with too-low requested threshold must stay healthy", tooLowDiagnostics as unknown as Record); assertCondition(zeroDiagnostics.heartbeatFreshTaskIds.includes(task.id), "fresh heartbeat must remain in fresh task ids", zeroDiagnostics as unknown as Record); assertCondition(zeroDiagnostics.heartbeatExpiredTaskIds.length === 0 && zeroDiagnostics.staleRecoveryCandidateTaskIds.length === 0 && zeroDiagnostics.heartbeatRiskTaskIds.length === 0, "fresh heartbeat must not be reported stale or risky", zeroDiagnostics as unknown as Record); return { name: "code-queue:heartbeat-threshold-normalized", ok: true, detail: { zeroDecision, tooLowDecision, normalizedStaleMs: zeroDiagnostics.schedulerHeartbeatStaleMs, state: zeroDiagnostics.state, effectiveLiveness: zeroDiagnostics.effectiveLiveness, heartbeatFreshTaskIds: zeroDiagnostics.heartbeatFreshTaskIds, heartbeatExpiredTaskIds: zeroDiagnostics.heartbeatExpiredTaskIds, staleRecoveryCandidateTaskIds: zeroDiagnostics.staleRecoveryCandidateTaskIds, }, }; } function checkActiveRunHeartbeatVisible(): FixtureCheck { const task = fixtureTask("codex_fixture_active_1", "running"); const run = activeRun(task.id); task.schedulerHeartbeat = buildSchedulerHeartbeat(task, run, { now: freshAt, owner: "D601", schedulerInstance: "code-queue-scheduler-fixture", agentPort: "codex", lastObservedAgentEventAt: freshAt, lastPersistedTraceAt: freshAt, }); const diagnostics = schedulerDiagnostics([task], [run]); assertCondition(diagnostics.activeHeartbeatTaskIds.includes(task.id), "active heartbeat task id must be visible", diagnostics as unknown as Record); assertCondition(diagnostics.schedulerActiveTaskIds.includes(task.id), "scheduler active task id must be visible", diagnostics as unknown as Record); assertCondition(diagnostics.lastSchedulerHeartbeatAt === freshAt, "last scheduler heartbeat must be surfaced", diagnostics as unknown as Record); return { name: "code-queue:active-run-heartbeat-visible", ok: true, detail: { state: diagnostics.state, schedulerActiveTaskIds: diagnostics.schedulerActiveTaskIds, activeHeartbeatTaskIds: diagnostics.activeHeartbeatTaskIds, lastSchedulerHeartbeatAt: diagnostics.lastSchedulerHeartbeatAt, heartbeat: task.schedulerHeartbeat, }, }; } function checkTraceGapNotStale(): FixtureCheck { const task = fixtureTask("codex_fixture_trace_gap_1", "running", heartbeat("codex_fixture_trace_gap_1", freshAt, { lastPersistedTraceAt: oldTraceAt, outputMaxSeq: 89112, })); const decision = staleRecoveryCandidate({ task, localActive: false, now }); const hasTraceGap = taskHasTraceGapButFreshHeartbeat(task, now); const diagnostics = schedulerDiagnostics([task], []); assertCondition(hasTraceGap, "trace gap with fresh owner heartbeat should be classified", { decision, diagnostics }); assertCondition(decision.allowed === false && decision.reason === "owner-heartbeat-fresh", "fresh heartbeat must block stale retry", { decision }); assertCondition(diagnostics.traceGapNotStaleTaskIds.includes(task.id), "diagnostics must expose trace gap as not stale", diagnostics as unknown as Record); assertCondition(!diagnostics.staleRecoveryCandidateTaskIds.includes(task.id), "trace gap must not enter stale recovery candidates", diagnostics as unknown as Record); return { name: "code-queue:trace-gap-not-stale", ok: true, detail: { decision, traceGapNotStaleTaskIds: diagnostics.traceGapNotStaleTaskIds, staleRecoveryCandidateTaskIds: diagnostics.staleRecoveryCandidateTaskIds, }, }; } function checkStaleActiveOwnerExpired(): FixtureCheck { const task = fixtureTask("codex_fixture_stale_1", "running", heartbeat("codex_fixture_stale_1", expiredAt, { lastObservedAgentEventAt: expiredAt, lastPersistedTraceAt: expiredAt, })); const decision = staleRecoveryCandidate({ task, localActive: false, now }); const diagnostics = schedulerDiagnostics([task], []); assertCondition(decision.allowed === true && decision.reason === "owner-heartbeat-expired", "expired owner heartbeat should be the stale recovery gate", { decision }); assertCondition(diagnostics.state === "stale-active", "diagnostics must mark stale active only after owner heartbeat expiry", diagnostics as unknown as Record); assertCondition(diagnostics.effectiveLiveness === "at-risk" && diagnostics.recommendedAction === "investigate-heartbeat-risk", "expired heartbeat diagnostics must surface at-risk liveness", diagnostics as unknown as Record); assertCondition(diagnostics.staleRecoveryCandidateTaskIds.includes(task.id), "expired owner heartbeat should create a stale candidate", diagnostics as unknown as Record); return { name: "code-queue:stale-active-owner-expired", ok: true, detail: { decision, state: diagnostics.state, effectiveLiveness: diagnostics.effectiveLiveness, recommendedAction: diagnostics.recommendedAction, staleRecoveryCandidateTaskIds: diagnostics.staleRecoveryCandidateTaskIds, }, }; } function recoverStaleActiveTaskForFixture(task: QueueTask): number { const decision = staleRecoveryCandidate({ task, localActive: false, now }); if (!decision.allowed) return 0; task.status = "retry_wait"; task.finishedAt = null; task.readAt = null; task.activeTurnId = null; task.lastError = "fixture stale-active recovery"; task.nextMode = "retry"; task.nextPrompt = "fixture recovery prompt"; task.updatedAt = now; return 1; } function checkStaleActiveRecoveryTransition(): FixtureCheck { const task = fixtureTask("codex_fixture_stale_recovery_1", "running", heartbeat("codex_fixture_stale_recovery_1", expiredAt, { lastObservedAgentEventAt: expiredAt, lastPersistedTraceAt: expiredAt, })); const freshTask = fixtureTask("codex_fixture_fresh_no_recovery_1", "running", heartbeat("codex_fixture_fresh_no_recovery_1", freshAt)); const diagnosticsBefore = schedulerDiagnostics([task], []); const recovered = recoverStaleActiveTaskForFixture(task); const freshRecovered = recoverStaleActiveTaskForFixture(freshTask); const diagnosticsAfter = schedulerDiagnostics([task], []); assertCondition(recovered === 1, "expired active task should be recovered exactly once", { recovered, task }); assertCondition(freshRecovered === 0 && freshTask.status === "running", "fresh owner heartbeat must block recovery", { freshRecovered, freshStatus: freshTask.status }); assertCondition(task.status === "retry_wait" && task.activeTurnId === null && task.nextMode === "retry", "recovered task should re-enter retry_wait with active turn cleared", { task }); assertCondition(diagnosticsBefore.staleRecoveryCandidateTaskIds.includes("codex_fixture_stale_recovery_1"), "before recovery should show stale candidate", diagnosticsBefore as unknown as Record); assertCondition(!diagnosticsAfter.staleRecoveryCandidateTaskIds.includes("codex_fixture_stale_recovery_1"), "after recovery should remove stale candidate", diagnosticsAfter as unknown as Record); return { name: "code-queue:stale-active-recovery-transition", ok: true, detail: { recovered, freshRecovered, status: task.status, activeTurnId: task.activeTurnId, nextMode: task.nextMode, candidatesBefore: diagnosticsBefore.staleRecoveryCandidateTaskIds, candidatesAfter: diagnosticsAfter.staleRecoveryCandidateTaskIds, }, }; } function checkControlPlaneSplitBrainDiagnostics(): FixtureCheck { const task = fixtureTask("codex_fixture_split_1", "running", heartbeat("codex_fixture_split_1", freshAt)); const diagnostics = buildExecutionDiagnostics({ now, controlPlane: "master-code-queue-mgr", executionStateSource: "postgres-control-plane", tasks: [task], activeRuns: [], activeRunSlotCount: 0, oaPublisher: null, }); assertCondition(diagnostics.state === "split-brain" && diagnostics.splitBrain === true, "master postgres-control-plane must report split-brain when DB active has fresh scheduler heartbeat", diagnostics as unknown as Record); assertCondition(diagnostics.splitBrainLive === true && diagnostics.effectiveLiveness === "live", "fresh split-brain diagnostics must be degraded but live", diagnostics as unknown as Record); assertCondition(diagnostics.recommendedAction === "continue-supervision", "fresh split-brain diagnostics must recommend continued supervision", diagnostics as unknown as Record); assertCondition(diagnostics.schedulerActiveRunSlotCount === 0 && diagnostics.databaseActiveTaskCount === 1, "split-brain fixture should preserve the exact control-plane divergence", diagnostics as unknown as Record); return { name: "code-queue:control-plane-split-brain-diagnostics", ok: true, detail: { state: diagnostics.state, splitBrain: diagnostics.splitBrain, splitBrainLive: diagnostics.splitBrainLive, effectiveLiveness: diagnostics.effectiveLiveness, recommendedAction: diagnostics.recommendedAction, executionStateSource: diagnostics.executionStateSource, databaseActiveTaskIds: diagnostics.databaseActiveTaskIds, schedulerActiveRunSlotCount: diagnostics.schedulerActiveRunSlotCount, heartbeatFreshTaskIds: diagnostics.heartbeatFreshTaskIds, }, }; } function checkOaPublisherDegradedVisible(): FixtureCheck { const oaPublisher = { pending: 3, lastError: "fixture OA publish retry", lastPublishedAt: null }; const diagnostics = schedulerDiagnostics([], [], oaPublisher); assertCondition(diagnostics.state === "degraded", "OA publisher pending/lastError must degrade diagnostics", diagnostics as unknown as Record); assertCondition(diagnostics.oaPublisher === oaPublisher, "OA publisher detail must remain visible", diagnostics as unknown as Record); return { name: "code-queue:oa-publisher-degraded-visible", ok: true, detail: { state: diagnostics.state, degraded: diagnostics.degraded, oaPublisher: diagnostics.oaPublisher as Record, }, }; } export function runCodeQueueLivenessFixtureChecks(only: string[] = []): { ok: boolean; checks: FixtureCheck[]; failures: Array<{ name: string; error: string }> } { const selected = new Set(only.filter((name) => name.trim().length > 0)); const runners: Array<[CodeQueueLivenessCheckName, () => FixtureCheck]> = [ ["code-queue:active-run-heartbeat-visible", checkActiveRunHeartbeatVisible], ["code-queue:trace-gap-not-stale", checkTraceGapNotStale], ["code-queue:heartbeat-threshold-normalized", checkHeartbeatThresholdNormalized], ["code-queue:stale-active-owner-expired", checkStaleActiveOwnerExpired], ["code-queue:stale-active-recovery-transition", checkStaleActiveRecoveryTransition], ["code-queue:control-plane-split-brain-diagnostics", checkControlPlaneSplitBrainDiagnostics], ["code-queue:oa-publisher-degraded-visible", checkOaPublisherDegradedVisible], ]; const checks: FixtureCheck[] = []; const failures: Array<{ name: string; error: string }> = []; for (const [name, run] of runners) { if (selected.size > 0 && !selected.has(name)) continue; try { checks.push(run()); } catch (error) { failures.push({ name, error: error instanceof Error ? error.message : String(error) }); checks.push({ name, ok: false, detail: { error: error instanceof Error ? error.message : String(error) } }); } } if (checks.length === 0) throw new Error(`no Code Queue liveness fixture checks matched: ${Array.from(selected).join(", ")}`); return { ok: failures.length === 0, checks, failures }; }