Compare commits

...

17 Commits

Author SHA1 Message Date
didericis 39e0976ace docs(prd): redesign label+color prompt as a curses modal window
lint / lint (push) Successful in 1m44s
- Single modal with two steps (label then color) instead of
  bare text prompts dropped to terminal
- Default label is <agent_name>-<slug_suffix>; first keystroke
  replaces the pre-fill rather than appending to it
- Color step shows a navigable list with live color preview;
  (none) selected by default; Esc skips
- Modal lives in tui.py and is shared between supervisor flow
  and cmd_start

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-06-07 12:01:11 -04:00
didericis 299579ab7b ci(prd): rename PRD to prd-new placeholder per new convention 2026-06-07 11:59:53 -04:00
didericis 3a10c38511 docs(prd): renumber PRD 0051 → 0054 (0051 slot taken by launch-selector on main) 2026-06-07 11:59:53 -04:00
didericis-claude db54f3d0b4 docs(prd): add PRD 0051 (named/labelled agents, renumbered from 0049) 2026-06-07 11:59:53 -04:00
Quality Badge Bot 8105e93031 chore: update quality badges
- Pylint: 9.93/10
- Pyright: 0 errors

[skip ci]
2026-06-07 15:57:03 +00:00
didericis 0d5c2f1a2e chore(ci): remove prd-check workflow
test / unit (pull_request) Successful in 28s
test / integration (pull_request) Successful in 42s
lint / lint (push) Successful in 1m42s
prd-number / assign-numbers (push) Successful in 23s
test / unit (push) Successful in 35s
test / integration (push) Successful in 52s
Update Quality Badges / update-badges (push) Successful in 1m40s
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-06-07 11:43:42 -04:00
didericis bba24d87f7 fix(lint): resolve pyright and pylint issues in provider/backend changes
lint / lint (push) Successful in 1m31s
prd-check / no-prd-new-on-main (pull_request) Failing after 21s
test / unit (pull_request) Successful in 28s
test / integration (pull_request) Successful in 43s
- Remove unused Bottle import from docker/backend.py (pyright)
- Suppress wrong-import-position on circular-import-avoiding
  deferred imports in backend/__init__.py (pylint C0413)
- Add encoding="utf-8" to read_text() in smolmachines provision
  test (pylint W1514)
- Suppress consider-using-with on TemporaryDirectory setUp pattern
  in both provision test files (pylint R1732)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-06-07 11:38:54 -04:00
didericis efb3af4a93 feat(agent-provider): user plugin discovery, Dockerfile cascade, and provider-owned ca/git provisioning
- Add _load_user_plugin: loads AgentProvider subclass from
  ~/.bot-bottle/contrib/<name>/agent_provider.py; get_provider()
  checks there first before falling back to built-ins
- Add Dockerfile cascade to docker prepare: per-bottle override →
  manifest dockerfile → user plugin Dockerfile → provider default
- Move provision_ca and provision_git from backend-specific
  provision/ modules to AgentProvider ABC as overridable defaults;
  delete docker/provision/ca.py, docker/provision/git.py,
  smolmachines/provision/ca.py, smolmachines/provision/git.py
- Add git_gate_insteadof_host/scheme properties to BottlePlan base;
  SmolmachinesBottlePlan overrides them to return agent_git_gate_host
  and "http" so provision_git works correctly on both backends
- Move SIGKILL retry from smolmachines provision/ca.py into
  SmolmachinesBottle.exec via _exec_raw helper — all exec calls
  on smolmachines now transparently retry once on exit 137
- Relax manifest_agent template validation to allow user-defined
  template names; keep auth_token/forward_host_credentials guards
  for built-in-only features
- Update tests: rewrite test_docker_provision_git_user and
  test_smolmachines_provision to call provider methods directly;
  add TestSmolmachinesBottleExec for SIGKILL retry coverage

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-06-07 11:35:35 -04:00
didericis 65746af720 docs(prd): expand user-provider-plugins to cover Dockerfile convention and provisioning methods 2026-06-07 11:35:35 -04:00
didericis d9e9d27e01 ci(prd): rename PRD to prd-new placeholder per new convention 2026-06-07 11:35:35 -04:00
didericis-claude 83351606c6 docs: bump PRD number from 0052 to 0053
Renames docs/prds/0052-user-provider-plugins.md to 0053-user-provider-plugins.md
and updates the heading inside the file. 0052 is now reserved for the egress
DLP addon.
2026-06-07 11:35:35 -04:00
didericis-claude d528f578aa fix: correct broken imports and fileno() guard after rebase
codex_auth.py was moved into contrib/codex/ but still used `.log`/
`.util` relative imports that resolved to the parent bot_bottle
package before the move — update to `...log` / `...util`.

_read_winsize() called sys.stdin.fileno() outside the OSError guard;
pytest's redirected stdin raises UnsupportedOperation (an OSError
subclass) there, breaking test_returns_first_tty_size. Move fileno()
inside the try block so any non-TTY stream is skipped cleanly.
2026-06-07 11:35:35 -04:00
didericis-claude cf3310e818 docs: PRD 0052 — user-defined agent provider plugins
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-06-07 11:35:35 -04:00
didericis-claude 74d6b25183 refactor: move codex_auth into contrib/codex
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-06-07 11:35:35 -04:00
didericis-claude dc837a5400 feat(supervise)!: remove egress-block MCP tool and runtime route-mutation
lint / lint (push) Successful in 1m39s
test / unit (push) Successful in 40s
test / integration (push) Successful in 1m1s
test / unit (pull_request) Successful in 40s
Update Quality Badges / update-badges (push) Successful in 1m45s
test / integration (pull_request) Successful in 57s
Drops `egress-block` from the supervise sidecar, removes
`_merge_single_route`, `add_route`, and `apply_routes_change` from
egress_apply.py, and strips the proposal/approve/reject flow for egress
from the supervise CLI. The list-egress-routes and capability-block tools
are unaffected. Tests updated throughout.

Closes #198
2026-06-07 09:56:39 -04:00
didericis-claude 4eff49c9c5 build: drop unused agent-image apt deps
Removes socat, openssh-client, and dnsutils from Dockerfile.claude
and Dockerfile.codex.

- socat was the privileged forwarder for the in-container ssh-agent
  that PRD 0009 removed; nothing in bot_bottle references it.
- openssh-client was needed back when the agent talked ssh:// to
  upstreams; git-gate's insteadOf rewrites now route every upstream
  through HTTP/git-protocol, and ssh-keygen runs host-side from the
  deploy-key provisioner.
- dnsutils was only used by tests/integration/test_sandbox_escape.py
  (attack 4b runs dig from inside the agent container).

