feat(orchestrator): list_live broker op + reconcile via broker (#468)

Chunk 3 of the host-control-server stack: grow the broker op vocabulary
(PRD gap 3), starting with `list_live`, and invert `reconcile` onto it.

- broker: split the closed op vocabulary into mutation (`launch`/`teardown`,
  carry a bottle id + static flags) and query (`list_live`, carries nothing
  but its op name) kinds. `verify_request` now enforces a **strict schema**
  (open question 1, resolved yes): unknown claim keys are rejected, a mutation
  must name its bottle, and a query that smuggles any id/flag is refused.
- broker verb: `LaunchBroker.list_live` / `SubmitBroker.list_live` return the
  backend's live source IPs; a backend enumeration failure is converted to the
  single `BrokerUnavailableError` "live set unknown" signal. `DockerBroker`
  enumerates its labelled containers; `StubBroker` derives from launches (or a
  test override).
- host controller: `POST /broker/live` verifies a signed `list_live` token and
  returns `{source_ips}`; `BrokerClient.list_live` is its drop-in client.
- reconcile: `OrchestratorCore.reconcile()` drops the `live_source_ips`
  parameter and pulls the live set from the broker itself — the tell that the
  orchestrator couldn't see the backend goes away. **Fail-safe**: if the broker
  can't return an authoritative set the sweep is skipped, never run against an
  empty/partial set (which would reap healthy rows). The `/reconcile` HTTP
  contract + `OrchestratorClient.reconcile` become a bare trigger.

The macOS launcher's Apple-container enumeration stays for now; it becomes the
host controller's `list_live` when launch itself moves behind the broker (the
pulled-forward chunk 5, next in the stack).

Tests + pyright clean; pylint 10.0 on broker.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

tests/unit/test_macos_consolidated_launch.py

@@ -190,8 +190,7 @@ class TestRegisterAgentReconciles(unittest.TestCase):
 
     def _register(self, client: Mock) -> None:
         with patch(f"{_MOD}.OrchestratorClient", return_value=client), \
