feat(topology): thread per-service config overrides through compose

MazeNET phase 2 step 2. Mirrors the flat-fleet service_config pattern
(DeckyConfig.service_config → composer → svc.compose_fragment) into the
topology compose pipeline, so a hand-authored decky can carry overrides
like {"ssh": {"password": "megapassword"}} and the ssh fragment reads
them just like the flat path does.

- _PlannedDecky gains service_config: dict[str, dict].
- persist() stores it under decky_config["service_config"].
- topology/compose.py passes cfg.get("service_config", {}).get(svc, {})
  to svc.compose_fragment(service_cfg=...).

Schema unchanged — service_config lives inside the existing
decky_config JSON blob. Zero changes in decnet/services/*.

decnet/topology/compose.py

@@ -62,6 +62,7 @@ def generate_topology_compose(hydrated: dict[str, Any]) -> dict:
         name = cfg["name"]
         ips_by_lan: dict[str, str] = cfg["ips_by_lan"]
         forwards_l3: bool = cfg.get("forwards_l3", False)
+        service_config: dict[str, dict] = cfg.get("service_config", {}) or {}
         svc_names: list[str] = decky["services"]
 
         base_key = name
@@ -92,7 +93,9 @@ def generate_topology_compose(hydrated: dict[str, Any]) -> dict:
             svc = get_service(svc_name)
             if svc is None or svc.fleet_singleton:
                 continue
-            fragment = svc.compose_fragment(name, service_cfg={})
+            fragment = svc.compose_fragment(
+                name, service_cfg=service_config.get(svc_name, {})
+            )
             if "build" in fragment:
                 fragment["build"].setdefault("args", {}).setdefault(
                     "BASE_IMAGE", _DEFAULT_BASE_IMAGE

decnet/topology/config.py

@@ -69,6 +69,10 @@ class _PlannedDecky:
     # Mapping LAN-name → assigned IP within that LAN's subnet.
     ips_by_lan: dict[str, str] = field(default_factory=dict)
     forwards_l3: bool = False  # only meaningful when present on ≥2 LANs
+    # Per-service config overrides: {service_name: {field: value}}.
+    # Mirrors ``DeckyConfig.service_config`` from the flat-fleet path;
+    # services read these via ``compose_fragment(service_cfg=...)``.
+    service_config: dict[str, dict] = field(default_factory=dict)
 
 
 @dataclass

decnet/topology/persistence.py

@@ -52,6 +52,7 @@ async def persist(repo: Any, plan: GeneratedTopology) -> str:
                     "services": decky.services,
                     "ips_by_lan": decky.ips_by_lan,
                     "forwards_l3": decky.forwards_l3,
+                    "service_config": decky.service_config,
                 },
                 "ip": primary_ip,
             }

tests/topology/test_service_config.py

@@ -0,0 +1,112 @@
+"""Per-decky, per-service config roundtrips through persist + compose."""
+from __future__ import annotations
+
+import pytest
+import yaml
+
+from decnet.topology.compose import generate_topology_compose
+from decnet.topology.config import TopologyConfig
+from decnet.topology.generator import generate
+from decnet.topology.persistence import hydrate, persist
+from decnet.web.db.factory import get_repository
+
+
+def _cfg(**kw) -> TopologyConfig:
+    base = dict(
+        name="svc",
+        depth=1,
+        branching_factor=1,
+        deckies_per_lan_min=1,
+        deckies_per_lan_max=1,
+        cross_edge_probability=0.0,
+        randomize_services=False,
+        services_explicit=["ssh"],
+        seed=5,
+    )
+    base.update(kw)
+    return TopologyConfig(**base)
+
+
+@pytest.fixture
+async def repo(tmp_path):
+    r = get_repository(db_path=str(tmp_path / "svc.db"))
+    await r.initialize()
+    return r
+
+
+@pytest.mark.anyio
+async def test_service_config_roundtrips(repo):
+    plan = generate(_cfg())
+    # Operator-style override, as the web editor would write it.
+    plan.deckies[0].service_config = {"ssh": {"password": "megapassword"}}
+    tid = await persist(repo, plan)
+
+    hydrated = await hydrate(repo, tid)
+    decky = next(
+        d for d in hydrated["deckies"] if d["name"] == plan.deckies[0].name
+    )
+    assert decky["decky_config"]["service_config"] == {
+        "ssh": {"password": "megapassword"}
+    }
+
+
+@pytest.mark.anyio
+async def test_service_config_reaches_compose_fragment(repo):
+    plan = generate(_cfg())
+    plan.deckies[0].service_config = {"ssh": {"password": "megapassword"}}
+    tid = await persist(repo, plan)
+
+    hydrated = await hydrate(repo, tid)
+    compose = generate_topology_compose(hydrated)
+    # The ssh fragment keys are "<decky>-ssh" (see compose.py:107).
+    ssh_key = f"{plan.deckies[0].name}-ssh"
+    frag = compose["services"][ssh_key]
+    env = frag.get("environment", {})
+    assert env.get("SSH_ROOT_PASSWORD") == "megapassword"
+
+
+@pytest.mark.anyio
+async def test_missing_service_config_defaults_work(repo):
+    """No service_config override → service falls back to its default."""
+    plan = generate(_cfg())
+    tid = await persist(repo, plan)
+    hydrated = await hydrate(repo, tid)
+    compose = generate_topology_compose(hydrated)
+    ssh_key = f"{plan.deckies[0].name}-ssh"
+    frag = compose["services"][ssh_key]
+    assert frag["environment"]["SSH_ROOT_PASSWORD"] == "admin"
+
+
+@pytest.mark.anyio
+async def test_unknown_nested_key_passes_through(repo):
+    """Forward-compat: unknown keys under a service reach the fragment
+    untouched (current services ignore them; future services may read)."""
+    plan = generate(_cfg())
+    plan.deckies[0].service_config = {
+        "ssh": {"password": "x", "future_flag": "hi"}
+    }
+    tid = await persist(repo, plan)
+    hydrated = await hydrate(repo, tid)
+    decky = next(
+        d for d in hydrated["deckies"] if d["name"] == plan.deckies[0].name
+    )
+    assert (
+        decky["decky_config"]["service_config"]["ssh"]["future_flag"] == "hi"
+    )
+
+
+@pytest.mark.anyio
+async def test_compose_file_yaml_is_loadable(repo):
+    """Regression: the compose dict roundtrips through yaml cleanly."""
+    plan = generate(_cfg())
+    plan.deckies[0].service_config = {"ssh": {"password": "roundtrip"}}
+    tid = await persist(repo, plan)
+    hydrated = await hydrate(repo, tid)
+    compose = generate_topology_compose(hydrated)
+    dumped = yaml.dump(compose, sort_keys=False)
+    reloaded = yaml.safe_load(dumped)
+    ssh_key = f"{plan.deckies[0].name}-ssh"
+    assert (
+        reloaded["services"][ssh_key]["environment"]["SSH_ROOT_PASSWORD"]
+        == "roundtrip"
+    )