pikasTech-unidesk/scripts/src/code-queue-liveness-fixtures.ts

// SPEC: PJ2026-01060308 cicd-yaml-targets draft-2026-06-25-cicd-yaml-targets.
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<string, unknown>;
}

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";
const fixtureProviderId = "D601";

function assertCondition(condition: unknown, message: string, detail: Record<string, unknown> = {}): 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: fixtureProviderId,
    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> = {}): 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<string, unknown>);
  assertCondition(tooLowDiagnostics.schedulerHeartbeatStaleMs === schedulerHeartbeatStaleMs, "too-low diagnostics threshold should normalize to default stale window", tooLowDiagnostics as unknown as Record<string, unknown>);
  assertCondition(zeroDiagnostics.state === "healthy" && zeroDiagnostics.effectiveLiveness === "healthy", "fresh heartbeat with zero requested threshold must stay healthy", zeroDiagnostics as unknown as Record<string, unknown>);
  assertCondition(tooLowDiagnostics.state === "healthy" && tooLowDiagnostics.effectiveLiveness === "healthy", "fresh heartbeat with too-low requested threshold must stay healthy", tooLowDiagnostics as unknown as Record<string, unknown>);
  assertCondition(zeroDiagnostics.heartbeatFreshTaskIds.includes(task.id), "fresh heartbeat must remain in fresh task ids", zeroDiagnostics as unknown as Record<string, unknown>);
  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<string, unknown>);
  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<string, unknown>);
  assertCondition(diagnostics.schedulerActiveTaskIds.includes(task.id), "scheduler active task id must be visible", diagnostics as unknown as Record<string, unknown>);
  assertCondition(diagnostics.lastSchedulerHeartbeatAt === freshAt, "last scheduler heartbeat must be surfaced", diagnostics as unknown as Record<string, unknown>);
  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<string, unknown>);
  assertCondition(!diagnostics.staleRecoveryCandidateTaskIds.includes(task.id), "trace gap must not enter stale recovery candidates", diagnostics as unknown as Record<string, unknown>);
  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<string, unknown>);
  assertCondition(diagnostics.effectiveLiveness === "at-risk" && diagnostics.recommendedAction === "investigate-heartbeat-risk", "expired heartbeat diagnostics must surface at-risk liveness", diagnostics as unknown as Record<string, unknown>);
  assertCondition(diagnostics.staleRecoveryCandidateTaskIds.includes(task.id), "expired owner heartbeat should create a stale candidate", diagnostics as unknown as Record<string, unknown>);
  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<string, unknown>);
  assertCondition(!diagnosticsAfter.staleRecoveryCandidateTaskIds.includes("codex_fixture_stale_recovery_1"), "after recovery should remove stale candidate", diagnosticsAfter as unknown as Record<string, unknown>);
  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<string, unknown>);
  assertCondition(diagnostics.splitBrainLive === true && diagnostics.effectiveLiveness === "live", "fresh split-brain diagnostics must be degraded but live", diagnostics as unknown as Record<string, unknown>);
  assertCondition(diagnostics.recommendedAction === "continue-supervision", "fresh split-brain diagnostics must recommend continued supervision", diagnostics as unknown as Record<string, unknown>);
  assertCondition(diagnostics.schedulerActiveRunSlotCount === 0 && diagnostics.databaseActiveTaskCount === 1, "split-brain fixture should preserve the exact control-plane divergence", diagnostics as unknown as Record<string, unknown>);
  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<string, unknown>);
  assertCondition(diagnostics.oaPublisher === oaPublisher, "OA publisher detail must remain visible", diagnostics as unknown as Record<string, unknown>);
  return {
    name: "code-queue:oa-publisher-degraded-visible",
    ok: true,
    detail: {
      state: diagnostics.state,
      degraded: diagnostics.degraded,
      oaPublisher: diagnostics.oaPublisher as Record<string, unknown>,
    },
  };
}

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 };
}