Node: report every radio heard on a call, not just the last (server-26#304)

P1 of the radio-ID plan. metadata_watcher records each transmission as
[radio, first seen, last seen] from channel rows and call_log grants in both
audio and console modes. Sightings of the same radio under 2 s apart are
merged, because the P0 probe found the ID in only 44% of polls during a
burst. call_end gains `radios: [{radio, start, end}]`.

main.py rebases the offsets into the trimmed audio file's frame, so
wall clock = audio_start_epoch + offset, the mapping C2 already uses. It
also sends the list on /upload (UploadMeta.radios), because C2 starts the
pipeline from /upload before call_end arrives.

srcaddr keeps its meaning (latest talker). Metadata only: no boundary,
timestamp or attribution change (drb-audio-review).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

drb-edge-node/app/internal/call_recorder.py

@@ -785,6 +785,7 @@ class CallRecorder:
         system_id: Optional[str] = None,
         audio_start_epoch: Optional[float] = None,
         audio_end_epoch: Optional[float] = None,
+        radios: Optional[str] = None,
     ) -> Optional[str]:
         """
         Upload once, inline. Returns the storage URL on success.
@@ -802,6 +803,7 @@ class CallRecorder:
         meta = UploadMeta(
             call_id=call_id, talkgroup_id=talkgroup_id, talkgroup_name=talkgroup_name,
             system_id=system_id, audio_start_epoch=audio_start_epoch, audio_end_epoch=audio_end_epoch,
+            radios=radios,
         )
         result = await self._post_inline(file_path, meta)
         self.spool.last_result = {"call_id": call_id, "kind": result.kind, "status": result.status,

drb-edge-node/app/internal/metadata_watcher.py

@@ -132,6 +132,11 @@ ATTRIBUTION_LOOKAHEAD_SECONDS = 2.0
 CONSOLE_HISTORY_SECONDS = 180.0
 CONSOLE_HISTORY_MAX = 1200
 
+# Sightings of the same radio closer together than this are one transmission.
+# The radio ID shows up in only ~44% of 0.5 s polls during a burst (server-26#304
+# P0 probe, 2026-11-08), so gaps of a few polls inside one key-up are normal.
+TX_MERGE_GAP_SECONDS = 2.0
+
 # How far back to look for a duplicate before appending a grant. OP25's call_log
 # deque drains on read so repeats should not happen, but a re-delivered entry
 # would otherwise inflate the transmission count and the attribution score.
@@ -207,6 +212,21 @@ def _talker(channel: Dict[str, Any]) -> Optional[tuple]:
     return tgid, rid
 
 
+def rebase_radios(radios: List[Dict[str, Any]], shift: float, span: float) -> List[Dict[str, Any]]:
+    """Move call-relative radio offsets into another frame (the trimmed audio file).
+
+    `shift` is call start minus the new origin; offsets land clamped to [0, span]
+    so a talker before the file's first sample sits at 0, not before it.
+    """
+    span = max(0.0, span)
+    return [
+        {**r,
+         "start": round(min(max(0.0, r["start"] + shift), span), 2),
+         "end": round(min(max(0.0, r["end"] + shift), span), 2)}
+        for r in radios
+    ]
+
+
 def _iso(epoch: Optional[float]) -> Optional[str]:
     if epoch is None:
         return None
@@ -254,6 +274,10 @@ class MetadataWatcher:
         self._current_srcaddr: Optional[int] = None
         self._started_at: Optional[float] = None   # audio onset, or grant epoch in fallback mode
         self._transmissions: int = 0
+        # Every radio heard on the open segment, in order: [rid, first_epoch,
+        # last_epoch]. srcaddr above keeps only the latest one; one recording
+        # often holds a dispatcher and a unit (server-26#304).
+        self._tx: List[List[Any]] = []
         # True when the open segment is governed by audio, False for the
         # console fallback. Fixed at open so capture flapping cannot switch the
         # rules underneath a live segment.
@@ -559,6 +583,7 @@ class MetadataWatcher:
             self._current_tgid_name = entry.get("tgtag") or ""
             self._current_freq = entry.get("freq")
             self._current_srcaddr = _as_int(entry.get("rid"))
+            self._note_tx(self._current_srcaddr, started_at)
             self._transmissions += 1
             logger.info(
                 f"Late attribution: segment {self._active_call_id} adopted tgid {tgid} from a grant "
@@ -571,11 +596,13 @@ class MetadataWatcher:
             # of a single conversation lands in one file.
             self._transmissions += 1
             self._refresh_meta_from_log(entry)
+            self._note_tx(_as_int(entry.get("rid")), started_at)
             return
 
         if self._same_channel(tgid, entry.get("freq"), entry.get("tgtag") or ""):
             self._transmissions += 1
             self._refresh_meta_from_log(entry)
+            self._note_tx(_as_int(entry.get("rid")), started_at)
             return
 
         # FORCED SPLIT. Two talkgroups can be back to back with no silence
@@ -626,6 +653,7 @@ class MetadataWatcher:
                 ours = True
                 if srcaddr is not None:
                     self._current_srcaddr = srcaddr
+                    self._note_tx(srcaddr, now)
                 self._last_activity = now
                 self._refresh_meta_from_channel(channel)
 
@@ -638,6 +666,7 @@ class MetadataWatcher:
                 self._current_tgid_name = active[0].get("tag") or ""
                 self._current_freq = active[0].get("freq")
                 self._current_srcaddr = _as_int(active[0].get("srcaddr"))
+                self._note_tx(self._current_srcaddr, now)
                 logger.info(f"Late attribution: segment {self._active_call_id} adopted tgid {tgid} from channel state.")
             return
 
@@ -709,6 +738,7 @@ class MetadataWatcher:
             self._last_tx_end = None
             self._last_activity = now
             self._refresh_meta_from_log(entry)
+            self._note_tx(_as_int(entry.get("rid")), started_at)
             return
 
         await self._close_segment(started_at + _tail_pad(), reason="tgid_change")
@@ -730,6 +760,7 @@ class MetadataWatcher:
                 tx_active = True
                 if srcaddr is not None:
                     self._current_srcaddr = srcaddr
+                    self._note_tx(srcaddr, now)
                 self._refresh_meta_from_channel(channel)
             elif chan_tgid is not None:
                 foreign_active_tgid = chan_tgid
@@ -792,6 +823,28 @@ class MetadataWatcher:
     # Segment open / close
     # ------------------------------------------------------------------
 
+    def _note_tx(self, rid: Optional[int], epoch: float) -> None:
+        """Record a sighting of `rid` talking on the open segment."""
+        if rid is None or self._active_call_id is None:
+            return
+        last = self._tx[-1] if self._tx else None
+        if last is not None and last[0] == rid and epoch - last[2] <= TX_MERGE_GAP_SECONDS:
+            last[2] = max(last[2], epoch)
+            return
+        self._tx.append([rid, epoch, epoch])
+
+    def _radios(self, start: float, end: float) -> List[Dict[str, Any]]:
+        """The segment's transmissions as offsets (seconds) from its start, clamped to it."""
+        span = max(0.0, end - start)
+        return [
+            {
+                "radio": rid,
+                "start": round(min(max(0.0, first - start), span), 2),
+                "end": round(min(max(0.0, last - start), span), 2),
+            }
+            for rid, first, last in self._tx
+        ]
+
     def _refresh_meta_from_log(self, entry: Dict[str, Any]) -> None:
         if not self._current_tgid_name:
             self._current_tgid_name = entry.get("tgtag") or ""
