19 Commits

Author SHA1 Message Date
didericis-claude a6733921c7 docs(prd): add PRD 0049 — named/labelled agents
test / unit (pull_request) Successful in 42s
test / integration (pull_request) Successful in 1m7s
Draft PRD for prompting operators for a custom label and optional
ANSI color at agent launch time, storing both in metadata.json, and
surfacing the label (in color) in the dashboard's active-agents pane.

Closes #171
2026-06-03 13:49:49 -04:00
didericis-codex f12b0f754e docs(prd): reactivate PRD 0049 without tmux alert
test / unit (pull_request) Successful in 37s
test / integration (pull_request) Successful in 55s
test / unit (push) Successful in 51s
test / integration (push) Successful in 59s
2026-06-03 17:35:10 +00:00
didericis-codex a593b157d6 fix(cli): remove supervise tmux alert handling
test / unit (pull_request) Successful in 40s
test / integration (pull_request) Successful in 55s
2026-06-03 17:34:41 +00:00
didericis-codex 15b54cdff2 docs(prd): reactivate PRD 0049 without queue highlight
test / unit (pull_request) Successful in 44s
test / integration (pull_request) Successful in 54s
2026-06-03 17:31:49 +00:00
didericis-codex d3bc463295 fix(cli): remove supervise queue highlight
test / unit (pull_request) Successful in 43s
test / integration (pull_request) Successful in 52s
2026-06-03 17:31:19 +00:00
didericis-codex 50ec920243 docs(prd): activate PRD 0049 strip dashboard to supervise
test / unit (pull_request) Successful in 37s
test / integration (pull_request) Successful in 55s
2026-06-03 17:27:56 +00:00
didericis-codex 4372b8a6dd docs(cli): update supervise code references
test / unit (pull_request) Successful in 35s
test / integration (pull_request) Successful in 52s
2026-06-03 17:27:14 +00:00
didericis-codex 63a7e63ce9 test(cli): clean up supervise test naming
test / unit (pull_request) Successful in 40s
test / integration (pull_request) Successful in 48s
2026-06-03 17:26:15 +00:00
didericis-codex c0e1f5fd70 docs(prd): supersede dashboard agent PRDs
test / unit (pull_request) Successful in 36s
test / integration (pull_request) Successful in 54s
2026-06-03 17:25:32 +00:00
didericis-codex 41570e04c0 test(cli): update supervise triage coverage
test / unit (pull_request) Successful in 40s
test / integration (pull_request) Successful in 51s
2026-06-03 17:25:09 +00:00
didericis-codex 6f0a42159f refactor(cli): rename dashboard command to supervise
test / unit (pull_request) Failing after 38s
test / integration (pull_request) Successful in 54s
2026-06-03 17:23:40 +00:00
didericis-codex 5c17f0de95 docs(prd): rename strip dashboard PRD to 0049 2026-06-03 17:19:41 +00:00
didericis 8a09e32fcc docs(prd): add PRD 0050 -- strip dashboard to supervisor tui
test / unit (pull_request) Successful in 51s
test / integration (pull_request) Successful in 54s
2026-06-03 13:15:05 -04:00
didericis-claude 83463f1cc8 docs(prd): activate PRD 0048 — SSH deploy-key provisioning
test / unit (push) Successful in 40s
test / integration (push) Successful in 41s
2026-06-03 11:58:36 -04:00
didericis-claude 0b5d59cf9e feat(prd-0048): implement SSH deploy-key provisioning with contrib/gitea
- manifest_git.py: add ProvisionedKeyConfig dataclass; extend GitEntry
  with ProvisionedKey field (optional); make IdentityFile default to ""
  so provisioned_key entries can be constructed without a static path;
  add _parse_provisioned_key_config; update from_repos_entry to accept
  provisioned_key as an alternative to identity (mutually exclusive,
  parser rejects both-or-neither)

- deploy_key_provisioner.py (new): DeployKeyProvisioner ABC with create()
  and delete() abstract methods; get_provisioner() factory with lazy
  contrib import for gitea

- contrib/gitea/deploy_key_provisioner.py (new): GiteaDeployKeyProvisioner
  generating ed25519 keypairs via ssh-keygen and managing them through
  the Gitea deploy-key API (POST/DELETE); 404 on delete is success;
  all other errors raise RuntimeError

- git_gate.py: add _provision_dynamic_key() called in GitGate.prepare()
  for entries with ProvisionedKey — generates key, writes private key
  and key ID files to stage_dir, patches GitGateUpstream.identity_file;
  add revoke_git_gate_provisioned_keys() for teardown — raises on failure

- docker/launch.py: call revoke_git_gate_provisioned_keys() in teardown()
  after stack.close() so revocation runs after containers stop and
  failures propagate (not suppressed)

- smolmachines/launch.py: extract _teardown_smolmachines() helper that
  catches stack.close() errors (warn + re-raise) then calls revocation;
  same fatal-on-failure contract as docker backend

- test_manifest_git.py: 9 new cases for provisioned_key parsing
- test_deploy_key_provisioner.py (new): factory smoke tests
- test_contrib_gitea_deploy_key.py (new): create/delete/error/split tests

Closes #169
2026-06-03 11:58:36 -04:00
didericis-claude 464012d97c docs(prd): address review feedback on PRD 0048
- Rename deploy_key → provisioned_key throughout (manifest key,
  dataclass names, internal field names, test descriptions)
- Revocation failure at teardown now halts cleanup and propagates
  loudly; a stranded key is a security concern that must surface
2026-06-03 11:58:36 -04:00
didericis-claude b5f8a27c47 docs(prd): add SSH deploy-key provisioning plan (PRD 0048)
Introduces the design for short-lived deploy keys provisioned at spin-up
and revoked at teardown, plus the contrib package structure for
platform-specific provisioner implementations. First contrib provider
targets the Gitea deploy-key API.

Closes #169
2026-06-03 11:58:36 -04:00
didericis-claude f0ca4e3527 refactor: extract dashboard state/model layer into dashboard_model.py
test / unit (pull_request) Successful in 41s
test / integration (pull_request) Successful in 47s
test / unit (push) Successful in 35s
test / integration (push) Successful in 44s
Splits the 2103-line dashboard.py into two modules. Pure data
structures (QueuedProposal), discovery helpers (discover_pending,
discover_active_agents), derived-value helpers (_is_recent,
_approval_status, _format_agent_row, _detail_lines, etc.), and
argv-builder helpers (_build_split_pane_argv, _build_respawn_pane_argv,
_build_resume_argv_with_fallback, _agent_runtime_args) all move to
dashboard_model.py. The curses TUI, $EDITOR integration, tmux
subprocess flows, and action handlers (approve, reject,
operator_edit_routes, operator_edit_allowlist) remain in dashboard.py,
which re-imports everything from dashboard_model so existing callers and
tests are unaffected.

Adds tests/unit/test_dashboard_model.py covering _approval_status,
_proposed_payload_label, and _suffix_for_tool — three helpers that had
no prior coverage. All 894 unit tests pass.

