feat(canary): allow custom canaries on MazeNET deckies via API

POST /api/v1/canary/tokens grows an optional topology_id field.  When
present, the server hydrates the topology, validates the named decky is
in it, and resolves the docker container via
planter.resolve_topology_container — <name>-ssh if the decky exposes ssh,
else the topology base container.  Absent ⇒ fleet semantics, unchanged.

The token row gets a nullable topology_id column (no migration helper
per pre-v1 policy).  GET /api/v1/canary/tokens accepts ?topology_id= as
a filter.  DELETE re-resolves the container at revoke time so a
redeployed topology is still reachable.

422 when the named decky isn't in the topology; 404 when the topology
itself doesn't exist.

decnet/web/db/models/canary.py

@@ -100,6 +100,12 @@ class CanaryToken(SQLModel, table=True):
     uuid: str = Field(default_factory=lambda: str(uuid4()), primary_key=True)
     kind: str = Field(index=True)  # CanaryKind literal at the API layer
     decky_name: str = Field(index=True)  # FleetDecky.name; no FK (composite PK)
+    # When NULL, the token is on a fleet decky (decky_name resolves to
+    # ``<name>-ssh``).  When set, it points at a MazeNET topology — the
+    # planter resolves the container via :func:`resolve_topology_container`.
+    # No FK: topologies are mutable and we don't want a row to vanish on
+    # cascade; the row is the historical record of placement.
+    topology_id: Optional[str] = Field(default=None, index=True)
     blob_uuid: Optional[str] = Field(
         default=None, foreign_key="canary_blobs.uuid", index=True,
     )