@@ -826,6 +879,8 @@ class MetadataWatcher:
         self._started_at = started_at
         self._transmissions = max(1, transmissions)
         self._audio_driven = audio_driven
+        self._tx = []
+        self._note_tx(srcaddr, started_at)
 
         # Console fallback assumes the transmission is still up; it learns
         # otherwise from the next channel scan.
@@ -876,6 +931,7 @@ class MetadataWatcher:
             "ended_at": _iso(end_epoch),
             "ended_at_epoch": end_epoch,
             "transmissions": self._transmissions,
+            "radios": self._radios(started_at if started_at is not None else end_epoch, end_epoch),
             "end_reason": reason,
             "attributed": attributed,
             "driver": "audio" if self._audio_driven else "console",
@@ -910,6 +966,7 @@ class MetadataWatcher:
         self._current_srcaddr = None
         self._started_at = None
         self._transmissions = 0
+        self._tx = []
         self._tx_active = False
         self._last_tx_end = None
         self._audio_driven = False
@@ -949,6 +1006,10 @@ class MetadataWatcher:
                 self._current_freq = found.freq
             if found.rid is not None and self._current_srcaddr is None:
                 self._current_srcaddr = found.rid
+                if not self._tx:
+                    # Only as the sole entry: the window's latest radio has no
+                    # timing of its own, so it must not reorder real sightings.
+                    self._note_tx(found.rid, start)
             self._transmissions = max(self._transmissions, found.grants)
             return
 

drb-edge-node/app/internal/upload_spool.py

