fix: route remote control through provider startup args

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

bot_bottle/backend/egress_apply.py

@@ -11,7 +11,7 @@ from pathlib import Path
 
 from ..bottle_state import egress_state_dir
 from ..egress import EGRESS_ROUTES_FILENAME
-from ..egress_addon_core import load_routes
+from ..egress_addon_core import LOG_OFF, load_config
 
 
 class EgressApplyError(RuntimeError):
@@ -33,11 +33,15 @@ class EgressApplicator(ABC):
     @staticmethod
     def validate_routes_content(content: str) -> None:
         try:
-            load_routes(content)
+            config = load_config(content)
         except ValueError as e:
             raise EgressApplyError(
                 f"proposed routes.yaml is not valid: {e}"
             ) from e
+        if config.log != LOG_OFF:
+            raise EgressApplyError(
+                "proposed routes.yaml must not change egress logging"
+            )
 
     @staticmethod
     def _routes_path(slug: str) -> Path:

bot_bottle/egress_addon_core.py

@@ -439,15 +439,6 @@ def route_to_yaml_dict(r: Route) -> dict[str, object]:
     return d
 
 
-def load_routes(text: str) -> tuple[Route, ...]:
-    """Parse YAML text → routes."""
-    try:
-        payload = parse_yaml_subset(text)
-    except YamlSubsetError as e:
-        raise ValueError(f"routes payload: invalid YAML: {e}") from e
-    return parse_routes(payload)
-
-
 def parse_config(payload: object) -> "Config":
     """Parse a full egress config payload (top-level log level + routes)."""
     if not isinstance(payload, dict):
@@ -862,7 +853,6 @@ __all__ = [
     "is_git_push_request",
     "is_git_fetch_request",
     "load_config",
-    "load_routes",
     "match_route",
     "outbound_scan_headers",
     "parse_config",

bot_bottle/supervise_server.py

@@ -47,11 +47,11 @@ from pathlib import Path
 try:
     # Same-directory imports inside the bundle container; these files are
     # COPYed flat under /app by Dockerfile.sidecars.
-    from egress_addon_core import load_routes
+    from egress_addon_core import LOG_OFF, load_config
     import supervise as _sv
 except ModuleNotFoundError:
     # Package imports for host-side tests and tooling.
-    from .egress_addon_core import load_routes
+    from .egress_addon_core import LOG_OFF, load_config
     from . import supervise as _sv
 
 
@@ -297,12 +297,17 @@ def validate_proposed_file(tool: str, content: str) -> None:
         pass
     elif tool in (_sv.TOOL_EGRESS_ALLOW, _sv.TOOL_EGRESS_BLOCK):
         try:
-            load_routes(content)
+            config = load_config(content)
         except ValueError as e:
             raise _RpcError(
                 ERR_INVALID_PARAMS,
                 f"{tool}: proposed routes.yaml is not valid: {e}",
             ) from e
+        if config.log != LOG_OFF:
+            raise _RpcError(
+                ERR_INVALID_PARAMS,
+                f"{tool}: proposed routes.yaml must not change egress logging",
+            )
     else:
         raise _RpcError(ERR_INVALID_PARAMS, f"unknown tool {tool!r}")
 

tests/unit/test_egress.py

@@ -322,7 +322,7 @@ class TestRenderRoutes(unittest.TestCase):
         self.assertEqual([], parse_yaml_subset(rendered)["routes"])
 
     def test_round_trip_through_addon_core(self):
-        from bot_bottle.egress_addon_core import load_routes
+        from bot_bottle.egress_addon_core import load_config
         b = _bottle([
             {"host": "api.github.com",
              "auth": {"scheme": "Bearer", "token_ref": "GH_PAT"},
@@ -333,7 +333,7 @@ class TestRenderRoutes(unittest.TestCase):
             {"host": "api.anthropic.com"},
         ])
         routes = egress_routes_for_bottle(b)
-        addon_routes = load_routes(egress_render_routes(routes))
+        addon_routes = load_config(egress_render_routes(routes)).routes
         self.assertEqual(3, len(addon_routes))
         self.assertEqual("Bearer", addon_routes[0].auth_scheme)
         self.assertEqual("EGRESS_TOKEN_0", addon_routes[0].token_env)
@@ -341,26 +341,26 @@ class TestRenderRoutes(unittest.TestCase):
         self.assertEqual("", addon_routes[2].auth_scheme)
 
     def test_dlp_round_trips(self):
-        from bot_bottle.egress_addon_core import load_routes
+        from bot_bottle.egress_addon_core import load_config
         b = _bottle([{"host": "x.example", "dlp": {
             "outbound_detectors": ["token_patterns"],
             "inbound_detectors": False,
         }}])
         routes = egress_routes_for_bottle(b)
         rendered = egress_render_routes(routes)
-        addon_routes = load_routes(rendered)
+        addon_routes = load_config(rendered).routes
         self.assertEqual(("token_patterns",), addon_routes[0].outbound_detectors)
         self.assertEqual((), addon_routes[0].inbound_detectors)
 
     def test_outbound_on_match_round_trips(self):
-        from bot_bottle.egress_addon_core import load_routes
+        from bot_bottle.egress_addon_core import load_config
         b = _bottle([{"host": "logs.example", "dlp": {
             "outbound_on_match": "redact",
         }}])
         routes = egress_routes_for_bottle(b)
         rendered = egress_render_routes(routes)
         self.assertIn('outbound_on_match: "redact"', rendered)
-        addon_routes = load_routes(rendered)
+        addon_routes = load_config(rendered).routes
         self.assertEqual("redact", addon_routes[0].outbound_on_match)
 
     def test_outbound_on_match_default_omitted_from_render(self):
@@ -370,12 +370,12 @@ class TestRenderRoutes(unittest.TestCase):
         self.assertNotIn("outbound_on_match", rendered)
 
     def test_git_fetch_policy_round_trips(self):
-        from bot_bottle.egress_addon_core import load_routes
+        from bot_bottle.egress_addon_core import load_config
         b = _bottle([{"host": "github.com", "git": {"fetch": True}}])
         routes = egress_routes_for_bottle(b)
         rendered = egress_render_routes(routes)
         self.assertEqual({"fetch": True}, self._parsed(routes)[0]["git"])
-        addon_routes = load_routes(rendered)
+        addon_routes = load_config(rendered).routes
         self.assertTrue(addon_routes[0].git_fetch)
 
     def test_log_zero_omitted_from_render(self):

tests/unit/test_egress_addon_core.py

@@ -32,7 +32,6 @@ from bot_bottle.egress_addon_core import (
     is_git_fetch_request,
     is_git_push_request,
     load_config,
-    load_routes,
     match_route,
     outbound_scan_headers,
     parse_config,
@@ -289,47 +288,6 @@ class TestParseDlp(unittest.TestCase):
             }]})
 
 
