Merge pull request 'Live wall: truth strip keyed to the pipeline, not node heartbeats (#232)' (#252) from feat/232-truth-strip into main

drb-frontend/components/LiveView.tsx

@@ -27,6 +27,8 @@ import { TimeScrubber } from "@/components/TimeScrubber";
 import { TranscriptTicker } from "@/components/live/TranscriptTicker";
 import { CameraDrill } from "@/components/live/CameraDrill";
 import { IncidentPanel, RegionPanel } from "@/components/live/WallPanels";
+import { TruthStrip } from "@/components/live/TruthStrip";
+import { usePipelineHealth } from "@/lib/usePipelineHealth";
 
 const MapView = dynamic(() => import("@/components/MapView"), { ssr: false });
 
@@ -93,12 +95,11 @@ export function LiveView() {
   const { calls: recentCalls, loading: callsLoading } = useCalls(500);
   const { aircraft } = useAircraft();
   const { vessels } = useVessels();
-  const [lastUpdated, setLastUpdated] = useState<Date | null>(null);
   const [clock, setClock] = useState<string>("");
 
-  useEffect(() => {
-    if (!nodesLoading) setLastUpdated(new Date());
-  }, [nodes, activeCalls, activeIncidents, nodesLoading]);
+  // No heartbeat-driven "last updated" any more (server-26#232): a node
+  // check-in kept '● Live' fresh while STT/correlation were dead (#196). The
+  // wall's liveness now comes from usePipelineHealth, keyed to pipeline output.
   useEffect(() => {
     const tick = () => setClock(new Date().toLocaleTimeString([], { hour12: false }));
     tick();
@@ -156,6 +157,13 @@ export function LiveView() {
   }, [pointsById, cams.cameras]);
   const camCountByIncident = useMemo(() => new Map(Array.from(nearById.entries()).map(([id, n]) => [id, n.length])), [nearById]);
   const events = useRegionEvents(regionBox);
+  const health = usePipelineHealth({
+    calls: recentCalls,
+    incidents: activeIncidents,
+    nodes,
+    eventsStatus: events.events_status,
+    cameraProviders: cams.providers,
+  });
   const onAir = useMemo(() => onAirIncidentIds(activeIncidents, activeCalls), [activeIncidents, activeCalls]);
 
   const selected = link.inc ? activeIncidents.find((i) => i.incident_id === link.inc) ?? null : null;
@@ -220,13 +228,14 @@ export function LiveView() {
 
   // ── KPIs ──────────────────────────────────────────────────────────────────
   const count = (sev: string) => activeIncidents.filter((i) => i.severity === sev).length;
-  const nodesUp = nodes.filter((n) => n.status !== "offline").length;
+  // "Nodes hearing": a call in the last N min, not merely not-offline — a node
+  // that checks in but hears nothing is not contributing to the picture.
+  const nodesHearing = health.nodesHearing;
   const eventsOff = !!events.events_status?.disabled;
 
-  const configuredButQuiet = !nodesLoading && nodes.length > 0 && activeIncidents.length === 0;
-  const mostRecentSeen = configuredButQuiet
-    ? nodes.map((n) => n.last_seen).filter((s): s is string => !!s).sort().at(-1)
-    : null;
+  // "Listening" only when every stage is healthy; otherwise the amber bar says why.
+  const configuredButQuiet = !nodesLoading && !callsLoading && nodes.length > 0
+    && activeIncidents.length === 0 && !health.degraded;
 
   const crumb = (label: string, active: boolean, enabled: boolean, onClick: () => void) => (
     <button type="button" disabled={!enabled} onClick={onClick} aria-current={active ? "true" : undefined}
@@ -251,13 +260,14 @@ export function LiveView() {
           <Kpi value={count("major")} label="major" />
           <Kpi value={count("moderate")} label="moderate" />
           <Kpi value={activeIncidents.length} label="active" />
-          <Kpi value={`${nodesUp}/${nodes.length}`} label="nodes" />
+          <Kpi value={`${nodesHearing}/${nodes.length}`} label={`nodes hearing (${health.staleMin}m)`} />
           <Kpi value={cams.cameras.length} label="cams" />
           <Kpi value={aircraft.length} label="aircraft" />
           <Kpi value={vessels.length} label="vessels" />
           <Kpi value={eventsOff ? "—" : events.events?.length ?? 0} label="road events" />
         </div>
         <div className="ml-auto flex items-center gap-3">
+          <TruthStrip health={health} placement="chips" />
           <button type="button" aria-pressed={link.tour} onClick={() => setLink({ ...link, cam: null, tour: !link.tour })}
             title="Wall mode: cycle through active incidents, worst first"
             className={`px-2 py-0.5 rounded border ${link.tour ? "bg-accent text-white border-accent" : "border-line text-ink-2 hover:text-ink"}`}>
@@ -271,9 +281,13 @@ export function LiveView() {
           <span className="font-mono text-ink-2 tabular-nums">{clock}</span>
         </div>
       </div>
+      <TruthStrip health={health} placement="bar" />
 
       <div className="flex-1 min-h-0 flex flex-col lg:flex-row">
         <div className="relative h-[55vh] lg:h-auto lg:flex-1 min-w-0 shrink-0">
+          {/* Degraded → grey the map and its incident rail (the rail lives inside
+              MapView, so it is wrapped here rather than edited). */}
+          <div className={`w-full h-full transition-[filter,opacity] ${health.degraded ? "grayscale opacity-60" : ""}`}>
           {nodesLoading || callsLoading ? (
             <div className="w-full h-full flex items-center justify-center text-ink-muted text-sm">Loading map…</div>
           ) : (
@@ -282,7 +296,6 @@ export function LiveView() {
               activeCalls={activeCalls}
               incidents={activeIncidents}
               calls={recentCalls}
-              lastUpdated={lastUpdated}
               selectedIncidentId={selected?.incident_id ?? null}
               onSelectIncident={selectIncident}
               nearCameras={near}
@@ -291,13 +304,14 @@ export function LiveView() {
               camCountByIncident={camCountByIncident}
             />
           )}
+          </div>
 
           {/* Configured-and-quiet — distinct from "no nodes at all" (chunk 10's
               Activation screen handles the zero-node case). */}
           {configuredButQuiet && (
             <div className="absolute top-12 left-1/2 -translate-x-1/2 z-[1001] pointer-events-none">
               <span className="bg-surface/90 border border-line rounded-full px-3 py-1 text-xs text-ink-2 whitespace-nowrap">
-                ● Listening{mostRecentSeen ? ` — last check-in ${timeAgo(new Date(mostRecentSeen))}` : ""}
+                ● Listening{health.newestCallAt ? ` — last call heard ${timeAgo(health.newestCallAt)}` : ""}
               </span>
             </div>
           )}

drb-frontend/components/live/TruthStrip.tsx

@@ -0,0 +1,53 @@
+"use client";
+
+/**
+ * Truth strip (server-26#232) — replaces the heartbeat-driven '● Live' pill.
+ *
+ * Healthy: compact per-stage chips (Radio · Transcripts · AI · Feeds), each
+ * with its last-output age. Any core stage dry past the threshold, or any
+ * /health/ai tier degraded: a full-width amber bar that says what broke and
+ * since when — "DEGRADED — radio heard, not transcribed since 12:15".
+ * Theme tokens only (sev-moderate is the amber), so both themes follow.
+ */
+import type { PipelineHealth, PipelineStage } from "@/lib/usePipelineHealth";
+
+function dotClass(s: PipelineStage["state"]): string {
+  if (s === "degraded") return "bg-sev-moderate";
+  if (s === "unknown") return "bg-ink-muted";
+  return "bg-ink-2";
+}
+
+export function TruthChips({ health }: { health: PipelineHealth }) {
+  return (
+    <div className="flex items-center gap-1.5 flex-wrap" role="status" aria-label="Pipeline health">
+      {health.stages.map((s) => (
+        <span key={s.key} title={s.detail}
+          className={`inline-flex items-center gap-1 px-1.5 py-0.5 rounded border text-[11px] whitespace-nowrap ${
+            s.state === "degraded" ? "border-sev-moderate text-ink bg-sev-moderate/10" : "border-line text-ink-2"}`}>
+          <span className={`w-1.5 h-1.5 rounded-full ${dotClass(s.state)}`} aria-hidden="true" />
+          {s.label}
+          <span className="sr-only">: {s.state === "ok" ? "healthy" : s.state}, {s.detail}</span>
+        </span>
+      ))}
+    </div>
+  );
+}
+
+export function TruthBar({ health }: { health: PipelineHealth }) {
+  if (!health.degraded || !health.message) return null;
+  return (
+    <div role="alert"
+      className="shrink-0 w-full px-3 py-1.5 border-b border-sev-moderate bg-sev-moderate/15 text-xs text-ink flex flex-wrap items-baseline gap-x-2">
+      <b className="font-bold tracking-wide">DEGRADED</b>
+      <span>— {health.message}</span>
+      <span className="ml-auto text-ink-muted whitespace-nowrap">
+        stale after {health.staleMin} min · map shows last-known state
+      </span>
+    </div>
+  );
+}
+
+/** Chips for the wall bar, plus the bar itself directly below it when degraded. */
+export function TruthStrip({ health, placement }: { health: PipelineHealth; placement: "chips" | "bar" }) {
+  return placement === "chips" ? <TruthChips health={health} /> : <TruthBar health={health} />;
+}

drb-frontend/lib/usePipelineHealth.ts

@@ -0,0 +1,271 @@
+"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<string, AiTierState>;
+}
+
+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<AiHealthResponse | null>(null);
+  const [reachable, setReachable] = useState<boolean | null>(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");
+  // Every configured node offline is not "Listening" — nothing can be heard.
+  const allNodesOffline = nodes.length > 0 && !anyNodeUp;
+  const radioDry = anyNodeUp && (!newestCallAt || now - newestCallAt.getTime() > absenceMs);
+  const radio: PipelineStage = {
+    key: "radio", label: "Radio", lastAt: newestCallAt,
+    state: nodes.length === 0 ? "unknown" : radioDry || allNodesOffline ? "degraded" : "ok",
+    detail: allNodesOffline
+      ? `all ${nodes.length} node${nodes.length === 1 ? "" : "s"} offline`
+      : newestCallAt ? `last call heard ${agoText(newestCallAt, now)}` : "no calls in the recent window",
+  };
+  // Bounded to configured nodes so a call from a since-removed node can't push
+  // the KPI past N/N.
+  const knownNodes = new Set(nodes.map((n) => n.node_id));
+  const nodesHearing = new Set(
+    calls.filter((c) => now - ms(c.started_at) <= staleMs && knownNodes.has(c.node_id)).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))}`
+      : `${newestAiAt ? `last incident write ${agoText(newestAiAt, now)}` : "no incident writes in the recent window"}${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 (allNodesOffline) parts.push(`${radio.detail}${newestCallAt ? `, last call heard ${clockTime(newestCallAt)}` : ""}`);
+  else if (radioDry) 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,
+  };
+}