feat(macos): split orchestrator and gateway into separate containers (PRD 0070)

The single Apple container existed only because two guests writing one
bot-bottle.db over virtiofs would race incoherent fcntl locks; #469 removed
that (the data plane no longer opens the DB), so the macOS backend now runs
the planes as two containers like docker:

  * bot-bottle-mac-orchestrator — lean control plane on the host-only
    `bot-bottle-mac-control` network only (image Dockerfile.orchestrator,
    `-m bot_bottle.orchestrator`). Sole mounter of the container-only DB
    volume; holds the signing key. The CLI + gateway reach it at its
    control-network address.
  * bot-bottle-mac-infra — the gateway, triple-homed on the NAT egress net,
    the host-only agent net, and the control net. Resolves the orchestrator by
    IP (Apple has no container DNS) via BOT_BOTTLE_ORCHESTRATOR_URL; holds the
    CA + gateway JWT.

Agents sit on the agent network only, so they have no route to the control
plane. ensure_networks gains the control network; MacosInfraService brings up
the orchestrator then the gateway; probe_orchestrator_url + ca_cert_pem target
the right containers. pyright 0 errors; unit suite green (2274). Needs a real
Apple-container host to validate the networking.

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

tests/unit/test_macos_infra.py

@@ -1,4 +1,4 @@
-"""Unit: the single macOS infra container (control plane + gateway, PRD 0070)."""
+"""Unit: macOS orchestrator + gateway containers (PRD 0070 plane split)."""
 
 from __future__ import annotations
 
@@ -24,86 +24,94 @@ def _fail(stderr: str = "boom") -> Mock:
     return Mock(returncode=1, stdout="", stderr=stderr)
 
 
-class TestInfraRun(unittest.TestCase):
-    def _run_container(self, svc: MacosInfraService) -> list[str]:
+def _spec(src: str, tgt: str, readonly: bool = False) -> str:
+    return f"type=bind,source={src},target={tgt}" + (",readonly" if readonly else "")
+
+
+class TestOrchestratorRun(unittest.TestCase):
+    def _run(self, svc: MacosInfraService) -> list[str]:
         run = Mock(return_value=_ok())
-
-        def _spec(src: str, tgt: str, readonly: bool = False) -> str:
-            return f"type=bind,source={src},target={tgt}" + (
-                ",readonly" if readonly else "")
-
         with patch(f"{_INFRA}.container_mod") as mod, \
-                patch(f"{_INFRA}.ensure_networks"):
+                patch(f"{_INFRA}.host_orchestrator_token", return_value="k"):
             mod.dns_server.return_value = "1.1.1.1"
             mod.bind_mount_spec.side_effect = _spec
             mod.run_container_argv = run
-            svc._run_container("h1")
+            svc._run_orchestrator_container("h1")
         return run.call_args.args[0]
 
-    def test_single_container_runs_both_processes(self) -> None:
-        """The whole point: one container starts the control plane AND the
-        gateway daemons, so one kernel owns the DB."""
-        argv = self._run_container(MacosInfraService(repo_root=Path("/r")))
-        script = argv[-1]
-        self.assertIn("bot_bottle.orchestrator", script)
-        # Gateway launches via the installed package (there is no
-        # /app/gateway_init.py file since the daemons moved into bot_bottle).
-        self.assertIn("bot_bottle.gateway.bootstrap", script)
-        self.assertIn("127.0.0.1", script)  # they reach each other on loopback
+    def test_runs_the_orchestrator_on_the_control_network_only(self) -> None:
+        argv = self._run(MacosInfraService(repo_root=Path("/r")))
+        nets = [argv[i + 1] for i, a in enumerate(argv) if a == "--network"]
+        self.assertEqual(["bot-bottle-mac-control"], nets)
+        # Image ENTRYPOINT is `-m bot_bottle.orchestrator`; these are its args.
+        self.assertIn("--broker", argv)
+        self.assertIn("stub", argv)
+        self.assertIn("bot-bottle-orchestrator:latest", argv)
 
     def test_db_is_a_container_only_volume(self) -> None:
-        """No host bind-mount of the DB — a named volume only this container
-        mounts, so the DB is never written by two kernels."""
-        argv = self._run_container(MacosInfraService(repo_root=Path("/r")))
+        argv = self._run(MacosInfraService(repo_root=Path("/r")))
         vols = [argv[i + 1] for i, a in enumerate(argv) if a == "--volume"]
         self.assertTrue(any(v.startswith(f"{INFRA_DB_VOLUME}:") for v in vols))