-                patch(f"{_UTIL}.provision_git_gate"), \
-                patch(f"{_MOD}.live_source_ips", return_value=["10.0.0.7"]):
+                patch(f"{_UTIL}.provision_git_gate"):
             register_agent(
                 _egress_plan(), _git_plan(),
                 source_ip="10.0.0.7", endpoint=_endpoint(),
@@ -213,7 +212,9 @@ class TestRegisterAgentReconciles(unittest.TestCase):
         client.register_bottle.side_effect = _register_bottle
         self._register(client)
         self.assertEqual(["reconcile", "register"], calls)
-        client.reconcile.assert_called_once_with(["10.0.0.7"])
+        # A bare trigger now — the orchestrator enumerates the live set itself
+        # (from the host controller), so the CLI passes no IPs.
+        client.reconcile.assert_called_once_with()
 
     def test_a_reconcile_failure_does_not_block_the_launch(self) -> None:
         from bot_bottle.orchestrator.client import OrchestratorClientError
@@ -221,18 +222,3 @@ class TestRegisterAgentReconciles(unittest.TestCase):
         client.reconcile.side_effect = OrchestratorClientError("unreachable")
         self._register(client)
         client.register_bottle.assert_called_once()
-
-    def test_enumeration_error_does_not_block_the_launch(self) -> None:
-        """A partial container listing must not abort the launch — skip
-        reconciliation and proceed, just as with an unreachable orchestrator."""
-        from bot_bottle.backend.macos_container.enumerate import EnumerationError
-        client = _client()
-        with patch(f"{_MOD}.OrchestratorClient", return_value=client), \
-                patch(f"{_UTIL}.provision_git_gate"), \
-                patch(f"{_MOD}.live_source_ips",
-                      side_effect=EnumerationError("container list failed")):
-            register_agent(
-                _egress_plan(), _git_plan(),
-                source_ip="10.0.0.7", endpoint=_endpoint(),
-            )
-        client.register_bottle.assert_called_once()

tests/unit/test_orchestrator_broker.py

@@ -2,6 +2,10 @@
 
 from __future__ import annotations
 
+import base64
+import hashlib
+import hmac
+import json
 import secrets
 import unittest
 
@@ -14,6 +18,21 @@ from bot_bottle.orchestrator.broker import (
 )
 
 
+def _sign_claims(claims: dict[str, object], secret: bytes) -> str:
+    """Sign an arbitrary claims dict (bypassing `sign_request`'s fixed fields)
+    so a test can craft off-schema payloads — e.g. an extra claim key, or a
+    query op carrying ids — with a *valid* signature, isolating the schema check
+    from the provenance check."""
+    def b64(data: bytes) -> str:
+        return base64.urlsafe_b64encode(data).rstrip(b"=").decode("ascii")
+    header = b64(json.dumps({"alg": "HS256", "typ": "JWT"}, sort_keys=True,
+                            separators=(",", ":")).encode())
+    payload = b64(json.dumps(claims, sort_keys=True, separators=(",", ":")).encode())
+    signing_input = f"{header}.{payload}"
+    sig = b64(hmac.new(secret, signing_input.encode("ascii"), hashlib.sha256).digest())
+    return f"{signing_input}.{sig}"
+
+
 class TestSignVerify(unittest.TestCase):
     def setUp(self) -> None:
         self.secret = secrets.token_bytes(16)
@@ -54,6 +73,41 @@ class TestSignVerify(unittest.TestCase):
         with self.assertRaises(BrokerAuthError):
             verify_request(token, self.secret)
 
+    def test_list_live_round_trip(self) -> None:
+        req = LaunchRequest(op="list_live")
+        got = verify_request(sign_request(req, self.secret), self.secret)
+        self.assertEqual("list_live", got.op)
+        self.assertEqual("", got.bottle_id)
+
+    def test_mutation_without_bottle_id_rejected(self) -> None:
+        # A launch/teardown must name its bottle even with a valid signature.
+        for op in ("launch", "teardown"):
+            token = _sign_claims(
+                {"op": op, "bottle_id": "", "jti": "x", "iat": 1}, self.secret)
+            with self.assertRaises(BrokerAuthError):
+                verify_request(token, self.secret)
+
+    def test_query_op_carrying_ids_or_flags_rejected(self) -> None:
+        # list_live is "no arguments" — a signed one that smuggles a bottle_id /
+        # source_ip / image_ref / slot is off-schema and refused.
+        for extra in (
+            {"bottle_id": "b1"}, {"source_ip": "10.0.0.1"},
+            {"image_ref": "img"}, {"slot": 2},
+        ):
+            claims: dict[str, object] = {"op": "list_live", "jti": "x", "iat": 1, **extra}
+            with self.assertRaises(BrokerAuthError):
+                verify_request(_sign_claims(claims, self.secret), self.secret)
+
+    def test_unknown_claim_key_rejected(self) -> None:
+        # Strict schema (open question 1): a well-signed token with an extra claim
+        # key is refused, so the surface can't widen past the fixed fields.
+        token = _sign_claims(
+            {"op": "launch", "bottle_id": "b1", "cmd": "rm -rf /", "jti": "x", "iat": 1},
+            self.secret,
+        )
+        with self.assertRaises(BrokerAuthError):
+            verify_request(token, self.secret)
+
 
 class TestStubBroker(unittest.TestCase):
     def setUp(self) -> None:
@@ -81,6 +135,37 @@ class TestStubBroker(unittest.TestCase):
             self.broker.submit(forged)
         self.assertEqual([], self.broker.launched)  # nothing acted on
 
+    def test_list_live_derives_from_launches(self) -> None:
+        # Default live set: launched, minus torn-down, by source IP.
+        self.broker.submit(
+            sign_request(LaunchRequest(op="launch", bottle_id="b1", source_ip="10.0.0.1"),
+                         self.secret))
+        self.broker.submit(
+            sign_request(LaunchRequest(op="launch", bottle_id="b2", source_ip="10.0.0.2"),
+                         self.secret))
+        self.broker.submit(
+            sign_request(LaunchRequest(op="teardown", bottle_id="b1", source_ip="10.0.0.1"),
+                         self.secret))
+        live = self.broker.list_live(
+            sign_request(LaunchRequest(op="list_live"), self.secret))
+        self.assertEqual(["10.0.0.2"], live)
+
+    def test_list_live_override(self) -> None:
+        self.broker.live_source_ips = ["10.0.0.9"]
+        live = self.broker.list_live(
+            sign_request(LaunchRequest(op="list_live"), self.secret))
+        self.assertEqual(["10.0.0.9"], live)
+
+    def test_submit_rejects_a_list_live_token(self) -> None:
+        # The verbs don't cross: a query token routed to submit is fail-closed.
+        with self.assertRaises(BrokerAuthError):
+            self.broker.submit(sign_request(LaunchRequest(op="list_live"), self.secret))
+
+    def test_list_live_rejects_a_mutation_token(self) -> None:
+        with self.assertRaises(BrokerAuthError):
+            self.broker.list_live(
+                sign_request(LaunchRequest(op="launch", bottle_id="b1"), self.secret))
+
 
 if __name__ == "__main__":
     unittest.main()

tests/unit/test_orchestrator_broker_client.py

@@ -113,5 +113,48 @@ class TestSubmit(unittest.TestCase):
                 self.c.submit("tok")
 
 
+class TestListLive(unittest.TestCase):
+    def setUp(self) -> None:
+        self.c = BrokerClient("http://host:8091")
+
+    def test_returns_source_ips(self) -> None:
+        with patch(_URLOPEN, return_value=_resp({"source_ips": ["10.0.0.1", "10.0.0.2"]})):
+            self.assertEqual(["10.0.0.1", "10.0.0.2"], self.c.list_live("tok"))
+
+    def test_posts_token_to_live_endpoint(self) -> None:
+        with patch(_URLOPEN, return_value=_resp({"source_ips": []})) as m:
+            self.c.list_live("signed-token")
+        request = m.call_args.args[0]
+        self.assertEqual("POST", request.get_method())
+        self.assertTrue(request.full_url.endswith("/broker/live"))
+        self.assertEqual({"token": "signed-token"}, json.loads(request.data))
+
+    def test_non_string_entries_are_filtered(self) -> None:
+        with patch(_URLOPEN, return_value=_resp({"source_ips": ["10.0.0.1", 5, None, ""]})):
+            self.assertEqual(["10.0.0.1"], self.c.list_live("tok"))
+
+    def test_missing_source_ips_raises(self) -> None:
+        # A response without the field is malformed — better to fail (reconcile
+        # skips) than to reconcile against a silent empty set.
+        with patch(_URLOPEN, return_value=_resp({})):
+            with self.assertRaises(BrokerClientError):
+                self.c.list_live("tok")
+
+    def test_401_raises_broker_auth_error(self) -> None:
+        with patch(_URLOPEN, side_effect=_http_error(401, {"error": "bad signature"})):
+            with self.assertRaises(BrokerAuthError):
+                self.c.list_live("forged")
+
+    def test_502_enumeration_failure_is_client_error(self) -> None:
+        with patch(_URLOPEN, side_effect=_http_error(502, {"error": "docker down"})):
+            with self.assertRaises(BrokerClientError):
+                self.c.list_live("tok")
+
+    def test_unreachable_is_unavailable(self) -> None:
+        with patch(_URLOPEN, side_effect=urllib.error.URLError("refused")):
+            with self.assertRaises(BrokerUnavailableError):
+                self.c.list_live("tok")
+
+
 if __name__ == "__main__":
     unittest.main()

tests/unit/test_orchestrator_client.py

@@ -148,26 +148,27 @@ class TestReconcile(unittest.TestCase):
     def setUp(self) -> None:
         self.c = OrchestratorClient("http://orch:8080")
 
-    def test_posts_live_ips_and_returns_reaped(self) -> None:
+    def test_triggers_sweep_and_returns_reaped(self) -> None:
         with patch(_URLOPEN, return_value=_resp(200, {"reaped": ["b1", "b2"]})) as m:
-            got = self.c.reconcile(["10.0.0.2", "10.0.0.3"])
+            got = self.c.reconcile()
         self.assertEqual(["b1", "b2"], got)
         sent = json.loads(m.call_args.args[0].data)
-        self.assertEqual(["10.0.0.2", "10.0.0.3"], sent["live_source_ips"])
+        # The live set is no longer sent — the orchestrator enumerates it.
+        self.assertNotIn("live_source_ips", sent)
         self.assertNotIn("grace_seconds", sent)  # omitted -> server default
 
     def test_grace_seconds_is_forwarded_when_given(self) -> None:
         with patch(_URLOPEN, return_value=_resp(200, {"reaped": []})) as m:
-            self.c.reconcile([], grace_seconds=30)
+            self.c.reconcile(grace_seconds=30)
         self.assertEqual(30, json.loads(m.call_args.args[0].data)["grace_seconds"])
 
     def test_malformed_reaped_is_tolerated(self) -> None:
         with patch(_URLOPEN, return_value=_resp(200, {"reaped": ["ok", 5, None]})):
-            self.assertEqual(["ok"], self.c.reconcile([]))
+            self.assertEqual(["ok"], self.c.reconcile())
         with patch(_URLOPEN, return_value=_resp(200, {})):
-            self.assertEqual([], self.c.reconcile([]))
+            self.assertEqual([], self.c.reconcile())
 
     def test_error_status_raises(self) -> None:
         with patch(_URLOPEN, side_effect=_http_error(500)):
             with self.assertRaises(OrchestratorClientError):
-                self.c.reconcile([])
+                self.c.reconcile()

tests/unit/test_orchestrator_docker_broker.py

@@ -8,6 +8,7 @@ from unittest.mock import Mock, patch
 
 from bot_bottle.orchestrator.broker import (
     BrokerAuthError,
+    BrokerUnavailableError,
     LaunchRequest,
     sign_request,
 )
@@ -96,5 +97,50 @@ class TestDockerBroker(unittest.TestCase):
             m.assert_not_called()
 
 
+class TestDockerBrokerListLive(unittest.TestCase):
+    def setUp(self) -> None:
+        self.secret = secrets.token_bytes(16)
+        self.broker = DockerBroker(self.secret)
+
+    def test_enumerates_labeled_container_ips(self) -> None:
+        # `docker ps` lists two labeled containers; each inspect yields an IP.
+        ps = Mock(returncode=0, stdout="c1\nc2\n", stderr="")
+        i1 = Mock(returncode=0, stdout="10.0.0.1 \n", stderr="")
+        i2 = Mock(returncode=0, stdout="10.0.0.2 \n", stderr="")
+        with patch.object(self.broker, "_docker", side_effect=[ps, i1, i2]) as m:
+            got = self.broker._list_live()
+        self.assertEqual(["10.0.0.1", "10.0.0.2"], got)
+        # First call is the label-filtered `docker ps`.
+        ps_argv = m.call_args_list[0].args[0]
+        self.assertEqual(["docker", "ps", "--filter", f"label={BOTTLE_ID_LABEL}"], ps_argv[:4])
+
+    def test_no_containers_is_empty(self) -> None:
+        with patch.object(self.broker, "_docker",
+                          return_value=Mock(returncode=0, stdout="", stderr="")):
+            self.assertEqual([], self.broker._list_live())
+
+    def test_ps_failure_raises(self) -> None:
+        with patch.object(self.broker, "_docker",
+                          return_value=Mock(returncode=1, stdout="", stderr="daemon down")):
+            with self.assertRaises(DockerBrokerError):
+                self.broker._list_live()
+
+    def test_inspect_failure_raises(self) -> None:
+        ps = Mock(returncode=0, stdout="c1\n", stderr="")
+        bad = Mock(returncode=1, stdout="", stderr="no such container")
+        with patch.object(self.broker, "_docker", side_effect=[ps, bad]):
+            with self.assertRaises(DockerBrokerError):
+                self.broker._list_live()
+
+    def test_enumeration_failure_surfaces_as_unavailable_via_list_live(self) -> None:
+        # Through the verb, a backend failure is the single "live set unknown"
+        # signal (BrokerUnavailableError) reconcile catches to skip the sweep.
+        token = sign_request(LaunchRequest(op="list_live"), self.secret)
+        with patch.object(self.broker, "_docker",
+                          return_value=Mock(returncode=1, stdout="", stderr="down")):
+            with self.assertRaises(BrokerUnavailableError):
+                self.broker.list_live(token)
+
+
 if __name__ == "__main__":
     unittest.main()

tests/unit/test_orchestrator_host_server.py

@@ -43,8 +43,9 @@ def _body(obj: object) -> bytes:
 
 
 class _RaisingBroker(LaunchBroker):
-    """A broker whose backend launch always fails — exercises the 502 path (an
-    operational backend failure, distinct from a fail-closed provenance 401)."""
+    """A broker whose backend launch/enumeration always fails — exercises the
+    502 path (an operational backend failure, distinct from a fail-closed
+    provenance 401)."""
 
     def _launch(self, req: LaunchRequest) -> None:
         raise RuntimeError("docker down")
@@ -52,6 +53,9 @@ class _RaisingBroker(LaunchBroker):
     def _teardown(self, req: LaunchRequest) -> None:
         raise RuntimeError("docker down")
 
+    def _list_live(self) -> list[str]:
+        raise RuntimeError("docker down")
+
 
 class TestDispatch(unittest.TestCase):
     def setUp(self) -> None:
@@ -100,6 +104,42 @@ class TestDispatch(unittest.TestCase):
         self.assertEqual(502, status)
         self.assertIn("backend launch failed", str(payload["error"]))
 
+    def test_broker_live_returns_source_ips(self) -> None:
+        # Two launched bottles -> the stub reports both as live.
+        self.broker.submit(self._token(op="launch", bottle_id="b1", source_ip="10.0.0.1"))
+        self.broker.submit(self._token(op="launch", bottle_id="b2", source_ip="10.0.0.2"))
+        token = self._token(op="list_live")
+        status, payload = dispatch(self.broker, "POST", "/broker/live", _body({"token": token}))
+        self.assertEqual(200, status)
+        self.assertEqual(
+            ["10.0.0.1", "10.0.0.2"],
+            sorted(typing.cast("list[str]", payload["source_ips"])),
+        )
+
+    def test_broker_live_forged_token_is_401(self) -> None:
+        forged = sign_request(LaunchRequest(op="list_live"), secrets.token_bytes(16))
+        status, payload = dispatch(self.broker, "POST", "/broker/live", _body({"token": forged}))
+        self.assertEqual(401, status)
+        self.assertIn("broker auth failed", str(payload["error"]))
+
+    def test_broker_live_rejects_a_mutation_token(self) -> None:
+        # A launch token routed to the query endpoint is a fail-closed 401 — the
+        # endpoints don't share a schema even though they share a secret.
+        token = self._token(op="launch", bottle_id="b1", image_ref="img")
+        status, _ = dispatch(self.broker, "POST", "/broker/live", _body({"token": token}))
+        self.assertEqual(401, status)
+
+    def test_broker_live_backend_failure_is_502(self) -> None:
+        broker = _RaisingBroker(self.secret)
+        token = self._token(op="list_live")
+        status, payload = dispatch(broker, "POST", "/broker/live", _body({"token": token}))
+        self.assertEqual(502, status)
+        self.assertIn("backend enumeration failed", str(payload["error"]))
+
+    def test_broker_live_missing_token_is_400(self) -> None:
+        status, _ = dispatch(self.broker, "POST", "/broker/live", _body({}))
+        self.assertEqual(400, status)
+
     def test_missing_token_is_400(self) -> None:
         status, _ = dispatch(self.broker, "POST", "/broker", _body({}))
         self.assertEqual(400, status)
@@ -190,6 +230,14 @@ class TestSeamRoundTrip(unittest.TestCase):
             client.submit(forged)
         self.assertEqual([], broker.launched)  # fail-closed across the wire
 
+    def test_list_live_enumerates_over_http(self) -> None:
+        secret = secrets.token_bytes(16)
+        broker = StubBroker(secret)
+        broker.live_source_ips = ["10.0.0.7", "10.0.0.8"]
+        client = self._serve(broker)
+        got = client.list_live(sign_request(LaunchRequest(op="list_live"), secret))
+        self.assertEqual(["10.0.0.7", "10.0.0.8"], got)
+
 
 class TestRequestLimits(unittest.TestCase):
     """The privileged listener must not let a caller that can merely reach the

tests/unit/test_orchestrator_server.py

@@ -604,13 +604,18 @@ if __name__ == "__main__":
 
 
 class TestReconcileRoute(unittest.TestCase):
-    """`POST /reconcile` — the host tells the orchestrator which bottles are
-    actually up, since the orchestrator can't see the backend from inside the
-    infra container."""
+    """`POST /reconcile` — a bare self-heal trigger. The orchestrator pulls its
+    own live set from the broker (`list_live`); the caller no longer supplies it,
+    so the request body carries at most `grace_seconds`."""
 
     def setUp(self) -> None:
         self._tmp = tempfile.TemporaryDirectory()
-        self.orch = _orchestrator(Path(self._tmp.name) / "r.db")
+        store = RegistryStore(Path(self._tmp.name) / "r.db")
+        store.migrate()
+        secret = secrets.token_bytes(16)
+        # Hold a typed StubBroker so tests can set its live set directly.
+        self.broker = StubBroker(secret)
+        self.orch = OrchestratorCore(store, self.broker, secret)
 
     def tearDown(self) -> None:
         self._tmp.cleanup()
@@ -627,30 +632,26 @@ class TestReconcileRoute(unittest.TestCase):
     def test_reaps_absent_and_reports_ids(self) -> None:
         dead = self._old("10.0.0.1")
         alive = self._old("10.0.0.2")
-        status, payload = dispatch(
-            self.orch, "POST", "/reconcile", _body({"live_source_ips": ["10.0.0.2"]}))
+        self.broker.live_source_ips = ["10.0.0.2"]  # broker: only .2 is up
+        status, payload = dispatch(self.orch, "POST", "/reconcile", _body({}))
         self.assertEqual(200, status)
         self.assertEqual([dead], payload["reaped"])
         self.assertIsNone(self.orch.registry.get(dead))
         self.assertIsNotNone(self.orch.registry.get(alive))
 
-    def test_missing_live_source_ips_is_400(self) -> None:
-        status, _ = dispatch(self.orch, "POST", "/reconcile", _body({}))
-        self.assertEqual(400, status)
+    def test_empty_body_triggers_the_sweep(self) -> None:
+        """No body is fine — the live set comes from the broker, not the body."""
+        dead = self._old("10.0.0.1")
+        self.broker.live_source_ips = []  # broker: nothing running
+        status, payload = dispatch(self.orch, "POST", "/reconcile", b"")
+        self.assertEqual(200, status)
+        self.assertEqual([dead], payload["reaped"])
 
     def test_grace_seconds_is_honoured(self) -> None:
         """A grace window wide enough to cover the row protects it."""
         self.orch.registry.register("10.0.0.3")
+        self.broker.live_source_ips = []
         status, payload = dispatch(
-            self.orch, "POST", "/reconcile",
-            _body({"live_source_ips": [], "grace_seconds": 3600}))
+            self.orch, "POST", "/reconcile", _body({"grace_seconds": 3600}))
         self.assertEqual(200, status)
         self.assertEqual([], payload["reaped"])
-
-    def test_non_string_entries_are_ignored(self) -> None:
-        dead = self._old("10.0.0.4")
-        status, payload = dispatch(
-            self.orch, "POST", "/reconcile",
-            _body({"live_source_ips": [None, 7, "10.0.0.9"]}))
-        self.assertEqual(200, status)
-        self.assertEqual([dead], payload["reaped"])

tests/unit/test_orchestrator_service.py

@@ -39,6 +39,9 @@ class _FailingBroker(LaunchBroker):
     def _teardown(self, req: LaunchRequest) -> None:
         pass
 
+    def _list_live(self) -> list[str]:
+        return []
+
 
 class _UnavailableBroker(LaunchBroker):
     """Verifies the token, then raises the *ambiguous* BrokerUnavailableError —
@@ -51,6 +54,9 @@ class _UnavailableBroker(LaunchBroker):
     def _teardown(self, req: LaunchRequest) -> None:
         pass
 
+    def _list_live(self) -> list[str]:
+        return []
+
 
 class TestOrchestrator(unittest.TestCase):
     def setUp(self) -> None:
@@ -357,8 +363,9 @@ class TestOrchestratorSupervise(unittest.TestCase):
         pid = self.orch.supervise_queue_proposal(
             rec.bottle_id, tool=TOOL_EGRESS_ALLOW,
             proposed_file="routes:\n  - host: google.com\n", justification="j")
-        # No live source IPs -> the bottle is reaped (grace 0 so it's immediate).
-        self.assertEqual([rec.bottle_id], self.orch.reconcile([], grace_seconds=0))
+        # No live source IPs (nothing launched through the stub) -> the bottle
+        # is reaped (grace 0 so it's immediate).
+        self.assertEqual([rec.bottle_id], self.orch.reconcile(grace_seconds=0))
         self.assertEqual(
             {"status": "unknown"}, self.orch.supervise_poll_response(rec.bottle_id, pid))
 
@@ -403,7 +410,9 @@ class TestOrchestratorReconcile(unittest.TestCase):
         live = self.orch.launch_bottle("10.243.0.2", tokens={"EGRESS_TOKEN_0": "keep"})
         self._age_all(600)
 
-        self.assertEqual([dead.bottle_id], self.orch.reconcile(["10.243.0.2"]))
+        # The broker reports only .2 as still running -> .1 is reaped.
+        self.broker.live_source_ips = ["10.243.0.2"]
+        self.assertEqual([dead.bottle_id], self.orch.reconcile())
         self.assertIsNone(self.store.get(dead.bottle_id))
         self.assertIsNotNone(self.store.get(live.bottle_id))
         # The in-memory egress credential goes with the row.
@@ -415,14 +424,31 @@ class TestOrchestratorReconcile(unittest.TestCase):
         broker error must not stop the sweep clearing the row."""
         self.orch.launch_bottle("10.243.0.1")
         self._age_all(600)
-        self.broker.launched.clear()
-        self.orch.reconcile([])
+        self.broker.live_source_ips = []  # broker reports nothing running
+        self.orch.reconcile()
         self.assertEqual([], self.broker.torn_down)
 
     def test_reconcile_keeps_everything_when_all_are_live(self) -> None:
         a = self.orch.launch_bottle("10.243.0.1")
         b = self.orch.launch_bottle("10.243.0.2")
         self._age_all(600)
-        self.assertEqual([], self.orch.reconcile(["10.243.0.1", "10.243.0.2"]))
+        self.broker.live_source_ips = ["10.243.0.1", "10.243.0.2"]
+        self.assertEqual([], self.orch.reconcile())
+        self.assertIsNotNone(self.store.get(a.bottle_id))
+        self.assertIsNotNone(self.store.get(b.bottle_id))
+
+    def test_reconcile_skipped_when_broker_cannot_enumerate(self) -> None:
+        """A broker that can't return an authoritative live set must NOT be
+        treated as "nothing is live" — that would reap every healthy bottle.
+        The sweep is skipped instead."""
+        a = self.orch.launch_bottle("10.243.0.1")
+        b = self.orch.launch_bottle("10.243.0.2")
+        self._age_all(600)
+
+        def _boom() -> list[str]:
+            raise RuntimeError("docker ps failed")
+
+        self.broker._list_live = _boom  # type: ignore[method-assign]
+        self.assertEqual([], self.orch.reconcile())
         self.assertIsNotNone(self.store.get(a.bottle_id))
         self.assertIsNotNone(self.store.get(b.bottle_id))