@@ -71,6 +71,10 @@ class UploadMeta:
     system_id: Optional[str] = None
     audio_start_epoch: Optional[float] = None
     audio_end_epoch: Optional[float] = None
+    # JSON list of {radio, start, end} — every radio heard on the call
+    # (server-26#304). Sent here as well as on call_end because C2 starts the
+    # pipeline from /upload, before call_end arrives.
+    radios: Optional[str] = None
 
 
 @dataclass
@@ -108,6 +112,8 @@ def _form(meta: UploadMeta) -> Dict[str, str]:
         form["audio_start_epoch"] = f"{meta.audio_start_epoch:.3f}"
     if meta.audio_end_epoch is not None:
         form["audio_end_epoch"] = f"{meta.audio_end_epoch:.3f}"
+    if meta.radios:
+        form["radios"] = meta.radios
     return form
 
 

drb-edge-node/app/main.py

@@ -1,4 +1,5 @@
 import asyncio
+import json
 from contextlib import asynccontextmanager
 from datetime import datetime, timezone
 from typing import Optional
@@ -9,7 +10,7 @@ from app.models import SystemConfig, normalize_sdr_pins, normalize_secondary_pri
 from app.internal.logger import logger
 from app.internal.mqtt_manager import mqtt_manager
 from app.internal import credentials
-from app.internal.metadata_watcher import metadata_watcher
+from app.internal.metadata_watcher import metadata_watcher, rebase_radios
 from app.internal.call_recorder import call_recorder
 from app.internal.discord_radio import radio_bot
 from app.internal.config_manager import (
@@ -93,6 +94,15 @@ async def on_call_end(data: dict):
         data["audio_end_at"] = _iso(recording.audio_end_epoch)
         data["audio_start_epoch"] = recording.audio_start_epoch
         data["audio_end_epoch"] = recording.audio_end_epoch
+        # Radio offsets were measured from the call start; C2 maps an offset
+        # to wall clock as audio_start_epoch + t (above), so put them in the
+        # audio file's frame (server-26#304).
+        if data.get("radios") and data.get("started_at_epoch") is not None:
+            data["radios"] = rebase_radios(
+                data["radios"],
+                shift=data["started_at_epoch"] - recording.audio_start_epoch,
+                span=recording.audio_end_epoch - recording.audio_start_epoch,
+            )
         data["audio_lead_trimmed"] = round(recording.lead_trimmed, 3)
         data["audio_tail_trimmed"] = round(recording.tail_trimmed, 3)
         if recording.clamped_seconds:
@@ -108,6 +118,7 @@ async def on_call_end(data: dict):
             system_id=node_cfg.assigned_system_id,
             audio_start_epoch=recording.audio_start_epoch,
             audio_end_epoch=recording.audio_end_epoch,
+            radios=json.dumps(data["radios"]) if data.get("radios") else None,
         )
         if audio_url:
             data["audio_url"] = audio_url

drb-edge-node/tests/test_metadata_watcher.py

@@ -27,6 +27,7 @@ from app.internal.metadata_watcher import (
     MAX_SEGMENT_SECONDS,
     MetadataWatcher,
     OP25_OFFLINE_GRACE,
+    rebase_radios,
 )
 from app.internal.op25_client import TerminalUpdate, parse_terminal_messages
 
@@ -1147,3 +1148,86 @@ async def test_console_history_is_bounded(hearing, clock, audio):
     clock.advance(CONSOLE_HISTORY_SECONDS + 1)
     await tick(hearing, update())
     assert all(e.epoch >= clock.now - CONSOLE_HISTORY_SECONDS for e in hearing._console)