-        # The repo source is bind-mounted read-only; the DB is not a bind mount.
-        mounts = [argv[i + 1] for i, a in enumerate(argv) if a == "--mount"]
-        self.assertTrue(all("bot-bottle.db" not in m for m in mounts))
 
-    def test_ca_is_persisted_on_the_host_not_the_container_volume(self) -> None:
-        """The CA survives infra recreation and cannot be removed by Apple
-        Container's volume-prune command."""
-        argv = self._run_container(MacosInfraService(repo_root=Path("/r")))
+    def test_orchestrator_has_no_ca_mount(self) -> None:
+        # The CA lives with the gateway, not the control plane.
+        argv = self._run(MacosInfraService(repo_root=Path("/r")))
         mounts = [argv[i + 1] for i, a in enumerate(argv) if a == "--mount"]
-        ca_mounts = [
-            m for m in mounts
-            if "target=/home/mitmproxy/.mitmproxy" in m
-        ]
-        self.assertEqual(1, len(ca_mounts))
-        self.assertIn("source=", ca_mounts[0])
-        self.assertIn("/gateway-ca", ca_mounts[0])
-        self.assertNotIn(",readonly", ca_mounts[0])
-
-    def test_nat_network_precedes_the_host_only_network(self) -> None:
-        argv = self._run_container(MacosInfraService(repo_root=Path("/r")))
-        nets = [argv[i + 1] for i, a in enumerate(argv) if a == "--network"]
-        self.assertEqual(["bot-bottle-mac-egress", "bot-bottle-mac-gateway"], nets)
+        self.assertFalse([m for m in mounts if "/home/mitmproxy" in m])
 
     def test_source_hash_is_labelled_for_recreate(self) -> None:
-        argv = self._run_container(MacosInfraService(repo_root=Path("/r")))
+        argv = self._run(MacosInfraService(repo_root=Path("/r")))
         self.assertIn("BOT_BOTTLE_SOURCE_HASH=h1", argv)
 
     def test_start_failure_raises(self) -> None:
         svc = MacosInfraService(repo_root=Path("/r"))
         with patch(f"{_INFRA}.container_mod") as mod, \
-                patch(f"{_INFRA}.ensure_networks"):
+                patch(f"{_INFRA}.host_orchestrator_token", return_value="k"):
             mod.dns_server.return_value = "1.1.1.1"
-            mod.bind_mount_spec.return_value = "m"
             mod.run_container_argv = Mock(return_value=_fail())
             with self.assertRaises(OrchestratorStartError):
-                svc._run_container("h1")
+                svc._run_orchestrator_container("h1")
+
+
+class TestGatewayRun(unittest.TestCase):
+    def _run(self, svc: MacosInfraService, url: str = "http://10.0.0.5:8099") -> list[str]:
+        run = Mock(return_value=_ok())
+        with patch(f"{_INFRA}.container_mod") as mod, \
+                patch(f"{_INFRA}.host_orchestrator_token", return_value="k"), \
+                patch(f"{_INFRA}.mint", return_value="jwt"):
+            mod.dns_server.return_value = "1.1.1.1"
+            mod.bind_mount_spec.side_effect = _spec
+            mod.run_container_argv = run
+            svc._ensure_gateway_container(url)
+        return run.call_args.args[0]
+
+    def test_gateway_is_triple_homed(self) -> None:
+        argv = self._run(MacosInfraService(repo_root=Path("/r")))
+        nets = [argv[i + 1] for i, a in enumerate(argv) if a == "--network"]
+        self.assertEqual(
+            ["bot-bottle-mac-egress", "bot-bottle-mac-gateway", "bot-bottle-mac-control"],
+            nets,
+        )
+
+    def test_gateway_resolves_orchestrator_by_url_and_holds_ca(self) -> None:
+        argv = self._run(MacosInfraService(repo_root=Path("/r")), "http://10.0.0.5:8099")
+        self.assertIn("BOT_BOTTLE_ORCHESTRATOR_URL=http://10.0.0.5:8099", argv)
+        mounts = [argv[i + 1] for i, a in enumerate(argv) if a == "--mount"]
+        self.assertTrue([m for m in mounts if "target=/home/mitmproxy/.mitmproxy" in m])
+        self.assertIn("bot-bottle-gateway:latest", argv)
+        self.assertIn(
+            "BOT_BOTTLE_GATEWAY_DAEMONS=egress,git-http,supervise", argv)
 
 
 class TestInfraEnsureRunning(unittest.TestCase):
