fix(orchestrator): reap registry rows whose bottle is no longer running

A registry row only ever left the registry two ways: an explicit
teardown_bottle (the launcher's cleanup callback) or the same-IP supersede
sweep in register(). Neither runs when the launching CLI dies hard, so the
row outlives its container.

That orphan is not inert. Source IPs are recycled by the backend's DHCP and
by_source_ip fail-closes on ambiguity, so a leftover row at a reused address
resolves no policy at all for the next bottle that lands there — and a
bottle with no policy denies every host, which surfaces to the agent as
"host X is not in the allowlist" for hosts that were never the problem.

Add reap_absent/reconcile and call it from the macOS launch path before
registering, so each launch self-heals the registry. Restores the invariant
the data plane needs: at most one active row per live address, and none for
a dead one. The second half matters as much as the first — when several rows
claim a *live* address the newest wins and the rest are swept, otherwise a
recycled address stays ambiguous, which is exactly the bricked state.

The host supplies the live set because the orchestrator runs inside the
infra container and cannot see the backend. A grace window exempts rows
younger than it, so reconciliation cannot race a bottle still coming up, and
a reconcile failure is logged rather than blocking an otherwise-fine launch.

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

tests/unit/test_macos_consolidated_launch.py

@@ -134,3 +134,69 @@ class TestTeardown(unittest.TestCase):
 
 if __name__ == "__main__":
     unittest.main()
+
+
+class TestLiveSourceIps(unittest.TestCase):
+    """The reconciliation input: the host enumerates its own bottles because
+    the orchestrator, inside the infra container, cannot see the backend."""
+
+    def _agents(self, *slugs: str) -> list[Mock]:
+        return [Mock(slug=s) for s in slugs]
+
+    def test_maps_slugs_to_container_addresses(self) -> None:
+        from bot_bottle.backend.macos_container.consolidated_launch import live_source_ips
+        with patch(f"{_MOD}.enumerate_active", return_value=self._agents("a", "b")), \
+                patch(f"{_MOD}.container_mod.try_container_ipv4_on_network",
+                      side_effect=["10.0.0.1", "10.0.0.2"]) as ip:
+            got = live_source_ips("net0")
+        self.assertEqual(["10.0.0.1", "10.0.0.2"], got)
+        self.assertEqual("bot-bottle-a", ip.call_args_list[0].args[0])
+
+    def test_containers_without_an_address_are_skipped(self) -> None:
+        """A container that hasn't been given a DHCP address yet contributes
+        nothing — the reap's grace window, not this list, protects it."""
+        from bot_bottle.backend.macos_container.consolidated_launch import live_source_ips
+        with patch(f"{_MOD}.enumerate_active", return_value=self._agents("a", "b")), \
+                patch(f"{_MOD}.container_mod.try_container_ipv4_on_network",
+                      side_effect=["", "10.0.0.2"]):
+            self.assertEqual(["10.0.0.2"], live_source_ips("net0"))
+
+
+class TestRegisterAgentReconciles(unittest.TestCase):
+    """Registration self-heals the registry first: an orphan row at a recycled
+    address makes attribution ambiguous, which resolves no policy at all and
+    denies every host for the bottle being launched."""
+
+    def _register(self, client: Mock) -> None:
+        with patch(f"{_MOD}.OrchestratorClient", return_value=client), \
+                patch(f"{_MOD}.provision_git_gate"), \
+                patch(f"{_MOD}.live_source_ips", return_value=["10.0.0.7"]):
+            register_agent(
+                _egress_plan(), _git_plan(),
+                source_ip="10.0.0.7", endpoint=_endpoint(),
+            )
+
+    def test_reconciles_before_registering(self) -> None:
+        client = _client()
+        calls: list[str] = []
+
+        def _reconcile(*_args: object, **_kwargs: object) -> list[str]:
+            calls.append("reconcile")
+            return []
+
+        def _register_bottle(*_args: object, **_kwargs: object) -> RegisteredBottle:
+            calls.append("register")
+            return RegisteredBottle("b1", "tok")
+
+        client.reconcile.side_effect = _reconcile
+        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"])
+
+    def test_a_reconcile_failure_does_not_block_the_launch(self) -> None:
+        from bot_bottle.orchestrator.client import OrchestratorClientError
+        client = _client()
+        client.reconcile.side_effect = OrchestratorClientError("unreachable")
+        self._register(client)
+        client.register_bottle.assert_called_once()

tests/unit/test_orchestrator_client.py

@@ -104,3 +104,32 @@ class TestHealthAndPolicy(unittest.TestCase):
 
 if __name__ == "__main__":
     unittest.main()
+
+
+class TestReconcile(unittest.TestCase):
+    def setUp(self) -> None:
+        self.c = OrchestratorClient("http://orch:8080")
+
+    def test_posts_live_ips_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"])
+        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"])
+        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.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([]))
+        with patch(_URLOPEN, return_value=_resp(200, {})):
+            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([])

