Wire AIS end to end: telemetry ingestion + live map overlay

node-26#9, same shape as the ADS-B commit. Adds POST /telemetry/ais (same node-key auth, same org_id-stamped upsert-by-key pattern, this time by mmsi into a new `vessels` collection) and its docInMyOrg() firestore rule. Frontend gets useVessels() (mirrors useAircraft(), longer staleness window since AIS position reports are minutes apart, not seconds) and an opt-in "Vessels" map overlay. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Wire ADS-B end to end: telemetry ingestion + live map overlay
2026-09-20 16:12:56 -04:00 · 2026-09-20 16:08:29 -04:00 · 2026-09-20 15:56:33 -04:00 · 2026-09-20 15:42:48 -04:00 · 2026-09-20 14:25:01 -04:00 · 2026-09-14 00:31:14 -04:00
24 changed files with 1230 additions and 255 deletions
@@ -47,3 +47,4 @@ Thumbs.db

 # Out of scope - not a deployed service (server-26#56)
 drb-telegram-bot/
+.claude/worktrees/
@@ -90,29 +90,19 @@ class Settings(BaseSettings):
    unit_continuity_max_idle_minutes: int = 20  # unit-continuity path: skip if incident idle > this
    recorrelation_scan_minutes: int = 60      # re-examine orphaned calls ended within this window
    tg_fast_path_idle_minutes: int = 90       # fast path: max minutes since incident last updated
-    # Dispatch channels only: tier-2 thin calls attach to a lone candidate idle < this.
-    # Was 10, which is long enough for the channel to have moved on to something else:
-    # on 2026-08-16 a "72 at Holland Station" incident absorbed a Grand Central train
-    # meet 9.6 min later, and a status check absorbed a records lookup at 9.7 min.
-    # Across that dump every correct thin attach was <= 3.4 min idle and every wrong
-    # one was >= 8.2, so 5 separates them with room on both sides. Genuine
-    # back-and-forth is handled by the 30-second tier-1 path above this.
-    # Second consumer (server-26#115): routers/upload.py's LLM-orphan-gate escape
-    # hatch (_recent_incident_on_same_talkgroup) reuses this same value, selected
-    # the same way (dispatch vs tactical) via _is_dispatch_channel. Retuning this
-    # for fast/thin reasons moves that gate's behavior too — check both call
-    # sites before changing it.
+    # Tier-2 thin calls attach to a lone candidate idle < this, on every
+    # channel (server-26#133/#134 removed the dispatch/tactical split — a
+    # channel's name doesn't change how much scrutiny it gets). Was 10, which
+    # is long enough for the channel to have moved on to something else: on
+    # 2026-08-16 a "72 at Holland Station" incident absorbed a Grand Central
+    # train meet 9.6 min later, and a status check absorbed a records lookup
+    # at 9.7 min. Every correct thin attach in that dump was <= 3.4 min idle
+    # and every wrong one was >= 8.2, so 5 separates them with room on both
+    # sides. Genuine back-and-forth is handled by the 30-second tier-1 path
+    # above this. Also the escape hatch in routers/upload.py's LLM-orphan gate
+    # (_recent_incident_on_same_talkgroup, server-26#115) — check both call
+    # sites before retuning this.
    tg_dispatch_thin_idle_minutes: int = 5
-    # Every other channel: tier-2 thin calls attach to a lone candidate idle < this.
-    # Non-dispatch talkgroups previously had NO tier-2 bound at all — they used the
-    # whole 90-minute tg_fast_path_idle_minutes window with no single-candidate
-    # requirement and no fit test, which is the widest version of the 2026-08-20
-    # over-merge. A tactical channel really is dedicated to one scene, so it earns
-    # a longer window than a dispatch backbone, but not an unbounded one: 15 sits
-    # inside the 20-minute tactical-default window in _call_fits_incident, so the
-    # no-evidence thin path is never more permissive than the fit-tested path on
-    # the same channel.
-    tg_thin_idle_minutes: int = 15

    # ── Hard caps: an incident past either of these stops accepting calls ──────
    # Enforced on every correlation path (see _incident_at_capacity). Pairwise fit
@@ -98,13 +98,6 @@ _PURSUIT_PROXIMITY_KM         = 20.0  # expanded radius for moving incidents
 # overlap with the candidate OR a distance under this tighter bar.
 _LOCATION_TIGHT_PROXIMITY_KM  = 0.2