-    def test_current_healthy_container_is_left_alone(self) -> None:
-        """Idempotent singleton: N launches must not churn the infra container
-        and drop every live bottle's control plane."""
+    def test_current_healthy_orchestrator_left_alone(self) -> None:
         svc = MacosInfraService(repo_root=Path("/r"))
         run = Mock()
         with patch(f"{_INFRA}.container_mod") as mod, \
                 patch(f"{_INFRA}.source_hash", return_value="h1"), \
-                patch.object(svc, "_run_container", run), \
+                patch.object(svc, "ensure_built"), \
+                patch.object(svc, "_run_orchestrator_container", run), \
+                patch.object(svc, "_ensure_gateway_container"), \
                 patch.object(svc, "is_healthy", return_value=True):
             mod.container_is_running.return_value = True
             mod.container_env.return_value = {"BOT_BOTTLE_SOURCE_HASH": "h1"}
@@ -113,13 +121,14 @@ class TestInfraEnsureRunning(unittest.TestCase):
         self.assertEqual("http://192.168.128.2:8099", endpoint.orchestrator_url)
         self.assertEqual("192.168.128.2", endpoint.gateway_ip)
 
-    def test_changed_source_recreates(self) -> None:
+    def test_changed_source_recreates_orchestrator(self) -> None:
         svc = MacosInfraService(repo_root=Path("/r"))
         run = Mock()
         with patch(f"{_INFRA}.container_mod") as mod, \
                 patch(f"{_INFRA}.source_hash", return_value="h2"), \
                 patch.object(svc, "ensure_built"), \
-                patch.object(svc, "_run_container", run), \
+                patch.object(svc, "_run_orchestrator_container", run), \
+                patch.object(svc, "_ensure_gateway_container"), \
                 patch.object(svc, "is_healthy", return_value=True):
             mod.container_is_running.return_value = True
             mod.container_env.return_value = {"BOT_BOTTLE_SOURCE_HASH": "h1"}
@@ -127,17 +136,15 @@ class TestInfraEnsureRunning(unittest.TestCase):
             svc.ensure_running()
         run.assert_called_once()
 
-    def test_wedged_but_current_container_is_recreated(self) -> None:
-        """Current source but a dead HTTP server must be recreated, not polled
-        to death forever — health, not just the source label, gates reuse."""
+    def test_wedged_but_current_orchestrator_recreated(self) -> None:
         svc = MacosInfraService(repo_root=Path("/r"))
         run = Mock()
-        health = Mock(side_effect=[False, True])
         with patch(f"{_INFRA}.container_mod") as mod, \
                 patch(f"{_INFRA}.source_hash", return_value="h1"), \
                 patch.object(svc, "ensure_built"), \
-                patch.object(svc, "_run_container", run), \
-                patch.object(svc, "is_healthy", health):
+                patch.object(svc, "_run_orchestrator_container", run), \
+                patch.object(svc, "_ensure_gateway_container"), \
+                patch.object(svc, "is_healthy", Mock(side_effect=[False, True])):
             mod.container_is_running.return_value = True
             mod.container_env.return_value = {"BOT_BOTTLE_SOURCE_HASH": "h1"}
             mod.try_container_ipv4_on_network.return_value = "192.168.128.2"
@@ -149,7 +156,8 @@ class TestInfraEnsureRunning(unittest.TestCase):
         with patch(f"{_INFRA}.container_mod") as mod, \
                 patch(f"{_INFRA}.source_hash", return_value="h1"), \
                 patch.object(svc, "ensure_built"), \
-                patch.object(svc, "_run_container"), \
+                patch.object(svc, "_run_orchestrator_container"), \
+                patch.object(svc, "_ensure_gateway_container"), \
                 patch.object(svc, "is_healthy", return_value=False):
             mod.container_is_running.return_value = False
             mod.try_container_ipv4_on_network.return_value = "192.168.128.2"
@@ -158,7 +166,7 @@ class TestInfraEnsureRunning(unittest.TestCase):
 
 
 class TestCaCertPem(unittest.TestCase):
-    def test_reads_ca_out_of_the_container(self) -> None:
+    def test_reads_ca_out_of_the_gateway_container(self) -> None:
         svc = MacosInfraService(repo_root=Path("/r"))
         with patch(f"{_INFRA}.container_mod") as mod:
             mod.run_container_argv.return_value = _ok("-----BEGIN CERTIFICATE-----\n")