-# --- load_routes ---------------------------------------------------------
-
-
-class TestLoadRoutes(unittest.TestCase):
-    def test_yaml_text_round_trip(self):
-        routes = load_routes(
-            'routes:\n'
-            '  - host: "api.example"\n'
-        )
-        self.assertEqual(1, len(routes))
-        self.assertEqual("api.example", routes[0].host)
-
-    def test_full_route_shape_parses(self):
-        routes = load_routes(
-            'routes:\n'
-            '  - host: "api.example"\n'
-            '    auth_scheme: "Bearer"\n'
-            '    token_env: "EGRESS_TOKEN_0"\n'
-            '    matches:\n'
-            '      - paths:\n'
-            '          - value: "/v1/"\n'
-            '          - type: "exact"\n'
-            '            value: "/messages"\n'
-        )
-        self.assertEqual(1, len(routes))
-        r = routes[0]
-        self.assertEqual("api.example", r.host)
-        self.assertEqual("Bearer", r.auth_scheme)
-        self.assertEqual("EGRESS_TOKEN_0", r.token_env)
-        self.assertEqual(1, len(r.matches))
-        self.assertEqual(2, len(r.matches[0].paths))
-
-    def test_empty_routes_list(self):
-        routes = load_routes("routes: []\n")
-        self.assertEqual((), routes)
-
-    def test_invalid_yaml_raises_value_error(self):
-        with self.assertRaises(ValueError):
-            load_routes("routes:\n\t- host: x\n")
-
-
 # --- load_config / parse_config ------------------------------------------
 
 
@@ -378,6 +336,33 @@ class TestLoadConfig(unittest.TestCase):
         with self.assertRaises(ValueError):
             parse_config("not a dict")
 
+    def test_empty_routes_list(self):
+        cfg = load_config("routes: []\n")
+        self.assertEqual((), cfg.routes)
+
+    def test_full_route_shape_parses(self):
+        cfg = load_config(
+            'routes:\n'
+            '  - host: "api.example"\n'
+            '    auth_scheme: "Bearer"\n'
+            '    token_env: "EGRESS_TOKEN_0"\n'
+            '    matches:\n'
+            '      - paths:\n'
+            '          - value: "/v1/"\n'
+            '          - type: "exact"\n'
+            '            value: "/messages"\n'
+        )
+        r = cfg.routes[0]
+        self.assertEqual("api.example", r.host)
+        self.assertEqual("Bearer", r.auth_scheme)
+        self.assertEqual("EGRESS_TOKEN_0", r.token_env)
+        self.assertEqual(1, len(r.matches))
+        self.assertEqual(2, len(r.matches[0].paths))
+
+    def test_invalid_yaml_raises_value_error(self):
+        with self.assertRaises(ValueError):
+            load_config("routes:\n\t- host: x\n")
+
 
 # --- evaluate_matches ---------------------------------------------------
 

tests/unit/test_egress_apply.py

@@ -54,6 +54,15 @@ class TestValidateRoutesContent(unittest.TestCase):
                 '    auth_scheme: "Bearer"\n'
             )
 
+    def test_rejects_log_full(self):
+        with self.assertRaises(EgressApplyError) as cm:
+            applicator.validate_routes_content(
+                'log: 2\n'
+                'routes:\n'
+                '  - host: "x.example"\n'
+            )
+        self.assertIn("must not change egress logging", str(cm.exception))
+
 
 class TestApplyRoutesChange(unittest.TestCase):
     def setUp(self):

tests/unit/test_supervise_server.py

@@ -67,6 +67,15 @@ class TestValidation(unittest.TestCase):
         with self.assertRaises(_RpcError):
             validate_proposed_file(_sv.TOOL_EGRESS_BLOCK, "routes: nope\n")
 
+    def test_egress_routes_yaml_rejects_log_full(self):
+        with self.assertRaises(_RpcError) as cm:
+            validate_proposed_file(
+                _sv.TOOL_EGRESS_ALLOW,
+                "log: 2\nroutes:\n  - host: example.com\n",
+            )
+        self.assertEqual(ERR_INVALID_PARAMS, cm.exception.code)
+        self.assertIn("must not change egress logging", cm.exception.message)
+
 
 # --- JSON-RPC parsing ------------------------------------------------------