-_DISPATCH_TG_RE = re.compile(
-    r"\bdispatch\b|\bdisp\b"
-    r"|\bpatched\b"   # patched channels aggregate multiple call streams
-    r"|\bprimary\b",  # "Primary" channels serve as shared backbones
-    re.IGNORECASE,
-)
-
 # Matches route/road identifiers in location strings for cross-system parent detection.
 # Groups: numbered routes (Route 202, NY-9, US-6, I-87, CR-35) and named parkways/highways.
 _ROAD_RE = re.compile(
@@ -494,13 +487,6 @@ def _resolve_incident_title(
    return {}


-def _is_dispatch_channel(talkgroup_name: Optional[str]) -> bool:
-    """True when the talkgroup is a shared dispatch backbone (not a tactical/working channel)."""
-    if not talkgroup_name:
-        return False
-    return bool(_DISPATCH_TG_RE.search(talkgroup_name))
-
-
 def _incident_idle_minutes(inc: dict, now: datetime) -> float:
    """Minutes since the incident was last updated (or started). Returns 9999 on parse error."""
    try:
@@ -973,14 +959,13 @@ def _run_decision(ctx: dict) -> dict:
    if talkgroup_id is not None and system_id:
        tg_str     = str(talkgroup_id)
        # talkgroup_name may be None when the upload form omits it (node sets it
-        # directly on the Firestore call doc).  Fall back to the call doc so that
-        # dispatch-channel strictness works regardless of how the call arrived.
+        # directly on the Firestore call doc). Fall back to the call doc so the
+        # log lines below still name the channel.
        effective_talkgroup_name = talkgroup_name or call_doc.get("talkgroup_name")
-        is_dispatch = _is_dispatch_channel(effective_talkgroup_name)
        if effective_talkgroup_name != talkgroup_name:
            logger.info(
                f"Correlator: talkgroup_name missing from request for call {call_id}, "
-                f"resolved from call doc: {effective_talkgroup_name!r} → is_dispatch={is_dispatch}"
+                f"resolved from call doc: {effective_talkgroup_name!r}"
            )

        tg_matches = [
@@ -1024,10 +1009,7 @@ def _run_decision(ctx: dict) -> dict:
            # single-candidate requirement and no fit test of any kind.  Four
            # hours is not a bound, and neither is ninety minutes.
            THIN_CONVERSATIONAL_SECS = 30
-            thin_window_min = (
-                settings.tg_dispatch_thin_idle_minutes if is_dispatch
-                else settings.tg_thin_idle_minutes
-            )
+            thin_window_min = settings.tg_dispatch_thin_idle_minutes
            very_recent = [
                inc for inc in tg_recent
                if _idle_gate_minutes(inc, now) * 60 <= THIN_CONVERSATIONAL_SECS
@@ -1049,8 +1031,7 @@ def _run_decision(ctx: dict) -> dict:
                if len(thin_pool) > 1:
                    logger.info(
                        f"Correlator fast-path thin (tier-2): {len(thin_pool)} active incidents "
-                        f"on {'dispatch' if is_dispatch else 'tactical'} channel — "
-                        f"ambiguous, skipping thin call {call_id}"
+                        f"— ambiguous, skipping thin call {call_id}"
                    )
                    thin_pool = []

@@ -1080,14 +1061,14 @@ def _run_decision(ctx: dict) -> dict:
            candidate = tg_recent[0]
            logger.info(
                f"Correlator fast/single: call {call_id} vs incident {candidate['incident_id']} "
-                f"tg_name={effective_talkgroup_name!r} is_dispatch={is_dispatch} "
+                f"tg_name={effective_talkgroup_name!r} "
                f"idle={round(_incident_idle_minutes(candidate, now), 1)}min "
                f"call_units={call_units} inc_units={candidate.get('units')} "
                f"call_coords={'yes' if coords else 'no'} inc_coords={'yes' if candidate.get('location_coords') else 'no'}"
            )
            fit, fit_signal = _call_fits_incident(
                candidate, call_units, call_vehicles, coords,
-                settings.location_proximity_km, is_dispatch=is_dispatch,
+                settings.location_proximity_km,
                call_embedding=call_embedding, now=now,
                reassignment=reassignment,
            )
@@ -1097,13 +1078,12 @@ def _run_decision(ctx: dict) -> dict:
                    "corr_path": "fast/single",
                    "corr_incident_idle_min": round(_incident_idle_minutes(candidate, now), 1),
                    "corr_fit_signal": fit_signal,
-                    "corr_is_dispatch": is_dispatch,
                }
                if fit_signal == "unit_overlap" and call_units:
                    corr_debug["corr_matched_units"] = _matching_units(call_units, candidate.get("units"))
                logger.info(
                    f"Correlator fast-path: call {call_id} → {candidate['incident_id']} "
-                    f"(signal={fit_signal}, is_dispatch={is_dispatch})"
+                    f"(signal={fit_signal})"
                )
            else:
                logger.info(
@@ -1119,14 +1099,14 @@ def _run_decision(ctx: dict) -> dict:
            # dispatch channel should create its own incident, not be force-merged.
            logger.info(
                f"Correlator fast/disambig: call {call_id} vs incident {candidate['incident_id']} "
-                f"tg_name={effective_talkgroup_name!r} is_dispatch={is_dispatch} "
+                f"tg_name={effective_talkgroup_name!r} "
                f"idle={round(_incident_idle_minutes(candidate, now), 1)}min "
                f"call_units={call_units} inc_units={candidate.get('units')} "
                f"call_coords={'yes' if coords else 'no'} inc_coords={'yes' if candidate.get('location_coords') else 'no'}"
            )
            fit, fit_signal = _call_fits_incident(
                candidate, call_units, call_vehicles, coords,
-                settings.location_proximity_km, is_dispatch=is_dispatch,
+                settings.location_proximity_km,
                call_embedding=call_embedding, now=now,
                reassignment=reassignment,
            )
@@ -1137,7 +1117,6 @@ def _run_decision(ctx: dict) -> dict:
                    "corr_incident_idle_min": round(_incident_idle_minutes(candidate, now), 1),
                    "corr_candidates": len(tg_recent),
                    "corr_fit_signal": fit_signal,
-                    "corr_is_dispatch": is_dispatch,
                }
                if fit_signal == "unit_overlap" and call_units:
                    corr_debug["corr_matched_units"] = _matching_units(call_units, candidate.get("units"))
@@ -1473,15 +1452,32 @@ async def _apply_and_log(decision: dict, ctx: dict) -> Optional[str]:
    equivalent to reading the flat fields today.
    """
    incident_id = await _apply_decision(decision, ctx)
+    if ctx.get("reassignment"):
+        await _release_reassigned_units(ctx, incident_id)
    corr_debug = decision.get("corr_debug") or {}
    if corr_debug:
        scene_index = ctx.get("scene_index", 0)
        updates = dict(corr_debug)
        updates["scenes"] = {
            str(scene_index): {
-                "transcript":   ctx.get("scene_transcript"),
-                "incident_id":  incident_id,
-                "corr_debug":   corr_debug,
+                "transcript":    ctx.get("scene_transcript"),
+                "incident_id":   incident_id,
+                "corr_debug":    corr_debug,
+                # server-26#139: this scene's OWN extracted incident_type/
+                # severity, as read by _call_is_substanceless's ctx at
+                # decision time — not the call doc's flat top-level field,
+                # which is last-scene-wins (server-26#96) and was the reason
+                # #138's "type" veto couldn't be told apart from cross-scene
+                # contamination without re-guessing from a live dump.
+                # NOTE: unlike incident_type, call_severity is already
+                # coerced to "routine" when extraction emitted nothing
+                # (ctx build: `severity or "routine"`) — a scene reading
+                # "routine" here doesn't distinguish "extraction said
+                # routine" from "extraction said nothing". Don't split a
+                # severity veto the way #138 splits the type veto without
+                # accounting for that.
+                "incident_type": ctx.get("incident_type"),
+                "severity":      ctx.get("call_severity"),
            }
        }
        try:
@@ -1716,7 +1712,6 @@ def _call_fits_incident(
    call_vehicles: list[str],
    call_coords: Optional[dict],
    proximity_km: float,
-    is_dispatch: bool = False,
    call_embedding: Optional[list] = None,
    now: Optional[datetime] = None,
    reassignment: bool = False,
@@ -1726,48 +1721,24 @@ def _call_fits_incident(
    the incident; signal names the specific evidence that drove the decision.

    fits=True  signals: "unit_overlap" | "vehicle_overlap" | "location_proximity"
-                        | "time_fallback" | "tactical_default"
    fits=False signals: "unit_loc_conflict" | "content_divergence"
-                        | "location_conflict" | "no_signal" | "tactical_idle"
+                        | "location_conflict" | "no_signal"

-    Original docstring (logic unchanged):
-    Return True if this call plausibly belongs to the given incident.
-
-    Evaluation order for dispatch channels (is_dispatch=True):
-    ─────────────────────────────────────────────────────────
-    1. Unit overlap
-       Same officer = same call.  On dispatch channels, also run a location
-       conflict guard: if both sides carry geocoded coords and they differ
-       significantly, the officer has moved to a new scene and the unit match
-       is a false positive.
-       When the call has NO geocoded coordinates AND the incident is old
-       (≥ 15 min), use content divergence as a location proxy: an officer at
-       a genuinely different scene will be talking about clearly different
-       things.  For recent incidents (< 15 min) we skip this proxy — the
-       officer may simply be giving an update without mentioning the address.
-
-    2. Vehicle overlap → True
-
-    3. Location proximity
-       Both sides geocoded and close → True.
-       Both sides geocoded and far apart (no other positive signal) → False.
-
-    4. No positive signals fired → dispatch fallback
-       a. Conversational continuity: idle < 2 min → True.
-          A call arriving within 2 minutes of the last incident activity almost
-          certainly belongs to the same dispatch thread.  "Baker, head over
-          there too" or "copy that" carries no incident-specific vocabulary but
-          is unambiguously a response to what was just said.  We do not require
-          embedding similarity here — embeddings capture word meaning, not
-          conversational context, and short operational messages will always
-          have lower similarity than the incident's accumulated content.
-       b. Older incident, no positive signals → False.
-          A shared dispatch channel must not absorb calls by default.
-
-    Tactical / working channel (is_dispatch=False):
-    ────────────────────────────────────────────────
-    Channel is dedicated to one scene.  No evidence of separation ≈ same call.
-    Default → True.
+    Evaluation order:
+    1. Unit overlap. Same officer = same call. Also runs a location-conflict
+       guard: geocoded on both sides and clearly different → the officer has
+       moved to a new scene, false positive. No geocode on the call (or on
+       the incident, with a call geocode) AND the incident is old (≥ 15 min)
+       → content divergence as a location proxy (embedding similarity < 0.82
+       → different scene). Skipped for recent incidents (< 15 min) — an
+       update without re-stating the address is normal.
+    2. Vehicle overlap → True.
+    3. Location proximity. Both geocoded and close → True; far apart with no
+       other positive signal → False.
+    4. No positive signal at all → False. A shared channel must not absorb
+       calls by default (server-26#134 — this used to default True within
+       20 min on any channel not name-matched as "dispatch"; a talkgroup
+       named tac/tactical is no less scrutinized in practice than any other).

    Thin calls (no units/vehicles/coords) never reach this function —
    they are intercepted before it in correlate_call.
@@ -1776,7 +1747,7 @@ def _call_fits_incident(
    # signed value: the re-correlation sweep anchors `now` to the call's own
    # started_at, which can be earlier than the incident's last activity and
    # send the signed value negative — silently defeating every `idle_min`
-    # gate below (content-divergence veto and the tactical default alike).
+    # gate below, the content-divergence veto included.
    # See `_idle_gate_minutes` docstring. The signed value is reported to
    # callers separately as `corr_incident_idle_min` (they compute it via
    # `_incident_idle_minutes` themselves) — nothing here needs it.
@@ -1787,44 +1758,43 @@ def _call_fits_incident(
    inc_units = _unit_keys(inc.get("units"))
    matched_units = _matching_units(call_units, inc.get("units"))
    if matched_units:
-        if is_dispatch:
-            if call_coords:
-                # Hard location conflict: geocoded on both sides and clearly different.
-                inc_coords_u = inc.get("location_coords")
-                if inc_coords_u:
-                    dist_km = _haversine_km(
-                        call_coords["lat"], call_coords["lng"],
-                        inc_coords_u["lat"], inc_coords_u["lng"],
-                    )
-                    if dist_km > proximity_km:
-                        logger.info(f"  fits[{inc_id}]: unit_overlap({matched_units}) but location_conflict dist={dist_km:.2f}km → unit_loc_conflict")
-                        return False, "unit_loc_conflict"
-                elif call_embedding and idle_min >= 15:
-                    # Call has geocode but incident doesn't — fall back to content
-                    # divergence as a location proxy.  Without this, stale incidents
-                    # that never geocoded absorb unrelated calls purely on unit
-                    # overlap (e.g. a patrol officer working a second scene 70 min
-                    # after the original call).
-                    inc_emb_u = inc.get("embedding")
-                    if inc_emb_u:
-                        sim = _cosine_similarity(call_embedding, inc_emb_u)
-                        if sim < 0.82:
-                            logger.info(f"  fits[{inc_id}]: unit_overlap({matched_units}) but content_divergence (has_call_coords/no_inc_coords) sim={sim:.3f} → content_divergence")
-                            return False, "content_divergence"
+        if call_coords:
+            # Hard location conflict: geocoded on both sides and clearly different.
+            inc_coords_u = inc.get("location_coords")
+            if inc_coords_u:
+                dist_km = _haversine_km(
+                    call_coords["lat"], call_coords["lng"],
+                    inc_coords_u["lat"], inc_coords_u["lng"],
+                )
+                if dist_km > proximity_km:
+                    logger.info(f"  fits[{inc_id}]: unit_overlap({matched_units}) but location_conflict dist={dist_km:.2f}km → unit_loc_conflict")
+                    return False, "unit_loc_conflict"
            elif call_embedding and idle_min >= 15:
-                # No geocode available AND old incident: use content divergence as a
-                # location-proxy veto.  After 15+ minutes an officer at a completely
-                # different scene will be discussing clearly different content.
-                # Skip this for recent incidents — an officer updating on the same
-                # scene without re-stating the address is normal and their update
-                # won't share much vocabulary with the original dispatch.
+                # Call has geocode but incident doesn't — fall back to content
+                # divergence as a location proxy.  Without this, stale incidents
+                # that never geocoded absorb unrelated calls purely on unit
+                # overlap (e.g. a patrol officer working a second scene 70 min
+                # after the original call).
                inc_emb_u = inc.get("embedding")
                if inc_emb_u:
                    sim = _cosine_similarity(call_embedding, inc_emb_u)
                    if sim < 0.82:
-                        logger.info(f"  fits[{inc_id}]: unit_overlap({matched_units}) but content_divergence sim={sim:.3f} → content_divergence")
+                        logger.info(f"  fits[{inc_id}]: unit_overlap({matched_units}) but content_divergence (has_call_coords/no_inc_coords) sim={sim:.3f} → content_divergence")
                        return False, "content_divergence"
-        logger.info(f"  fits[{inc_id}]: unit_overlap matched={matched_units} is_dispatch={is_dispatch} → unit_overlap")
+        elif call_embedding and idle_min >= 15:
+            # No geocode available AND old incident: use content divergence as a
+            # location-proxy veto.  After 15+ minutes an officer at a completely
+            # different scene will be discussing clearly different content.
+            # Skip this for recent incidents — an officer updating on the same
+            # scene without re-stating the address is normal and their update
+            # won't share much vocabulary with the original dispatch.
+            inc_emb_u = inc.get("embedding")
+            if inc_emb_u:
+                sim = _cosine_similarity(call_embedding, inc_emb_u)
+                if sim < 0.82:
+                    logger.info(f"  fits[{inc_id}]: unit_overlap({matched_units}) but content_divergence sim={sim:.3f} → content_divergence")
+                    return False, "content_divergence"
+        logger.info(f"  fits[{inc_id}]: unit_overlap matched={matched_units} → unit_overlap")
        return True, "unit_overlap"

    # ── 2. Vehicle overlap ────────────────────────────────────────────────────
@@ -1878,29 +1848,84 @@ def _call_fits_incident(
            return False, "location_conflict"

    # ── 4. No positive signals ────────────────────────────────────────────────
+    # Requires at least one positive signal (unit, vehicle, or location match).
+    # A substantive call with no matching signals is more likely a separate
+    # incident than a follow-up — two dispatches can arrive within seconds of
+    # each other on a busy channel. Content-free thin calls are handled before
+    # this function via the thin path in correlate_call, with a tighter
+    # 30-second recency window.
    logger.info(
-        f"  fits[{inc_id}]: no positive signal — is_dispatch={is_dispatch} idle={idle_min:.1f}min "
+        f"  fits[{inc_id}]: no positive signal — idle={idle_min:.1f}min "
        f"inc_units={list(inc_units)} call_units={call_units} "
        f"inc_vehicles={list(inc_vehicles)} call_vehicles={call_vehicles} "
        f"call_coords={call_coords is not None} inc_coords={inc_coords is not None}"
    )
-    if is_dispatch:
-        # Dispatch channels require at least one positive signal (unit, vehicle,
-        # or location match). A substantive call with no matching signals is more
-        # likely a separate incident than a follow-up to the current one — two
-        # dispatches can arrive within seconds of each other on a busy channel.
-        # Content-free thin calls are handled before this function via the thin
-        # path in correlate_call, with a tighter 30-second recency window.
-        return False, "no_signal"
+    return False, "no_signal"

-    # Tactical channel: one scene per channel.
-    # Within 20 min of the last incident activity, link by default — same
-    # working channel almost certainly means same scene.
-    # After 20 min of silence, require at least one positive signal; the same
-    # frequency can be reused for a new unrelated incident later in the shift.
-    if idle_min < 20.0:
-        return True, "tactical_default"
-    return False, "tactical_idle"
+
+def _apply_unit_clearance(inc: dict, cleared: list[str]) -> tuple[list[str], list[str], bool]:
+    """
+    Merge `cleared` into inc's units_active/units_cleared. Shared by
+    _update_incident (explicit 10-8/back-in-service extraction) and
+    _release_reassigned_units (server-26#<pending> pattern B: a unit accepting
+    a new dispatch, reassignment=True, is real-world evidence they're off
+    their prior call even without an explicit clearance phrase).
+
+    Returns (units_active, units_cleared, auto_resolved) — auto_resolved is
+    True when every tracked unit has now cleared, matching the resolve gate
+    at the bottom of _update_incident.
+    """
+    units_active  = list(inc.get("units_active")  or [])
+    units_cleared = list(inc.get("units_cleared") or [])
+    for u in cleared:
+        if u in units_active:
+            units_active.remove(u)
+        if u not in units_cleared:
+            units_cleared.append(u)
+    auto_resolved = bool(units_cleared) and not units_active
+    return units_active, units_cleared, auto_resolved
+
+
+async def _release_reassigned_units(ctx: dict, exclude_incident_id: Optional[str]) -> None:
+    """
+    server-26#<pending>: reassignment=True means a unit is accepting a NEW
+    dispatch — real-world evidence they're off whatever they were on before,
+    even when they never say an explicit 10-8/clear phrase (dispatch: "are
+    you able to clear and take a run at X" / unit: "10-4" carries no
+    self-reported clearance language intelligence.py's cleared_units
+    extraction looks for). Without this, that unit's prior incident is only
+    ever closed by the 90-minute idle sweep, not a real clear.
+
+    Scoped to OTHER active incidents (exclude_incident_id keeps this call's
+    own outcome untouched) with unit overlap in units_active — mirrors the
+    unit-continuity candidate scan at :1142 but releases instead of links.
+    """
+    call_units = ctx.get("call_units")
+    if not call_units:
+        return
+    system_id = ctx.get("system_id")
+    now       = ctx["now"]
+    unit_set  = _unit_keys(call_units)
+    for inc in ctx.get("all_active") or []:
+        if inc.get("incident_id") == exclude_incident_id:
+            continue
+        if system_id and system_id not in (inc.get("system_ids") or []):
+            continue
+        matched = [u for u in (inc.get("units_active") or []) if _normalize_unit(u) in unit_set]
+        if not matched:
+            continue
+        units_active, units_cleared, auto_resolved = _apply_unit_clearance(inc, matched)
+        updates = {"units_active": units_active, "units_cleared": units_cleared}
+        if auto_resolved:
+            updates["status"] = "resolved"
+            updates["resolved_at"] = now.isoformat()
+        await fstore.doc_set("incidents", inc["incident_id"], updates)
+        logger.info(
+            f"Correlator: reassignment released unit(s) {matched} from incident "
+            f"{inc['incident_id']}" + (" (auto-resolved)" if auto_resolved else "")
+        )
+        if auto_resolved:
+            await maybe_resolve_parent(inc["incident_id"])


 async def _update_incident(
@@ -1946,11 +1971,8 @@ async def _update_incident(
    for u in call_units:
        if u not in units_cleared and u not in units_active:
            units_active.append(u)
-    for u in (cleared_units or []):
-        if u in units_active:
-            units_active.remove(u)
-        if u not in units_cleared:
-            units_cleared.append(u)
+    inc_with_active_update = {**inc, "units_active": units_active, "units_cleared": units_cleared}
+    units_active, units_cleared, _ = _apply_unit_clearance(inc_with_active_update, cleared_units or [])

    # The incident's label and its pin are resolved together, as one value.
    location          = clean_location(location)
@@ -64,9 +64,9 @@ Response format — a JSON object with a "scenes" array. Each scene:
 Rules:
 - location: prefer intersections > addresses > mile markers > route+town > route alone > town alone. Dispatch-provided addresses take priority over unit-reported positions. Empty string if none.
 - tags: describe WHAT happened, not WHERE. Specific, lowercase, hyphenated. Do not use location names, road names, talkgroup names, or place names as tags (wrong: "lower-macy's", "canvas-route-6", "route-202"; right: "suspect-search", "shoplifting", "vehicle-pursuit"). Do not repeat incident_type as a tag.
- units: ONLY identifiers that appear verbatim in the transcript. Use speaker role inference to distinguish units being dispatched from units acknowledging — both should be included. Never infer or guess unit IDs not present in the text.
+- units: ONLY identifiers that appear verbatim in the transcript. Use speaker role inference to distinguish units being dispatched from units acknowledging — both should be included. Never infer or guess unit IDs not present in the text. If a unit ID format is given below, use it to recognise a unit spoken in a shortened or partial form (e.g. just the phonetic name alone) as the same unit — but still only extract what is actually said, never fabricate the full form.
 - Do not invent details not present in the transcript.
- incident_type: let the talkgroup channel be your primary signal. Use "fire" ONLY if the talkgroup is clearly a fire/rescue channel OR the transcript explicitly describes active fire, smoke, flames, or structure fire activation. Police or EMS referencing a fire scene → use "police" or "ems". When the channel is a police channel and nothing in the transcript contradicts it, return "police" — do NOT fall back to "other" merely because the transmission is administrative. Reserve "other" for traffic that genuinely belongs to no emergency service (rail operations, public works, utility coordination). Reserve "unknown" for transcripts too garbled to place at all.
+- incident_type: FIRST decide whether this transmission has any incident behind it at all, using the same bar as the "routine" severity rule below — pure administrative/status traffic with nothing describable happening: post/unit check-ins, roll call, bare acknowledgements ("10-4", "copy", "received"), records/report exchanges, "show me admin"/"show me available", a status ten-code with no event attached. If it is administrative/status-only, return "unknown" — this applies on EVERY channel, including a police channel; do not let the channel default override it (server-26#138: forcing a channel default onto content-free chatter is what let radio housekeeping open incidents). Only once real event content is present, let the talkgroup channel be your primary signal for WHICH type. Use "fire" ONLY if the talkgroup is clearly a fire/rescue channel OR the transcript explicitly describes active fire, smoke, flames, or structure fire activation. Police or EMS referencing a fire scene → use "police" or "ems". When the channel is a police channel, a real event is present, and nothing in the transcript contradicts it, return "police". Reserve "other" for a real event that genuinely belongs to no emergency service (rail operations, public works, utility coordination) — not for administrative chatter, which is "unknown" per above regardless of channel. Also reserve "unknown" for transcripts too garbled to place at all.
 - severity: ALWAYS return one of the four values. Judge the underlying event, not how dramatic the words sound.
    "routine"  — administrative/status traffic with no incident behind it: mileage and transport logging, radio checks, acknowledgements, shift changes, track block/power requests, records lookups.
    "minor"    — a real but low-stakes call: lift assist, parking complaint, past-tense larceny report, noise complaint, welfare check.
@@ -74,18 +74,15 @@ Rules:
    "major"    — life safety or major property loss: structure fire, vehicle pursuit, shots fired, entrapment, cardiac arrest, officer needing assistance.
 - ten_codes: interpret radio codes using the department reference provided below. Do not guess codes not listed.
 - resolved: true only when the scene explicitly signals "Code 4", "all clear", "10-42", "in custody", "patient transported", "fire out", "GOA", "negative contact", "scene clear".
- cleared_units: only include units that explicitly stated their own back-in-service status in this recording (e.g. "Unit 7, 10-8", "Baker-1 available", "E-14 back in service", or the department ten-code for available/back-in-service listed above). Silence or absence of a unit is NOT clearance. A scene-wide Code 4 belongs in resolved=true, not here — cleared_units is for individual unit availability signals only.
+- cleared_units: include a unit whose back-in-service/available status is stated in this recording — either the unit self-reporting (e.g. "Unit 7, 10-8", "Baker-1 available", "E-14 back in service", or the department ten-code for available/back-in-service listed above) OR dispatch confirming that SPECIFIC unit's status back to them (e.g. the unit asks "how do you show me" and dispatch replies "showing you available" / "in service"). The unit ID must be identifiable either way — a bare "clear" or "10-8" with no unit attached to it is NOT clearance; do not guess which unit said it. Silence or absence of a unit is NOT clearance. A scene-wide Code 4 belongs in resolved=true, not here — cleared_units is for individual unit availability signals only.
 - reassignment: only true when a unit is explicitly being pulled to a completely new call or location. A unit going en route to their first dispatch is NOT a reassignment. Routine status updates, acknowledgements, and scene updates are NOT reassignments.

 System: {system_id}
 Talkgroup: {talkgroup_name}
-{ten_codes_block}{vocabulary_block}{transcript_block}"""
+{ten_codes_block}{vocabulary_block}{unit_format_block}{transcript_block}"""

-# The incident_type enum offered to the model in EXTRACTION_PROMPT.  Kept here
-# rather than only in the prompt so a model that invents a value cannot write it
-# into incident.type.  "unknown" is deliberately absent — it is a real answer
-# from the model but not a usable type, and is normalised to None alongside
-# anything unrecognised.
+# "unknown" is deliberately absent — normalises to None, which is what lets
+# the creation gate veto a content-free call (server-26#138).
 _VALID_INCIDENT_TYPES = frozenset({"fire", "ems", "police", "accident", "other"})

 # Geographic bias radius for geocoding — half-width in degrees (~55 km)
@@ -156,6 +153,23 @@ def _build_ten_codes_block(ten_codes: dict[str, str]) -> str:
    return f"Department ten-codes:\n{lines}\n\n"


+def _build_unit_format_block(unit_format_hint: Optional[str]) -> str:
+    """
+    server-26#<pending> — unit ID formats vary per department (e.g. Yorktown:
+    "<district>-<phonetic>", "5-David", sometimes spoken as bare "David";
+    County: "<location>-<number>", "SAM-1", "airport-3", "parks-4") with no
+    shared pattern across systems. Without a per-system hint, the model has
+    no way to recognise a unit ID it hasn't seen phrased that way before, and
+    that failure compounds into cleared_units and reassignment detection,
+    both of which depend on first recognising which token IS the unit.
+    Owner-authored free text per system (systems/{id}.unit_format_hint via
+    PUT /systems/{id}/unit-format) — no auto-induction yet.
+    """
+    if not unit_format_hint:
+        return ""
+    return f"This system's unit ID format: {unit_format_hint}\n\n"
+
+
 async def extract_scenes(
    call_id: str,
    transcript: str,
@@ -182,12 +196,15 @@ async def extract_scenes(
    """
    vocabulary: list[str] = []
    ten_codes: dict[str, str] = {}
+    unit_format_hint: str = ""
    if system_id:
-        # Single cached read — vocabulary and ten_codes live on the same document.
+        # Single cached read — vocabulary, ten_codes and unit_format_hint all
+        # live on the same document.
        system_doc = await fstore.doc_get_cached("systems", system_id)
        if system_doc:
-            vocabulary = system_doc.get("vocabulary") or []
-            ten_codes  = system_doc.get("ten_codes") or {}
+            vocabulary       = system_doc.get("vocabulary") or []
+            ten_codes        = system_doc.get("ten_codes") or {}
+            unit_format_hint = system_doc.get("unit_format_hint") or ""

    if _is_garbage_transcript(transcript):
        logger.warning(
@@ -254,6 +271,7 @@ async def extract_scenes(
    raw_scenes: list[dict] = await asyncio.to_thread(
        _sync_extract,
        transcript, talkgroup_name, talkgroup_id, system_id, segments, vocabulary, ten_codes,
+        unit_format_hint,
    )

    if not raw_scenes:
@@ -677,6 +695,7 @@ def _sync_extract(
    segments: Optional[list[dict]],
    vocabulary: Optional[list[str]] = None,
    ten_codes: Optional[dict[str, str]] = None,
+    unit_format_hint: Optional[str] = None,
 ) -> list[dict]:
    """Call GPT-4o-mini and return a list of scene dicts."""
    from app.config import settings
@@ -694,6 +713,7 @@ def _sync_extract(
        system_id=system_id or "unknown",
        ten_codes_block=_build_ten_codes_block(ten_codes or {}),
        vocabulary_block=build_gpt_vocab_block(vocabulary or []),
+        unit_format_block=_build_unit_format_block(unit_format_hint),
    )

    try:
@@ -111,6 +111,8 @@ class MQTTHandler:
                "assigned_system_id": None,
                "approval_status": "pending",
                "node_type": payload.get("node_type", "fixed"),
+                "secondary_sdr_mode": payload.get("secondary_sdr_mode", "none"),
+                "sdr_count": payload.get("sdr_count", 1),
                "enforce_override_timeout": payload.get("enforce_override_timeout", True),
                "is_overridden": False,
                "override_system_id": None,
@@ -141,6 +143,11 @@ class MQTTHandler:
            updates["node_type"] = node_type
            updates["enforce_override_timeout"] = enforce_timeout

+            if "secondary_sdr_mode" in payload:
+                updates["secondary_sdr_mode"] = payload["secondary_sdr_mode"]
+            if "sdr_count" in payload:
+                updates["sdr_count"] = payload["sdr_count"]
+
            if node_type == "portable":
                updates["is_overridden"] = False
                updates["override_system_id"] = None
@@ -20,6 +20,30 @@ from app.internal.logger import logger
 from app.internal import firestore as fstore
 from app.config import settings

+# server-26#131: minimum time since the real-time pipeline (routers/upload.py
+# _run_intelligence_pipeline) marked intelligence_started_at before the sweep
+# will touch a call, even though it already looks orphaned. STT + scene
+# extraction + correlation is a multi-second-to-low-minutes chain (Whisper,
+# then a Gemini call per scene); without this buffer the sweep could pick up
+# a call mid-pipeline — no incident_id/corr_path written yet — and correlate
+# it a second time, independently, sometimes onto a different incident than
+# the real-time path lands on. That race is what #131 found (same call in
+# two incidents' call_ids, ~2% of linked calls). A call with no
+# intelligence_started_at at all (pre-#131 call doc, or the marker write
+# itself failed) is NOT held back by this — absence isn't evidence of an
+# in-flight pipeline, and #131's own bug predates this field existing.
+#
+# 15, not 5: neither OpenAI's Whisper client nor Gemini's call in
+# llm_correlator.py sets a request timeout (server-26#153), so a hung call can
+# run well past a few minutes on SDK-default retries, and this constant is a
+# guess against that unbounded tail, not a measured bound. Raising it costs
+# nothing on the recovery side: a call that finished processing (linked OR
+# genuinely orphaned) always has corr_path set (_apply_and_log writes it even
+# on the orphan action), so it's already excluded by the
+# `not c.get("corr_path")` filter below and never reaches this check at all —
+# this constant only ever delays calls that are still actually running.
+MIN_MINUTES_SINCE_PIPELINE_START = 15
+
 # Standard link-only retry budget before a call is tombstoned corr_path="unlinked".
 MAX_SWEEP_ATTEMPTS = 3
 # server-26#115 — a call the consensus LLM-orphan gate parked (llm=orphan vs
@@ -52,8 +76,22 @@ async def recorrelation_loop() -> None:
            logger.error(f"Re-correlation sweep failed: {e}")


+def _pipeline_likely_still_running(call: dict, now: datetime) -> bool:
+    """server-26#131 — True when the real-time pipeline marked
+    intelligence_started_at recently enough that it's probably still mid-flight
+    (STT / scene extraction / correlation), so the sweep should not race it.
+    No marker at all (older call doc, or the marker write itself failed)
+    returns False — absence isn't evidence of an in-flight pipeline."""
+    started = _parse_dt(call.get("intelligence_started_at"))
+    if not started:
+        return False
+    age_minutes = (now - started).total_seconds() / 60
+    return age_minutes < MIN_MINUTES_SINCE_PIPELINE_START
+
+
 async def _run_sweep_pass() -> None:
-    cutoff = datetime.now(timezone.utc) - timedelta(minutes=settings.recorrelation_scan_minutes)
+    now = datetime.now(timezone.utc)
+    cutoff = now - timedelta(minutes=settings.recorrelation_scan_minutes)

    # Server-side range query: only calls that ended within the scan window.
    # Filter incident_id=null client-side (Firestore can't query for missing fields).
@@ -77,6 +115,7 @@ async def _run_sweep_pass() -> None:
        # a second route into the over-merge the thin fix above addresses.
        and not c.get("skip_reason")
        and c.get("corr_sweep_count", 0) < _max_sweep_attempts(c)
+        and not _pipeline_likely_still_running(c, now)
    ]

    if not orphans:
@@ -17,7 +17,7 @@ from app.internal.auth import (
    require_node_service_or_firebase_token,
 )
 from app.routers import nodes, systems, calls, upload, tokens, incidents, alerts, admin, trips, places, links, users
-from app.routers import enrollment, media, org, waitlist
+from app.routers import enrollment, media, org, waitlist, telemetry
 from app.internal import dynsec
 from app.internal import firestore as fstore

@@ -120,6 +120,7 @@ app.include_router(nodes.router,     dependencies=[Depends(require_service_or_fi
 # write routes inside carry their own require_admin_token, so nodes get read
 # access only.
 app.include_router(systems.router,   dependencies=[Depends(require_node_service_or_firebase_token)])
+app.include_router(telemetry.router, dependencies=[Depends(require_node_service_or_firebase_token)])
 app.include_router(calls.router,     dependencies=[Depends(require_service_or_firebase_token)])
 app.include_router(tokens.router,    dependencies=[Depends(require_service_or_firebase_token)])
 app.include_router(incidents.router, dependencies=[Depends(require_service_or_firebase_token)])
@@ -62,12 +62,43 @@ class NodeRecord(BaseModel):
    last_seen: Optional[datetime] = None
    assigned_system_id: Optional[str] = None
    node_type: str = "fixed"  # fixed or portable
+    secondary_sdr_mode: str = "none"  # none | adsb | ais | op25_2 — requires a second physical SDR
+    sdr_count: int = 1  # self-reported by the node's checkin, best-effort
    enforce_override_timeout: bool = True
    is_overridden: bool = False
    override_system_id: Optional[str] = None
    override_timeout_at: Optional[datetime] = None


+class AircraftTrack(BaseModel):
+    """Live ADS-B position, one doc per icao. Overwritten on every sighting —
+    this is a live-map snapshot, not a history (see node-26#9)."""
+    icao: str
+    org_id: Optional[str] = None
+    node_id: str
+    callsign: Optional[str] = None
+    lat: Optional[float] = None
+    lon: Optional[float] = None
+    altitude_ft: Optional[float] = None
+    ground_speed_kt: Optional[float] = None
+    track_deg: Optional[float] = None
+    last_seen: datetime
+
+
+class VesselTrack(BaseModel):
+    """Live AIS position, one doc per mmsi. Same live-snapshot shape as
+    AircraftTrack — overwritten on every sighting (see node-26#9)."""
+    mmsi: str
+    org_id: Optional[str] = None
+    node_id: str
+    name: Optional[str] = None
+    lat: Optional[float] = None
+    lon: Optional[float] = None
+    speed_kt: Optional[float] = None
+    heading_deg: Optional[float] = None
+    last_seen: datetime
+
+
 class CommandPayload(BaseModel):
    action: str                      # discord_join / discord_leave / op25_restart
    guild_id: Optional[str] = None
@@ -109,6 +109,11 @@ async def debug_correlation(
            "scene_index":            scene_index,
            "transcript":             scene.get("transcript"),
            "incident_id":            scene.get("incident_id"),
+            # server-26#139: this scene's OWN incident_type/severity, as seen
+            # by _call_is_substanceless at decision time — not the call doc's
+            # flat top-level field, which is last-scene-wins (server-26#96).
+            "incident_type":          scene.get("incident_type"),
+            "severity":               scene.get("severity"),
            "corr_path":              corr_debug.get("corr_path"),
            "corr_incident_idle_min": corr_debug.get("corr_incident_idle_min"),
            "corr_distance_km":       corr_debug.get("corr_distance_km"),
@@ -195,6 +195,7 @@ async def assign_system(
 class NodeUpdateBody(BaseModel):
    node_type: Optional[str] = None
    enforce_override_timeout: Optional[bool] = None
+    secondary_sdr_mode: Optional[str] = None  # none | adsb | ais | op25_2


@router.patch("/{node_id}")
@@ -227,6 +228,8 @@ async def update_node(
            }
            if updated_node.get("ppm_override") is not None:
                push_payload["ppm_override"] = updated_node["ppm_override"]
+            if updated_node.get("secondary_sdr_mode") is not None:
+                push_payload["secondary_sdr_mode"] = updated_node["secondary_sdr_mode"]
            mqtt_handler.push_config(node_id, push_payload)

    return {"ok": True}
@@ -24,6 +24,10 @@ class TenCodesBody(BaseModel):
    ten_codes: Dict[str, str]


+class UnitFormatBody(BaseModel):
+    unit_format_hint: str
+
+
 class PendingTermBody(BaseModel):
    talkgroup_id: int
    term: str
@@ -155,6 +159,38 @@ async def update_ten_codes(
    return {"ok": True, "ten_codes": body.ten_codes}


+# ── Unit ID format hint ─────────────────────────────────────────────────────────
+
+@router.get("/{system_id}/unit-format")
+async def get_unit_format(system_id: str):
+    """Return the unit-ID format hint for a system."""
+    system = await fstore.doc_get("systems", system_id)
+    if not system:
+        raise HTTPException(404, f"System '{system_id}' not found.")
+    return {"unit_format_hint": system.get("unit_format_hint") or ""}
+
+
+@router.put("/{system_id}/unit-format")
+async def update_unit_format(
+    system_id: str,
+    body: UnitFormatBody,
+    _: dict = Depends(require_admin_token),
+):
+    """
+    Set the free-text unit-ID format hint fed into intelligence.py's
+    extraction prompt (server-26#<pending>). Departments have no shared unit
+    ID convention — e.g. "5-David"/bare "David" vs "SAM-1"/"airport-3" — and
+    the extraction prompt has no way to recognise a format it hasn't been
+    told about. Own route for the same reason ten-codes has one: not carried
+    by the systems form, so folding it into PUT /{id} would wipe it.
+    """
+    existing = await fstore.doc_get("systems", system_id)
+    if not existing:
+        raise HTTPException(404, f"System '{system_id}' not found.")
+    await fstore.doc_update("systems", system_id, {"unit_format_hint": body.unit_format_hint})
+    return {"ok": True, "unit_format_hint": body.unit_format_hint}
+
+
 # ── Area context ──────────────────────────────────────────────────────────────

@router.get("/{system_id}/area-context")
@@ -0,0 +1,125 @@
+from datetime import datetime, timezone
+from typing import List, Optional
+
+from fastapi import APIRouter, Depends, HTTPException
+from pydantic import BaseModel
+
+from app.internal import firestore as fstore
+from app.internal.auth import require_node_service_or_firebase_token
+from app.internal.logger import logger
+
+router = APIRouter(prefix="/telemetry", tags=["telemetry"])
+
+
+class AircraftReport(BaseModel):
+    icao: str
+    callsign: Optional[str] = None
+    lat: Optional[float] = None
+    lon: Optional[float] = None
+    altitude_ft: Optional[float] = None
+    ground_speed_kt: Optional[float] = None
+    track_deg: Optional[float] = None
+
+
+class AdsbUploadBody(BaseModel):
+    aircraft: List[AircraftReport]
+
+
+@router.post("/adsb")
+async def upload_adsb(
+    body: AdsbUploadBody,
+    decoded: dict = Depends(require_node_service_or_firebase_token),
+):
+    """
+    Node-initiated: a second-SDR ADS-B decoder (node-26#9) periodically posts
+    its current aircraft snapshot here. One doc per icao, last-seen-wins —
+    this is a live-map overlay, not a flight history.
+    """
+    node_id = decoded.get("node_id")
+    if not node_id:
+        raise HTTPException(400, "This endpoint requires node identity, not a service/admin token.")
+
+    node = await fstore.doc_get_cached("nodes", node_id)
+    org_id = node.get("org_id") if node else None
+    now = datetime.now(timezone.utc).isoformat()
+
+    writes = []
+    for ac in body.aircraft:
+        if not ac.icao:
+            continue
+        doc = {
+            "icao": ac.icao,
+            "node_id": node_id,
+            "callsign": ac.callsign,
+            "lat": ac.lat,
+            "lon": ac.lon,
+            "altitude_ft": ac.altitude_ft,
+            "ground_speed_kt": ac.ground_speed_kt,
+            "track_deg": ac.track_deg,
+            "last_seen": now,
+        }
+        if org_id:
+            doc["org_id"] = org_id
+        writes.append(("aircraft", ac.icao, doc))
+
+    for collection, doc_id, doc in writes:
+        try:
+            await fstore.doc_set(collection, doc_id, doc, merge=True)
+        except Exception as e:
+            logger.warning(f"Failed to upsert {collection}/{doc_id} from node {node_id}: {e}")
+
+    return {"ok": True, "count": len(writes)}
+
+
+class VesselReport(BaseModel):
+    mmsi: str
+    name: Optional[str] = None
+    lat: Optional[float] = None
+    lon: Optional[float] = None
+    speed_kt: Optional[float] = None
+    heading_deg: Optional[float] = None
+
+
+class AisUploadBody(BaseModel):
+    vessels: List[VesselReport]
+
+
+@router.post("/ais")
+async def upload_ais(
+    body: AisUploadBody,
+    decoded: dict = Depends(require_node_service_or_firebase_token),
+):
+    """Same shape as /telemetry/adsb, one doc per mmsi in `vessels`."""
+    node_id = decoded.get("node_id")
+    if not node_id:
+        raise HTTPException(400, "This endpoint requires node identity, not a service/admin token.")
+
+    node = await fstore.doc_get_cached("nodes", node_id)
+    org_id = node.get("org_id") if node else None
+    now = datetime.now(timezone.utc).isoformat()
+
+    writes = []
+    for v in body.vessels:
+        if not v.mmsi:
+            continue
+        doc = {
+            "mmsi": v.mmsi,
+            "node_id": node_id,
+            "name": v.name,
+            "lat": v.lat,
+            "lon": v.lon,
+            "speed_kt": v.speed_kt,
+            "heading_deg": v.heading_deg,
+            "last_seen": now,
+        }
+        if org_id:
+            doc["org_id"] = org_id
+        writes.append(("vessels", v.mmsi, doc))
+
+    for collection, doc_id, doc in writes:
+        try:
+            await fstore.doc_set(collection, doc_id, doc, merge=True)
+        except Exception as e:
+            logger.warning(f"Failed to upsert {collection}/{doc_id} from node {node_id}: {e}")
+
+    return {"ok": True, "count": len(writes)}
@@ -111,55 +111,17 @@ async def upload_call_audio(

 def _recent_incident_on_same_talkgroup(ctx: dict) -> bool:
    """
-    True when one of the already-loaded recent incidents is running on this
-    call's own system + talkgroup AND was active within the last
-    `settings.tg_dispatch_thin_idle_minutes` minutes. Covers the "unit
-    dispatched on the dispatch channel, thin acknowledgement 10-30s later"
-    case: the ack carries no substance of its own but plainly belongs to the
-    job just opened.
+    True when a recent incident is running on this call's own system +
+    talkgroup, within `settings.tg_dispatch_thin_idle_minutes` (5 min) —
+    applied uniformly regardless of the talkgroup's name (server-26#134).
+    Covers "unit dispatched, thin ack 10-30s later": the ack has no
+    substance of its own but plainly belongs to the job just opened.

-    Always uses `settings.tg_dispatch_thin_idle_minutes` (5 min), regardless
-    of what the talkgroup is named. An earlier version of this branched on
-    `_is_dispatch_channel` (mirroring incident_correlator.py's fast/thin idle-
-    window selection) to use a longer 15-minute window on anything not
-    literally named "dispatch"/"patched"/"primary" — owner correction,
-    2026-09-13, from direct scanning experience: a talkgroup named "tac"/
-    "tactical" genuinely does see materially different traffic only during a
-    real incident, and that's rare; the overwhelming majority of traffic on
-    ANY monitored channel — including high-risk stops and pursuits — runs on
-    the main channel regardless of what it's named. `_is_dispatch_channel`'s
-    string-match is a naming-convention guess, not a detector of actual
-    channel behavior, and trusting it here meant a busy single-channel
-    department not literally named "dispatch" would silently get the more
-    permissive window and reproduce #115's original bug. One constant,
-    applied uniformly, is the safer default; a real low-volume channel is the
-    rare case, not the norm, so erring toward the tighter window costs little.
-    NOT applied to incident_correlator.py's own fast/thin selection (out of
-    scope for this pass — a bigger, decision-changing surface, worth its own
-    review rather than changing under this fix).
-
-    This used to be a plain "does any recent incident exist on this
-    talkgroup" check against a 2-hour window (`correlation_window_hours`).
-    Measured live in production (server-26#115, CORRELATION_REVIEW_0912.md,
-    window #3): on a busy dispatch channel producing 3-13 incidents per 2h,
-    that condition is satisfied almost unconditionally, so the surrounding
-    LLM-orphan gate never fired on exactly the channels it exists to
-    protect (0/24 target-shaped calls gated in a 4h window). The docstring's
-    own intent was always "10-30 seconds", not "hours" — a few minutes is
-    the right shape.
-
-    Reads ctx["recent"] — the same window-filtered candidate list the rules
-    engine already loaded — so this adds no Firestore read.
-
-    Known limitation (server-26#115): ctx["recent"] is derived from
-    `all_active` in `_build_context` — incidents with `status=="active"`
-    for the call's org, with over-capacity incidents already dropped by
-    `_drop_capped` — not a full scan of the `incidents` collection. A
-    same-talkgroup incident that has already auto-resolved (no longer
-    "active") or hit `incident_max_calls`/`incident_max_duration_minutes`
-    will NOT appear here even though it is chronologically recent. A proper
-    fix needs a dedicated Firestore query that is not status/capacity
-    filtered — a new read, out of scope for this pass.
+    Reads ctx["recent"] (the rules engine's own candidate list — no extra
+    Firestore read). That list is status=="active" incidents only, so an
+    already-resolved or capacity-capped same-talkgroup incident won't be
+    seen here even if chronologically recent (server-26#115, unresolved —
+    would need a dedicated non-status-filtered query).

    Whether this limitation explains the 2/24 unexplained gate misses in the
    window #3 measurement is UNANSWERED, not confirmed either way — a prior
@@ -451,6 +413,25 @@ async def _run_intelligence_pipeline(
    """
    from app.internal import transcription, intelligence, incident_correlator, alerter, talkgroups

+    # server-26#131: mark that real-time processing has started for this call
+    # BEFORE any of the slow steps below (STT, scene extraction, correlation).
+    # The re-correlation sweep (internal/recorrelation_sweep.py) scans for
+    # calls that still look orphaned within a wide window (recorrelation_scan_
+    # minutes, default 60) — with no guard here, a call whose real-time
+    # pipeline is still mid-flight (still transcribing, still waiting on a
+    # Gemini call) has no incident_id/corr_path written yet, so the sweep's
+    # orphan filter can't tell "never processed" from "processing right now"
+    # and correlates it a second time, independently, sometimes landing on a
+    # different incident than the real-time path — the exact duplicate-link
+    # bug #131 found (same call in two incidents' call_ids, ~2% of linked
+    # calls). Best-effort: a write failure here must not abort the pipeline.
+    try:
+        await fstore.doc_set("calls", call_id, {
+            "intelligence_started_at": datetime.now(timezone.utc).isoformat()
+        })
+    except Exception as e:
+        logger.warning(f"Could not mark intelligence_started_at for call {call_id}: {e}")
+
    # The node only sends talkgroup_name when OP25 had it in the loaded tags
    # file, so it arrives empty for exactly the talkgroups C2 can name from the
    # system config. Resolve it once, here, at the single funnel both /upload
@@ -31,8 +31,10 @@ from app.internal.incident_correlator import (

 NOW = datetime(2026, 8, 20, 7, 0, 0, tzinfo=timezone.utc)

-# TG 383 from the dump: "Ch 1 (Patched with 155.310)". _DISPATCH_TG_RE matches
-# "patched", so this is a shared dispatch backbone carrying the whole department.
+# TG 383 from the dump: "Ch 1 (Patched with 155.310)", a shared dispatch
+# backbone carrying the whole department. Kept as two distinct fixture names
+# for readability even though the channel's name no longer affects behavior
+# (server-26#134).
 DISPATCH_TG = "Ch 1 (Patched with 155.310)"
 TACTICAL_TG = "Fireground 2"

@@ -140,12 +142,9 @@ def test_thin_call_with_no_overlap_does_not_attach_on_a_dispatch_channel():
    assert _run_decision(_ctx(all_active=[inc], recent=[inc]))["action"] == "orphan"


-def test_thin_call_with_no_overlap_does_not_attach_on_a_tactical_channel():
-    """
-    The widest version of the bug: non-dispatch talkgroups skipped the tiering
-    entirely and used the whole 90-minute fast-path window with no
-    single-candidate requirement, so ANY thin call joined whatever was newest.
-    """
+def test_thin_call_with_no_overlap_does_not_attach_on_a_tactical_named_channel():
+    """A channel's name no longer changes anything (server-26#134) — same
+    assertion as the dispatch-named case above, different fixture name."""
    inc = _incident(idle_minutes=40)
    decision = _run_decision(_ctx(
        all_active=[inc], recent=[inc], talkgroup_name=TACTICAL_TG,
@@ -153,9 +152,18 @@ def test_thin_call_with_no_overlap_does_not_attach_on_a_tactical_channel():
    assert decision["action"] == "orphan"


-def test_tactical_thin_call_still_attaches_inside_its_own_window():
-    """Bounded, not removed — a "10-4" on a working channel is still context."""
-    inc = _incident(idle_minutes=settings.tg_thin_idle_minutes - 1)
+def test_tactical_named_channel_uses_the_dispatch_window_now():
+    """server-26#134: 14 min was inside the old 15-min tactical window; now
+    every channel uses the 5-min window regardless of name."""
+    inc = _incident(idle_minutes=14)
+    decision = _run_decision(_ctx(
+        all_active=[inc], recent=[inc], talkgroup_name=TACTICAL_TG,
+    ))
+    assert decision["action"] == "orphan"
+
+
+def test_tactical_named_channel_still_attaches_inside_the_dispatch_window():
+    inc = _incident(idle_minutes=settings.tg_dispatch_thin_idle_minutes - 1)
    decision = _run_decision(_ctx(
        all_active=[inc], recent=[inc], talkgroup_name=TACTICAL_TG,
    ))
@@ -224,47 +232,27 @@ def test_back_dated_thin_call_does_not_sail_through_the_recency_gate():

 def test_back_dated_call_does_not_bypass_the_content_divergence_veto(monkeypatch):
    """
-    Same `9d376ffe` failure mode, but exercised directly against
-    `_call_fits_incident` on a dispatch channel: unit overlap plus a
-    back-dated call (incident updated 45 minutes AFTER the call's own
-    `started_at`, which the sweep passes as `now`) used to make the signed
-    idle -45, so `idle_min >= 15` read False and the content-divergence
-    veto never ran — unit overlap alone forced the merge regardless of
-    what the call was actually about. With the gate fixed to compare
-    distance, idle_min is 45 (>= 15), the veto runs, and a divergent
-    embedding (patched below so the assertion doesn't depend on numpy
-    being installed in this environment) fails it.
+    Same `9d376ffe` failure mode, exercised directly against
+    `_call_fits_incident`: unit overlap plus a back-dated call (incident
+    updated 45 minutes AFTER the call's own `started_at`, which the sweep
+    passes as `now`) used to make the signed idle -45, so `idle_min >= 15`
+    read False and the content-divergence veto never ran — unit overlap
+    alone forced the merge regardless of what the call was actually about.
+    With the gate fixed to compare distance, idle_min is 45 (>= 15), the
+    veto runs, and a divergent embedding (patched below so the assertion
+    doesn't depend on numpy being installed in this environment) fails it.
    """
    monkeypatch.setattr(correlator_mod, "_cosine_similarity", lambda a, b: 0.0)
    inc = _incident(idle_minutes=-45, units=["6-Adam"])
    inc["embedding"] = [1.0, 0.0]
    fits, signal = _call_fits_incident(
        inc, call_units=["6-Adam"], call_vehicles=[], call_coords=None,
-        proximity_km=settings.location_proximity_km, is_dispatch=True,
+        proximity_km=settings.location_proximity_km,
        call_embedding=[0.0, 1.0], now=NOW,
    )
    assert (fits, signal) == (False, "content_divergence")


-def test_back_dated_call_on_tactical_channel_does_not_get_the_default():
-    """
-    Tactical-channel counterpart: no unit/vehicle/location signal, so the
-    function falls through to step 4's `idle_min < 20.0` default. A
-    back-dated call (incident updated 45 minutes after the call's own
-    started_at) used to read idle_min as -45, which is always < 20.0, so
-    `tactical_default` fired unconditionally no matter how stale the
-    incident actually was relative to this call. Fixed, idle_min is the
-    45-minute distance, which is not < 20.0.
-    """
-    inc = _incident(idle_minutes=-45)
-    fits, signal = _call_fits_incident(
-        inc, call_units=[], call_vehicles=[], call_coords=None,
-        proximity_km=settings.location_proximity_km, is_dispatch=False,
-        call_embedding=None, now=NOW,
-    )
-    assert (fits, signal) == (False, "tactical_idle")
-
-
 # ---------------------------------------------------------------------------
 # 4. Hard caps — path-independent, because pairwise fit tests can't see shape
 # ---------------------------------------------------------------------------
@@ -0,0 +1,198 @@
+"""
+server-26#<pending> — pattern B clearance: a unit accepting a NEW dispatch
+("dispatch: are you able to clear and take a run at X / unit: 10-4") carries
+no self-reported clearance language intelligence.py's cleared_units
+extraction looks for (that only catches pattern A, "Unit 7, 10-8"). Before
+this fix, reassignment=True only ever suppressed the unit from re-linking to
+their prior incident (upload.py's corr_units=[] on reassignment) — nothing
+ever released them from it, so it sat "active" until the 90-minute idle
+sweep timed it out instead of being marked cleared by a real event.
+
+`_release_reassigned_units` closes that gap: when a scene is a reassignment,
+scan the OTHER active incidents for unit overlap and release the unit there,
+using the same units_active/units_cleared merge (`_apply_unit_clearance`)
+that explicit 10-8 extraction already used via `_update_incident`.
+"""
+from datetime import datetime, timedelta, timezone
+from unittest.mock import patch
+
+import pytest
+
+from app.internal.incident_correlator import (
+    _apply_unit_clearance, _release_reassigned_units,
+)
+
+NOW = datetime(2026, 9, 20, 12, 0, 0, tzinfo=timezone.utc)
+
+
+def _incident(incident_id="inc-1", units_active=None, units_cleared=None,
+              system_ids=("sys-1",), **overrides):
+    inc = {
+        "incident_id": incident_id,
+        "system_ids": list(system_ids),
+        "units_active": list(units_active or []),
+        "units_cleared": list(units_cleared or []),
+        "status": "active",
+        "updated_at": (NOW - timedelta(minutes=5)).isoformat(),
+    }
+    inc.update(overrides)
+    return inc
+
+
+def _ctx(call_units, all_active, system_id="sys-1", now=NOW):
+    return {"call_units": call_units, "all_active": all_active, "system_id": system_id, "now": now}
+
+
+# ---------------------------------------------------------------------------
+# _apply_unit_clearance — pure merge logic
+# ---------------------------------------------------------------------------
+
+def test_clearance_moves_unit_from_active_to_cleared():
+    inc = _incident(units_active=["6-3"], units_cleared=[])
+    active, cleared, resolved = _apply_unit_clearance(inc, ["6-3"])
+    assert active == []
+    assert cleared == ["6-3"]
+    assert resolved is True
+
+
+def test_clearance_leaves_other_active_units_alone():
+    inc = _incident(units_active=["6-3", "6-7"], units_cleared=[])
+    active, cleared, resolved = _apply_unit_clearance(inc, ["6-3"])
+    assert active == ["6-7"]
+    assert cleared == ["6-3"]
+    assert resolved is False  # 6-7 still active
+
+
+def test_clearing_a_unit_not_tracked_as_active_is_a_noop_for_active_list():
+    inc = _incident(units_active=["6-7"], units_cleared=[])
+    active, cleared, resolved = _apply_unit_clearance(inc, ["ghost-unit"])
+    assert active == ["6-7"]
+    assert cleared == ["ghost-unit"]
+    assert resolved is False
+
+
+def test_no_units_ever_tracked_does_not_auto_resolve():
+    # An incident that never had a unit signal at all — clearing nothing
+    # must not manufacture a resolve.
+    inc = _incident(units_active=[], units_cleared=[])
+    active, cleared, resolved = _apply_unit_clearance(inc, [])
+    assert resolved is False
+
+
+# ---------------------------------------------------------------------------
+# _release_reassigned_units — reassignment releases the unit from its
+# PRIOR incident, scoped correctly, without touching that incident's calls
+# ---------------------------------------------------------------------------
+
+@pytest.mark.asyncio
+async def test_reassignment_clears_unit_from_prior_incident():
+    prior = _incident(incident_id="inc-prior", units_active=["6-3", "6-7"])
+    ctx = _ctx(call_units=["6-3"], all_active=[prior])
+
+    doc_sets = []
+    async def fake_doc_set(collection, doc_id, data, merge=True):
+        doc_sets.append((collection, doc_id, data))
+
+    with patch("app.internal.incident_correlator.fstore") as mock_fstore:
+        mock_fstore.doc_set = fake_doc_set
+        await _release_reassigned_units(ctx, exclude_incident_id="inc-new")
+
+    assert len(doc_sets) == 1
+    collection, doc_id, data = doc_sets[0]
+    assert collection == "incidents" and doc_id == "inc-prior"
+    assert data["units_active"] == ["6-7"]
+    assert data["units_cleared"] == ["6-3"]
+    assert "status" not in data  # 6-7 still active — not auto-resolved
+
+
+@pytest.mark.asyncio
+async def test_reassignment_auto_resolves_when_last_unit_clears():
+    prior = _incident(incident_id="inc-prior", units_active=["6-3"])
+    ctx = _ctx(call_units=["6-3"], all_active=[prior])
+
+    doc_sets = []
+    async def fake_doc_set(collection, doc_id, data, merge=True):
+        doc_sets.append((collection, doc_id, data))
+    async def fake_doc_get(collection, doc_id):
+        return None  # no parent — maybe_resolve_parent exits immediately
+
+    with patch("app.internal.incident_correlator.fstore") as mock_fstore:
+        mock_fstore.doc_set = fake_doc_set
+        mock_fstore.doc_get = fake_doc_get
+        await _release_reassigned_units(ctx, exclude_incident_id=None)
+
+    collection, doc_id, data = doc_sets[0]
+    assert data["status"] == "resolved"
+    assert "resolved_at" in data
+
+
+@pytest.mark.asyncio
+async def test_reassignment_never_touches_the_calls_own_incident():
+    # The call's own decision (link/new) already handled its own incident —
+    # excluding it here prevents double-writing or self-clearing on it.
+    same = _incident(incident_id="inc-new", units_active=["6-3"])
+    ctx = _ctx(call_units=["6-3"], all_active=[same])
+
+    doc_sets = []
+    async def fake_doc_set(collection, doc_id, data, merge=True):
+        doc_sets.append((collection, doc_id, data))
+
+    with patch("app.internal.incident_correlator.fstore") as mock_fstore:
+        mock_fstore.doc_set = fake_doc_set
+        await _release_reassigned_units(ctx, exclude_incident_id="inc-new")
+
+    assert doc_sets == []
+
+
+@pytest.mark.asyncio
+async def test_reassignment_does_not_cross_systems():
+    other_system = _incident(incident_id="inc-other-sys", units_active=["6-3"], system_ids=("sys-2",))
+    ctx = _ctx(call_units=["6-3"], all_active=[other_system], system_id="sys-1")
+
+    doc_sets = []
+    async def fake_doc_set(collection, doc_id, data, merge=True):
+        doc_sets.append((collection, doc_id, data))
+
+    with patch("app.internal.incident_correlator.fstore") as mock_fstore:
+        mock_fstore.doc_set = fake_doc_set
+        await _release_reassigned_units(ctx, exclude_incident_id=None)
+
+    assert doc_sets == []
+
+
+@pytest.mark.asyncio
+async def test_reassignment_with_no_call_units_is_a_noop():
+    prior = _incident(incident_id="inc-prior", units_active=["6-3"])
+    ctx = _ctx(call_units=[], all_active=[prior])
+
+    doc_sets = []
+    async def fake_doc_set(collection, doc_id, data, merge=True):
+        doc_sets.append((collection, doc_id, data))
+
+    with patch("app.internal.incident_correlator.fstore") as mock_fstore:
+        mock_fstore.doc_set = fake_doc_set
+        await _release_reassigned_units(ctx, exclude_incident_id=None)
+
+    assert doc_sets == []
+
+
+@pytest.mark.asyncio
+async def test_reassignment_matches_units_by_normalized_key():
+    # "5-David" vs "5David" — same unit, different transcription — must
+    # still match via the existing _normalize_unit key, not exact string eq.
+    prior = _incident(incident_id="inc-prior", units_active=["5-David"])
+    ctx = _ctx(call_units=["5 David"], all_active=[prior])
+
+    doc_sets = []
+    async def fake_doc_set(collection, doc_id, data, merge=True):
+        doc_sets.append((collection, doc_id, data))
+    async def fake_doc_get(collection, doc_id):
+        return None  # no parent — maybe_resolve_parent exits immediately
+
+    with patch("app.internal.incident_correlator.fstore") as mock_fstore:
+        mock_fstore.doc_set = fake_doc_set
+        mock_fstore.doc_get = fake_doc_get
+        await _release_reassigned_units(ctx, exclude_incident_id=None)
+
+    assert len(doc_sets) == 1
+    assert doc_sets[0][2]["units_cleared"] == ["5-David"]
@@ -0,0 +1,54 @@
+"""
+server-26#<pending> — no per-system unit-ID format awareness existed anywhere
+in the pipeline (vocabulary_learner's "known local terms" is a flat glossary,
+not a structured format). Departments use incompatible unit ID conventions
+(Yorktown: "5-David", sometimes spoken as bare "David"; County:
+"SAM-1"/"airport-3"/"parks-4", a location word + number) and the extraction
+prompt had no way to be told which one a given system uses. This pins the
+prompt-block builder and the template wiring that carries it.
+"""
+from app.internal.intelligence import (
+    _PROMPT_TEMPLATE, _build_unit_format_block, _build_ten_codes_block,
+    _build_transcript_block,
+)
+
+
+def test_empty_hint_produces_no_block():
+    assert _build_unit_format_block(None) == ""
+    assert _build_unit_format_block("") == ""
+
+
+def test_hint_is_labelled_and_fed_to_the_model_verbatim():
+    block = _build_unit_format_block(
+        "Yorktown: <district>-<phonetic name>, e.g. 5-David. Sometimes spoken as just the name alone."
+    )
+    assert "unit ID format" in block
+    assert "5-David" in block
+
+
+def test_prompt_template_renders_with_all_blocks_including_empty_unit_format():
+    # Regression guard: a missing placeholder in .format() raises KeyError at
+    # request time, not import time — this is the cheapest way to catch that
+    # before it reaches a live call.
+    rendered = _PROMPT_TEMPLATE.format(
+        transcript_block=_build_transcript_block("1. Test.", None),
+        talkgroup_name="Test TG",
+        system_id="sys-1",
+        ten_codes_block=_build_ten_codes_block({}),
+        vocabulary_block="",
+        unit_format_block=_build_unit_format_block(""),
+    )
+    assert "Test TG" in rendered
+    assert "1. Test." in rendered
+
+
+def test_prompt_template_renders_with_a_populated_unit_format_block():
+    rendered = _PROMPT_TEMPLATE.format(
+        transcript_block=_build_transcript_block("1. Test.", None),
+        talkgroup_name="Test TG",
+        system_id="sys-1",
+        ten_codes_block=_build_ten_codes_block({}),
+        vocabulary_block="",
+        unit_format_block=_build_unit_format_block("County: <location>-<number>, e.g. SAM-1, airport-3."),
+    )
+    assert "SAM-1" in rendered
@@ -123,10 +123,66 @@ async def test_single_scene_call_still_gets_a_scenes_map_equivalent_to_flat_fiel
            "transcript": "10-4",
            "incident_id": None,
            "corr_debug": {"corr_path": "fast/thin", "corr_consensus": "rules_only"},
+            "incident_type": None,
+            "severity": None,
        }
    }


+@pytest.mark.asyncio
+async def test_scene_entry_captures_its_own_incident_type_not_a_sibling_scenes():
+    """
+    server-26#139: _call_is_substanceless's "type" veto reads ctx["incident_type"]
+    at decision time, but that value was never persisted per-scene — only the
+    last-scene-wins flat field, which #138's dump analysis couldn't
+    distinguish from cross-scene contamination. Pins _apply_and_log's write
+    side: each scene's own scenes.<n> entry carries its own incident_type/
+    severity, distinct from any other scene on the same call. Does NOT cover
+    whether the ctx handed to _call_is_substanceless is the same object that
+    reaches here — that linkage is pinned by test_consensus_gate.py and
+    test_incident_identity.py, not this file.
+    """
+    docs: dict[tuple, dict] = {}
+
+    async def fake_doc_set(collection, doc_id, data, merge=True):
+        docs.setdefault((collection, doc_id), {})
+        _merge(docs[(collection, doc_id)], data)
+
+    decision0 = {
+        "action": "orphan", "matched_incident": None, "incident_type": None,
+        "corr_debug": {"corr_path": "new", "corr_consensus": "tiebreak", "corr_gate_veto": "type"},
+    }
+    ctx0 = {
+        "call_id": "call-5", "scene_index": 0, "scene_transcript": "10-4, clear",
+        "incident_type": "traffic-stop", "call_severity": "routine",
+    }
+
+    decision1 = {
+        "action": "orphan", "matched_incident": None, "incident_type": None,
+        "corr_debug": {"corr_path": "new", "corr_consensus": "agreed"},
+    }
+    ctx1 = {
+        "call_id": "call-5", "scene_index": 1, "scene_transcript": "roll call",
+        "incident_type": None, "call_severity": "moderate",
+    }
+
+    with patch.object(incident_correlator, "fstore") as mock_fstore:
+        mock_fstore.doc_set = fake_doc_set
+        await incident_correlator._apply_and_log(decision0, ctx0)
+        await incident_correlator._apply_and_log(decision1, ctx1)
+
+    doc = docs[("calls", "call-5")]
+    scenes = doc["scenes"]
+    assert scenes["0"]["incident_type"] == "traffic-stop"
+    assert scenes["0"]["severity"] == "routine"
+    assert scenes["1"]["incident_type"] is None
+    assert scenes["1"]["severity"] == "moderate"
+    # _apply_and_log only ever flat-merges corr_debug's own keys (:1460) — a
+    # future corr_debug["incident_type"] would silently clobber
+    # intelligence.py's flat field, so this is asserted, not just commented.
+    assert "incident_type" not in doc
+
+
@pytest.mark.asyncio
 async def test_empty_corr_debug_writes_nothing_same_as_before():
    """Preserve the pre-#96 short-circuit: no corr_debug means no write at
@@ -0,0 +1,97 @@
+"""
+server-26#131 — the re-correlation sweep's orphan filter checked incident_id/
+incident_ids/corr_path but had no way to tell "never processed" apart from
+"real-time pipeline (routers/upload.py _run_intelligence_pipeline) is still
+mid-flight". Racing the sweep against an in-flight real-time correlation could
+land the same call on two different incidents — the exact duplicate-link bug
+#131 found in 3 live dumps (~2% of linked calls). This pins the fix: a call
+whose intelligence_started_at marker is recent is held back from the sweep
+regardless of how orphaned it otherwise looks.
+"""
+from datetime import datetime, timezone, timedelta
+from unittest.mock import patch
+
+import pytest
+
+from app.internal import recorrelation_sweep
+
+
+def _iso(dt: datetime) -> str:
+    return dt.isoformat()
+
+
+class TestPipelineLikelyStillRunning:
+    def test_recent_marker_is_still_running(self):
+        now = datetime.now(timezone.utc)
+        call = {"intelligence_started_at": _iso(now - timedelta(minutes=1))}
+        assert recorrelation_sweep._pipeline_likely_still_running(call, now) is True
+
+    def test_old_marker_is_not_still_running(self):
+        now = datetime.now(timezone.utc)
+        call = {"intelligence_started_at": _iso(now - timedelta(minutes=30))}
+        assert recorrelation_sweep._pipeline_likely_still_running(call, now) is False
+
+    def test_marker_exactly_at_the_threshold_is_not_held_back(self):
+        # age_minutes < MIN_MINUTES_SINCE_PIPELINE_START (strict), so exactly
+        # at the threshold is old enough to release — pins the boundary so it
+        # can't drift to <= by accident and silently double the hold time.
+        now = datetime.now(timezone.utc)
+        threshold = recorrelation_sweep.MIN_MINUTES_SINCE_PIPELINE_START
+        call = {"intelligence_started_at": _iso(now - timedelta(minutes=threshold))}
+        assert recorrelation_sweep._pipeline_likely_still_running(call, now) is False
+
+    def test_no_marker_at_all_is_not_held_back(self):
+        """A pre-#131 call doc, or the marker write itself failed — absence
+        isn't evidence of an in-flight pipeline, so the sweep must still be
+        able to pick these up (that's its whole job)."""
+        now = datetime.now(timezone.utc)
+        assert recorrelation_sweep._pipeline_likely_still_running({}, now) is False
+
+    def test_unparseable_marker_is_not_held_back(self):
+        now = datetime.now(timezone.utc)
+        call = {"intelligence_started_at": "not-a-timestamp"}
+        assert recorrelation_sweep._pipeline_likely_still_running(call, now) is False
+
+
+@pytest.mark.asyncio
+async def test_sweep_pass_skips_a_call_whose_pipeline_just_started():
+    """Integration-shaped: a call that looks orphaned by every OTHER filter
+    (no incident_ids, no corr_path, no skip_reason, under the attempt budget)
+    but has a fresh intelligence_started_at must not reach correlate_call —
+    that's the race #131 found."""
+    now = datetime.now(timezone.utc)
+    racing_call = {
+        "call_id": "call-racing",
+        "started_at": _iso(now - timedelta(minutes=2)),
+        "ended_at": _iso(now - timedelta(minutes=1)),
+        "intelligence_started_at": _iso(now - timedelta(seconds=30)),
+    }
+    genuinely_orphaned_call = {
+        "call_id": "call-genuine-orphan",
+        "started_at": _iso(now - timedelta(minutes=20)),
+        "ended_at": _iso(now - timedelta(minutes=19)),
+        "intelligence_started_at": _iso(now - timedelta(minutes=19)),
+    }
+
+    async def fake_collection_where(collection, clauses):
+        assert collection == "calls"
+        return [racing_call, genuinely_orphaned_call]
+
+    correlate_calls: list[str] = []
+
+    async def fake_correlate_call(**kwargs):
+        correlate_calls.append(kwargs["call_id"])
+        return None  # no match — exercises the "not linked" branch too
+
+    doc_sets: list[tuple] = []
+
+    async def fake_doc_set(collection, doc_id, data, merge=True):
+        doc_sets.append((collection, doc_id, data))
+
+    with patch.object(recorrelation_sweep, "fstore") as mock_fstore, \
+         patch("app.internal.incident_correlator.correlate_call", fake_correlate_call):
+        mock_fstore.collection_where = fake_collection_where
+        mock_fstore.doc_set = fake_doc_set
+        await recorrelation_sweep._run_sweep_pass()
+
+    assert correlate_calls == ["call-genuine-orphan"]
@@ -0,0 +1,94 @@
+"""
+node-26#9 — second-SDR ADS-B telemetry ingestion.
+
+Two things matter here: the endpoint requires node identity (a service/admin
+token has no node_id to attribute the sighting to, so it must 400 rather than
+silently write an orphan doc), and org_id gets stamped from the node's own
+Firestore doc so firestore.rules' docInMyOrg() can gate the frontend's read —
+the same defensive-stamp pattern upload.py already uses for `calls`.
+"""
+from unittest.mock import AsyncMock, patch
+
+from fastapi.testclient import TestClient
+
+from app.main import app
+from app.internal.auth import require_node_service_or_firebase_token
+from app.routers import telemetry
+
+client = TestClient(app)
+
+
+def _override(decoded: dict):
+    app.dependency_overrides[require_node_service_or_firebase_token] = lambda: decoded
+
+
+def teardown_function():
+    app.dependency_overrides.pop(require_node_service_or_firebase_token, None)
+
+
+def test_service_token_without_node_id_is_rejected():
+    _override({"service": True})
+    resp = client.post("/telemetry/adsb", json={"aircraft": []})
+    assert resp.status_code == 400
+
+
+def test_node_upload_upserts_and_stamps_org_id():
+    _override({"node": True, "node_id": "node-1"})
+    with patch.object(telemetry.fstore, "doc_get_cached", AsyncMock(return_value={"org_id": "org-A"})), \
+         patch.object(telemetry.fstore, "doc_set", AsyncMock()) as mock_set:
+        resp = client.post("/telemetry/adsb", json={
+            "aircraft": [{"icao": "A1B2C3", "callsign": "UAL123", "lat": 41.1, "lon": -73.8}],
+        })
+    assert resp.status_code == 200
+    assert resp.json() == {"ok": True, "count": 1}
+    mock_set.assert_awaited_once()
+    (collection, doc_id, doc), kwargs = mock_set.await_args
+    assert collection == "aircraft"
+    assert doc_id == "A1B2C3"
+    assert doc["node_id"] == "node-1"
+    assert doc["org_id"] == "org-A"
+    assert kwargs.get("merge") is True
+
+
+def test_node_upload_skips_entries_missing_icao():
+    _override({"node": True, "node_id": "node-1"})
+    with patch.object(telemetry.fstore, "doc_get_cached", AsyncMock(return_value=None)), \
+         patch.object(telemetry.fstore, "doc_set", AsyncMock()) as mock_set:
+        resp = client.post("/telemetry/adsb", json={"aircraft": [{"icao": ""}]})
+    assert resp.status_code == 200
+    assert resp.json() == {"ok": True, "count": 0}
+    mock_set.assert_not_awaited()
+
+
+def test_ais_service_token_without_node_id_is_rejected():
+    _override({"service": True})
+    resp = client.post("/telemetry/ais", json={"vessels": []})
+    assert resp.status_code == 400
+
+
+def test_ais_node_upload_upserts_and_stamps_org_id():
+    _override({"node": True, "node_id": "node-1"})
+    with patch.object(telemetry.fstore, "doc_get_cached", AsyncMock(return_value={"org_id": "org-A"})), \
+         patch.object(telemetry.fstore, "doc_set", AsyncMock()) as mock_set:
+        resp = client.post("/telemetry/ais", json={
+            "vessels": [{"mmsi": "123456789", "name": "MV TEST", "lat": 41.0, "lon": -73.9}],
+        })
+    assert resp.status_code == 200
+    assert resp.json() == {"ok": True, "count": 1}
+    mock_set.assert_awaited_once()
+    (collection, doc_id, doc), kwargs = mock_set.await_args
+    assert collection == "vessels"
+    assert doc_id == "123456789"
+    assert doc["node_id"] == "node-1"
+    assert doc["org_id"] == "org-A"
+    assert kwargs.get("merge") is True
+
+
+def test_ais_node_upload_skips_entries_missing_mmsi():
+    _override({"node": True, "node_id": "node-1"})
+    with patch.object(telemetry.fstore, "doc_get_cached", AsyncMock(return_value=None)), \
+         patch.object(telemetry.fstore, "doc_set", AsyncMock()) as mock_set:
+        resp = client.post("/telemetry/ais", json={"vessels": [{"mmsi": ""}]})
+    assert resp.status_code == 200
+    assert resp.json() == {"ok": True, "count": 0}
+    mock_set.assert_not_awaited()
@@ -15,6 +15,8 @@ import L from "leaflet";
 import type { CallRecord, IncidentRecord, NodeRecord, NodeStatus } from "@/lib/types";
 import { isKnownSeverity, SEVERITY_COLORS, SEVERITY_LABEL, type Severity } from "@/lib/severity";
 import { MachineOutputNotice } from "@/components/ui/MachineOutputNotice";
+import { useAircraft } from "@/lib/useAircraft";
+import { useVessels } from "@/lib/useVessels";

 // ── Leaflet icon fix ──────────────────────────────────────────────────────────
 delete (L.Icon.Default.prototype as unknown as Record<string, unknown>)._getIconUrl;
@@ -90,6 +92,73 @@ function nodeIcon(status: NodeStatus): L.DivIcon {
  });
 }

+// ── Aircraft icon — node-26#9 second-SDR ADS-B overlay ────────────────────────
+function aircraftIcon(trackDeg: number | null): L.DivIcon {
+  const size = 16;
+  const rotation = trackDeg ?? 0;
+  return L.divIcon({
+    className: "",
+    html: `<div style="width:${size}px;height:${size}px;transform:rotate(${rotation}deg)"><svg width="${size}" height="${size}" viewBox="0 0 24 24" fill="var(--accent)" stroke="var(--surface)" stroke-width="1"><path d="M12 2 L15 11 L22 15 L15 15.5 L14 21 L17 22.5 L12 21.5 L7 22.5 L10 21 L9 15.5 L2 15 L9 11 Z"/></svg></div>`,
+    iconSize: [size, size],
+    iconAnchor: [size / 2, size / 2],
+  });
+}
+
+function AircraftLayer() {
+  const { aircraft } = useAircraft();
+  return (
+    <>
+      {aircraft
+        .filter((a) => a.lat != null && a.lon != null)
+        .map((a) => (
+          <Marker key={a.icao} position={[a.lat as number, a.lon as number]} icon={aircraftIcon(a.track_deg)}>
+            <Popup minWidth={160}>
+              <div className="space-y-1">
+                <div className="font-semibold">{a.callsign || a.icao}</div>
+                <div className="text-xs text-ink-muted">ICAO {a.icao}</div>
+                {a.altitude_ft != null && <div className="text-xs">Altitude: {Math.round(a.altitude_ft)} ft</div>}
+                {a.ground_speed_kt != null && <div className="text-xs">Speed: {Math.round(a.ground_speed_kt)} kt</div>}
+              </div>
+            </Popup>
+          </Marker>
+        ))}
+    </>
+  );
+}
+
+// ── Vessel icon — node-26#9 second-SDR AIS overlay ─────────────────────────────
+function vesselIcon(headingDeg: number | null): L.DivIcon {
+  const size = 14;
+  const rotation = headingDeg ?? 0;
+  return L.divIcon({
+    className: "",
+    html: `<div style="width:${size}px;height:${size}px;transform:rotate(${rotation}deg)"><svg width="${size}" height="${size}" viewBox="0 0 24 24" fill="var(--accent)" stroke="var(--surface)" stroke-width="1"><path d="M12 2 L18 14 L18 20 L6 20 L6 14 Z"/></svg></div>`,
+    iconSize: [size, size],
+    iconAnchor: [size / 2, size / 2],
+  });
+}
+
+function VesselLayer() {
+  const { vessels } = useVessels();
+  return (
+    <>
+      {vessels
+        .filter((v) => v.lat != null && v.lon != null)
+        .map((v) => (
+          <Marker key={v.mmsi} position={[v.lat as number, v.lon as number]} icon={vesselIcon(v.heading_deg)}>
+            <Popup minWidth={160}>
+              <div className="space-y-1">
+                <div className="font-semibold">{v.name || v.mmsi}</div>
+                <div className="text-xs text-ink-muted">MMSI {v.mmsi}</div>
+                {v.speed_kt != null && <div className="text-xs">Speed: {Math.round(v.speed_kt)} kt</div>}
+              </div>
+            </Popup>
+          </Marker>
+        ))}
+    </>
+  );
+}
+
 function nodeFanIcon(members: NodeRecord[]): L.DivIcon {
  const n = members.length;
  const CARD = 13;
@@ -577,6 +646,20 @@ export default function MapView({ nodes, activeCalls, incidents = [], calls = []
            </FeatureGroup>
          </LayersControl.Overlay>

+          {/* Overlay: Aircraft — node-26#9 second-SDR ADS-B live snapshot, opt-in */}
+          <LayersControl.Overlay name="Aircraft">
+            <FeatureGroup>
+              <AircraftLayer />
+            </FeatureGroup>
+          </LayersControl.Overlay>
+
+          {/* Overlay: Vessels — node-26#9 second-SDR AIS live snapshot, opt-in */}
+          <LayersControl.Overlay name="Vessels">
+            <FeatureGroup>
+              <VesselLayer />
+            </FeatureGroup>
+          </LayersControl.Overlay>
+
          {/* Overlay: Weather Radar — NEXRAD via Iowa Env Mesonet; key forces remount on refresh */}
          <LayersControl.Overlay name="Weather Radar">
            <TileLayer
@@ -53,12 +53,39 @@ export interface NodeRecord {
  hardware_preset?: string;
  ppm_override?: number | null;
  node_type?: string;
+  secondary_sdr_mode?: string;
+  sdr_count?: number;
  enforce_override_timeout?: boolean;
  is_overridden?: boolean;
  override_system_id?: string | null;
  override_timeout_at?: string | null;
 }

+export interface AircraftTrack {
+  icao: string;
+  org_id?: string;
+  node_id: string;
+  callsign: string | null;
+  lat: number | null;
+  lon: number | null;
+  altitude_ft: number | null;
+  ground_speed_kt: number | null;
+  track_deg: number | null;
+  last_seen: string;
+}
+
+export interface VesselTrack {
+  mmsi: string;
+  org_id?: string;
+  node_id: string;
+  name: string | null;
+  lat: number | null;
+  lon: number | null;
+  speed_kt: number | null;
+  heading_deg: number | null;
+  last_seen: string;
+}
+
 export interface VocabularyPendingTerm {
  term: string;
  source: "induction" | "correction";
@@ -0,0 +1,52 @@
+"use client";
+
+import { useEffect, useState } from "react";
+import { collection, onSnapshot, query, where, FirestoreError } from "firebase/firestore";
+import { onAuthStateChanged } from "firebase/auth";
+import { db, auth } from "@/lib/firebase";
+import { useAuth } from "@/components/AuthProvider";
+import type { AircraftTrack } from "@/lib/types";
+
+// `aircraft` docs are a live snapshot (one per icao, overwritten on every
+// sighting, node-26#9) — nothing prunes a doc when a plane leaves range, so
+// staleness is filtered client-side rather than assuming the collection only
+// ever holds current traffic.
+const STALE_AFTER_MS = 2 * 60 * 1000;
+
+export function useAircraft() {
+  const [aircraft, setAircraft] = useState<AircraftTrack[]>([]);
+  const [loading, setLoading] = useState(true);
+  const [error, setError] = useState<string | null>(null);
+  const { orgId } = useAuth();
+
+  useEffect(() => {
+    let unsubFirestore: (() => void) | undefined;
+
+    const unsubAuth = onAuthStateChanged(auth, (user) => {
+      if (unsubFirestore) { unsubFirestore(); unsubFirestore = undefined; }
+
+      if (!user || !orgId) {
+        setAircraft([]);
+        setLoading(false);
+        return;
+      }
+
+      const q = query(collection(db, "aircraft"), where("org_id", "==", orgId));
+      unsubFirestore = onSnapshot(q, (snap) => {
+        const now = Date.now();
+        const fresh = snap.docs
+          .map((d) => d.data() as AircraftTrack)
+          .filter((a) => now - new Date(a.last_seen).getTime() < STALE_AFTER_MS);
+        setAircraft(fresh);
+        setLoading(false);
+      }, (err: FirestoreError) => { console.error("useAircraft:", err); setError(err.message); setLoading(false); });
+    });
+
+    return () => {
+      unsubAuth();
+      if (unsubFirestore) unsubFirestore();
+    };
+  }, [orgId]);
+
+  return { aircraft, loading, error };
+}
@@ -0,0 +1,53 @@
+"use client";
+
+import { useEffect, useState } from "react";
+import { collection, onSnapshot, query, where, FirestoreError } from "firebase/firestore";
+import { onAuthStateChanged } from "firebase/auth";
+import { db, auth } from "@/lib/firebase";
+import { useAuth } from "@/components/AuthProvider";
+import type { VesselTrack } from "@/lib/types";
+
+// Same shape as useAircraft — `vessels` is a live snapshot (one per mmsi,
+// overwritten on every sighting, node-26#9), nothing prunes a doc when a
+// vessel goes out of range, so staleness is filtered client-side. AIS
+// position reports are much less frequent than ADS-B (minutes, not
+// seconds), so this window is longer than useAircraft's.
+const STALE_AFTER_MS = 10 * 60 * 1000;
+
+export function useVessels() {
+  const [vessels, setVessels] = useState<VesselTrack[]>([]);
+  const [loading, setLoading] = useState(true);
+  const [error, setError] = useState<string | null>(null);
+  const { orgId } = useAuth();
+
+  useEffect(() => {
+    let unsubFirestore: (() => void) | undefined;
+
+    const unsubAuth = onAuthStateChanged(auth, (user) => {
+      if (unsubFirestore) { unsubFirestore(); unsubFirestore = undefined; }
+
+      if (!user || !orgId) {
+        setVessels([]);
+        setLoading(false);
+        return;
+      }
+
+      const q = query(collection(db, "vessels"), where("org_id", "==", orgId));
+      unsubFirestore = onSnapshot(q, (snap) => {
+        const now = Date.now();
+        const fresh = snap.docs
+          .map((d) => d.data() as VesselTrack)
+          .filter((v) => now - new Date(v.last_seen).getTime() < STALE_AFTER_MS);
+        setVessels(fresh);
+        setLoading(false);
+      }, (err: FirestoreError) => { console.error("useVessels:", err); setError(err.message); setLoading(false); });
+    });
+
+    return () => {
+      unsubAuth();
+      if (unsubFirestore) unsubFirestore();
+    };
+  }, [orgId]);
+
+  return { vessels, loading, error };
+}
@@ -95,6 +95,18 @@ service cloud.firestore {
      allow write: if false;
    }

+    // Live map overlays fed by a node's second SDR (node-26#9). Snapshot
+    // docs, one per icao/mmsi, last-seen-wins — not a history collection.
+    match /aircraft/{icao} {
+      allow read: if docInMyOrg();
+      allow write: if false;
+    }
+
+    match /vessels/{mmsi} {
+      allow read: if docInMyOrg();
+      allow write: if false;
+    }
+
    match /alert_events/{alertId} {
      allow read: if docInMyOrg();
      allow write: if false;
Author	SHA1	Message	Date
Logan CusanoandClaude Sonnet 5	3d2b722c64	Wire AIS end to end: telemetry ingestion + live map overlay node-26#9, same shape as the ADS-B commit. Adds POST /telemetry/ais (same node-key auth, same org_id-stamped upsert-by-key pattern, this time by mmsi into a new `vessels` collection) and its docInMyOrg() firestore rule. Frontend gets useVessels() (mirrors useAircraft(), longer staleness window since AIS position reports are minutes apart, not seconds) and an opt-in "Vessels" map overlay. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>	2026-09-20 16:12:56 -04:00
Logan CusanoandClaude Sonnet 5	5537b095df	Wire ADS-B end to end: telemetry ingestion + live map overlay node-26#9. Adds POST /telemetry/adsb (node-key authed via require_node_service_or_firebase_token) that upserts one Firestore doc per icao into a new `aircraft` collection, org_id stamped from the reporting node the same way upload.py defensively stamps `calls`. firestore.rules gets a matching docInMyOrg()-gated read rule. Frontend: useAircraft() mirrors useNodes()'s onSnapshot pattern, filtering docs older than 2 minutes client-side since nothing prunes a stale aircraft doc server-side yet. MapView gets an opt-in "Aircraft" overlay (unchecked by default, like the weather radar layer) rendering a rotated plane glyph per sighting. Unverified via typecheck — no Node.js/npm on this authoring box yet (see CLAUDE.md Testing reality). Server side is pytest-covered (test_telemetry.py). Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>	2026-09-20 16:08:29 -04:00
Logan CusanoandClaude Sonnet 5	8892e824fc	Add second-SDR fields: secondary_sdr_mode + sdr_count on NodeRecord Server-side half of node-26#9. NodeRecord gains secondary_sdr_mode (none\|adsb\|ais\|op25_2) and sdr_count; checkin ingestion stores both, and PATCH /nodes/{id} accepts and re-pushes secondary_sdr_mode the same way hardware_preset/ppm_override already work, so it isn't wiped by a system reassignment (see server-26#111 for the pre-existing bug that pattern avoids repeating). Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>	2026-09-20 15:56:33 -04:00
Logan CusanoandClaude Sonnet 5	3f69879437	intelligence: stop the police-channel default from typing content-free chatter (#138 ) Build & Deploy / Build & push images (push) Successful in 4m8s Details Build & Deploy / Deploy to VM (push) Failing after 1m12s Details Build & Deploy / Report a failed deploy (push) Successful in 1s Details EXTRACTION_PROMPT's incident_type rule always returned "police" for anything on a police channel unless contradicted, so pure administrative chatter (post check-ins, roll call, bare acknowledgements) got a truthy incident_type and defeated the creation gate's "type" veto (_call_is_substanceless) ~82% of the time (measured server-26#138/CORRELATION_REVIEW_0914.md, confirmed against live 9-14 data: 9/9 type-veto examples checked all had severity "routine" — same population the severity rubric already correctly identifies as content-free, incident_type just wasn't using that signal). Rule now checks for actual event content FIRST, on every channel, before applying the channel-inference defaults; content-free traffic returns "unknown" (already normalizes to None) instead of a channel default, letting the existing gate correctly veto it. No correlator/gate code touched — CORRELATION_REVIEW_0914.md's own recommendation was to fix this in the prompt, not _call_is_substanceless, to avoid risking a real event getting gated out. Verified: 401 pass, 0 fail. Live effect to be confirmed against fresh traffic (AI features just re-enabled this session after being off since 9/14) — tracking in SESSION_STATE.md. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>	2026-09-20 15:42:48 -04:00
Logan CusanoandClaude Sonnet 5	fb0bb15c22	correlator/intelligence: close the incident-clearance gap (dispatch-to-10-8 lifecycle) Build & Deploy / Build & push images (push) Successful in 4m10s Details Build & Deploy / Deploy to VM (push) Failing after 3m25s Details Build & Deploy / Report a failed deploy (push) Successful in 1s Details Two independent fixes, found by tracing why incidents never actually close (only 2/281 incidents in 5 correlation debug windows ever got a non-empty units_cleared; 183 resolved via the 90-min idle sweep instead of a real clear): 1. Pattern B (re-dispatch accept, no explicit 10-8): reassignment=True already fired correctly and suppressed the unit from re-linking to its prior incident, but nothing ever released the unit FROM that incident — it just sat "active" until the idle sweep timed it out. _release_reassigned_units now scans other active incidents for unit overlap on a reassignment and clears the unit there, reusing the same units_active/units_cleared merge (factored out as _apply_unit_clearance) that explicit 10-8 extraction uses. 2. Pattern A (self-clear) extraction was inconsistent for two reasons: no per-system unit ID format awareness anywhere in the pipeline (formats vary by department with zero shared convention), and the cleared_units prompt rule only accepted a unit self-reporting, missing dispatch confirming a unit's status back to them. Added system.unit_format_hint (owner-authored free text, GET/PUT /systems/{id}/unit-format, no auto-induction yet) fed into the extraction prompt, and broadened the cleared_units rule while still requiring an identifiable unit ID (guards against bare "10-8"/"clear" noise, including Whisper hallucination runs already caught upstream by _is_garbage_transcript). Verified: 401 pass, 0 fail (local Linux venv ~/venvs/drb-5c — see CLAUDE.md testing-reality note). server-26#pending — not yet filed, Gitea unreachable from this sandbox. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>	2026-09-20 14:25:01 -04:00
logan	a9197709f8	Merge pull request 'correlator: stop the re-correlation sweep racing an in-flight upload (#131 )' (#154 ) from fix/131-sweep-race-duplicate-link into main Build & Deploy / Build & push images (push) Successful in 4m4s Details Build & Deploy / Deploy to VM (push) Failing after 1m9s Details Build & Deploy / Report a failed deploy (push) Successful in 1s Details	2026-09-14 00:31:14 -04:00
Logan CusanoandClaude Sonnet 5	8dd636af8f	correlator: stop the re-correlation sweep racing an in-flight upload (#131 ) server-26#131: same call_id ends up in TWO incidents' call_ids, byte-identical extracted data, ~2% of linked calls across 3 live dumps. Root cause: the sweep's orphan filter (incident_id/incident_ids/corr_path all absent) can't tell 'never processed' apart from 'real-time pipeline is still mid-flight' -- a call whose STT/scene-extraction/correlation chain (routers/upload.py _run_intelligence_pipeline) hasn't finished yet has none of those fields set, so the sweep picks it up and correlates it independently, sometimes onto a different incident than the real-time path lands on. Confirmed in review: _update_incident/_create_incident append call_id to an incident's call_ids unconditionally, with no cross-incident dedup guard -- preventing the second correlation attempt is the only lever available at this layer. Fix: _run_intelligence_pipeline marks intelligence_started_at on the call doc before any slow step; the sweep holds back any call whose marker is under 15 minutes old (raised from an initial 5 -- see below), regardless of how orphaned it otherwise looks. No marker at all (pre-#131 call doc, or the marker write itself failed) is not held back -- absence isn't evidence of an in-flight pipeline, and that's #131's own pre-existing population. Also covers the /calls/{id}/reprocess path, which calls the same _run_intelligence_pipeline. 15 min, not 5: neither the OpenAI Whisper client nor the Gemini call in llm_correlator.py sets a request timeout (filed server-26#153), so 5 min was a guess against an unbounded tail -- drb-correlation-review flagged this. Raising it is free on the recovery side: a call that finished processing (linked or genuinely orphaned) always has corr_path set and is already excluded by the sweep's other filter, so this constant only ever delays calls that are still actually running. DEFERRED.md row 52 (outside this repo, Version 5C root) updated to flag its ~6 min timing figure as stale. recorrelation_sweep.py had zero test coverage before this. New file covers the guard function's boundary (age < threshold vs exactly-at vs old vs missing vs unparseable) and one integration-shaped test proving a racing call never reaches correlate_call while a genuinely-orphaned call still does. Sandboxed pytest: 381 -> 387. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Tbknwttzou4s46PAykmtix	2026-09-14 00:30:24 -04:00
logan	e27f8f6636	Merge pull request 'correlator/admin: capture per-scene incident_type + severity (#139 )' (#152 ) from fix/139-scene-incident-type into main Build & Deploy / Build & push images (push) Successful in 4m8s Details Build & Deploy / Deploy to VM (push) Failing after 1m40s Details Build & Deploy / Report a failed deploy (push) Successful in 1s Details	2026-09-14 00:14:53 -04:00
Logan CusanoandClaude Sonnet 5	f23026b9ab	correlator: address drb-correlation-review notes on #139 - Document call_severity's routine-coercion asymmetry with incident_type (extraction-said-routine vs extraction-said-nothing look identical). - Test docstring no longer overclaims the ctx-linkage it doesn't cover; points to the tests that do (test_consensus_gate.py, test_incident_identity.py). - scene1 now uses a distinct severity so the test actually exercises both fields symmetrically; the 'no flat top-level clobber' claim is now asserted, not just commented. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Tbknwttzou4s46PAykmtix	2026-09-14 00:14:32 -04:00
Logan CusanoandClaude Sonnet 5	7717fcccdd	correlator/admin: capture per-scene incident_type + severity (#139 ) _call_is_substanceless's "type" veto reads ctx["incident_type"] at decision time, but that value was never persisted per-scene — only the last-scene-wins flat field, which #138's window-4 dump analysis couldn't tell apart from cross-scene contamination without re-guessing from a live dump. Adds incident_type/severity to _apply_and_log's per-scene write and to admin.py's _scene_summary allowlist (the debug-dump reader has its own field allowlist, separate from the write side — silently would not have surfaced otherwise). Sandboxed pytest: 380 -> 381. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Tbknwttzou4s46PAykmtix	2026-09-14 00:09:26 -04:00
logan	454fe7e81c	Merge pull request 'correlator: remove is_dispatch and the tactical fit path entirely (#134 )' (#135 ) from fix/134-drop-tactical-fit-path into main Build & Deploy / Build & push images (push) Successful in 4m6s Details Build & Deploy / Deploy to VM (push) Failing after 1m56s Details Build & Deploy / Report a failed deploy (push) Successful in 1s Details	2026-09-13 14:55:04 -04:00
Logan CusanoandClaude Sonnet 5	422e9a4dc8	correlator: fix two comments left describing the removed tactical path (#134 ) Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Tbknwttzou4s46PAykmtix	2026-09-13 14:54:54 -04:00
Logan CusanoandClaude Sonnet 5	b1884852d5	correlator: remove is_dispatch and the tactical fit path entirely (#134 ) Full removal, not a hardcoded flag: _is_dispatch_channel, _DISPATCH_TG_RE, the is_dispatch parameter, and _call_fits_incident's tactical branch are gone. One evaluation path for every channel. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Tbknwttzou4s46PAykmtix	2026-09-13 14:49:55 -04:00
Logan CusanoandClaude Sonnet 5	0473e6a583	correlator: always run the dispatch-strict fit test, not name-guessed (#134 ) is_dispatch was computed from _is_dispatch_channel(talkgroup_name) and picked between two _call_fits_incident evaluation orders: dispatch (requires a positive signal, runs location-conflict/content-divergence vetoes on unit overlap) vs tactical (skips both vetoes, defaults to True on no signal at all within 20 min). Per #133's reasoning, a name not literally containing dispatch/patched/primary got the unvetoed, default-True path solely because of its label. Hardcoded is_dispatch=True at its one real call site; the tactical branch and its own tests stay in place, unreached. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Tbknwttzou4s46PAykmtix	2026-09-13 14:38:57 -04:00