feat: accept link=<link_id> on the replay range and frames endpoints

Narrows the merged frame stream to one capture source BEFORE counting,
slicing and bucketing (frame_count / frames | buckets all recomputed on
the narrowed set), AND-composing with the display filter. A pure
identity filter applied before any engine work — only the selected
link's pcap gets a sharkd pass, so link+filter is cheaper than filter
alone.

Two boundaries by contract with the Web UI:
- sources[] stays the tag's stable inventory: every capture source
  listed with engine-free TOTAL counts, unaffected by link/filter — a
  source dropdown must not shrink when the view narrows (this also
  settles sources[].count on total counts rather than post-filter
  matches, which no spec ever required)
- an unknown link_id matches nothing: frame_count 0, start null, empty
  frames/buckets — the same shape as a zero-match display filter,
  deliberately not a 404; an empty link param is treated as absent

tests/api/routes/controller/test_marker_replay.py

@@ -209,6 +209,47 @@ class TestReplayRoutes:
         assert response.status_code == status.HTTP_404_NOT_FOUND
         assert "rebuilt" in response.json()["message"]
 
+    async def test_range_link_param_narrows_and_keeps_sources(
+        self, app: FastAPI, client: AsyncClient, project: Project, monkeypatch
+    ) -> None:
+
+        r1, r2 = UDPLink(project), UDPLink(project)
+        project._links.update({r1.id: r1, r2.id: r2})
+
+        def _wire(link, node_id, frames):
+            link._markers["icmp"] = {"bpf": "icmp", "tag": 7, "enabled": False, "color": None,
+                                     "highlight_duration": None, "capture_node_id": node_id,
+                                     "direction": None, "data_link_type": "DLT_EN10MB"}
+            _write_pcap(f"{project.markers_directory}/{node_id}_{link.id}_icmp.pcap", frames)
+
+        _wire(r1, "n1", [(1693472000, 0, b"a" * 60), (1693472002, 0, b"a" * 60)])
+        _wire(r2, "n2", [(1693472001, 0, b"b" * 60)])
+
+        async def fake_columns(pcap, filter_expr):
+            return {1: _cols(), 2: _cols()}
+
+        monkeypatch.setattr(marker_replay, "_columns_for", fake_columns)
+
+        response = await client.get(
+            app.url_path_for("replay_tag_range", project_id=project.id, tag=7),
+            params={"link": r1.id},
+        )
+        body = response.json()
+        assert body["frame_count"] == 2
+        assert {f["link_id"] for f in body["frames"]} == {r1.id}
+        # The source dropdown keeps the whole tag inventory.
+        assert sorted(s["count"] for s in body["sources"]) == [1, 2]
+
+        # Unknown link: 200 with an empty timeline, same shape as a zero-match
+        # filter — not a 404.
+        response = await client.get(
+            app.url_path_for("replay_tag_range", project_id=project.id, tag=7),
+            params={"link": "00000000-0000-0000-0000-000000000000"},
+        )
+        body = response.json()
+        assert body["frame_count"] == 0 and body["frames"] == [] and body["start"] is None
+        assert len(body["sources"]) == 2
+
     @sharkd_present
     async def test_range_columns_and_filter_end_to_end(
         self, app: FastAPI, client: AsyncClient, project: Project, no_residual_sessions

tests/controller/test_marker_replay.py

@@ -301,7 +301,77 @@ class TestTimeline:
         assert timeline["end"] == "1693472002.000000"
         # frame numbers keep their ORIGINAL pcap identity through the filter.
         assert [f["frame_number"] for f in timeline["frames"]] == [1, 3]
-        assert timeline["sources"][0]["count"] == 2
+        # sources[] is the stable inventory: engine-free TOTAL counts,
+        # unaffected by link/filter (the WebUI source dropdown must not
+        # shrink when the view narrows).
+        assert timeline["sources"][0]["count"] == 3
+
+
+class TestLinkFilter:
+    """``link_id`` narrows the frame stream to one capture source before
+    counting/slicing/bucketing; sources stay the full inventory."""
+
+    def _project(self, tmp_path, monkeypatch, columns_override=None):
+        _write_pcap(tmp_path / "n1_linkA_icmp.pcap", [
+            (1693472000, 500000, b"a" * 60),
+            (1693472002, 000000, b"a" * 60),
+        ])
+        _write_pcap(tmp_path / "n2_linkB_icmp.pcap", [
+            (1693472001, 000000, b"b" * 60),
+            (1693472002, 000000, b"b" * 60),
+        ])
+
+        async def fake_columns(pcap, filter_expr):
+            if columns_override is not None:
+                return await columns_override(os.path.basename(pcap), filter_expr)
+            src = "10.0.0.1" if "n1" in pcap else "10.0.0.2"
+            return {1: _cols(src=src), 2: _cols(src=src)}
+
+        monkeypatch.setattr(marker_replay, "_columns_for", fake_columns)
+        return _fake_project(tmp_path, {
+            "linkA/icmp": _marker_entry(tag=7, enabled=False, node_id="n1"),
+            "linkB/icmp": _marker_entry(tag=7, enabled=False, node_id="n2"),
+        })
+
+    async def test_link_narrows_before_count_and_slice(self, tmp_path, monkeypatch):
+        timeline = await build_timeline(self._project(tmp_path, monkeypatch), tag=7, link_id="linkA")
+        assert timeline["frame_count"] == 2
+        assert timeline["start"] == "1693472000.500000"
+        assert [f["link_id"] for f in timeline["frames"]] == ["linkA", "linkA"]
+        # sources stay the FULL inventory with engine-free totals.
+        assert sorted((s["link_id"], s["count"]) for s in timeline["sources"]) == [
+            ("linkA", 2), ("linkB", 2)
+        ]
+
+    async def test_unknown_link_is_empty_success(self, tmp_path, monkeypatch):
+        timeline = await build_timeline(self._project(tmp_path, monkeypatch), tag=7, link_id="nope")
+        assert timeline["frame_count"] == 0
+        assert timeline["start"] is None and timeline["end"] is None
+        assert timeline["frames"] == []
+        assert len(timeline["sources"]) == 2
+
+    async def test_link_and_filter_compose_as_and(self, tmp_path, monkeypatch):
+        # The injected "matching set" contains only frame 2 per source —
+        # combined with link=linkA the view is exactly that one frame.
+        async def override(basename, filter_expr):
+            assert filter_expr == "tcp"
+            return {2: _cols(proto="TCP")}
+
+        project = self._project(tmp_path, monkeypatch, columns_override=override)
+        timeline = await build_timeline(project, tag=7, filter_expr="tcp", link_id="linkA")
+        assert timeline["frame_count"] == 1
+        assert timeline["frames"][0]["frame_number"] == 2
+        assert timeline["frames"][0]["link_id"] == "linkA"
+
+    async def test_empty_link_string_means_absent(self, tmp_path, monkeypatch):
+        timeline = await build_timeline(self._project(tmp_path, monkeypatch), tag=7, link_id="")
+        assert timeline["frame_count"] == 4
+
+    async def test_query_frames_window_over_link_stream(self, tmp_path, monkeypatch):
+        project = self._project(tmp_path, monkeypatch)
+        result = await query_frames(project, tag=7, ts="1693472002.000000",
+                                    window_ms=0, link_id="linkB")
+        assert [f["link_id"] for f in result["frames"]] == ["linkB"]
 
 
 class TestQueryFrames: