"use client"; /** * Pipeline truth for the live wall (server-26#232). * * The old '● Live · Ns ago' pill was reset by any node heartbeat, so the wall * read live while transcription/correlation were dead (#196). This hook keys * health to the PIPELINE instead: radio → STT → AI, plus the context feeds. * * Two sources, deliberately both: * - Data timestamps on what the wall already subscribes to (recent calls, * active incidents). These survive a c2-core restart. * - GET /health/ai (main.py, auth-free) — per-tier provider state. It is * in-memory per process and resets on restart, so it can only ADD a * degraded verdict, never clear one the data shows. */ import { useEffect, useMemo, useState } from "react"; import type { CallRecord, CameraProviderFeed, IncidentRecord, NodeRecord, Ny511FeedStatus } from "@/lib/types"; const BASE = process.env.NEXT_PUBLIC_C2_URL ?? "http://localhost:8000"; const HEALTH_POLL_MS = 60_000; const TICK_MS = 30_000; function envMinutes(): number { const n = Number(process.env.NEXT_PUBLIC_WALL_STALE_MIN); return Number.isFinite(n) && n > 0 ? n : 10; } /** Minutes a stage may be dry before the wall calls itself degraded. */ export const WALL_STALE_MIN = envMinutes(); /** * Radio silence and "no incident written" are weaker evidence than "radio * heard but not transcribed" — a quiet county at 4am is not an outage, and the * correlator orphans in doubt — so those two stages get a longer leash. */ const ABSENCE_FACTOR = 3; /** * Calls Whisper returned nothing for (silence, kerchunks, degenerate output) * get no transcript by design, so one or two blank calls after the last * transcript are normal. It takes this many to call STT dry. */ const MIN_UNSERVED = 3; // ── /health/ai response (app/internal/ai_health.py snapshot()) ────────────── export interface AiTierState { degraded: boolean; permanent: boolean; provider: string | null; model: string | null; problem: string | null; fix: string | null; first_seen: string | null; last_seen: string | null; consecutive_failures: number; alerted: boolean; } export interface AiHealthResponse { tiers: Record; } export type StageKey = "radio" | "stt" | "ai" | "feeds"; export type StageState = "ok" | "degraded" | "unknown"; export interface PipelineStage { key: StageKey; label: string; state: StageState; /** Newest evidence this stage produced output, or null if none in the window. */ lastAt: Date | null; /** One-line human explanation, shown as the chip tooltip / bar text. */ detail: string; } export interface PipelineHealth { stages: PipelineStage[]; /** True when a core stage (radio/STT/AI) is dry or an AI tier is degraded. */ degraded: boolean; /** Bar text, e.g. "radio heard, not transcribed since 12:15 · transcription tier degraded (openai: …)". */ message: string | null; degradedTiers: { tier: string; state: AiTierState }[]; /** null until the first /health/ai answer; false if it could not be reached. */ aiHealthReachable: boolean | null; /** Distinct nodes with a call started in the last WALL_STALE_MIN minutes. */ nodesHearing: number; newestCallAt: Date | null; staleMin: number; } function useAiHealth(): { data: AiHealthResponse | null; reachable: boolean | null } { const [data, setData] = useState(null); const [reachable, setReachable] = useState(null); useEffect(() => { let cancelled = false; const load = () => { if (document.hidden) return; fetch(`${BASE}/health/ai`, { cache: "no-store" }) .then((r) => (r.ok ? r.json() : Promise.reject(new Error(`HTTP ${r.status}`)))) .then((j: AiHealthResponse) => { if (cancelled) return; setData(j && typeof j.tiers === "object" ? j : { tiers: {} }); setReachable(true); }) .catch((e) => { if (cancelled) return; console.error("health/ai:", e); setReachable(false); }); }; load(); const id = setInterval(load, HEALTH_POLL_MS); const onVis = () => { if (!document.hidden) load(); }; document.addEventListener("visibilitychange", onVis); return () => { cancelled = true; clearInterval(id); document.removeEventListener("visibilitychange", onVis); }; }, []); return { data, reachable }; } const ms = (iso: string | null | undefined): number => { if (!iso) return NaN; const t = Date.parse(iso); return Number.isFinite(t) ? t : NaN; }; const maxDate = (vals: number[]): Date | null => { const ok = vals.filter((v) => Number.isFinite(v)); return ok.length ? new Date(Math.max(...ok)) : null; }; export function clockTime(d: Date): string { return d.toLocaleTimeString([], { hour: "2-digit", minute: "2-digit", hour12: false }); } function agoText(d: Date | null, now: number): string { if (!d) return "none in window"; const s = Math.max(0, Math.floor((now - d.getTime()) / 1000)); if (s < 60) return `${s}s ago`; if (s < 3600) return `${Math.floor(s / 60)}m ago`; return `${Math.floor(s / 3600)}h ago`; } const hasTranscript = (c: CallRecord) => !!(c.transcript || c.transcript_corrected); const isAiTier = (t: string) => t !== "transcription"; export interface PipelineInput { calls: CallRecord[]; incidents: IncidentRecord[]; nodes: NodeRecord[]; eventsStatus?: Ny511FeedStatus; cameraProviders: CameraProviderFeed[]; } export function usePipelineHealth(input: PipelineInput): PipelineHealth { const { calls, incidents, nodes, eventsStatus, cameraProviders } = input; const { data: ai, reachable } = useAiHealth(); // Staleness is a function of wall-clock time, not only of new data arriving. const [now, setNow] = useState(() => Date.now()); useEffect(() => { const id = setInterval(() => setNow(Date.now()), TICK_MS); return () => clearInterval(id); }, []); return useMemo( () => derivePipelineHealth({ calls, incidents, nodes, eventsStatus, cameraProviders }, ai, reachable, now), [calls, incidents, nodes, eventsStatus, cameraProviders, ai, reachable, now], ); } /** Pure derivation — separated from the hook so it can be exercised without React. */ export function derivePipelineHealth( input: PipelineInput, ai: AiHealthResponse | null, reachable: boolean | null, now: number, ): PipelineHealth { const { calls, incidents, nodes, eventsStatus, cameraProviders } = input; const staleMs = WALL_STALE_MIN * 60_000; const absenceMs = staleMs * ABSENCE_FACTOR; // Canonical, finished recordings only: a second node's copy of the same // transmission (duplicate_of) is never transcribed, and an in-flight call // has not been uploaded yet. const real = calls.filter((c) => !c.duplicate_of && c.status !== "active" && Number.isFinite(ms(c.started_at))); const tiers = Object.entries(ai?.tiers ?? {}); const degradedTiers = tiers.filter(([, s]) => s?.degraded).map(([tier, state]) => ({ tier, state })); const tierText = (tier: string, s: AiTierState) => `${tier} tier degraded${s.provider ? ` (${s.provider}${s.model ? ` ${s.model}` : ""})` : ""}${s.problem ? `: ${s.problem}` : ""}`; // ── Radio ──────────────────────────────────────────────────────────────── const newestCallAt = maxDate(calls.map((c) => ms(c.started_at))); const anyNodeUp = nodes.some((n) => n.status !== "offline"); const radioDry = anyNodeUp && (!newestCallAt || now - newestCallAt.getTime() > absenceMs); const radio: PipelineStage = { key: "radio", label: "Radio", lastAt: newestCallAt, state: nodes.length === 0 ? "unknown" : radioDry ? "degraded" : "ok", detail: newestCallAt ? `last call heard ${agoText(newestCallAt, now)}` : "no calls in the recent window", }; const nodesHearing = new Set( calls.filter((c) => now - ms(c.started_at) <= staleMs).map((c) => c.node_id), ).size; // ── STT: radio heard, nothing transcribed ──────────────────────────────── const newestTranscriptAt = maxDate(real.filter(hasTranscript).map((c) => ms(c.started_at))); const unserved = real .filter((c) => !hasTranscript(c) && (!newestTranscriptAt || ms(c.started_at) > newestTranscriptAt.getTime())) .map((c) => ms(c.started_at)); const oldestUnserved = unserved.length ? Math.min(...unserved) : NaN; const sttDry = unserved.length >= MIN_UNSERVED && now - oldestUnserved > staleMs; const sttTier = degradedTiers.find((t) => t.tier === "transcription"); const sttSince = newestTranscriptAt ?? (Number.isFinite(oldestUnserved) ? new Date(oldestUnserved) : null); const stt: PipelineStage = { key: "stt", label: "Transcripts", lastAt: newestTranscriptAt, state: sttDry || sttTier ? "degraded" : real.length ? "ok" : "unknown", detail: sttDry ? `radio heard, not transcribed since ${sttSince ? clockTime(sttSince) : "—"} (${unserved.length} calls waiting)` : newestTranscriptAt ? `last transcript ${agoText(newestTranscriptAt, now)}` : "no transcripts in the recent window", }; // ── AI: transcribed, but nothing correlated ───────────────────────────── const newestAiAt = maxDate([ ...incidents.flatMap((i) => [ms(i.updated_at), ms(i.last_thin_at)]), ...real.filter((c) => (c.incident_ids?.length ?? 0) > 0 || !!c.incident_id).map((c) => ms(c.started_at)), ]); const transcribedSinceAi = real .filter((c) => hasTranscript(c) && (!newestAiAt || ms(c.started_at) > newestAiAt.getTime())) .map((c) => ms(c.started_at)); const oldestUncorrelated = transcribedSinceAi.length ? Math.min(...transcribedSinceAi) : NaN; const aiDry = transcribedSinceAi.length >= MIN_UNSERVED && now - oldestUncorrelated > absenceMs; const aiTiers = degradedTiers.filter((t) => isAiTier(t.tier)); const ai_: PipelineStage = { key: "ai", label: "AI", lastAt: newestAiAt, state: aiDry || aiTiers.length ? "degraded" : reachable === false && !newestAiAt ? "unknown" : "ok", detail: aiDry ? `transcribed, no incident written since ${newestAiAt ? clockTime(newestAiAt) : clockTime(new Date(oldestUncorrelated))}` : `last incident write ${agoText(newestAiAt, now)}${reachable === false ? " · /health/ai unreachable" : ""}`, }; // ── Context feeds (chip only — a dead camera feed doesn't blind the pipeline) ── const feedProblems: string[] = []; if (eventsStatus && !eventsStatus.disabled) { if (eventsStatus.error) feedProblems.push(`511: ${eventsStatus.error}`); else if (eventsStatus.fetched_at && now / 1000 - eventsStatus.fetched_at > absenceMs / 1000) feedProblems.push(`511 data ${Math.round((now / 1000 - eventsStatus.fetched_at) / 60)}m old`); } for (const p of cameraProviders) if (p.error) feedProblems.push(`${p.name}: ${p.error}`); const feeds: PipelineStage = { key: "feeds", label: "Feeds", lastAt: null, state: feedProblems.length ? "degraded" : "ok", detail: feedProblems.length ? feedProblems.join(" · ") : "511 / camera providers OK", }; // ── Verdict ───────────────────────────────────────────────────────────── const parts: string[] = []; if (radio.state === "degraded") parts.push(`nodes checking in, no radio heard since ${newestCallAt ? clockTime(newestCallAt) : "—"}`); if (sttDry) parts.push(stt.detail); if (aiDry) parts.push(ai_.detail); for (const t of degradedTiers) parts.push(tierText(t.tier, t.state)); const degraded = parts.length > 0; return { stages: [radio, stt, ai_, feeds], degraded, message: degraded ? parts.join(" · ") : null, degradedTiers, aiHealthReachable: reachable, nodesHearing, newestCallAt, staleMin: WALL_STALE_MIN, }; }