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. 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 5 minutes old, 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. 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:19:13 -04:00
19 changed files with 22 additions and 968 deletions
@@ -1452,32 +1452,15 @@ 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"),
+                "transcript":   ctx.get("scene_transcript"),
-                "incident_id":   incident_id,
+                "incident_id":  incident_id,
-                "corr_debug":    corr_debug,
+                "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:
@@ -1863,71 +1846,6 @@ def _call_fits_incident(
    return False, "no_signal"
 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(
    inc: dict,
    call_id: str,
@@ -1971,8 +1889,11 @@ 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)
-    inc_with_active_update = {**inc, "units_active": units_active, "units_cleared": units_cleared}
+    for u in (cleared_units or []):
-    units_active, units_cleared, _ = _apply_unit_clearance(inc_with_active_update, cleared_units or [])
+        if u in units_active:
            units_active.remove(u)
        if u not in units_cleared:
            units_cleared.append(u)
    # 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. 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.
+- 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.
 - Do not invent details not present in the transcript.
- 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.
+- 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.
 - 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,15 +74,18 @@ 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: 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.
+- 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.
 - 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}{unit_format_block}{transcript_block}"""
+{ten_codes_block}{vocabulary_block}{transcript_block}"""
-# "unknown" is deliberately absent — normalises to None, which is what lets
+# The incident_type enum offered to the model in EXTRACTION_PROMPT.  Kept here
-# the creation gate veto a content-free call (server-26#138).
+# 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.
 _VALID_INCIDENT_TYPES = frozenset({"fire", "ems", "police", "accident", "other"})
 # Geographic bias radius for geocoding — half-width in degrees (~55 km)
@@ -153,23 +156,6 @@ 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,
@@ -196,15 +182,12 @@ async def extract_scenes(
    """
    vocabulary: list[str] = []
    ten_codes: dict[str, str] = {}
    unit_format_hint: str = ""
    if system_id:
-        # Single cached read — vocabulary, ten_codes and unit_format_hint all
+        # Single cached read — vocabulary and ten_codes live on the same document.
        # live on the same document.
        system_doc = await fstore.doc_get_cached("systems", system_id)
        if system_doc:
-            vocabulary       = system_doc.get("vocabulary") or []
+            vocabulary = system_doc.get("vocabulary") or []
-            ten_codes        = system_doc.get("ten_codes") 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(
@@ -271,7 +254,6 @@ 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:
@@ -695,7 +677,6 @@ 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
@@ -713,7 +694,6 @@ 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,8 +111,6 @@ 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,
@@ -143,11 +141,6 @@ 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
@@ -32,17 +32,7 @@ from app.config import settings
 # 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.
-#
+MIN_MINUTES_SINCE_PIPELINE_START = 5
 # 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
@@ -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, telemetry
+from app.routers import enrollment, media, org, waitlist
 from app.internal import dynsec
 from app.internal import firestore as fstore
@@ -120,7 +120,6 @@ 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,43 +62,12 @@ 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,11 +109,6 @@ 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,7 +195,6 @@ 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}")
@@ -228,8 +227,6 @@ 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,10 +24,6 @@ class TenCodesBody(BaseModel):
    ten_codes: Dict[str, str]
 class UnitFormatBody(BaseModel):
    unit_format_hint: str
 class PendingTermBody(BaseModel):
    talkgroup_id: int
    term: str
@@ -159,38 +155,6 @@ 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")
@@ -1,125 +0,0 @@
 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)}
@@ -1,198 +0,0 @@
 """
 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"]
@@ -1,54 +0,0 @@
 """
 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,66 +123,10 @@ 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
@@ -1,94 +0,0 @@
 """
 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,8 +15,6 @@ 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;
@@ -92,73 +90,6 @@ 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;
@@ -646,20 +577,6 @@ 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,39 +53,12 @@ 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";
@@ -1,52 +0,0 @@
 "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 };
 }
@@ -1,53 +0,0 @@
 "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,18 +95,6 @@ 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;