@@ -188,6 +194,10 @@ class CanaryTokenCreateRequest(BaseModel):
     router so the 400 carries a clear detail message.
     """
     decky_name: str = PydanticField(..., min_length=1)
+    # When set, ``decky_name`` is interpreted as a MazeNET topology decky
+    # name; the server validates membership and resolves the container
+    # accordingly.  Absent ⇒ fleet semantics (today's behavior).
+    topology_id: Optional[str] = None
     kind: CanaryKind
     placement_path: str = PydanticField(..., min_length=1)
     blob_uuid: Optional[str] = None
@@ -202,6 +212,7 @@ class CanaryTokenResponse(BaseModel):
     uuid: str
     kind: CanaryKind
     decky_name: str
+    topology_id: Optional[str] = None
     blob_uuid: Optional[str]
     instrumenter: Optional[str]
     generator: Optional[str]

decnet/web/db/repository.py

@@ -936,6 +936,7 @@ class BaseRepository(ABC):
         decky_name: Optional[str] = None,
         state: Optional[str] = None,
         kind: Optional[str] = None,
+        topology_id: Optional[str] = None,
     ) -> list[dict[str, Any]]:
         raise NotImplementedError
 

decnet/web/db/sqlmodel_repo/canary.py

@@ -122,6 +122,7 @@ class CanaryMixin:
         decky_name: Optional[str] = None,
         state: Optional[str] = None,
         kind: Optional[str] = None,
+        topology_id: Optional[str] = None,
     ) -> list[dict[str, Any]]:
         async with self._session() as session:
             stmt = select(CanaryToken)
@@ -131,6 +132,8 @@ class CanaryMixin:
                 stmt = stmt.where(CanaryToken.state == state)
             if kind is not None:
                 stmt = stmt.where(CanaryToken.kind == kind)
+            if topology_id is not None:
+                stmt = stmt.where(CanaryToken.topology_id == topology_id)
             stmt = stmt.order_by(desc(CanaryToken.placed_at))
             result = await session.execute(stmt)
             return [r.model_dump(mode="json") for r in result.scalars().all()]

decnet/web/router/canary/api_tokens.py

@@ -61,6 +61,33 @@ def _row_to_response(row: dict[str, Any]) -> CanaryTokenResponse:
     return CanaryTokenResponse(**row)
 
 
+async def _resolve_topology_target(
+    topology_id: str, decky_name: str,
+) -> str:
+    """Validate (topology_id, decky_name) and return the docker container.
+
+    404 if the topology doesn't exist; 422 if the named decky isn't in it.
+    Hoisted into ``decky_io/resolve.py`` in workstream 2 so the file-drop
+    endpoint can share it; for now it's local to the canary router.
+    """
+    from decnet.topology.persistence import hydrate
+    hydrated = await hydrate(repo, topology_id)
+    if hydrated is None:
+        raise HTTPException(status_code=404, detail="topology not found")
+    for decky in hydrated["deckies"]:
+        cfg = decky.get("decky_config") or {}
+        name = cfg.get("name") or decky.get("name")
+        if name == decky_name:
+            services = decky.get("services") or []
+            return planter.resolve_topology_container(
+                topology_id, decky_name, services,
+            )
+    raise HTTPException(
+        status_code=422,
+        detail=f"decky {decky_name!r} is not in topology {topology_id!r}",
+    )
+
+
 def _trigger_row_to_response(row: dict[str, Any]) -> CanaryTriggerResponse:
     # Decode raw_headers JSON for the response shape.
     headers = row.get("raw_headers") or "{}"
@@ -105,6 +132,14 @@ async def api_create_token(
     except ValueError as e:
         raise HTTPException(status_code=400, detail=str(e)) from e
 
+    # Resolve the docker container before any expensive work — surfacing
+    # 404/422 here keeps a typo from minting a half-baked token row.
+    container: str | None = None
+    if req.topology_id:
+        container = await _resolve_topology_target(
+            req.topology_id, req.decky_name,
+        )
+
     slug = token_urlsafe(16)
     ctx = CanaryContext(
         callback_token=slug, http_base=_http_base(), dns_zone=_dns_zone(),
@@ -145,6 +180,7 @@ async def api_create_token(
         "uuid": token_uuid,
         "kind": kind,
         "decky_name": req.decky_name,
+        "topology_id": req.topology_id,
         "blob_uuid": req.blob_uuid,
         "instrumenter": instrumenter_name,
         "generator": req.generator,
@@ -154,7 +190,10 @@ async def api_create_token(
         "created_by": admin.get("uuid", "unknown"),
         "state": "planted",
     })
-    await planter.plant(req.decky_name, artifact, token_uuid=token_uuid, repo=repo)
+    await planter.plant(
+        req.decky_name, artifact,
+        token_uuid=token_uuid, repo=repo, container=container,
+    )
     row = await repo.get_canary_token(token_uuid)
     return _row_to_response(row)
 
@@ -173,10 +212,12 @@ async def api_list_tokens(
     decky_name: str | None = Query(default=None),
     state: str | None = Query(default=None),
     kind: str | None = Query(default=None),
+    topology_id: str | None = Query(default=None),
     viewer: dict = Depends(require_viewer),
 ) -> CanaryTokensResponse:
     rows = await repo.list_canary_tokens(
         decky_name=decky_name, state=state, kind=kind,
+        topology_id=topology_id,
     )
     return CanaryTokensResponse(
         tokens=[_row_to_response(r) for r in rows],
@@ -311,8 +352,21 @@ async def api_revoke_token(
     row = await repo.get_canary_token(uuid)
     if row is None:
         raise HTTPException(status_code=404, detail="token not found")
+    # Re-resolve the container at revoke time: the topology may have
+    # been redeployed since placement.  If it's gone entirely we fall
+    # through to the planter's fleet default — the call will fail
+    # best-effort and the row still flips to revoked.
+    container: str | None = None
+    topology_id = row.get("topology_id")
+    if topology_id:
+        try:
+            container = await _resolve_topology_target(
+                topology_id, row["decky_name"],
+            )
+        except HTTPException:
+            container = None
     await planter.revoke(
         row["decky_name"], row["placement_path"],
-        token_uuid=uuid, repo=repo,
+        token_uuid=uuid, repo=repo, container=container,
     )
     return MessageResponse(message="ok")

tests/api/canary/test_canary_tokens_api.py

@@ -344,6 +344,231 @@ async def test_triggers_list_for_token(
     assert res.status_code == 404
 
 
+# ---------------- topology (MazeNET) deckies ------------------------------
+
+
+def _patch_subprocess_capture():
+    """Subprocess patcher that records argv for assertion in tests."""
+    captured: list[list[str]] = []
+
+    async def _fake(*argv, **kw):  # noqa: ANN001
+        captured.append(list(argv))
+        return _FakeProc(rc=0)
+
+    return patch.object(asyncio, "create_subprocess_exec", _fake), captured
+
+
+def _hydrate_returning(deckies: list[dict]):
+    async def _fake_hydrate(_repo, _topology_id):
+        return {
+            "topology": {"id": _topology_id},
+            "lans": [],
+            "deckies": deckies,
+            "edges": [],
+        }
+    return _fake_hydrate
+
+
+@pytest.mark.asyncio
+async def test_create_token_on_topology_decky_with_ssh_resolves_ssh_container(
+    client: httpx.AsyncClient, auth_token: str, monkeypatch
+) -> None:
+    monkeypatch.setenv("DECNET_CANARY_HTTP_BASE", "https://canary.test")
+    topo_id = "abcdef0123456789"
+    monkeypatch.setattr(
+        "decnet.topology.persistence.hydrate",
+        _hydrate_returning([{
+            "uuid": "u1", "name": "web1",
+            "decky_config": {"name": "web1"},
+            "services": ["ssh", "http"],
+        }]),
+    )
+    patcher, captured = _patch_subprocess_capture()
+    with patcher:
+        res = await client.post(
+            f"{_BASE}/tokens",
+            json={
+                "decky_name": "web1",
+                "topology_id": topo_id,
+                "kind": "http",
+                "placement_path": "/etc/canary.env",
+                "generator": "env_file",
+            },
+            headers=_hdr(auth_token),
+        )
+    assert res.status_code == 201, res.text
+    body = res.json()
+    assert body["topology_id"] == topo_id
+    # docker exec -i <container> sh -c <script>
+    assert captured and captured[0][3] == "web1-ssh"
+
+
+@pytest.mark.asyncio
+async def test_create_token_on_topology_decky_without_ssh_uses_base_container(
+    client: httpx.AsyncClient, auth_token: str, monkeypatch
+) -> None:
+    monkeypatch.setenv("DECNET_CANARY_HTTP_BASE", "https://canary.test")
+    topo_id = "fedcba9876543210"
+    monkeypatch.setattr(
+        "decnet.topology.persistence.hydrate",
+        _hydrate_returning([{
+            "uuid": "u1", "name": "router",
+            "decky_config": {"name": "router"},
+            "services": ["dns"],
+        }]),
+    )
+    patcher, captured = _patch_subprocess_capture()
+    with patcher:
+        res = await client.post(
+            f"{_BASE}/tokens",
+            json={
+                "decky_name": "router",
+                "topology_id": topo_id,
+                "kind": "http",
+                "placement_path": "/etc/canary.env",
+                "generator": "env_file",
+            },
+            headers=_hdr(auth_token),
+        )
+    assert res.status_code == 201, res.text
+    assert captured[0][3] == "decnet_t_fedcba98_router"
+
+
+@pytest.mark.asyncio
+async def test_create_token_404_when_topology_unknown(
+    client: httpx.AsyncClient, auth_token: str, monkeypatch
+) -> None:
+    async def _no_topology(_repo, _topology_id):
+        return None
+    monkeypatch.setattr(
+        "decnet.topology.persistence.hydrate", _no_topology,
+    )
+    res = await client.post(
+        f"{_BASE}/tokens",
+        json={
+            "decky_name": "web1",
+            "topology_id": "ghost",
+            "kind": "http",
+            "placement_path": "/x.env",
+            "generator": "env_file",
+        },
+        headers=_hdr(auth_token),
+    )
+    assert res.status_code == 404
+
+
+@pytest.mark.asyncio
+async def test_create_token_422_when_decky_not_in_topology(
+    client: httpx.AsyncClient, auth_token: str, monkeypatch
+) -> None:
+    monkeypatch.setattr(
+        "decnet.topology.persistence.hydrate",
+        _hydrate_returning([{
+            "uuid": "u1", "name": "other",
+            "decky_config": {"name": "other"},
+            "services": [],
+        }]),
+    )
+    res = await client.post(
+        f"{_BASE}/tokens",
+        json={
+            "decky_name": "web1",
+            "topology_id": "abcdef0123456789",
+            "kind": "http",
+            "placement_path": "/x.env",
+            "generator": "env_file",
+        },
+        headers=_hdr(auth_token),
+    )
+    assert res.status_code == 422
+
+
+@pytest.mark.asyncio
+async def test_revoke_token_re_resolves_container_from_topology_id(
+    client: httpx.AsyncClient, auth_token: str, monkeypatch
+) -> None:
+    monkeypatch.setenv("DECNET_CANARY_HTTP_BASE", "https://canary.test")
+    topo_id = "11112222333344445555"
+    monkeypatch.setattr(
+        "decnet.topology.persistence.hydrate",
+        _hydrate_returning([{
+            "uuid": "u1", "name": "router",
+            "decky_config": {"name": "router"},
+            "services": [],
+        }]),
+    )
+    # Create the token on a topology decky.
+    create_patcher, _ = _patch_subprocess_capture()
+    with create_patcher:
+        res = await client.post(
+            f"{_BASE}/tokens",
+            json={
+                "decky_name": "router",
+                "topology_id": topo_id,
+                "kind": "http",
+                "placement_path": "/x.env",
+                "generator": "env_file",
+            },
+            headers=_hdr(auth_token),
+        )
+    assert res.status_code == 201
+    token = res.json()
+    # Revoke and assert the captured argv targets the topology base
+    # container, not <name>-ssh.
+    revoke_patcher, captured = _patch_subprocess_capture()
+    with revoke_patcher:
+        rev = await client.delete(
+            f"{_BASE}/tokens/{token['uuid']}", headers=_hdr(auth_token),
+        )
+    assert rev.status_code == 200, rev.text
+    assert captured and captured[0][2] == "decnet_t_11112222_router"
+
+
+@pytest.mark.asyncio
+async def test_list_tokens_filters_by_topology_id(
+    client: httpx.AsyncClient, auth_token: str, monkeypatch
+) -> None:
+    monkeypatch.setenv("DECNET_CANARY_HTTP_BASE", "https://canary.test")
+    topo_id = "topotopotopotopo"
+    monkeypatch.setattr(
+        "decnet.topology.persistence.hydrate",
+        _hydrate_returning([{
+            "uuid": "u1", "name": "web1",
+            "decky_config": {"name": "web1"},
+            "services": ["ssh"],
+        }]),
+    )
+    # Create one fleet token (no topology_id) and one topology token.
+    with _patch_subprocess(rc=0):
+        await client.post(
+            f"{_BASE}/tokens",
+            json={
+                "decky_name": "fleet1", "kind": "http",
+                "placement_path": "/etc/a.env", "generator": "env_file",
+            },
+            headers=_hdr(auth_token),
+        )
+        await client.post(
+            f"{_BASE}/tokens",
+            json={
+                "decky_name": "web1", "topology_id": topo_id,
+                "kind": "http", "placement_path": "/etc/b.env",
+                "generator": "env_file",
+            },
+            headers=_hdr(auth_token),
+        )
+    # Filter to topology tokens only.
+    res = await client.get(
+        f"{_BASE}/tokens", params={"topology_id": topo_id},
+        headers=_hdr(auth_token),
+    )
+    assert res.status_code == 200
+    body = res.json()
+    decky_names = {t["decky_name"] for t in body["tokens"]}
+    assert decky_names == {"web1"}
+    assert all(t["topology_id"] == topo_id for t in body["tokens"])
+
+
 # ---------------- auth ----------------------------------------------------