tests/unit/test_orchestrator_control_plane.py

@@ -8,11 +8,13 @@ from __future__ import annotations
 
 import json
 import secrets
+import sqlite3
 import tempfile
 import threading
 import unittest
 import urllib.error
 import urllib.request
+from contextlib import closing
 from pathlib import Path
 from unittest.mock import patch
 
@@ -264,6 +266,7 @@ class TestControlPlaneAuth(unittest.TestCase):
             ("POST", "/bottles", _body({"source_ip": "10.0.0.1"})),
             ("PUT", "/bottles/x/policy", _body({"policy": "routes: []"})),
             ("DELETE", "/bottles/x", b""),
+            ("POST", "/reconcile", _body({"live_source_ips": []})),
             ("POST", "/resolve", _body({"source_ip": "10.0.0.1", "identity_token": "t"})),
             ("POST", "/attribute", _body({"source_ip": "10.0.0.1", "identity_token": "t"})),
             ("GET", "/supervise/proposals", b""),
@@ -387,3 +390,56 @@ class TestDispatchSupervise(unittest.TestCase):
 
 if __name__ == "__main__":
     unittest.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."""
+
+    def setUp(self) -> None:
+        self._tmp = tempfile.TemporaryDirectory()
+        self.orch = _orchestrator(Path(self._tmp.name) / "r.db")
+
+    def tearDown(self) -> None:
+        self._tmp.cleanup()
+
+    def _old(self, source_ip: str) -> str:
+        rec = self.orch.registry.register(source_ip)
+        with closing(sqlite3.connect(self.orch.registry.db_path)) as conn:
+            conn.execute(
+                "UPDATE orchestrator_bottles SET created_at = 0.0 WHERE bottle_id = ?",
+                (rec.bottle_id,))
+            conn.commit()
+        return rec.bottle_id
+
+    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.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_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")
+        status, payload = dispatch(
+            self.orch, "POST", "/reconcile",
+            _body({"live_source_ips": [], "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_registry.py

@@ -4,7 +4,9 @@ from __future__ import annotations
 
 import sqlite3
 import tempfile
+import time
 import unittest
+from contextlib import closing
 from pathlib import Path
 
 from bot_bottle.orchestrator.registry import (
@@ -169,3 +171,81 @@ class TestRegistryStore(unittest.TestCase):
 
 if __name__ == "__main__":
     unittest.main()
+
+
+class TestReapAbsent(unittest.TestCase):
+    """`reap_absent` — the self-heal for rows whose bottle is gone.
+
+    An orphan is not merely untidy: source IPs get recycled, and
+    `by_source_ip` fail-closes on ambiguity, so a leftover row at a reused
+    address resolves *no* policy for the next bottle that lands there and
+    every host it asks for is denied.
+    """
+
+    def setUp(self) -> None:
+        self._tmp = tempfile.TemporaryDirectory()
+        self.db = Path(self._tmp.name) / "registry.db"
+        self.store = RegistryStore(self.db)
+        self.store.migrate()
+
+    def tearDown(self) -> None:
+        self._tmp.cleanup()
+
+    def _aged(self, source_ip: str, *, age: float) -> BottleRecord:
+        """Register a bottle and backdate it past the grace window."""
+        rec = self.store.register(source_ip)
+        with closing(sqlite3.connect(self.db)) as conn:
+            conn.execute(
+                "UPDATE orchestrator_bottles SET created_at = ? WHERE bottle_id = ?",
+                (time.time() - age, rec.bottle_id),
+            )
+            conn.commit()
+        return rec
+
+    def test_reaps_row_with_no_live_container(self) -> None:
+        gone = self._aged("10.243.0.9", age=600)
+        reaped = self.store.reap_absent([])
+        self.assertEqual([gone.bottle_id], [r.bottle_id for r in reaped])
+        self.assertIsNone(self.store.get(gone.bottle_id))
+
+    def test_keeps_row_whose_ip_is_live(self) -> None:
+        alive = self._aged("10.243.0.9", age=600)
+        self.assertEqual([], self.store.reap_absent(["10.243.0.9"]))
+        self.assertIsNotNone(self.store.get(alive.bottle_id))
+
+    def test_grace_window_protects_an_in_flight_launch(self) -> None:
+        """A bottle registered moments ago is never reaped, even though the
+        caller's enumeration didn't see its address yet."""
+        fresh = self.store.register("10.243.0.10")
+        self.assertEqual([], self.store.reap_absent([]))
+        self.assertIsNotNone(self.store.get(fresh.bottle_id))
+
+    def test_reaping_the_orphan_unbricks_the_reused_address(self) -> None:
+        """The regression this exists for: an orphan at an address that vmnet
+        later hands to a new bottle makes `by_source_ip` ambiguous, so the new
+        bottle resolves no policy at all."""
+        orphan = self._aged("10.243.0.11", age=600)
+        # A new bottle lands on the recycled address. Force the row in directly
+        # so `register`'s own supersede sweep doesn't mask the ambiguity.
+        with closing(sqlite3.connect(self.db)) as conn:
+            conn.execute(
+                "INSERT INTO orchestrator_bottles "
+                "(bottle_id, source_ip, identity_token, state, created_at, metadata, policy) "
+                "VALUES ('newbottle', '10.243.0.11', 'tok-new', 'active', ?, '', 'routes: []')",
+                (time.time(),),
+            )
+            conn.commit()
+        self.assertIsNone(self.store.by_source_ip("10.243.0.11"))  # bricked
+
+        reaped = self.store.reap_absent(["10.243.0.11"], grace_seconds=60)
+        self.assertEqual([orphan.bottle_id], [r.bottle_id for r in reaped])
+        rec = self.store.by_source_ip("10.243.0.11")
+        assert rec is not None
+        self.assertEqual("newbottle", rec.bottle_id)
+
+    def test_ignores_empty_ips_in_the_live_set(self) -> None:
+        gone = self._aged("10.243.0.12", age=600)
+        self.assertEqual(
+            [gone.bottle_id],
+            [r.bottle_id for r in self.store.reap_absent(["", "10.243.0.99"])],
+        )

tests/unit/test_orchestrator_service.py

@@ -4,8 +4,10 @@ from __future__ import annotations
 
 import json
 import secrets
+import sqlite3
 import tempfile
 import unittest
+from contextlib import closing
 from pathlib import Path
 from unittest.mock import patch
 
@@ -304,3 +306,55 @@ class TestOrchestratorSupervise(unittest.TestCase):
 
 if __name__ == "__main__":
     unittest.main()
+
+
+class TestOrchestratorReconcile(unittest.TestCase):
+    """`reconcile` — drop rows for bottles that are no longer running."""
+
+    def setUp(self) -> None:
+        self._tmp = tempfile.TemporaryDirectory()
+        self.secret = secrets.token_bytes(16)
+        self.db = Path(self._tmp.name) / "r.db"
+        self.store = RegistryStore(self.db)
+        self.store.migrate()
+        self.broker = StubBroker(self.secret)
+        self.orch = Orchestrator(self.store, self.broker, self.secret)
+
+    def tearDown(self) -> None:
+        self._tmp.cleanup()
+
+    def _age_all(self, seconds: float) -> None:
+        """Backdate every row past the reap grace window."""
+        with closing(sqlite3.connect(self.db)) as conn:
+            conn.execute(
+                "UPDATE orchestrator_bottles SET created_at = created_at - ?", (seconds,))
+            conn.commit()
+
+    def test_reaps_dead_bottle_and_forgets_its_tokens(self) -> None:
+        dead = self.orch.launch_bottle("10.243.0.1", tokens={"EGRESS_TOKEN_0": "s3cret"})
+        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"]))
+        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.
+        self.assertEqual({}, self.orch.tokens_for(dead.bottle_id))
+        self.assertEqual({"EGRESS_TOKEN_0": "keep"}, self.orch.tokens_for(live.bottle_id))
+
+    def test_reconcile_does_not_broker_a_teardown(self) -> None:
+        """The container is already gone — there is nothing to stop, and a
+        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.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.assertIsNotNone(self.store.get(a.bottle_id))
+        self.assertIsNotNone(self.store.get(b.bottle_id))