feat(dashboard): routes edit TUI verb for operator-initiated changes (PRD 0014)

Phase 4 of PRD 0014. Adds the proactive routes-edit path that
doesn't require a pending proposal:

- discover_cred_proxy_slugs() lists running cred-proxy sidecars by
  parsing docker ps output. Returns [] when docker is unreachable
  or not installed (no exception escapes).
- operator_edit_routes(slug, new_content) wraps apply_routes_change
  and writes an audit entry tagged ACTION_OPERATOR_EDIT (so a
  future reader can distinguish operator-initiated changes from
  agent-proposal approvals in the log).
- New 'e' keybinding in the main TUI: discover slugs, prompt if
  multiple (or use the only one directly), fetch current routes,
  open in $EDITOR, apply on save. CredProxyApplyError lands in the
  status line; the operator can retry.

Tests cover audit-entry shape, failure path, and docker-missing
recovery for slug discovery.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
2026-05-25 04:47:22 -04:00
parent f3a1b4d667
commit 81277e9d81
2 changed files with 136 additions and 1 deletions
+87 -1
View File
@@ -25,6 +25,7 @@ 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 (
@@ -57,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
@@ -132,6 +160,27 @@ def reject(qp: QueuedProposal, *, reason: str) -> 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(
qp: QueuedProposal,
*,
@@ -244,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]
@@ -313,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:
@@ -408,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)
+49
View File
@@ -268,6 +268,55 @@ class TestCredProxyApplyWiring(_FakeHomeMixin, unittest.TestCase):
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'