refactor: split _signal_bundle_reload per backend, move macos egress to macos_container

bot_bottle/backend/docker/egress_apply.py

@@ -11,7 +11,7 @@ import os
 import subprocess
 from pathlib import Path
 
-from ...bottle_state import egress_state_dir, read_metadata
+from ...bottle_state import egress_state_dir
 from ...egress import EGRESS_ROUTES_FILENAME, EGRESS_ROUTES_IN_CONTAINER
 from ...egress_addon_core import load_routes
 from ...log import warn
@@ -63,46 +63,20 @@ def _routes_path(slug: str) -> Path:
 
 def _signal_bundle_reload(slug: str) -> None:
     container = sidecar_bundle_container_name(slug)
-    backend = ""
+    result = subprocess.run(
-    metadata = read_metadata(slug)
+        ["docker", "kill", "--signal", "HUP", container],
-    if metadata is not None:
+        capture_output=True, text=True, check=False, env=os.environ,
-        backend = metadata.backend
+    )
-
+    if result.returncode != 0:
-    candidates: list[list[str]]
-    if backend == "macos-container":
-        candidates = [["container", "kill", "--signal", "HUP", container]]
-    elif backend:
-        candidates = [["docker", "kill", "--signal", "HUP", container]]
-    else:
-        candidates = [
-            ["docker", "kill", "--signal", "HUP", container],
-            ["container", "kill", "--signal", "HUP", container],
-        ]
-
-    last_error = ""
-    for argv in candidates:
-        try:
-            result = subprocess.run(
-                argv,
-                capture_output=True,
-                text=True,
-                check=False,
-                env=os.environ,
-            )
-        except FileNotFoundError as e:
-            last_error = str(e)
-            continue
-        if result.returncode == 0:
-            return
         last_error = (result.stderr or "").strip() or (result.stdout or "").strip()
-    warn(
+        warn(
-        f"egress: routes updated on disk for {slug}, but bundle reload failed: "
+            f"egress: routes updated on disk for {slug}, but bundle reload failed: "
-        f"{last_error or 'no reload command succeeded'}"
+            f"{last_error or 'docker kill failed'}"
-    )
+        )
-    raise EgressApplyError(
+        raise EgressApplyError(
-        f"could not reload egress bundle {container}: "
+            f"could not reload egress bundle {container}: "
-        f"{last_error or 'no reload command succeeded'}"
+            f"{last_error or 'docker kill failed'}"
-    )
+        )
 
 
 __all__ = [

bot_bottle/backend/macos_container/egress_apply.py

@@ -0,0 +1,54 @@
+"""Host-side egress apply for the macos-container backend.
+
+Uses `container kill --signal HUP` (Apple Container framework) instead
+of `docker kill` to signal the sidecar bundle.
+"""
+
+from __future__ import annotations
+
+import os
+import subprocess
+from pathlib import Path
+
+from ...bottle_state import egress_state_dir
+from ...egress import EGRESS_ROUTES_FILENAME
+from ...log import warn
+from ..docker.egress_apply import EgressApplyError, validate_routes_content
+from .launch import sidecar_container_name
+
+
+def apply_routes_change(slug: str, content: str) -> tuple[str, str]:
+    """Persist `content` to the live routes file and reload egress."""
+    validate_routes_content(content)
+    routes_path = _routes_path(slug)
+    routes_path.parent.mkdir(parents=True, exist_ok=True)
+    before = routes_path.read_text(encoding="utf-8") if routes_path.exists() else ""
+    routes_path.write_text(content, encoding="utf-8")
+    routes_path.chmod(0o600)
+    _signal_bundle_reload(slug)
+    return before, content
+
+
+def _routes_path(slug: str) -> Path:
+    return egress_state_dir(slug) / EGRESS_ROUTES_FILENAME
+
+
+def _signal_bundle_reload(slug: str) -> None:
+    container = sidecar_container_name(slug)
+    result = subprocess.run(
+        ["container", "kill", "--signal", "HUP", container],
+        capture_output=True, text=True, check=False, env=os.environ,
+    )
+    if result.returncode != 0:
+        last_error = (result.stderr or "").strip() or (result.stdout or "").strip()
+        warn(
+            f"egress: routes updated on disk for {slug}, but bundle reload failed: "
+            f"{last_error or 'container kill failed'}"
+        )
+        raise EgressApplyError(
+            f"could not reload egress bundle {container}: "
+            f"{last_error or 'container kill failed'}"
+        )
+
+
+__all__ = ["apply_routes_change"]

bot_bottle/cli/supervise.py

@@ -21,14 +21,17 @@ from datetime import datetime, timezone
 from pathlib import Path
 
 from .. import supervise as _supervise
-# from ..bottle_state import read_metadata
+from ..bottle_state import read_metadata
 # from ..backend.docker.capability_apply import (
 #     CapabilityApplyError,
 #     apply_capability_change,
 # )
 from ..backend.docker.egress_apply import (
     EgressApplyError,
-    apply_routes_change,
+    apply_routes_change as _docker_apply_routes_change,
+)
+from ..backend.macos_container.egress_apply import (
+    apply_routes_change as _macos_apply_routes_change,
 )
 from ..log import Die, error, info
 
@@ -73,6 +76,13 @@ class QueuedProposal:
 ApplyError = (CapabilityApplyError, EgressApplyError)
 
 
+def apply_routes_change(slug: str, content: str) -> tuple[str, str]:
+    meta = read_metadata(slug)
+    if meta is not None and meta.backend == "macos-container":
+        return _macos_apply_routes_change(slug, content)
+    return _docker_apply_routes_change(slug, content)
+
+
 def discover_pending() -> list[QueuedProposal]:
     """Walk ~/.bot-bottle/queue/* and collect pending proposals."""
     queue_root = _supervise.bot_bottle_root() / "queue"

tests/unit/test_egress_apply.py

@@ -97,33 +97,6 @@ class TestApplyRoutesChange(unittest.TestCase):
             calls[0],
         )
 
-    def test_updates_legacy_routes_file_when_existing_bottle_mounted_it(self):
-        legacy_path = (
-            Path(self._tmp.name)
-            / ".bot-bottle/state/dev/egress/egress_routes.yaml"
-        )
-        legacy_path.parent.mkdir(parents=True)
-        legacy_path.write_text("routes: []\n", encoding="utf-8")
-
-        with patch(
-            "bot_bottle.backend.docker.egress_apply.subprocess.run",
-            return_value=SimpleNamespace(returncode=0, stdout="", stderr=""),
-        ):
-            before, after = apply_routes_change(
-                "dev",
-                "routes:\n  - host: google.com\n",
-            )
-
-        self.assertEqual("routes: []\n", before)
-        self.assertEqual("routes:\n  - host: google.com\n", after)
-        self.assertEqual(
-            "routes:\n  - host: google.com\n",
-            legacy_path.read_text(encoding="utf-8"),
-        )
-        self.assertFalse(
-            (Path(self._tmp.name) / ".bot-bottle/state/dev/egress/routes.yaml").exists(),
-        )
-
 
 if __name__ == "__main__":
     unittest.main()