refactor: extract EgressApplicator base class shared between backends

Pulls the duplicated apply_routes_change / validate_routes_content /
_routes_path logic into EgressApplicator (ABC) in backend/egress_apply.py.
DockerEgressApplicator and MacOSContainerEgressApplicator override the
single abstract _signal_bundle_reload method with their respective kill
commands. Module-level shims preserve the existing public API.

bot_bottle/backend/docker/egress_apply.py

@@ -9,19 +9,13 @@ from __future__ import annotations
 
 import os
 import subprocess
-from pathlib import Path
 
-from ...bottle_state import egress_state_dir
+from ...egress import EGRESS_ROUTES_IN_CONTAINER
-from ...egress import EGRESS_ROUTES_FILENAME, EGRESS_ROUTES_IN_CONTAINER
-from ...egress_addon_core import load_routes
 from ...log import warn
+from ..egress_apply import EgressApplicator, EgressApplyError
 from .sidecar_bundle import sidecar_bundle_container_name
 
 
-class EgressApplyError(RuntimeError):
-    pass
-
-
 def fetch_current_routes(slug: str) -> str:
     container = sidecar_bundle_container_name(slug)
     r = subprocess.run(
@@ -36,32 +30,8 @@ def fetch_current_routes(slug: str) -> str:
     return r.stdout
 
 
-def apply_routes_change(slug: str, content: str) -> tuple[str, str]:
+class DockerEgressApplicator(EgressApplicator):
-    """Persist `content` to the live routes file and reload egress."""
+    def _signal_bundle_reload(self, slug: str) -> None:
-    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 validate_routes_content(content: str) -> None:
-    try:
-        load_routes(content)
-    except ValueError as e:
-        raise EgressApplyError(
-            f"proposed routes.yaml is not valid: {e}"
-        ) from e
-
-
-def _routes_path(slug: str) -> Path:
-    return egress_state_dir(slug) / EGRESS_ROUTES_FILENAME
-
-
-def _signal_bundle_reload(slug: str) -> None:
         container = sidecar_bundle_container_name(slug)
         result = subprocess.run(
             ["docker", "kill", "--signal", "HUP", container],
@@ -79,7 +49,18 @@ def _signal_bundle_reload(slug: str) -> None:
             )
 
 
+_applicator = DockerEgressApplicator()
+
+
+def apply_routes_change(slug: str, content: str) -> tuple[str, str]:
+    return _applicator.apply_routes_change(slug, content)
+
+
+validate_routes_content = EgressApplicator.validate_routes_content
+
+
 __all__ = [
+    "DockerEgressApplicator",
     "EgressApplyError",
     "apply_routes_change",
     "fetch_current_routes",

bot_bottle/backend/egress_apply.py

@@ -0,0 +1,50 @@
+"""Shared base class for host-side egress apply across backends.
+
+Each backend subclasses EgressApplicator and overrides _signal_bundle_reload
+with the backend-specific kill command.
+"""
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from pathlib import Path
+
+from ..bottle_state import egress_state_dir
+from ..egress import EGRESS_ROUTES_FILENAME
+from ..egress_addon_core import load_routes
+
+
+class EgressApplyError(RuntimeError):
+    pass
+
+
+class EgressApplicator(ABC):
+    def apply_routes_change(self, slug: str, content: str) -> tuple[str, str]:
+        """Persist `content` to the live routes file and reload egress."""
+        self.validate_routes_content(content)
+        routes_path = self._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)
+        self._signal_bundle_reload(slug)
+        return before, content
+
+    @staticmethod
+    def validate_routes_content(content: str) -> None:
+        try:
+            load_routes(content)
+        except ValueError as e:
+            raise EgressApplyError(
+                f"proposed routes.yaml is not valid: {e}"
+            ) from e
+
+    @staticmethod
+    def _routes_path(slug: str) -> Path:
+        return egress_state_dir(slug) / EGRESS_ROUTES_FILENAME
+
+    @abstractmethod
+    def _signal_bundle_reload(self, slug: str) -> None: ...
+
+
+__all__ = ["EgressApplicator", "EgressApplyError"]

bot_bottle/backend/macos_container/egress_apply.py

@@ -8,32 +8,14 @@ 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 ..egress_apply import EgressApplicator, EgressApplyError
 from .launch import sidecar_container_name
 
 
-def apply_routes_change(slug: str, content: str) -> tuple[str, str]:
+class MacOSContainerEgressApplicator(EgressApplicator):
-    """Persist `content` to the live routes file and reload egress."""
+    def _signal_bundle_reload(self, slug: str) -> None:
-    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],
@@ -51,4 +33,11 @@ def _signal_bundle_reload(slug: str) -> None:
             )
 
 
-__all__ = ["apply_routes_change"]
+_applicator = MacOSContainerEgressApplicator()
+
+
+def apply_routes_change(slug: str, content: str) -> tuple[str, str]:
+    return _applicator.apply_routes_change(slug, content)
+
+
+__all__ = ["MacOSContainerEgressApplicator", "EgressApplyError", "apply_routes_change"]