+
+
+# ---------------------------------------------------------------------------
+# Every radio on the call (server-26#304) — not just the last srcaddr
+# ---------------------------------------------------------------------------
+
+@pytest.mark.asyncio
+async def test_call_end_lists_every_radio_in_order(watcher, clock):
+    """Dispatcher, unit, dispatcher on one talkgroup: three transmissions."""
+    start = clock.now
+    await tick(watcher, update(call_log=[grant(1234, start, rid=555)],
+                               channels=[channel(tgid=1234, srcaddr=555)]))
+    clock.advance(1.0)
+    await tick(watcher, update(channels=[channel(tgid=1234, srcaddr=555)]))
+    clock.advance(1.0)
+    await tick(watcher, update(channels=[channel(tgid=1234, srcaddr=777)]))
+    clock.advance(0.5)
+    await tick(watcher, update(channels=[channel(tgid=1234, srcaddr=777)]))
+    clock.advance(1.0)
+    await tick(watcher, update(channels=[channel(tgid=1234, srcaddr=555)]))
+    clock.advance(0.5)
+    await tick(watcher, update(channels=[channel(tgid=1234, srcaddr=0, hold_tgid=1234)]))
+    clock.advance(settings.call_idle_timeout + 0.5)
+    await tick(watcher, update(channels=[channel()]))
+
+    payload = watcher.on_call_end.call_args[0][0]
+    assert [r["radio"] for r in payload["radios"]] == [555, 777, 555]
+    assert payload["radios"][0] == {"radio": 555, "start": 0.0, "end": 1.0}
+    assert payload["radios"][1] == {"radio": 777, "start": 2.0, "end": 2.5}
+    # srcaddr keeps its old meaning (the latest talker) for existing readers.
+    assert payload["srcaddr"] == 555
+
+
+@pytest.mark.asyncio
+async def test_missed_polls_inside_one_keyup_stay_one_transmission(watcher, clock):
+    """The radio ID drops out of some polls mid-transmission; that is not a new key-up."""
+    await tick(watcher, update(call_log=[grant(1234, clock.now, rid=555)],
+                               channels=[channel(tgid=1234, srcaddr=555)]))
+    clock.advance(0.5)
+    await tick(watcher, update(channels=[channel(tgid=1234, srcaddr=0, hold_tgid=1234)]))
+    clock.advance(1.0)
+    await tick(watcher, update(channels=[channel(tgid=1234, srcaddr=555)]))
+    clock.advance(0.5)
+    await tick(watcher, update(channels=[channel(tgid=1234, srcaddr=0, hold_tgid=1234)]))
+    clock.advance(settings.call_idle_timeout + 0.5)
+    await tick(watcher, update(channels=[channel()]))
+
+    radios = watcher.on_call_end.call_args[0][0]["radios"]
+    assert radios == [{"radio": 555, "start": 0.0, "end": 1.5}]
+
+
+@pytest.mark.asyncio
+async def test_radio_list_resets_between_calls(watcher, clock):
+    await tick(watcher, update(call_log=[grant(1234, clock.now, rid=555)]))
+    clock.advance(0.2)
+    await tick(watcher, update(call_log=[grant(4321, clock.now, rid=888)]))
+
+    first = watcher.on_call_end.call_args[0][0]
+    assert [r["radio"] for r in first["radios"]] == [555]
+    assert watcher._tx == [[888, clock.now, clock.now]]
+
+
+@pytest.mark.asyncio
+async def test_audio_mode_records_a_radio_seen_only_in_a_grant(hearing, clock, audio):
+    """A short key-up between channel polls is only visible as a call_log grant."""
+    audio.speak()
+    await tick(hearing, update(call_log=[grant(1234, clock.now, rid=555)],
+                               channels=[channel(tgid=1234, srcaddr=555)]))
+    clock.advance(1.0)
+    audio.speak()
+    await tick(hearing, update(call_log=[grant(1234, clock.now, rid=777)]))
+    clock.advance(settings.call_silence_timeout + 0.5)
+    await tick(hearing, update(channels=[channel()]))
+
+    radios = hearing.on_call_end.call_args[0][0]["radios"]
+    assert [r["radio"] for r in radios] == [555, 777]
+
+
+def test_rebase_radios_moves_offsets_into_the_audio_frame():
+    """The file starts 0.25 s of pre-roll before the call, then 1.0 s of lead was trimmed."""
+    radios = [{"radio": 555, "start": 0.0, "end": 2.0}, {"radio": 777, "start": 3.0, "end": 9.0}]
+    out = rebase_radios(radios, shift=-0.75, span=6.0)
+    assert out == [{"radio": 555, "start": 0.0, "end": 1.25}, {"radio": 777, "start": 2.25, "end": 6.0}]

drb-edge-node/tests/test_upload_spool.py

@@ -228,3 +228,10 @@ async def test_wake_makes_everything_due_now(tmp_path):
     sp.add(_recording(tmp_path), UploadMeta(call_id="c1"), UploadResult(RETRY, status=401))
     sp.wake()
     assert await sp.drain_once() == 1
+
+
+def test_form_carries_the_radio_list_and_omits_it_when_empty():
+    """C2 starts correlation from /upload, so the radio list rides on it (server-26#304)."""
+    radios = json.dumps([{"radio": 555, "start": 0.0, "end": 1.0}])
+    assert upload_spool._form(UploadMeta(call_id="c1", radios=radios))["radios"] == radios
+    assert "radios" not in upload_spool._form(UploadMeta(call_id="c1"))