Splits python3/python3-pip/python3-venv onto a separate layer with
a comment noting they're app-specific and a candidate to move to a
downstream image.
2026-06-07 09:50:27 -04:00
didericis 965d5073c3 ci(prd): add prd-new placeholder convention and numbering workflow
Implements #213: PRDs use prd-new-<slug>.md while a PR is open; a
post-merge workflow on main assigns sequential numbers and renames the
file. A required PR check blocks prd-new-*.md from landing on main
without going through the workflow.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-06-06 22:02:21 -04:00
36 changed files with 1143 additions and 1175 deletions
+123
View File
@@ -0,0 +1,123 @@
# Assign sequential numbers to prd-new-*.md files on merge to main.
#
# When a PR merges to main and includes prd-new-*.md files this workflow:
# 1. Finds the next available NNNN number by scanning existing PRDs.
# 2. Renames each prd-new-*.md to NNNN-<slug>.md.
# 3. Updates the title header (# PRD prd-new: → # PRD NNNN:).
# 4. Flips Status: Draft → Active when the merge commit also touched
# files outside docs/prds/ (i.e. the implementation shipped together
# with the PRD).
# 5. Commits the renaming back to main.
#
# No-op if the push contained no prd-new-*.md files.
name: prd-number
on:
push:
branches:
- main
paths:
- 'docs/prds/prd-new-*.md'
jobs:
assign-numbers:
runs-on: ubuntu-latest
permissions:
contents: write
steps:
- name: Checkout
uses: actions/checkout@v4
with:
fetch-depth: 2
token: ${{ secrets.GITHUB_TOKEN }}
- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: "3.12"
- name: Configure git
run: |
git config user.name "github-actions[bot]"
git config user.email "github-actions[bot]@users.noreply.github.com"
- name: Assign PRD numbers
run: |
python3 - <<'EOF'
import os
import re
import subprocess
import sys
from pathlib import Path
prds_dir = Path("docs/prds")
# Files added in the latest commit (HEAD vs HEAD~1).
result = subprocess.run(
["git", "diff", "--name-only", "--diff-filter=A", "HEAD~1", "HEAD"],
capture_output=True, text=True, check=True,
)
added = [Path(p) for p in result.stdout.splitlines()]
new_prds = [p for p in added if p.parent == prds_dir
and re.match(r"prd-new-.+\.md$", p.name)]
if not new_prds:
print("No prd-new-*.md files added in this commit — nothing to do.")
sys.exit(0)
# Determine whether non-PRD files were also changed (for Status flip).
all_changed = subprocess.run(
["git", "diff", "--name-only", "HEAD~1", "HEAD"],
capture_output=True, text=True, check=True,
).stdout.splitlines()
non_prd_changed = any(
not f.startswith("docs/prds/") for f in all_changed
)
# Find next available number.
existing = sorted(
int(m.group(1))
for p in prds_dir.glob("*.md")
if (m := re.match(r"^(\d{4})-", p.name))
)
next_num = (max(existing) + 1) if existing else 1
for prd_path in sorted(new_prds):
slug = re.sub(r"^prd-new-", "", prd_path.stem)
new_name = f"{next_num:04d}-{slug}.md"
new_path = prds_dir / new_name
print(f" {prd_path.name} → {new_name}")
content = prd_path.read_text()
# Update title header.
content = re.sub(
r"^(#\s+PRD\s+)prd-new(:)",
rf"\g<1>{next_num:04d}\2",
content,
count=1,
flags=re.MULTILINE,
)
# Conditionally flip Status.
if non_prd_changed:
content = re.sub(
r"(\*\*Status:\*\*\s*)Draft",
r"\g<1>Active",
content,
count=1,
)
new_path.write_text(content)
subprocess.run(["git", "rm", str(prd_path)], check=True)
subprocess.run(["git", "add", str(new_path)], check=True)
next_num += 1
subprocess.run(
["git", "commit", "-m", "ci(prd): assign sequential numbers to new PRDs"],
check=True,
)
subprocess.run(["git", "push"], check=True)
EOF
+5 -4
View File
@@ -36,10 +36,11 @@ the container lifecycle and the copying of skills and env vars into it.
- Three kinds of doc, each with its own conventions in-folder; see - Three kinds of doc, each with its own conventions in-folder; see
`docs/README.md` for when to write which: `docs/README.md` for when to write which:
- **PRDs** (`docs/prds/`) — one feature per file, numbered - **PRDs** (`docs/prds/`) — one feature per file. While a PR is open
`NNNN-kebab.md`. A `Status:` line tracks lifecycle: Draft → Active the file is named `prd-new-<kebab>.md`; CI assigns a sequential
(shipped to `main`) → Superseded/Retargeted. Format in number on merge to `main` and renames it. A `Status:` line tracks
`docs/prds/README.md`. lifecycle: Draft → Active (shipped to `main`) →
Superseded/Retargeted. Format in `docs/prds/README.md`.
- **Research notes** (`docs/research/`) — opinionated investigations; - **Research notes** (`docs/research/`) — opinionated investigations;
unnumbered kebab-case, freeform and verdict-first. See unnumbered kebab-case, freeform and verdict-first. See
`docs/research/README.md`. `docs/research/README.md`.
+13 -7
View File
@@ -16,14 +16,20 @@ FROM node:22-slim
# features (status checks, commits, PR creation) — without git in the # features (status checks, commits, PR creation) — without git in the
# image, those features fail in surprising ways once the user does any # image, those features fail in surprising ways once the user does any
# real work. ca-certificates is already in the slim base; listed for # real work. ca-certificates is already in the slim base; listed for
# clarity in case the base ever drops it. socat is the privileged # clarity in case the base ever drops it. curl is here so any
# forwarder for the in-container ssh-agent (see bot_bottle/ssh.py): the agent # HTTPS_PROXY-aware tool (curl itself, plus anything that shells out
# runs as root and rejects non-root connections, so socat sits between # to it) works against egress's bumped TLS without the agent needing
# node and the agent socket. curl is here so any HTTPS_PROXY-aware # local DNS.
# tool (curl itself, plus anything that shells out to it) works
# against egress's bumped TLS without the agent needing local DNS.
RUN apt-get update \ RUN apt-get update \
&& apt-get install -y --no-install-recommends git ca-certificates openssh-client socat curl dnsutils python3 python3-pip python3-venv \ && apt-get install -y --no-install-recommends git ca-certificates curl \
&& rm -rf /var/lib/apt/lists/*
# App-specific deps. Python isn't required by claude-code itself
# (claude-code is a Node CLI), but is convenient for the agent to
# shell out to for ad-hoc scripts. Kept on its own layer so it can
# be moved to a downstream image if the base ever needs to shrink.
RUN apt-get update \
&& apt-get install -y --no-install-recommends python3 python3-pip python3-venv \
&& rm -rf /var/lib/apt/lists/* && rm -rf /var/lib/apt/lists/*
# Install claude-code globally. Pinned to the version verified in the v1 # Install claude-code globally. Pinned to the version verified in the v1
+9 -1
View File
@@ -6,7 +6,15 @@
FROM node:22-slim FROM node:22-slim
RUN apt-get update \ RUN apt-get update \
&& apt-get install -y --no-install-recommends git ca-certificates openssh-client socat curl dnsutils python3 python3-pip python3-venv \ && apt-get install -y --no-install-recommends git ca-certificates curl \
&& rm -rf /var/lib/apt/lists/*
# App-specific deps. Python isn't required by codex itself
# (codex is a Node CLI), but is convenient for the agent to shell
# out to for ad-hoc scripts. Kept on its own layer so it can be
# moved to a downstream image if the base ever needs to shrink.
RUN apt-get update \
&& apt-get install -y --no-install-recommends python3 python3-pip python3-venv \
&& rm -rf /var/lib/apt/lists/* && rm -rf /var/lib/apt/lists/*
RUN npm install -g --no-fund --no-audit @openai/codex@0.136.0 \ RUN npm install -g --no-fund --no-audit @openai/codex@0.136.0 \
+1 -1
View File
@@ -5,7 +5,7 @@
# bot-bottle # bot-bottle
[![test](https://gitea.dideric.is/didericis/bot-bottle/actions/workflows/test.yml/badge.svg?branch=main)](https://gitea.dideric.is/didericis/bot-bottle/actions?workflow=test.yml) [![test](https://gitea.dideric.is/didericis/bot-bottle/actions/workflows/test.yml/badge.svg?branch=main)](https://gitea.dideric.is/didericis/bot-bottle/actions?workflow=test.yml)
[![pylint](https://img.shields.io/badge/pylint-9.92%2F10-brightgreen)](https://github.com/PyCQA/pylint) [![pylint](https://img.shields.io/badge/pylint-9.93%2F10-brightgreen)](https://github.com/PyCQA/pylint)
[![pyright](https://img.shields.io/badge/pyright-0%20errors-brightgreen)](https://github.com/microsoft/pyright) [![pyright](https://img.shields.io/badge/pyright-0%20errors-brightgreen)](https://github.com/microsoft/pyright)
**Problem:** Developer wants to run a coding agent without supervision, but they don't want a prompt injected or misbehaving agent wrecking their environment or exfiltrating sensitive data. **Problem:** Developer wants to run a coding agent without supervision, but they don't want a prompt injected or misbehaving agent wrecking their environment or exfiltrating sensitive data.
+124 -3
View File
@@ -19,6 +19,10 @@ Per PRD 0050 the per-provider implementations live under
from __future__ import annotations from __future__ import annotations
import importlib.util
import os
import shlex
import tempfile
from abc import ABC, abstractmethod from abc import ABC, abstractmethod
from dataclasses import dataclass, field from dataclasses import dataclass, field
from pathlib import Path from pathlib import Path
@@ -174,13 +178,130 @@ class AgentProvider(ABC):
the supervise sidecar is reachable. No-op when the supervise sidecar is reachable. No-op when
`plan.supervise_plan is None`.""" `plan.supervise_plan is None`."""
def provision_ca(self, bottle: "Bottle", plan: "BottlePlan") -> None:
"""Install the egress MITM CA into the agent's trust store.
Default: Debian-style — cp the cert to the standard source path,
run update-ca-certificates, log the fingerprint. Override for
non-Debian base images or non-standard trust mechanisms."""
from .backend.util import AGENT_CA_PATH, log_ca_fingerprint, select_ca_cert
from .log import die
cert_host_path, label = select_ca_cert(plan.egress_plan)
bottle.cp_in(str(cert_host_path), AGENT_CA_PATH)
r = bottle.exec(
f"chmod 644 {AGENT_CA_PATH} && update-ca-certificates",
user="root",
)
if r.returncode != 0:
die(
f"update-ca-certificates failed (exit {r.returncode}): "
f"stdout={(r.stdout or '').strip()!r} "
f"stderr={(r.stderr or '').strip()!r}"
)
log_ca_fingerprint(cert_host_path, label)
def provision_git(self, bottle: "Bottle", plan: "BottlePlan") -> None:
"""Configure git inside the agent container.
Default: Debian/node — copies .git when --cwd is set, writes the
git-gate insteadOf gitconfig, sets user.name/email as node.
Override for images that run as a different user or use a
non-standard home directory."""
from .log import info
workspace = plan.workspace_plan
if workspace.enabled and workspace.copy_git and workspace.has_host_git_dir:
guest_workspace_git = f"{workspace.guest_path}/.git"
host_git = str(workspace.host_path / ".git")
info(f"copying {host_git} -> {bottle.name}:{guest_workspace_git}")
bottle.exec(f"mkdir -p {shlex.quote(workspace.guest_path)}", user="root")
bottle.cp_in(host_git, guest_workspace_git)
bottle.exec(
f"chown -R {shlex.quote(workspace.owner)} "
f"{shlex.quote(guest_workspace_git)}",
user="root",
)
manifest_bottle = plan.spec.manifest.bottle_for(plan.spec.agent_name)
if manifest_bottle.git:
from .git_gate import GIT_GATE_HOSTNAME, git_gate_render_gitconfig
gate_host = getattr(plan, "git_gate_insteadof_host", GIT_GATE_HOSTNAME)
gate_scheme = getattr(plan, "git_gate_insteadof_scheme", "git")
content = git_gate_render_gitconfig(
manifest_bottle.git, gate_host, scheme=gate_scheme,
)
guest_gitconfig = f"{plan.guest_home}/.gitconfig"
with tempfile.NamedTemporaryFile(
"w", dir=str(plan.stage_dir), prefix="gitconfig.", delete=False,
) as f:
f.write(content)
config_file = Path(f.name)
os.chmod(config_file, 0o600)
info(
f"writing {guest_gitconfig} with "
f"{len(manifest_bottle.git)} insteadOf rule(s)"
)
bottle.cp_in(str(config_file), guest_gitconfig)
bottle.exec(
f"chown node:node {shlex.quote(guest_gitconfig)} && "
f"chmod 644 {shlex.quote(guest_gitconfig)}",
user="root",
)
gu = manifest_bottle.git_user
if not gu.is_empty():
if gu.name:
info(f"git config --global user.name = {gu.name!r}")
bottle.exec(
f"git config --global user.name {shlex.quote(gu.name)}",
user="node",
)
if gu.email:
info(f"git config --global user.email = {gu.email!r}")
bottle.exec(
f"git config --global user.email {shlex.quote(gu.email)}",
user="node",
)
def _load_user_plugin(template: str) -> AgentProvider | None:
"""Check ~/.bot-bottle/contrib/<template>/agent_provider.py for a
user-defined AgentProvider subclass. Returns an instance if found,
None if the plugin directory doesn't exist, raises ValueError if
the file exists but exports no AgentProvider subclass."""
plugin_path = (
Path.home() / ".bot-bottle" / "contrib" / template / "agent_provider.py"
)
if not plugin_path.exists():
return None
spec = importlib.util.spec_from_file_location(
f"_user_contrib_{template}.agent_provider", plugin_path
)
if spec is None or spec.loader is None:
raise ValueError(f"user plugin at {plugin_path} could not be loaded")
mod = importlib.util.module_from_spec(spec)
spec.loader.exec_module(mod) # type: ignore[union-attr]
for obj in vars(mod).values():
if (
isinstance(obj, type)
and issubclass(obj, AgentProvider)
and obj is not AgentProvider
):
return obj()
raise ValueError(
f"user plugin at {plugin_path} defines no AgentProvider subclass"
)
def get_provider(template: str) -> AgentProvider: def get_provider(template: str) -> AgentProvider:
"""Resolve a provider template name to its plugin instance. """Resolve a provider template name to its plugin instance.
Lazy-imports the contrib module so importing this module doesn't Checks ~/.bot-bottle/contrib/<template>/agent_provider.py first so
pull provider-specific code paths in. Mirrors the contrib users can shadow a built-in for local testing. Falls through to the
convention PRD 0048 established for deploy key provisioners.""" built-in registry; raises ValueError for unknown names with no
matching user plugin."""
user_plugin = _load_user_plugin(template)
if user_plugin is not None:
return user_plugin
if template == PROVIDER_CLAUDE: if template == PROVIDER_CLAUDE:
from .contrib.claude.agent_provider import ClaudeAgentProvider from .contrib.claude.agent_provider import ClaudeAgentProvider
return ClaudeAgentProvider() return ClaudeAgentProvider()
+18 -18
View File
@@ -78,6 +78,20 @@ class BottlePlan(ABC):
stage_dir: Path stage_dir: Path
guest_home: str guest_home: str
git_gate_plan: GitGatePlan git_gate_plan: GitGatePlan
@property
def git_gate_insteadof_host(self) -> str:
"""Host (and optional port) used in git-gate insteadOf URLs.
Docker uses the compose-network DNS alias; smolmachines
overrides with a loopback IP:port since TSI has no DNS."""
return "git-gate"
@property
def git_gate_insteadof_scheme(self) -> str:
"""URL scheme for git-gate insteadOf rewrites. 'git' for
Docker (git daemon); 'http' for smolmachines (HTTP proxy
over a published host port)."""
return "git"
egress_plan: EgressPlan egress_plan: EgressPlan
supervise_plan: SupervisePlan | None supervise_plan: SupervisePlan | None
agent_provision: AgentProvisionPlan agent_provision: AgentProvisionPlan
@@ -339,36 +353,22 @@ class BottleBackend(ABC, Generic[PlanT, CleanupT]):
HTTPS_PROXY (claude-code, git over HTTPS, npm, curl) is HTTPS_PROXY (claude-code, git over HTTPS, npm, curl) is
intercepted without per-tool reconfiguration.""" intercepted without per-tool reconfiguration."""
provider = get_provider(plan.agent_provision.template) provider = get_provider(plan.agent_provision.template)
self.provision_ca(plan, bottle) provider.provision_ca(bottle, plan)
prompt_path = provider.provision_prompt(plan, bottle) prompt_path = provider.provision_prompt(plan, bottle)
provider.provision(plan, bottle) provider.provision(plan, bottle)
provider.provision_skills(plan, bottle) provider.provision_skills(plan, bottle)
self.provision_workspace(plan, bottle) self.provision_workspace(plan, bottle)
self.provision_git(plan, bottle) provider.provision_git(bottle, plan)
provider.provision_supervise_mcp( provider.provision_supervise_mcp(
plan, bottle, self.supervise_mcp_url(plan), plan, bottle, self.supervise_mcp_url(plan),
) )
return prompt_path return prompt_path
def provision_ca(self, plan: PlanT, bottle: "Bottle") -> None:
"""Install the per-bottle CA into the agent's trust store so
the agent trusts the bumped CONNECT cert egress presents.
Default impl is a no-op so
backends that don't yet support TLS interception (every backend
except Docker today) aren't forced to implement it. The Docker
backend overrides to docker-cp the cert in and run
`update-ca-certificates`."""
def provision_workspace(self, plan: PlanT, bottle: "Bottle") -> None: def provision_workspace(self, plan: PlanT, bottle: "Bottle") -> None:
"""Copy the operator workspace into the running bottle when """Copy the operator workspace into the running bottle when
the backend cannot bake it into the agent image. Default is the backend cannot bake it into the agent image. Default is
no-op for backends like Docker that handle this before launch.""" no-op for backends like Docker that handle this before launch."""
@abstractmethod
def provision_git(self, plan: PlanT, bottle: "Bottle") -> None:
"""Copy the host's cwd `.git` directory into the running
bottle if the user requested --cwd. No-op otherwise."""
def supervise_mcp_url(self, plan: PlanT) -> str: def supervise_mcp_url(self, plan: PlanT) -> str:
"""Return the agent-side URL of the per-bottle supervise """Return the agent-side URL of the per-bottle supervise
sidecar, or "" when this bottle has no sidecar. The provider sidecar, or "" when this bottle has no sidecar. The provider
@@ -411,8 +411,8 @@ class BottleBackend(ABC, Generic[PlanT, CleanupT]):
# Import concrete backend classes AFTER the base types are defined, so # Import concrete backend classes AFTER the base types are defined, so
# each backend module can pull BottleSpec / BottlePlan / BottleBackend # each backend module can pull BottleSpec / BottlePlan / BottleBackend
# via `from . import ...` without hitting a partially-initialized module. # via `from . import ...` without hitting a partially-initialized module.
from .docker import DockerBottleBackend # noqa: E402 from .docker import DockerBottleBackend # noqa: E402 # pylint: disable=wrong-import-position
from .smolmachines import SmolmachinesBottleBackend # noqa: E402 from .smolmachines import SmolmachinesBottleBackend # noqa: E402 # pylint: disable=wrong-import-position
# The dict is heterogeneous: each value is a BottleBackend specialized # The dict is heterogeneous: each value is a BottleBackend specialized
+1 -11
View File
@@ -25,7 +25,7 @@ from pathlib import Path
from typing import Generator, Sequence from typing import Generator, Sequence
from ...supervise import SUPERVISE_HOSTNAME, SUPERVISE_PORT from ...supervise import SUPERVISE_HOSTNAME, SUPERVISE_PORT
from .. import ActiveAgent, Bottle, BottleBackend, BottleSpec from .. import ActiveAgent, BottleBackend, BottleSpec
from . import cleanup as _cleanup from . import cleanup as _cleanup
from . import enumerate as _enumerate from . import enumerate as _enumerate
from . import launch as _launch from . import launch as _launch
@@ -33,10 +33,6 @@ from . import prepare as _prepare
from .bottle import DockerBottle from .bottle import DockerBottle
from .bottle_cleanup_plan import DockerBottleCleanupPlan from .bottle_cleanup_plan import DockerBottleCleanupPlan
from .bottle_plan import DockerBottlePlan from .bottle_plan import DockerBottlePlan
from .provision import ca as _ca
from .provision import git as _git
class DockerBottleBackend(BottleBackend["DockerBottlePlan", "DockerBottleCleanupPlan"]): class DockerBottleBackend(BottleBackend["DockerBottlePlan", "DockerBottleCleanupPlan"]):
"""Docker backend implementation. Selected by BOT_BOTTLE_BACKEND """Docker backend implementation. Selected by BOT_BOTTLE_BACKEND
(default).""" (default)."""
@@ -60,12 +56,6 @@ class DockerBottleBackend(BottleBackend["DockerBottlePlan", "DockerBottleCleanup
with _launch.launch(plan, provision=self.provision) as bottle: with _launch.launch(plan, provision=self.provision) as bottle:
yield bottle yield bottle
def provision_ca(self, plan: DockerBottlePlan, bottle: Bottle) -> None:
_ca.provision_ca(plan, bottle)
def provision_git(self, plan: DockerBottlePlan, bottle: Bottle) -> None:
_git.provision_git(plan, bottle)
def supervise_mcp_url(self, plan: DockerBottlePlan) -> str: def supervise_mcp_url(self, plan: DockerBottlePlan) -> str:
"""Docker bottles reach the supervise sidecar via the """Docker bottles reach the supervise sidecar via the
compose-network alias `supervise:9100`. No per-bottle URL compose-network alias `supervise:9100`. No per-bottle URL
+5 -191
View File
@@ -1,70 +1,20 @@
"""Host-side helper to apply a routes.yaml change to a running """Host-side helper for egress sidecar inspection (issue #198).
egress sidecar (PRD 0014 retargeted by PRD 0017 chunk 3, PRD 0053).
Used by the supervise dashboard when the operator approves an `_merge_single_route`, `add_route`, and `apply_routes_change` were
egress-block proposal. Fetches current routes.yaml, validates, removed when the egress-block MCP tool was dropped. The remaining
writes into the sidecar, then SIGHUPs to reload. helpers support runtime inspection and validation of the routes file
without modifying it at runtime.
""" """
from __future__ import annotations from __future__ import annotations
import json
import subprocess import subprocess
from pathlib import Path
from typing import cast
from ...egress import EGRESS_ROUTES_IN_CONTAINER from ...egress import EGRESS_ROUTES_IN_CONTAINER
from ...egress_addon_core import load_routes from ...egress_addon_core import load_routes
from ...yaml_subset import YamlSubsetError, parse_yaml_subset
from .bottle_state import egress_state_dir
from .sidecar_bundle import sidecar_bundle_container_name from .sidecar_bundle import sidecar_bundle_container_name
def _render_routes_payload(routes_list: list[dict[str, object]]) -> str:
"""Render a list-of-dicts routes payload as YAML matching the
shape `egress_render_routes` produces."""
if not routes_list:
return "routes: []\n"
lines: list[str] = ["routes:"]
for entry in routes_list:
host = str(entry.get("host", ""))
lines.append(f' - host: "{host}"')
auth_scheme = entry.get("auth_scheme")
token_env = entry.get("token_env")
if auth_scheme and token_env:
lines.append(f' auth_scheme: "{auth_scheme}"')
lines.append(f' token_env: "{token_env}"')
matches_obj = entry.get("matches")
if isinstance(matches_obj, list) and matches_obj:
lines.append(" matches:")
for match_entry in matches_obj:
me = cast(dict[str, object], match_entry)
first_key = True
if "paths" in me:
lines.append(" - paths:")
first_key = False
for pd in cast(list[dict[str, str]], me["paths"]):
if "type" in pd:
lines.append(f' - type: "{pd["type"]}"')
lines.append(f' value: "{pd["value"]}"')
else:
lines.append(f' - value: "{pd["value"]}"')
if "methods" in me:
methods_str = ", ".join(
f'"{m}"' for m in cast(list[str], me["methods"])
)
prefix = " - " if first_key else " "
lines.append(f'{prefix}methods: [{methods_str}]')
first_key = False
if first_key:
lines.append(" - {}")
return "\n".join(lines) + "\n"
def _egress_routes_host_path(slug: str) -> Path:
return egress_state_dir(slug) / "egress_routes.yaml"
class EgressApplyError(RuntimeError): class EgressApplyError(RuntimeError):
pass pass
@@ -92,144 +42,8 @@ def validate_routes_content(content: str) -> None:
) from e ) from e
def apply_routes_change(slug: str, new_content: str) -> tuple[str, str]:
container = sidecar_bundle_container_name(slug)
before = fetch_current_routes(slug)
validate_routes_content(new_content)
target = _egress_routes_host_path(slug)
target.parent.mkdir(parents=True, exist_ok=True)
target.write_text(new_content)
target.chmod(0o644)
sig = subprocess.run(
["docker", "kill", "--signal", "HUP", container],
capture_output=True, text=True, check=False,
)
if sig.returncode != 0:
raise EgressApplyError(
f"failed to SIGHUP {container}: "
f"{(sig.stderr or '').strip()}"
)
return before, new_content
def _merge_single_route(
current_yaml: str, new_route: dict[str, object],
) -> str:
"""Merge a single proposed route into the current routes.yaml.
- Host absent → append the route.
- Host present → union the match paths (proposed existing).
Auth is preserved from existing route.
"""
try:
cfg = parse_yaml_subset(current_yaml)
except YamlSubsetError as e:
raise EgressApplyError(
f"current routes.yaml is not valid YAML: {e}"
) from e
routes = cfg.get("routes")
if not isinstance(routes, list):
raise EgressApplyError(
"current routes.yaml: 'routes' is not a list"
)
routes_typed = cast(list[object], routes)
new_host = str(new_route.get("host", "")).lower()
if not new_host:
raise EgressApplyError(
"proposed route is missing 'host'"
)
# Build proposed matches from the input
proposed_matches = new_route.get("matches")
if proposed_matches is None:
# Accept legacy path_allowlist from agent proposals and convert
proposed_paths = new_route.get("path_allowlist")
if isinstance(proposed_paths, list) and proposed_paths:
proposed_matches = [{"paths": [{"value": p} for p in proposed_paths]}]
for entry in routes_typed:
if not isinstance(entry, dict):
continue
entry_typed = cast(dict[str, object], entry)
if str(entry_typed.get("host", "")).lower() == new_host:
# Merge matches: union path values from proposed into existing
if isinstance(proposed_matches, list) and proposed_matches:
existing_matches = entry_typed.get("matches")
if not isinstance(existing_matches, list):
existing_matches = []
# Simple merge: collect all existing path values, add new ones
existing_paths: set[str] = set()
for me in existing_matches:
me_typed = cast(dict[str, object], me) if isinstance(me, dict) else {}
paths = me_typed.get("paths")
if isinstance(paths, list):
for p in paths:
p_typed = cast(dict[str, object], p) if isinstance(p, dict) else {}
val = p_typed.get("value")
if isinstance(val, str):
existing_paths.add(val)
new_paths: list[str] = []
for me in proposed_matches:
me_typed = cast(dict[str, object], me) if isinstance(me, dict) else {}
paths = me_typed.get("paths")
if isinstance(paths, list):
for p in paths:
p_typed = cast(dict[str, object], p) if isinstance(p, dict) else {}
val = p_typed.get("value")
if isinstance(val, str) and val not in existing_paths:
new_paths.append(val)
existing_paths.add(val)
if new_paths:
existing_matches.append(
{"paths": [{"value": p} for p in new_paths]}
)
entry_typed["matches"] = existing_matches
break
else:
entry_typed: dict[str, object] = {"host": new_route.get("host")} # type: ignore
if isinstance(proposed_matches, list) and proposed_matches:
entry_typed["matches"] = proposed_matches
auth = new_route.get("auth")
if isinstance(auth, dict) and auth.get("scheme") and auth.get("token_ref"): # type: ignore
auth_typed = cast(dict[str, object], auth)
existing_slots = sorted({
str(r_entry.get("token_env", ""))
for r_entry_obj in routes_typed
if isinstance(r_entry_obj, dict)
for r_entry in [cast(dict[str, object], r_entry_obj)]
if r_entry.get("token_env")
})
next_idx = len(existing_slots)
entry_typed["auth_scheme"] = str(cast(object, auth_typed.get("scheme")))
entry_typed["token_env"] = f"EGRESS_TOKEN_{next_idx}"
routes_typed.append(entry_typed)
return _render_routes_payload(cast(list[dict[str, object]], routes_typed))
def add_route(slug: str, proposed_route_json: str) -> tuple[str, str]:
try:
proposed = json.loads(proposed_route_json)
except json.JSONDecodeError as e:
raise EgressApplyError(
f"proposed route is not valid JSON: {e}"
) from e
if not isinstance(proposed, dict):
raise EgressApplyError(
"proposed route must be a JSON object"
)
current = fetch_current_routes(slug)
merged = _merge_single_route(current, proposed)
return apply_routes_change(slug, merged)
__all__ = [ __all__ = [
"EgressApplyError", "EgressApplyError",
"add_route",
"apply_routes_change",
"fetch_current_routes", "fetch_current_routes",
"validate_routes_content", "validate_routes_content",
] ]
+10 -1
View File
@@ -15,7 +15,7 @@ from datetime import datetime, timezone
from dataclasses import replace from dataclasses import replace
from pathlib import Path from pathlib import Path
from ...agent_provider import agent_provision_plan, runtime_for from ...agent_provider import PROVIDER_TEMPLATES, agent_provision_plan, runtime_for
from ...egress import Egress from ...egress import Egress
from ...env import ResolvedEnv, resolve_env from ...env import ResolvedEnv, resolve_env
from ...git_gate import GitGate from ...git_gate import GitGate
@@ -100,6 +100,15 @@ def resolve_plan(
elif provider_runtime.dockerfile: elif provider_runtime.dockerfile:
image_default = provider_runtime.image image_default = provider_runtime.image
dockerfile_path = provider_runtime.dockerfile dockerfile_path = provider_runtime.dockerfile
elif provider.template not in PROVIDER_TEMPLATES:
user_dockerfile = (
Path.home() / ".bot-bottle" / "contrib" / provider.template / "Dockerfile"
)
if user_dockerfile.is_file():
image_default = f"bot-bottle-{provider.template}:{slug}"
dockerfile_path = str(user_dockerfile)
else:
image_default = provider_runtime.image
else: else:
image_default = provider_runtime.image image_default = provider_runtime.image
image = os.environ.get("BOT_BOTTLE_IMAGE", image_default) image = os.environ.get("BOT_BOTTLE_IMAGE", image_default)
@@ -2,10 +2,11 @@
Per PRD 0050 the per-provider provisioning steps (prompt, skills, Per PRD 0050 the per-provider provisioning steps (prompt, skills,
declarative provision-plan apply, supervise MCP registration) live on declarative provision-plan apply, supervise MCP registration) live on
the `AgentProvider` plugin under `bot_bottle/contrib/`. The modules the `AgentProvider` plugin under `bot_bottle/contrib/`. CA and git
left in this subpackage handle only the steps that are provisioning also moved to the AgentProvider ABC (with Debian/node
backend-specific: defaults); user plugins override them for non-standard images.
- ca.py — install per-bottle CA bundle into the guest trust store No modules remain in this subpackage — the directory is kept so that
- git.py — copy host cwd `.git` into the guest when --cwd is used existing imports of `from .provision import ...` don't need updating
if new backend-specific provisioners are added later.
""" """
-40
View File
@@ -1,40 +0,0 @@
"""Install the per-bottle egress MITM CA into the agent container's
trust store.
By the time this provisioner runs, `egress_tls_init` has generated
the egress CA and the path is re-bound into `plan.egress_plan`.
Cert lands on Debian's standard source path
(`/usr/local/share/ca-certificates/`); `update-ca-certificates`
rebuilds `/etc/ssl/certs/ca-certificates.crt`, which is what curl,
Python `ssl`, and OpenSSL-based tools all read by default. The env
trio set on the agent's `docker run` covers Node
(`NODE_EXTRA_CA_CERTS`) and Python `requests` /
`SSL_CERT_FILE`-honoring libraries that don't load the system
bundle.
The fingerprint is computed via stdlib (`ssl.PEM_cert_to_DER_cert`
+ `hashlib.sha256`) and logged once to stderr. The private key
stays on the host (under `stage_dir`) until teardown wipes the
stage dir; nothing in the agent ever sees it."""
from __future__ import annotations
from ... import Bottle
from ...util import AGENT_CA_PATH, log_ca_fingerprint, select_ca_cert
from ..bottle_plan import DockerBottlePlan
def provision_ca(plan: DockerBottlePlan, bottle: Bottle) -> None:
"""Copy the agent-facing CA cert into the agent, rebuild the
trust bundle, emit a one-line fingerprint log. Called from
`BottleBackend.provision` after the agent container is up."""
cert_host_path, label = select_ca_cert(plan.egress_plan)
bottle.cp_in(str(cert_host_path), AGENT_CA_PATH)
bottle.exec(
f"chmod 644 {AGENT_CA_PATH} && update-ca-certificates",
user="root",
)
log_ca_fingerprint(cert_host_path, label)
-106
View File
@@ -1,106 +0,0 @@
"""Git provisioning inside a running Docker bottle.
Three concerns, all about git in the agent:
1. If --cwd was passed AND the host cwd has a .git, copy that .git
into the planned guest workspace so the agent operates on the
user's repo.
2. If the bottle declares `git` entries (PRD 0008), write a
~/.gitconfig with insteadOf rules so every git operation
against a declared upstream (push, fetch, clone, pull,
ls-remote) transparently hits the per-agent git-gate. The
gate mirrors the upstream in both directions, so URL
rewriting is symmetric.
3. If the bottle declares `git.user` (issue #86), set
`git config --global user.{name,email}` inside the bottle so
the agent's commits are attributed to that identity.
"""
from __future__ import annotations
import shlex
from ....git_gate import GIT_GATE_HOSTNAME, git_gate_render_gitconfig
from ....log import info
from ... import Bottle
from ..bottle_plan import DockerBottlePlan
def provision_git(plan: DockerBottlePlan, bottle: Bottle) -> None:
"""Set up git inside the bottle. Runs all three subcases; each
no-ops when its condition isn't met."""
_provision_cwd_git(plan, bottle)
_provision_git_gate_config(plan, bottle)
_provision_git_user(plan, bottle)
def _provision_cwd_git(plan: DockerBottlePlan, bottle: Bottle) -> None:
"""If --cwd was set and the host cwd has a .git directory, copy
it into /home/node/workspace/.git and fix ownership. No-op
otherwise."""
workspace = plan.workspace_plan
if not (workspace.enabled and workspace.copy_git and workspace.has_host_git_dir):
return
guest_workspace_git = f"{workspace.guest_path}/.git"
host_git = str(workspace.host_path / ".git")
info(f"copying {host_git} -> {bottle.name}:{guest_workspace_git}")
bottle.cp_in(host_git, guest_workspace_git)
bottle.exec(
f"chown -R {shlex.quote(workspace.owner)} {shlex.quote(guest_workspace_git)}",
user="root",
)
def _provision_git_gate_config(plan: DockerBottlePlan, bottle: Bottle) -> None:
"""Write ~/.gitconfig in the bottle with the git-gate
insteadOf rules. No-op when the bottle has no `git` entries."""
manifest_bottle = plan.spec.manifest.bottle_for(plan.spec.agent_name)
if not manifest_bottle.git:
return
container_gitconfig = f"{plan.guest_home}/.gitconfig"
content = git_gate_render_gitconfig(manifest_bottle.git, GIT_GATE_HOSTNAME)
config_file = plan.stage_dir / "agent_gitconfig"
config_file.write_text(content)
config_file.chmod(0o600)
info(f"writing {container_gitconfig} with {len(manifest_bottle.git)} insteadOf rule(s)")
bottle.cp_in(str(config_file), container_gitconfig)
bottle.exec(
f"chown node:node {shlex.quote(container_gitconfig)} && "
f"chmod 644 {shlex.quote(container_gitconfig)}",
user="root",
)
def _provision_git_user(plan: DockerBottlePlan, bottle: Bottle) -> None:
"""Apply `git config --global user.{name,email}` inside the
bottle so the agent's commits are attributed to the operator-
chosen identity instead of the agent image's default
(which is no user — git would refuse to commit at all
until the agent ran its own `git config`).
Runs as the `node` user so `--global` lands in
`/home/node/.gitconfig` (matching the existing
`_provision_git_gate_config` write location). No-op when the
bottle didn't declare `git.user`.
Each field set independently — name-only or email-only
configs only run the `git config` line for the field
present."""
manifest_bottle = plan.spec.manifest.bottle_for(plan.spec.agent_name)
gu = manifest_bottle.git_user
if gu.is_empty():
return
if gu.name:
info(f"git config --global user.name = {gu.name!r}")
bottle.exec(
f"git config --global user.name {shlex.quote(gu.name)}",
user="node",
)
if gu.email:
info(f"git config --global user.email = {gu.email!r}")
bottle.exec(
f"git config --global user.email {shlex.quote(gu.email)}",
user="node",
)
@@ -22,8 +22,6 @@ from . import smolvm as _smolvm
from .bottle import SmolmachinesBottle from .bottle import SmolmachinesBottle
from .bottle_cleanup_plan import SmolmachinesBottleCleanupPlan from .bottle_cleanup_plan import SmolmachinesBottleCleanupPlan
from .bottle_plan import SmolmachinesBottlePlan from .bottle_plan import SmolmachinesBottlePlan
from .provision import ca as _ca
from .provision import git as _git
from .provision import workspace as _workspace from .provision import workspace as _workspace
@@ -55,21 +53,11 @@ class SmolmachinesBottleBackend(
with _launch.launch(plan, provision=self.provision) as bottle: with _launch.launch(plan, provision=self.provision) as bottle:
yield bottle yield bottle
def provision_ca(
self, plan: SmolmachinesBottlePlan, bottle: Bottle
) -> None:
_ca.provision_ca(plan, bottle)
def provision_workspace( def provision_workspace(
self, plan: SmolmachinesBottlePlan, bottle: Bottle self, plan: SmolmachinesBottlePlan, bottle: Bottle
) -> None: ) -> None:
_workspace.provision_workspace(plan, bottle) _workspace.provision_workspace(plan, bottle)
def provision_git(
self, plan: SmolmachinesBottlePlan, bottle: Bottle
) -> None:
_git.provision_git(plan, bottle)
def supervise_mcp_url(self, plan: SmolmachinesBottlePlan) -> str: def supervise_mcp_url(self, plan: SmolmachinesBottlePlan) -> str:
"""The smolmachines guest reaches the supervise sidecar via a """The smolmachines guest reaches the supervise sidecar via a
host-published random port the launch step pinned earlier host-published random port the launch step pinned earlier
+17 -3
View File
@@ -19,6 +19,7 @@ from __future__ import annotations
import subprocess import subprocess
import sys import sys
import time
from typing import Mapping, cast from typing import Mapping, cast
from ...agent_provider import PromptMode, prompt_args from ...agent_provider import PromptMode, prompt_args
@@ -131,6 +132,11 @@ class SmolmachinesBottle(Bottle):
self.agent_argv(argv, tty=tty), check=False, self.agent_argv(argv, tty=tty), check=False,
).returncode ).returncode
# smolvm/libkrun can SIGKILL an otherwise-normal exec during
# early-VM provisioning. Retry once after a short settle so
# callers (provision_ca, etc.) don't have to handle it themselves.
_SIGKILL_EXIT = 128 + 9
def exec(self, script: str, *, user: str = "node") -> ExecResult: def exec(self, script: str, *, user: str = "node") -> ExecResult:
"""Run a POSIX shell script as `user` (default `node`) and """Run a POSIX shell script as `user` (default `node`) and
capture the result. Matches the docker backend's `exec`, capture the result. Matches the docker backend's `exec`,
@@ -141,14 +147,22 @@ class SmolmachinesBottle(Bottle):
`runuser -u <user> -- env ... /bin/sh -c <script>` switches UID `runuser -u <user> -- env ... /bin/sh -c <script>` switches UID
without invoking a login shell, then sets HOME / USER and the without invoking a login shell, then sets HOME / USER and the
bottle env in the child process.""" bottle env in the child process.
Retries once on SIGKILL (exit 137) — libkrun occasionally
kills short-lived execs during VM bring-up."""
r = self._exec_raw(script, user=user)
if r.returncode == self._SIGKILL_EXIT:
time.sleep(1.0)
r = self._exec_raw(script, user=user)
return r
def _exec_raw(self, script: str, *, user: str = "node") -> ExecResult:
argv = [ argv = [
"--", "runuser", "-u", user, "--", "--", "runuser", "-u", user, "--",
"env", *_env_assignments_for(user, self._guest_env), "env", *_env_assignments_for(user, self._guest_env),
"/bin/sh", "-c", script, "/bin/sh", "-c", script,
] ]
# Call smolvm directly because this path needs the host-side
# subprocess capture shape used by the Docker backend.
r = subprocess.run( r = subprocess.run(
["smolvm", "machine", "exec", "--name", self.name] + argv, ["smolvm", "machine", "exec", "--name", self.name] + argv,
capture_output=True, text=True, check=False, capture_output=True, text=True, check=False,
@@ -82,6 +82,14 @@ class SmolmachinesBottlePlan(BottlePlan):
agent_git_gate_host: str = "" agent_git_gate_host: str = ""
agent_supervise_url: str = "" agent_supervise_url: str = ""
@property
def git_gate_insteadof_host(self) -> str:
return self.agent_git_gate_host
@property
def git_gate_insteadof_scheme(self) -> str:
return "http"
@property @property
def agent_command(self) -> str: def agent_command(self) -> str:
return self.agent_provision.command return self.agent_provision.command
@@ -2,11 +2,12 @@
Per PRD 0050 the per-provider provisioning steps (prompt, skills, Per PRD 0050 the per-provider provisioning steps (prompt, skills,
declarative provision-plan apply, supervise MCP registration) live on declarative provision-plan apply, supervise MCP registration) live on
the `AgentProvider` plugin under `bot_bottle/contrib/`. The modules the `AgentProvider` plugin under `bot_bottle/contrib/`. CA and git
left in this subpackage handle only the steps that are provisioning also moved to the AgentProvider ABC (with Debian/node
backend-specific: defaults); user plugins override them for non-standard images.
The module left in this subpackage handles the remaining backend-
specific step:
- ca.py install per-bottle CA bundle into the guest trust store
- git.py copy host cwd `.git` into the guest when --cwd is used
- workspace.py copy the operator workspace into the guest - workspace.py copy the operator workspace into the guest
""" """
@@ -1,90 +0,0 @@
"""Install the per-bottle egress MITM CA into the smolmachines
guest's trust store (PRD 0023 chunk 4d).
Mirrors `backend.docker.provision.ca`: copy the egress CA to
Debian's `/usr/local/share/ca-certificates/` path,
`update-ca-certificates` to rebuild the trust bundle, and log the
fingerprint once.
`smolvm machine exec` runs commands as root in the VM (no `-u`
flag exists; the VM init is root), so we don't need the explicit
`-u 0` the docker backend uses on its `docker exec` calls."""
from __future__ import annotations
import time
from ....log import die
from ...util import (
AGENT_CA_BUNDLE,
AGENT_CA_PATH,
log_ca_fingerprint,
select_ca_cert,
)
from ... import Bottle, ExecResult
from ..bottle_plan import SmolmachinesBottlePlan
_SIGKILL_EXIT = 128 + 9
def provision_ca(plan: SmolmachinesBottlePlan, bottle: Bottle) -> None:
"""Copy the agent-facing CA cert into the guest, rebuild the
trust bundle, emit a one-line fingerprint log. Called from
`BottleBackend.provision` after the smolvm guest is up."""
cert_host_path, label = select_ca_cert(plan.egress_plan)
bottle.cp_in(str(cert_host_path), AGENT_CA_PATH)
# Mode 0644 — readable to non-root tools in the guest.
# update-ca-certificates rebuilds the bundle at AGENT_CA_BUNDLE,
# which is what curl / Python ssl / OpenSSL-based tools read by
# default. The env trio (NODE_EXTRA_CA_CERTS / SSL_CERT_FILE /
# REQUESTS_CA_BUNDLE) on the guest_env covers Node + Python
# `requests` / libraries that don't load the system bundle.
#
r = _install_ca(bottle)
if r.returncode == _SIGKILL_EXIT:
# smolvm/libkrun can SIGKILL an otherwise-normal exec
# during early-VM provisioning. `update-ca-certificates`
# is idempotent, so retry the same install once after a
# short settle delay before treating it as fatal.
time.sleep(1.0)
r = _install_ca(bottle)
if r.returncode != 0:
# update-ca-certificates not adding our cert is fatal —
# claude-code's TLS handshake against the egress-MITM'd
# api.anthropic.com would fail downstream. Bail early
# with what we can see (output is captured so we can
# surface it).
die(
f"update-ca-certificates didn't add the agent CA "
f"(exit {r.returncode}): "
f"stdout={(r.stdout or '').strip()!r} "
f"stderr={(r.stderr or '').strip()!r}"
)
log_ca_fingerprint(cert_host_path, label)
def _install_ca(bottle: Bottle) -> ExecResult:
# chown + chmod + update-ca-certificates + bundle
# verification run in one exec so we only pay one
# round trip; the `&&` chaining surfaces the first failure
# as the return code. The verify check is more stable than
# requiring "1 added" in stdout: a retry after a
# partially-completed first run may legitimately report "0
# added" while the cert is already installed.
return bottle.exec(
f"chown root:root {AGENT_CA_PATH} && "
f"chmod 644 {AGENT_CA_PATH} && "
f"update-ca-certificates && "
f"openssl verify -CAfile {AGENT_CA_BUNDLE} {AGENT_CA_PATH}",
user="root",
)
# Re-exported for the launch/provision_ca caller + tests. The path
# constants live in the shared `backend.util` (Debian's
# `update-ca-certificates` layout is the same in both backends).
__all__ = ["AGENT_CA_BUNDLE", "AGENT_CA_PATH", "provision_ca"]
@@ -1,133 +0,0 @@
"""Git provisioning inside a running smolmachines bottle
(PRD 0023 chunk 4d).
Three concerns, all about git in the agent:
1. If --cwd was passed AND the host cwd has a .git, copy that
.git into the planned guest workspace so the agent operates on
the user's repo.
2. If the bottle declares `git` entries (PRD 0008), write a
~/.gitconfig with insteadOf rules so every git operation
against a declared upstream transparently hits the per-bottle
git-gate. The gate mirrors the upstream in both directions,
so URL rewriting is symmetric.
3. If the bottle declares `git.user` (issue #86), set
`git config --global user.{name,email}` inside the guest so
the agent's commits are attributed to that identity.
Differs from `backend.docker.provision.git` in one address detail:
the TSI-allowlisted guest can only reach the bundle's pinned IP
(no DNS resolver in the /32 allowlist), so the insteadOf URLs
are `http://<bundle_ip>:<port>/<name>.git` rather than the
docker backend's `git://git-gate/<name>.git`. The render itself
is the shared `git_gate_render_gitconfig` on the platform-neutral
git_gate module."""
from __future__ import annotations
import os
import shlex
import tempfile
from pathlib import Path
from ....git_gate import git_gate_render_gitconfig
from ....log import info
from ... import Bottle
from ..bottle_plan import SmolmachinesBottlePlan
def provision_git(plan: SmolmachinesBottlePlan, bottle: Bottle) -> None:
"""Set up git inside the guest. Runs all three subcases; each
no-ops when its condition isn't met."""
_provision_cwd_git(plan, bottle)
_provision_git_gate_config(plan, bottle)
_provision_git_user(plan, bottle)
def _provision_cwd_git(plan: SmolmachinesBottlePlan, bottle: Bottle) -> None:
"""If --cwd was set and the host cwd has a .git directory, copy
it into <guest_home>/workspace/.git and fix ownership. No-op
otherwise."""
workspace = plan.workspace_plan
if not (workspace.enabled and workspace.copy_git and workspace.has_host_git_dir):
return
guest_workspace_git = f"{workspace.guest_path}/.git"
host_git = str(workspace.host_path / ".git")
info(f"copying {host_git} -> {bottle.name}:{guest_workspace_git}")
# mkdir -p the workspace dir so cp_in lands the .git
# directly there even on first-time bottles.
bottle.exec(f"mkdir -p {shlex.quote(workspace.guest_path)}", user="root")
bottle.cp_in(host_git, guest_workspace_git)
# cp_in lands files as root; the agent runs as node so
# the workspace tree must be chowned over.
bottle.exec(
f"chown -R {shlex.quote(workspace.owner)} {shlex.quote(guest_workspace_git)}",
user="root",
)
def _provision_git_gate_config(
plan: SmolmachinesBottlePlan, bottle: Bottle
) -> None:
"""Write ~/.gitconfig in the guest with the git-gate insteadOf
rules. No-op when the bottle has no `git` entries."""
manifest_bottle = plan.spec.manifest.bottle_for(plan.spec.agent_name)
if not manifest_bottle.git:
return
# `<loopback alias>:<host port>` form: the bundle's git-gate
# HTTP port is published on host loopback at launch time so
# the smolvm guest (which can only reach macOS networking via
# TSI, not the docker bridge IP) can dial it. launch.py
# populates `plan.agent_git_gate_host` after bundle bringup.
content = git_gate_render_gitconfig(
manifest_bottle.git, plan.agent_git_gate_host, scheme="http",
)
guest_gitconfig = f"{plan.guest_home}/.gitconfig"
# Stage the file under the plan's stage_dir so cp_in
# has a stable host path. The plan's stage_dir is cleaned up
# by start.py's session-end teardown.
with tempfile.NamedTemporaryFile(
"w", dir=str(plan.stage_dir), prefix="gitconfig.",
delete=False,
) as f:
f.write(content)
config_file = Path(f.name)
os.chmod(config_file, 0o600)
info(f"writing {guest_gitconfig} with {len(manifest_bottle.git)} insteadOf rule(s)")
bottle.cp_in(str(config_file), guest_gitconfig)
bottle.exec(
f"chown node:node {shlex.quote(guest_gitconfig)} && "
f"chmod 644 {shlex.quote(guest_gitconfig)}",
user="root",
)
def _provision_git_user(
plan: SmolmachinesBottlePlan, bottle: Bottle,
) -> None:
"""Apply `git config --global user.{name,email}` inside the
guest as the node user so --global lands in the same
`/home/node/.gitconfig` that `_provision_git_gate_config`
writes to. No-op when the bottle didn't declare `git.user`.
SmolmachinesBottle.exec(user="node") automatically sets
HOME=/home/node so --global writes to /home/node/.gitconfig."""
manifest_bottle = plan.spec.manifest.bottle_for(plan.spec.agent_name)
gu = manifest_bottle.git_user
if gu.is_empty():
return
if gu.name:
info(f"git config --global user.name = {gu.name!r}")
bottle.exec(
f"git config --global user.name {shlex.quote(gu.name)}",
user="node",
)
if gu.email:
info(f"git config --global user.email = {gu.email!r}")
bottle.exec(
f"git config --global user.email {shlex.quote(gu.email)}",
user="node",
)
@@ -68,8 +68,9 @@ def _read_winsize() -> tuple[int, int] | None:
- tmux respawn-pane: tmux sets all three to the pane's PTY. - tmux respawn-pane: tmux sets all three to the pane's PTY.
- non-TTY (someone piped stdin in tests): none are; the - non-TTY (someone piped stdin in tests): none are; the
sync just no-ops, which is the right behavior.""" sync just no-ops, which is the right behavior."""
for fd in (sys.stdin.fileno(), sys.stdout.fileno(), sys.stderr.fileno()): for stream in (sys.stdin, sys.stdout, sys.stderr):
try: try:
fd = stream.fileno()
data = fcntl.ioctl(fd, termios.TIOCGWINSZ, b"\x00" * 8) data = fcntl.ioctl(fd, termios.TIOCGWINSZ, b"\x00" * 8)
except OSError: except OSError:
continue continue
+5 -14
View File
@@ -2,9 +2,8 @@
act on them (approve / modify / reject). act on them (approve / modify / reject).
Curses-based TUI; modify-then-approve shells out to $EDITOR. The Curses-based TUI; modify-then-approve shells out to $EDITOR. The
approval handlers wire to the per-tool remediation engines: approval handler wires to PRD 0016 (capability-block), which rebuilds
PRD 0014 (egress) writes routes.yaml + SIGHUPs egress; PRD 0016 the bottle Dockerfile. The egress-block tool was removed in issue #198.
(capability) rebuilds the bottle Dockerfile.
""" """
from __future__ import annotations from __future__ import annotations
@@ -26,7 +25,6 @@ from ..backend.docker.capability_apply import (
CapabilityApplyError, CapabilityApplyError,
apply_capability_change, apply_capability_change,
) )
from ..backend.docker.egress_apply import EgressApplyError, add_route
from ..log import Die, error, info from ..log import Die, error, info
from ..supervise import ( from ..supervise import (
COMPONENT_FOR_TOOL, COMPONENT_FOR_TOOL,
@@ -37,7 +35,6 @@ from ..supervise import (
STATUS_MODIFIED, STATUS_MODIFIED,
STATUS_REJECTED, STATUS_REJECTED,
TOOL_CAPABILITY_BLOCK, TOOL_CAPABILITY_BLOCK,
TOOL_EGRESS_BLOCK,
archive_proposal, archive_proposal,
list_pending_proposals, list_pending_proposals,
render_diff, render_diff,
@@ -61,7 +58,7 @@ class QueuedProposal:
# Errors any remediation engine may raise. Caught by the TUI key # Errors any remediation engine may raise. Caught by the TUI key
# handlers and surfaced in the status line so a failed apply keeps # handlers and surfaced in the status line so a failed apply keeps
# the proposal pending rather than crashing curses. # the proposal pending rather than crashing curses.
ApplyError = (EgressApplyError, CapabilityApplyError) ApplyError = (CapabilityApplyError,)
def discover_pending() -> list[QueuedProposal]: def discover_pending() -> list[QueuedProposal]:
@@ -82,9 +79,7 @@ def discover_pending() -> list[QueuedProposal]:
def _approval_status(qp: QueuedProposal, verb: str) -> str: def _approval_status(qp: QueuedProposal, verb: str) -> str:
"""Status-line text after a successful approval.""" """Status-line text after a successful approval."""
base = f"{verb} {qp.proposal.tool} for [{qp.proposal.bottle_slug}]" 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 f"{base}; resume: ./cli.py resume {qp.proposal.bottle_slug}"
return base
def _detail_lines( def _detail_lines(
@@ -132,11 +127,7 @@ def approve(
file_to_apply = final_file if final_file is not None else qp.proposal.proposed_file file_to_apply = final_file if final_file is not None else qp.proposal.proposed_file
diff_before, diff_after = "", "" diff_before, diff_after = "", ""
if qp.proposal.tool == TOOL_EGRESS_BLOCK: if qp.proposal.tool == TOOL_CAPABILITY_BLOCK:
diff_before, diff_after = add_route(
qp.proposal.bottle_slug, file_to_apply,
)
elif qp.proposal.tool == TOOL_CAPABILITY_BLOCK:
_meta = read_metadata(qp.proposal.bottle_slug) _meta = read_metadata(qp.proposal.bottle_slug)
if _meta is not None and not _meta.compose_project: if _meta is not None and not _meta.compose_project:
raise CapabilityApplyError( raise CapabilityApplyError(
+1 -1
View File
@@ -23,7 +23,7 @@ from ...agent_provider import (
AgentProvisionFile, AgentProvisionFile,
AgentProvisionPlan, AgentProvisionPlan,
) )
from ...codex_auth import codex_host_access_token, write_codex_dummy_auth_file from .codex_auth import codex_host_access_token, write_codex_dummy_auth_file
from ...egress import CODEX_HOST_CREDENTIAL_TOKEN_REF, EgressRoute from ...egress import CODEX_HOST_CREDENTIAL_TOKEN_REF, EgressRoute
from ...log import die, info, warn from ...log import die, info, warn
@@ -15,8 +15,8 @@ from datetime import datetime, timezone
from pathlib import Path from pathlib import Path
from typing import cast from typing import cast
from .log import die from ...log import die
from .util import expand_tilde from ...util import expand_tilde
def codex_auth_path(host_env: dict[str, str] | None = None) -> Path: def codex_auth_path(host_env: dict[str, str] | None = None) -> Path:
+12 -5
View File
@@ -49,11 +49,6 @@ class AgentProvider:
f"bottle '{bottle_name}' agent_provider.template must be a " f"bottle '{bottle_name}' agent_provider.template must be a "
f"non-empty string" f"non-empty string"
) )
if template not in PROVIDER_TEMPLATES:
raise ManifestError(
f"bottle '{bottle_name}' agent_provider.template {template!r} "
f"is not one of {', '.join(sorted(PROVIDER_TEMPLATES))}"
)
dockerfile = d.get("dockerfile", "") dockerfile = d.get("dockerfile", "")
if not isinstance(dockerfile, str): if not isinstance(dockerfile, str):
raise ManifestError( raise ManifestError(
@@ -66,6 +61,12 @@ class AgentProvider:
f"bottle '{bottle_name}' agent_provider.auth_token must be a " f"bottle '{bottle_name}' agent_provider.auth_token must be a "
f"string (was {type(auth_token).__name__})" f"string (was {type(auth_token).__name__})"
) )
if auth_token and template not in PROVIDER_TEMPLATES:
raise ManifestError(
f"bottle '{bottle_name}' agent_provider.auth_token is only "
f"supported for built-in templates "
f"({', '.join(sorted(PROVIDER_TEMPLATES))})"
)
if auth_token and template != "claude": if auth_token and template != "claude":
raise ManifestError( raise ManifestError(
f"bottle '{bottle_name}' agent_provider.auth_token is only " f"bottle '{bottle_name}' agent_provider.auth_token is only "
@@ -77,6 +78,12 @@ class AgentProvider:
f"bottle '{bottle_name}' agent_provider.forward_host_credentials " f"bottle '{bottle_name}' agent_provider.forward_host_credentials "
f"must be a boolean (was {type(forward_host_credentials).__name__})" f"must be a boolean (was {type(forward_host_credentials).__name__})"
) )
if forward_host_credentials and template not in PROVIDER_TEMPLATES:
raise ManifestError(
f"bottle '{bottle_name}' agent_provider.forward_host_credentials "
f"is only supported for built-in templates "
f"({', '.join(sorted(PROVIDER_TEMPLATES))})"
)
if forward_host_credentials and template != "codex": if forward_host_credentials and template != "codex":
raise ManifestError( raise ManifestError(
f"bottle '{bottle_name}' agent_provider.forward_host_credentials " f"bottle '{bottle_name}' agent_provider.forward_host_credentials "
+2 -7
View File
@@ -48,11 +48,9 @@ from pathlib import Path
SUPERVISE_HOSTNAME = "supervise" SUPERVISE_HOSTNAME = "supervise"
SUPERVISE_PORT = 9100 SUPERVISE_PORT = 9100
TOOL_EGRESS_BLOCK = "egress-block"
TOOL_CAPABILITY_BLOCK = "capability-block" TOOL_CAPABILITY_BLOCK = "capability-block"
TOOL_LIST_EGRESS_ROUTES = "list-egress-routes" TOOL_LIST_EGRESS_ROUTES = "list-egress-routes"
TOOLS: tuple[str, ...] = ( TOOLS: tuple[str, ...] = (
TOOL_EGRESS_BLOCK,
TOOL_CAPABILITY_BLOCK, TOOL_CAPABILITY_BLOCK,
TOOL_LIST_EGRESS_ROUTES, TOOL_LIST_EGRESS_ROUTES,
) )
@@ -70,10 +68,8 @@ EGRESS_INTROSPECT_URL = "http://_egress.local/allowlist"
# capability-block has no on-disk config the operator edits in place # capability-block has no on-disk config the operator edits in place
# (the Dockerfile is rebuilt, not patched), so it has no audit log # (the Dockerfile is rebuilt, not patched), so it has no audit log
# here — those changes are captured by git history + the rebuild # here — those changes are captured by git history + the rebuild
# record laid down in PRD 0016. # record laid down in PRD 0016. egress-block was removed in issue #198.
COMPONENT_FOR_TOOL: dict[str, str] = { COMPONENT_FOR_TOOL: dict[str, str] = {}
TOOL_EGRESS_BLOCK: "egress",
}
STATUS_APPROVED = "approved" STATUS_APPROVED = "approved"
STATUS_MODIFIED = "modified" STATUS_MODIFIED = "modified"
@@ -555,7 +551,6 @@ __all__ = [
"EGRESS_FORWARD_PROXY", "EGRESS_FORWARD_PROXY",
"EGRESS_INTROSPECT_URL", "EGRESS_INTROSPECT_URL",
"TOOL_CAPABILITY_BLOCK", "TOOL_CAPABILITY_BLOCK",
"TOOL_EGRESS_BLOCK",
"TOOL_LIST_EGRESS_ROUTES", "TOOL_LIST_EGRESS_ROUTES",
"archive_proposal", "archive_proposal",
"audit_dir", "audit_dir",
+6 -142
View File
@@ -1,8 +1,10 @@
"""Supervise sidecar HTTP server (PRD 0013). """Supervise sidecar HTTP server (PRD 0013).
Per-bottle MCP server exposing two tools `egress-block`, Per-bottle MCP server exposing tools the agent calls to propose config
`capability-block` that the agent calls to propose config changes changes when stuck. The egress-block tool was removed in issue #198;
when stuck. Each tool call: the remaining tools are `capability-block` and `list-egress-routes`.
Each queued tool call:
1. Validates the proposed file syntactically. 1. Validates the proposed file syntactically.
2. Writes a Proposal to /run/supervise/queue/ (bind-mounted from 2. Writes a Proposal to /run/supervise/queue/ (bind-mounted from
@@ -133,69 +135,6 @@ def jsonrpc_error(request_id: object, code: int, message: str) -> bytes:
TOOL_DEFINITIONS: list[dict[str, object]] = [ TOOL_DEFINITIONS: list[dict[str, object]] = [
{
"name": _sv.TOOL_EGRESS_BLOCK,
"description": (
"Call when egress refused your HTTPS request — host "
"without a matching route, or a request that did not match "
"the route's matches rules (typically a 403 from the "
"proxy). Propose a SINGLE route to add: the host you "
"need + (optionally) a path_allowlist of path prefixes + "
"(optionally) an auth block. The supervisor merges the "
"route into the live table at approval time — you do NOT "
"need to see or reproduce the existing routes. If the "
"host already has a route, the proposed paths are unioned "
"with the existing ones (host stays single-route). The "
"operator approves or rejects in the supervise TUI. On "
"approval the supervisor writes the merged routes.yaml "
"and SIGHUPs egress (no dropped connections)."
),
"inputSchema": {
"type": "object",
"properties": {
"host": {
"type": "string",
"description": (
"The hostname to allow (e.g. 'api.github.com'). "
"Case-insensitive on match."
),
},
"path_allowlist": {
"type": "array",
"items": {"type": "string"},
"description": (
"Optional URL path prefixes the route permits. "
"Each must start with '/'. Omit to allow all "
"paths under this host (bare-pass route). "
"Internally converted to matches entries."
),
},
"auth": {
"type": "object",
"description": (
"Optional credential injection. {scheme, "
"token_ref}: scheme is 'Bearer' or 'token'; "
"token_ref names the host env var holding the "
"secret value. Omit to add a host without "
"credential injection. Ignored if the host "
"already has a route (operator decides auth "
"changes, not the agent)."
),
"properties": {
"scheme": {"type": "string"},
"token_ref": {"type": "string"},
},
"required": ["scheme", "token_ref"],
"additionalProperties": False,
},
"justification": {
"type": "string",
"description": "Why this host needs to be allowed.",
},
},
"required": ["host", "justification"],
},
},
{ {
"name": _sv.TOOL_LIST_EGRESS_ROUTES, "name": _sv.TOOL_LIST_EGRESS_ROUTES,
"description": ( "description": (
@@ -254,11 +193,6 @@ PROPOSED_FILE_FIELD: dict[str, str] = {
# --- Validation ------------------------------------------------------------ # --- Validation ------------------------------------------------------------
# Auth schemes accepted on egress-block proposals — match the
# manifest-side EGRESS_AUTH_SCHEMES.
_AUTH_SCHEMES = ("Bearer", "token")
def validate_proposed_file(tool: str, content: str) -> None: def validate_proposed_file(tool: str, content: str) -> None:
"""Syntactic validation. The operator is the real gate; this just """Syntactic validation. The operator is the real gate; this just
catches obvious paste-errors / wrong-tool selections before they catches obvious paste-errors / wrong-tool selections before they
@@ -273,70 +207,6 @@ def validate_proposed_file(tool: str, content: str) -> None:
raise _RpcError(ERR_INVALID_PARAMS, f"unknown tool {tool!r}") raise _RpcError(ERR_INVALID_PARAMS, f"unknown tool {tool!r}")
def _validate_and_bundle_egress_route(
args: dict[str, object],
) -> str:
"""Validate egress-block input fields and bundle them into
a JSON string that becomes the Proposal.proposed_file. Raises
_RpcError on bad input the agent retries with a fixed shape."""
tool = _sv.TOOL_EGRESS_BLOCK
host = args.get("host")
if not isinstance(host, str) or not host.strip():
raise _RpcError(
ERR_INVALID_PARAMS,
f"{tool}: 'host' is required and must be a non-empty string",
)
payload: dict[str, object] = {"host": host}
path_allow_raw = args.get("path_allowlist")
if path_allow_raw is not None:
if not isinstance(path_allow_raw, list):
raise _RpcError(
ERR_INVALID_PARAMS,
f"{tool}: 'path_allowlist' must be an array of strings",
)
prefixes: list[str] = []
for i, p in enumerate(path_allow_raw):
if not isinstance(p, str):
raise _RpcError(
ERR_INVALID_PARAMS,
f"{tool}: path_allowlist[{i}] must be a string",
)
if not p.startswith("/"):
raise _RpcError(
ERR_INVALID_PARAMS,
f"{tool}: path_allowlist[{i}] {p!r} must start with '/'",
)
prefixes.append(p)
if prefixes:
payload["path_allowlist"] = prefixes
auth_raw = args.get("auth")
if auth_raw is not None:
if not isinstance(auth_raw, dict):
raise _RpcError(
ERR_INVALID_PARAMS,
f"{tool}: 'auth' must be an object with 'scheme' and 'token_ref'",
)
scheme = auth_raw.get("scheme")
token_ref = auth_raw.get("token_ref")
if not isinstance(scheme, str) or scheme not in _AUTH_SCHEMES:
raise _RpcError(
ERR_INVALID_PARAMS,
f"{tool}: auth.scheme must be one of "
f"{', '.join(_AUTH_SCHEMES)} (got {scheme!r})",
)
if not isinstance(token_ref, str) or not token_ref:
raise _RpcError(
ERR_INVALID_PARAMS,
f"{tool}: auth.token_ref must be a non-empty string "
f"naming the host env var holding the token",
)
payload["auth"] = {"scheme": scheme, "token_ref": token_ref}
return json.dumps(payload, indent=2) + "\n"
# --- MCP handlers ---------------------------------------------------------- # --- MCP handlers ----------------------------------------------------------
@@ -422,13 +292,7 @@ def handle_tools_call(
f"{name}: 'justification' is required and must be a non-empty string", f"{name}: 'justification' is required and must be a non-empty string",
) )
if name == _sv.TOOL_EGRESS_BLOCK: if name in PROPOSED_FILE_FIELD:
# Structured input → JSON bundle on Proposal.proposed_file.
# The dashboard's apply step (egress_apply.add_route)
# parses this JSON, fetches the current routes, merges in
# the new one, and writes the merged file.
proposed_file = _validate_and_bundle_egress_route(args_raw)
elif name in PROPOSED_FILE_FIELD:
file_field = PROPOSED_FILE_FIELD[name] file_field = PROPOSED_FILE_FIELD[name]
proposed_file = args_raw.get(file_field) proposed_file = args_raw.get(file_field)
if not isinstance(proposed_file, str): if not isinstance(proposed_file, str):
+7 -4
View File
@@ -7,9 +7,12 @@ document vs. a research note or a decision record).
## Naming and numbering ## Naming and numbering
`NNNN-kebab-title.md`, zero-padded and sequential (`0024-…`, `0025-…`). New PRDs use a `prd-new-<kebab-title>.md` placeholder name while the PR
Numbers are never reused; gaps are fine (there is no 0005). The number is open. On merge to `main` a CI workflow assigns the next sequential
is assigned at creation and stays fixed for the life of the doc. number (`0024-…`, `0025-…`), renames the file, and updates the title
header. Numbers are never reused; gaps are fine.
Once numbered, the filename stays fixed for the life of the doc.
## Status ## Status
@@ -23,7 +26,7 @@ The `Status:` line near the top tracks the PRD's lifecycle:
## Format ## Format
```markdown ```markdown
# PRD NNNN: <short title> # PRD prd-new: <short title> ← placeholder; CI fills in the number on merge
- **Status:** Draft - **Status:** Draft
- **Author:** <who> - **Author:** <who>
+319
View File
@@ -0,0 +1,319 @@
# PRD prd-new: Named / Labelled Agents
- **Status:** Draft
- **Author:** didericis
- **Created:** 2026-06-03
- **Issue:** #171
## Summary
At agent launch time, present the operator with a curses modal to optionally
set a human-readable label and color for the agent before it launches. The
modal pre-fills the label with the current agent name pattern (e.g.
`implementer-a3f9`) and leaves color unset; Enter with no changes accepts
those defaults. Store both in the bottle's `metadata.json`. Display the label —
rendered in the chosen color — in the supervisor'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 supervisor'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 (supervisor picker or CLI argument), a
curses modal appears before the backend picker / preflight. The modal
pre-fills the label with `<agent_name>-<slug_suffix>` (the same pattern
currently shown in the agents pane). No color is pre-selected.
2. In the modal, any printable keystroke immediately replaces the pre-filled
label and starts building the new name. Backspace edits normally. Enter
at any point confirms — accepting the pre-fill if nothing was typed, or
the in-progress text otherwise.
3. After the label field is confirmed, the modal presents color selection:
a list of the 16 ANSI color names the operator can navigate with arrow
keys, or Enter with no selection to skip color entirely.
4. `label` and `color` are stored in `BottleMetadata` and written to the
bottle's `metadata.json`. Both fields default to `""` (empty / unset).
5. `ActiveAgent` carries `label` and `color`; `enumerate_active()` reads them
from `metadata.json`.
6. The supervisor's 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.
7. `BottleSpec` carries `label` and `color`; both backends' `prepare` steps
copy them into `BottleMetadata`.
8. `ClaudeAgentProvider.provision_plan()` writes `label``"name"` and
`color``"color"` into the generated `claude.json`. Fields are omitted
when empty.
9. The supervisor's `_new_agent_flow` includes the modal between agent
selection and the backend picker.
10. `cmd_start` (CLI) shows the same modal (via the shared `tui` module)
before `_launch_bottle`; passes `label` / `color` into `BottleSpec`.
11. 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 supervisor agents pane.
- Validating or constraining label content beyond the 64-byte printable cap.
- Persisting color-pair state across supervisor 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).
## Design
### Data flow
```
operator input (modal)
BottleSpec.label, BottleSpec.color
├─► docker/prepare.py → BottleMetadata.label / .color → metadata.json
├─► smolmachines/prepare.py → BottleMetadata.label / .color → metadata.json
└─► contrib/claude/agent_provider.py → claude.json {"name": label, "color": color}
(omitted when empty)
supervisor refresh
enumerate_active() → read_metadata(slug) → ActiveAgent.label / .color
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.
### Agent row rendering
The display name logic becomes:
```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). Color
pairs are allocated lazily and cached in a `dict[str, int]` that lives for
the duration of the supervisor 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).
### The label+color modal
A single curses modal (`name_color_modal` in `bot_bottle/cli/tui.py`) handles
both label and color in two sequential steps within the same window. It is
called identically from the supervisor flow and from `cmd_start`.
```python
label, color = name_color_modal(default_label=f"{agent_name}-{slug_suffix}")
```
**Step 1 — label.** The window renders:
```
Name agent
──────────────────────────────────────
implementer-a3f9
──────────────────────────────────────
[any key] edit [Enter] confirm
```
The pre-filled text is shown in the input field. Any printable keystroke
immediately clears the pre-fill and starts a new name from that character
(first-keystroke-replaces semantics). Subsequent keystrokes append normally.
Backspace edits from the right. Enter confirms — accepting the pre-fill if
the field was never edited, or the typed text otherwise.
**Step 2 — color.** After confirming the label, the window transitions to:
```
Name agent
──────────────────────────────────────
implementer-a3f9 ← confirmed label
──────────────────────────────────────
Color (optional)
> (none)
red
green
blue
──────────────────────────────────────
[↑↓] move [Enter] select [Esc] skip
```
The list starts with `(none)` selected. Arrow keys move the cursor; Enter
confirms the highlighted choice; Esc or `q` skips color (equivalent to
selecting `(none)`). Each color name in the list is rendered in its own color
so the operator can preview the palette.
The function returns `(label, color)` — both strings, `color` is `""` when
`(none)` is selected or the step is skipped.
### Slug suffix for the default label
The default label is `<agent_name>-<slug_suffix>`, where `slug_suffix` is the
last four characters of the slug (the same short hash shown in the agents
pane). Both the supervisor and `cmd_start` have access to the slug after
`bottle_identity()` is called; the default label is computed there and passed
to `name_color_modal`.
In `cmd_start` the slug is not yet known at the time the modal appears (it is
minted inside `prepare`). The modal is therefore called with
`default_label=args.name` (the manifest agent key) as a simpler fallback; the
supervisor path, which has the slug available from `_new_agent_flow`, uses the
full `<agent_name>-<slug_suffix>` form.
### Claude Code config injection
Per PRD 0050, the `claude.json` trust-marker file is written by
`ClaudeAgentProvider.provision_plan()` in
`bot_bottle/contrib/claude/agent_provider.py`. Add `label: str = ""` and
`color: str = ""` keyword parameters to `provision_plan()` on both the
`AgentProvider` ABC and `ClaudeAgentProvider`, and to the
`agent_provision_plan()` shim in `agent_provider.py`. Both `prepare.py`
modules pass `spec.label` / `spec.color`; other providers accept the params
and ignore them.
In `ClaudeAgentProvider.provision_plan()`:
```python
payload = {
"hasCompletedOnboarding": True,
"theme": "dark",
"bypassPermissionsModeAccepted": True,
"projects": claude_projects,
}
if label:
payload["name"] = label
if color:
payload["color"] = color
claude_config.write_text(json.dumps(payload, indent=2) + "\n")
```
## 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` and `smolmachines/prepare.py`: copy `spec.label` /
`spec.color` into `BottleMetadata`; pass them to `agent_provision_plan()`.
- `docker/enumerate.py` and smolmachines equivalent: copy `metadata.label` /
`metadata.color` into `ActiveAgent`.
- Add `label: str = ""` and `color: str = ""` keyword params to
`AgentProvider.provision_plan()` (ABC), `ClaudeAgentProvider.provision_plan()`
(uses them in the `claude.json` write), and the `agent_provision_plan()` shim.
`CodexAgentProvider` accepts the params and ignores them.
- No prompt changes; no UI changes. All existing behavior is identical.
### Chunk 2 — modal + display
- `bot_bottle/cli/tui.py`: add `name_color_modal(default_label)` implementing
the two-step curses window described above.
- Supervisor `_new_agent_flow`: call `name_color_modal` after agent selection,
pass `label` / `color` into `BottleSpec`.
- `cmd_start`: call `name_color_modal(default_label=args.name)` before
`_launch_bottle`; pass `label` / `color` into `BottleSpec`.
- Supervisor agent row: add `_color_pair_for` helper; update row rendering to
use `a.label` with color.
## Open questions
None.
+269
View File
@@ -0,0 +1,269 @@
# PRD prd-new: User-defined agent provider plugins
- **Status:** Draft
- **Author:** claude
- **Created:** 2026-06-04
## Summary
The `get_provider()` registry in `bot_bottle/agent_provider.py` is a closed list —
only `"claude"` and `"codex"` are valid templates, validated at manifest-load time and
again at launch. Users who want to run a different agent (Gemini, Aider, a custom
local model wrapper) cannot add a provider without forking the package.
This PRD opens the registry to user-defined plugins. A plugin placed at
`~/.bot-bottle/contrib/<name>/` is discovered and loaded at launch time. The manifest
accepts any non-empty template string that names a built-in or resolves to a user
plugin at that path.
Alongside discovery, this PRD moves CA and git provisioning out of the Docker backend
and into the `AgentProvider` ABC as overridable methods. The current standalone
`provision/ca.py` and `provision/git.py` files in the Docker backend are deleted;
their logic becomes the default implementations on the ABC. This lets exotic provider
images (different base OS, different user, non-standard trust mechanism) override
provisioning freely without the abstraction fighting them.
The preceding commit on this PR moves `codex_auth.py` from `bot_bottle/` into
`bot_bottle/contrib/codex/` — a clean-up that fits naturally here since this PR
also clarifies that `contrib/` is the per-provider home.
## Problem
Users building unconventional setups hit a hard wall: the template validation in
`manifest_agent.AgentProvider.from_dict` rejects any string not in `PROVIDER_TEMPLATES`.
There is no escape hatch short of editing bot-bottle's source.
PRD 0050 moved provider logic into `contrib/` specifically so a third provider would
be "cheap to add" — but "cheap" today still means a pull request against the bot-bottle
repo, not a drop-in file in the user's home directory. The filesystem layout is already
the right shape; the discovery step is missing.
Beyond discovery, the Docker backend's `provision_ca` and `provision_git` functions
bake in Debian-specific commands (`update-ca-certificates`) and a hardcoded container
user (`node`). A user plugin that runs as a different user, or on a different base OS,
silently gets the wrong provisioning with no way to correct it short of forking.
## Goals / Success Criteria
1. A user places `~/.bot-bottle/contrib/<name>/agent_provider.py` — a file that exports
a class inheriting `AgentProvider` — sets `agent_provider.template: <name>` in a
bottle's frontmatter, and launches a bottle using that provider with no changes to
the bot-bottle source.
2. The plugin directory may also contain a `Dockerfile` at
`~/.bot-bottle/contrib/<name>/Dockerfile`; the existing three-tier Dockerfile cascade
(per-bottle override → manifest `dockerfile:` field → provider default) uses this
path as the provider default for user plugins.
3. The manifest validator accepts any non-empty template string. Unknown templates that
resolve to no user plugin still raise a clear error, but at launch (via `get_provider`)
rather than at manifest-load time.
4. Built-in provider knobs (`auth_token` → claude only; `forward_host_credentials`
codex only) are guarded to built-in template names. Bottles using a user provider
may set neither knob.
5. `get_provider(template)` checks `~/.bot-bottle/contrib/<template>/agent_provider.py`
before the built-ins, so a user can shadow a built-in for local testing.
6. A clear `ValueError` is raised if the user plugin file exists but contains no
`AgentProvider` subclass.
7. `AgentProvider` gains `provision_ca(self, bottle, plan)` and
`provision_git(self, bottle, plan)` with default implementations that reproduce
current Docker/Debian/node behavior. Built-in providers inherit the defaults
unchanged. User plugins override either method when their image diverges.
8. `bot_bottle/backend/docker/provision/ca.py` and
`bot_bottle/backend/docker/provision/git.py` are deleted. The Docker backend base
class calls `provider.provision_ca(bottle, plan)` and
`provider.provision_git(bottle, plan)` directly.
## Non-goals
- Packaging or distributing user plugins as installable Python packages.
- A plugin registry, index, or discovery beyond the filesystem path convention.
- Adding a third built-in provider.
- Validating that user plugin images, Dockerfiles, or commands exist before launch
(same policy as built-ins).
- Sandboxing user plugin code — plugins run with full Python interpreter access.
- Per-provider opt-out of the egress sidecar or network provisioning (follow-on).
## Scope
### In scope
- `get_provider(template: str) -> AgentProvider` gains a `_load_user_plugin(template)`
step that checks `~/.bot-bottle/contrib/<template>/agent_provider.py` first, then
falls through to the built-in look-ups.
- `_load_user_plugin` uses `importlib.util.spec_from_file_location` to load the module
and returns the first `AgentProvider` subclass found in its `__dict__`. Raises
`ValueError` if the file exists but exports no subclass.
- The Dockerfile cascade in the Docker backend's `resolve_plan()` uses
`~/.bot-bottle/contrib/<template>/Dockerfile` as the provider default for user
plugins (the same slot currently occupied by `Dockerfile.claude` / `Dockerfile.codex`
for built-ins).
- `manifest_agent.AgentProvider.from_dict`: the `template not in PROVIDER_TEMPLATES`
check is removed; the two built-in-specific knob guards (`auth_token` → claude,
`forward_host_credentials` → codex) are tightened to `template in PROVIDER_TEMPLATES`
so they are skipped for user-defined names.
- `PROVIDER_TEMPLATES` remains in `agent_provider.py` as the set of built-in names for
use by tests and any enumeration callers.
- `AgentProvider` ABC gains:
```python
def provision_ca(self, bottle: Bottle, plan: BottlePlan) -> None: ...
def provision_git(self, bottle: Bottle, plan: BottlePlan) -> None: ...
```
Default implementations reproduce the current `provision/ca.py` and
`provision/git.py` logic exactly (Debian `update-ca-certificates`, `node` user,
`/home/node` home).
- `bot_bottle/backend/docker/provision/ca.py` and
`bot_bottle/backend/docker/provision/git.py` deleted. The Docker backend base
class substitutes direct calls to the provider methods.
- Unit tests for the discovery path:
- Plugin found and loaded → correct `AgentProvider` instance returned.
- Plugin file exists but exports no subclass → `ValueError`.
- Unknown template with no user plugin → `ValueError` from `get_provider`.
- Built-in template name still works normally even when no user plugin exists.
- Unit tests for the provisioning delegation:
- A provider subclass that overrides `provision_ca` has its override called.
- A provider subclass that overrides `provision_git` has its override called.
- One paragraph added to `README.md` under a new "Custom providers" section describing
the `~/.bot-bottle/contrib/<name>/` convention (both `agent_provider.py` and
`Dockerfile`), the `provision_ca` / `provision_git` override points, and pointing at
the existing contrib providers as reference implementations.
### Out of scope
- Hot-reloading plugins during a running session.
- Plugin versioning or dependency declaration.
- Changes to the smolmachines backend provisioning path.
## Proposed design
### Discovery in `get_provider`
```python
import importlib.util
def get_provider(template: str) -> AgentProvider:
user_plugin = _load_user_plugin(template)
if user_plugin is not None:
return user_plugin
if template == PROVIDER_CLAUDE:
from .contrib.claude.agent_provider import ClaudeAgentProvider
return ClaudeAgentProvider()
if template == PROVIDER_CODEX:
from .contrib.codex.agent_provider import CodexAgentProvider
return CodexAgentProvider()
raise ValueError(f"unknown agent provider template: {template!r}")
def _load_user_plugin(template: str) -> AgentProvider | None:
plugin_path = (
Path.home() / ".bot-bottle" / "contrib" / template / "agent_provider.py"
)
if not plugin_path.exists():
return None
spec = importlib.util.spec_from_file_location(
f"_user_contrib_{template}.agent_provider", plugin_path
)
if spec is None or spec.loader is None:
raise ValueError(f"user plugin at {plugin_path} could not be loaded")
mod = importlib.util.module_from_spec(spec)
spec.loader.exec_module(mod) # type: ignore[union-attr]
for obj in vars(mod).values():
if (
isinstance(obj, type)
and issubclass(obj, AgentProvider)
and obj is not AgentProvider
):
return obj()
raise ValueError(
f"user plugin at {plugin_path} defines no AgentProvider subclass"
)
```
### Dockerfile convention for user plugins
`resolve_plan()` in the Docker backend already has a three-tier cascade. For user
plugins the provider-default slot is filled by:
```python
Path.home() / ".bot-bottle" / "contrib" / template / "Dockerfile"
```
Per-bottle overrides and manifest `dockerfile:` fields continue to take precedence.
### Provisioning methods on `AgentProvider`
```python
class AgentProvider(ABC):
...
def provision_ca(self, bottle: Bottle, plan: BottlePlan) -> None:
"""Install the egress MITM CA into the agent container's trust store.
Override for non-Debian base images or non-standard trust mechanisms."""
cert_host_path, label = select_ca_cert(plan.egress_plan)
bottle.cp_in(str(cert_host_path), AGENT_CA_PATH)
bottle.exec(
f"chmod 644 {AGENT_CA_PATH} && update-ca-certificates",
user="root",
)
log_ca_fingerprint(cert_host_path, label)
def provision_git(self, bottle: Bottle, plan: BottlePlan) -> None:
"""Configure git inside the agent container.
Override for images that run as a different user or use a non-standard home."""
_provision_cwd_git(plan, bottle)
_provision_git_gate_config(plan, bottle)
_provision_git_user(plan, bottle)
```
The Docker backend base class replaces the direct calls to the old standalone
functions with:
```python
provider.provision_ca(bottle, plan)
provider.provision_git(bottle, plan)
```
### Manifest validation change
In `manifest_agent.AgentProvider.from_dict`, remove the hard rejection:
```python
# Before
if template not in PROVIDER_TEMPLATES:
raise ManifestError(
f"bottle '{bottle_name}' agent_provider.template {template!r} "
f"is not one of {', '.join(sorted(PROVIDER_TEMPLATES))}"
)
# After — removed entirely; get_provider() raises at launch for unknown names
```
Guard the built-in knob checks with `template in PROVIDER_TEMPLATES`:
```python
if auth_token and template == "claude": # unchanged
...
if auth_token and template not in PROVIDER_TEMPLATES:
raise ManifestError(
f"bottle '{bottle_name}' agent_provider.auth_token is only "
f"supported for built-in templates ({', '.join(sorted(PROVIDER_TEMPLATES))})"
)
if forward_host_credentials and template == "codex": # unchanged
...
if forward_host_credentials and template not in PROVIDER_TEMPLATES:
raise ManifestError(
f"bottle '{bottle_name}' agent_provider.forward_host_credentials "
f"is only supported for built-in templates"
)
```
## Open questions
1. **`BOT_BOTTLE_CONTRIB_DIR` env var.** Omitted for now — `~/.bot-bottle/contrib/`
is consistent with the rest of the user config layout. Revisit if the need surfaces.
## References
- PRD 0050 — agent provider contrib (established `contrib/` as the per-provider home)
- PRD 0048 — SSH deploy key provisioning (the `contrib/` convention)
- `bot_bottle/agent_provider.py``get_provider`, `PROVIDER_TEMPLATES`, `AgentProvider` ABC
- `bot_bottle/manifest_agent.py` — template validation that this PRD relaxes
- `bot_bottle/backend/docker/provision/ca.py` — current CA provisioner (to be deleted)
- `bot_bottle/backend/docker/provision/git.py` — current git provisioner (to be deleted)
+1 -1
View File
@@ -9,7 +9,7 @@ import unittest
from datetime import datetime, timezone from datetime import datetime, timezone
from pathlib import Path from pathlib import Path
from bot_bottle.codex_auth import ( from bot_bottle.contrib.codex.codex_auth import (
codex_auth_path, codex_auth_path,
codex_dummy_auth_json, codex_dummy_auth_json,
codex_host_access_token, codex_host_access_token,
+37 -17
View File
@@ -1,10 +1,9 @@
"""Unit: docker backend `_provision_git_user` (issue #86). """Unit: AgentProvider.provision_git — git-user and cwd .git copy (issue #86).
Mocks `bottle.exec` / `bottle.cp_in` and asserts on the script Mocks bottle.exec / bottle.cp_in and asserts on the dispatched script
strings and user parameter. The cwd + git-gate passes are covered shape. provision_git is now a method on AgentProvider (default impl);
indirectly by the existing integration-shaped tests in the internal passes (_provision_cwd_git, _provision_git_gate_config,
test_smolmachines_provision; this file targets just the git_user _provision_git_user) are no longer exposed as separate helpers."""
pass."""
from __future__ import annotations from __future__ import annotations
@@ -13,16 +12,39 @@ import unittest
from pathlib import Path from pathlib import Path
from unittest.mock import MagicMock from unittest.mock import MagicMock
from bot_bottle.agent_provider import AgentProvisionPlan from bot_bottle.agent_provider import (
AgentProvider,
AgentProvisionPlan,
AgentProviderRuntime,
)
from bot_bottle.backend import Bottle, BottleSpec, ExecResult from bot_bottle.backend import Bottle, BottleSpec, ExecResult
from bot_bottle.backend.docker.bottle_plan import DockerBottlePlan from bot_bottle.backend.docker.bottle_plan import DockerBottlePlan
from bot_bottle.backend.docker.provision import git as _git
from bot_bottle.egress import EgressPlan from bot_bottle.egress import EgressPlan
from bot_bottle.git_gate import GitGatePlan from bot_bottle.git_gate import GitGatePlan
from bot_bottle.manifest import Manifest from bot_bottle.manifest import Manifest
from bot_bottle.workspace import workspace_plan from bot_bottle.workspace import workspace_plan
class _Provider(AgentProvider):
"""Minimal concrete subclass for testing the default provision_git."""
@property
def runtime(self) -> AgentProviderRuntime:
return AgentProviderRuntime(
template="test", command="test", image="", dockerfile="",
prompt_mode="append_file", bypass_args=(), resume_args=(),
remote_control_args=(),
)
def provision_plan(self, **kwargs): # type: ignore[override]
raise NotImplementedError
def provision_skills(self, plan, bottle): ... # type: ignore[override]
def provision_prompt(self, plan, bottle): ... # type: ignore[override]
def provision(self, plan, bottle): ... # type: ignore[override]
def provision_supervise_mcp(self, plan, bottle, supervise_url): ... # type: ignore[override]
_PROVIDER = _Provider()
def _plan(*, git_user: dict | None = None, # type: ignore def _plan(*, git_user: dict | None = None, # type: ignore
copy_cwd: bool = False, copy_cwd: bool = False,
user_cwd: str = "/tmp/x", user_cwd: str = "/tmp/x",
@@ -87,8 +109,6 @@ def _make_bottle(name: str = "bot-bottle-demo-abc12") -> MagicMock:
def _git_config_exec_calls(bottle: MagicMock) -> list[tuple[str, str]]: def _git_config_exec_calls(bottle: MagicMock) -> list[tuple[str, str]]:
"""Filter bottle.exec calls to git-config invocations.
Returns list of (script, user) tuples."""
out = [] out = []
for c in bottle.exec.call_args_list: for c in bottle.exec.call_args_list:
script = c.args[0] if c.args else c.kwargs.get("script", "") script = c.args[0] if c.args else c.kwargs.get("script", "")
@@ -100,7 +120,7 @@ def _git_config_exec_calls(bottle: MagicMock) -> list[tuple[str, str]]:
class TestProvisionGitUser(unittest.TestCase): class TestProvisionGitUser(unittest.TestCase):
def setUp(self): def setUp(self):
self._tmp = tempfile.TemporaryDirectory(prefix="cb-prov-git-user.") self._tmp = tempfile.TemporaryDirectory(prefix="cb-prov-git-user.") # pylint: disable=consider-using-with
self.stage = Path(self._tmp.name) self.stage = Path(self._tmp.name)
def tearDown(self): def tearDown(self):
@@ -108,7 +128,7 @@ class TestProvisionGitUser(unittest.TestCase):
def test_noop_when_no_git_user(self): def test_noop_when_no_git_user(self):
bottle = _make_bottle() bottle = _make_bottle()
_git._provision_git_user(_plan(stage_dir=self.stage), bottle) _PROVIDER.provision_git(bottle, _plan(stage_dir=self.stage))
self.assertEqual([], _git_config_exec_calls(bottle)) self.assertEqual([], _git_config_exec_calls(bottle))
def test_copies_cwd_git_to_workspace_plan_path(self): def test_copies_cwd_git_to_workspace_plan_path(self):
@@ -116,7 +136,7 @@ class TestProvisionGitUser(unittest.TestCase):
(cwd / ".git").mkdir(parents=True) (cwd / ".git").mkdir(parents=True)
plan = _plan(copy_cwd=True, user_cwd=str(cwd), stage_dir=self.stage) plan = _plan(copy_cwd=True, user_cwd=str(cwd), stage_dir=self.stage)
bottle = _make_bottle() bottle = _make_bottle()
_git._provision_cwd_git(plan, bottle) _PROVIDER.provision_git(bottle, plan)
bottle.cp_in.assert_called_once_with( bottle.cp_in.assert_called_once_with(
f"{cwd}/.git", f"{cwd}/.git",
@@ -125,10 +145,10 @@ class TestProvisionGitUser(unittest.TestCase):
chown_calls = [ chown_calls = [
c for c in bottle.exec.call_args_list c for c in bottle.exec.call_args_list
if "chown" in (c.args[0] if c.args else "") if "chown" in (c.args[0] if c.args else "")
and "/home/node/workspace/.git" in (c.args[0] if c.args else "")
] ]
self.assertEqual(1, len(chown_calls)) self.assertEqual(1, len(chown_calls))
self.assertIn("node:node", chown_calls[0].args[0]) self.assertIn("node:node", chown_calls[0].args[0])
self.assertIn("/home/node/workspace/.git", chown_calls[0].args[0])
def test_sets_name_and_email(self): def test_sets_name_and_email(self):
plan = _plan( plan = _plan(
@@ -136,7 +156,7 @@ class TestProvisionGitUser(unittest.TestCase):
stage_dir=self.stage, stage_dir=self.stage,
) )
bottle = _make_bottle() bottle = _make_bottle()
_git._provision_git_user(plan, bottle) _PROVIDER.provision_git(bottle, plan)
calls = _git_config_exec_calls(bottle) calls = _git_config_exec_calls(bottle)
self.assertEqual(2, len(calls)) self.assertEqual(2, len(calls))
for script, user in calls: for script, user in calls:
@@ -150,7 +170,7 @@ class TestProvisionGitUser(unittest.TestCase):
def test_name_only_sets_only_name(self): def test_name_only_sets_only_name(self):
plan = _plan(git_user={"name": "Bot"}, stage_dir=self.stage) plan = _plan(git_user={"name": "Bot"}, stage_dir=self.stage)
bottle = _make_bottle() bottle = _make_bottle()
_git._provision_git_user(plan, bottle) _PROVIDER.provision_git(bottle, plan)
calls = _git_config_exec_calls(bottle) calls = _git_config_exec_calls(bottle)
self.assertEqual(1, len(calls)) self.assertEqual(1, len(calls))
self.assertIn("user.name", calls[0][0]) self.assertIn("user.name", calls[0][0])
@@ -161,7 +181,7 @@ class TestProvisionGitUser(unittest.TestCase):
git_user={"email": "bot@example.com"}, stage_dir=self.stage, git_user={"email": "bot@example.com"}, stage_dir=self.stage,
) )
bottle = _make_bottle() bottle = _make_bottle()
_git._provision_git_user(plan, bottle) _PROVIDER.provision_git(bottle, plan)
calls = _git_config_exec_calls(bottle) calls = _git_config_exec_calls(bottle)
self.assertEqual(1, len(calls)) self.assertEqual(1, len(calls))
self.assertIn("user.email", calls[0][0]) self.assertIn("user.email", calls[0][0])
+3 -136
View File
@@ -1,27 +1,19 @@
"""Unit: validate_routes_content (PRD 0014 retargeted by PRD 0017 """Unit: validate_routes_content (issue #198: _merge_single_route and
chunk 3, PRD 0053). docker exec / cp / kill paths are covered by the add_route removed; docker exec / cp / kill paths are covered by the
integration test.""" integration test)."""
import unittest import unittest
from bot_bottle.backend.docker.egress_apply import ( from bot_bottle.backend.docker.egress_apply import (
EgressApplyError, EgressApplyError,
_merge_single_route,
validate_routes_content, validate_routes_content,
) )
from bot_bottle.yaml_subset import parse_yaml_subset
_ROUTES_EMPTY = "routes: []\n" _ROUTES_EMPTY = "routes: []\n"
_ROUTES_ONE = 'routes:\n - host: "api.anthropic.com"\n' _ROUTES_ONE = 'routes:\n - host: "api.anthropic.com"\n'
def _routes(parsed: str) -> list[dict]: # type: ignore
"""Parse a YAML routes string and pull out the routes list, so
tests can assert on shape directly."""
return parse_yaml_subset(parsed)["routes"] # type: ignore
class TestValidateRoutesContent(unittest.TestCase): class TestValidateRoutesContent(unittest.TestCase):
def test_accepts_minimal_route_table(self): def test_accepts_minimal_route_table(self):
validate_routes_content(_ROUTES_EMPTY) validate_routes_content(_ROUTES_EMPTY)
@@ -60,130 +52,5 @@ class TestValidateRoutesContent(unittest.TestCase):
) )
class TestMergeSingleRoute(unittest.TestCase):
BASE = _ROUTES_ONE
def test_appends_route_when_host_absent(self):
merged = _merge_single_route(self.BASE, {"host": "github.com"})
hosts = [r["host"] for r in _routes(merged)]
self.assertEqual(["api.anthropic.com", "github.com"], hosts)
def test_appends_matches(self):
merged = _merge_single_route(
self.BASE,
{"host": "github.com", "matches": [
{"paths": [{"value": "/repos/x/"}]}
]},
)
new_route = _routes(merged)[-1]
self.assertIn("matches", new_route)
def test_appends_legacy_path_allowlist_as_matches(self):
merged = _merge_single_route(
self.BASE,
{"host": "github.com", "path_allowlist": ["/repos/x/"]},
)
new_route = _routes(merged)[-1]
self.assertIn("matches", new_route)
def test_appends_auth_with_token_env_slot(self):
merged = _merge_single_route(
self.BASE,
{
"host": "api.github.com",
"auth": {"scheme": "Bearer", "token_ref": "GH"},
},
)
new_route = _routes(merged)[-1]
self.assertEqual("Bearer", new_route["auth_scheme"])
self.assertEqual("EGRESS_TOKEN_0", new_route["token_env"])
def test_auth_slot_increments_past_existing(self):
base = (
'routes:\n'
' - host: "api.anthropic.com"\n'
' auth_scheme: "Bearer"\n'
' token_env: "EGRESS_TOKEN_0"\n'
)
merged = _merge_single_route(base, {
"host": "api.github.com",
"auth": {"scheme": "Bearer", "token_ref": "GH"},
})
new_route = _routes(merged)[-1]
self.assertEqual("EGRESS_TOKEN_1", new_route["token_env"])
def test_existing_host_merges_match_paths_as_union(self):
base = (
'routes:\n'
' - host: "github.com"\n'
' matches:\n'
' - paths:\n'
' - value: "/a/"\n'
)
merged = _merge_single_route(base, {
"host": "github.com",
"matches": [{"paths": [{"value": "/b/"}]}],
})
routes = _routes(merged)
self.assertEqual(1, len(routes))
all_paths: list[str] = []
for me in routes[0].get("matches", []):
for p in me.get("paths", []):
all_paths.append(p["value"])
self.assertIn("/a/", all_paths)
self.assertIn("/b/", all_paths)
def test_existing_host_dedup_match_paths(self):
base = (
'routes:\n'
' - host: "github.com"\n'
' matches:\n'
' - paths:\n'
' - value: "/a/"\n'
)
merged = _merge_single_route(base, {
"host": "github.com",
"matches": [{"paths": [{"value": "/a/"}, {"value": "/b/"}]}],
})
all_paths: list[str] = []
for me in _routes(merged)[0].get("matches", []):
for p in me.get("paths", []):
all_paths.append(p["value"])
self.assertEqual(1, all_paths.count("/a/"))
self.assertIn("/b/", all_paths)
def test_existing_host_preserves_existing_auth_ignores_proposed(self):
base = (
'routes:\n'
' - host: "api.github.com"\n'
' auth_scheme: "Bearer"\n'
' token_env: "EGRESS_TOKEN_0"\n'
)
merged = _merge_single_route(base, {
"host": "api.github.com",
"auth": {"scheme": "token", "token_ref": "DIFFERENT"},
})
route = _routes(merged)[0]
self.assertEqual("Bearer", route["auth_scheme"])
self.assertEqual("EGRESS_TOKEN_0", route["token_env"])
def test_host_match_is_case_insensitive(self):
base = 'routes:\n - host: "GitHub.com"\n'
merged = _merge_single_route(base, {
"host": "github.com",
"matches": [{"paths": [{"value": "/x/"}]}],
})
routes = _routes(merged)
self.assertEqual(1, len(routes))
def test_missing_host_raises(self):
with self.assertRaises(EgressApplyError):
_merge_single_route(self.BASE, {})
def test_invalid_current_yaml_raises(self):
with self.assertRaises(EgressApplyError):
_merge_single_route("routes:\n\tbad", {"host": "x.example"})
if __name__ == "__main__": if __name__ == "__main__":
unittest.main() unittest.main()
+89 -36
View File
@@ -14,21 +14,23 @@ from pathlib import Path
from unittest.mock import MagicMock, patch from unittest.mock import MagicMock, patch
from bot_bottle.agent_provider import ( from bot_bottle.agent_provider import (
AgentProvider,
AgentProviderRuntime,
AgentProvisionCommand, AgentProvisionCommand,
AgentProvisionDir, AgentProvisionDir,
AgentProvisionFile, AgentProvisionFile,
AgentProvisionPlan, AgentProvisionPlan,
) )
from bot_bottle.backend import Bottle, BottleSpec, ExecResult from bot_bottle.backend import Bottle, BottleSpec, ExecResult
from bot_bottle.backend.smolmachines.bottle import SmolmachinesBottle
from bot_bottle.backend.smolmachines.bottle_plan import ( from bot_bottle.backend.smolmachines.bottle_plan import (
SmolmachinesBottlePlan, SmolmachinesBottlePlan,
) )
from bot_bottle.backend.smolmachines.provision import ( from bot_bottle.backend.smolmachines.provision import (
ca as _ca,
git as _git,
workspace as _workspace, workspace as _workspace,
) )
from bot_bottle.backend.smolmachines.launch import _bundle_launch_spec from bot_bottle.backend.smolmachines.launch import _bundle_launch_spec
from bot_bottle.backend.util import AGENT_CA_PATH
from bot_bottle.egress import EgressPlan, EgressRoute from bot_bottle.egress import EgressPlan, EgressRoute
from bot_bottle.git_gate import GitGatePlan, GitGateUpstream from bot_bottle.git_gate import GitGatePlan, GitGateUpstream
from bot_bottle.manifest import GitEntry, Manifest from bot_bottle.manifest import GitEntry, Manifest
@@ -36,6 +38,26 @@ from bot_bottle.supervise import SupervisePlan
from bot_bottle.workspace import workspace_plan from bot_bottle.workspace import workspace_plan
class _Provider(AgentProvider):
"""Minimal concrete subclass for testing the default provision_ca/provision_git."""
@property
def runtime(self) -> AgentProviderRuntime:
return AgentProviderRuntime(
template="test", command="test", image="", dockerfile="",
prompt_mode="append_file", bypass_args=(), resume_args=(),
remote_control_args=(),
)
def provision_plan(self, **kwargs): # type: ignore[override]
raise NotImplementedError
def provision_skills(self, plan, bottle): ... # type: ignore[override]
def provision_prompt(self, plan, bottle): ... # type: ignore[override]
def provision(self, plan, bottle): ... # type: ignore[override]
def provision_supervise_mcp(self, plan, bottle, supervise_url): ... # type: ignore[override]
_PROVIDER = _Provider()
def _make_bottle( def _make_bottle(
name: str = "bot-bottle-demo-abc12", name: str = "bot-bottle-demo-abc12",
exec_result: ExecResult | None = None, exec_result: ExecResult | None = None,
@@ -232,7 +254,7 @@ class TestProvisionCA(unittest.TestCase):
cp_in + exec in the right order.""" cp_in + exec in the right order."""
def setUp(self): def setUp(self):
self._tmp = tempfile.TemporaryDirectory(prefix="cb-prov-ca.") self._tmp = tempfile.TemporaryDirectory(prefix="cb-prov-ca.") # pylint: disable=consider-using-with
self.tmp = Path(self._tmp.name) self.tmp = Path(self._tmp.name)
self.egress_ca = self.tmp / "egress-ca.pem" self.egress_ca = self.tmp / "egress-ca.pem"
_write_self_signed_cert(self.egress_ca) _write_self_signed_cert(self.egress_ca)
@@ -252,10 +274,10 @@ class TestProvisionCA(unittest.TestCase):
def test_egress_ca_always_installed(self): def test_egress_ca_always_installed(self):
plan = _plan(egress_ca_path=self.egress_ca) plan = _plan(egress_ca_path=self.egress_ca)
bottle = _make_bottle(exec_result=self._UPDATE_OK) bottle = _make_bottle(exec_result=self._UPDATE_OK)
_ca.provision_ca(plan, bottle) _PROVIDER.provision_ca(bottle, plan)
bottle.cp_in.assert_called_once_with( bottle.cp_in.assert_called_once_with(
str(self.egress_ca), str(self.egress_ca),
_ca.AGENT_CA_PATH, AGENT_CA_PATH,
) )
bottle.exec.assert_called_once() bottle.exec.assert_called_once()
script = bottle.exec.call_args.args[0] script = bottle.exec.call_args.args[0]
@@ -263,28 +285,59 @@ class TestProvisionCA(unittest.TestCase):
self.assertIn("update-ca-certificates", script) self.assertIn("update-ca-certificates", script)
self.assertEqual("root", bottle.exec.call_args.kwargs.get("user")) self.assertEqual("root", bottle.exec.call_args.kwargs.get("user"))
def test_retries_smolvm_sigkill_during_update_ca(self):
plan = _plan(egress_ca_path=self.egress_ca)
killed = ExecResult(
returncode=137,
stdout="Updating certificates in /etc/ssl/certs...\n",
stderr="",
)
bottle = _make_bottle()
bottle.exec.side_effect = [killed, self._UPDATE_OK]
with patch(
"bot_bottle.backend.smolmachines.provision.ca.time.sleep"
) as sleep:
_ca.provision_ca(plan, bottle)
self.assertEqual(2, bottle.exec.call_count)
sleep.assert_called_once_with(1.0)
def test_dies_when_egress_cert_missing(self): def test_dies_when_egress_cert_missing(self):
plan = _plan(egress_ca_path=self.tmp / "does-not-exist.pem") plan = _plan(egress_ca_path=self.tmp / "does-not-exist.pem")
bottle = _make_bottle() bottle = _make_bottle()
with self.assertRaises(SystemExit): with self.assertRaises(SystemExit):
_ca.provision_ca(plan, bottle) _PROVIDER.provision_ca(bottle, plan)
class TestSmolmachinesBottleExec(unittest.TestCase):
"""SmolmachinesBottle.exec retries once on SIGKILL (exit 137)."""
_SIGKILL = subprocess.CompletedProcess(
args=[], returncode=137, stdout="", stderr="",
)
_SUCCESS = subprocess.CompletedProcess(
args=[], returncode=0, stdout="done", stderr="",
)
def test_retries_on_sigkill(self):
bottle = SmolmachinesBottle("test-machine")
with patch(
"bot_bottle.backend.smolmachines.bottle.subprocess.run",
side_effect=[self._SIGKILL, self._SUCCESS],
) as mock_run, patch(
"bot_bottle.backend.smolmachines.bottle.time.sleep"
) as mock_sleep:
result = bottle.exec("echo hi")
self.assertEqual(0, result.returncode)
self.assertEqual(2, mock_run.call_count)
mock_sleep.assert_called_once_with(1.0)
def test_no_retry_on_success(self):
bottle = SmolmachinesBottle("test-machine")
with patch(
"bot_bottle.backend.smolmachines.bottle.subprocess.run",
return_value=self._SUCCESS,
) as mock_run:
result = bottle.exec("echo hi")
self.assertEqual(0, result.returncode)
self.assertEqual(1, mock_run.call_count)
def test_no_retry_on_other_error(self):
fail = subprocess.CompletedProcess(args=[], returncode=1, stdout="", stderr="err")
bottle = SmolmachinesBottle("test-machine")
with patch(
"bot_bottle.backend.smolmachines.bottle.subprocess.run",
return_value=fail,
) as mock_run:
result = bottle.exec("bad-cmd")
self.assertEqual(1, result.returncode)
self.assertEqual(1, mock_run.call_count)
class TestProvisionGit(unittest.TestCase): class TestProvisionGit(unittest.TestCase):
@@ -293,7 +346,7 @@ class TestProvisionGit(unittest.TestCase):
when its condition doesn't hold.""" when its condition doesn't hold."""
def setUp(self): def setUp(self):
self._tmp = tempfile.TemporaryDirectory(prefix="cb-prov-git.") self._tmp = tempfile.TemporaryDirectory(prefix="cb-prov-git.") # pylint: disable=consider-using-with
self.stage = Path(self._tmp.name) self.stage = Path(self._tmp.name)
def tearDown(self): def tearDown(self):
@@ -301,20 +354,20 @@ class TestProvisionGit(unittest.TestCase):
def test_noop_when_no_cwd_and_no_git_entries(self): def test_noop_when_no_cwd_and_no_git_entries(self):
bottle = _make_bottle() bottle = _make_bottle()
_git.provision_git(_plan(stage_dir=self.stage), bottle) _PROVIDER.provision_git(bottle, _plan(stage_dir=self.stage))
bottle.cp_in.assert_not_called() bottle.cp_in.assert_not_called()
bottle.exec.assert_not_called() bottle.exec.assert_not_called()
def test_copies_cwd_git_when_copy_cwd_and_git_present(self): def test_copies_cwd_git_when_copy_cwd_and_git_present(self):
# Stage a fake host .git dir under user_cwd so the path- # Stage a fake host .git dir under user_cwd so the path-
# check in _provision_cwd_git fires. # check in provision_git fires.
cwd = self.stage / "cwd" cwd = self.stage / "cwd"
(cwd / ".git").mkdir(parents=True) (cwd / ".git").mkdir(parents=True)
plan = _plan( plan = _plan(
copy_cwd=True, user_cwd=str(cwd), stage_dir=self.stage, copy_cwd=True, user_cwd=str(cwd), stage_dir=self.stage,
) )
bottle = _make_bottle() bottle = _make_bottle()
_git.provision_git(plan, bottle) _PROVIDER.provision_git(bottle, plan)
bottle.cp_in.assert_called_once_with( bottle.cp_in.assert_called_once_with(
f"{cwd}/.git", f"{cwd}/.git",
"/home/node/workspace/.git", "/home/node/workspace/.git",
@@ -330,7 +383,7 @@ class TestProvisionGit(unittest.TestCase):
def test_skips_cwd_when_copy_cwd_false(self): def test_skips_cwd_when_copy_cwd_false(self):
plan = _plan(copy_cwd=False, stage_dir=self.stage) plan = _plan(copy_cwd=False, stage_dir=self.stage)
bottle = _make_bottle() bottle = _make_bottle()
_git.provision_git(plan, bottle) _PROVIDER.provision_git(bottle, plan)
bottle.cp_in.assert_not_called() bottle.cp_in.assert_not_called()
def test_writes_gitconfig_with_ip_port_form_for_smolmachines(self): def test_writes_gitconfig_with_ip_port_form_for_smolmachines(self):
@@ -348,13 +401,13 @@ class TestProvisionGit(unittest.TestCase):
agent_git_gate_host="127.0.0.1:9418", agent_git_gate_host="127.0.0.1:9418",
) )
bottle = _make_bottle() bottle = _make_bottle()
_git.provision_git(plan, bottle) _PROVIDER.provision_git(bottle, plan)
# The staged gitconfig path is whatever NamedTemporaryFile # The staged gitconfig path is whatever NamedTemporaryFile
# picked; we read its contents. # picked; we read its contents.
cp_call = bottle.cp_in.call_args cp_call = bottle.cp_in.call_args
staged_path = Path(cp_call.args[0]) staged_path = Path(cp_call.args[0])
self.assertEqual(self.stage, staged_path.parent) self.assertEqual(self.stage, staged_path.parent)
content = staged_path.read_text() content = staged_path.read_text(encoding="utf-8")
self.assertIn( self.assertIn(
'[url "http://127.0.0.1:9418/bot-bottle.git"]', content, '[url "http://127.0.0.1:9418/bot-bottle.git"]', content,
) )
@@ -392,7 +445,7 @@ class TestBundleLaunchSpec(unittest.TestCase):
class TestProvisionGitUser(unittest.TestCase): class TestProvisionGitUser(unittest.TestCase):
"""`_provision_git_user` runs `git config --global` inside the """`provision_git` runs `git config --global` inside the
guest as the node user. SmolmachinesBottle.exec sets HOME and guest as the node user. SmolmachinesBottle.exec sets HOME and
USER automatically for the requested user, so --global lands USER automatically for the requested user, so --global lands
in /home/node/.gitconfig. No-op when the bottle didn't declare in /home/node/.gitconfig. No-op when the bottle didn't declare
@@ -411,7 +464,7 @@ class TestProvisionGitUser(unittest.TestCase):
def test_noop_when_no_git_user(self): def test_noop_when_no_git_user(self):
bottle = _make_bottle() bottle = _make_bottle()
_git._provision_git_user(_plan(), bottle) _PROVIDER.provision_git(bottle, _plan())
self.assertEqual([], self._git_config_calls(bottle)) self.assertEqual([], self._git_config_calls(bottle))
def test_sets_name_and_email_as_node(self): def test_sets_name_and_email_as_node(self):
@@ -420,7 +473,7 @@ class TestProvisionGitUser(unittest.TestCase):
"email": "eric@dideric.is", "email": "eric@dideric.is",
}) })
bottle = _make_bottle() bottle = _make_bottle()
_git._provision_git_user(plan, bottle) _PROVIDER.provision_git(bottle, plan)
calls = self._git_config_calls(bottle) calls = self._git_config_calls(bottle)
self.assertEqual(2, len(calls)) self.assertEqual(2, len(calls))
# Both run as node so SmolmachinesBottle.exec sets HOME=/home/node # Both run as node so SmolmachinesBottle.exec sets HOME=/home/node
@@ -436,7 +489,7 @@ class TestProvisionGitUser(unittest.TestCase):
def test_name_only(self): def test_name_only(self):
plan = _plan(git_user={"name": "Bot"}) plan = _plan(git_user={"name": "Bot"})
bottle = _make_bottle() bottle = _make_bottle()
_git._provision_git_user(plan, bottle) _PROVIDER.provision_git(bottle, plan)
calls = self._git_config_calls(bottle) calls = self._git_config_calls(bottle)
self.assertEqual(1, len(calls)) self.assertEqual(1, len(calls))
self.assertIn("user.name", calls[0][0]) self.assertIn("user.name", calls[0][0])
@@ -445,7 +498,7 @@ class TestProvisionGitUser(unittest.TestCase):
def test_email_only(self): def test_email_only(self):
plan = _plan(git_user={"email": "bot@example.com"}) plan = _plan(git_user={"email": "bot@example.com"})
bottle = _make_bottle() bottle = _make_bottle()
_git._provision_git_user(plan, bottle) _PROVIDER.provision_git(bottle, plan)
calls = self._git_config_calls(bottle) calls = self._git_config_calls(bottle)
self.assertEqual(1, len(calls)) self.assertEqual(1, len(calls))
self.assertIn("user.email", calls[0][0]) self.assertIn("user.email", calls[0][0])
@@ -454,7 +507,7 @@ class TestProvisionGitUser(unittest.TestCase):
class TestProvisionWorkspace(unittest.TestCase): class TestProvisionWorkspace(unittest.TestCase):
def setUp(self): def setUp(self):
self._tmp = tempfile.TemporaryDirectory(prefix="cb-prov-workspace.") self._tmp = tempfile.TemporaryDirectory(prefix="cb-prov-workspace.") # pylint: disable=consider-using-with
self.stage = Path(self._tmp.name) self.stage = Path(self._tmp.name)
def tearDown(self): def tearDown(self):
+12 -14
View File
@@ -17,7 +17,6 @@ from bot_bottle.supervise import (
STATUS_MODIFIED, STATUS_MODIFIED,
STATUS_REJECTED, STATUS_REJECTED,
TOOL_CAPABILITY_BLOCK, TOOL_CAPABILITY_BLOCK,
TOOL_EGRESS_BLOCK,
archive_proposal, archive_proposal,
audit_log_path, audit_log_path,
list_pending_proposals, list_pending_proposals,
@@ -37,16 +36,16 @@ FIXED_TS = datetime(2026, 5, 25, 12, 0, 0, tzinfo=timezone.utc)
def _proposal( def _proposal(
tool: str = TOOL_EGRESS_BLOCK, tool: str = TOOL_CAPABILITY_BLOCK,
proposed: str = "{}", proposed: str = "FROM python:3.13\n",
justification: str = "need a route", justification: str = "need a capability",
) -> Proposal: ) -> Proposal:
return Proposal.new( return Proposal.new(
bottle_slug="dev", bottle_slug="dev",
tool=tool, tool=tool,
proposed_file=proposed, proposed_file=proposed,
justification=justification, justification=justification,
current_file_hash=sha256_hex("{}"), current_file_hash=sha256_hex(proposed),
now=FIXED_TS, now=FIXED_TS,
) )
@@ -57,7 +56,7 @@ class TestProposalRoundtrip(unittest.TestCase):
self.assertTrue(p.id) self.assertTrue(p.id)
self.assertEqual("2026-05-25T12:00:00+00:00", p.arrival_timestamp) self.assertEqual("2026-05-25T12:00:00+00:00", p.arrival_timestamp)
self.assertEqual("dev", p.bottle_slug) self.assertEqual("dev", p.bottle_slug)
self.assertEqual(TOOL_EGRESS_BLOCK, p.tool) self.assertEqual(TOOL_CAPABILITY_BLOCK, p.tool)
def test_to_from_dict_roundtrip(self): def test_to_from_dict_roundtrip(self):
p = _proposal() p = _proposal()
@@ -142,14 +141,14 @@ class TestQueueIO(unittest.TestCase):
def test_list_pending_sorted_by_arrival(self): def test_list_pending_sorted_by_arrival(self):
# Fabricate two with explicit timestamps. # Fabricate two with explicit timestamps.
a = Proposal.new( a = Proposal.new(
bottle_slug="dev", tool=TOOL_EGRESS_BLOCK, bottle_slug="dev", tool=TOOL_CAPABILITY_BLOCK,
proposed_file="{}", justification="early", proposed_file="FROM python:3.13\n", justification="early",
current_file_hash="x", current_file_hash="x",
now=datetime(2026, 5, 25, 10, 0, 0, tzinfo=timezone.utc), now=datetime(2026, 5, 25, 10, 0, 0, tzinfo=timezone.utc),
) )
b = Proposal.new( b = Proposal.new(
bottle_slug="dev", tool=TOOL_EGRESS_BLOCK, bottle_slug="dev", tool=TOOL_CAPABILITY_BLOCK,
proposed_file="{}", justification="late", proposed_file="FROM python:3.13\n", justification="late",
current_file_hash="x", current_file_hash="x",
now=datetime(2026, 5, 25, 14, 0, 0, tzinfo=timezone.utc), now=datetime(2026, 5, 25, 14, 0, 0, tzinfo=timezone.utc),
) )
@@ -318,16 +317,15 @@ class TestToolConstants(unittest.TestCase):
def test_tools_tuple_matches_individual_constants(self): def test_tools_tuple_matches_individual_constants(self):
self.assertEqual( self.assertEqual(
( (
TOOL_EGRESS_BLOCK,
TOOL_CAPABILITY_BLOCK, TOOL_CAPABILITY_BLOCK,
supervise.TOOL_LIST_EGRESS_ROUTES, supervise.TOOL_LIST_EGRESS_ROUTES,
), ),
supervise.TOOLS, supervise.TOOLS,
) )
def test_component_map_covers_egress_remediation_only(self): def test_component_map_has_no_entries(self):
self.assertIn(TOOL_EGRESS_BLOCK, supervise.COMPONENT_FOR_TOOL) # egress-block removed in issue #198; capability-block never had one.
self.assertNotIn(TOOL_CAPABILITY_BLOCK, supervise.COMPONENT_FOR_TOOL) self.assertEqual({}, supervise.COMPONENT_FOR_TOOL)
class _StubSupervise(supervise.Supervise): class _StubSupervise(supervise.Supervise):
+18 -152
View File
@@ -1,12 +1,10 @@
"""Unit: supervise headless paths (PRD 0013 phase 4, PRD 0014). """Unit: supervise headless paths (PRD 0013 phase 4, PRD 0016).
The curses TUI itself isn't exercised here — these tests cover the The curses TUI itself isn't exercised here — these tests cover the
discovery + approve/reject + audit-write paths that the TUI's key discovery + approve/reject paths that the TUI's key handlers call into.
handlers call into.
add_route is stubbed at the supervise CLI module level so the tests egress-block (add_route) was removed in issue #198; the TestEgressApplyWiring
don't need a running egress sidecar; the real docker exec/cp/SIGHUP class and all stubs for add_route have been dropped accordingly.
plumbing is covered by the integration test.
""" """
import os import os
@@ -17,7 +15,6 @@ from pathlib import Path
from bot_bottle import supervise from bot_bottle import supervise
from bot_bottle.backend.docker.capability_apply import CapabilityApplyError from bot_bottle.backend.docker.capability_apply import CapabilityApplyError
from bot_bottle.backend.docker.egress_apply import EgressApplyError
from bot_bottle.cli import supervise as supervise_cli from bot_bottle.cli import supervise as supervise_cli
from bot_bottle.supervise import ( from bot_bottle.supervise import (
Proposal, Proposal,
@@ -25,7 +22,6 @@ from bot_bottle.supervise import (
STATUS_MODIFIED, STATUS_MODIFIED,
STATUS_REJECTED, STATUS_REJECTED,
TOOL_CAPABILITY_BLOCK, TOOL_CAPABILITY_BLOCK,
TOOL_EGRESS_BLOCK,
read_audit_entries, read_audit_entries,
read_response, read_response,
sha256_hex, sha256_hex,
@@ -35,9 +31,8 @@ from bot_bottle.supervise import (
FIXED = datetime(2026, 5, 25, 12, 0, 0, tzinfo=timezone.utc) FIXED = datetime(2026, 5, 25, 12, 0, 0, tzinfo=timezone.utc)
def _proposal(slug: str = "dev", tool: str = TOOL_EGRESS_BLOCK) -> Proposal: def _proposal(slug: str = "dev", tool: str = TOOL_CAPABILITY_BLOCK) -> Proposal:
payloads = { payloads = {
TOOL_EGRESS_BLOCK: '{"routes": []}\n',
TOOL_CAPABILITY_BLOCK: "FROM python:3.13\n", TOOL_CAPABILITY_BLOCK: "FROM python:3.13\n",
} }
payload = payloads.get(tool, "") payload = payloads.get(tool, "")
@@ -88,14 +83,14 @@ class TestDiscoverPending(_FakeHomeMixin, unittest.TestCase):
def test_sorted_by_arrival_across_bottles(self): def test_sorted_by_arrival_across_bottles(self):
early = Proposal.new( early = Proposal.new(
bottle_slug="api", tool=TOOL_EGRESS_BLOCK, bottle_slug="api", tool=TOOL_CAPABILITY_BLOCK,
proposed_file="{}", justification="early", proposed_file="FROM python:3.13\n", justification="early",
current_file_hash="h", current_file_hash="h",
now=datetime(2026, 5, 25, 10, 0, 0, tzinfo=timezone.utc), now=datetime(2026, 5, 25, 10, 0, 0, tzinfo=timezone.utc),
) )
late = Proposal.new( late = Proposal.new(
bottle_slug="dev", tool=TOOL_EGRESS_BLOCK, bottle_slug="dev", tool=TOOL_CAPABILITY_BLOCK,
proposed_file="{}", justification="late", proposed_file="FROM python:3.13\n", justification="late",
current_file_hash="h", current_file_hash="h",
now=datetime(2026, 5, 25, 14, 0, 0, tzinfo=timezone.utc), now=datetime(2026, 5, 25, 14, 0, 0, tzinfo=timezone.utc),
) )
@@ -120,48 +115,38 @@ class TestDiscoverPending(_FakeHomeMixin, unittest.TestCase):
class TestApproveReject(_FakeHomeMixin, unittest.TestCase): class TestApproveReject(_FakeHomeMixin, unittest.TestCase):
def setUp(self): def setUp(self):
self._setup_fake_home() self._setup_fake_home()
self._original_add_route = supervise_cli.add_route
self._original_apply_capability = supervise_cli.apply_capability_change 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.
supervise_cli.add_route = lambda slug, content: ( # type: ignore
'{"routes": []}\n', '{"routes": [{"host": "x"}]}\n',
)
supervise_cli.apply_capability_change = lambda slug, content: ( # type: ignore supervise_cli.apply_capability_change = lambda slug, content: ( # type: ignore
"FROM old\n", content, "FROM old\n", content,
) )
def tearDown(self): def tearDown(self):
supervise_cli.add_route = self._original_add_route
supervise_cli.apply_capability_change = self._original_apply_capability supervise_cli.apply_capability_change = self._original_apply_capability
self._teardown_fake_home() self._teardown_fake_home()
def _enqueue(self, tool: str = TOOL_EGRESS_BLOCK): def _enqueue(self, tool: str = TOOL_CAPABILITY_BLOCK):
p = _proposal(tool=tool) p = _proposal(tool=tool)
qdir = supervise.queue_dir_for_slug("dev") qdir = supervise.queue_dir_for_slug("dev")
qdir.mkdir(parents=True, exist_ok=True) qdir.mkdir(parents=True, exist_ok=True)
supervise.write_proposal(qdir, p) supervise.write_proposal(qdir, p)
return supervise_cli.QueuedProposal(proposal=p, queue_dir=qdir) return supervise_cli.QueuedProposal(proposal=p, queue_dir=qdir)
def test_approve_writes_response_and_audit(self): def test_approve_writes_response(self):
qp = self._enqueue() qp = self._enqueue()
supervise_cli.approve(qp) supervise_cli.approve(qp)
resp = read_response(qp.queue_dir, qp.proposal.id) # capability-block is archived on approve, so the response file
# moves to processed/ before the caller can read it.
resp = read_response(qp.queue_dir / "processed", qp.proposal.id)
self.assertEqual(STATUS_APPROVED, resp.status) self.assertEqual(STATUS_APPROVED, resp.status)
self.assertIsNone(resp.final_file) self.assertIsNone(resp.final_file)
entries = read_audit_entries("egress", "dev")
self.assertEqual(1, len(entries))
self.assertEqual("approved", entries[0].operator_action)
def test_approve_with_final_file_marks_modified(self): def test_approve_with_final_file_marks_modified(self):
qp = self._enqueue() qp = self._enqueue()
supervise_cli.approve(qp, final_file='{"routes": [{"path": "/x/"}]}\n', notes="tweaked") supervise_cli.approve(qp, final_file="FROM bookworm\n", notes="tweaked")
resp = read_response(qp.queue_dir, qp.proposal.id) resp = read_response(qp.queue_dir / "processed", qp.proposal.id)
self.assertEqual(STATUS_MODIFIED, resp.status) self.assertEqual(STATUS_MODIFIED, resp.status)
self.assertEqual('{"routes": [{"path": "/x/"}]}\n', resp.final_file) self.assertEqual("FROM bookworm\n", resp.final_file)
self.assertEqual("tweaked", resp.notes) self.assertEqual("tweaked", resp.notes)
entries = read_audit_entries("egress", "dev")
self.assertEqual("modified", entries[0].operator_action)
def test_reject_writes_rejection(self): def test_reject_writes_rejection(self):
qp = self._enqueue() qp = self._enqueue()
@@ -169,113 +154,13 @@ class TestApproveReject(_FakeHomeMixin, unittest.TestCase):
resp = read_response(qp.queue_dir, qp.proposal.id) resp = read_response(qp.queue_dir, qp.proposal.id)
self.assertEqual(STATUS_REJECTED, resp.status) self.assertEqual(STATUS_REJECTED, resp.status)
self.assertEqual("nope", resp.notes) self.assertEqual("nope", resp.notes)
entries = read_audit_entries("egress", "dev")
self.assertEqual("rejected", entries[0].operator_action)
self.assertEqual("nope", entries[0].operator_notes)
def test_capability_block_skips_audit_log(self): def test_no_audit_log_for_capability_block(self):
qp = self._enqueue(tool=TOOL_CAPABILITY_BLOCK) qp = self._enqueue(tool=TOOL_CAPABILITY_BLOCK)
supervise_cli.approve(qp) supervise_cli.approve(qp)
# No audit log for capability-block (per PRD 0013 / 0016).
self.assertEqual([], read_audit_entries("egress", "dev")) self.assertEqual([], read_audit_entries("egress", "dev"))
class TestEgressApplyWiring(_FakeHomeMixin, unittest.TestCase):
"""PRD 0017 chunk 3: approve() on an egress-block proposal
must call add_route (single-route merge) with the right args
and surface its failures."""
def setUp(self):
self._setup_fake_home()
self._original_add_route = supervise_cli.add_route
def tearDown(self):
supervise_cli.add_route = self._original_add_route
self._teardown_fake_home()
def _enqueue_egress(self, proposed: str = '{"host": "x.example"}\n'):
p = Proposal.new(
bottle_slug="dev", tool=TOOL_EGRESS_BLOCK,
proposed_file=proposed,
justification="need a route",
current_file_hash=sha256_hex(proposed),
now=FIXED,
)
qdir = supervise.queue_dir_for_slug("dev")
qdir.mkdir(parents=True, exist_ok=True)
supervise.write_proposal(qdir, p)
return supervise_cli.QueuedProposal(proposal=p, queue_dir=qdir)
def test_egress_block_calls_add_route_with_proposed_json(self):
calls = []
supervise_cli.add_route = lambda slug, content: ( # type: ignore
calls.append((slug, content)) or ("before", "after")
)
qp = self._enqueue_egress(
proposed='{"host": "new.example", "path_allowlist": ["/x/"]}\n'
)
supervise_cli.approve(qp)
self.assertEqual(1, len(calls))
slug, content = calls[0]
self.assertEqual("dev", slug)
# The single-route JSON the agent proposed reaches add_route
# unchanged — add_route fetches current state + merges.
self.assertEqual(
'{"host": "new.example", "path_allowlist": ["/x/"]}\n',
content,
)
def test_modify_passes_final_file_to_add_route(self):
calls = []
supervise_cli.add_route = lambda slug, content: ( # type: ignore
calls.append(content) or ("before", "after")
)
qp = self._enqueue_egress()
supervise_cli.approve(
qp,
final_file='{"host": "edited.example"}\n',
notes="tweaked",
)
self.assertEqual(['{"host": "edited.example"}\n'], calls)
def test_apply_failure_blocks_response_and_audit(self):
supervise_cli.add_route = lambda slug, content: (_ for _ in ()).throw( # type: ignore
EgressApplyError("docker exec failed")
)
qp = self._enqueue_egress()
with self.assertRaises(EgressApplyError):
supervise_cli.approve(qp)
# No response file (proposal stays pending).
self.assertEqual(
[qp.proposal.id],
[p.id for p in supervise.list_pending_proposals(qp.queue_dir)],
)
# No audit entry.
self.assertEqual([], read_audit_entries("egress", "dev"))
def test_real_diff_lands_in_audit(self):
supervise_cli.add_route = lambda slug, content: ( # type: ignore
'{"routes": []}\n', # before
'{"routes": [{"host": "new.example"}]}\n', # after
)
qp = self._enqueue_egress(proposed='{"host": "new.example"}\n')
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):
qp = self._enqueue_egress()
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)
entries = read_audit_entries("egress", "dev")
self.assertEqual(1, len(entries))
self.assertEqual("", entries[0].diff)
class TestCapabilityApplyWiring(_FakeHomeMixin, unittest.TestCase): class TestCapabilityApplyWiring(_FakeHomeMixin, unittest.TestCase):
"""PRD 0016 Phase 3: approve() on a capability-block proposal """PRD 0016 Phase 3: approve() on a capability-block proposal
calls apply_capability_change, archives the proposal afterward calls apply_capability_change, archives the proposal afterward
@@ -328,17 +213,12 @@ class TestCapabilityApplyWiring(_FakeHomeMixin, unittest.TestCase):
supervise_cli.apply_capability_change = lambda slug, content: ("FROM old\n", content) # type: ignore supervise_cli.apply_capability_change = lambda slug, content: ("FROM old\n", content) # type: ignore
qp = self._enqueue_capability() qp = self._enqueue_capability()
supervise_cli.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("egress", "dev"))
def test_proposal_archived_after_apply(self): def test_proposal_archived_after_apply(self):
supervise_cli.apply_capability_change = lambda slug, content: ("FROM old\n", content) # type: ignore supervise_cli.apply_capability_change = lambda slug, content: ("FROM old\n", content) # type: ignore
qp = self._enqueue_capability() qp = self._enqueue_capability()
supervise_cli.approve(qp) supervise_cli.approve(qp)
# Sidecar would normally archive after delivering the response,
# 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)) self.assertEqual([], supervise.list_pending_proposals(qp.queue_dir))
processed = list((qp.queue_dir / "processed").glob("*.json")) processed = list((qp.queue_dir / "processed").glob("*.json"))
self.assertEqual(2, len(processed)) self.assertEqual(2, len(processed))
@@ -346,20 +226,8 @@ class TestCapabilityApplyWiring(_FakeHomeMixin, unittest.TestCase):
class TestEditInEditor(unittest.TestCase): class TestEditInEditor(unittest.TestCase):
def test_runs_editor_returns_edited_content(self): def test_runs_editor_returns_edited_content(self):
# Fake "editor" is /bin/sh -c 'cat <<EOF > $1 ... EOF'
original_editor = os.environ.get("EDITOR") original_editor = os.environ.get("EDITOR")
try: try:
# Use a fake editor that overwrites the file with a known
# marker. EDITOR is split with shlex equivalence by
# subprocess.run when invoked as a list — keep it as a
# single program path that takes the file as argv[1].
os.environ["EDITOR"] = (
"/bin/sh -c 'printf %s \"edited\" > \"$0\"'"
)
# subprocess.run with the str as the first list element
# would try to find a binary literally named "/bin/sh -c ..."
# — that won't work. Use shell mode trick: wrap in a script.
# Easier: build a tiny helper script.
with tempfile.NamedTemporaryFile( with tempfile.NamedTemporaryFile(
mode="w", suffix=".sh", delete=False, prefix="fake-editor.", mode="w", suffix=".sh", delete=False, prefix="fake-editor.",
) as script: ) as script:
@@ -381,7 +249,6 @@ class TestEditInEditor(unittest.TestCase):
def test_returns_none_when_unchanged(self): def test_returns_none_when_unchanged(self):
original_editor = os.environ.get("EDITOR") original_editor = os.environ.get("EDITOR")
try: try:
# No-op editor: touch the file (leaves it unchanged).
with tempfile.NamedTemporaryFile( with tempfile.NamedTemporaryFile(
mode="w", suffix=".sh", delete=False, prefix="noop-editor.", mode="w", suffix=".sh", delete=False, prefix="noop-editor.",
) as script: ) as script:
@@ -445,7 +312,6 @@ class TestCapabilityBlockSmolmachinesGuard(_FakeHomeMixin, unittest.TestCase):
supervise_cli.approve(qp) # must not raise supervise_cli.approve(qp) # must not raise
def test_no_metadata_falls_through_to_docker_path(self): def test_no_metadata_falls_through_to_docker_path(self):
# No metadata at all → assume Docker (backward-compatible).
qp = self._enqueue_capability("dev") qp = self._enqueue_capability("dev")
supervise_cli.approve(qp) # must not raise supervise_cli.approve(qp) # must not raise
+12 -12
View File
@@ -141,7 +141,6 @@ class TestHandleToolsList(unittest.TestCase):
names = [t["name"] for t in result["tools"]] # type: ignore[index] names = [t["name"] for t in result["tools"]] # type: ignore[index]
self.assertEqual( self.assertEqual(
sorted([ sorted([
_sv.TOOL_EGRESS_BLOCK,
_sv.TOOL_CAPABILITY_BLOCK, _sv.TOOL_CAPABILITY_BLOCK,
_sv.TOOL_LIST_EGRESS_ROUTES, _sv.TOOL_LIST_EGRESS_ROUTES,
]), ]),
@@ -206,10 +205,10 @@ class TestHandleToolsCall(unittest.TestCase):
try: try:
result = handle_tools_call( result = handle_tools_call(
{ {
"name": _sv.TOOL_EGRESS_BLOCK, "name": _sv.TOOL_CAPABILITY_BLOCK,
"arguments": { "arguments": {
"host": "example.com", "dockerfile": "FROM python:3.13\n",
"justification": "need a route", "justification": "need git",
}, },
}, },
self.config, self.config,
@@ -250,8 +249,8 @@ class TestHandleToolsCall(unittest.TestCase):
with self.assertRaises(_RpcError): with self.assertRaises(_RpcError):
handle_tools_call( handle_tools_call(
{ {
"name": _sv.TOOL_EGRESS_BLOCK, "name": _sv.TOOL_CAPABILITY_BLOCK,
"arguments": {"host": "example.com"}, "arguments": {"dockerfile": "FROM python:3.13\n"},
}, },
self.config, self.config,
) )
@@ -261,9 +260,9 @@ class TestHandleToolsCall(unittest.TestCase):
try: try:
handle_tools_call( handle_tools_call(
{ {
"name": _sv.TOOL_EGRESS_BLOCK, "name": _sv.TOOL_CAPABILITY_BLOCK,
"arguments": { "arguments": {
"host": "example.com", "dockerfile": "FROM python:3.13\n",
"justification": "x", "justification": "x",
}, },
}, },
@@ -285,10 +284,10 @@ class TestHandleToolsCall(unittest.TestCase):
) )
result = handle_tools_call( result = handle_tools_call(
{ {
"name": _sv.TOOL_EGRESS_BLOCK, "name": _sv.TOOL_CAPABILITY_BLOCK,
"arguments": { "arguments": {
"host": "example.com", "dockerfile": "FROM python:3.13\n",
"justification": "need a route", "justification": "need a capability",
}, },
}, },
config, config,
@@ -412,7 +411,8 @@ class TestHttpEndToEnd(unittest.TestCase):
self.assertEqual("2.0", result["jsonrpc"]) self.assertEqual("2.0", result["jsonrpc"])
self.assertEqual(1, result["id"]) self.assertEqual(1, result["id"])
names = [t["name"] for t in result["result"]["tools"]] # type: ignore[index] names = [t["name"] for t in result["result"]["tools"]] # type: ignore[index]
self.assertIn(_sv.TOOL_EGRESS_BLOCK, names) self.assertIn(_sv.TOOL_CAPABILITY_BLOCK, names)
self.assertNotIn("egress-block", names)
def test_unknown_method_returns_jsonrpc_error(self): def test_unknown_method_returns_jsonrpc_error(self):
result = self._post_jsonrpc( result = self._post_jsonrpc(