Closes #158
2026-06-03 15:52:27 +00:00
didericis-claude ca6d257f30 test(git-gate): add shell-escaping regression tests (issue #159)
test / unit (pull_request) Successful in 36s
test / integration (pull_request) Successful in 44s
test / unit (push) Successful in 35s
test / integration (push) Successful in 42s
Cover all six pathological character classes (single-quote,
double-quote, space, semicolon, newline, backtick) in both
upstream URL and name positions.  Each case validates rendered
output via `sh -n` and asserts the original value is preserved
verbatim after shlex.quote encoding.  Also add `sh -n` smoke
tests for the static pre-receive and access-hook scripts.
2026-06-03 14:51:23 +00:00
29 changed files with 2416 additions and 2906 deletions
+8
View File
@@ -43,6 +43,7 @@ from pathlib import Path
from typing import Callable, Generator
from ...egress import egress_resolve_token_values
from ...git_gate import revoke_git_gate_provisioned_keys
from ...log import info, warn
from . import network as network_mod
from . import util as docker_mod
@@ -51,6 +52,7 @@ from .bottle_plan import DockerBottlePlan
from .bottle_state import (
bottle_state_dir,
egress_state_dir,
git_gate_state_dir,
pipelock_state_dir,
)
from .compose import (
@@ -84,6 +86,9 @@ def launch(
Teardown on exit."""
stack = ExitStack()
_bottle_for_revoke = plan.spec.manifest.bottle_for(plan.spec.agent_name)
_git_gate_dir_for_revoke = git_gate_state_dir(plan.slug)
def teardown() -> None:
try:
stack.close()
@@ -92,6 +97,9 @@ def launch(
f"teardown failed for container {plan.container_name}"
f" (compose-down): {exc!r}"
)
revoke_git_gate_provisioned_keys(
_bottle_for_revoke, _git_gate_dir_for_revoke
)
try:
# Step 1: agent image build. Sidecar images get built lazily by
+24
View File
@@ -53,6 +53,9 @@ from ..docker.pipelock import (
PIPELOCK_PORT as _PIPELOCK_PORT_STR,
pipelock_tls_init,
)
from ...git_gate import revoke_git_gate_provisioned_keys
from ...log import warn
from ..docker.bottle_state import git_gate_state_dir
from . import loopback_alias as _loopback
from . import sidecar_bundle as _bundle
from . import smolvm as _smolvm
@@ -120,7 +123,28 @@ def launch(
agent_prompt_mode=plan.agent_prompt_mode,
)
finally:
_teardown_smolmachines(stack, plan)
def _teardown_smolmachines(
stack: ExitStack,
plan: SmolmachinesBottlePlan,
) -> None:
"""Unwind the ExitStack, then revoke any provisioned deploy keys.
ExitStack errors are caught and logged (non-fatal) so that key
revocation always runs. Revocation errors propagate — a stranded
deploy key is a security concern the operator must address."""
teardown_exc: BaseException | None = None
try:
stack.close()
except BaseException as exc:
teardown_exc = exc
warn(f"smolmachines teardown failed: {exc!r}")
bottle = plan.spec.manifest.bottle_for(plan.spec.agent_name)
revoke_git_gate_provisioned_keys(bottle, git_gate_state_dir(plan.slug))
if teardown_exc is not None:
raise teardown_exc
def _allocate_resources(
+5 -5
View File
@@ -1,6 +1,6 @@
"""Main CLI dispatcher.
Commands: cleanup, dashboard, edit, info, init, list, resume, start
Commands: cleanup, edit, info, init, list, resume, start, supervise
"""
from __future__ import annotations
@@ -12,24 +12,24 @@ from ..manifest import ManifestError
from ._common import PROG
from . import list as _list_mod
from .cleanup import cmd_cleanup
from .dashboard import cmd_dashboard
from .edit import cmd_edit
from .info import cmd_info
from .init import cmd_init
from .resume import cmd_resume
from .start import cmd_start
from .supervise import cmd_supervise
cmd_list = _list_mod.cmd_list
COMMANDS = {
"cleanup": cmd_cleanup,
"dashboard": cmd_dashboard,
"edit": cmd_edit,
"info": cmd_info,
"init": cmd_init,
"list": cmd_list,
"resume": cmd_resume,
"start": cmd_start,
"supervise": cmd_supervise,
}
@@ -37,13 +37,13 @@ def usage() -> None:
sys.stderr.write(f"usage: {PROG} <command> [args...]\n\n")
sys.stderr.write("Commands:\n")
sys.stderr.write(" cleanup stop and remove all active bot-bottle containers\n")
sys.stderr.write(" dashboard view + approve/modify/reject pending supervise proposals (PRD 0013)\n")
sys.stderr.write(" edit open an agent in vim for editing\n")
sys.stderr.write(" info print env, skills, and prompt details for a named agent\n")
sys.stderr.write(" init interactively create a new agent and add it to bot-bottle.json\n")
sys.stderr.write(" list list available agents or active containers\n")
sys.stderr.write(" resume re-launch a bottle by its identity (continues state from PRD 0016)\n")
sys.stderr.write(" start boot a container for a named agent and attach an interactive session\n\n")
sys.stderr.write(" start boot a container for a named agent and attach an interactive session\n")
sys.stderr.write(" supervise view + approve/modify/reject pending supervise proposals (PRD 0013)\n\n")
sys.stderr.write(f"Run '{PROG} <command> --help' for command-specific usage.\n")
File diff suppressed because it is too large Load Diff
+11 -25
View File
@@ -2,10 +2,8 @@
interactive claude-code session. The container is torn down when the
session ends.
The launch core is shared with `cli.py resume <identity>` and (PRD
0020 chunk 1+) the dashboard's in-process start flow: see the
public helpers `prepare_with_preflight`, `attach_agent`, and the
private orchestrator `_launch_bottle`.
The launch core is shared with `cli.py resume <identity>` through
the private orchestrator `_launch_bottle`.
"""
from __future__ import annotations
@@ -71,7 +69,7 @@ def cmd_start(argv: list[str]) -> int:
)
# --- Public helpers shared with the dashboard (PRD 0020) -----------------
# --- Launch helpers ------------------------------------------------------
def prepare_with_preflight(
@@ -84,14 +82,11 @@ def prepare_with_preflight(
backend_name: str | None = None,
) -> tuple[DockerBottlePlan | None, str]:
"""Run `backend.prepare`, render the preflight summary via the
injected callable, prompt y/N via the injected callable. The CLI
binds these to stderr/stdin; the dashboard binds them to a
curses modal.
injected callable, prompt y/N via the injected callable.
`backend_name` selects which backend prepares the plan
(`None` → `$BOT_BOTTLE_BACKEND` → `docker`). Dashboard
passes the value from its new-agent backend-picker modal; the
CLI passes whatever `--backend` resolved to.
(`None` → `$BOT_BOTTLE_BACKEND` → `docker`). The CLI passes
whatever `--backend` resolved to.
Returns `(plan, identity)`. `plan` is None on dry-run or
operator-N, but `identity` is set as soon as `backend.prepare`
@@ -122,16 +117,10 @@ def attach_agent(
agent process's exit code.
`resume=True` adds `--continue` so claude picks up its most
recent session non-interactively (no session-picker prompt)
the right shape for the dashboard's Enter re-attach (PRD 0020
chunk 3), where a bottle typically has exactly one session.
First-attach paths (`./cli.py start`, the dashboard's new-agent
flow) leave it False.
recent session non-interactively (no session-picker prompt).
First-attach paths (`./cli.py start`) leave it False.
Used as the inner step of `./cli.py start` (one-shot) and by the
dashboard, which calls it from inside a `curses.endwin → … →
stdscr.refresh()` handoff so the curses surface gets out of the
terminal's way while the agent has it."""
Used as the inner step of `./cli.py start`."""
runtime = runtime_for(agent_provider_template)
info(
f"attaching interactive {agent_provider_template} session "
@@ -148,8 +137,7 @@ def attach_agent(
def capture_claude_session_state(identity: str, exit_code: int) -> None:
"""Inside the launch context, while the container is still
alive: snapshot the transcript and mark for preservation if
claude crashed. Public for the dashboard's death-handling path
(PRD 0020 open question 3)."""
claude crashed."""
# FIXME: this captures Claude-specific session state. A follow-up
# spike should explore freezing provider-neutral container state
# instead of relying on each agent's transcript layout.
@@ -162,9 +150,7 @@ def capture_claude_session_state(identity: str, exit_code: int) -> None:
def settle_state(identity: str) -> None:
"""Post-teardown housekeeping: print the resume hint if the
state was preserved, otherwise reap the per-bottle state dir.
Public so the dashboard's explicit-stop path calls the same
settlement the CLI uses on context exit."""
state was preserved, otherwise reap the per-bottle state dir."""
if not identity:
return
if is_preserved(identity):
+577
View File
@@ -0,0 +1,577 @@
"""supervise: list pending supervise proposals across all bottles and
act on them (approve / modify / reject).
Curses-based TUI; modify-then-approve shells out to $EDITOR. The
approval handlers wire to the per-tool remediation engines:
PRD 0014 (egress, retargeted from cred-proxy in PRD 0017
chunk 3) writes routes.yaml + SIGHUPs egress; PRD 0015
(pipelock) writes the allowlist + restarts pipelock; PRD 0016
(capability) rebuilds the bottle Dockerfile.
"""
from __future__ import annotations
import argparse
import curses
import os
import subprocess
import sys
import tempfile
import traceback
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from .. import supervise as _supervise
from ..backend.docker.bottle_state import read_metadata
from ..backend.docker.capability_apply import (
CapabilityApplyError,
apply_capability_change,
)
from ..backend.docker.egress_apply import EgressApplyError, add_route
from ..backend.docker.pipelock_apply import (
PipelockApplyError,
apply_allowlist_change,
fetch_current_allowlist,
parse_allowlist_content,
render_allowlist_content,
)
from ..log import Die, error, info
from ..supervise import (
COMPONENT_FOR_TOOL,
AuditEntry,
Proposal,
Response,
STATUS_APPROVED,
STATUS_MODIFIED,
STATUS_REJECTED,
TOOL_CAPABILITY_BLOCK,
TOOL_EGRESS_BLOCK,
TOOL_PIPELOCK_BLOCK,
archive_proposal,
list_pending_proposals,
render_diff,
write_audit_entry,
write_response,
)
from ._common import PROG
_REFRESH_INTERVAL_MS = 1000
@dataclass(frozen=True)
class QueuedProposal:
"""A pending proposal plus the queue dir it was found in."""
proposal: Proposal
queue_dir: Path
# 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, PipelockApplyError, CapabilityApplyError)
def discover_pending() -> list[QueuedProposal]:
"""Walk ~/.bot-bottle/queue/* and collect pending proposals."""
queue_root = _supervise.bot_bottle_root() / "queue"
if not queue_root.is_dir():
return []
out: list[QueuedProposal] = []
for slug_dir in sorted(queue_root.iterdir()):
if not slug_dir.is_dir():
continue
for proposal in list_pending_proposals(slug_dir):
out.append(QueuedProposal(proposal=proposal, queue_dir=slug_dir))
out.sort(key=lambda q: q.proposal.arrival_timestamp)
return out
def _approval_status(qp: QueuedProposal, verb: str) -> str:
"""Status-line text after a successful approval."""
base = f"{verb} {qp.proposal.tool} for [{qp.proposal.bottle_slug}]"
if qp.proposal.tool == TOOL_CAPABILITY_BLOCK:
return f"{base}; resume: ./cli.py resume {qp.proposal.bottle_slug}"
return base
def _detail_lines(
qp: QueuedProposal,
*,
green_attr: int = 0,
) -> list[tuple[str, int]]:
"""Return the detail-view body as (text, curses-attr) tuples."""
p = qp.proposal
out: list[tuple[str, int]] = [
(f"bottle: {p.bottle_slug}", 0),
(f"tool: {p.tool}", 0),
(f"id: {p.id}", 0),
(f"arrived: {p.arrival_timestamp}", 0),
(f"queue: {qp.queue_dir}", 0),
("", 0),
("justification:", 0),
]
out.extend((" " + line, 0) for line in p.justification.splitlines() or [""])
out.extend([
("", 0),
(_proposed_payload_label(p.tool) + ":", 0),
])
out.extend((line, 0) for line in p.proposed_file.splitlines() or [""])
if p.tool == TOOL_PIPELOCK_BLOCK:
host = _failed_url_host(p.proposed_file)
if host:
out.append(("", 0))
out.append((host, green_attr))
return out
def _failed_url_host(url: str) -> str:
"""Best-effort hostname extraction from a pipelock-block proposal."""
import urllib.parse
try:
return urllib.parse.urlsplit(url.strip()).hostname or ""
except ValueError:
return ""
def _proposed_payload_label(tool: str) -> str:
if tool == TOOL_PIPELOCK_BLOCK:
return "failed URL"
return "proposed file"
def _suffix_for_tool(tool: str) -> str:
if tool == TOOL_CAPABILITY_BLOCK:
return ".dockerfile"
return ".txt"
# --- Operator actions ------------------------------------------------------
def approve(
qp: QueuedProposal,
*,
notes: str = "",
final_file: str | None = None,
) -> None:
"""Apply the proposal, write the waiting response, and audit it."""
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_EGRESS_BLOCK:
diff_before, diff_after = add_route(
qp.proposal.bottle_slug, file_to_apply,
)
elif qp.proposal.tool == TOOL_PIPELOCK_BLOCK:
diff_before, diff_after = _apply_pipelock_url(
qp.proposal.bottle_slug, file_to_apply,
)
elif 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,
)
response = Response(
proposal_id=qp.proposal.id,
status=status,
notes=notes,
final_file=final_file,
)
write_response(qp.queue_dir, response)
_write_audit(
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."""
response = Response(
proposal_id=qp.proposal.id,
status=STATUS_REJECTED,
notes=reason,
final_file=None,
)
write_response(qp.queue_dir, response)
_write_audit(qp, action=STATUS_REJECTED, notes=reason, diff_before="", diff_after="")
def _apply_pipelock_url(slug: str, failed_url: str) -> tuple[str, str]:
"""Merge a pipelock-block failed URL's host into the allowlist."""
import urllib.parse
parsed = urllib.parse.urlsplit(failed_url.strip())
host = parsed.hostname or ""
if not host:
raise PipelockApplyError(
f"proposed failed_url has no extractable host: {failed_url!r}"
)
current = fetch_current_allowlist(slug)
hosts = parse_allowlist_content(current)
if host not in hosts:
hosts.append(host)
return apply_allowlist_change(slug, render_allowlist_content(hosts))
def _write_audit(
qp: QueuedProposal,
*,
action: str,
notes: str,
diff_before: str,
diff_after: str,
) -> None:
"""Audit log for egress / pipelock tools."""
component = COMPONENT_FOR_TOOL.get(qp.proposal.tool)
if component is None:
return
write_audit_entry(AuditEntry(
timestamp=datetime.now(timezone.utc).isoformat(),
bottle_slug=qp.proposal.bottle_slug,
component=component,
operator_action=action,
operator_notes=notes,
justification=qp.proposal.justification,
diff=render_diff(diff_before, diff_after, label=component),
))
# --- $EDITOR integration --------------------------------------------------
def edit_in_editor(content: str, *, suffix: str = ".tmp") -> str | None:
"""Open `content` in $EDITOR and return edited content, if changed."""
editor = os.environ.get("EDITOR", "vim")
with tempfile.NamedTemporaryFile(
mode="w", suffix=suffix, delete=False, prefix="supervise-modify.",
) as f:
f.write(content)
path = f.name
try:
subprocess.run([editor, path], check=False)
with open(path) as f:
edited = f.read()
return edited if edited != content else None
finally:
try:
os.unlink(path)
except OSError:
pass
# --- TUI -------------------------------------------------------------------
def cmd_supervise(argv: list[str]) -> int:
parser = argparse.ArgumentParser(prog=f"{PROG} supervise", add_help=True)
parser.add_argument(
"--once", action="store_true",
help="list pending proposals once and exit (no TUI)",
)
args = parser.parse_args(argv)
if args.once:
return _list_once()
try:
curses.wrapper(_main_loop)
except KeyboardInterrupt:
return 130
except Die as e:
if e.message:
error(e.message)
else:
error("supervise exited on a fatal error (no detail captured).")
return e.code if isinstance(e.code, int) else 1
except Exception as e:
log_path = _write_crash_log(e)
error(f"supervise crashed: {type(e).__name__}: {e}")
error(f"full traceback written to {log_path}")
return 1
return 0
def _write_crash_log(exc: BaseException) -> Path:
"""Persist `exc`'s traceback to a stable file under ~/.bot-bottle/."""
stamp = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
body = "".join(
traceback.format_exception(type(exc), exc, exc.__traceback__)
)
entry = f"=== supervise crash {stamp} ===\n{body}\n"
try:
log_dir = _supervise.bot_bottle_root() / "logs"
log_dir.mkdir(parents=True, exist_ok=True)
path = log_dir / "supervise-crash.log"
with path.open("a", encoding="utf-8") as fh:
fh.write(entry)
return path
except OSError:
fd, tmp = tempfile.mkstemp(
prefix="bot-bottle-supervise-crash-", suffix=".log",
)
with os.fdopen(fd, "w", encoding="utf-8") as fh:
fh.write(entry)
return Path(tmp)
def _list_once() -> int:
pending = discover_pending()
if not pending:
info("no pending proposals")
return 0
for qp in pending:
sys.stdout.write(
f"{qp.proposal.arrival_timestamp} "
f"[{qp.proposal.bottle_slug}] "
f"{qp.proposal.tool} "
f"{qp.proposal.id}\n"
)
sys.stdout.write(f" {qp.proposal.justification}\n")
return 0
def _try_init_green() -> int:
"""Initialise a green color pair and return its attr, or 0."""
try:
curses.start_color()
curses.use_default_colors()
curses.init_pair(1, curses.COLOR_GREEN, -1)
return curses.color_pair(1)
except curses.error:
return 0
def _main_loop(stdscr: "curses._CursesWindow") -> None:
curses.curs_set(0)
stdscr.timeout(_REFRESH_INTERVAL_MS)
green_attr = _try_init_green()
selected = 0
status_line = ""
seen_ids: set[str] = set()
while True:
pending = discover_pending()
if selected >= len(pending):
selected = max(0, len(pending) - 1)
live_ids = {qp.proposal.id for qp in pending}
newly_arrived = live_ids - seen_ids
if seen_ids and newly_arrived:
try:
curses.beep()
except curses.error:
pass
for i, qp in enumerate(pending):
if qp.proposal.id in newly_arrived:
selected = i
break
seen_ids = live_ids
_render(
stdscr, pending, selected, status_line,
green_attr=green_attr,
)
try:
key = stdscr.getch()
except KeyboardInterrupt:
return
if key == -1:
continue
status_line = ""
if key in (ord("q"), 27):
return
if not pending:
continue
qp = pending[selected]
if key in (curses.KEY_DOWN, ord("j")):
selected = min(selected + 1, len(pending) - 1)
elif key in (curses.KEY_UP, ord("k")):
selected = max(selected - 1, 0)
elif key in (curses.KEY_ENTER, 10, 13):
_detail_view(stdscr, qp, green_attr=green_attr)
elif key == ord("a"):
try:
approve(qp)
status_line = _approval_status(qp, "approved")
except ApplyError 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:
try:
approve(qp, final_file=edited, notes="operator modified before approving")
status_line = _approval_status(qp, "modified+approved")
except ApplyError as e:
status_line = f"apply failed: {e}"
elif key == ord("r"):
reason = _prompt(stdscr, "reject reason: ")
if reason:
reject(qp, reason=reason)
status_line = f"rejected {qp.proposal.tool} for [{qp.proposal.bottle_slug}]"
else:
status_line = "reject aborted (empty reason)"
def _render(
stdscr: "curses._CursesWindow",
pending: list[QueuedProposal],
selected: int,
status_line: str,
*,
green_attr: int = 0,
) -> None:
stdscr.erase()
h, w = stdscr.getmaxyx()
header = f"bot-bottle supervise ({len(pending)} pending)"
stdscr.addnstr(0, 0, header, w - 1, curses.A_BOLD)
stdscr.hline(1, 0, curses.ACS_HLINE, w)
row = 2
if not pending:
stdscr.addnstr(
row, 2,
"no pending proposals; agents will queue here when they call a "
"supervise tool",
w - 4,
)
else:
for i, qp in enumerate(pending):
if row >= h - 3:
break
p = qp.proposal
ts_short = (
p.arrival_timestamp.split("T", 1)[1][:8]
if "T" in p.arrival_timestamp else p.arrival_timestamp
)
cursor = "> " if i == selected else " "
line = (
f"{cursor}{ts_short} "
f"[{p.bottle_slug}] {p.tool:<18} {p.id[:8]} "
f"{_proposed_payload_label(p.tool)}"
)
attr = curses.A_REVERSE if i == selected else curses.A_NORMAL
stdscr.addnstr(row, 0, line, w - 1, attr)
row += 1
if row >= h - 3:
break
if p.justification:
stdscr.addnstr(row, 4, p.justification[: max(0, w - 5)], w - 5)
row += 1
footer = "[j/k] move [Enter] view [a] approve [m] modify [r] reject [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:
stdscr.addnstr(h - 3, 0, status_line, w - 1, curses.A_BOLD)
stdscr.refresh()
def _detail_view(
stdscr: "curses._CursesWindow",
qp: QueuedProposal,
*,
green_attr: int = 0,
) -> None:
"""Render the full proposal. Scrollable. Press q to return."""
lines = _detail_lines(qp, green_attr=green_attr)
offset = 0
while True:
stdscr.erase()
h, w = stdscr.getmaxyx()
for i, (text, attr) in enumerate(lines[offset:offset + h - 1]):
stdscr.addnstr(i, 0, text, w - 1, attr)
stdscr.addnstr(
h - 1, 0,
"[j/k] scroll [g/G] top/bottom [a] approve [m] modify [r] reject [q] back",
w - 1, curses.A_DIM,
)
stdscr.refresh()
key = stdscr.getch()
if key in (ord("q"), 27):
return
if key in (curses.KEY_DOWN, ord("j")):
offset = min(offset + 1, max(0, len(lines) - 1))
elif key in (curses.KEY_UP, ord("k")):
offset = max(offset - 1, 0)
elif key == ord("g"):
offset = 0
elif key == ord("G"):
offset = max(0, len(lines) - 1)
elif key == ord("a"):
try:
approve(qp)
except ApplyError:
pass
return
elif key == ord("m"):
edited = _modify(stdscr, qp)
if edited is not None:
try:
approve(qp, final_file=edited, notes="operator modified before approving")
except ApplyError:
pass
return
elif key == ord("r"):
reason = _prompt(stdscr, "reject reason: ")
if reason:
reject(qp, reason=reason)
return
def _modify(stdscr: "curses._CursesWindow", qp: QueuedProposal) -> str | None:
"""Suspend curses, open $EDITOR on the proposed file, return edited content."""
suffix = _suffix_for_tool(qp.proposal.tool)
curses.endwin()
try:
edited = edit_in_editor(qp.proposal.proposed_file, suffix=suffix)
finally:
stdscr.refresh()
return edited
def _prompt(stdscr: "curses._CursesWindow", label: str) -> str:
"""One-line input at the bottom of the screen."""
curses.curs_set(1)
h, _ = stdscr.getmaxyx()
stdscr.move(h - 2, 0)
stdscr.clrtoeol()
stdscr.addstr(h - 2, 0, label)
stdscr.refresh()
curses.echo()
try:
raw = stdscr.getstr(h - 2, len(label), 200)
finally:
curses.noecho()
curses.curs_set(0)
return raw.decode("utf-8", errors="replace").strip()
__all__ = [
"QueuedProposal",
"approve",
"cmd_supervise",
"discover_pending",
"edit_in_editor",
"reject",
]
View File
@@ -0,0 +1,121 @@
"""Gitea deploy-key provisioner (PRD 0048, contrib).
Generates ed25519 keypairs via `ssh-keygen` and registers / deletes
them using the Gitea deploy-key HTTP API. No new Python dependencies —
only stdlib `urllib.request` and `subprocess`."""
from __future__ import annotations
import json
import subprocess
import tempfile
import urllib.error
import urllib.request
from pathlib import Path
from ...deploy_key_provisioner import DeployKeyProvisioner
class GiteaDeployKeyProvisioner(DeployKeyProvisioner):
"""Manages deploy keys on a Gitea instance."""
def __init__(self, *, token: str, api_url: str) -> None:
self._token = token
self._api_url = api_url.rstrip("/")
def create(self, owner_repo: str, title: str) -> tuple[str, bytes]:
"""Generate an ed25519 keypair, register the public half as a
repo deploy key, and return `(key_id, private_key_bytes)`.
The key is registered with `read_only=False` because git-gate
needs push access to forward gitleaks-scanned refs upstream."""
with tempfile.TemporaryDirectory() as tmpdir:
key_path = Path(tmpdir) / "key"
subprocess.run(
[
"ssh-keygen", "-t", "ed25519",
"-f", str(key_path),
"-N", "",
],
check=True,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
private_key = key_path.read_bytes()
public_key = key_path.with_suffix(".pub").read_text().strip()
owner, repo = _split_owner_repo(owner_repo)
url = f"{self._api_url}/api/v1/repos/{owner}/{repo}/keys"
payload = json.dumps({
"key": public_key,
"read_only": False,
"title": title,
}).encode()
req = urllib.request.Request(
url,
data=payload,
headers={
"Authorization": f"token {self._token}",
"Content-Type": "application/json",
},
method="POST",
)
try:
with urllib.request.urlopen(req) as resp:
body = json.loads(resp.read())
except urllib.error.HTTPError as exc:
_body = _read_error_body(exc)
raise RuntimeError(
f"failed to create deploy key for {owner_repo}: "
f"HTTP {exc.code}{_body}"
) from exc
except urllib.error.URLError as exc:
raise RuntimeError(
f"failed to create deploy key for {owner_repo}: {exc.reason}"
) from exc
return str(body["id"]), private_key
def delete(self, owner_repo: str, key_id: str) -> None:
"""Delete the deploy key. HTTP 404 (already gone) is success.
All other errors raise RuntimeError so teardown halts."""
owner, repo = _split_owner_repo(owner_repo)
url = f"{self._api_url}/api/v1/repos/{owner}/{repo}/keys/{key_id}"
req = urllib.request.Request(
url,
headers={"Authorization": f"token {self._token}"},
method="DELETE",
)
try:
with urllib.request.urlopen(req):
pass
except urllib.error.HTTPError as exc:
if exc.code == 404:
return
_body = _read_error_body(exc)
raise RuntimeError(
f"failed to delete deploy key {key_id} for {owner_repo}: "
f"HTTP {exc.code}{_body}"
) from exc
except urllib.error.URLError as exc:
raise RuntimeError(
f"failed to delete deploy key {key_id} for {owner_repo}: "
f"{exc.reason}"
) from exc
def _split_owner_repo(owner_repo: str) -> tuple[str, str]:
"""Split `'owner/repo'` into `('owner', 'repo')`."""
parts = owner_repo.split("/", 1)
if len(parts) != 2 or not all(parts):
raise ValueError(
f"expected 'owner/repo' format, got {owner_repo!r}"
)
return parts[0], parts[1]
def _read_error_body(exc: urllib.error.HTTPError) -> str:
try:
return exc.read().decode("utf-8", errors="replace")
except Exception:
return ""
+52
View File
@@ -0,0 +1,52 @@
"""Deploy-key provisioner interface and factory (PRD 0048).
The core defines the abstract contract; concrete implementations live
in `bot_bottle/contrib/<provider>/deploy_key_provisioner.py`. The
factory `get_provisioner` imports contrib modules lazily so that a
missing optional dependency in one provider doesn't break unrelated
features."""
from __future__ import annotations
from abc import ABC, abstractmethod
class DeployKeyProvisioner(ABC):
"""Manages a single deploy-key lifecycle on a remote forge."""
@abstractmethod
def create(self, owner_repo: str, title: str) -> tuple[str, bytes]:
"""Generate a keypair and register the public half as a
deploy key on the forge.
`owner_repo` is the `<owner>/<repo>` path (no `.git` suffix).
`title` is the human-readable label shown in the forge UI.
Returns `(key_id, private_key_bytes)` where `key_id` is opaque
to the caller and is only ever passed back to `delete`."""
@abstractmethod
def delete(self, owner_repo: str, key_id: str) -> None:
"""Delete the registered deploy key.
Must not raise if the key is already absent (HTTP 404 is
success). Must raise for all other failures so teardown halts."""
def get_provisioner(
provider: str, token: str, api_url: str
) -> DeployKeyProvisioner:
"""Instantiate the contrib provisioner for `provider`.
Raises `ManifestError` for unknown providers so the error surfaces
at parse time rather than at runtime."""
if provider == "gitea":
from bot_bottle.contrib.gitea.deploy_key_provisioner import (
GiteaDeployKeyProvisioner,
)
return GiteaDeployKeyProvisioner(token=token, api_url=api_url)
from .manifest_util import ManifestError
raise ManifestError(
f"unknown provisioned_key provider: {provider!r}; "
f"available: gitea"
)
+89 -1
View File
@@ -29,11 +29,14 @@ backend-specific and lives on concrete subclasses (see
from __future__ import annotations
import dataclasses
import os
import shlex
from abc import ABC, abstractmethod
from dataclasses import dataclass
from pathlib import Path
from .log import info
from .manifest import Bottle, GitEntry
@@ -357,6 +360,80 @@ exit 0
"""
def _provision_dynamic_key(
entry: GitEntry,
slug: str,
stage_dir: Path,
) -> str:
"""Generate a fresh ed25519 keypair, register the public half with
the forge, and persist the private key + key ID under `stage_dir`.
Returns the host-side path to the private key file so the caller
can inject it into the GitGateUpstream as `identity_file`."""
from .deploy_key_provisioner import get_provisioner
pk = entry.ProvisionedKey
assert pk is not None
token = os.environ.get(pk.token_env)
if token is None:
raise RuntimeError(
f"git-gate.repos[{entry.Name!r}] provisioned_key.token_env"
f" = {pk.token_env!r}: env var is not set"
)
api_url = pk.api_url or f"https://{entry.UpstreamHost}"
provisioner = get_provisioner(pk.provider, token, api_url)
owner_repo = entry.UpstreamPath
if owner_repo.endswith(".git"):
owner_repo = owner_repo[:-4]
title = f"bot-bottle:{slug}:{entry.Name}"
info(f"provisioning deploy key for git-gate.repos[{entry.Name!r}]")
key_id, private_key_bytes = provisioner.create(owner_repo, title)
key_file = stage_dir / f"{entry.Name}-key"
key_file.write_bytes(private_key_bytes)
key_file.chmod(0o600)
id_file = stage_dir / f"{entry.Name}-deploy-key-id"
id_file.write_text(key_id)
id_file.chmod(0o600)
info(f"provisioned deploy key {key_id} for git-gate.repos[{entry.Name!r}]")
return str(key_file)
def revoke_git_gate_provisioned_keys(bottle: Bottle, stage_dir: Path) -> None:
"""Revoke all deploy keys provisioned for `bottle` during prepare.
Called at teardown after containers stop. Raises if any revocation
fails a stranded key is a security concern that the operator must
address manually."""
from .deploy_key_provisioner import get_provisioner
for entry in bottle.git:
if entry.ProvisionedKey is None:
continue
pk = entry.ProvisionedKey
id_file = stage_dir / f"{entry.Name}-deploy-key-id"
if not id_file.exists():
continue
key_id = id_file.read_text().strip()
token = os.environ.get(pk.token_env)
if token is None:
raise RuntimeError(
f"git-gate.repos[{entry.Name!r}] provisioned_key.token_env"
f" = {pk.token_env!r}: env var is not set;"
f" cannot revoke deploy key {key_id}"
)
api_url = pk.api_url or f"https://{entry.UpstreamHost}"
provisioner = get_provisioner(pk.provider, token, api_url)
owner_repo = entry.UpstreamPath
if owner_repo.endswith(".git"):
owner_repo = owner_repo[:-4]
info(f"revoking deploy key {key_id} for git-gate.repos[{entry.Name!r}]")
provisioner.delete(owner_repo, key_id)
info(f"revoked deploy key {key_id} for git-gate.repos[{entry.Name!r}]")
class GitGate(ABC):
"""The per-agent git-gate. Encapsulates the host-side prepare
(upstream lift + entrypoint/hook render); the sidecar's
@@ -368,10 +445,21 @@ class GitGate(ABC):
entrypoint, pre-receive hook, and access-hook scripts (mode
600) under `stage_dir`. Pure host-side, no docker subprocess.
For `provisioned_key` entries, also generates and registers
a fresh deploy key via the forge API and writes the private key
+ key ID to `stage_dir`.
Returned plan is incomplete: the launch step must fill
`internal_network` / `egress_network` via `dataclasses.replace`
before passing the plan to `.start`."""
upstreams = git_gate_upstreams_for_bottle(bottle)
upstreams_list = list(git_gate_upstreams_for_bottle(bottle))
for i, entry in enumerate(bottle.git):
if entry.ProvisionedKey is not None:
key_file = _provision_dynamic_key(entry, slug, stage_dir)
upstreams_list[i] = dataclasses.replace(
upstreams_list[i], identity_file=key_file
)
upstreams = tuple(upstreams_list)
entrypoint = stage_dir / "git_gate_entrypoint.sh"
entrypoint.write_text(git_gate_render_entrypoint(upstreams))
entrypoint.chmod(0o600)
+90 -11
View File
@@ -4,6 +4,7 @@ from __future__ import annotations
import re
from dataclasses import dataclass
from typing import Optional
from .manifest_util import ManifestError, as_json_object
@@ -61,6 +62,24 @@ def validate_unique_git_names(bottle_name: str, git: tuple[GitEntry, ...]) -> No
seen[g.Name] = None
@dataclass(frozen=True)
class ProvisionedKeyConfig:
"""Configuration for automatic deploy-key lifecycle management
(PRD 0048). Used when a git-gate.repos entry opts out of a
static identity file and instead wants a fresh SSH keypair
generated at spin-up and revoked at teardown.
`provider` names the contrib sub-package to load (e.g. `gitea`).
`token_env` is the name of a host-side env var carrying the API
token; the value is read at provision time, never stored on the
plan. `api_url` is the forge's HTTP API root; if empty, it is
derived from the upstream URL's host at provision time."""
provider: str
token_env: str
api_url: str = ""
@dataclass(frozen=True)
class GitEntry:
"""One upstream the per-agent git-gate (PRD 0008) is allowed to
@@ -74,14 +93,15 @@ class GitEntry:
stashed in the `Upstream*` fields so the git-gate render step
doesn't have to re-parse.
Manifest source: `git-gate.repos.<Name>` (PRD 0047). The YAML keys
are `url`, `identity`, and `host_key`; the internal field names are
stable across that rename."""
Manifest source: `git-gate.repos.<Name>` (PRD 0047/0048). Exactly
one of `identity` (static key path) or `provisioned_key` (automatic
lifecycle) must be present. The internal field names are stable."""
Name: str
Upstream: str
IdentityFile: str
IdentityFile: str = ""
KnownHostKey: str = ""
ProvisionedKey: Optional[ProvisionedKeyConfig] = None
RemoteKey: str = ""
UpstreamUser: str = ""
UpstreamHost: str = ""
@@ -94,8 +114,9 @@ class GitEntry:
) -> "GitEntry":
"""Parse one entry from `git-gate.repos.<repo_name>`.
YAML keys: `url` (required), `identity` (required),
`host_key` (optional). The repo_name becomes `Name`."""
YAML keys: `url` (required), exactly one of `identity` or
`provisioned_key` (required), `host_key` (optional).
The repo_name becomes `Name`."""
if not repo_name:
raise ManifestError(
f"bottle '{bottle_name}' git-gate.repos has an empty key"
@@ -108,21 +129,44 @@ class GitEntry:
label = f"git-gate.repos[{repo_name!r}]"
d = as_json_object(raw, f"bottle '{bottle_name}' {label}")
for k in d:
if k not in {"url", "identity", "host_key"}:
if k not in {"url", "identity", "provisioned_key", "host_key"}:
raise ManifestError(
f"bottle '{bottle_name}' {label} has unknown key {k!r}; "
f"allowed: url, identity, host_key"
f"allowed: url, identity, provisioned_key, host_key"
)
upstream = d.get("url")
if not isinstance(upstream, str) or not upstream:
raise ManifestError(
f"bottle '{bottle_name}' {label} missing required string field 'url'"
)
ident = d.get("identity")
if not isinstance(ident, str) or not ident:
has_identity = "identity" in d
has_provisioned = "provisioned_key" in d
if has_identity and has_provisioned:
raise ManifestError(
f"bottle '{bottle_name}' {label} missing required string field 'identity'"
f"bottle '{bottle_name}' {label} must set exactly one of "
f"'identity' or 'provisioned_key'; got both."
)
if not has_identity and not has_provisioned:
raise ManifestError(
f"bottle '{bottle_name}' {label} must set exactly one of "
f"'identity' or 'provisioned_key'; got neither."
)
ident = ""
provisioned_key: Optional[ProvisionedKeyConfig] = None
if has_identity:
raw_ident = d.get("identity")
if not isinstance(raw_ident, str) or not raw_ident:
raise ManifestError(
f"bottle '{bottle_name}' {label} 'identity' must be a non-empty string"
)
ident = raw_ident
else:
provisioned_key = _parse_provisioned_key_config(
bottle_name, label, d["provisioned_key"]
)
khk = _opt_str(
d.get("host_key"),
f"bottle '{bottle_name}' {label} host_key",
@@ -135,6 +179,7 @@ class GitEntry:
Upstream=upstream,
IdentityFile=ident,
KnownHostKey=khk,
ProvisionedKey=provisioned_key,
RemoteKey=host,
UpstreamUser=user,
UpstreamHost=host,
@@ -143,6 +188,40 @@ class GitEntry:
)
def _parse_provisioned_key_config(
bottle_name: str, label: str, raw: object
) -> ProvisionedKeyConfig:
d = as_json_object(raw, f"bottle '{bottle_name}' {label}.provisioned_key")
for k in d:
if k not in {"provider", "token_env", "api_url"}:
raise ManifestError(
f"bottle '{bottle_name}' {label}.provisioned_key has unknown key {k!r}; "
f"allowed: provider, token_env, api_url"
)
provider = d.get("provider")
if not isinstance(provider, str) or not provider:
raise ManifestError(
f"bottle '{bottle_name}' {label}.provisioned_key missing required "
f"string field 'provider'"
)
token_env = d.get("token_env")
if not isinstance(token_env, str) or not token_env:
raise ManifestError(
f"bottle '{bottle_name}' {label}.provisioned_key missing required "
f"string field 'token_env'"
)
api_url_raw = d.get("api_url", "")
if not isinstance(api_url_raw, str):
raise ManifestError(
f"bottle '{bottle_name}' {label}.provisioned_key 'api_url' must be a string"
)
return ProvisionedKeyConfig(
provider=provider,
token_env=token_env,
api_url=api_url_raw,
)
@dataclass(frozen=True)
class GitUser:
"""Per-bottle `git config --global user.name` / `user.email`
+2 -2
View File
@@ -12,8 +12,8 @@ agent calls when it hits a stuck-recovery category:
Each tool call: the agent passes the full proposed file plus a
justification text. The sidecar validates the proposal syntactically,
writes it to the host's per-bottle queue dir, and holds the tool-call
connection open. The operator's TUI dashboard
(bot_bottle.cli.dashboard) sees the proposal, accepts
connection open. The operator's supervise TUI
(bot_bottle.cli.supervise) sees the proposal, accepts
approve / modify / reject, and writes a response file alongside the
proposal. The sidecar sees the response and returns `{status, notes}`
to the agent.
+1 -1
View File
@@ -1,6 +1,6 @@
# PRD 0019: Active agents in the dashboard, agent-scoped edit verbs
- **Status:** Active
- **Status:** Superseded by [PRD 0049](0049-strip-dashboard-to-supervisor-tui.md)
- **Author:** didericis
- **Created:** 2026-05-26
@@ -1,6 +1,6 @@
# PRD 0020: Start and attach to agents from inside the dashboard
- **Status:** Active
- **Status:** Superseded by [PRD 0049](0049-strip-dashboard-to-supervisor-tui.md)
- **Author:** didericis
- **Created:** 2026-05-26
+1 -1
View File
@@ -1,6 +1,6 @@
# PRD 0021: Dashboard as left tmux pane, selected agent as right pane
- **Status:** Active
- **Status:** Superseded by [PRD 0049](0049-strip-dashboard-to-supervisor-tui.md)
- **Author:** didericis
- **Created:** 2026-05-26
@@ -0,0 +1,296 @@
# PRD 0048: SSH Deploy-Key Provisioning
- **Status:** Active
- **Author:** didericis-claude
- **Created:** 2026-06-03
- **Issue:** #169
## Summary
Replace per-repo static SSH identity files with short-lived ed25519 deploy
keys that are generated at spin-up and revoked at teardown. Introduce
`bot_bottle/contrib/` as the package for platform-specific provisioners and
ship the first contrib sub-package: `bot_bottle/contrib/gitea/` with
`GiteaDeployKeyProvisioner`. A new `provisioned_key:` block in `git-gate.repos`
entries opts a repo into automatic key lifecycle management; `identity:` stays
valid for operators who supply their own key material.
## Problem
The current `git-gate.repos` entries require an `identity:` field pointing to
a host-side SSH private key (PRD 0047). Keys are static: the operator generates
them once, registers them with the upstream forge, and the same key is reused
across every bottle spin-up. This has several consequences:
- **No automatic revocation.** If a bottle misbehaves or a key leaks, the
operator must notice and manually delete the key from the forge. There is no
teardown hook that does it.
- **Broad blast radius.** A forge deploy key typically grants write access for
the lifetime of the key. A static key that survives bottle teardown continues
to grant that access.
- **Manual rotation burden.** Operators must manage key files on disk, keeping
them secure, rotating them on a schedule, and distributing them across hosts
that run `./cli.py start`.
## Goals / Success Criteria
- `git-gate.repos` entries accept `provisioned_key:` as an alternative to
`identity:`. The parser rejects entries that have both, or neither.
- `provisioned_key.provider: gitea` provisions and revokes deploy keys via the
Gitea HTTP API.
- At prepare time the provisioner generates a fresh ed25519 keypair, registers
the public half as a repo-scoped deploy key, and makes the private key
available to git-gate at the path it expects — the rest of the pipeline is
unchanged.
- At teardown the provisioner deletes the registered deploy key. Failure to
delete halts teardown and propagates the error loudly.
- `bot_bottle/contrib/` is introduced as the package for platform-specific
implementations; the core defines the abstract interface; contrib sub-packages
provide concrete implementations.
- Existing `identity:`-based repos continue to work without change.
- The unit test suite passes unchanged for `identity:` paths; new tests cover
`provisioned_key:` parse, validation, and provisioner dispatch.
## Non-goals
- GitHub, GitLab, or other forge providers (a future contrib sub-package each).
- Dashboard UI for listing or revoking orphaned deploy keys.
- SSH CA certificate approach (rejected in the issue thread in favour of
per-repo deploy keys for simpler revocation, smaller blast radius, and forge
compatibility).
- Key rotation mid-session (keys live for exactly one spin-up / teardown cycle).
- Any change to how `identity:` repos are provisioned.
## Design
### Manifest changes (builds on PRD 0047)
`git-gate.repos.<name>` currently accepts exactly:
```
url (required string)
identity (required string)
host_key (optional string)
```
After this PRD:
```
url (required string)
identity (optional string — mutually exclusive with provisioned_key)
provisioned_key (optional object — mutually exclusive with identity)
host_key (optional string)
```
Exactly one of `identity` or `provisioned_key` must be present. The parser
emits a targeted error for each violation:
```
bottle 'dev' git-gate.repos['bot-bottle'] must set exactly one of
'identity' or 'provisioned_key'; got neither.
bottle 'dev' git-gate.repos['bot-bottle'] must set exactly one of
'identity' or 'provisioned_key'; got both.
```
`provisioned_key` object schema:
```yaml
provisioned_key:
provider: gitea # required; names the contrib module to load
token_env: GITEA_TOKEN # required; name of a host env var holding the API token
api_url: https://... # optional; defaults to https://<host from url>
```
| Field | Type | Notes |
|-------|------|-------|
| `provider` | required string | Must match a sub-package under `bot_bottle/contrib/` |
| `token_env` | required string | Resolved at provision time via `os.environ`; never stored in plan |
| `api_url` | optional string | Override when the API endpoint differs from the git host |
**Example bottle manifest:**
```yaml
git-gate:
user:
name: implementer-bot
email: eric+implementer@dideric.is
repos:
bot-bottle:
url: ssh://git@gitea.dideric.is:30009/didericis/bot-bottle.git
provisioned_key:
provider: gitea
token_env: GITEA_DEPLOY_TOKEN
host_key: "ssh-rsa AAAA..."
```
### `contrib` package structure
```
bot_bottle/
contrib/
__init__.py # empty; no core symbols
gitea/
__init__.py # empty
deploy_key_provisioner.py
```
`contrib` is a flat namespace of forge/platform sub-packages. Each sub-package
is self-contained; the core imports from contrib lazily (inside factory
functions) so that missing optional dependencies in a contrib sub-package don't
break unrelated features.
### Core interface
New file: `bot_bottle/deploy_key_provisioner.py`
```python
from abc import ABC, abstractmethod
class DeployKeyProvisioner(ABC):
@abstractmethod
def create(self, owner_repo: str, title: str) -> tuple[str, bytes]:
"""Generate a keypair and register the public half.
owner_repo: '<owner>/<repo>' portion of the git upstream URL.
title: human-readable label shown in the forge key list.
Returns (key_id, private_key_pem) where key_id is opaque to
the caller and is only passed back to delete()."""
@abstractmethod
def delete(self, owner_repo: str, key_id: str) -> None:
"""Delete the registered deploy key.
Must not raise if the key is already absent (HTTP 404 is success).
Must raise for all other failures so that teardown halts."""
def get_provisioner(provider: str, token: str, api_url: str) -> DeployKeyProvisioner:
"""Instantiate the named contrib provisioner.
Raises ManifestError for unknown providers so the error is caught
at parse time rather than at runtime."""
if provider == "gitea":
from bot_bottle.contrib.gitea.deploy_key_provisioner import (
GiteaDeployKeyProvisioner,
)
return GiteaDeployKeyProvisioner(token=token, api_url=api_url)
from .manifest_util import ManifestError
raise ManifestError(f"unknown provisioned_key provider: {provider!r}")
```
### Gitea contrib implementation
`bot_bottle/contrib/gitea/deploy_key_provisioner.py`:
`create(owner_repo, title)`:
1. Generate an ed25519 keypair via `ssh-keygen -t ed25519 -f <tmpfile> -N ''`
(uses the SSH tooling already required by git-gate; no new Python dependency).
2. Read the private key bytes and the `.pub` file.
3. `POST /api/v1/repos/{owner}/{repo}/keys` with the public key, `title`, and
`read_only: false` (deploy keys always need push access for git-gate).
4. Return `(str(response["id"]), private_key_bytes)`.
`delete(owner_repo, key_id)`:
1. `DELETE /api/v1/repos/{owner}/{repo}/keys/{id}`.
2. Treat HTTP 404 as success (key already gone).
3. Raise `RuntimeError` for any other non-2xx response or network error,
including the status code and response body in the message.
HTTP calls use `urllib.request` from the stdlib; no new runtime dependency.
### `GitEntry` dataclass changes
`bot_bottle/manifest_git.py`:
- Add `ProvisionedKeyConfig` dataclass:
```python
@dataclass(frozen=True)
class ProvisionedKeyConfig:
provider: str
token_env: str
api_url: str # empty string means "derive from UpstreamHost"
```
- `GitEntry`:
- `IdentityFile: str` unchanged internally; empty string when
`provisioned_key` is used; set at provision time, not parse time.
- New field: `ProvisionedKey: ProvisionedKeyConfig | None = None`
- `from_repos_entry` validates the mutually-exclusive constraint and parses
the `provisioned_key` block when present.
### `GitGateUpstream` / prepare-time changes
`bot_bottle/git_gate.py` and `bot_bottle/backend/docker/provision/git.py`:
The existing path writes the identity file path into `GitGateUpstream.IdentityFile`
and docker-cp's it into `/git-gate/creds/<name>-key`. That path stays unchanged
for `identity:` repos.
For `provisioned_key:` repos, a new helper `provision_deploy_key(entry,
stage_dir, bottle_name)` runs before the git-gate sidecar starts:
1. Resolve `token = os.environ[entry.ProvisionedKey.token_env]`. Missing key
raises `RuntimeError` with a clear message naming the env var.
2. Resolve `api_url = entry.ProvisionedKey.api_url or f"https://{entry.UpstreamHost}"`.
3. Instantiate `get_provisioner(entry.ProvisionedKey.provider, token, api_url)`.
4. Call `provisioner.create(entry.UpstreamPath.lstrip("/"), title)` where
`title = f"bot-bottle:{bottle_name}:{entry.Name}"`.
5. Write private key to `stage_dir / f"{entry.Name}-key"` (mode 0o600).
6. Write key ID to `stage_dir / f"{entry.Name}-deploy-key-id"` (plain text).
7. Return the key file path; caller sets `GitGateUpstream.IdentityFile` to it.
`owner_repo` is extracted from `entry.UpstreamPath` (the path component of the
`ssh://` URL, e.g. `/didericis/bot-bottle.git``didericis/bot-bottle`).
### Teardown changes
`bot_bottle/backend/docker/cleanup.py` (or the equivalent teardown path):
After the git-gate sidecar stops, for each `GitEntry` with `ProvisionedKey`
set:
1. Check that `stage_dir / f"{entry.Name}-deploy-key-id"` exists; skip if
absent (provision never ran or already cleaned up).
2. Resolve token and API URL as above.
3. Instantiate provisioner and call `provisioner.delete(owner_repo, key_id)`.
4. On success, log at INFO. On failure, allow the exception to propagate —
teardown halts and the error surfaces to the operator.
A stranded deploy key is a security concern: the operator must know about it
and address it manually. Silent continuation is not acceptable.
The private key file in `stage_dir` is cleaned up as part of normal stage-dir
teardown (no extra step needed).
## Testing strategy
```
python3 -m unittest discover -s tests/unit
```
New / modified test files:
- `tests/unit/test_manifest_git.py` — add cases for:
- `provisioned_key:` accepted with valid `provider`, `token_env`, optional `api_url`
- Both `identity` and `provisioned_key` present → `ManifestError`
- Neither `identity` nor `provisioned_key` present → `ManifestError`
- Unknown key inside `provisioned_key` block → `ManifestError`
- Missing `provider` or `token_env` inside `provisioned_key``ManifestError`
- `tests/unit/test_deploy_key_provisioner.py` — new:
- `get_provisioner("gitea", ...)` returns `GiteaDeployKeyProvisioner`
- `get_provisioner("unknown", ...)` raises `ManifestError`
- `tests/unit/test_contrib_gitea_deploy_key.py` — new (using `unittest.mock`
to stub `urllib.request.urlopen` and `subprocess.run`):
- `create()` calls `ssh-keygen`, POSTs to correct endpoint, returns key ID
- `delete()` DELETEs to correct endpoint
- `delete()` tolerates HTTP 404 (already-deleted key)
- `delete()` raises `RuntimeError` on non-404 HTTP error
## Open questions
None.
+283
View File
@@ -0,0 +1,283 @@
# PRD 0049: Named / Labelled Agents
- **Status:** Draft
- **Author:** didericis
- **Created:** 2026-06-03
- **Issue:** #171
## Summary
At agent launch time, prompt the operator for a short human-readable label
(defaulting to the manifest agent key) and an optional color from the 16-color
ANSI palette. Store both in the bottle's `metadata.json`. Display the label —
rendered in the chosen color — in the dashboard's active-agents pane, replacing
the bare manifest key. Inject the label and color into the in-container
`claude.json` as `name` / `color` so Claude Code can surface them in its own
harness when upstream support lands.
## Problem
The dashboard's agents pane identifies each running instance by its manifest
agent key (e.g., `implementer`) plus a random slug suffix. When an operator
runs three `implementer` bottles simultaneously — one each for three different
repos — the pane shows:
```
[docker] a3f9 implementer started 14:02:11 [egress,pipelock]
[docker] b81c implementer started 14:03:45 [egress,pipelock]
[docker] d220 implementer started 14:05:01 [egress,pipelock]
```
There is no way to tell which bottle is working on which task without attaching
to each one in turn. The slug is opaque; the manifest key is shared. Operators
working a multi-bottle session resort to keeping a mental map of slug→task,
which breaks the moment they switch windows.
## Goals / Success Criteria
1. After the operator selects an agent name (dashboard picker or CLI argument),
they are prompted for a label. The prompt suggests the manifest key as the
default; pressing Enter (or providing no input) accepts it. The label may
contain any printable characters up to 64 bytes.
2. After the label prompt, the operator is optionally prompted for a color from
the 16-color ANSI palette (names: `black`, `red`, `green`, `yellow`, `blue`,
`magenta`, `cyan`, `white`, `bright-black`, `bright-red`, `bright-green`,
`bright-yellow`, `bright-blue`, `bright-magenta`, `bright-cyan`,
`bright-white`). Pressing Enter without a selection skips color entirely.
3. `label` and `color` are stored in `BottleMetadata` and written to the
bottle's `metadata.json`. Both fields default to `""` (empty / unset).
4. `ActiveAgent` carries `label` and `color`; `enumerate_active()` reads them
from `metadata.json`.
5. `_format_agent_row` uses the label when non-empty (falling back to
`agent_name`). If a non-empty color is set and the terminal supports it, the
label substring is rendered in that color.
6. `BottleSpec` carries `label` and `color`; the docker backend's `prepare`
step copies them into `BottleMetadata`.
7. `agent_provider.py` writes `label``"name"` and `color``"color"` into
the generated `claude.json`, alongside the existing fields. Fields are
omitted when empty.
8. The dashboard's `_new_agent_flow` (PRD 0020) includes the label+color step
between agent selection and the backend picker.
9. `cmd_start` (CLI) includes the label+color step after argument validation
and before prepare-with-preflight.
10. All existing unit tests stay green; no new tests are required for this
change (the label/color fields are thin plumbing with no branching logic
worth unit-testing beyond the already-tested metadata read/write path).
## Non-goals
- Showing the agent label inside the Claude Code TUI (status line, terminal
title, custom header). That requires upstream Claude Code / codex support.
Writing to `claude.json` is best-effort scaffolding for when that lands.
- Per-bottle color affecting anything outside the dashboard agents pane (e.g.,
proposal-pane highlights, log prefixes).
- Validating or constraining label content beyond the 64-byte printable cap.
- Persisting color-pair state across dashboard restarts (color pairs are
initialized fresh each session).
- Editing the label or color of an already-running bottle.
- Exposing label/color via `./cli.py list` (out of scope for v1; trivial to
add later since the field will be in metadata).
## Design
### Data flow
```
operator input
BottleSpec.label, BottleSpec.color
├─► docker/prepare.py → BottleMetadata.label / .color → metadata.json
└─► agent_provider.py → claude.json {"name": label, "color": color}
(omitted when empty)
dashboard refresh
enumerate_active() → read_metadata(slug) → ActiveAgent.label / .color
_format_agent_row → label (colored) in the row string
```
### BottleSpec changes
```python
@dataclass(frozen=True)
class BottleSpec:
manifest: Manifest
agent_name: str
copy_cwd: bool
user_cwd: str
identity: str = ""
label: str = "" # operator-chosen display name; defaults to agent_name at render time
color: str = "" # one of the 16 ANSI color names, or "" for terminal default
```
`label` and `color` default to `""` so all existing callers remain valid with
no changes.
### BottleMetadata changes
Add two new fields with backward-compatible defaults:
```python
@dataclass
class BottleMetadata:
identity: str
agent_name: str
cwd: str
copy_cwd: bool
started_at: str
compose_project: str
backend: str
label: str = ""
color: str = ""
```
`metadata.json` written by older bot-bottle versions won't have these keys;
`read_metadata` already uses `dict.get` with defaults, so existing slugs load
cleanly with `label=""`, `color=""`.
### ActiveAgent changes
```python
@dataclass(frozen=True)
class ActiveAgent:
backend_name: str
slug: str
agent_name: str
started_at: str
services: tuple[str, ...]
label: str = ""
color: str = ""
```
`enumerate_active()` copies `label` and `color` out of `BottleMetadata` when
constructing each `ActiveAgent`. The smolmachines backend gets the same
additions for symmetry; it reads from its own metadata path.
### Dashboard row rendering
`_format_agent_row` already falls through cleanly on missing fields. The
change is:
```python
display_name = a.label if a.label else a.agent_name
```
Color rendering uses the existing `_try_init_green()` pattern as a model.
A `_color_pair_for(color_name)` helper initialises a fresh curses color pair
for the requested named color and returns its attr (or 0 on failure). Each
unique color in the active agent list gets its own pair index. Color pairs are
allocated lazily and cached in a `dict[str, int]` that lives for the duration
of the dashboard session.
The 16 ANSI color name → curses constant mapping:
| Name | curses constant |
|------|----------------|
| `black` | `curses.COLOR_BLACK` |
| `red` | `curses.COLOR_RED` |
| `green` | `curses.COLOR_GREEN` |
| `yellow` | `curses.COLOR_YELLOW` |
| `blue` | `curses.COLOR_BLUE` |
| `magenta` | `curses.COLOR_MAGENTA` |
| `cyan` | `curses.COLOR_CYAN` |
| `white` | `curses.COLOR_WHITE` |
| `bright-*` | same constant + `curses.A_BOLD` |
Terminals that don't support color fall back to plain text (the helper returns
0, which ORed in is a no-op — same pattern as `_try_init_green`).
### Label + color prompt — dashboard
In `_new_agent_flow`, after `_picker_modal` returns a non-None name and before
`_backend_picker_modal`:
```python
label, color = _label_color_modal(stdscr, default_label=picked)
```
`_label_color_modal` uses `curses.endwin()` → text-mode prompts → restore
(the same drop-and-resume pattern as the existing editor flow and preflight
Y/N). Two sequential prompts:
```
bot-bottle: agent label [implementer]: <operator types>
bot-bottle: color (red/green/blue/… or Enter to skip): <operator types>
```
Invalid color names are silently ignored (treated as empty). The function
returns `(label, color)` — both strings, both possibly `""`.
### Label + color prompt — CLI
In `cmd_start`, after argument parsing and before `_launch_bottle`:
```python
label = _text_prompt_label(args.name)
color = _text_prompt_color()
```
`_text_prompt_label(default)` writes `"bot-bottle: agent label [{default}]: "`
to stderr and returns the stripped input (or `default` if blank).
`_text_prompt_color()` writes the color prompt and returns the stripped input
(or `""` if blank or invalid).
Both use `read_tty_line()` (already in `start.py`) for the read.
### Claude Code config injection
In `agent_provider.py`, where `claude_config.write_text(...)` is called,
expand the JSON dict conditionally:
```python
payload = {
"hasCompletedOnboarding": True,
"theme": "dark",
"bypassPermissionsModeAccepted": True,
"projects": claude_projects,
}
if spec.label:
payload["name"] = spec.label
if spec.color:
payload["color"] = spec.color
claude_config.write_text(json.dumps(payload, indent=2) + "\n")
```
`spec` here is the `AgentProvisionSpec` (or equivalent) that `agent_provider`
already receives; it needs `label` and `color` threaded in from `BottleSpec`
through whatever plan/provision object the provider operates on.
## Implementation chunks
Two PRs, each independently mergeable.
### Chunk 1 — schema + storage
- Add `label: str = ""` and `color: str = ""` to `BottleSpec`,
`BottleMetadata`, and `ActiveAgent`.
- `docker/prepare.py`: copy `spec.label` / `spec.color` into `BottleMetadata`.
- `docker/enumerate.py`: copy `metadata.label` / `metadata.color` into
`ActiveAgent`.
- `agent_provider.py` (or the plan object it reads): thread label/color through
to `claude.json` write.
- Smolmachines backend: parallel changes to metadata read/write and
`ActiveAgent` construction.
- No prompt changes; no UI changes. All existing behavior is identical.
### Chunk 2 — prompts + display
- `start.py`: add `_text_prompt_label` and `_text_prompt_color`; call them in
`cmd_start` before `_launch_bottle`; pass `label` / `color` into `BottleSpec`.
- `dashboard.py`: add `_label_color_modal` (drop-and-resume); call it in
`_new_agent_flow`; pass label/color into `BottleSpec`; add
`_color_pair_for` helper; update `_format_agent_row` to use `a.label` with
color rendering.
## Open questions
None.
@@ -0,0 +1,343 @@
- **Status:** Active
- **Author:** didericis
- **Created:** 2026-06-03
- **Issue:** #174
## Summary
The `./cli.py dashboard` command has grown from its PRD 0013 roots
(triage supervise proposals) into a parallel-agent control surface
(PRDs 0019/0020/0021): an active-agents pane, agent picker + start,
re-attach, per-bottle stop, tmux split-pane handoff, operator-
initiated `routes`/`pipelock` edits. Each chunk is reasonable on its
own; together they make the dashboard the largest CLI file in the
repo and the thing most likely to break on a rough edge (curses /
tmux / docker-exec / metadata-discovery interactions).
This PRD reverses that scope creep. The dashboard is reduced to the
**supervise-plane triage TUI** it was in PRDs 00130016: list pending
proposals, approve / modify / reject each one, write audit entries,
deliver the response that unblocks the agent's tool call. Everything
that's about *starting / re-entering / stopping* bottles, or about
*operator-initiated* config edits, comes out. The command is renamed
`./cli.py supervise` so the name matches what it does after the cut.
Future agent-management UX is explicitly punted: if and when a
control surface for parallel agents resurfaces, the working
assumption (per the issue) is that a web GUI — usable from mobile
— is a better second pass than another round of curses iteration.
That decision is not in this PRD's scope; this PRD only removes the
half-built local-curses path so we stop maintaining it.
## Problem
Three concrete pains, all downstream of the dashboard's growth:
1. **Surface area vs. polish.** `dashboard.py` is ~1740 lines;
`dashboard_model.py` adds another ~420. The interactions among
curses, modals, tmux split-pane, docker-exec handoff, agent
provider templates, metadata-driven re-attach, and
ExitStack-free bottle ownership are intricate enough that
shipping the next polish increment costs more than it returns.
2. **No clear ownership of "starts and stops bottles".** Today
that responsibility is split: `./cli.py start` owns one-shot
sessions; the dashboard owns multi-session bottles it started
itself; `./cli.py cleanup` owns everything else. The dashboard
tracking its own `bottles: dict[str, (cm, bottle, identity)]`
that doesn't survive a quit is a confusing third lane.
3. **Wrong target shape for a "manage many agents" UI.** The
parallel-agent experience the dashboard reaches for is mobile-
meaningful — checking in on agents from a phone is the high-
value case — and curses inside an SSH session is the wrong
tool for that. Continuing to polish a local-only TUI delays
the right next investment.
The triage half of the dashboard isn't suffering from any of these.
Pending proposals are a small, well-scoped, real workload, and the
PRD 00130016 surface for handling them is the right shape. The
problem is everything that got bolted onto that core after.
## Goals / Success Criteria
1. The supervise TUI starts up, lists pending proposals across all
running bottles, and supports approve / modify / reject + the
`--once` non-interactive mode — exactly as PRDs 00130016
specified, minus everything 0019/0020/0021 added.
2. The CLI subcommand is renamed `supervise` (was `dashboard`). The
old name is not aliased — this PRD is intentionally a
compat/breaking change (the issue carries the
`Compat/Breaking` label).
3. `dashboard.py` shrinks to a single proposal-triage curses loop:
no agents pane, no Tab pane switching, no agent picker, no
start / re-attach / stop verbs, no tmux split-pane, no
`e`/`p` operator-edit verbs, no per-process `bottles` dict.
4. `dashboard_model.py` is collapsed into whatever
`supervise.py` (CLI) needs; the model module is removed if it
has no purpose after the cut.
5. The proposal-side apply paths in `bot_bottle/backend/docker/
egress_apply.py`, `pipelock_apply.py`, and `capability_apply.py`
are unchanged — they are still called by the approve path.
6. The supervise-sidecar / proposal-queue protocol (PRD 0013) is
unchanged: the agent's experience is identical.
7. The previously-active PRDs that this one undoes are marked
`Superseded by PRD 0049`:
- PRD 0019 — active-agents pane + agent-scoped edit verbs
- PRD 0020 — start / re-attach / stop from the dashboard
- PRD 0021 — tmux split-pane
## Non-goals
- **A web GUI for managing agents.** The issue floats this as a
second pass; this PRD does not design or commit to it. The cut
is "remove the path we no longer want to invest in", not
"build the replacement".
- **A separate CLI for operator-initiated routes / pipelock
edits.** Today those edits live as `e` / `p` keys inside the
dashboard. After this PRD they don't exist anywhere — operators
who need ad-hoc edits use the same path the agents do (call the
supervise tool from inside the bottle) or hand-edit the host-
side files and restart the sidecar. Adding a `./cli.py routes
edit <slug>` verb is a follow-up if the loss bites.
- **Removing `./cli.py start` or changing its semantics.** Start
remains the one-shot launch path. PRD 0020's bottle-outlives-
process model is removed; the only path to a long-running
bottle is `./cli.py start` (foreground) plus `cli.py cleanup`
for teardown.
- **Removing the supervise-sidecar protocol or any of the three
block-remediation engines.** PRDs 00130016 stay Active. The
agent's view of the world doesn't change.
- **Renaming `dashboard` anywhere other than the CLI entry
point.** The dashboard-related docs (PRDs, decision records,
research notes) keep their historical references — they
describe the state of the world at the time they were written,
and the Status: Superseded line is the marker that the world
has moved on.
- **Migrating the proposal-queue file layout.** The queue still
lives at `~/.bot-bottle/queue/<slug>/`; the audit log still
lives at `~/.bot-bottle/audit/<component>-<slug>.log`. The CLI
surface changes; the on-disk surface does not.
## Scope
### In scope
- **Rename the subcommand.** `./cli.py dashboard` becomes
`./cli.py supervise`. The module moves from `bot_bottle/cli/
dashboard.py` to `bot_bottle/cli/supervise.py`. The dispatcher
in `bot_bottle/cli/__init__.py` and the help text both update.
- **Strip the curses loop to proposal-only.** The remaining
surface is: list pending proposals (with the new-arrival bell
from PRD 0013), Enter for detail view,
`a`/`m`/`r` for approve / modify / reject, `q` to quit. No
agents pane, no Tab, no agent picker, no `n`/`x`/`e`/`p`, no
tmux dispatch, no `bottles` dict on the main loop.
- **Drop unused helpers.** `_picker_modal`, `_preflight_modal`,
`_backend_picker_modal`, `_new_agent_flow`, `_attach_to_bottle`,
`_attach_in_tmux`, `_attach_via_handoff`, `_tmux_*`,
`_ensure_right_pane`, `_redirect_stderr_to_file`,
`_route_op_to_right_pane`, `_stop_bottle_flow`,
`_operator_edit_*_flow`, `operator_edit_routes`,
`operator_edit_allowlist`, and their imports come out.
- **Collapse the model module.** `dashboard_model.py`'s
proposal-side helpers (`QueuedProposal`, `discover_pending`,
`_approval_status`, `_detail_lines`,
`_failed_url_host`, `_proposed_payload_label`,
`_suffix_for_tool`, `_REFRESH_INTERVAL_MS`) move back into
`supervise.py` (CLI) or into `bot_bottle/supervise.py`
(the daemon-side module) — wherever they fit. The agents /
picker / tmux helpers in that module (`PANE_*`,
`_filter_agents`, `_running_counts`, `_format_agent_row`,
`_selection_status`, `_selected_agent`, `_bottle_for_slug`,
`_pick_next_after_stop`, `_agent_runtime_args`,
`_build_resume_argv_with_fallback`, `_build_split_pane_argv`,
`_build_respawn_pane_argv`, `_in_tmux`,
`discover_active_agents`) are deleted.
- **Mark superseded PRDs.** The Status line on PRDs 0019, 0020,
and 0021 changes to `Superseded by [PRD 0049](0049-strip-
dashboard-to-supervisor-tui.md)`.
- **Test cleanup.** Any test that targets a removed surface (the
agent picker, the tmux split helpers, the start-from-dashboard
flow, the operator-edit flows, `discover_active_agents`)
comes out. Tests covering proposal triage stay.
- **Help / usage strings.** `bot_bottle/cli/__init__.py`'s usage
block updates the command name and one-liner.
### Out of scope
- Any new feature in the supervise TUI. The cut is purely
subtractive (except for the rename).
- Behavior changes in `./cli.py start`, `cli.py cleanup`,
`cli.py resume`, `cli.py list`, `cli.py info`, `cli.py edit`,
`cli.py init` — unchanged.
- Changes to the supervise sidecar (`supervise_server.py`,
`supervise.py` daemon module). The wire protocol stays.
- Changes to the routes / pipelock / capability apply engines.
- Migration helpers, deprecation warnings, or a transitional
`dashboard` alias for `supervise`. The label on the issue says
Compat/Breaking; the rename is a hard cutover.
## Proposed design
### Final shape of the TUI
After this PRD the `./cli.py supervise` curses surface is:
```
bot-bottle supervise (3 pending)
─────────────────────────────────────────────────────────
> 03:14:22 [implementer-cy7a6] egress-block abc123… add
github.com/foo
03:13:55 [researcher-9xqs1] pipelock-block def456… allow
registry.npmjs.org
03:13:10 [implementer-cy7a6] capability-block ghi789… install
ripgrep
─────────────────────────────────────────────────────────
[j/k] move [Enter] view [a] approve [m] modify [r] reject [q] quit
```
- One pane. No Tab. `j` / `k` / arrows move through the queue.
- Enter opens the existing detail view (justification +
proposed-file body + the green pipelock host-extraction hint).
`a` / `m` / `r` work from both the list view and the detail
view, same as today.
- `q` / Esc quits. There are no dashboard-owned bottles, so no
per-process teardown decision — `q` just exits.
- The new-arrival bell stays, because it is a real win for the
operator's "I was typing at claude and a proposal landed" case.
No tmux-specific focus management remains.
### Code organisation
After the cut, the CLI module looks roughly like:
```
bot_bottle/cli/supervise.py
- cmd_supervise(argv)
- _list_once() # --once mode
- _main_loop(stdscr) # proposal-only
- _render(stdscr, pending, ...)
- _detail_view(stdscr, qp, ...)
- _modify(stdscr, qp)
- _prompt(stdscr, label)
- _write_crash_log(exc)
- approve(qp, *, notes, final_file)
- reject(qp, *, reason)
- QueuedProposal, discover_pending
- _detail_lines, _approval_status,
_failed_url_host,
_proposed_payload_label,
_suffix_for_tool
```
`dashboard_model.py` has no purpose once the agents / picker /
tmux helpers are gone, so it is removed and the surviving
proposal-side helpers move into `supervise.py` directly. The
PRD-0013 refactor that split model out (`refactor: extract
dashboard state/model layer into dashboard_model.py`) was
load-bearing for the bigger dashboard surface; with the surface
shrunk back, the split is no longer justified.
### Removed PRDs: how to mark them
The three superseded PRDs keep their bodies intact. Only the
Status line at the top changes:
```
- **Status:** Superseded by [PRD
0049](0049-strip-dashboard-to-supervisor-tui.md)
```
The PRD's own Goals / Success Criteria are left as the historical
record of what the feature shipped — readers tracing back from the
code or the git log land in a PRD that explains what once was, with
a clear pointer forward. No PRD body is rewritten.
### Tests to keep, tests to remove
Keep:
- `tests/cli/test_dashboard*.py` cases that exercise
`discover_pending`, `approve`, `reject`, `_detail_lines`,
`_approval_status`, `_failed_url_host`,
`_proposed_payload_label`, `_suffix_for_tool`,
`_modify` / `edit_in_editor`.
- `tests/cli/test_dashboard_once.py` (or equivalent) — the
`--once` listing mode.
Remove:
- Any test of `_picker_modal`, `_preflight_modal`,
`_backend_picker_modal`, `_new_agent_flow`, `_attach_*`,
`_tmux_*`, `_route_op_to_right_pane`,
`_redirect_stderr_to_file`, `_stop_bottle_flow`,
`_operator_edit_*`, `_filter_agents`, `_running_counts`,
`_format_agent_row`, `_selection_status`,
`_selected_agent`, `_bottle_for_slug`,
`_pick_next_after_stop`, `_agent_runtime_args`,
`_build_*_argv`, `discover_active_agents`.
- The test files that exist solely to cover those (e.g.,
`test_dashboard_picker.py`, `test_dashboard_tmux.py`,
`test_dashboard_attach.py`, `test_dashboard_agents.py`
whichever of these exist after the file walk).
Files are renamed `test_supervise_*.py` to mirror the module
rename. The rename is mechanical; no test logic changes.
## Implementation chunks
Sized for a single PR each.
1. **Strip + rename in one cut.** Move `bot_bottle/cli/
dashboard.py` to `bot_bottle/cli/supervise.py`, delete the
removed helpers, delete `dashboard_model.py`, inline the
surviving helpers, update the dispatcher + usage in
`bot_bottle/cli/__init__.py`, rename tests to match, mark
PRDs 0019/0020/0021 as superseded. One commit per logical
piece inside the PR (rename, strip, supersede notes,
tests).
2. **Activate PRD 0049.** Flip this PRD's Status line from
Draft to Active in the same PR as chunk 1 once the
implementation lands. (The repo convention is that a PRD's
shipping commit is also the Status flip — see the recent
`docs(prd): activate PRD 0048…` commit shape.)
The PR closes issue #174.
## Open questions
1. **`e` / `p` operator-initiated edits — gone for good or
moved to a separate CLI verb?** The PRD removes them with no
replacement. The simplest replacement is `./cli.py routes
edit <slug>` and `./cli.py pipelock edit <slug>`, sharing
the existing `apply_routes_change` / `apply_allowlist_change`
engines. If the loss is felt within the first parallel
run after this lands, that follow-up is a small PR. Leaving
it for a separate PRD so this one stays subtractive.
2. **`--once` output shape.** The text listing today emits one
proposal per line. Worth keeping exactly as-is for
scripting consumers; this PRD does not change it. Flagging
only because the rename could tempt a tweak.
3. **Audit-log entry shape for an unprompted edit applied via
a future `routes edit` CLI verb.** Today's
`operator_edit_routes` writes an `ACTION_OPERATOR_EDIT`
audit entry. With those flows removed the constant has no
callers inside this PRD's scope. Keep the constant exported
from `supervise.py` (it's already an `__all__` member) so a
follow-up CLI verb can re-use the same audit shape without
re-introducing dead code first.
## References
- Issue
[#174](https://gitea.dideric.is/didericis/bot-bottle/issues/174)
— the request: "strip the dashboard down into just a TUI for
managing agent requests for new egress routes and new
capabilities."
- PRD 0013 — supervise plane foundation (the floor this PRD
reverts the dashboard to).
- PRDs 0014 / 0015 / 0016 — block-remediation engines that the
supervise TUI continues to drive on approve.
- PRDs 0019 / 0020 / 0021 — the bolted-on capabilities this PRD
removes.
-46
View File
@@ -277,51 +277,5 @@ class TestBottleMetadataBackend(_FakeHomeMixin, unittest.TestCase):
self.assertEqual("", loaded.backend)
class TestBottleForSlugBackend(_FakeHomeMixin, unittest.TestCase):
"""PRD 0040: _bottle_for_slug constructs the right bottle type."""
def setUp(self):
self._setup_fake_home()
def tearDown(self):
self._teardown_fake_home()
def test_docker_metadata_returns_docker_bottle(self):
from bot_bottle.backend.docker.bottle import DockerBottle
from bot_bottle.cli.dashboard import _bottle_for_slug
write_metadata(BottleMetadata(
identity="dev-d1",
agent_name="dev",
cwd="",
copy_cwd=False,
started_at="2026-06-02T00:00:00+00:00",
compose_project="bot-bottle-dev-d1",
backend="docker",
))
bottle, _ = _bottle_for_slug("dev-d1", {}, None)
self.assertIsInstance(bottle, DockerBottle)
def test_smolmachines_metadata_returns_smolmachines_bottle(self):
from bot_bottle.backend.smolmachines.bottle import SmolmachinesBottle
from bot_bottle.cli.dashboard import _bottle_for_slug
write_metadata(BottleMetadata(
identity="dev-s1",
agent_name="dev",
cwd="",
copy_cwd=False,
started_at="2026-06-02T00:00:00+00:00",
compose_project="",
backend="smolmachines",
))
bottle, _ = _bottle_for_slug("dev-s1", {}, None)
self.assertIsInstance(bottle, SmolmachinesBottle)
def test_no_metadata_defaults_to_docker_bottle(self):
from bot_bottle.backend.docker.bottle import DockerBottle
from bot_bottle.cli.dashboard import _bottle_for_slug
bottle, _ = _bottle_for_slug("unknown-slug", {}, None)
self.assertIsInstance(bottle, DockerBottle)
if __name__ == "__main__":
unittest.main()
+166
View File
@@ -0,0 +1,166 @@
"""Unit: GiteaDeployKeyProvisioner (PRD 0048, contrib/gitea)."""
from __future__ import annotations
import json
import unittest
import urllib.error
from io import BytesIO
from pathlib import Path
from tempfile import mkdtemp
from unittest.mock import MagicMock, call, patch
from bot_bottle.contrib.gitea.deploy_key_provisioner import (
GiteaDeployKeyProvisioner,
_split_owner_repo,
)
def _provisioner() -> GiteaDeployKeyProvisioner:
return GiteaDeployKeyProvisioner(
token="test-token", api_url="https://gitea.example.com"
)
def _urlopen_response(body: dict, status: int = 200) -> MagicMock:
resp = MagicMock()
resp.read.return_value = json.dumps(body).encode()
resp.status = status
resp.__enter__ = lambda s: s
resp.__exit__ = MagicMock(return_value=False)
return resp
def _http_error(code: int, body: str = "") -> urllib.error.HTTPError:
return urllib.error.HTTPError(
url="http://x",
code=code,
msg="err",
hdrs=None, # type: ignore[arg-type]
fp=BytesIO(body.encode()),
)
class TestCreate(unittest.TestCase):
def test_create_calls_ssh_keygen_and_posts_to_api(self):
provisioner = _provisioner()
fake_key_id = 42
fake_private = b"PRIVATE_KEY"
fake_public = "ssh-ed25519 AAAA fake"
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=fake_private,
), patch(
"bot_bottle.contrib.gitea.deploy_key_provisioner.Path.read_text",
return_value=fake_public + "\n",
):
mock_urlopen.return_value = _urlopen_response({"id": fake_key_id})
key_id, private_bytes = provisioner.create(
"didericis/bot-bottle", "bot-bottle:slug:repo"
)
# ssh-keygen called with ed25519
mock_run.assert_called_once()
run_args = mock_run.call_args.args[0]
self.assertIn("ssh-keygen", run_args)
self.assertIn("-t", run_args)
self.assertIn("ed25519", run_args)
# POST body contains public key
post_call = mock_urlopen.call_args.args[0]
payload = json.loads(post_call.data)
self.assertEqual(fake_public, payload["key"])
self.assertFalse(payload["read_only"])
# Correct URL
self.assertIn(
"/api/v1/repos/didericis/bot-bottle/keys", post_call.full_url
)
self.assertEqual(str(fake_key_id), key_id)
self.assertEqual(fake_private, private_bytes)
def test_create_raises_on_http_error(self):
provisioner = _provisioner()
with patch(
"bot_bottle.contrib.gitea.deploy_key_provisioner.subprocess.run"
), patch(
"bot_bottle.contrib.gitea.deploy_key_provisioner.urllib.request.urlopen",
side_effect=_http_error(403, "forbidden"),
), patch(
"bot_bottle.contrib.gitea.deploy_key_provisioner.Path.read_bytes",
return_value=b"pk",
), patch(
"bot_bottle.contrib.gitea.deploy_key_provisioner.Path.read_text",
return_value="ssh-ed25519 AAAA\n",
):
with self.assertRaises(RuntimeError) as ctx:
provisioner.create("owner/repo", "title")
self.assertIn("403", str(ctx.exception))
class TestDelete(unittest.TestCase):
def test_delete_calls_correct_endpoint(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("didericis/bot-bottle", "99")
req = mock_urlopen.call_args.args[0]
self.assertIn("/api/v1/repos/didericis/bot-bottle/keys/99", req.full_url)
self.assertEqual("DELETE", req.get_method())
def test_delete_tolerates_404(self):
provisioner = _provisioner()
with patch(
"bot_bottle.contrib.gitea.deploy_key_provisioner.urllib.request.urlopen",
side_effect=_http_error(404),
):
provisioner.delete("owner/repo", "123") # must not raise
def test_delete_raises_on_non_404_http_error(self):
provisioner = _provisioner()
with patch(
"bot_bottle.contrib.gitea.deploy_key_provisioner.urllib.request.urlopen",
side_effect=_http_error(500, "internal server error"),
):
with self.assertRaises(RuntimeError) as ctx:
provisioner.delete("owner/repo", "7")
self.assertIn("500", str(ctx.exception))
def test_delete_raises_on_url_error(self):
provisioner = _provisioner()
with patch(
"bot_bottle.contrib.gitea.deploy_key_provisioner.urllib.request.urlopen",
side_effect=urllib.error.URLError("connection refused"),
):
with self.assertRaises(RuntimeError) as ctx:
provisioner.delete("owner/repo", "7")
self.assertIn("connection refused", str(ctx.exception))
class TestSplitOwnerRepo(unittest.TestCase):
def test_simple(self):
self.assertEqual(("owner", "repo"), _split_owner_repo("owner/repo"))
def test_raises_on_missing_slash(self):
with self.assertRaises(ValueError):
_split_owner_repo("noslash")
def test_raises_on_empty_owner(self):
with self.assertRaises(ValueError):
_split_owner_repo("/repo")
def test_raises_on_empty_repo(self):
with self.assertRaises(ValueError):
_split_owner_repo("owner/")
if __name__ == "__main__":
unittest.main()
-492
View File
@@ -1,492 +0,0 @@
"""Unit: dashboard's row-formatting + selection helpers (PRD 0019)."""
from __future__ import annotations
import tempfile
import unittest
from pathlib import Path
from unittest import mock
from bot_bottle import supervise
from bot_bottle.cli import dashboard
class _FakeHomeMixin:
def _setup_fake_home(self) -> None:
self._tmp = tempfile.TemporaryDirectory(prefix="dashboard-aa-test.")
original = supervise.bot_bottle_root
def fake_root() -> Path:
return Path(self._tmp.name) / ".bot-bottle"
supervise.bot_bottle_root = fake_root # type: ignore[assignment]
self._restore_home = lambda: setattr(supervise, "bot_bottle_root", original)
def _teardown_fake_home(self) -> None:
self._restore_home()
self._tmp.cleanup()
class TestFormatAgentRow(unittest.TestCase):
"""One-line row formatting for the agents pane (PRD 0019 chunk 2)."""
def _agent(self, **overrides) -> dashboard.ActiveAgent:
defaults = dict(
backend_name="docker",
slug="dev-abc12",
agent_name="implementer",
started_at="2026-05-26T02:55:01+00:00",
services=("egress", "git-gate", "pipelock", "supervise"),
)
defaults.update(overrides)
return dashboard.ActiveAgent(**defaults)
def test_renders_slug_name_time_services(self):
s = dashboard._format_agent_row(self._agent(), 200)
self.assertIn("dev-abc12", s)
self.assertIn("implementer", s)
self.assertIn("02:55:01", s)
self.assertIn("egress,git-gate,pipelock,supervise", s)
def test_starting_label_when_no_services(self):
# Race window: compose project is up but containers haven't
# been picked up by `docker ps` yet.
s = dashboard._format_agent_row(self._agent(services=()), 200)
self.assertIn("(starting)", s)
def test_filters_agent_service_from_display(self):
# The `agent` service is always present for an active bottle;
# listing it is noise. The row should show only the sidecars.
s = dashboard._format_agent_row(
self._agent(services=("agent", "pipelock", "supervise")), 200,
)
self.assertIn("[pipelock,supervise]", s)
self.assertNotIn("agent,", s)
self.assertNotIn(",agent", s)
def test_only_agent_service_shows_starting(self):
# A bottle whose only running service is `agent` (sidecars
# still warming up) renders as `(starting)`.
s = dashboard._format_agent_row(self._agent(services=("agent",)), 200)
self.assertIn("(starting)", s)
def test_question_mark_when_no_started_at(self):
s = dashboard._format_agent_row(self._agent(started_at=""), 200)
self.assertIn("started ?", s)
def test_truncates_to_maxw(self):
s = dashboard._format_agent_row(self._agent(), 30)
self.assertLessEqual(len(s), 30)
self.assertTrue(s.endswith(""))
class TestSelectionStatus(unittest.TestCase):
"""Idle-state status-line text for the agents-pane focus
(PRD 0019 chunk 3). Empty when the proposals pane is focused;
surfaces the selected agent (or a clear placeholder) when the
agents pane is focused."""
def _agent(self, slug: str) -> dashboard.ActiveAgent:
return dashboard.ActiveAgent(
backend_name="docker",
slug=slug, agent_name="x", started_at="", services=(),
)
def test_empty_when_proposals_focused(self):
s = dashboard._selection_status(
dashboard.PANE_PROPOSALS, [self._agent("a-1")], 0,
)
self.assertEqual("", s)
def test_no_agents_message_when_agents_pane_empty(self):
s = dashboard._selection_status(dashboard.PANE_AGENTS, [], 0)
self.assertEqual("[no active agents]", s)
def test_shows_selected_slug(self):
agents = [self._agent("a-1"), self._agent("b-2"), self._agent("c-3")]
s = dashboard._selection_status(dashboard.PANE_AGENTS, agents, 1)
self.assertEqual("[selected: b-2]", s)
def test_out_of_bounds_falls_back_to_no_selection(self):
agents = [self._agent("only")]
s = dashboard._selection_status(dashboard.PANE_AGENTS, agents, 99)
self.assertEqual("[no agent selected]", s)
class TestFilterAgents(unittest.TestCase):
"""Pure-function picker filter (PRD 0020 chunk 2). Curses-free
so we can exercise the substring + case-insensitivity rules
directly."""
NAMES = ["implementer", "researcher", "triage-bot", "ImplDeluxe"]
def test_empty_query_returns_all(self):
self.assertEqual(self.NAMES, dashboard._filter_agents("", self.NAMES))
def test_substring_match(self):
self.assertEqual(
["implementer", "ImplDeluxe"],
dashboard._filter_agents("impl", self.NAMES),
)
def test_case_insensitive(self):
self.assertEqual(
["implementer", "ImplDeluxe"],
dashboard._filter_agents("IMPL", self.NAMES),
)
def test_no_match_returns_empty(self):
self.assertEqual([], dashboard._filter_agents("zzz", self.NAMES))
def test_preserves_input_order(self):
# Filtering should never re-sort; the picker draws in the
# order the manifest exposed.
out = dashboard._filter_agents("e", ["beta", "alpha", "echo"])
self.assertEqual(["beta", "echo"], out)
class TestDashboardManifestLoading(unittest.TestCase):
def test_new_agent_flow_empty_manifest_has_no_picker_entries(self):
manifest = dashboard.Manifest.from_json_obj({"bottles": {}, "agents": {}})
with mock.patch("bot_bottle.cli.dashboard._picker_modal", return_value=None) as picker:
status = dashboard._new_agent_flow(
None, manifest, {}, [], tmux_state=None, # type: ignore[arg-type]
)
picker.assert_called_once()
self.assertEqual([], picker.call_args.args[1])
self.assertIn("no agents configured", status)
class TestRunningCounts(unittest.TestCase):
"""Per-agent running-count surfaced in the picker so the
operator sees `(N running)` before picking. Counts come from
the dashboard's current `discover_active_agents` snapshot."""
def _agent(self, agent_name: str) -> dashboard.ActiveAgent:
return dashboard.ActiveAgent(
backend_name="docker",
slug=f"{agent_name}-abc",
agent_name=agent_name,
started_at="",
services=(),
)
def test_empty_when_no_active_agents(self):
self.assertEqual({}, dashboard._running_counts({}, []))
def test_one_per_unique_agent_name(self):
agents = [self._agent("a"), self._agent("b"), self._agent("c")]
self.assertEqual(
{"a": 1, "b": 1, "c": 1},
dashboard._running_counts({}, agents),
)
def test_counts_collisions(self):
agents = [
self._agent("implementer"),
self._agent("implementer"),
self._agent("researcher"),
]
self.assertEqual(
{"implementer": 2, "researcher": 1},
dashboard._running_counts({}, agents),
)
class TestSelectedAgent(unittest.TestCase):
"""`_selected_agent` is what chunk 4's e/p key handlers use to
decide whether to fire and which agent to target."""
def _agent(self, slug: str, services: tuple[str, ...] = ()) -> dashboard.ActiveAgent:
return dashboard.ActiveAgent(
backend_name="docker",
slug=slug, agent_name="x", started_at="", services=services,
)
def test_none_when_proposals_focused(self):
agents = [self._agent("a-1")]
self.assertIsNone(
dashboard._selected_agent(dashboard.PANE_PROPOSALS, agents, 0),
)
def test_none_when_no_agents(self):
self.assertIsNone(
dashboard._selected_agent(dashboard.PANE_AGENTS, [], 0),
)
def test_returns_indexed_agent_when_in_range(self):
agents = [self._agent("a-1"), self._agent("b-2")]
result = dashboard._selected_agent(dashboard.PANE_AGENTS, agents, 1)
self.assertIsNotNone(result)
assert result is not None # for type checker
self.assertEqual("b-2", result.slug)
def test_none_when_index_out_of_range(self):
agents = [self._agent("only")]
self.assertIsNone(
dashboard._selected_agent(dashboard.PANE_AGENTS, agents, 99),
)
class TestBottleForSlug(unittest.TestCase):
"""Re-attach target resolution (PRD 0020 chunk 3). Dashboard-
owned bottles return the stored handle as-is; non-owned bottles
get a synthesized DockerBottle backed by the slug-derived
container name."""
def test_owned_bottle_returns_held_handle(self):
sentinel = object()
bottles = {"dev-abc": (None, sentinel, "dev-abc")}
bottle, _ = dashboard._bottle_for_slug("dev-abc", bottles, None)
self.assertIs(sentinel, bottle)
def test_unowned_synthesizes_docker_bottle(self):
bottle, _ = dashboard._bottle_for_slug("dev-xyz", {}, None)
# The synth wraps the slug-derived container name.
self.assertEqual("bot-bottle-dev-xyz", bottle.name)
def test_unowned_without_manifest_omits_prompt_path(self):
bottle, hint = dashboard._bottle_for_slug("dev-xyz", {}, None)
self.assertEqual("", hint)
class TestPickNextAfterStop(unittest.TestCase):
"""After `x` stops a bottle, the dashboard slides focus to
the next agent the one filling the stopped row, or the
new last row if the stopped was last. Pure helper, easy
to unit-test."""
def _agent(self, slug: str) -> dashboard.ActiveAgent:
return dashboard.ActiveAgent(
backend_name="docker",
slug=slug, agent_name=slug, started_at="", services=(),
)
def test_empty_list_returns_none(self):
self.assertIsNone(
dashboard._pick_next_after_stop([], 0, "anything"),
)
def test_only_agent_being_stopped_returns_none(self):
# Stopping the last agent → nothing to focus.
agents = [self._agent("only")]
self.assertIsNone(
dashboard._pick_next_after_stop(agents, 0, "only"),
)
def test_middle_row_slides_up_to_same_index(self):
agents = [self._agent("a"), self._agent("b"), self._agent("c")]
# Cursor was on "b" at index 1; stopping "b" → "c" now sits
# at index 1 and takes focus.
out = dashboard._pick_next_after_stop(agents, 1, "b")
self.assertEqual((1, self._agent("c")), out)
def test_last_row_wraps_to_new_last(self):
agents = [self._agent("a"), self._agent("b"), self._agent("c")]
# Cursor on "c" at index 2; stopping "c" leaves a 2-agent
# list — index 2 is out of bounds, fall back to new last (1).
out = dashboard._pick_next_after_stop(agents, 2, "c")
self.assertEqual((1, self._agent("b")), out)
def test_first_row(self):
agents = [self._agent("a"), self._agent("b")]
out = dashboard._pick_next_after_stop(agents, 0, "a")
self.assertEqual((0, self._agent("b")), out)
def test_clamps_negative_selection(self):
# Defensive: a stale negative index doesn't crash.
agents = [self._agent("a"), self._agent("b")]
out = dashboard._pick_next_after_stop(agents, -1, "a")
self.assertEqual((0, self._agent("b")), out)
class TestTmuxPaneArgvBuilders(unittest.TestCase):
"""Pure argv builders for the tmux split-pane integration
(PRD 0021 chunk 2). The subprocess invocation itself is
environment-dependent; here we lock the wrapping shape so
a regression surfaces in CI without needing a real tmux."""
DOCKER_ARGV = [
"docker", "exec", "-it",
"bot-bottle-dev-abc",
"claude", "--dangerously-skip-permissions", "--continue",
]
def test_split_pane_argv_horizontal_with_pane_id_capture(self):
argv = dashboard._build_split_pane_argv(self.DOCKER_ARGV)
self.assertEqual(
["tmux", "split-window", "-h",
"-P", "-F", "#{pane_id}",
*self.DOCKER_ARGV],
argv,
)
def test_respawn_pane_argv_kills_existing_process(self):
argv = dashboard._build_respawn_pane_argv("%12", self.DOCKER_ARGV)
self.assertEqual(
["tmux", "respawn-pane", "-k", "-t", "%12", *self.DOCKER_ARGV],
argv,
)
def test_respawn_pane_argv_threads_pane_id_unmodified(self):
# Pane ids contain `%`; make sure we pass them straight
# through to `-t` without quoting or substitution surprises.
argv = dashboard._build_respawn_pane_argv("%abc.123", ["sh"])
self.assertIn("%abc.123", argv)
class TestResumeArgvWithFallback(unittest.TestCase):
"""The `claude --continue || claude` shell fallback for the
tmux re-attach path. Without it, an agent that's been spun
up but never typed at crashes the pane on Enter because
--continue has no session to resume."""
def _bottle(self, prompt_path: str | None = None):
from bot_bottle.backend.docker.bottle import DockerBottle
return DockerBottle(
container="bot-bottle-dev-abc",
teardown=lambda: None,
prompt_path_in_container=prompt_path,
)
def test_wraps_in_sh_c_with_or_fallback(self):
argv = dashboard._build_resume_argv_with_fallback(self._bottle())
# Must end with `sh -c '<cmd> --continue || <cmd>'`.
self.assertEqual(
["docker", "exec", "-it", "bot-bottle-dev-abc", "sh", "-c"],
argv[:6],
)
inner = argv[6]
self.assertIn("--continue", inner)
self.assertIn("||", inner)
# Both branches mention claude.
self.assertEqual(2, inner.count("claude"))
def test_inner_args_quoted_safely(self):
# Paths with spaces would break naive concatenation.
bottle = self._bottle("/home/with space/.prompt")
argv = dashboard._build_resume_argv_with_fallback(bottle)
inner = argv[-1]
# shlex.quote should single-quote any token with a space.
self.assertIn("'/home/with space/.prompt'", inner)
def test_includes_skip_permissions(self):
argv = dashboard._build_resume_argv_with_fallback(self._bottle())
self.assertIn("--dangerously-skip-permissions", argv[-1])
def test_includes_prompt_file_flag_when_set(self):
bottle = self._bottle("/home/node/.bot-bottle-prompt.txt")
argv = dashboard._build_resume_argv_with_fallback(bottle)
self.assertIn("--append-system-prompt-file", argv[-1])
self.assertIn("/home/node/.bot-bottle-prompt.txt", argv[-1])
class TestClaudeRuntimeArgs(unittest.TestCase):
"""The argv passed to `bottle.agent_argv` on each
attach. Locked here so the tmux + foreground paths build
identical agent invocations."""
def test_default_skip_permissions_only(self):
self.assertEqual(
["--dangerously-skip-permissions"],
dashboard._agent_runtime_args(resume=False),
)
def test_resume_appends_continue(self):
self.assertEqual(
["--dangerously-skip-permissions", "--continue"],
dashboard._agent_runtime_args(resume=True),
)
def test_remote_control(self):
args = dashboard._agent_runtime_args(
resume=False, remote_control=True,
)
self.assertIn("--remote-control", args)
class TestStopBottleFlow(unittest.TestCase):
"""Explicit per-bottle stop (PRD 0020 chunk 4). The non-owned
path is the one safe to test without curses + docker the
owned path drives `cm.__exit__` against a real launch context
and belongs in integration tests."""
def test_non_owned_returns_cleanup_hint(self):
# stdscr is None here on purpose — the non-owned branch
# returns before any curses call.
msg = dashboard._stop_bottle_flow(
stdscr=None, # type: ignore[arg-type]
bottles={},
slug="ghost-zzz",
)
self.assertIn("not dashboard-owned", msg)
self.assertIn("./cli.py cleanup", msg)
def test_non_owned_does_not_touch_tmux_state(self):
# PRD 0021: a stop on an unknown slug should never clear
# the right-pane occupant tracking, even if the slugs
# happen to match (defensive — non-owned can't be in the
# right pane via the dashboard's normal flow anyway).
tmux_state = {"pane_id": "%5", "slug": "live-bbb"}
dashboard._stop_bottle_flow(
stdscr=None, # type: ignore[arg-type]
bottles={},
slug="ghost-zzz",
tmux_state=tmux_state,
)
self.assertEqual({"pane_id": "%5", "slug": "live-bbb"}, tmux_state)
class TestOperatorEditFlowGuards(_FakeHomeMixin, unittest.TestCase):
"""Chunk-4 contract: the edit flow refuses when the selected
agent doesn't have the required sidecar running. The discover-
and-prompt scaffolding is gone, so the gating happens here
instead of in the key handler."""
def setUp(self) -> None:
self._setup_fake_home()
def tearDown(self) -> None:
self._teardown_fake_home()
def _agent(self, services: tuple[str, ...]) -> dashboard.ActiveAgent:
return dashboard.ActiveAgent(
backend_name="docker",
slug="dev-abc12",
agent_name="impl",
started_at="",
services=services,
)
def test_routes_edit_refuses_without_egress(self):
# Bottle without bottle.egress.routes → no egress sidecar.
msg = dashboard._operator_edit_flow(
stdscr=None, # type: ignore[arg-type]
agent=self._agent(("pipelock", "supervise")),
required_service="egress",
label="routes",
fetch=lambda _: "x",
apply=lambda _slug, _content: None,
suffix=".yaml",
)
self.assertIn("no running egress sidecar", msg)
self.assertIn("dev-abc12", msg)
def test_pipelock_edit_refuses_when_pipelock_missing(self):
# Belt-and-braces — pipelock should always be there, but
# the race window between `compose up` and `docker ps`
# update is real.
msg = dashboard._operator_edit_flow(
stdscr=None, # type: ignore[arg-type]
agent=self._agent(()),
required_service="pipelock",
label="pipelock",
fetch=lambda _: "x",
apply=lambda _slug, _content: None,
suffix=".txt",
)
self.assertIn("no running pipelock sidecar", msg)
if __name__ == "__main__":
unittest.main()
-39
View File
@@ -1,39 +0,0 @@
"""Unit: dashboard's new-proposal highlight window.
The curses rendering itself is exercised manually; this isolates
the pure decision `is the proposal still in its post-arrival
highlight window?`"""
import unittest
from bot_bottle.cli import dashboard
class TestIsRecent(unittest.TestCase):
def test_just_seen_is_recent(self):
self.assertTrue(dashboard._is_recent("p1", {"p1": 100.0}, now=100.5))
def test_seen_within_window(self):
# Default window is 5s.
self.assertTrue(
dashboard._is_recent("p1", {"p1": 100.0}, now=104.9),
)
def test_seen_past_window_is_not_recent(self):
self.assertFalse(
dashboard._is_recent("p1", {"p1": 100.0}, now=106.0),
)
def test_unknown_proposal_is_not_recent(self):
self.assertFalse(
dashboard._is_recent("p2", {"p1": 100.0}, now=100.5),
)
def test_none_args_safe_default(self):
self.assertFalse(dashboard._is_recent("p1", None, None))
self.assertFalse(dashboard._is_recent("p1", {"p1": 100.0}, None))
self.assertFalse(dashboard._is_recent("p1", None, 100.5))
if __name__ == "__main__":
unittest.main()
+29
View File
@@ -0,0 +1,29 @@
"""Unit: deploy_key_provisioner factory (PRD 0048)."""
from __future__ import annotations
import unittest
from unittest.mock import patch
from bot_bottle.deploy_key_provisioner import DeployKeyProvisioner, get_provisioner
from bot_bottle.manifest import ManifestError
class TestGetProvisioner(unittest.TestCase):
def test_gitea_returns_gitea_provisioner(self):
from bot_bottle.contrib.gitea.deploy_key_provisioner import (
GiteaDeployKeyProvisioner,
)
p = get_provisioner("gitea", token="tok", api_url="https://gitea.example.com")
self.assertIsInstance(p, GiteaDeployKeyProvisioner)
self.assertIsInstance(p, DeployKeyProvisioner)
def test_unknown_provider_raises_manifest_error(self):
with self.assertRaises(ManifestError) as ctx:
get_provisioner("github", token="tok", api_url="https://github.com")
self.assertIn("github", str(ctx.exception))
self.assertIn("provisioned_key provider", str(ctx.exception))
if __name__ == "__main__":
unittest.main()
+98
View File
@@ -282,5 +282,103 @@ class TestPrepare(unittest.TestCase):
self.assertIn("exec git daemon", content)
class TestShellEscaping(unittest.TestCase):
"""Regression tests: all three render functions must produce syntactically
valid sh code even when names and upstream URLs contain shell-special
characters. Tests construct GitGateUpstream directly bypassing manifest
name validation so the rendering layer is exercised in isolation."""
_MALICIOUS_URL_CASES = [
("single_quote", "ssh://git@host/path'with'quotes.git"),
("double_quote", 'ssh://git@host/path"with"quotes.git'),
("space", "ssh://git@host/path with spaces.git"),
("semicolon", "ssh://git@host/path;evil.git"),
("newline", "ssh://git@host/path\nwith\nnewlines.git"),
("backtick", "ssh://git@host/path`whoami`.git"),
]
_MALICIOUS_NAME_CASES = [
("single_quote", "repo'name"),
("double_quote", 'repo"name'),
("space", "repo name"),
("semicolon", "repo;name"),
("newline", "repo\nname"),
("backtick", "repo`name"),
]
def _make_upstream(self, url: str, name: str = "myrepo") -> GitGateUpstream:
return GitGateUpstream(
name=name,
upstream_url=url,
upstream_host="host",
upstream_port="22",
identity_file="/key",
known_host_key="",
)
def _assert_valid_sh(self, script: str, label: str = "") -> None:
import subprocess
fd, path = tempfile.mkstemp(suffix=".sh")
try:
with os.fdopen(fd, "w") as f:
f.write(script)
result = subprocess.run(
["sh", "-n", path], capture_output=True, text=True,
)
self.assertEqual(
0, result.returncode,
f"sh -n failed{(' for ' + label) if label else ''}: {result.stderr}",
)
finally:
os.unlink(path)
def test_hook_renders_valid_sh(self):
self._assert_valid_sh(git_gate_render_hook(), "pre-receive hook")
def test_access_hook_renders_valid_sh(self):
self._assert_valid_sh(git_gate_render_access_hook(), "access hook")
def test_entrypoint_with_pathological_upstream_url_renders_valid_sh(self):
for label, url in self._MALICIOUS_URL_CASES:
with self.subTest(char=label):
script = git_gate_render_entrypoint((self._make_upstream(url),))
self._assert_valid_sh(script, label)
def test_entrypoint_upstream_url_value_preserved_after_quoting(self):
import shlex as _shlex
for label, url in self._MALICIOUS_URL_CASES:
with self.subTest(char=label):
script = git_gate_render_entrypoint((self._make_upstream(url),))
# The quoted form of the URL must appear verbatim in the script so
# the shell reconstructs exactly the original value at runtime.
expected = f"init_repo {_shlex.quote('myrepo')} {_shlex.quote(url)}"
self.assertIn(
expected, script,
f"{label}: expected quoted form not found in script",
)
def test_entrypoint_with_pathological_name_renders_valid_sh(self):
for label, name in self._MALICIOUS_NAME_CASES:
with self.subTest(char=label):
script = git_gate_render_entrypoint((
self._make_upstream("ssh://git@github.com/foo/bar.git", name=name),
))
self._assert_valid_sh(script, label)
def test_entrypoint_name_value_preserved_after_quoting(self):
import shlex as _shlex
url = "ssh://git@github.com/foo/bar.git"
for label, name in self._MALICIOUS_NAME_CASES:
with self.subTest(char=label):
script = git_gate_render_entrypoint((
self._make_upstream(url, name=name),
))
expected = f"init_repo {_shlex.quote(name)} {_shlex.quote(url)}"
self.assertIn(
expected, script,
f"{label}: expected quoted form not found in script",
)
if __name__ == "__main__":
unittest.main()
+107
View File
@@ -243,6 +243,113 @@ class TestGitEntryCrossValidation(unittest.TestCase):
self.assertIn("PRD 0047", msg)
class TestProvisionedKey(unittest.TestCase):
"""git-gate.repos entries that use provisioned_key (PRD 0048)."""
def test_provisioned_key_minimal(self):
m = Manifest.from_json_obj(_manifest({
"bot-bottle": {
"url": "ssh://git@gitea.dideric.is:30009/didericis/bot-bottle.git",
"provisioned_key": {
"provider": "gitea",
"token_env": "GITEA_TOKEN",
},
},
}))
e = m.bottles["dev"].git[0]
self.assertEqual("bot-bottle", e.Name)
self.assertIsNotNone(e.ProvisionedKey)
assert e.ProvisionedKey is not None
self.assertEqual("gitea", e.ProvisionedKey.provider)
self.assertEqual("GITEA_TOKEN", e.ProvisionedKey.token_env)
self.assertEqual("", e.ProvisionedKey.api_url)
self.assertEqual("", e.IdentityFile)
def test_provisioned_key_with_api_url(self):
m = Manifest.from_json_obj(_manifest({
"repo": {
"url": "ssh://git@gitea.example.com/org/repo.git",
"provisioned_key": {
"provider": "gitea",
"token_env": "MY_TOKEN",
"api_url": "https://gitea.example.com",
},
},
}))
pk = m.bottles["dev"].git[0].ProvisionedKey
assert pk is not None
self.assertEqual("https://gitea.example.com", pk.api_url)
def test_both_identity_and_provisioned_key_dies(self):
with self.assertRaises(ManifestError) as ctx:
Manifest.from_json_obj(_manifest({
"foo": {
"url": "ssh://git@github.com/foo.git",
"identity": "/dev/null",
"provisioned_key": {"provider": "gitea", "token_env": "T"},
},
}))
self.assertIn("exactly one of", str(ctx.exception))
self.assertIn("got both", str(ctx.exception))
def test_neither_identity_nor_provisioned_key_dies(self):
with self.assertRaises(ManifestError) as ctx:
Manifest.from_json_obj(_manifest({
"foo": {"url": "ssh://git@github.com/foo.git"},
}))
self.assertIn("exactly one of", str(ctx.exception))
self.assertIn("got neither", str(ctx.exception))
def test_unknown_key_in_provisioned_key_block_dies(self):
with self.assertRaises(ManifestError):
Manifest.from_json_obj(_manifest({
"foo": {
"url": "ssh://git@github.com/foo.git",
"provisioned_key": {
"provider": "gitea",
"token_env": "T",
"key_type": "rsa", # not allowed
},
},
}))
def test_missing_provider_dies(self):
with self.assertRaises(ManifestError):
Manifest.from_json_obj(_manifest({
"foo": {
"url": "ssh://git@github.com/foo.git",
"provisioned_key": {"token_env": "T"},
},
}))
def test_missing_token_env_dies(self):
with self.assertRaises(ManifestError):
Manifest.from_json_obj(_manifest({
"foo": {
"url": "ssh://git@github.com/foo.git",
"provisioned_key": {"provider": "gitea"},
},
}))
def test_provisioned_key_entry_has_no_identity_file(self):
m = Manifest.from_json_obj(_manifest({
"foo": {
"url": "ssh://git@github.com/didericis/foo.git",
"provisioned_key": {"provider": "gitea", "token_env": "T"},
},
}))
self.assertEqual("", m.bottles["dev"].git[0].IdentityFile)
def test_identity_entry_has_no_provisioned_key(self):
m = Manifest.from_json_obj(_manifest({
"foo": {
"url": "ssh://git@github.com/foo.git",
"identity": "/dev/null",
},
}))
self.assertIsNone(m.bottles["dev"].git[0].ProvisionedKey)
class TestEmptyGitGateField(unittest.TestCase):
def test_no_git_gate_field_yields_empty_tuple(self):
m = Manifest.from_json_obj({
@@ -1,12 +1,12 @@
"""Unit: dashboard headless paths (PRD 0013 phase 4, PRD 0014).
"""Unit: supervise 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.
add_route is stubbed at the supervise CLI module level so the tests
don't need a running egress sidecar; the real docker exec/cp/SIGHUP
plumbing is covered by the integration test.
"""
import os
@@ -19,7 +19,7 @@ from bot_bottle import supervise
from bot_bottle.backend.docker.capability_apply import CapabilityApplyError
from bot_bottle.backend.docker.egress_apply import EgressApplyError
from bot_bottle.backend.docker.pipelock_apply import PipelockApplyError
from bot_bottle.cli import dashboard
from bot_bottle.cli import supervise as supervise_cli
from bot_bottle.supervise import (
Proposal,
STATUS_APPROVED,
@@ -61,7 +61,7 @@ class _FakeHomeMixin:
"""Patch supervise.bot_bottle_root to a temp dir for the test."""
def _setup_fake_home(self):
self._tmp = tempfile.TemporaryDirectory(prefix="dashboard-test.")
self._tmp = tempfile.TemporaryDirectory(prefix="supervise-test.")
original = supervise.bot_bottle_root
def fake_root() -> Path:
@@ -83,14 +83,14 @@ class TestDiscoverPending(_FakeHomeMixin, unittest.TestCase):
self._teardown_fake_home()
def test_empty_when_no_queues(self):
self.assertEqual([], dashboard.discover_pending())
self.assertEqual([], supervise_cli.discover_pending())
def test_walks_all_slug_subdirs(self):
for slug in ("dev", "api"):
qdir = supervise.queue_dir_for_slug(slug)
qdir.mkdir(parents=True)
supervise.write_proposal(qdir, _proposal(slug=slug))
pending = dashboard.discover_pending()
pending = supervise_cli.discover_pending()
self.assertEqual({"dev", "api"}, {qp.proposal.bottle_slug for qp in pending})
def test_sorted_by_arrival_across_bottles(self):
@@ -110,7 +110,7 @@ class TestDiscoverPending(_FakeHomeMixin, unittest.TestCase):
qdir = supervise.queue_dir_for_slug(p.bottle_slug)
qdir.mkdir(parents=True, exist_ok=True)
supervise.write_proposal(qdir, p)
pending = dashboard.discover_pending()
pending = supervise_cli.discover_pending()
self.assertEqual([early.id, late.id], [qp.proposal.id for qp in pending])
def test_excludes_already_responded(self):
@@ -121,34 +121,34 @@ class TestDiscoverPending(_FakeHomeMixin, unittest.TestCase):
supervise.write_response(qdir, supervise.Response(
proposal_id=p.id, status=STATUS_APPROVED, notes="",
))
self.assertEqual([], dashboard.discover_pending())
self.assertEqual([], supervise_cli.discover_pending())
class TestApproveReject(_FakeHomeMixin, unittest.TestCase):
def setUp(self):
self._setup_fake_home()
self._original_add_route = dashboard.add_route
self._original_apply_allowlist = dashboard.apply_allowlist_change
self._original_fetch_allowlist = dashboard.fetch_current_allowlist
self._original_apply_capability = dashboard.apply_capability_change
self._original_add_route = supervise_cli.add_route
self._original_apply_allowlist = supervise_cli.apply_allowlist_change
self._original_fetch_allowlist = supervise_cli.fetch_current_allowlist
self._original_apply_capability = supervise_cli.apply_capability_change
# Default stubs: succeed with deterministic before/after so the
# audit log shows a non-empty diff.
dashboard.add_route = lambda slug, content: (
supervise_cli.add_route = lambda slug, content: (
'{"routes": []}\n', '{"routes": [{"host": "x"}]}\n',
)
dashboard.apply_allowlist_change = lambda slug, content: (
supervise_cli.apply_allowlist_change = lambda slug, content: (
"old.example\n", content,
)
dashboard.fetch_current_allowlist = lambda slug: "old.example\n"
dashboard.apply_capability_change = lambda slug, content: (
supervise_cli.fetch_current_allowlist = lambda slug: "old.example\n"
supervise_cli.apply_capability_change = lambda slug, content: (
"FROM old\n", content,
)
def tearDown(self):
dashboard.add_route = self._original_add_route
dashboard.apply_allowlist_change = self._original_apply_allowlist
dashboard.fetch_current_allowlist = self._original_fetch_allowlist
dashboard.apply_capability_change = self._original_apply_capability
supervise_cli.add_route = self._original_add_route
supervise_cli.apply_allowlist_change = self._original_apply_allowlist
supervise_cli.fetch_current_allowlist = self._original_fetch_allowlist
supervise_cli.apply_capability_change = self._original_apply_capability
self._teardown_fake_home()
def _enqueue(self, tool: str = TOOL_EGRESS_BLOCK):
@@ -156,11 +156,11 @@ class TestApproveReject(_FakeHomeMixin, unittest.TestCase):
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)
return supervise_cli.QueuedProposal(proposal=p, queue_dir=qdir)
def test_approve_writes_response_and_audit(self):
qp = self._enqueue()
dashboard.approve(qp)
supervise_cli.approve(qp)
resp = read_response(qp.queue_dir, qp.proposal.id)
self.assertEqual(STATUS_APPROVED, resp.status)
self.assertIsNone(resp.final_file)
@@ -170,7 +170,7 @@ class TestApproveReject(_FakeHomeMixin, unittest.TestCase):
def test_approve_with_final_file_marks_modified(self):
qp = self._enqueue()
dashboard.approve(qp, final_file='{"routes": [{"path": "/x/"}]}\n', notes="tweaked")
supervise_cli.approve(qp, final_file='{"routes": [{"path": "/x/"}]}\n', notes="tweaked")
resp = read_response(qp.queue_dir, qp.proposal.id)
self.assertEqual(STATUS_MODIFIED, resp.status)
self.assertEqual('{"routes": [{"path": "/x/"}]}\n', resp.final_file)
@@ -180,7 +180,7 @@ class TestApproveReject(_FakeHomeMixin, unittest.TestCase):
def test_reject_writes_rejection(self):
qp = self._enqueue()
dashboard.reject(qp, reason="nope")
supervise_cli.reject(qp, reason="nope")
resp = read_response(qp.queue_dir, qp.proposal.id)
self.assertEqual(STATUS_REJECTED, resp.status)
self.assertEqual("nope", resp.notes)
@@ -190,7 +190,7 @@ class TestApproveReject(_FakeHomeMixin, unittest.TestCase):
def test_capability_block_skips_audit_log(self):
qp = self._enqueue(tool=TOOL_CAPABILITY_BLOCK)
dashboard.approve(qp)
supervise_cli.approve(qp)
# No audit log for capability-block (per PRD 0013 / 0016).
# cred-proxy and pipelock logs both empty.
self.assertEqual([], read_audit_entries("egress", "dev"))
@@ -198,7 +198,7 @@ class TestApproveReject(_FakeHomeMixin, unittest.TestCase):
def test_pipelock_audit_distinct_from_egress(self):
qp = self._enqueue(tool=TOOL_PIPELOCK_BLOCK)
dashboard.approve(qp)
supervise_cli.approve(qp)
self.assertEqual(1, len(read_audit_entries("pipelock", "dev")))
self.assertEqual(0, len(read_audit_entries("egress", "dev")))
@@ -210,10 +210,10 @@ class TestEgressApplyWiring(_FakeHomeMixin, unittest.TestCase):
def setUp(self):
self._setup_fake_home()
self._original_add_route = dashboard.add_route
self._original_add_route = supervise_cli.add_route
def tearDown(self):
dashboard.add_route = self._original_add_route
supervise_cli.add_route = self._original_add_route
self._teardown_fake_home()
def _enqueue_egress(self, proposed: str = '{"host": "x.example"}\n'):
@@ -227,17 +227,17 @@ class TestEgressApplyWiring(_FakeHomeMixin, unittest.TestCase):
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)
return supervise_cli.QueuedProposal(proposal=p, queue_dir=qdir)
def test_egress_block_calls_add_route_with_proposed_json(self):
calls = []
dashboard.add_route = lambda slug, content: (
supervise_cli.add_route = lambda slug, content: (
calls.append((slug, content)) or ("before", "after")
)
qp = self._enqueue_egress(
proposed='{"host": "new.example", "path_allowlist": ["/x/"]}\n'
)
dashboard.approve(qp)
supervise_cli.approve(qp)
self.assertEqual(1, len(calls))
slug, content = calls[0]
self.assertEqual("dev", slug)
@@ -250,11 +250,11 @@ class TestEgressApplyWiring(_FakeHomeMixin, unittest.TestCase):
def test_modify_passes_final_file_to_add_route(self):
calls = []
dashboard.add_route = lambda slug, content: (
supervise_cli.add_route = lambda slug, content: (
calls.append(content) or ("before", "after")
)
qp = self._enqueue_egress()
dashboard.approve(
supervise_cli.approve(
qp,
final_file='{"host": "edited.example"}\n',
notes="tweaked",
@@ -262,12 +262,12 @@ class TestEgressApplyWiring(_FakeHomeMixin, unittest.TestCase):
self.assertEqual(['{"host": "edited.example"}\n'], calls)
def test_apply_failure_blocks_response_and_audit(self):
dashboard.add_route = lambda slug, content: (_ for _ in ()).throw(
supervise_cli.add_route = lambda slug, content: (_ for _ in ()).throw(
EgressApplyError("docker exec failed")
)
qp = self._enqueue_egress()
with self.assertRaises(EgressApplyError):
dashboard.approve(qp)
supervise_cli.approve(qp)
# No response file (proposal stays pending).
self.assertEqual(
[qp.proposal.id],
@@ -277,25 +277,20 @@ class TestEgressApplyWiring(_FakeHomeMixin, unittest.TestCase):
self.assertEqual([], read_audit_entries("egress", "dev"))
def test_real_diff_lands_in_audit(self):
dashboard.add_route = lambda slug, content: (
supervise_cli.add_route = lambda slug, content: (
'{"routes": []}\n', # before
'{"routes": [{"host": "new.example"}]}\n', # after
)
qp = self._enqueue_egress(proposed='{"host": "new.example"}\n')
dashboard.approve(qp)
supervise_cli.approve(qp)
entries = read_audit_entries("egress", "dev")
self.assertEqual(1, len(entries))
self.assertIn('+{"routes": [{"host": "new.example"}]}', 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_egress()
dashboard.reject(qp, reason="no thanks")
self.assertEqual([], called)
supervise_cli.reject(qp, reason="no thanks")
# 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)
@@ -306,18 +301,18 @@ class TestEgressApplyWiring(_FakeHomeMixin, unittest.TestCase):
class TestPipelockApplyWiring(_FakeHomeMixin, unittest.TestCase):
"""PRD 0015 Phase 2 + PR #25 follow-up: approve() on a
pipelock-block proposal carries the failed URL; the dashboard
pipelock-block proposal carries the failed URL; the supervise TUI
extracts the host, merges it into the running allowlist, and
calls apply_allowlist_change with the merged content."""
def setUp(self):
self._setup_fake_home()
self._original_apply = dashboard.apply_allowlist_change
self._original_fetch = dashboard.fetch_current_allowlist
self._original_apply = supervise_cli.apply_allowlist_change
self._original_fetch = supervise_cli.fetch_current_allowlist
def tearDown(self):
dashboard.apply_allowlist_change = self._original_apply
dashboard.fetch_current_allowlist = self._original_fetch
supervise_cli.apply_allowlist_change = self._original_apply
supervise_cli.fetch_current_allowlist = self._original_fetch
self._teardown_fake_home()
def _enqueue_pipelock(self, failed_url: str = "https://api.github.com/repos/foo/bar"):
@@ -331,17 +326,17 @@ class TestPipelockApplyWiring(_FakeHomeMixin, unittest.TestCase):
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)
return supervise_cli.QueuedProposal(proposal=p, queue_dir=qdir)
def test_url_host_merged_into_current_allowlist(self):
dashboard.fetch_current_allowlist = lambda slug: "existing.example\n"
supervise_cli.fetch_current_allowlist = lambda slug: "existing.example\n"
applied = []
dashboard.apply_allowlist_change = lambda slug, content: (
supervise_cli.apply_allowlist_change = lambda slug, content: (
applied.append((slug, content))
or ("existing.example\n", content)
)
qp = self._enqueue_pipelock("https://api.github.com/repos/foo/bar")
dashboard.approve(qp)
supervise_cli.approve(qp)
# apply_allowlist_change was called with the merged content:
# existing host + the URL's host (no path, since pipelock is
# hostname-only).
@@ -353,27 +348,27 @@ class TestPipelockApplyWiring(_FakeHomeMixin, unittest.TestCase):
self.assertNotIn("/repos/foo/bar", content) # path stripped
def test_host_already_in_allowlist_is_idempotent(self):
dashboard.fetch_current_allowlist = lambda slug: "api.github.com\n"
supervise_cli.fetch_current_allowlist = lambda slug: "api.github.com\n"
applied = []
dashboard.apply_allowlist_change = lambda slug, content: (
supervise_cli.apply_allowlist_change = lambda slug, content: (
applied.append(content)
or ("api.github.com\n", content)
)
qp = self._enqueue_pipelock("https://api.github.com/some/path")
dashboard.approve(qp)
supervise_cli.approve(qp)
# Still applied, but the content is unchanged from current —
# before/after diff is empty.
self.assertEqual(1, len(applied))
self.assertEqual("api.github.com\n", applied[0])
def test_apply_failure_blocks_response_and_audit(self):
dashboard.fetch_current_allowlist = lambda slug: "existing.example\n"
dashboard.apply_allowlist_change = lambda slug, content: (_ for _ in ()).throw(
supervise_cli.fetch_current_allowlist = lambda slug: "existing.example\n"
supervise_cli.apply_allowlist_change = lambda slug, content: (_ for _ in ()).throw(
PipelockApplyError("docker exec failed")
)
qp = self._enqueue_pipelock()
with self.assertRaises(PipelockApplyError):
dashboard.approve(qp)
supervise_cli.approve(qp)
self.assertEqual(
[qp.proposal.id],
[p.id for p in supervise.list_pending_proposals(qp.queue_dir)],
@@ -381,12 +376,12 @@ class TestPipelockApplyWiring(_FakeHomeMixin, unittest.TestCase):
self.assertEqual([], read_audit_entries("pipelock", "dev"))
def test_url_without_host_raises(self):
dashboard.fetch_current_allowlist = lambda slug: ""
supervise_cli.fetch_current_allowlist = lambda slug: ""
# supervise_server's validator would catch this; if a broken
# URL ever makes it through, the dashboard surfaces it too.
# URL ever makes it through, the supervise TUI surfaces it too.
qp = self._enqueue_pipelock("https:///nohost")
with self.assertRaises(PipelockApplyError):
dashboard.approve(qp)
supervise_cli.approve(qp)
class TestCapabilityApplyWiring(_FakeHomeMixin, unittest.TestCase):
@@ -397,10 +392,10 @@ class TestCapabilityApplyWiring(_FakeHomeMixin, unittest.TestCase):
def setUp(self):
self._setup_fake_home()
self._original = dashboard.apply_capability_change
self._original = supervise_cli.apply_capability_change
def tearDown(self):
dashboard.apply_capability_change = self._original
supervise_cli.apply_capability_change = self._original
self._teardown_fake_home()
def _enqueue_capability(self, proposed: str = "FROM python:3.13\nRUN apk add ripgrep\n"):
@@ -414,112 +409,50 @@ class TestCapabilityApplyWiring(_FakeHomeMixin, unittest.TestCase):
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)
return supervise_cli.QueuedProposal(proposal=p, queue_dir=qdir)
def test_capability_block_calls_apply_with_proposed_file(self):
calls = []
dashboard.apply_capability_change = lambda slug, content: (
supervise_cli.apply_capability_change = lambda slug, content: (
calls.append((slug, content)) or ("FROM old\n", content)
)
qp = self._enqueue_capability("FROM bookworm\n")
dashboard.approve(qp)
supervise_cli.approve(qp)
self.assertEqual([("dev", "FROM bookworm\n")], calls)
def test_apply_failure_blocks_response_and_keeps_pending(self):
dashboard.apply_capability_change = lambda slug, content: (_ for _ in ()).throw(
supervise_cli.apply_capability_change = lambda slug, content: (_ for _ in ()).throw(
CapabilityApplyError("teardown failed")
)
qp = self._enqueue_capability()
with self.assertRaises(CapabilityApplyError):
dashboard.approve(qp)
supervise_cli.approve(qp)
self.assertEqual(
[qp.proposal.id],
[p.id for p in supervise.list_pending_proposals(qp.queue_dir)],
)
def test_no_audit_log_for_capability(self):
dashboard.apply_capability_change = lambda slug, content: ("FROM old\n", content)
supervise_cli.apply_capability_change = lambda slug, content: ("FROM old\n", content)
qp = self._enqueue_capability()
dashboard.approve(qp)
supervise_cli.approve(qp)
# capability-block has no audit log per PRD 0013 — its record
# lives in the per-bottle Dockerfile + transcript state.
self.assertEqual([], read_audit_entries("egress", "dev"))
self.assertEqual([], read_audit_entries("pipelock", "dev"))
def test_proposal_archived_after_apply(self):
dashboard.apply_capability_change = lambda slug, content: ("FROM old\n", content)
supervise_cli.apply_capability_change = lambda slug, content: ("FROM old\n", content)
qp = self._enqueue_capability()
dashboard.approve(qp)
supervise_cli.approve(qp)
# Sidecar would normally archive after delivering the response,
# but it's gone by then. The dashboard archives so
# but it's gone by then. The supervise TUI archives so
# discover_pending stops surfacing the resolved proposal.
self.assertEqual([], supervise.list_pending_proposals(qp.queue_dir))
processed = list((qp.queue_dir / "processed").glob("*.json"))
self.assertEqual(2, len(processed))
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("egress", "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(
EgressApplyError("nope")
)
with self.assertRaises(EgressApplyError):
dashboard.operator_edit_routes("dev", '{"routes": []}\n')
self.assertEqual([], read_audit_entries("egress", "dev"))
class TestOperatorEditAllowlist(_FakeHomeMixin, unittest.TestCase):
"""PRD 0015 Phase 3: operator-initiated pipelock allowlist edit."""
def setUp(self):
self._setup_fake_home()
self._original = dashboard.apply_allowlist_change
def tearDown(self):
dashboard.apply_allowlist_change = self._original
self._teardown_fake_home()
def test_writes_audit_with_operator_edit_action(self):
dashboard.apply_allowlist_change = lambda slug, content: (
"old.example\n", content,
)
dashboard.operator_edit_allowlist("dev", "old.example\nnew.example\n")
entries = read_audit_entries("pipelock", "dev")
self.assertEqual(1, len(entries))
self.assertEqual(supervise.ACTION_OPERATOR_EDIT, entries[0].operator_action)
self.assertIn("+new.example", entries[0].diff)
def test_failure_does_not_write_audit(self):
dashboard.apply_allowlist_change = lambda slug, content: (_ for _ in ()).throw(
PipelockApplyError("nope")
)
with self.assertRaises(PipelockApplyError):
dashboard.operator_edit_allowlist("dev", "x.example\n")
self.assertEqual([], read_audit_entries("pipelock", "dev"))
class TestEditInEditor(unittest.TestCase):
def test_runs_editor_returns_edited_content(self):
# Fake "editor" is /bin/sh -c 'cat <<EOF > $1 ... EOF'
@@ -544,7 +477,7 @@ class TestEditInEditor(unittest.TestCase):
os.chmod(editor_script, 0o755)
os.environ["EDITOR"] = editor_script
try:
result = dashboard.edit_in_editor("original")
result = supervise_cli.edit_in_editor("original")
self.assertEqual("edited", result)
finally:
os.unlink(editor_script)
@@ -566,7 +499,7 @@ class TestEditInEditor(unittest.TestCase):
os.chmod(editor_script, 0o755)
os.environ["EDITOR"] = editor_script
try:
result = dashboard.edit_in_editor("original")
result = supervise_cli.edit_in_editor("original")
self.assertIsNone(result)
finally:
os.unlink(editor_script)
@@ -583,19 +516,19 @@ class TestCapabilityBlockSmolmachinesGuard(_FakeHomeMixin, unittest.TestCase):
def setUp(self):
self._setup_fake_home()
self._original_apply_capability = dashboard.apply_capability_change
dashboard.apply_capability_change = lambda slug, content: ("", content)
self._original_apply_capability = supervise_cli.apply_capability_change
supervise_cli.apply_capability_change = lambda slug, content: ("", content)
def tearDown(self):
dashboard.apply_capability_change = self._original_apply_capability
supervise_cli.apply_capability_change = self._original_apply_capability
self._teardown_fake_home()
def _enqueue_capability(self, slug: str = "dev") -> "dashboard.QueuedProposal":
def _enqueue_capability(self, slug: str = "dev") -> "supervise_cli.QueuedProposal":
p = _proposal(slug=slug, tool=TOOL_CAPABILITY_BLOCK)
qdir = supervise.queue_dir_for_slug(slug)
qdir.mkdir(parents=True, exist_ok=True)
supervise.write_proposal(qdir, p)
return dashboard.QueuedProposal(proposal=p, queue_dir=qdir)
return supervise_cli.QueuedProposal(proposal=p, queue_dir=qdir)
def _write_metadata(self, slug: str, compose_project: str) -> None:
from bot_bottle.backend.docker.bottle_state import BottleMetadata, write_metadata
@@ -612,18 +545,18 @@ class TestCapabilityBlockSmolmachinesGuard(_FakeHomeMixin, unittest.TestCase):
self._write_metadata("dev", compose_project="")
qp = self._enqueue_capability("dev")
with self.assertRaises(CapabilityApplyError) as ctx:
dashboard.approve(qp)
supervise_cli.approve(qp)
self.assertIn("smolmachines", str(ctx.exception))
def test_docker_bottle_calls_apply_capability_change(self):
self._write_metadata("dev", compose_project="bot-bottle-dev")
qp = self._enqueue_capability("dev")
dashboard.approve(qp) # must not raise
supervise_cli.approve(qp) # must not raise
def test_no_metadata_falls_through_to_docker_path(self):
# No metadata at all → assume Docker (backward-compatible).
qp = self._enqueue_capability("dev")
dashboard.approve(qp) # must not raise
supervise_cli.approve(qp) # must not raise
if __name__ == "__main__":
@@ -1,6 +1,6 @@
"""Unit: dashboard launch/crash failure logging (issue #100).
"""Unit: supervise launch/crash failure logging (issue #100).
The dashboard runs under curses, so anything written to stderr while the
The supervise TUI runs under curses, so anything written to stderr while the
TUI owns the terminal is wiped when the terminal is restored. These
tests lock the recovery paths: a config error (`Die`) is re-surfaced
after the wrapper returns, and an unexpected crash is persisted to a
@@ -17,7 +17,7 @@ from pathlib import Path
from unittest import mock
from bot_bottle import supervise
from bot_bottle.cli import dashboard
from bot_bottle.cli import supervise as supervise_cli
from bot_bottle.log import Die, die
@@ -44,7 +44,7 @@ class _FakeHomeMixin:
~/.bot-bottle."""
def _setup_fake_home(self):
self._tmp = tempfile.TemporaryDirectory(prefix="dash-crash-test.")
self._tmp = tempfile.TemporaryDirectory(prefix="supervise-crash-test.")
self._orig_root = supervise.bot_bottle_root
self._root = Path(self._tmp.name) / ".bot-bottle"
supervise.bot_bottle_root = lambda: self._root # type: ignore[assignment]
@@ -54,7 +54,7 @@ class _FakeHomeMixin:
self._tmp.cleanup()
class TestCmdDashboardErrorPaths(_FakeHomeMixin, unittest.TestCase):
class TestCmdSuperviseErrorPaths(_FakeHomeMixin, unittest.TestCase):
def setUp(self):
self._setup_fake_home()
@@ -63,42 +63,42 @@ class TestCmdDashboardErrorPaths(_FakeHomeMixin, unittest.TestCase):
def test_keyboard_interrupt_returns_130(self):
with mock.patch.object(
dashboard.curses, "wrapper", side_effect=KeyboardInterrupt
supervise_cli.curses, "wrapper", side_effect=KeyboardInterrupt
):
self.assertEqual(130, dashboard.cmd_dashboard([]))
self.assertEqual(130, supervise_cli.cmd_supervise([]))
def test_die_resurfaces_message_after_curses(self):
buf = io.StringIO()
with mock.patch.object(
dashboard.curses, "wrapper",
supervise_cli.curses, "wrapper",
side_effect=Die(1, "manifest parse error at line 3"),
):
with contextlib.redirect_stderr(buf):
rc = dashboard.cmd_dashboard([])
rc = supervise_cli.cmd_supervise([])
self.assertEqual(1, rc)
self.assertIn("manifest parse error at line 3", buf.getvalue())
def test_die_without_message_has_fallback(self):
buf = io.StringIO()
with mock.patch.object(dashboard.curses, "wrapper", side_effect=Die(1)):
with mock.patch.object(supervise_cli.curses, "wrapper", side_effect=Die(1)):
with contextlib.redirect_stderr(buf):
rc = dashboard.cmd_dashboard([])
rc = supervise_cli.cmd_supervise([])
self.assertEqual(1, rc)
self.assertIn("fatal error", buf.getvalue())
def test_unexpected_exception_writes_crash_log(self):
buf = io.StringIO()
with mock.patch.object(
dashboard.curses, "wrapper",
supervise_cli.curses, "wrapper",
side_effect=ValueError("kaboom in render"),
):
with contextlib.redirect_stderr(buf):
rc = dashboard.cmd_dashboard([])
rc = supervise_cli.cmd_supervise([])
self.assertEqual(1, rc)
out = buf.getvalue()
self.assertIn("dashboard crashed: ValueError: kaboom in render", out)
self.assertIn("supervise crashed: ValueError: kaboom in render", out)
self.assertIn("full traceback written to", out)
log_path = self._root / "logs" / "dashboard-crash.log"
log_path = self._root / "logs" / "supervise-crash.log"
self.assertTrue(log_path.exists())
content = log_path.read_text()
self.assertIn("kaboom in render", content)
@@ -116,10 +116,10 @@ class TestWriteCrashLog(_FakeHomeMixin, unittest.TestCase):
try:
raise RuntimeError("explode")
except RuntimeError as e:
path = dashboard._write_crash_log(e)
self.assertEqual(self._root / "logs" / "dashboard-crash.log", path)
path = supervise_cli._write_crash_log(e)
self.assertEqual(self._root / "logs" / "supervise-crash.log", path)
text = path.read_text()
self.assertIn("=== dashboard crash", text)
self.assertIn("=== supervise crash", text)
self.assertIn("RuntimeError: explode", text)
def test_falls_back_to_tempfile_when_home_unwritable(self):
@@ -131,7 +131,7 @@ class TestWriteCrashLog(_FakeHomeMixin, unittest.TestCase):
try:
raise RuntimeError("explode2")
except RuntimeError as e:
path = dashboard._write_crash_log(e)
path = supervise_cli._write_crash_log(e)
self.assertTrue(path.exists())
self.assertIn("explode2", path.read_text())
@@ -1,4 +1,4 @@
"""Unit: dashboard's detail-view line builder.
"""Unit: supervise's detail-view line builder.
_detail_lines returns (text, attr) tuples. Most are plain; for
pipelock-block proposals it appends a "→ would allow host: <host>"
@@ -8,7 +8,7 @@ which hostname will land in pipelock's allowlist on approval."""
import unittest
from bot_bottle import supervise
from bot_bottle.cli import dashboard
from bot_bottle.cli import supervise as supervise_cli
from bot_bottle.supervise import (
Proposal,
TOOL_CAPABILITY_BLOCK,
@@ -18,7 +18,7 @@ from bot_bottle.supervise import (
)
def _qp(tool: str, payload: str) -> dashboard.QueuedProposal:
def _qp(tool: str, payload: str) -> supervise_cli.QueuedProposal:
from datetime import datetime, timezone
from pathlib import Path
p = Proposal.new(
@@ -29,14 +29,14 @@ def _qp(tool: str, payload: str) -> dashboard.QueuedProposal:
current_file_hash=sha256_hex(payload),
now=datetime(2026, 5, 25, 12, 0, 0, tzinfo=timezone.utc),
)
return dashboard.QueuedProposal(proposal=p, queue_dir=Path("/tmp/q"))
return supervise_cli.QueuedProposal(proposal=p, queue_dir=Path("/tmp/q"))
class TestPipelockHostHighlight(unittest.TestCase):
GREEN = 0xDEADBEEF # arbitrary sentinel; _detail_lines passes through
def test_appends_green_host_line_for_pipelock_block(self):
lines = dashboard._detail_lines(
lines = supervise_cli._detail_lines(
_qp(TOOL_PIPELOCK_BLOCK, "https://api.github.com/repos/foo/bar"),
green_attr=self.GREEN,
)
@@ -47,14 +47,14 @@ class TestPipelockHostHighlight(unittest.TestCase):
self.assertEqual(["api.github.com"], green_lines)
def test_no_green_lines_for_egress_block(self):
lines = dashboard._detail_lines(
lines = supervise_cli._detail_lines(
_qp(TOOL_EGRESS_BLOCK, '{"routes": []}'),
green_attr=self.GREEN,
)
self.assertEqual([], [t for t, a in lines if a == self.GREEN])
def test_no_green_lines_for_capability_block(self):
lines = dashboard._detail_lines(
lines = supervise_cli._detail_lines(
_qp(TOOL_CAPABILITY_BLOCK, "FROM python:3.13\n"),
green_attr=self.GREEN,
)
@@ -63,8 +63,8 @@ class TestPipelockHostHighlight(unittest.TestCase):
def test_skips_host_line_when_url_unparseable(self):
# Shouldn't happen in production — supervise_server validates
# the URL before queuing — but if a malformed payload ever
# reaches the dashboard, don't render a misleading host line.
lines = dashboard._detail_lines(
# reaches the supervise TUI, don't render a misleading host line.
lines = supervise_cli._detail_lines(
_qp(TOOL_PIPELOCK_BLOCK, "garbage-not-a-url"),
green_attr=self.GREEN,
)
@@ -73,7 +73,7 @@ class TestPipelockHostHighlight(unittest.TestCase):
def test_no_green_attr_passed_still_renders_host(self):
# Even without color support (green_attr=0), the host line
# is still present — it just won't be coloured.
lines = dashboard._detail_lines(
lines = supervise_cli._detail_lines(
_qp(TOOL_PIPELOCK_BLOCK, "https://api.github.com/x"),
green_attr=0,
)
@@ -86,14 +86,14 @@ class TestFailedUrlHost(unittest.TestCase):
def test_extracts_hostname(self):
self.assertEqual(
"api.github.com",
dashboard._failed_url_host("https://api.github.com/repos/foo"),
supervise_cli._failed_url_host("https://api.github.com/repos/foo"),
)
def test_returns_empty_for_unparseable(self):
self.assertEqual("", dashboard._failed_url_host("not a url"))
self.assertEqual("", supervise_cli._failed_url_host("not a url"))
def test_returns_empty_for_url_without_host(self):
self.assertEqual("", dashboard._failed_url_host("https:///nohost"))
self.assertEqual("", supervise_cli._failed_url_host("https:///nohost"))
if __name__ == "__main__":