test(topology): cover compose labels and tolerate docker filter kwarg

test_compose asserts the new decnet.topology.* labels land on both base
deckies (role=base, no service marker) and service fragments
(service=true). The stub docker client in test_deploy grew a filters
kwarg so it keeps matching the real .networks.list(filters=...) call
signature now used by the deployer.

tests/topology/test_compose.py

@@ -89,6 +89,39 @@ async def test_compose_forwards_l3_sets_sysctl(repo):
         assert "NET_ADMIN" in base["cap_add"]
 
 
+@pytest.mark.anyio
+async def test_compose_labels_service_containers_for_collector(repo):
+    """Service fragments must carry ``decnet.topology.service=true`` so
+    the host-side collector picks up their log streams — the old fleet
+    state file never mentions topology containers."""
+    plan = generate(_cfg())
+    tid = await persist(repo, plan)
+    hydrated = await hydrate(repo, tid)
+    data = generate_topology_compose(hydrated)
+
+    service_keys = [
+        k for k in data["services"]
+        if "-" in k and k not in {d["decky_config"]["name"] for d in hydrated["deckies"]}
+    ]
+    assert service_keys, "expected at least one service container"
+    for k in service_keys:
+        labels = data["services"][k].get("labels") or {}
+        assert labels.get("decnet.topology.service") == "true", (
+            f"service {k!r} missing collector-discovery label: {labels}"
+        )
+        assert labels.get("decnet.topology.id") == tid
+        assert "decnet.topology.decky" in labels
+        assert "decnet.topology.service_name" in labels
+
+    # Base containers get their own label (role=base) but MUST NOT carry
+    # the service marker — otherwise the collector double-attaches.
+    base_keys = {d["decky_config"]["name"] for d in hydrated["deckies"]}
+    for k in base_keys:
+        labels = data["services"][k].get("labels") or {}
+        assert labels.get("decnet.topology.role") == "base"
+        assert labels.get("decnet.topology.service") != "true"
+
+
 def test_teardown_order_is_leaf_first():
     lans = [
         {"name": "LAN-00"},

tests/topology/test_deploy.py

@@ -75,7 +75,7 @@ async def test_deploy_failure_transitions_to_failed(repo, tmp_path, monkeypatch)
     class _BoomClient:
         def __init__(self):
             self.networks = self
-        def list(self, names=None):  # noqa: ARG002
+        def list(self, names=None, filters=None):  # noqa: ARG002
             return []
         def create(self, *a, **kw):  # noqa: ARG002
             raise RuntimeError("boom: docker daemon unreachable")
@@ -107,7 +107,7 @@ async def test_teardown_from_failed_marks_torn_down(repo, tmp_path, monkeypatch)
     class _StubClient:
         def __init__(self):
             self.networks = self
-        def list(self, names=None):  # noqa: ARG002
+        def list(self, names=None, filters=None):  # noqa: ARG002
             return []
 
     with patch("decnet.engine.deployer.docker.from_env", return_value=_StubClient()):