fix(workers): advertise the port the control plane actually bound

An enrollment token carries the endpoint a worker will dial, but
default_endpoint() read the CONFIGURED port rather than the bound one. Start
on any other port and every token points somewhere nothing is listening —
the worker retries forever against a dead address with backoff, so it looks
like a network problem rather than a wrong number.

Found by running the feature end to end on a non-default port, which is also
the second bug in this seam: the first was advertising a .local hostname
gRPC's resolver cannot resolve. Both were about what the token tells a
worker to dial, so both now have regression tests.

backend/worker/service.py

@@ -93,6 +93,10 @@ class ControlPlane:
         self._server = None
         self._tasks: list[asyncio.Task] = []
         self._started = False
+        # The port we actually bound, which is not necessarily the configured
+        # one — an enrollment token carries this, so advertising the config
+        # value instead hands workers an endpoint nothing is listening on.
+        self._port: Optional[int] = None
 
     @property
     def running(self) -> bool:
@@ -129,9 +133,10 @@ class ControlPlane:
             artifact_dir=locations["artifacts"],
             cert_fingerprint=self.credentials.fingerprint,
         )
+        self._port = port or control_port()
         self._server = await serve(
             self.servicer,
-            port=port or control_port(),
+            port=self._port,
             certificate_pem=self.credentials.certificate_pem,
             private_key_pem=self.credentials.private_key_pem,
         )
@@ -140,7 +145,7 @@ class ControlPlane:
             asyncio.create_task(self._dispatch_loop(), name="worker-dispatch"),
         ]
         self._started = True
-        logger.info("Remote worker control plane started on port %d", port or control_port())
+        logger.info("Remote worker control plane started on port %d", self._port)
 
     async def stop(self) -> None:
         for task in self._tasks:
@@ -153,6 +158,7 @@ class ControlPlane:
             # would delay app shutdown for work that survives anyway.
             await self._server.stop(grace=2.0)
             self._server = None
+        self._port = None
         self._started = False
 
     async def _sweep_loop(self) -> None:
@@ -226,7 +232,7 @@ class ControlPlane:
             or tls.primary_ip()
             or "127.0.0.1"
         )
-        return f"{host}:{control_port()}"
+        return f"{host}:{self._port or control_port()}"
 
     def snapshot(self, *, now: Optional[float] = None) -> dict:
         """Everything the workers UI needs in one call."""

tests/test_worker_service_api.py

@@ -301,3 +301,37 @@ def test_consent_is_recorded_explicitly(client, db):
 def test_task_listing_is_empty_when_stopped(client):
     body = client.get("/workers/tasks").json()
     assert body == {"tasks": [], "queue_depth": 0}
+
+
+@pytest.mark.asyncio
+async def test_enrollment_advertises_the_port_actually_bound(db, monkeypatch, tmp_path):
+    """A token carries the endpoint a worker will dial. Advertising the
+    configured port while listening on another hands workers an address
+    nothing answers on — found by running the thing on a non-default port.
+    """
+    monkeypatch.setattr(
+        service,
+        "paths",
+        lambda: {
+            "root": str(tmp_path),
+            "certificate": str(tmp_path / "cp.crt"),
+            "private_key": str(tmp_path / "cp.key"),
+            "worker_key": str(tmp_path / "w.key"),
+            "artifacts": str(tmp_path / "artifacts"),
+        },
+    )
+    monkeypatch.delenv("OMNIVOICE_WORKER_PORT", raising=False)
+    monkeypatch.delenv("OMNIVOICE_WORKER_ENDPOINT_HOST", raising=False)
+
+    plane = service.ControlPlane()
+    await plane.start(port=7601)
+    try:
+        assert plane.default_endpoint().endswith(":7601")
+        assert plane.create_enrollment().endpoint.endswith(":7601")
+    finally:
+        await plane.stop()
+
+
+def test_endpoint_falls_back_to_the_configured_port_when_stopped(monkeypatch):
+    monkeypatch.delenv("OMNIVOICE_WORKER_PORT", raising=False)
+    assert service.ControlPlane().default_endpoint().endswith(f":{service.DEFAULT_PORT}")