feat: add ripgrep to agent images

Closes #258.

`egress_render_routes` and `_render_match_entry` now pass all manifest
strings (host, auth_scheme, token_env, path/header values) through
`_yaml_str_escape` before interpolating into double-quoted YAML scalars,
preventing stray `"` or newlines from corrupting routes.yaml.

`git_gate_render_gitconfig` now calls `_gitconfig_validate_value` on
each Upstream value (and the derived alias) before writing the
`insteadOf` line, rejecting any value containing a newline that would
inject arbitrary gitconfig keys.

.coveragerc

@@ -1,9 +0,0 @@
-[run]
-branch = True
-source = .
-
-[report]
-omit =
-    bot_bottle/egress_addon.py
-    bot_bottle/cli/tui.py
-    tests/*

.gitea/workflows/test.yml

@@ -39,14 +39,8 @@ jobs:
         with:
           python-version: "3.12"
 
-      - name: Install dev requirements
-        run: python3 -m pip install -r requirements-dev.txt
-
       - name: Run unit tests
-        run: python3 -m coverage run -m unittest discover -t . -s tests/unit -v
-
-      - name: Report unit coverage
-        run: python3 -m coverage report -m
+        run: python3 -m unittest discover -t . -s tests/unit -v
 
   integration:
     runs-on: ubuntu-latest

.gitignore

@@ -22,4 +22,3 @@ venv/
 .pytest_cache/
 .mypy_cache/
 .ruff_cache/
-.coverage

bot_bottle/cli/supervise.py

@@ -2,8 +2,9 @@
 act on them (approve / modify / reject).
 
 Curses-based TUI; modify-then-approve shells out to $EDITOR. The
-Egress proposals are queued for operator review as full routes.yaml
-updates.
+approval handler wires to PRD 0016 (capability-block), which rebuilds
+the bottle Dockerfile. Egress proposals are queued for operator review
+as full routes.yaml updates.
 """
 
 from __future__ import annotations
@@ -21,6 +22,10 @@ from pathlib import Path
 
 from .. import supervise as _supervise
 from ..bottle_state import read_metadata
+# from ..backend.docker.capability_apply import (
+#     CapabilityApplyError,
+#     apply_capability_change,
+# )
 from ..backend.docker.egress_apply import (
     EgressApplyError,
     applicator as _docker_applicator,
@@ -33,6 +38,10 @@ from ..backend.smolmachines.egress_apply import (
 )
 from ..log import Die, error, info
 
+
+class CapabilityApplyError(RuntimeError):
+    """Placeholder while capability_apply is disabled."""
+
 from ..supervise import (
     COMPONENT_FOR_TOOL,
     AuditEntry,
@@ -41,10 +50,12 @@ from ..supervise import (
     STATUS_APPROVED,
     STATUS_MODIFIED,
     STATUS_REJECTED,
+    TOOL_CAPABILITY_BLOCK,
     TOOL_EGRESS_ALLOW,
     TOOL_EGRESS_BLOCK,
     TOOL_GITLEAKS_ALLOW,
     TOOL_EGRESS_TOKEN_ALLOW,
+    archive_proposal,
     list_pending_proposals,
     render_diff,
     write_audit_entry,
@@ -72,7 +83,7 @@ class QueuedProposal:
 # Errors any remediation engine may raise. Caught by the TUI key
 # handlers and surfaced in the status line so a failed apply keeps
 # the proposal pending rather than crashing curses.
-ApplyError = (EgressApplyError,)
+ApplyError = (CapabilityApplyError, EgressApplyError)
 
 
 def apply_routes_change(slug: str, content: str) -> tuple[str, str]:
@@ -132,6 +143,8 @@ def _detail_lines(
 
 
 def _suffix_for_tool(tool: str) -> str:
+    if tool == TOOL_CAPABILITY_BLOCK:
+        return ".dockerfile"
     if tool in (TOOL_EGRESS_ALLOW, TOOL_EGRESS_BLOCK):
         return ".yaml"
     if tool in (TOOL_GITLEAKS_ALLOW, TOOL_EGRESS_TOKEN_ALLOW):
@@ -153,6 +166,17 @@ def approve(
     file_to_apply = final_file if final_file is not None else qp.proposal.proposed_file
 
     diff_before, diff_after = "", ""
+    # if qp.proposal.tool == TOOL_CAPABILITY_BLOCK:
+    #     _meta = read_metadata(qp.proposal.bottle_slug)
+    #     if _meta is not None and not _meta.compose_project:
+    #         raise CapabilityApplyError(
+    #             "capability-block remediation is not supported for smolmachines "
+    #             "bottles. Reject this proposal or handle the capability change "
+    #             "manually, then restart the bottle."
+    #         )
+    #     diff_before, diff_after = apply_capability_change(
+    #         qp.proposal.bottle_slug, file_to_apply,
+    #     )
     if qp.proposal.tool in (TOOL_EGRESS_ALLOW, TOOL_EGRESS_BLOCK):
         diff_before, diff_after = apply_routes_change(
             qp.proposal.bottle_slug,
@@ -170,6 +194,9 @@ def approve(
         qp, action=status, notes=notes,
         diff_before=diff_before, diff_after=diff_after,
     )
+    if qp.proposal.tool == TOOL_CAPABILITY_BLOCK:
+        archive_proposal(qp.queue_dir, qp.proposal.id)
+
 
 def reject(qp: QueuedProposal, *, reason: str) -> None:
     """Write a rejection response and an audit entry."""
@@ -319,7 +346,7 @@ def _list_once() -> int:
     return 0
 
 
-def _try_init_green() -> int:  # pragma: no cover
+def _try_init_green() -> int:
     """Initialise a green color pair and return its attr, or 0."""
     try:
         curses.start_color()
@@ -330,7 +357,7 @@ def _try_init_green() -> int:  # pragma: no cover
         return 0
 
 
-def _main_loop(stdscr: "curses._CursesWindow") -> None:  # type: ignore  # pragma: no cover
+def _main_loop(stdscr: "curses._CursesWindow") -> None:  # type: ignore
     curses.curs_set(0)
     stdscr.timeout(_REFRESH_INTERVAL_MS)
     green_attr = _try_init_green()
@@ -420,7 +447,7 @@ def _render(
     status_line: str,
     *,
     green_attr: int = 0,  # noqa: F841 — unused, but required by interface
-) -> None:  # pragma: no cover
+) -> None:
     stdscr.erase()
     h, w = stdscr.getmaxyx()
     header = f"bot-bottle supervise  ({len(pending)} pending)"
@@ -471,7 +498,7 @@ def _detail_view(
     qp: QueuedProposal,
     *,
     green_attr: int = 0,
-) -> None:  # pragma: no cover
+) -> None:
     """Render the full proposal. Scrollable. Press q to return."""
     lines = _detail_lines(qp, green_attr=green_attr)
     offset = 0
@@ -523,7 +550,7 @@ def _detail_view(
             return
 
 
-def _modify(stdscr: "curses._CursesWindow", qp: QueuedProposal) -> str | None:  # type: ignore  # pragma: no cover
+def _modify(stdscr: "curses._CursesWindow", qp: QueuedProposal) -> str | None:  # type: ignore
     """Suspend curses, open $EDITOR on the proposed file, return edited content."""
     suffix = _suffix_for_tool(qp.proposal.tool)
     curses.endwin()
@@ -534,7 +561,7 @@ def _modify(stdscr: "curses._CursesWindow", qp: QueuedProposal) -> str | None:
     return edited
 
 
-def _prompt(stdscr: "curses._CursesWindow", label: str) -> str:  # type: ignore  # pragma: no cover
+def _prompt(stdscr: "curses._CursesWindow", label: str) -> str:  # type: ignore
     """One-line input at the bottom of the screen."""
     curses.curs_set(1)
     h, _ = stdscr.getmaxyx()

bot_bottle/contrib/claude/Dockerfile

@@ -21,7 +21,7 @@ FROM node:22-slim
 # to it) works against egress's bumped TLS without the agent needing
 # local DNS.
 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 ripgrep \
   && rm -rf /var/lib/apt/lists/*
 
 # App-specific deps. Python isn't required by claude-code itself

bot_bottle/contrib/codex/Dockerfile

@@ -6,7 +6,7 @@
 FROM node:22-slim
 
 RUN apt-get update \
-  && apt-get install -y --no-install-recommends git ca-certificates curl procps \
+  && apt-get install -y --no-install-recommends git ca-certificates curl procps ripgrep \
   && rm -rf /var/lib/apt/lists/*
 
 # App-specific deps. Python isn't required by codex itself

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/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'            value: "{pd_dict["value"]}"')
+                lines.append(f'          - type: "{_yaml_str_escape(pd_dict["type"])}"')
+                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'            value: "{hd_dict["value"]}"')
+            lines.append(f'          - name: "{_yaml_str_escape(hd_dict["name"])}"')
+            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'    token_env: "{f["token_env"]}"')
+            lines.append(f'    auth_scheme: "{_yaml_str_escape(str(f["auth_scheme"]))}"')
+            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/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
-# interrupted during push, git daemon should reap the receive-pack
-# child instead of keeping the gate wedged indefinitely.
-GIT_GATE_DAEMON_TIMEOUT_SECS = 15
+# Shared timeout (seconds) for all git-gate subprocess and CGI calls:
+# git daemon (--timeout/--init-timeout), the access-hook subprocess in
+# git_http_backend, and the git http-backend CGI subprocess.
+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"  --init-timeout={GIT_GATE_DAEMON_TIMEOUT_SECS} \\",
+        f"  --timeout={GIT_GATE_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/supervise_server.py

@@ -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,7 +301,7 @@ 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.
@@ -299,17 +310,17 @@ def validate_proposed_file(tool: str, content: str) -> None:
         try:
             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 _RpcError(
+            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 ----------------------------------------------------------
@@ -382,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",
         )
@@ -401,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,
@@ -416,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"
@@ -436,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 {
@@ -512,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
 
@@ -520,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
-            sys.stderr.write(f"supervise: internal error: {e}\n")
+        except _RpcInternalError as e:
+            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
 
@@ -543,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)

requirements-dev.txt

@@ -4,4 +4,3 @@
 
 pylint>=3.0.0
 pyright>=1.1.300
-coverage>=7.0.0

tests/integration/test_sandbox_escape.py

@@ -92,9 +92,9 @@ class TestSandboxEscape(unittest.TestCase):
                     "on PATH: curl -sSL https://smolmachines.com/install.sh | sh"
                 )
 
-        # Throwaway static key for the git-gate fixture. It need not
-        # be a real SSH key: test 5 reaches gitleaks before any SSH
-        # attempt anyway.
+        # Throwaway "identity file" for the git-gate's `identity` field.
+        # It need not be a real SSH key: test 5 reaches gitleaks before
+        # any SSH attempt anyway.
         fd, kp = tempfile.mkstemp(prefix="sandbox-test-key.")
         os.close(fd)
         cls._key_path = Path(kp)
@@ -123,10 +123,7 @@ class TestSandboxEscape(unittest.TestCase):
                     "git-gate": {"repos": {
                         "throwaway": {
                             "url": "ssh://git@unreachable.invalid:22/throwaway.git",
-                            "key": {
-                                "provider": "static",
-                                "path": str(cls._key_path),
-                            },
+                            "identity": str(cls._key_path),
                         },
                     }},
                 },

tests/integration/test_smolmachines_launch.py

@@ -198,7 +198,6 @@ class TestSmolmachinesLaunch(unittest.TestCase):
         # connect fails, which is the property chunk 3 will
         # preserve once egress is actually running.
         r = self.bottle.exec(
-            "env -u HTTPS_PROXY -u HTTP_PROXY -u https_proxy -u http_proxy "
             f"curl -s --show-error --max-time 3 http://{self.plan.bundle_ip}:9099 "
             "2>&1 || true"
         )

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_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,
@@ -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_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_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_supervise_server.py

@@ -50,15 +50,15 @@ from bot_bottle.supervise_server import (
 
 
 class TestValidation(unittest.TestCase):
+    def test_capability_block_accepts_anything_nonempty(self):
+        validate_proposed_file(
+            _sv.TOOL_CAPABILITY_BLOCK,
+            "FROM python:3.13\nRUN apk add git\n",
+        )
+
     def test_empty_proposed_file_rejected_for_tools_with_file_field(self):
         with self.assertRaises(_RpcError):
-            validate_proposed_file(_sv.TOOL_EGRESS_ALLOW, "   \n\t")
-
-    def test_capability_block_rejected_as_unknown_tool(self):
-        with self.assertRaises(_RpcError) as cm:
-            validate_proposed_file("capability-block", "FROM python:3.13\n")
-        self.assertEqual(ERR_INVALID_PARAMS, cm.exception.code)
-        self.assertIn("unknown tool", cm.exception.message)
+            validate_proposed_file(_sv.TOOL_CAPABILITY_BLOCK, "   \n\t")
 
     def test_egress_routes_yaml_is_validated(self):
         validate_proposed_file(
@@ -127,9 +127,9 @@ class TestRpcInternalErrorOnIoFailure(unittest.TestCase):
         with self.assertRaises(_RpcInternalError) as cm:
             handle_tools_call(
                 {
-                    "name": _sv.TOOL_EGRESS_ALLOW,
+                    "name": _sv.TOOL_CAPABILITY_BLOCK,
                     "arguments": {
-                        "routes_yaml": "routes:\n  - host: example.com\n",
+                        "dockerfile": "FROM python:3.13\n",
                         "justification": "x",
                     },
                 },
@@ -219,6 +219,7 @@ class TestHandleToolsList(unittest.TestCase):
         self.assertEqual(
             sorted([
                 _sv.TOOL_EGRESS_ALLOW,
+                _sv.TOOL_CAPABILITY_BLOCK,
                 _sv.TOOL_EGRESS_BLOCK,
                 _sv.TOOL_LIST_EGRESS_ROUTES,
             ]),
@@ -294,10 +295,10 @@ class TestHandleToolsCall(unittest.TestCase):
         try:
             result = handle_tools_call(
                 {
-                    "name": _sv.TOOL_EGRESS_BLOCK,
+                    "name": _sv.TOOL_CAPABILITY_BLOCK,
                     "arguments": {
-                        "routes_yaml": "routes:\n  - host: example.com\n",
-                        "justification": "need example.com",
+                        "dockerfile": "FROM python:3.13\n",
+                        "justification": "need git",
                     },
                 },
                 self.config,
@@ -334,9 +335,9 @@ class TestHandleToolsCall(unittest.TestCase):
         try:
             result = handle_tools_call(
                 {
-                    "name": _sv.TOOL_EGRESS_ALLOW,
+                    "name": _sv.TOOL_CAPABILITY_BLOCK,
                     "arguments": {
-                        "routes_yaml": "routes:\n  - host: example.com\n",
+                        "dockerfile": "FROM python:3.13\n",
                         "justification": "needed for tests",
                     },
                 },
@@ -358,52 +359,20 @@ class TestHandleToolsCall(unittest.TestCase):
         with self.assertRaises(_RpcError):
             handle_tools_call(
                 {
-                    "name": _sv.TOOL_EGRESS_ALLOW,
-                    "arguments": {"routes_yaml": "routes:\n  - host: example.com\n"},
+                    "name": _sv.TOOL_CAPABILITY_BLOCK,
+                    "arguments": {"dockerfile": "FROM python:3.13\n"},
                 },
                 self.config,
             )
 
-    def test_missing_name_raises(self):
-        with self.assertRaises(_RpcError) as cm:
-            handle_tools_call({"arguments": {}}, self.config)
-        self.assertEqual(ERR_INVALID_PARAMS, cm.exception.code)
-
-    def test_arguments_must_be_object(self):
-        with self.assertRaises(_RpcError) as cm:
-            handle_tools_call(
-                {
-                    "name": _sv.TOOL_EGRESS_ALLOW,
-                    "arguments": [],
-                },
-                self.config,
-            )
-        self.assertEqual(ERR_INVALID_PARAMS, cm.exception.code)
-        self.assertIn("must be an object", cm.exception.message)
-
-    def test_capability_block_call_raises_unknown_tool(self):
-        with self.assertRaises(_RpcError) as cm:
-            handle_tools_call(
-                {
-                    "name": "capability-block",
-                    "arguments": {
-                        "dockerfile": "FROM python:3.13\n",
-                        "justification": "need git",
-                    },
-                },
-                self.config,
-            )
-        self.assertEqual(ERR_INVALID_PARAMS, cm.exception.code)
-        self.assertIn("unknown tool", cm.exception.message)
-
     def test_archives_proposal_after_response(self):
         responder = self._respond_when_proposal_appears(_sv.STATUS_APPROVED)
         try:
             handle_tools_call(
                 {
-                    "name": _sv.TOOL_EGRESS_ALLOW,
+                    "name": _sv.TOOL_CAPABILITY_BLOCK,
                     "arguments": {
-                        "routes_yaml": "routes:\n  - host: example.com\n",
+                        "dockerfile": "FROM python:3.13\n",
                         "justification": "x",
                     },
                 },
@@ -425,10 +394,10 @@ class TestHandleToolsCall(unittest.TestCase):
         )
         result = handle_tools_call(
             {
-                "name": _sv.TOOL_EGRESS_ALLOW,
+                "name": _sv.TOOL_CAPABILITY_BLOCK,
                 "arguments": {
-                    "routes_yaml": "routes:\n  - host: example.com\n",
-                    "justification": "need egress",
+                    "dockerfile": "FROM python:3.13\n",
+                    "justification": "need a capability",
                 },
             },
             config,
@@ -443,31 +412,6 @@ class TestHandleToolsCall(unittest.TestCase):
 
 
 class TestHandleListEgressRoutes(unittest.TestCase):
-    def test_success_returns_body_text(self):
-        class _Resp:
-            def __enter__(self):
-                return self
-
-            def __exit__(self, exc_type: type[BaseException] | None, exc: BaseException | None, tb: object) -> bool:
-                return False
-
-            def read(self):
-                return b"[{\"host\": \"example.com\"}]"
-
-        class _Opener:
-            def open(self, *args, **kwargs):  # noqa: ANN001, ANN002, ANN003  # type: ignore
-                return _Resp()
-
-        with patch.object(supervise_server.urllib.request, "build_opener", return_value=_Opener()):
-            result = handle_list_egress_routes(
-                {},
-                ServerConfig(bottle_slug="dev", queue_dir=Path("/unused")),
-            )
-
-        self.assertFalse(result["isError"])  # type: ignore[index]
-        text = result["content"][0]["text"]  # type: ignore[index]
-        self.assertIn("example.com", text)
-
     def test_url_error_returns_tool_error(self):
         class _Opener:
             def open(self, *args, **kwargs):  # noqa: ANN001, ANN002, ANN003  # type: ignore
@@ -527,13 +471,6 @@ class TestFormatResponseText(unittest.TestCase):
         self.assertIn("the operator modified", text.lower())
 
 
-class TestFormatPendingResponseText(unittest.TestCase):
-    def test_formats_timeout_message(self):
-        text = supervise_server.format_pending_response_text(12.5)
-        self.assertIn("status: pending", text)
-        self.assertIn("12.5s", text)
-
-
 # --- End-to-end HTTP sanity ------------------------------------------------
 
 
@@ -584,7 +521,7 @@ class TestHttpEndToEnd(unittest.TestCase):
         self.assertEqual("2.0", result["jsonrpc"])
         self.assertEqual(1, result["id"])
         names = [t["name"] for t in result["result"]["tools"]]  # type: ignore[index]
-        self.assertNotIn("capability-block", names)
+        self.assertIn(_sv.TOOL_CAPABILITY_BLOCK, names)
         self.assertIn(_sv.TOOL_EGRESS_ALLOW, names)
         self.assertIn(_sv.TOOL_EGRESS_BLOCK, names)
 
@@ -604,9 +541,9 @@ class TestHttpEndToEnd(unittest.TestCase):
                 "id": 99,
                 "method": "tools/call",
                 "params": {
-                    "name": _sv.TOOL_EGRESS_ALLOW,
+                    "name": _sv.TOOL_CAPABILITY_BLOCK,
                     "arguments": {
-                        "routes_yaml": "routes:\n  - host: example.com\n",
+                        "dockerfile": "FROM python:3.13\n",
                         "justification": "x",
                     },
                 },