refactor: drop redundant single-parent fast path in _resolve_one_bottle

_fold_parents with one name returns after the first resolve; the
single-element branch was a verbatim copy of the general path.

bot_bottle/agent_provider.py

@@ -61,7 +61,6 @@ class AgentProviderRuntime:
     prompt_mode: PromptMode
     bypass_args: tuple[str, ...]
     resume_args: tuple[str, ...]
-    remote_control_args: tuple[str, ...]
 
 
 @dataclass(frozen=True)
@@ -391,7 +390,7 @@ def prompt_args(
     if prompt_mode == "append_file":
         return ["--append-system-prompt-file", prompt_path]
     if prompt_mode == "read_prompt_file":
-        if argv and "resume" in argv:
+        if argv and ("resume" in argv or "remote-control" in argv):
             return []
         return [f"Read and follow the instructions in {prompt_path}."]
     if prompt_mode == "print_read_prompt_file":

bot_bottle/backend/__init__.py

@@ -109,9 +109,8 @@ class BottlePlan(ABC):
     def workspace_plan(self) -> WorkspacePlan:
         return workspace_plan(self.spec, guest_home=self.guest_home)
 
-    def print(self, *, remote_control: bool) -> None:
+    def print(self) -> None:
         """Render the y/N preflight summary to stderr."""
-        del remote_control
         spec = self.spec
         manifest = self.manifest
         agent = manifest.agent

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/backend/macos_container/util.py

@@ -68,6 +68,11 @@ def build_image(ref: str, context: str, *, dockerfile: str = "") -> None:
     _ensure_builder_dns()
     args = [_CONTAINER, "build", "-t", ref, "--dns", dns_server()]
     if dockerfile:
+        # `container build` resolves -f relative to the current working
+        # directory, not the build context. Anchor a relative Dockerfile to
+        # the context so builds work from any cwd.
+        if not os.path.isabs(dockerfile):
+            dockerfile = os.path.join(context, dockerfile)
         args.extend(["-f", dockerfile])
     args.append(context)
     subprocess.run(args, check=True)

bot_bottle/cli/resume.py

@@ -28,7 +28,6 @@ from .start import _launch_bottle
 def cmd_resume(argv: list[str]) -> int:
     parser = argparse.ArgumentParser(prog=f"{PROG} resume", add_help=True)
     parser.add_argument("--dry-run", action="store_true")
-    parser.add_argument("--remote-control", action="store_true")
     parser.add_argument(
         "identity",
         help="bottle identity from a prior `start` (see its session-end output)",
@@ -56,6 +55,5 @@ def cmd_resume(argv: list[str]) -> int:
     return _launch_bottle(
         spec,
         dry_run=args.dry_run,
-        remote_control=args.remote_control,
         backend_name=backend_name,
     )

bot_bottle/cli/start.py

@@ -42,7 +42,6 @@ def cmd_start(argv: list[str]) -> int:
     parser = argparse.ArgumentParser(prog=f"{PROG} start", add_help=True)
     parser.add_argument("--dry-run", action="store_true")
     parser.add_argument("--cwd", action="store_true", help="copy host cwd into the running bottle")
-    parser.add_argument("--remote-control", action="store_true")
     parser.add_argument(
         "--backend",
         choices=known_backend_names(),
@@ -89,7 +88,6 @@ def cmd_start(argv: list[str]) -> int:
     return _launch_bottle(
         spec,
         dry_run=dry_run,
-        remote_control=args.remote_control,
         backend_name=backend_name,
     )
 
@@ -134,7 +132,7 @@ def prepare_with_preflight(
 
 
 def attach_agent(
-    bottle: Bottle, *, remote_control: bool = False, resume: bool = False,
+    bottle: Bottle, *, resume: bool = False,
     agent_provider_template: str = "claude",
     startup_args: tuple[str, ...] = (),
 ) -> int:
@@ -153,8 +151,6 @@ def attach_agent(
         "(Ctrl-D or 'exit' to leave; container will be removed)"
     )
     agent_args = list(runtime.bypass_args)
-    if remote_control:
-        agent_args.extend(runtime.remote_control_args)
     agent_args.extend(startup_args)
     if resume:
         agent_args.extend(runtime.resume_args)
@@ -218,9 +214,9 @@ def _text_prompt_yes() -> bool:
     return reply in ("y", "Y", "yes", "YES")
 
 
-def _text_render_preflight(*, remote_control: bool):
+def _text_render_preflight():
     def _render(plan: DockerBottlePlan) -> None:
-        plan.print(remote_control=remote_control)
+        plan.print()
     return _render
 
 
@@ -228,7 +224,6 @@ def _launch_bottle(
     spec: BottleSpec,
     *,
     dry_run: bool,
-    remote_control: bool,
     backend_name: str | None = None,
 ) -> int:
     """Shared launch core for `start` and `resume`. Builds the plan,
@@ -240,7 +235,7 @@ def _launch_bottle(
         plan, identity = prepare_with_preflight(
             spec,
             stage_dir=stage_dir,
-            render_preflight=_text_render_preflight(remote_control=remote_control),
+            render_preflight=_text_render_preflight(),
             prompt_yes=_text_prompt_yes,
             dry_run=dry_run,
             backend_name=backend_name,
@@ -253,7 +248,6 @@ def _launch_bottle(
             agent_provider_template = getattr(plan, "agent_provider_template", "claude")
             exit_code = attach_agent(
                 bottle,
-                remote_control=remote_control,
                 agent_provider_template=agent_provider_template,
                 startup_args=plan.agent_provision.startup_args,
             )

bot_bottle/contrib/claude/agent_provider.py

@@ -91,7 +91,6 @@ _RUNTIME = AgentProviderRuntime(
     prompt_mode="append_file",
     bypass_args=("--dangerously-skip-permissions",),
     resume_args=("--continue",),
-    remote_control_args=("--remote-control",),
 )
 
 

bot_bottle/contrib/codex/Dockerfile

@@ -1,12 +1,12 @@
 # bot-bottle Codex provider image.
 #
 # Mirrors the default Claude image shape: Node LTS, git/network tooling,
-# non-root node user, and the provider CLI installed globally.
+# non-root node user, and the provider CLI installed for that user.
 
 FROM node:22-slim
 
 RUN apt-get update \
-  && apt-get install -y --no-install-recommends git ca-certificates curl \
+  && apt-get install -y --no-install-recommends git ca-certificates curl procps \
   && rm -rf /var/lib/apt/lists/*
 
 # App-specific deps. Python isn't required by codex itself
@@ -17,12 +17,15 @@ RUN apt-get update \
   && apt-get install -y --no-install-recommends python3 python3-pip python3-venv \
   && rm -rf /var/lib/apt/lists/*
 
-RUN npm install -g --no-fund --no-audit @openai/codex@0.136.0 \
-  && npm cache clean --force
-
 USER node
 WORKDIR /home/node
 
-RUN mkdir -p /home/node/.codex
+ENV PATH="/home/node/.local/bin:${PATH}"
+
+# Remote-control support requires the standalone Codex install layout
+# under ~/.codex/packages/standalone/current. The npm package can run
+# the TUI, but remote-control commands expect this installer-owned path.
+RUN mkdir -p /home/node/.codex \
+  && curl -fsSL https://chatgpt.com/codex/install.sh | sh
 
 CMD ["codex"]

bot_bottle/contrib/codex/agent_provider.py

@@ -55,7 +55,6 @@ _RUNTIME = AgentProviderRuntime(
     prompt_mode="read_prompt_file",
     bypass_args=("--dangerously-bypass-approvals-and-sandbox",),
     resume_args=("resume", "--last"),
-    remote_control_args=(),
 )
 
 

bot_bottle/contrib/gitea/deploy_key_provisioner.py

@@ -21,6 +21,11 @@ from pathlib import Path
 
 from ...deploy_key_provisioner import DeployKeyCollisionError, DeployKeyProvisioner
 
+# Timeout for ssh-keygen and Gitea API HTTP calls. A hung Gitea instance at
+# prepare time would stall bottle launch indefinitely without this bound.
+_API_TIMEOUT_SECS = 30
+_KEYGEN_TIMEOUT_SECS = 10
+
 
 class GiteaDeployKeyProvisioner(DeployKeyProvisioner):
     """Manages deploy keys on a Gitea instance."""
@@ -46,6 +51,7 @@ class GiteaDeployKeyProvisioner(DeployKeyProvisioner):
                 check=True,
                 stdout=subprocess.DEVNULL,
                 stderr=subprocess.DEVNULL,
+                timeout=_KEYGEN_TIMEOUT_SECS,
             )
             private_key = key_path.read_bytes()
             public_key = key_path.with_suffix(".pub").read_text().strip()
@@ -67,7 +73,7 @@ class GiteaDeployKeyProvisioner(DeployKeyProvisioner):
             method="POST",
         )
         try:
-            with urllib.request.urlopen(req) as resp:
+            with urllib.request.urlopen(req, timeout=_API_TIMEOUT_SECS) as resp:
                 body = json.loads(resp.read())
         except urllib.error.HTTPError as exc:
             _body = _read_error_body(exc)
@@ -98,7 +104,7 @@ class GiteaDeployKeyProvisioner(DeployKeyProvisioner):
             method="DELETE",
         )
         try:
-            with urllib.request.urlopen(req):
+            with urllib.request.urlopen(req, timeout=_API_TIMEOUT_SECS):
                 pass
         except urllib.error.HTTPError as exc:
             if exc.code == 404:

bot_bottle/contrib/pi/agent_provider.py

@@ -166,7 +166,6 @@ _RUNTIME = AgentProviderRuntime(
     prompt_mode="append_system_prompt",
     bypass_args=(),
     resume_args=(),
-    remote_control_args=(),
 )
 
 

bot_bottle/egress.py

@@ -210,6 +210,17 @@ def egress_token_env_map(
     return out
 
 
+def _yaml_str_escape(s: str) -> str:
+    """Escape a string for use inside a YAML double-quoted scalar."""
+    return (
+        s.replace("\\", "\\\\")
+         .replace('"', '\\"')
+         .replace("\n", "\\n")
+         .replace("\r", "\\r")
+         .replace("\t", "\\t")
+    )
+
+
 def _route_to_yaml_fields(r: Route) -> dict[str, object]:
     fields: dict[str, object] = {"host": r.host}
     if r.auth_scheme and r.token_env:
@@ -272,12 +283,12 @@ def _render_match_entry(entry: dict[str, object]) -> list[str]:
         for pd in entry["paths"]:  # type: ignore[union-attr]
             pd_dict: dict[str, str] = pd  # type: ignore[assignment]
             if "type" in pd_dict:
-                lines.append(f'          - type: "{pd_dict["type"]}"')
+                lines.append(f'          - type: "{_yaml_str_escape(pd_dict["type"])}"')
-                lines.append(f'            value: "{pd_dict["value"]}"')
+                lines.append(f'            value: "{_yaml_str_escape(pd_dict["value"])}"')
             else:
-                lines.append(f'          - value: "{pd_dict["value"]}"')
+                lines.append(f'          - value: "{_yaml_str_escape(pd_dict["value"])}"')
     if "methods" in entry:
-        methods_str = ", ".join(f'"{m}"' for m in entry["methods"])  # type: ignore[union-attr]
+        methods_str = ", ".join(f'"{_yaml_str_escape(m)}"' for m in entry["methods"])  # type: ignore[union-attr]
         prefix = "      - " if first_key else "        "
         lines.append(f'{prefix}methods: [{methods_str}]')
         first_key = False
@@ -287,8 +298,8 @@ def _render_match_entry(entry: dict[str, object]) -> list[str]:
         first_key = False
         for hd in entry["headers"]:  # type: ignore[union-attr]
             hd_dict: dict[str, str] = hd  # type: ignore[assignment]
-            lines.append(f'          - name: "{hd_dict["name"]}"')
+            lines.append(f'          - name: "{_yaml_str_escape(hd_dict["name"])}"')
-            lines.append(f'            value: "{hd_dict["value"]}"')
+            lines.append(f'            value: "{_yaml_str_escape(hd_dict["value"])}"')
     if first_key:
         lines.append("      - {}")
     return lines
@@ -308,10 +319,10 @@ def egress_render_routes(
         return "\n".join(lines) + "\n"
     for r in routes:
         f = _route_to_yaml_fields(r)
-        lines.append(f'  - host: "{f["host"]}"')
+        lines.append(f'  - host: "{_yaml_str_escape(str(f["host"]))}"')
         if "auth_scheme" in f:
-            lines.append(f'    auth_scheme: "{f["auth_scheme"]}"')
+            lines.append(f'    auth_scheme: "{_yaml_str_escape(str(f["auth_scheme"]))}"')
-            lines.append(f'    token_env: "{f["token_env"]}"')
+            lines.append(f'    token_env: "{_yaml_str_escape(str(f["token_env"]))}"')
         if "matches" in f:
             lines.append("    matches:")
             for entry in f["matches"]:  # type: ignore[union-attr]
@@ -331,7 +342,7 @@ def egress_render_routes(
                     items_str = ", ".join(f'"{x}"' for x in dv)
                     lines.append(f"      {dk}: [{items_str}]")
                 elif isinstance(dv, str):
-                    lines.append(f'      {dk}: "{dv}"')
+                    lines.append(f'      {dk}: "{_yaml_str_escape(dv)}"')
     return "\n".join(lines) + "\n"
 
 

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/git_gate.py

@@ -43,10 +43,10 @@ from .manifest import ManifestBottle, ManifestGitEntry
 # Short network alias for git-gate inside the sidecar bundle. The
 # agent's `.gitconfig` insteadOf rewrites resolve through this name.
 GIT_GATE_HOSTNAME = "git-gate"
-# Bound half-open git client sessions. If an agent/tool runner is
+# Shared timeout (seconds) for all git-gate subprocess and CGI calls:
-# interrupted during push, git daemon should reap the receive-pack
+# git daemon (--timeout/--init-timeout), the access-hook subprocess in
-# child instead of keeping the gate wedged indefinitely.
+# git_http_backend, and the git http-backend CGI subprocess.
-GIT_GATE_DAEMON_TIMEOUT_SECS = 15
+GIT_GATE_TIMEOUT_SECS = 15
 
 
 @dataclass(frozen=True)
@@ -112,6 +112,15 @@ def git_gate_upstreams_for_bottle(bottle: ManifestBottle) -> tuple[GitGateUpstre
     )
 
 
+def _gitconfig_validate_value(field: str, value: str) -> None:
+    """Raise ValueError if value contains characters that break gitconfig line syntax."""
+    if "\n" in value or "\r" in value:
+        raise ValueError(
+            f"git-gate: {field} contains a newline, which would inject "
+            f"arbitrary gitconfig keys; rejecting manifest entry"
+        )
+
+
 def git_gate_render_gitconfig(
     entries: tuple[ManifestGitEntry, ...], gate_host: str, *, scheme: str = "git",
 ) -> str:
@@ -136,6 +145,7 @@ def git_gate_render_gitconfig(
         "# fetch-from-upstream-before-every-upload-pack via access-hook).\n",
     ]
     for entry in entries:
+        _gitconfig_validate_value(f"repos[{entry.Name!r}].url", entry.Upstream)
         out.append(f'[url "{scheme}://{gate_host}/{entry.Name}.git"]\n')
         out.append(f"\tinsteadOf = {entry.Upstream}\n")
         if entry.RemoteKey and entry.RemoteKey != entry.UpstreamHost:
@@ -148,6 +158,7 @@ def git_gate_render_gitconfig(
                 f"ssh://{entry.UpstreamUser}@{entry.RemoteKey}{port}/"
                 f"{entry.UpstreamPath}"
             )
+            _gitconfig_validate_value(f"repos[{entry.Name!r}].url (resolved alias)", alias)
             out.append(f"\tinsteadOf = {alias}\n")
     return "".join(out)
 
@@ -217,8 +228,8 @@ def git_gate_render_entrypoint(upstreams: tuple[GitGateUpstream, ...]) -> str:
         "",
         "exec git daemon \\",
         "  --reuseaddr \\",
-        f"  --timeout={GIT_GATE_DAEMON_TIMEOUT_SECS} \\",
+        f"  --timeout={GIT_GATE_TIMEOUT_SECS} \\",
-        f"  --init-timeout={GIT_GATE_DAEMON_TIMEOUT_SECS} \\",
+        f"  --init-timeout={GIT_GATE_TIMEOUT_SECS} \\",
         "  --base-path=/git \\",
         "  --export-all \\",
         "  --enable=receive-pack \\",

bot_bottle/git_http_backend.py

@@ -16,6 +16,8 @@ from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from pathlib import Path
 from urllib.parse import urlsplit
 
+from .git_gate import GIT_GATE_TIMEOUT_SECS
+
 
 DEFAULT_PORT = 9420
 
@@ -47,6 +49,7 @@ class GitHttpHandler(BaseHTTPRequestHandler):
                 [hook_path, "upload-pack", str(repo_dir), peer, peer],
                 capture_output=True,
                 check=False,
+                timeout=GIT_GATE_TIMEOUT_SECS,
             )
             if hook.returncode != 0:
                 detail = (hook.stderr or hook.stdout).decode(
@@ -110,6 +113,7 @@ class GitHttpHandler(BaseHTTPRequestHandler):
             env=env,
             capture_output=True,
             check=False,
+            timeout=GIT_GATE_TIMEOUT_SECS,
         )
         self._write_cgi_response(proc.stdout)
 
@@ -148,7 +152,13 @@ class GitHttpHandler(BaseHTTPRequestHandler):
             key, _, value = line.decode("latin1").partition(":")
             value = value.strip()
             if key.lower() == "status":
-                status = int(value.split()[0])
+                try:
+                    status = int(value.split()[0])
+                except (ValueError, IndexError):
+                    self.log_message(
+                        "malformed CGI Status header %r; using 500", value,
+                    )
+                    status = 500
             else:
                 headers.append((key, value))
         self.send_response(status)

bot_bottle/manifest_extends.py

@@ -49,33 +49,125 @@ def _resolve_one_bottle(
         repos_cache[name] = _resolve_repos_raw({}, child_raw)
         return bottle
 
-    if not isinstance(parent_name_raw, str):
+    # Normalize to list, accepting both str and list[str].
+    raw_list: list[object]
+    if isinstance(parent_name_raw, str):
+        raw_list = [parent_name_raw]
+    elif isinstance(parent_name_raw, list):
+        raw_list = parent_name_raw
+    else:
         raise ManifestError(
-            f"bottle '{name}' extends must be a string "
+            f"bottle '{name}' extends must be a string or list of strings "
             f"(was {type(parent_name_raw).__name__})"
         )
-    parent_name: str = parent_name_raw
+
-    if parent_name == name:
+    # Validate each entry before resolving any of them.
-        raise ManifestError(
+    parent_names: list[str] = []
-            f"bottle '{name}' extends itself; remove the "
+    for i, pname in enumerate(raw_list):
-            f"self-reference"
+        if not isinstance(pname, str):
-        )
+            raise ManifestError(
-    if parent_name not in raws:
+                f"bottle '{name}' extends[{i}] must be a string "
-        avail = ", ".join(sorted(raws.keys())) or "(none)"
+                f"(was {type(pname).__name__})"
-        raise ManifestError(
+            )
-            f"bottle '{name}' extends '{parent_name}' which is not "
+        parent_names.append(pname)
-            f"defined. Available bottles: {avail}"
+        if pname == name:
-        )
+            raise ManifestError(
-    parent = _resolve_one_bottle(
+                f"bottle '{name}' extends itself; remove the self-reference"
-        parent_name, raws, cache, repos_cache, seen + (name,)
+            )
+        if pname not in raws:
+            avail = ", ".join(sorted(raws.keys())) or "(none)"
+            raise ManifestError(
+                f"bottle '{name}' extends '{pname}' which is not "
+                f"defined. Available bottles: {avail}"
+            )
+
+    combined_parent, combined_repos_raw = _fold_parents(
+        parent_names, raws, cache, repos_cache, seen + (name,)
     )
-    merged_repos_raw = _resolve_repos_raw(repos_cache[parent_name], child_raw)
+    merged_repos_raw = _resolve_repos_raw(combined_repos_raw, child_raw)
-    bottle = _merge_bottles(parent, child_raw, merged_repos_raw, name)
+    bottle = _merge_bottles(combined_parent, child_raw, merged_repos_raw, name)
     cache[name] = bottle
     repos_cache[name] = merged_repos_raw
     return bottle
 
 
+def _fold_parents(
+    parent_names: list[str],
+    raws: dict[str, dict[str, object]],
+    cache: dict[str, ManifestBottle],
+    repos_cache: dict[str, dict[str, object]],
+    seen: tuple[str, ...],
+) -> tuple[ManifestBottle, dict[str, object]]:
+    """Resolve each parent and fold them left-to-right.
+
+    Later parents win over earlier ones on conflict.  The `seen` tuple
+    carries the current bottle's name so cycle detection works across
+    every parent edge in the multi-parent graph."""
+    first = parent_names[0]
+    effective = _resolve_one_bottle(first, raws, cache, repos_cache, seen)
+    effective_repos_raw = repos_cache[first]
+    for pname in parent_names[1:]:
+        later = _resolve_one_bottle(pname, raws, cache, repos_cache, seen)
+        later_repos_raw = repos_cache[pname]
+        effective, effective_repos_raw = _fold_two_bottles(
+            effective, effective_repos_raw, later, later_repos_raw
+        )
+    return effective, effective_repos_raw
+
+
+def _fold_two_bottles(
+    earlier: ManifestBottle,
+    earlier_repos_raw: dict[str, object],
+    later: ManifestBottle,
+    later_repos_raw: dict[str, object],
+) -> tuple[ManifestBottle, dict[str, object]]:
+    """Combine two resolved parent bottles; later wins over earlier."""
+    from .manifest import ManifestBottle, ManifestGitUser
+    from .manifest_egress import ManifestEgressConfig
+    from .manifest_git import parse_git_gate_config
+    from .manifest_util import as_json_object
+
+    merged_env = {**earlier.env, **later.env}
+
+    merged_git_user = ManifestGitUser(
+        name=later.git_user.name or earlier.git_user.name,
+        email=later.git_user.email or earlier.git_user.email,
+    )
+
+    # Repos: union by name; for same-name entries, later wins per-field.
+    # Unlike _resolve_repos_raw, an empty later_repos_raw means "no repos
+    # declared" — it does NOT clear the earlier parent's repos.
+    names = list(earlier_repos_raw) + [
+        n for n in later_repos_raw if n not in earlier_repos_raw
+    ]
+    merged_repos_raw: dict[str, object] = {
+        n: {
+            **as_json_object(earlier_repos_raw.get(n, {}), "earlier parent repo"),
+            **as_json_object(later_repos_raw.get(n, {}), "later parent repo"),
+        }
+        for n in names
+    }
+    if merged_repos_raw:
+        merged_git, _ = parse_git_gate_config("_fold", {"repos": merged_repos_raw})
+    else:
+        merged_git = ()
+
+    # Egress: routes concatenate; scalar fields use last-wins.
+    merged_egress = ManifestEgressConfig(
+        routes=earlier.egress.routes + later.egress.routes,
+        Log=later.egress.Log,
+    )
+
+    return ManifestBottle(
+        env=merged_env,
+        agent_provider=later.agent_provider,
+        git=merged_git,
+        git_user=merged_git_user,
+        egress=merged_egress,
+        supervise=later.supervise,
+    ), merged_repos_raw
+
+
 def _merge_bottles(
     parent: ManifestBottle,
     child_raw: dict[str, object],

bot_bottle/manifest_loader.py

@@ -87,5 +87,7 @@ def load_bottle_chain_from_dir(
         parent = fm.get("extends")
         if isinstance(parent, str):
             to_load.append(parent)
+        elif isinstance(parent, list):
+            to_load.extend(p for p in parent if isinstance(p, str))
 
     return resolve_bottles(raws)[bottle_name]

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
 
 
@@ -90,19 +90,19 @@ def parse_jsonrpc(body: bytes) -> JsonRpcRequest:
     try:
         raw = json.loads(body)
     except json.JSONDecodeError as e:
-        raise _RpcError(ERR_PARSE, f"parse error: {e}") from e
+        raise _RpcClientError(ERR_PARSE, f"parse error: {e}") from e
     if not isinstance(raw, dict):
-        raise _RpcError(ERR_INVALID_REQUEST, "request must be a JSON object")
+        raise _RpcClientError(ERR_INVALID_REQUEST, "request must be a JSON object")
     if raw.get("jsonrpc") != JSONRPC_VERSION:
-        raise _RpcError(ERR_INVALID_REQUEST, "jsonrpc field must be '2.0'")
+        raise _RpcClientError(ERR_INVALID_REQUEST, "jsonrpc field must be '2.0'")
     method = raw.get("method")
     if not isinstance(method, str):
-        raise _RpcError(ERR_INVALID_REQUEST, "method must be a string")
+        raise _RpcClientError(ERR_INVALID_REQUEST, "method must be a string")
     params = raw.get("params", {})
     if params is None:
         params = {}
     if not isinstance(params, dict):
-        raise _RpcError(ERR_INVALID_PARAMS, "params must be an object")
+        raise _RpcClientError(ERR_INVALID_PARAMS, "params must be an object")
     rpc_id = raw.get("id", _NO_ID)
     is_notification = rpc_id is _NO_ID
     return JsonRpcRequest(
@@ -117,12 +117,23 @@ _NO_ID = object()
 
 
 class _RpcError(Exception):
+    """Base class for all typed RPC errors that surface as JSON-RPC error responses."""
     def __init__(self, code: int, message: str):
         super().__init__(message)
         self.code = code
         self.message = message
 
 
+class _RpcClientError(_RpcError):
+    """Caller sent a bad request; returned verbatim, no server-side logging."""
+
+
+class _RpcInternalError(_RpcError):
+    """Server-side fault; logged at ERROR with cause, always returns ERR_INTERNAL."""
+    def __init__(self, message: str) -> None:
+        super().__init__(ERR_INTERNAL, message)
+
+
 def jsonrpc_result(request_id: object, result: object) -> bytes:
     payload = {"jsonrpc": JSONRPC_VERSION, "id": request_id, "result": result}
     return (json.dumps(payload) + "\n").encode("utf-8")
@@ -290,21 +301,26 @@ def validate_proposed_file(tool: str, content: str) -> None:
     catches obvious paste-errors / wrong-tool selections before they
     enter the queue."""
     if not content.strip():
-        raise _RpcError(ERR_INVALID_PARAMS, f"{tool}: proposed file is empty")
+        raise _RpcClientError(ERR_INVALID_PARAMS, f"{tool}: proposed file is empty")
     if tool == _sv.TOOL_CAPABILITY_BLOCK:
         # Dockerfiles are too varied to validate syntactically beyond
         # non-empty. The operator reads the diff in the TUI.
         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(
+            raise _RpcClientError(
                 ERR_INVALID_PARAMS,
                 f"{tool}: proposed routes.yaml is not valid: {e}",
             ) from e
+        if config.log != LOG_OFF:
+            raise _RpcClientError(
+                ERR_INVALID_PARAMS,
+                f"{tool}: proposed routes.yaml must not change egress logging",
+            )
     else:
-        raise _RpcError(ERR_INVALID_PARAMS, f"unknown tool {tool!r}")
+        raise _RpcClientError(ERR_INVALID_PARAMS, f"unknown tool {tool!r}")
 
 
 # --- MCP handlers ----------------------------------------------------------
@@ -377,17 +393,17 @@ def handle_tools_call(
     doesn't need operator approval."""
     name = params.get("name")
     if not isinstance(name, str):
-        raise _RpcError(ERR_INVALID_PARAMS, "tools/call missing 'name'")
+        raise _RpcClientError(ERR_INVALID_PARAMS, "tools/call missing 'name'")
     if name == _sv.TOOL_LIST_EGRESS_ROUTES:
         return handle_list_egress_routes(typing.cast(dict[str, object], params.get("arguments", {})), config)
 
     args_raw = params.get("arguments", {})
     if not isinstance(args_raw, dict):
-        raise _RpcError(ERR_INVALID_PARAMS, "tools/call 'arguments' must be an object")
+        raise _RpcClientError(ERR_INVALID_PARAMS, "tools/call 'arguments' must be an object")
 
     justification = args_raw.get("justification")
     if not isinstance(justification, str) or not justification.strip():
-        raise _RpcError(
+        raise _RpcClientError(
             ERR_INVALID_PARAMS,
             f"{name}: 'justification' is required and must be a non-empty string",
         )
@@ -396,13 +412,13 @@ def handle_tools_call(
         file_field = PROPOSED_FILE_FIELD[name]
         proposed_file = args_raw.get(file_field)
         if not isinstance(proposed_file, str):
-            raise _RpcError(
+            raise _RpcClientError(
                 ERR_INVALID_PARAMS,
                 f"{name}: '{file_field}' is required and must be a string",
             )
         validate_proposed_file(name, proposed_file)
     else:
-        raise _RpcError(ERR_INVALID_PARAMS, f"unknown tool {name!r}")
+        raise _RpcClientError(ERR_INVALID_PARAMS, f"unknown tool {name!r}")
 
     proposal = _sv.Proposal.new(
         bottle_slug=config.bottle_slug,
@@ -411,7 +427,10 @@ def handle_tools_call(
         justification=justification,
         current_file_hash=_sv.sha256_hex(proposed_file),
     )
-    _sv.write_proposal(config.queue_dir, proposal)
+    try:
+        _sv.write_proposal(config.queue_dir, proposal)
+    except OSError as e:
+        raise _RpcInternalError(f"failed to write proposal to queue: {e}") from e
     sys.stderr.write(
         f"supervise: queued proposal {proposal.id} ({name}) "
         f"for bottle {config.bottle_slug}; waiting for operator...\n"
@@ -431,7 +450,10 @@ def handle_tools_call(
             "content": [{"type": "text", "text": text}],
             "isError": False,
         }
-    _sv.archive_proposal(config.queue_dir, proposal.id)
+    try:
+        _sv.archive_proposal(config.queue_dir, proposal.id)
+    except OSError as e:
+        raise _RpcInternalError(f"failed to archive proposal: {e}") from e
 
     text = format_response_text(response)
     return {
@@ -507,7 +529,7 @@ class MCPHandler(http.server.BaseHTTPRequestHandler):
 
         try:
             req = parse_jsonrpc(body)
-        except _RpcError as e:
+        except _RpcClientError as e:
             self._write_jsonrpc(jsonrpc_error(None, e.code, e.message))
             return
 
@@ -515,11 +537,19 @@ class MCPHandler(http.server.BaseHTTPRequestHandler):
 
         try:
             result = self._dispatch(req, config)
-        except _RpcError as e:
+        except _RpcClientError as e:
             self._write_jsonrpc(jsonrpc_error(req.id, e.code, e.message))
             return
-        except Exception as e:  # noqa: W0718 — catch-all for RPC dispatch errors
+        except _RpcInternalError as e:
-            sys.stderr.write(f"supervise: internal error: {e}\n")
+            cause = e.__cause__
+            detail = f": {cause}" if cause else ""
+            sys.stderr.write(f"supervise: internal error: {e.message}{detail}\n")
+            sys.stderr.flush()
+            self._write_jsonrpc(jsonrpc_error(req.id, ERR_INTERNAL, "internal error"))
+            return
+        except Exception as e:  # noqa: W0718 — unexpected errors
+            sys.stderr.write(f"supervise: unexpected error: {type(e).__name__}: {e}\n")
+            sys.stderr.flush()
             self._write_jsonrpc(jsonrpc_error(req.id, ERR_INTERNAL, "internal error"))
             return
 
@@ -538,7 +568,7 @@ class MCPHandler(http.server.BaseHTTPRequestHandler):
             return handle_tools_list(req.params)
         if method == "tools/call":
             return handle_tools_call(req.params, config)
-        raise _RpcError(ERR_METHOD_NOT_FOUND, f"method not found: {method}")
+        raise _RpcClientError(ERR_METHOD_NOT_FOUND, f"method not found: {method}")
 
     def _write_jsonrpc(self, body: bytes) -> None:
         self.send_response(200)

docs/prds/prd-new-multi-parent-extends.md

@@ -0,0 +1,166 @@
+# PRD prd-new: Multi-parent `extends:` for bottles
+
+- **Status:** Draft
+- **Author:** didericis
+- **Created:** 2026-06-25
+- **Issue:** #268
+- **Extends:** PRD 0025 (`0025-bottle-extends.md`)
+
+## Summary
+
+Allow a bottle's `extends:` field to accept either a single bottle name (existing
+behavior) or a list of bottle names (new). Multiple parents are resolved
+independently and folded left-to-right into a single effective parent before the
+child is merged on top. This lets orthogonal concerns (base env, networking/egress,
+agent provider) live in separate bottles and be composed without forcing them into a
+linear chain.
+
+## Problem
+
+PRD 0025 shipped single-parent `extends:` and listed "No multi-parent inheritance"
+as a non-goal. In practice, users want to compose multiple orthogonal bottles — a
+base environment, a networking profile, and an agent-provider override — without
+creating a three-level linear chain that couples unrelated parents to each other.
+The linear chain workaround has two problems:
+
+1. **Ordering constraint.** `networking extends base` works, but then
+   `agent extends networking` can't also pick up `base` without going through
+   `networking`, coupling two unrelated concerns.
+
+2. **Quadratic duplication.** N orthogonal bottles require O(N²) chain variants
+   (one chain per permutation of applied concerns).
+
+Multi-parent `extends:` removes both constraints: each orthogonal concern stays in
+its own bottle, and the child bottle is the only place that names the combination.
+
+## Goals / Success Criteria
+
+- `extends:` accepts a list of strings in addition to a plain string.
+- Backward compat: existing single-string `extends:` is unchanged.
+- Parents are resolved left-to-right; later entries win on conflict.
+- Child wins over all parents (unchanged from PRD 0025).
+- Cycle detection covers multi-parent graphs, not just linear chains.
+- Diamond inheritance: a shared ancestor is resolved once (via the existing cache).
+- Invalid list entries (non-string, undefined bottle, self-reference) die at parse
+  with clear messages.
+- `manifest_loader.py`'s `load_bottle_chain_from_dir` enqueues all parents from a
+  list `extends:` so the resolver sees every bottle in the graph.
+
+## Non-goals
+
+- No change to the agent-vs-bottle trust boundary (PRD 0025 "Alternatives
+  considered" option 2 stays rejected).
+- No MRO / C3 linearization. Left-to-right fold is sufficient for the expected use
+  cases.
+- No preflight display of per-field provenance across multiple parents (same open
+  question as PRD 0025; remains a follow-up).
+
+## Design
+
+### Schema
+
+`extends:` now accepts either form:
+
+```yaml
+# single parent (unchanged)
+extends: base
+
+# multiple parents (new)
+extends: [base, networking]
+```
+
+Both forms are normalized to a list internally. A list with one element behaves
+identically to the string form.
+
+### Merge rules for multi-parent fold
+
+Parents are folded pairwise left-to-right before the child merge. For each step in
+the fold, the "earlier" bottle is the running accumulator and the "later" bottle is
+the next parent. Rules per field:
+
+| Field              | Fold rule                                                    |
+|--------------------|--------------------------------------------------------------|
+| `env`              | dict merge; later wins on key collision                      |
+| `git-gate.user`    | per-field overlay; later's non-empty fields win              |
+| `git-gate.repos`   | union by name; for same-name entries, later wins per-field   |
+| `egress.routes`    | concatenate (earlier first, later appended)                  |
+| `egress.log`       | later wins (last-wins)                                       |
+| `agent_provider`   | later wins (last-wins)                                       |
+| `supervise`        | later wins (last-wins)                                       |
+
+After the fold, the combined parent is merged against the child using the existing
+PRD 0025 rules (child always wins). The child's `egress.routes` appends to the
+combined parent's concatenated routes; `validate_egress_routes` runs once on the
+final merged set and catches duplicate hosts.
+
+### Algorithm
+
+```
+extends: [p1, p2, p3]
+
+fold:
+  combined = resolve(p1)
+  combined = fold_two(combined, resolve(p2))
+  combined = fold_two(combined, resolve(p3))
+
+merge:
+  result = _merge_bottles(combined, child_raw, name)
+```
+
+`fold_two(earlier, later)` applies the rules in the table above. Cycle detection
+(the `seen` tuple) is passed to each parent resolution call unchanged — if any
+parent's chain circles back to the current bottle, it is caught. The `cache` dict
+ensures a shared ancestor is only resolved once across all parents.
+
+### Error cases
+
+| Condition                              | Error message shape                                              |
+|----------------------------------------|------------------------------------------------------------------|
+| `extends` is not a string or list      | `extends must be a string or list of strings (was <type>)`       |
+| A list entry is not a string           | `extends[<i>] must be a string (was <type>)`                     |
+| A list entry names an undefined bottle | `extends '<name>' which is not defined. Available bottles: ...`  |
+| A list entry is the bottle itself      | `extends itself; remove the self-reference`                      |
+| Cycle through any parent edge          | `is in an extends cycle: <chain>`                                |
+
+## Implementation
+
+### `bot_bottle/manifest_extends.py`
+
+- `_resolve_one_bottle`: accept `str | list[str]` for `extends`; normalize to list;
+  validate each entry; for a single-entry list fall through to the existing
+  single-parent path; for multiple entries call `_fold_parents` then
+  `_merge_bottles`.
+- `_fold_parents(parent_names, raws, cache, repos_cache, seen)`: resolve each
+  parent and fold pairwise left-to-right; return `(effective_bottle,
+  effective_repos_raw)`.
+- `_fold_two_bottles(earlier, earlier_repos_raw, later, later_repos_raw)`: apply
+  the fold rules above; return `(folded_bottle, folded_repos_raw)`.
+
+### `bot_bottle/manifest_loader.py`
+
+- `load_bottle_chain_from_dir`: when `extends` is a list, enqueue all parent names
+  for loading (previously only `isinstance(parent, str)` was handled).
+
+### `tests/unit/test_manifest_extends.py`
+
+- `TestExtendsErrors.test_non_string_extends_dies`: update to use an integer
+  `extends` value (a list is now valid).
+- New class `TestExtendsMultiParent` covering all cases listed in the issue.
+
+## Testing strategy
+
+Unit tests via `ManifestIndex.from_json_obj` (same resolver surface used by all
+paths). No integration test changes needed — downstream code consumes the already-
+merged bottle and is unchanged.
+
+Test cases:
+- Two-parent list: env union, egress routes concat, git repos union
+- Last-parent-wins on scalar (supervise, agent_provider)
+- Child wins over all parents on conflict
+- Diamond: two parents share an ancestor; ancestor resolved once
+- Single-element list: identical to string form
+- Non-string extends value → ManifestError
+- Non-string list entry → ManifestError
+- Undefined bottle in list → ManifestError
+- Self-reference in list → ManifestError
+- Cycle through multi-parent edge → ManifestError

tests/unit/test_cli_start_settle.py

@@ -102,6 +102,27 @@ class TestAttachAgent(unittest.TestCase):
             bottle.argv,
         )
 
+    def test_remote_control_is_provider_startup_arg(self):
+        class Bottle:
+            argv: list[str] = []
+
+            def exec_agent(self, argv: list[str], *, tty: bool = True) -> int:
+                self.argv = list(argv)
+                return 0
+
+        bottle = Bottle()
+        exit_code = start_mod.attach_agent(
+            bottle,  # type: ignore[arg-type]
+            agent_provider_template="codex",
+            startup_args=("remote-control",),
+        )
+
+        self.assertEqual(0, exit_code)
+        self.assertEqual(
+            ["--dangerously-bypass-approvals-and-sandbox", "remote-control"],
+            bottle.argv,
+        )
+
 
 if __name__ == "__main__":
     unittest.main()

tests/unit/test_contrib_codex_provider.py

@@ -29,6 +29,9 @@ from bot_bottle.supervise import SupervisePlan
 
 
 _URL = "http://supervise:9100/"
+_CODEX_DOCKERFILE = (
+    Path(__file__).resolve().parents[2] / "bot_bottle/contrib/codex/Dockerfile"
+)
 
 
 def _make_bottle(exec_result: ExecResult | None = None) -> MagicMock:
@@ -276,6 +279,12 @@ class TestCodexProvision(unittest.TestCase):
             )
 
 
+class TestCodexDockerfile(unittest.TestCase):
+    def test_installs_procps_for_remote_control_pid_management(self):
+        dockerfile = _CODEX_DOCKERFILE.read_text()
+        self.assertIn("procps", dockerfile)
+
+
 class TestCodexSuperviseMcp(unittest.TestCase):
     def test_noop_when_supervise_disabled(self):
         bottle = _make_bottle()

tests/unit/test_contrib_gitea_deploy_key.py

@@ -10,6 +10,8 @@ from unittest.mock import MagicMock, patch
 
 from bot_bottle.contrib.gitea.deploy_key_provisioner import (
     GiteaDeployKeyProvisioner,
+    _API_TIMEOUT_SECS,
+    _KEYGEN_TIMEOUT_SECS,
     _split_owner_repo,
 )
 from bot_bottle.deploy_key_provisioner import DeployKeyCollisionError
@@ -83,6 +85,25 @@ class TestCreate(unittest.TestCase):
         self.assertEqual(str(fake_key_id), key_id)
         self.assertEqual(fake_private, private_bytes)
 
+    def test_create_passes_timeout_to_ssh_keygen_and_urlopen(self):
+        provisioner = _provisioner()
+        with patch(
+            "bot_bottle.contrib.gitea.deploy_key_provisioner.subprocess.run"
+        ) as mock_run, patch(
+            "bot_bottle.contrib.gitea.deploy_key_provisioner.urllib.request.urlopen"
+        ) as mock_urlopen, patch(
+            "bot_bottle.contrib.gitea.deploy_key_provisioner.Path.read_bytes",
+            return_value=b"PRIVATE",
+        ), patch(
+            "bot_bottle.contrib.gitea.deploy_key_provisioner.Path.read_text",
+            return_value="ssh-ed25519 AAAA\n",
+        ):
+            mock_urlopen.return_value = _urlopen_response({"id": 1})
+            provisioner.create("owner/repo", "title")
+
+        self.assertEqual(_KEYGEN_TIMEOUT_SECS, mock_run.call_args.kwargs.get("timeout"))
+        self.assertEqual(_API_TIMEOUT_SECS, mock_urlopen.call_args.kwargs.get("timeout"))
+
     def test_create_raises_on_http_error(self):
         provisioner = _provisioner()
         with patch(
@@ -139,6 +160,16 @@ class TestDelete(unittest.TestCase):
         self.assertIn("/api/v1/repos/didericis/bot-bottle/keys/99", req.full_url)
         self.assertEqual("DELETE", req.get_method())
 
+    def test_delete_passes_timeout_to_urlopen(self):
+        provisioner = _provisioner()
+        with patch(
+            "bot_bottle.contrib.gitea.deploy_key_provisioner.urllib.request.urlopen"
+        ) as mock_urlopen:
+            mock_urlopen.return_value = _urlopen_response({})
+            provisioner.delete("owner/repo", "7")
+
+        self.assertEqual(_API_TIMEOUT_SECS, mock_urlopen.call_args.kwargs.get("timeout"))
+
     def test_delete_tolerates_404(self):
         provisioner = _provisioner()
         with patch(

tests/unit/test_docker_bottle.py

@@ -136,6 +136,16 @@ class TestClaudeArgv(unittest.TestCase):
             argv,
         )
 
+    def test_codex_remote_control_startup_arg_does_not_receive_initial_prompt(self):
+        argv = _codex_bottle("/home/node/.bot-bottle-prompt.txt").agent_argv(
+            ["--dangerously-bypass-approvals-and-sandbox", "remote-control"],
+        )
+        self.assertEqual(
+            ["docker", "exec", "-it", "bot-bottle-dev-abc", "codex",
+             "--dangerously-bypass-approvals-and-sandbox", "remote-control"],
+            argv,
+        )
+
     def test_codex_resume_does_not_append_initial_prompt(self):
         argv = _codex_bottle("/home/node/.bot-bottle-prompt.txt").agent_argv(
             ["--dangerously-bypass-approvals-and-sandbox", "resume", "--last"],

tests/unit/test_docker_provision_git_user.py

@@ -31,7 +31,6 @@ class _Provider(AgentProvider):
         return AgentProviderRuntime(
             template="test", command="test", image="",
             prompt_mode="append_file", bypass_args=(), resume_args=(),
-            remote_control_args=(),
         )
     def provision_plan(self, **kwargs):  # type: ignore[override]
         raise NotImplementedError

tests/unit/test_egress.py

@@ -10,6 +10,7 @@ from bot_bottle.egress import (
     Egress,
     EgressPlan,
     EgressRoute,
+    _yaml_str_escape,
     egress_agent_env_entries,
     egress_manifest_routes,
     egress_render_routes,
@@ -322,7 +323,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 +334,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 +342,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 +371,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):
@@ -419,6 +420,76 @@ class TestRenderRoutes(unittest.TestCase):
         self.assertEqual(LOG_BLOCKS, cfg.log)
 
 
+class TestYamlStrEscape(unittest.TestCase):
+    """_yaml_str_escape produces safe YAML double-quoted scalar content."""
+
+    def test_plain_string_unchanged(self):
+        self.assertEqual("api.example.com", _yaml_str_escape("api.example.com"))
+
+    def test_double_quote_escaped(self):
+        self.assertEqual('\\"', _yaml_str_escape('"'))
+
+    def test_backslash_escaped(self):
+        self.assertEqual("\\\\", _yaml_str_escape("\\"))
+
+    def test_newline_escaped(self):
+        self.assertEqual("\\n", _yaml_str_escape("\n"))
+
+    def test_carriage_return_escaped(self):
+        self.assertEqual("\\r", _yaml_str_escape("\r"))
+
+    def test_tab_escaped(self):
+        self.assertEqual("\\t", _yaml_str_escape("\t"))
+
+    def test_combined(self):
+        self.assertEqual('\\"\\n\\\\', _yaml_str_escape('"\n\\'))
+
+
+class TestRenderRoutesEscaping(unittest.TestCase):
+    """Stray quotes/newlines in manifest strings do not corrupt routes.yaml."""
+
+    @staticmethod
+    def _parsed(routes) -> list[dict]:  # type: ignore
+        return parse_yaml_subset(egress_render_routes(routes))["routes"]  # type: ignore
+
+    def test_host_with_double_quote_round_trips(self):
+        routes = (EgressRoute(host='bad"host.example'),)
+        parsed = self._parsed(routes)
+        self.assertEqual('bad"host.example', parsed[0]["host"])
+
+    def test_host_with_newline_round_trips(self):
+        routes = (EgressRoute(host="host\nextra.example"),)
+        parsed = self._parsed(routes)
+        self.assertEqual("host\nextra.example", parsed[0]["host"])
+
+    def test_auth_scheme_with_double_quote_round_trips(self):
+        routes = (EgressRoute(
+            host="api.example",
+            auth_scheme='Bear"er',
+            token_env="EGRESS_TOKEN_0",
+        ),)
+        parsed = self._parsed(routes)
+        self.assertEqual('Bear"er', parsed[0]["auth_scheme"])
+
+    def test_path_value_with_double_quote_round_trips(self):
+        from bot_bottle.egress_addon_core import PathMatch, MatchEntry
+        routes = (EgressRoute(
+            host="api.example",
+            matches=(MatchEntry(paths=(PathMatch(type="prefix", value='/v1/"quoted"/'),)),),
+        ),)
+        parsed = self._parsed(routes)
+        self.assertEqual('/v1/"quoted"/', parsed[0]["matches"][0]["paths"][0]["value"])
+
+    def test_header_value_with_double_quote_round_trips(self):
+        from bot_bottle.egress_addon_core import HeaderMatch, MatchEntry
+        routes = (EgressRoute(
+            host="api.example",
+            matches=(MatchEntry(headers=(HeaderMatch(name="x-h", value='val"ue'),)),),
+        ),)
+        parsed = self._parsed(routes)
+        self.assertEqual('val"ue', parsed[0]["matches"][0]["headers"][0]["value"])
+
+
 class TestResolveTokenValues(unittest.TestCase):
     def test_reads_host_env(self):
         out = egress_resolve_token_values(

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_git_http_backend.py

@@ -9,6 +9,7 @@ import urllib.request
 from pathlib import Path
 from unittest import mock
 
+from bot_bottle.git_gate import GIT_GATE_TIMEOUT_SECS
 from bot_bottle.git_http_backend import GitHttpHandler, MAX_BODY_BYTES
 
 
@@ -150,6 +151,61 @@ class TestGitHttpBackend(unittest.TestCase):
             )
             self.assertEqual("git/test", env["HTTP_USER_AGENT"])
 
+    def test_subprocess_calls_include_timeout(self):
+        """Both subprocess.run calls (access-hook and git http-backend) must
+        pass timeout= so a hung upstream cannot wedge the sidecar."""
+        from http.server import ThreadingHTTPServer
+
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            (root / "repo.git").mkdir()
+
+            old_root = os.environ.get("GIT_PROJECT_ROOT")
+            os.environ["GIT_PROJECT_ROOT"] = str(root)
+            self.addCleanup(self._restore_env, old_root)
+            old_hook = os.environ.get("GIT_GATE_ACCESS_HOOK")
+            hook = root / "access-hook"
+            hook.write_text("#!/bin/sh\nexit 0\n")
+            hook.chmod(0o700)
+            os.environ["GIT_GATE_ACCESS_HOOK"] = str(hook)
+            self.addCleanup(self._restore_hook, old_hook)
+
+            server = ThreadingHTTPServer(("127.0.0.1", 0), GitHttpHandler)
+            thread = threading.Thread(target=server.serve_forever, daemon=True)
+            thread.start()
+            self.addCleanup(server.shutdown)
+            self.addCleanup(server.server_close)
+
+            backend_response = (
+                b"Status: 200 OK\r\n"
+                b"Content-Type: application/x-git-upload-pack-result\r\n"
+                b"\r\n"
+                b"0000"
+            )
+            calls = [
+                subprocess.CompletedProcess(["hook"], 0, b"", b""),
+                subprocess.CompletedProcess(["git"], 0, backend_response, b""),
+            ]
+            with mock.patch(
+                "bot_bottle.git_http_backend.subprocess.run",
+                side_effect=calls,
+            ) as run:
+                req = urllib.request.Request(
+                    f"http://127.0.0.1:{server.server_port}"
+                    "/repo.git/git-upload-pack",
+                    data=b"",
+                    method="POST",
+                )
+                with urllib.request.urlopen(req, timeout=5):
+                    pass
+
+            for call in run.call_args_list:
+                self.assertEqual(
+                    GIT_GATE_TIMEOUT_SECS,
+                    call.kwargs.get("timeout"),
+                    f"subprocess.run call missing timeout: {call}",
+                )
+
     def test_access_hook_denial_is_logged_to_stdout(self):
         """When the access-hook exits non-zero we still return 403 to the
         client, but the hook's stderr must also appear on the handler's
@@ -256,6 +312,57 @@ class TestGitHttpBackend(unittest.TestCase):
             os.environ["GIT_GATE_ACCESS_HOOK"] = value
 
 
+class TestMalformedStatusHeader(unittest.TestCase):
+    """Malformed CGI Status: headers must not propagate as unhandled exceptions;
+    the handler should fall back to HTTP 500."""
+
+    def setUp(self):
+        from http.server import ThreadingHTTPServer
+        import tempfile
+        self._tmp = tempfile.mkdtemp()
+        os.environ["GIT_PROJECT_ROOT"] = self._tmp
+        self._server = ThreadingHTTPServer(("127.0.0.1", 0), GitHttpHandler)
+        self._thread = threading.Thread(
+            target=self._server.serve_forever, daemon=True,
+        )
+        self._thread.start()
+        self._port = self._server.server_port
+
+    def tearDown(self):
+        self._server.shutdown()
+        self._server.server_close()
+        os.environ.pop("GIT_PROJECT_ROOT", None)
+        import shutil
+        shutil.rmtree(self._tmp, ignore_errors=True)
+
+    def _get_with_backend_response(self, cgi_response: bytes) -> int:
+        with mock.patch(
+            "bot_bottle.git_http_backend.subprocess.run",
+            return_value=mock.Mock(returncode=0, stdout=cgi_response),
+        ):
+            req = urllib.request.Request(
+                f"http://127.0.0.1:{self._port}/repo.git/info/refs",
+                method="GET",
+            )
+            try:
+                with urllib.request.urlopen(req, timeout=3) as resp:
+                    return resp.status
+            except urllib.error.HTTPError as e:  # type: ignore
+                return e.code
+
+    def test_empty_status_value_returns_500(self):
+        status = self._get_with_backend_response(
+            b"Status: \r\nContent-Type: text/plain\r\n\r\n"
+        )
+        self.assertEqual(500, status)
+
+    def test_non_numeric_status_returns_500(self):
+        status = self._get_with_backend_response(
+            b"Status: bad\r\nContent-Type: text/plain\r\n\r\n"
+        )
+        self.assertEqual(500, status)
+
+
 class TestContentLengthBounds(unittest.TestCase):
     """PRD 0041: malformed or oversized Content-Length is rejected before
     git http-backend is invoked."""

tests/unit/test_macos_container_util.py

@@ -73,6 +73,33 @@ resolver #2
         )
         self.assertTrue(run.call_args_list[-1].kwargs["check"])
 
+    def test_build_image_anchors_relative_dockerfile_to_context(self):
+        status = util.subprocess.CompletedProcess(
+            args=[],
+            returncode=0,
+            stdout=(
+                '[{"status":{"state":"running"},'
+                '"configuration":{"dns":{"nameservers":["9.9.9.9"]}}}]'
+            ),
+            stderr="",
+        )
+        with patch.object(util.subprocess, "run", return_value=status) as run, \
+             patch.object(util.os, "environ", {
+                 "BOT_BOTTLE_MACOS_CONTAINER_DNS": "9.9.9.9",
+             }):
+            util.build_image(
+                "bot-bottle-sidecars:latest",
+                "/repo",
+                dockerfile="Dockerfile.sidecars",
+            )
+        self.assertEqual(
+            [
+                "container", "build", "-t", "bot-bottle-sidecars:latest",
+                "--dns", "9.9.9.9", "-f", "/repo/Dockerfile.sidecars", "/repo",
+            ],
+            run.call_args_list[-1].args[0],
+        )
+
     def test_commit_container_execs_tar_and_builds_image(self):
         # stderr is bytes because subprocess.run uses stderr=PIPE without text=True
         completed = util.subprocess.CompletedProcess(

tests/unit/test_manifest_extends.py

@@ -423,9 +423,182 @@ class TestExtendsErrors(unittest.TestCase):
         )
         self.assertIn("extends cycle", msg)
 
-    def test_non_string_extends_dies(self):
+    def test_non_string_non_list_extends_dies(self):
-        msg = _error_message(_build, child={"extends": ["base"]})
+        msg = _error_message(_build, child={"extends": 123})
-        self.assertIn("extends must be a string", msg)
+        self.assertIn("extends must be a string or list of strings", msg)
+
+    def test_list_entry_non_string_dies(self):
+        msg = _error_message(_build, child={"extends": [123]})
+        self.assertIn("extends[0] must be a string", msg)
+
+
+class TestExtendsMultiParent(unittest.TestCase):
+    """extends: [p1, p2, ...] — multi-parent composition (issue #268)."""
+
+    _GIT_A = {"url": "ssh://git@host-a/a.git", "key": {"provider": "static", "path": "/k"}}
+    _GIT_B = {"url": "ssh://git@host-b/b.git", "key": {"provider": "static", "path": "/k"}}
+
+    def test_single_element_list_same_as_string(self):
+        m = _build(
+            base={"env": {"X": "1"}},
+            child={"extends": ["base"]},
+        )
+        self.assertEqual({"X": "1"}, dict(m.bottles["child"].env))
+
+    def test_two_parents_env_union(self):
+        m = _build(
+            p1={"env": {"A": "1"}},
+            p2={"env": {"B": "2"}},
+            child={"extends": ["p1", "p2"]},
+        )
+        self.assertEqual({"A": "1", "B": "2"}, dict(m.bottles["child"].env))
+
+    def test_two_parents_env_last_wins_on_collision(self):
+        m = _build(
+            p1={"env": {"X": "from-p1"}},
+            p2={"env": {"X": "from-p2"}},
+            child={"extends": ["p1", "p2"]},
+        )
+        self.assertEqual("from-p2", m.bottles["child"].env["X"])
+
+    def test_child_wins_over_all_parents(self):
+        m = _build(
+            p1={"env": {"X": "from-p1"}},
+            p2={"env": {"X": "from-p2"}},
+            child={"extends": ["p1", "p2"], "env": {"X": "from-child"}},
+        )
+        self.assertEqual("from-child", m.bottles["child"].env["X"])
+
+    def test_two_parents_supervise_last_wins(self):
+        m = _build(
+            p1={"supervise": False},
+            p2={"supervise": True},
+            child={"extends": ["p1", "p2"]},
+        )
+        self.assertTrue(m.bottles["child"].supervise)
+
+    def test_child_supervise_overrides_all_parents(self):
+        m = _build(
+            p1={"supervise": True},
+            p2={"supervise": True},
+            child={"extends": ["p1", "p2"], "supervise": False},
+        )
+        self.assertFalse(m.bottles["child"].supervise)
+
+    def test_two_parents_egress_routes_concatenated(self):
+        m = _build(
+            p1={"egress": {"routes": [{"host": "a.example.com"}]}},
+            p2={"egress": {"routes": [{"host": "b.example.com"}]}},
+            child={"extends": ["p1", "p2"]},
+        )
+        hosts = [r.Host for r in m.bottles["child"].egress.routes]
+        self.assertEqual(["a.example.com", "b.example.com"], hosts)
+
+    def test_child_egress_appends_after_combined_parents(self):
+        m = _build(
+            p1={"egress": {"routes": [{"host": "a.example.com"}]}},
+            p2={"egress": {"routes": [{"host": "b.example.com"}]}},
+            child={
+                "extends": ["p1", "p2"],
+                "egress": {"routes": [{"host": "c.example.com"}]},
+            },
+        )
+        hosts = [r.Host for r in m.bottles["child"].egress.routes]
+        self.assertEqual(["a.example.com", "b.example.com", "c.example.com"], hosts)
+
+    def test_two_parents_git_repos_union(self):
+        m = _build(
+            p1={"git-gate": {"repos": {"a": self._GIT_A}}},
+            p2={"git-gate": {"repos": {"b": self._GIT_B}}},
+            child={"extends": ["p1", "p2"]},
+        )
+        names = {e.Name for e in m.bottles["child"].git}
+        self.assertEqual({"a", "b"}, names)
+
+    def test_two_parents_git_same_name_later_wins_per_field(self):
+        # Both parents declare the same repo name. p2's `key` wins; p1's
+        # `host_key` is preserved because p2 doesn't override it.
+        p1_entry = {
+            "url": "ssh://git@host-a/repo.git",
+            "host_key": "ecdsa AAAA",
+            "key": {"provider": "static", "path": "/k1"},
+        }
+        p2_entry = {
+            "url": "ssh://git@host-a/repo.git",  # required, same url
+            "key": {"provider": "gitea", "forge_token_env": "TOK"},
+        }
+        m = _build(
+            p1={"git-gate": {"repos": {"repo": p1_entry}}},
+            p2={"git-gate": {"repos": {"repo": p2_entry}}},
+            child={"extends": ["p1", "p2"]},
+        )
+        entries = m.bottles["child"].git
+        self.assertEqual(1, len(entries))
+        e = entries[0]
+        self.assertEqual("ssh://git@host-a/repo.git", e.Upstream)
+        self.assertEqual("ecdsa AAAA", e.KnownHostKey)
+        self.assertEqual("gitea", e.Key.provider)
+
+    def test_p1_repos_preserved_when_p2_has_none(self):
+        m = _build(
+            p1={"git-gate": {"repos": {"a": self._GIT_A}}},
+            p2={"env": {"X": "1"}},
+            child={"extends": ["p1", "p2"]},
+        )
+        names = [e.Name for e in m.bottles["child"].git]
+        self.assertEqual(["a"], names)
+
+    def test_diamond_shared_ancestor_resolved_once(self):
+        # a <- b, a <- c; child extends [b, c]
+        # `a` must be resolved once and cached.
+        m = _build(
+            a={"env": {"FROM_A": "1"}, "supervise": False},
+            b={"extends": "a", "env": {"FROM_B": "1"}},
+            c={"extends": "a", "env": {"FROM_C": "1"}},
+            child={"extends": ["b", "c"]},
+        )
+        child = m.bottles["child"]
+        self.assertEqual("1", child.env["FROM_A"])
+        self.assertEqual("1", child.env["FROM_B"])
+        self.assertEqual("1", child.env["FROM_C"])
+        # supervise=False from `a` threads through both b and c; c is the
+        # later parent so its effective supervise (False) wins.
+        self.assertFalse(child.supervise)
+
+    def test_three_parents_env_fold_order(self):
+        m = _build(
+            p1={"env": {"X": "1", "A": "a"}},
+            p2={"env": {"X": "2", "B": "b"}},
+            p3={"env": {"X": "3", "C": "c"}},
+            child={"extends": ["p1", "p2", "p3"]},
+        )
+        env = dict(m.bottles["child"].env)
+        self.assertEqual("3", env["X"])
+        self.assertEqual("a", env["A"])
+        self.assertEqual("b", env["B"])
+        self.assertEqual("c", env["C"])
+
+    def test_undefined_bottle_in_list_dies(self):
+        msg = _error_message(
+            _build,
+            base={"env": {}},
+            child={"extends": ["base", "ghost"]},
+        )
+        self.assertIn("extends 'ghost'", msg)
+        self.assertIn("not defined", msg)
+
+    def test_self_reference_in_list_dies(self):
+        msg = _error_message(_build, child={"extends": ["child"]})
+        self.assertIn("extends itself", msg)
+
+    def test_cycle_through_multi_parent_edge_dies(self):
+        msg = _error_message(
+            _build,
+            a={"extends": ["b", "c"]},
+            b={},
+            c={"extends": "a"},
+        )
+        self.assertIn("extends cycle", msg)
 
 
 class TestExtendsAvailableInBottleKeys(unittest.TestCase):

tests/unit/test_plan_print_parity.py

@@ -130,7 +130,7 @@ def _capture_print(plan: DockerBottlePlan | SmolmachinesBottlePlan) -> list[str]
     orig = sys.stderr
     sys.stderr = buf
     try:
-        plan.print(remote_control=False)
+        plan.print()
     finally:
         sys.stderr = orig
     return buf.getvalue().splitlines()

tests/unit/test_provision_git.py

@@ -8,6 +8,7 @@ import unittest
 
 from bot_bottle.git_gate import (
     GIT_GATE_HOSTNAME,
+    _gitconfig_validate_value,
     git_gate_render_gitconfig,
 )
 from bot_bottle.manifest import ManifestIndex
@@ -90,5 +91,42 @@ class TestGitGateGitconfigRender(unittest.TestCase):
         self.assertNotIn("gitea.dideric.is", out)
 
 
+class TestGitconfigValidateValue(unittest.TestCase):
+    """_gitconfig_validate_value rejects values that would inject gitconfig keys."""
+
+    def test_normal_url_passes(self):
+        _gitconfig_validate_value("url", "ssh://git@github.com/owner/repo.git")
+
+    def test_newline_in_url_raises(self):
+        with self.assertRaises(ValueError):
+            _gitconfig_validate_value("url", "ssh://git@github.com/owner/\nrepo.git")
+
+    def test_carriage_return_in_url_raises(self):
+        with self.assertRaises(ValueError):
+            _gitconfig_validate_value("url", "ssh://git@github.com/\rrepo.git")
+
+    def test_error_message_names_field(self):
+        with self.assertRaises(ValueError, msg="error should name the field") as ctx:
+            _gitconfig_validate_value("repos['bad'].url", "ssh://host/\npath")
+        self.assertIn("repos['bad'].url", str(ctx.exception))
+
+
+class TestGitconfigRenderRejectsNewlineInUpstream(unittest.TestCase):
+    """git_gate_render_gitconfig raises on Upstream values with newlines."""
+
+    def test_newline_in_upstream_raises(self):
+        m = ManifestIndex.from_json_obj({
+            "bottles": {"dev": {"git-gate": {"repos": {
+                "evil": {
+                    "url": "ssh://git@github.com/owner/\nfake-key = injected\nrepo.git",
+                    "key": {"provider": "static", "path": "/dev/null"},
+                },
+            }}}},
+            "agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}},
+        })
+        with self.assertRaises(ValueError):
+            git_gate_render_gitconfig(m.bottles["dev"].git, GIT_GATE_HOSTNAME)
+
+
 if __name__ == "__main__":
     unittest.main()

tests/unit/test_smolmachines_provision.py

@@ -42,7 +42,6 @@ class _Provider(AgentProvider):
         return AgentProviderRuntime(
             template="test", command="test", image="",
             prompt_mode="append_file", bypass_args=(), resume_args=(),
-            remote_control_args=(),
         )
     def provision_plan(self, **kwargs):  # type: ignore[override]
         raise NotImplementedError

tests/unit/test_supervise_server.py

@@ -20,6 +20,7 @@ import supervise as _sv  # noqa: E402  # type: ignore
 
 from bot_bottle import supervise_server  # noqa: E402
 from bot_bottle.supervise_server import (
+    ERR_INTERNAL,
     ERR_INVALID_PARAMS,
     ERR_INVALID_REQUEST,
     ERR_METHOD_NOT_FOUND,
@@ -29,7 +30,9 @@ from bot_bottle.supervise_server import (
     PROPOSED_FILE_FIELD,
     ServerConfig,
     TOOL_DEFINITIONS,
+    _RpcClientError,
     _RpcError,
+    _RpcInternalError,
     _response_timeout_from_env,
     format_response_text,
     handle_initialize,
@@ -67,6 +70,74 @@ 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)
+
+
+# --- Error taxonomy --------------------------------------------------------
+
+
+class TestRpcErrorTaxonomy(unittest.TestCase):
+    def test_rpc_client_error_is_rpc_error(self):
+        e = _RpcClientError(ERR_INVALID_PARAMS, "bad param")
+        self.assertIsInstance(e, _RpcError)
+        self.assertEqual(ERR_INVALID_PARAMS, e.code)
+        self.assertEqual("bad param", e.message)
+
+    def test_rpc_internal_error_is_rpc_error(self):
+        e = _RpcInternalError("disk full")
+        self.assertIsInstance(e, _RpcError)
+        self.assertEqual(ERR_INTERNAL, e.code)
+        self.assertEqual("disk full", e.message)
+
+    def test_rpc_internal_error_preserves_cause(self):
+        cause = OSError("no space left on device")
+        try:
+            raise _RpcInternalError("failed to write") from cause
+        except _RpcInternalError as e:
+            self.assertIs(cause, e.__cause__)
+
+    def test_parse_error_is_client_error(self):
+        with self.assertRaises(_RpcClientError):
+            parse_jsonrpc(b"{bad json")
+
+    def test_validation_error_is_client_error(self):
+        with self.assertRaises(_RpcClientError):
+            validate_proposed_file(_sv.TOOL_EGRESS_ALLOW, "routes: nope\n")
+
+    def test_unknown_tool_in_tools_call_is_client_error(self):
+        config = ServerConfig(bottle_slug="dev", queue_dir=Path("/unused"))
+        with self.assertRaises(_RpcClientError) as cm:
+            handle_tools_call({"name": "no-such-tool", "arguments": {}}, config)
+        self.assertEqual(ERR_INVALID_PARAMS, cm.exception.code)
+
+
+class TestRpcInternalErrorOnIoFailure(unittest.TestCase):
+    def test_write_proposal_os_error_raises_internal(self):
+        config = ServerConfig(
+            bottle_slug="dev",
+            queue_dir=Path("/dev/null/cannot-exist"),
+        )
+        with self.assertRaises(_RpcInternalError) as cm:
+            handle_tools_call(
+                {
+                    "name": _sv.TOOL_CAPABILITY_BLOCK,
+                    "arguments": {
+                        "dockerfile": "FROM python:3.13\n",
+                        "justification": "x",
+                    },
+                },
+                config,
+            )
+        self.assertEqual(ERR_INTERNAL, cm.exception.code)
+        self.assertIsNotNone(cm.exception.__cause__)
+
 
 # --- JSON-RPC parsing ------------------------------------------------------
 
@@ -460,6 +531,26 @@ class TestHttpEndToEnd(unittest.TestCase):
         )
         self.assertEqual(ERR_METHOD_NOT_FOUND, result["error"]["code"])  # type: ignore[index]
 
+    def test_internal_error_returns_err_internal_over_http(self):
+        with patch.object(
+            supervise_server._sv, "write_proposal",
+            side_effect=OSError("disk full"),
+        ):
+            result = self._post_jsonrpc({
+                "jsonrpc": "2.0",
+                "id": 99,
+                "method": "tools/call",
+                "params": {
+                    "name": _sv.TOOL_CAPABILITY_BLOCK,
+                    "arguments": {
+                        "dockerfile": "FROM python:3.13\n",
+                        "justification": "x",
+                    },
+                },
+            })
+        self.assertIn("error", result)
+        self.assertEqual(ERR_INTERNAL, result["error"]["code"])  # type: ignore[index]
+
     def test_health_endpoint(self):
         conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=5)
         try: