claude_bottle/backend/docker/cred_proxy_apply.py

@@ -0,0 +1,123 @@
+"""Host-side helper to apply a routes.json change to a running
+cred-proxy sidecar (PRD 0014).
+
+Used by the supervise dashboard when the operator approves a
+cred-proxy-block proposal (or runs the operator-initiated `routes
+edit <bottle>` verb). Fetches the current routes.json via `docker
+exec cat`, validates the new JSON, writes it into the sidecar via
+`docker cp`, then `docker kill --signal HUP` to make the in-sidecar
+SIGHUP handler (PRD 0014 Phase 1) reload without dropping
+connections.
+
+Raises CredProxyApplyError on any failure — the dashboard surfaces
+the message and keeps the proposal pending so the operator can
+retry.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import subprocess
+import tempfile
+from pathlib import Path
+
+from .cred_proxy import (
+    CRED_PROXY_ROUTES_IN_CONTAINER,
+    cred_proxy_container_name,
+)
+
+
+class CredProxyApplyError(RuntimeError):
+    """Raised when fetch / apply fails. Caller renders to the
+    operator; does not crash the dashboard."""
+
+
+def fetch_current_routes(slug: str) -> str:
+    """Read the live routes.json from the running cred-proxy sidecar
+    for `slug`. Returns the file content as a string. Raises
+    CredProxyApplyError if the sidecar isn't reachable or the read
+    fails."""
+    container = cred_proxy_container_name(slug)
+    r = subprocess.run(
+        ["docker", "exec", container, "cat", CRED_PROXY_ROUTES_IN_CONTAINER],
+        capture_output=True, text=True, check=False,
+    )
+    if r.returncode != 0:
+        raise CredProxyApplyError(
+            f"could not read routes.json from {container}: "
+            f"{(r.stderr or '').strip() or 'container not running?'}"
+        )
+    return r.stdout
+
+
+def validate_routes_json(content: str) -> None:
+    """Syntactic check before SIGHUP — the sidecar's reload also
+    validates, but failing here keeps the old routes live and gives
+    the operator a clearer error than 'reload failed' in the
+    sidecar logs."""
+    try:
+        parsed = json.loads(content)
+    except json.JSONDecodeError as e:
+        raise CredProxyApplyError(
+            f"proposed routes.json is not valid JSON: {e}"
+        ) from e
+    if not isinstance(parsed, dict) or not isinstance(parsed.get("routes"), list):
+        raise CredProxyApplyError(
+            "proposed routes.json must be an object with a 'routes' array"
+        )
+
+
+def apply_routes_change(slug: str, new_content: str) -> tuple[str, str]:
+    """Apply `new_content` to the cred-proxy sidecar for `slug`:
+      1. Fetch current routes.json (for the before-diff).
+      2. Validate the new JSON.
+      3. Write to a temp file, `docker cp` into the sidecar.
+      4. `docker kill --signal HUP` so cred-proxy reloads.
+
+    Returns (before, after) where `after` == `new_content`. Raises
+    CredProxyApplyError on any step; the existing routes in the
+    sidecar are unchanged if the failure is before docker cp, and
+    are reverted in spirit if SIGHUP fails (cp landed but reload
+    didn't fire — caller's next attempt will SIGHUP again)."""
+    container = cred_proxy_container_name(slug)
+    before = fetch_current_routes(slug)
+    validate_routes_json(new_content)
+
+    fd, tmp_path = tempfile.mkstemp(prefix="cb-routes.", suffix=".json")
+    try:
+        with os.fdopen(fd, "w") as f:
+            f.write(new_content)
+        cp = subprocess.run(
+            ["docker", "cp", tmp_path, f"{container}:{CRED_PROXY_ROUTES_IN_CONTAINER}"],
+            capture_output=True, text=True, check=False,
+        )
+        if cp.returncode != 0:
+            raise CredProxyApplyError(
+                f"failed to copy routes.json into {container}: "
+                f"{(cp.stderr or '').strip()}"
+            )
+        sig = subprocess.run(
+            ["docker", "kill", "--signal", "HUP", container],
+            capture_output=True, text=True, check=False,
+        )
+        if sig.returncode != 0:
+            raise CredProxyApplyError(
+                f"failed to SIGHUP {container}: "
+                f"{(sig.stderr or '').strip()}"
+            )
+    finally:
+        try:
+            Path(tmp_path).unlink()
+        except OSError:
+            pass
+
+    return before, new_content
+
+
+__all__ = [
+    "CredProxyApplyError",
+    "apply_routes_change",
+    "fetch_current_routes",
+    "validate_routes_json",
+]

claude_bottle/cli/dashboard.py

@@ -22,6 +22,11 @@ from datetime import datetime, timezone
 from pathlib import Path
 
 from .. import supervise as _supervise
+from ..backend.docker.cred_proxy_apply import (
+    CredProxyApplyError,
+    apply_routes_change,
+    fetch_current_routes,
+)
 from ..log import info
 from ..supervise import (
     ACTION_OPERATOR_EDIT,
@@ -33,6 +38,7 @@ from ..supervise import (
     STATUS_MODIFIED,
     STATUS_REJECTED,
     TOOL_CAPABILITY_BLOCK,
+    TOOL_CRED_PROXY_BLOCK,
     list_pending_proposals,
     render_diff,
     write_audit_entry,
@@ -52,6 +58,33 @@ class QueuedProposal:
     queue_dir: Path
 
 
+def discover_cred_proxy_slugs() -> list[str]:
+    """Slugs of bottles with a running cred-proxy sidecar. Used by
+    the operator-initiated `routes edit` verb to know which bottles
+    are editable. Empty list if docker isn't reachable or not
+    installed."""
+    try:
+        r = subprocess.run(
+            [
+                "docker", "ps",
+                "--filter", "name=^claude-bottle-cred-proxy-",
+                "--format", "{{.Names}}",
+            ],
+            capture_output=True, text=True, check=False,
+        )
+    except FileNotFoundError:
+        return []
+    if r.returncode != 0:
+        return []
+    prefix = "claude-bottle-cred-proxy-"
+    out: list[str] = []
+    for line in (r.stdout or "").splitlines():
+        line = line.strip()
+        if line.startswith(prefix):
+            out.append(line[len(prefix):])
+    return sorted(out)
+
+
 def discover_pending() -> list[QueuedProposal]:
     """Walk ~/.claude-bottle/queue/* and collect pending proposals
     from every bottle's queue. Sorted by arrival time across the
@@ -78,9 +111,28 @@ def approve(
     notes: str = "",
     final_file: str | None = None,
 ) -> None:
-    """Write an approval response and an audit entry. If `final_file`
-    is provided the status is `modified`; otherwise `approved`."""
+    """Apply the proposal to the running sidecar, write the response
+    file the agent's tool call is waiting on, and append an audit
+    entry. If `final_file` is provided the status is `modified`;
+    otherwise `approved`.
+
+    Raises CredProxyApplyError if the cred-proxy-block apply fails
+    (sidecar down, invalid JSON survived the operator's modify).
+    On failure no response is written and no audit entry is
+    appended — the proposal stays pending so the operator can fix
+    the input and retry."""
     status = STATUS_MODIFIED if final_file is not None else STATUS_APPROVED
+    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_CRED_PROXY_BLOCK:
+        diff_before, diff_after = apply_routes_change(
+            qp.proposal.bottle_slug, file_to_apply,
+        )
+    # pipelock-block + capability-block remediation lands in PRDs
+    # 0015 + 0016; for 0014 they remain no-op approvals and the
+    # audit diff stays empty.
+
     response = Response(
         proposal_id=qp.proposal.id,
         status=status,
@@ -88,11 +140,16 @@ def approve(
         final_file=final_file,
     )
     write_response(qp.queue_dir, response)
-    _write_audit(qp, action=status, notes=notes, final_file=final_file)
+    _write_audit(
+        qp, action=status, notes=notes,
+        diff_before=diff_before, diff_after=diff_after,
+    )
 
 
 def reject(qp: QueuedProposal, *, reason: str) -> None:
-    """Write a rejection response and an audit entry."""
+    """Write a rejection response and an audit entry. No remediation
+    apply happens on reject — the agent sees the rejection and
+    decides whether to retry / give up."""
     response = Response(
         proposal_id=qp.proposal.id,
         status=STATUS_REJECTED,
@@ -100,7 +157,28 @@ def reject(qp: QueuedProposal, *, reason: str) -> None:
         final_file=None,
     )
     write_response(qp.queue_dir, response)
-    _write_audit(qp, action=STATUS_REJECTED, notes=reason, final_file=None)
+    _write_audit(qp, action=STATUS_REJECTED, notes=reason, diff_before="", diff_after="")
+
+
+def operator_edit_routes(slug: str, new_content: str) -> tuple[str, str]:
+    """Apply an operator-initiated routes.json change (no agent
+    proposal). Used by the `routes edit <bottle>` TUI verb and
+    available for scripted use. Returns (before, after) like
+    apply_routes_change. Writes an audit entry tagged
+    ACTION_OPERATOR_EDIT to distinguish from tool-call approvals.
+
+    Raises CredProxyApplyError on failure."""
+    before, after = apply_routes_change(slug, new_content)
+    write_audit_entry(AuditEntry(
+        timestamp=datetime.now(timezone.utc).isoformat(),
+        bottle_slug=slug,
+        component="cred-proxy",
+        operator_action=ACTION_OPERATOR_EDIT,
+        operator_notes="",
+        justification="",
+        diff=render_diff(before, after, label="cred-proxy"),
+    ))
+    return before, after
 
 
 def _write_audit(
@@ -108,19 +186,21 @@ def _write_audit(
     *,
     action: str,
     notes: str,
-    final_file: str | None,
+    diff_before: str,
+    diff_after: str,
 ) -> None:
     """Audit log for cred-proxy / pipelock tools. capability-block has
     no audit log (its changes are captured by the bottle's rebuild
-    record + git history per PRD 0016)."""
+    record + git history per PRD 0016).
+
+    For cred-proxy-block approvals the (before, after) come from the
+    apply_routes_change return — a real fetched-from-sidecar diff.
+    For rejections, or for tools whose remediation hasn't landed yet
+    (pipelock in 0014, capability anywhere), both are empty strings
+    and the audit diff renders as empty."""
     component = COMPONENT_FOR_TOOL.get(qp.proposal.tool)
     if component is None:
-        # capability-block: skip audit log; 0016 records via rebuild.
         return
-    # v1 audit diff is empty: 0013's no-op handler doesn't have the
-    # actual current-on-disk file to diff against, only the agent's
-    # proposed file. 0014 / 0015 fill in the real diff against the
-    # live routes.json / allowlist after writing the change.
     write_audit_entry(AuditEntry(
         timestamp=datetime.now(timezone.utc).isoformat(),
         bottle_slug=qp.proposal.bottle_slug,
@@ -128,11 +208,7 @@ def _write_audit(
         operator_action=action,
         operator_notes=notes,
         justification=qp.proposal.justification,
-        diff=render_diff(
-            "",
-            final_file if final_file is not None else qp.proposal.proposed_file,
-            label=component,
-        ),
+        diff=render_diff(diff_before, diff_after, label=component),
     ))
 
 
@@ -217,6 +293,9 @@ def _main_loop(stdscr: "curses._CursesWindow") -> None:
 
         if key in (ord("q"), 27):  # q or ESC
             return
+        if key == ord("e"):
+            status_line = _operator_edit_routes_flow(stdscr)
+            continue
         if not pending:
             continue
         qp = pending[selected]
@@ -228,15 +307,21 @@ def _main_loop(stdscr: "curses._CursesWindow") -> None:
         elif key in (curses.KEY_ENTER, 10, 13, ord("v")):
             _detail_view(stdscr, qp)
         elif key == ord("a"):
-            approve(qp)
-            status_line = f"approved {qp.proposal.tool} for [{qp.proposal.bottle_slug}]"
+            try:
+                approve(qp)
+                status_line = f"approved {qp.proposal.tool} for [{qp.proposal.bottle_slug}]"
+            except CredProxyApplyError as e:
+                status_line = f"apply failed: {e}"
         elif key == ord("m"):
             edited = _modify(stdscr, qp)
             if edited is None:
                 status_line = "modify aborted (no change)"
             else:
-                approve(qp, final_file=edited, notes="operator modified before approving")
-                status_line = f"modified+approved {qp.proposal.tool} for [{qp.proposal.bottle_slug}]"
+                try:
+                    approve(qp, final_file=edited, notes="operator modified before approving")
+                    status_line = f"modified+approved {qp.proposal.tool} for [{qp.proposal.bottle_slug}]"
+                except CredProxyApplyError as e:
+                    status_line = f"apply failed: {e}"
         elif key == ord("r"):
             reason = _prompt(stdscr, "reject reason: ")
             if reason:
@@ -280,7 +365,7 @@ def _render(
             attr = curses.A_REVERSE if i == selected else curses.A_NORMAL
             stdscr.addnstr(row, 0, line, w - 1, attr)
 
-    footer = "[Enter] view  [a] approve  [m] modify  [r] reject  [j/k] move  [q] quit"
+    footer = "[Enter] view  [a] approve  [m] modify  [r] reject  [e] routes edit  [j/k] move  [q] quit"
     stdscr.hline(h - 2, 0, curses.ACS_HLINE, w)
     stdscr.addnstr(h - 1, 0, footer, w - 1, curses.A_DIM)
     if status_line:
@@ -316,12 +401,18 @@ def _detail_view(stdscr: "curses._CursesWindow", qp: QueuedProposal) -> None:
         elif key == ord("G"):
             offset = max(0, len(lines) - 1)
         elif key == ord("a"):
-            approve(qp)
+            try:
+                approve(qp)
+            except CredProxyApplyError:
+                pass  # Status surfaces back in the list view's render.
             return
         elif key == ord("m"):
             edited = _modify(stdscr, qp)
             if edited is not None:
-                approve(qp, final_file=edited, notes="operator modified before approving")
+                try:
+                    approve(qp, final_file=edited, notes="operator modified before approving")
+                except CredProxyApplyError:
+                    pass
             return
         elif key == ord("r"):
             reason = _prompt(stdscr, "reject reason: ")
@@ -369,6 +460,40 @@ def _suffix_for_tool(tool: str) -> str:
     return ".txt"
 
 
+def _operator_edit_routes_flow(stdscr: "curses._CursesWindow") -> str:
+    """Operator-initiated routes.json edit. Discover running
+    cred-proxy sidecars, pick one (single → use directly; multi →
+    prompt), fetch the current routes, open in $EDITOR, apply on
+    save. Returns a status-line message."""
+    slugs = discover_cred_proxy_slugs()
+    if not slugs:
+        return "no running cred-proxy sidecars to edit"
+    if len(slugs) == 1:
+        slug = slugs[0]
+    else:
+        slug = _prompt(stdscr, f"bottle ({', '.join(slugs)}): ")
+        if not slug:
+            return "routes edit aborted"
+        if slug not in slugs:
+            return f"unknown bottle {slug!r}"
+    try:
+        current = fetch_current_routes(slug)
+    except CredProxyApplyError as e:
+        return f"fetch failed: {e}"
+    curses.endwin()
+    try:
+        edited = edit_in_editor(current, suffix=".json")
+    finally:
+        stdscr.refresh()
+    if edited is None:
+        return f"routes for [{slug}] unchanged"
+    try:
+        operator_edit_routes(slug, edited)
+    except CredProxyApplyError as e:
+        return f"apply failed: {e}"
+    return f"updated routes for [{slug}]"
+
+
 def _prompt(stdscr: "curses._CursesWindow", label: str) -> str:
     """One-line input at the bottom of the screen."""
     curses.curs_set(1)

claude_bottle/cred_proxy_server.py

@@ -30,6 +30,7 @@ import http.client
 import http.server
 import json
 import os
+import signal
 import socketserver
 import sys
 import typing
@@ -398,6 +399,56 @@ def load_tokens(routes: tuple[Route, ...], environ: typing.Mapping[str, str]) ->
     return out
 
 
+def reload_routes(
+    server: "CredProxyServer",
+    routes_path: str,
+    *,
+    environ: typing.Mapping[str, str] | None = None,
+) -> tuple[bool, str]:
+    """Re-read routes.json + tokens and swap them onto `server`. Used
+    by the SIGHUP handler (PRD 0014) so the operator can update the
+    routes file in-place and have cred-proxy pick up the change
+    without dropping in-flight connections.
+
+    Returns (ok, message). On failure the server's existing routes
+    stay in place — better to keep serving the old config than to
+    leave the proxy with no routes after a typo.
+
+    Atomic swap: Python attribute reassignment is atomic, and the
+    request handler reads `server.routes`/`server.tokens` once at
+    the top of `_proxy()` so an in-flight request keeps the version
+    it captured. New requests see the new routes."""
+    env = environ if environ is not None else os.environ
+    try:
+        new_routes = load_routes(routes_path)
+        new_tokens = load_tokens(new_routes, env)
+    except (OSError, ValueError, json.JSONDecodeError) as e:
+        return False, f"reload failed: {e}"
+    server.routes = new_routes
+    server.tokens = new_tokens
+    return True, (
+        f"reloaded {len(new_routes)} route(s): "
+        f"{', '.join(r.path for r in new_routes)}"
+    )
+
+
+def install_sighup_handler(server: "CredProxyServer", routes_path: str) -> None:
+    """Wire SIGHUP to reload_routes. No-op on platforms without
+    SIGHUP (Windows). The handler swallows exceptions so a bad
+    reload doesn't crash the long-lived sidecar."""
+    if not hasattr(signal, "SIGHUP"):
+        return
+
+    def handler(signum: int, frame: object) -> None:
+        del signum, frame
+        ok, message = reload_routes(server, routes_path)
+        prefix = "cred-proxy: SIGHUP " + ("ok: " if ok else "failed: ")
+        sys.stderr.write(prefix + message + "\n")
+        sys.stderr.flush()
+
+    signal.signal(signal.SIGHUP, handler)
+
+
 def serve(
     *,
     routes_path: str = DEFAULT_ROUTES_PATH,
@@ -414,6 +465,7 @@ def serve(
     server = CredProxyServer((bind, port), CredProxyHandler)
     server.routes = routes
     server.tokens = tokens
+    install_sighup_handler(server, routes_path)
     sys.stderr.write(
         f"cred-proxy listening on {bind}:{port}; "
         f"{len(routes)} route(s): "

docs/prds/0014-cred-proxy-block-remediation.md

@@ -0,0 +1,64 @@
+# PRD 0014: cred-proxy block remediation
+
+- **Status:** Draft
+- **Author:** didericis
+- **Created:** 2026-05-25
+- **Parent:** PRD 0012
+- **Depends on:** PRD 0013
+
+## Summary
+
+Wires the **cred-proxy block** path (PRD 0012 *Stuck categories*) end-to-end. cred-proxy gains SIGHUP-based hot reload of `routes.json`. The supervisor, on approval of a `cred-proxy-block` proposal, writes the new `routes.json` to the host and SIGHUPs cred-proxy — no restart, no dropped connections. The TUI gains a proactive `routes edit <bottle>` verb for operator-initiated edits unrelated to a tool call. The cred-proxy audit log (format defined in PRD 0013) is filled in with real entries on every edit.
+
+## Problem
+
+See PRD 0012. This PRD specifically addresses: with 0013 in place, the operator can approve a `cred-proxy-block` proposal but nothing happens — `routes.json` doesn't change and cred-proxy doesn't notice. This PRD closes the loop.
+
+## Goals / Success Criteria
+
+A real cred-proxy block recovers end-to-end: the agent's HTTP request fails with a 403, the agent calls `cred-proxy-block` with a proposed `routes.json` and a justification, the operator approves in the TUI, the supervisor writes the new file and SIGHUPs cred-proxy, the agent retries against the now-reloaded proxy and proceeds. In-flight connections to cred-proxy do not drop during the reload.
+
+## Non-goals
+
+- Pipelock or capability handling (covered by 0015 and 0016).
+- Auto-rotation of expired tokens. The operator decides whether to issue a new token; this PRD just delivers approved config changes to cred-proxy.
+
+## Scope
+
+### In scope
+
+- SIGHUP reload of `routes.json` on cred-proxy. ~30 lines added to the server.
+- Supervisor write path: on operator approval of a `cred-proxy-block` proposal, write the proposed `routes.json` to the host-side path cred-proxy reads, then send SIGHUP.
+- `routes edit <bottle>` TUI verb: open the bottle's `routes.json` in `$EDITOR`, write + SIGHUP on save. Not gated on a pending proposal.
+- cred-proxy audit log entries: every routes edit (from a tool-call approval or from a proactive `routes edit`) appends an entry with timestamp, diff, justification (if from tool call), and operator action.
+
+### Out of scope
+
+- Restart-based reload as a fallback. SIGHUP only.
+- Pipelock equivalents (PRD 0015).
+
+## Proposed Design
+
+### New services / components
+
+- **`routes edit <bottle>` TUI verb.** Opens the bottle's current `routes.json` in `$EDITOR`. On save, the supervisor writes the new file and SIGHUPs cred-proxy. Useful when the operator wants to add a route without waiting for an agent prompt.
+
+### Existing code touched
+
+- **cred-proxy** (PRD 0010) — gains a SIGHUP signal handler that re-reads `routes.json` without dropping connections or breaking in-flight calls.
+- **MCP sidecar** (PRD 0013) — the `cred-proxy-block` approval handler stops being a no-op; on approval, calls the supervisor's write+SIGHUP path.
+- **`cli.py`** — dashboard subcommand gains the `routes edit` verb.
+
+### Data model changes
+
+None beyond PRD 0013. The audit log format is defined there; this PRD fills it in.
+
+## Open questions
+
+- **SIGHUP race window.** An agent that retries within msec of the SIGHUP may hit old routes once before the reload completes, fail, and retry against the new routes. Assumption is that normal HTTP retry semantics absorb this; worth confirming under real usage rather than designing around it preemptively.
+
+## References
+
+- PRD 0010 — cred-proxy.
+- PRD 0012 — stuck-agent recovery flow overview.
+- PRD 0013 — supervise plane foundation (prerequisite).

tests/integration/test_cred_proxy_sighup.py

@@ -0,0 +1,223 @@
+"""Integration: SIGHUP reload + host-side apply_routes_change
+(PRD 0014).
+
+Brings up a real cred-proxy sidecar with one route, then uses
+apply_routes_change (docker cp + SIGHUP) to swap to a different
+route. Verifies cred-proxy actually serves the new routes after the
+reload (and 404s the old ones).
+
+Avoids a real upstream by routing to unreachable hostnames — the
+proxy's 502 "upstream connection failed" is a sufficient signal that
+the route matched. 404 means no route matched.
+
+apply_routes_change uses docker exec / cp / kill (not bind mounts),
+so this test should work in docker-in-docker environments too — no
+skip decorator beyond skip_unless_docker.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import shutil
+import subprocess
+import tempfile
+import time
+import unittest
+from pathlib import Path
+
+from claude_bottle.backend.docker.cred_proxy import (
+    CRED_PROXY_PORT,
+    DockerCredProxy,
+    build_cred_proxy_image,
+    cred_proxy_container_name,
+)
+from claude_bottle.backend.docker.cred_proxy_apply import (
+    CredProxyApplyError,
+    apply_routes_change,
+    fetch_current_routes,
+)
+from claude_bottle.backend.docker.network import (
+    network_create_egress,
+    network_create_internal,
+    network_remove,
+)
+from claude_bottle.cred_proxy import (
+    CRED_PROXY_HOSTNAME,
+    CredProxyPlan,
+    CredProxyRoute,
+)
+from tests._docker import skip_unless_docker
+
+
+CURL_IMAGE = "curlimages/curl:latest"
+
+
+@skip_unless_docker()
+class TestCredProxySighupReload(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        r = subprocess.run(
+            ["docker", "pull", CURL_IMAGE],
+            stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False,
+        )
+        if r.returncode != 0:
+            raise unittest.SkipTest(f"could not pull {CURL_IMAGE}")
+        build_cred_proxy_image()
+
+    def setUp(self):
+        self.slug = f"cb-test-sighup-{os.getpid()}-{int(time.time())}"
+        self.proxy_name = ""
+        self.internal_net = ""
+        self.egress_net = ""
+        self.work_dir = Path(tempfile.mkdtemp(prefix="cred-proxy-sighup."))
+        # Token value for both initial and post-SIGHUP routes — they
+        # share the same TokenRef so they share CRED_PROXY_TOKEN_0 in
+        # the container's environ.
+        os.environ["CB_SIGHUP_TEST_TOKEN"] = "test-token"
+
+    def tearDown(self):
+        os.environ.pop("CB_SIGHUP_TEST_TOKEN", None)
+        if self.proxy_name:
+            subprocess.run(
+                ["docker", "rm", "-f", self.proxy_name],
+                stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False,
+            )
+        for n in (self.internal_net, self.egress_net):
+            if n:
+                network_remove(n)
+        shutil.rmtree(self.work_dir, ignore_errors=True)
+
+    def _bring_up_with_route(self, path: str, upstream: str) -> None:
+        self.internal_net = network_create_internal(self.slug)
+        self.egress_net = network_create_egress(self.slug)
+        route = CredProxyRoute(
+            path=path,
+            upstream=upstream,
+            auth_scheme="Bearer",
+            token_env="CRED_PROXY_TOKEN_0",
+            token_ref="CB_SIGHUP_TEST_TOKEN",
+        )
+        routes_path = self.work_dir / "routes.json"
+        from claude_bottle.cred_proxy import cred_proxy_render_routes
+        routes_path.write_text(cred_proxy_render_routes((route,)))
+        routes_path.chmod(0o600)
+        plan = CredProxyPlan(
+            slug=self.slug,
+            routes_path=routes_path,
+            routes=(route,),
+            token_env_map={"CRED_PROXY_TOKEN_0": "CB_SIGHUP_TEST_TOKEN"},
+            internal_network=self.internal_net,
+            egress_network=self.egress_net,
+            # No pipelock for this test — the proxy talks directly to
+            # the egress network. Upstreams are unreachable so the
+            # 502s confirm the route table.
+        )
+        self.proxy_name = DockerCredProxy().start(plan)
+        # Wait until the proxy is serving (it's the only way I have
+        # to know python has bound to the port).
+        deadline = time.monotonic() + 10.0
+        while time.monotonic() < deadline:
+            code = self._curl("/__probe/")
+            if code in (404, 502):  # serving — either response proves it's up
+                return
+            time.sleep(0.2)
+        raise AssertionError("cred-proxy never came up")
+
+    def _curl(self, path: str) -> int | None:
+        """Return the HTTP status from a curl-in-container request to
+        the cred-proxy, or None on connection failure."""
+        r = subprocess.run(
+            [
+                "docker", "run", "--rm",
+                "--network", self.internal_net,
+                CURL_IMAGE,
+                "-sS", "-o", "/dev/null",
+                "-w", "%{http_code}",
+                "--max-time", "8",
+                f"http://{CRED_PROXY_HOSTNAME}:{CRED_PROXY_PORT}{path}",
+            ],
+            capture_output=True, text=True, check=False,
+        )
+        if r.returncode != 0:
+            return None
+        try:
+            return int(r.stdout.strip())
+        except ValueError:
+            return None
+
+    def test_sighup_swaps_routes(self):
+        """Initial route /a/ matches (502 from unreachable upstream);
+        /b/ 404s. After apply_routes_change with /b/ only, the table
+        flips: /a/ 404s, /b/ matches."""
+        self._bring_up_with_route("/a/", "https://unreachable-a.example")
+
+        self.assertEqual(502, self._curl("/a/foo"))
+        self.assertEqual(404, self._curl("/b/foo"))
+
+        new_routes = json.dumps({"routes": [{
+            "path": "/b/",
+            "upstream": "https://unreachable-b.example",
+            "auth_scheme": "Bearer",
+            "token_env": "CRED_PROXY_TOKEN_0",
+        }]}) + "\n"
+
+        before, after = apply_routes_change(self.slug, new_routes)
+        self.assertIn("/a/", before)
+        self.assertEqual(new_routes, after)
+
+        # SIGHUP propagates as a Python signal — runs at the next
+        # bytecode boundary on the main thread. Give it a moment.
+        deadline = time.monotonic() + 5.0
+        flipped = False
+        while time.monotonic() < deadline:
+            if self._curl("/a/foo") == 404 and self._curl("/b/foo") == 502:
+                flipped = True
+                break
+            time.sleep(0.2)
+        self.assertTrue(flipped, "SIGHUP reload did not propagate to the route table")
+
+    def test_in_flight_connections_survive_sighup(self):
+        """SIGHUP must reload without dropping the bound socket. The
+        signal handler runs on the main thread; in-flight worker
+        threads keep the routes they captured at request start.
+        Verified by issuing a request right after SIGHUP and seeing
+        the new route in effect (the listener never restarted)."""
+        self._bring_up_with_route("/a/", "https://unreachable.example")
+        # Fetching the current routes also proves the proxy is up.
+        current = fetch_current_routes(self.slug)
+        self.assertIn("/a/", current)
+
+        new_routes = json.dumps({"routes": [{
+            "path": "/c/",
+            "upstream": "https://unreachable-c.example",
+            "auth_scheme": "Bearer",
+            "token_env": "CRED_PROXY_TOKEN_0",
+        }]}) + "\n"
+        apply_routes_change(self.slug, new_routes)
+
+        deadline = time.monotonic() + 5.0
+        while time.monotonic() < deadline:
+            if self._curl("/c/foo") == 502:
+                return
+            time.sleep(0.2)
+        self.fail("new route not picked up after SIGHUP")
+
+    def test_apply_with_invalid_json_raises(self):
+        self._bring_up_with_route("/a/", "https://unreachable.example")
+        with self.assertRaises(CredProxyApplyError) as cm:
+            apply_routes_change(self.slug, "{not json")
+        self.assertIn("not valid JSON", str(cm.exception))
+
+    def test_apply_against_missing_sidecar_raises(self):
+        # Don't bring up the sidecar; the slug points at nothing.
+        with self.assertRaises(CredProxyApplyError):
+            apply_routes_change(
+                self.slug,
+                '{"routes": [{"path": "/x/", "upstream": "https://example.com",'
+                ' "auth_scheme": "Bearer", "token_env": "CRED_PROXY_TOKEN_0"}]}',
+            )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/test_cred_proxy_apply.py

@@ -0,0 +1,39 @@
+"""Unit: validate_routes_json (PRD 0014 Phase 2). docker exec / cp /
+kill paths are covered by the integration test."""
+
+import unittest
+
+from claude_bottle.backend.docker.cred_proxy_apply import (
+    CredProxyApplyError,
+    validate_routes_json,
+)
+
+
+class TestValidateRoutesJson(unittest.TestCase):
+    def test_accepts_routes_array(self):
+        validate_routes_json('{"routes": []}')
+        validate_routes_json(
+            '{"routes": [{"path": "/a/", "upstream": "https://example.com",'
+            ' "auth_scheme": "Bearer", "token_env": "T0"}]}'
+        )
+
+    def test_rejects_bad_json(self):
+        with self.assertRaises(CredProxyApplyError) as cm:
+            validate_routes_json("{not json")
+        self.assertIn("not valid JSON", str(cm.exception))
+
+    def test_rejects_non_object_top_level(self):
+        with self.assertRaises(CredProxyApplyError):
+            validate_routes_json("[]")
+
+    def test_rejects_missing_routes_key(self):
+        with self.assertRaises(CredProxyApplyError):
+            validate_routes_json('{"other": []}')
+
+    def test_rejects_non_list_routes(self):
+        with self.assertRaises(CredProxyApplyError):
+            validate_routes_json('{"routes": "not a list"}')
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/test_cred_proxy_server.py

@@ -1,15 +1,20 @@
 """Unit: cred-proxy server pure functions — route parsing, route
-selection, header injection (PRD 0010)."""
+selection, header injection (PRD 0010); SIGHUP reload (PRD 0014)."""
 
+import json
+import tempfile
 import unittest
+from pathlib import Path
 
 from claude_bottle.cred_proxy_server import (
+    CredProxyServer,
     Route,
     build_forward_headers,
     filter_response_headers,
     is_git_push_request,
     load_tokens,
     parse_routes,
+    reload_routes,
     select_route,
 )
 
@@ -258,5 +263,77 @@ class TestLoadTokens(unittest.TestCase):
         self.assertEqual({"T_0": ""}, out)
 
 
+class TestReloadRoutes(unittest.TestCase):
+    """SIGHUP reload helper (PRD 0014).
+
+    Drives the same code path the signal handler invokes, but
+    without actually sending a signal — keeps the test
+    deterministic. The signal binding is just `signal.signal(SIGHUP,
+    handler)`; install_sighup_handler is exercised by the
+    integration test."""
+
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory(prefix="cp-reload-test.")
+        self.routes_path = Path(self._tmp.name) / "routes.json"
+        self.routes_path.write_text(json.dumps({"routes": [
+            {"path": "/a/", "upstream": "https://a.example",
+             "auth_scheme": "Bearer", "token_env": "T0"},
+        ]}))
+        # Bind to :0 so the test doesn't need a fixed port.
+        self.server = CredProxyServer(("127.0.0.1", 0), _NullHandler)
+        self.server.routes = parse_routes(json.loads(self.routes_path.read_text()))
+        self.server.tokens = {"T0": "old"}
+
+    def tearDown(self):
+        self.server.server_close()
+        self._tmp.cleanup()
+
+    def test_reload_swaps_routes_and_tokens(self):
+        self.routes_path.write_text(json.dumps({"routes": [
+            {"path": "/a/", "upstream": "https://a.example",
+             "auth_scheme": "Bearer", "token_env": "T0"},
+            {"path": "/b/", "upstream": "https://b.example",
+             "auth_scheme": "Bearer", "token_env": "T1"},
+        ]}))
+        ok, msg = reload_routes(
+            self.server, str(self.routes_path),
+            environ={"T0": "new0", "T1": "new1"},
+        )
+        self.assertTrue(ok, msg)
+        self.assertEqual(2, len(self.server.routes))
+        self.assertEqual({"T0": "new0", "T1": "new1"}, self.server.tokens)
+        self.assertIn("reloaded 2 route(s)", msg)
+
+    def test_failed_reload_keeps_old_routes(self):
+        original_routes = self.server.routes
+        original_tokens = self.server.tokens
+        self.routes_path.write_text("not valid json {")
+        ok, msg = reload_routes(
+            self.server, str(self.routes_path),
+            environ={"T0": "ignored"},
+        )
+        self.assertFalse(ok)
+        self.assertIn("reload failed", msg)
+        self.assertIs(original_routes, self.server.routes)
+        self.assertIs(original_tokens, self.server.tokens)
+
+    def test_failed_reload_on_missing_file_keeps_old_routes(self):
+        original_routes = self.server.routes
+        self.routes_path.unlink()
+        ok, _ = reload_routes(
+            self.server, str(self.routes_path), environ={},
+        )
+        self.assertFalse(ok)
+        self.assertIs(original_routes, self.server.routes)
+
+
+class _NullHandler:  # noqa: D401 — test helper, not a real handler
+    """Dummy handler class; the reload tests never actually serve a
+    request, so the handler is never instantiated."""
+
+    def __init__(self, *args, **kwargs):
+        raise RuntimeError("should not be called in reload tests")
+
+
 if __name__ == "__main__":
     unittest.main()

tests/unit/test_dashboard.py

@@ -1,8 +1,12 @@
-"""Unit: dashboard headless paths (PRD 0013 phase 4).
+"""Unit: dashboard headless paths (PRD 0013 phase 4, PRD 0014).
 
 The curses TUI itself isn't exercised here — these tests cover the
 discovery + approve/reject + audit-write paths that the TUI's key
 handlers call into.
+
+apply_routes_change is stubbed at the dashboard module level so the
+tests don't need a running cred-proxy sidecar; the real docker
+exec/cp/SIGHUP plumbing is covered by the integration test.
 """
 
 import os
@@ -12,6 +16,7 @@ from datetime import datetime, timezone
 from pathlib import Path
 
 from claude_bottle import supervise
+from claude_bottle.backend.docker.cred_proxy_apply import CredProxyApplyError
 from claude_bottle.cli import dashboard
 from claude_bottle.supervise import (
     Proposal,
@@ -110,8 +115,15 @@ class TestDiscoverPending(_FakeHomeMixin, unittest.TestCase):
 class TestApproveReject(_FakeHomeMixin, unittest.TestCase):
     def setUp(self):
         self._setup_fake_home()
+        self._original_apply = dashboard.apply_routes_change
+        # Default stub: succeed with deterministic before/after so the
+        # audit log shows a non-empty diff.
+        dashboard.apply_routes_change = lambda slug, content: (
+            '{"routes": []}\n', content,
+        )
 
     def tearDown(self):
+        dashboard.apply_routes_change = self._original_apply
         self._teardown_fake_home()
 
     def _enqueue(self, tool: str = TOOL_CRED_PROXY_BLOCK):
@@ -166,6 +178,145 @@ class TestApproveReject(_FakeHomeMixin, unittest.TestCase):
         self.assertEqual(0, len(read_audit_entries("cred-proxy", "dev")))
 
 
+class TestCredProxyApplyWiring(_FakeHomeMixin, unittest.TestCase):
+    """PRD 0014 Phase 3: approve() on a cred-proxy-block proposal
+    must call apply_routes_change with the right args and surface
+    its failures."""
+
+    def setUp(self):
+        self._setup_fake_home()
+        self._original_apply = dashboard.apply_routes_change
+
+    def tearDown(self):
+        dashboard.apply_routes_change = self._original_apply
+        self._teardown_fake_home()
+
+    def _enqueue_cred_proxy(self, proposed: str = '{"routes": []}\n'):
+        p = Proposal.new(
+            bottle_slug="dev", tool=TOOL_CRED_PROXY_BLOCK,
+            proposed_file=proposed,
+            justification="need a route",
+            current_file_hash=sha256_hex(proposed),
+            now=FIXED,
+        )
+        qdir = supervise.queue_dir_for_slug("dev")
+        qdir.mkdir(parents=True, exist_ok=True)
+        supervise.write_proposal(qdir, p)
+        return dashboard.QueuedProposal(proposal=p, queue_dir=qdir)
+
+    def test_cred_proxy_block_calls_apply_with_proposed_file(self):
+        calls = []
+        dashboard.apply_routes_change = lambda slug, content: (
+            calls.append((slug, content)) or ("before", content)
+        )
+        qp = self._enqueue_cred_proxy(proposed='{"routes": [{"path": "/new/"}]}\n')
+        dashboard.approve(qp)
+        self.assertEqual(1, len(calls))
+        slug, content = calls[0]
+        self.assertEqual("dev", slug)
+        self.assertEqual('{"routes": [{"path": "/new/"}]}\n', content)
+
+    def test_modify_passes_final_file_to_apply(self):
+        calls = []
+        dashboard.apply_routes_change = lambda slug, content: (
+            calls.append(content) or ("before", content)
+        )
+        qp = self._enqueue_cred_proxy()
+        dashboard.approve(qp, final_file='{"routes": [{"path": "/edited/"}]}\n', notes="tweaked")
+        self.assertEqual(['{"routes": [{"path": "/edited/"}]}\n'], calls)
+
+    def test_apply_failure_blocks_response_and_audit(self):
+        dashboard.apply_routes_change = lambda slug, content: (_ for _ in ()).throw(
+            CredProxyApplyError("docker exec failed")
+        )
+        qp = self._enqueue_cred_proxy()
+        with self.assertRaises(CredProxyApplyError):
+            dashboard.approve(qp)
+        # No response file (proposal stays pending).
+        self.assertEqual(
+            [qp.proposal.id],
+            [p.id for p in supervise.list_pending_proposals(qp.queue_dir)],
+        )
+        # No audit entry.
+        self.assertEqual([], read_audit_entries("cred-proxy", "dev"))
+
+    def test_real_diff_lands_in_audit(self):
+        dashboard.apply_routes_change = lambda slug, content: (
+            '{"routes": []}\n',  # before
+            '{"routes": [{"path": "/new/"}]}\n',  # after
+        )
+        qp = self._enqueue_cred_proxy(proposed='{"routes": [{"path": "/new/"}]}\n')
+        dashboard.approve(qp)
+        entries = read_audit_entries("cred-proxy", "dev")
+        self.assertEqual(1, len(entries))
+        self.assertIn('+{"routes": [{"path": "/new/"}]}', entries[0].diff)
+        self.assertIn('-{"routes": []}', entries[0].diff)
+
+    def test_reject_does_not_call_apply(self):
+        called = []
+        dashboard.apply_routes_change = lambda slug, content: (
+            called.append(True) or ("", content)
+        )
+        qp = self._enqueue_cred_proxy()
+        dashboard.reject(qp, reason="no thanks")
+        self.assertEqual([], called)
+        # Reject still writes a response + audit entry with empty diff.
+        resp = read_response(qp.queue_dir, qp.proposal.id)
+        self.assertEqual(STATUS_REJECTED, resp.status)
+        entries = read_audit_entries("cred-proxy", "dev")
+        self.assertEqual(1, len(entries))
+        self.assertEqual("", entries[0].diff)
+
+
+class TestOperatorEditRoutes(_FakeHomeMixin, unittest.TestCase):
+    """PRD 0014 Phase 4: operator-initiated routes edit (not gated
+    on a pending proposal)."""
+
+    def setUp(self):
+        self._setup_fake_home()
+        self._original_apply = dashboard.apply_routes_change
+
+    def tearDown(self):
+        dashboard.apply_routes_change = self._original_apply
+        self._teardown_fake_home()
+
+    def test_writes_audit_with_operator_edit_action(self):
+        dashboard.apply_routes_change = lambda slug, content: (
+            '{"routes": []}\n', content,
+        )
+        dashboard.operator_edit_routes("dev", '{"routes": [{"path": "/x/"}]}\n')
+        entries = read_audit_entries("cred-proxy", "dev")
+        self.assertEqual(1, len(entries))
+        self.assertEqual(supervise.ACTION_OPERATOR_EDIT, entries[0].operator_action)
+        self.assertEqual("", entries[0].justification)
+        self.assertIn("+", entries[0].diff)
+
+    def test_failure_does_not_write_audit(self):
+        dashboard.apply_routes_change = lambda slug, content: (_ for _ in ()).throw(
+            CredProxyApplyError("nope")
+        )
+        with self.assertRaises(CredProxyApplyError):
+            dashboard.operator_edit_routes("dev", '{"routes": []}\n')
+        self.assertEqual([], read_audit_entries("cred-proxy", "dev"))
+
+
+class TestDiscoverCredProxySlugs(unittest.TestCase):
+    """Slug-extraction parsing — exercises only the parsing path; the
+    docker ps invocation itself is environment-dependent (and tested
+    implicitly by the integration test)."""
+
+    def test_returns_empty_when_docker_unavailable(self):
+        # Force a failure by setting PATH to a dir with no docker
+        # binary. The discover helper swallows the non-zero rc.
+        import os
+        original = os.environ.get("PATH", "")
+        os.environ["PATH"] = "/nonexistent-no-docker-here"
+        try:
+            self.assertEqual([], dashboard.discover_cred_proxy_slugs())
+        finally:
+            os.environ["PATH"] = original
+
+
 class TestEditInEditor(unittest.TestCase):
     def test_runs_editor_returns_edited_content(self):
         # Fake "editor" is /bin/sh -c 'cat <<EOF > $1 ... EOF'