Compare commits

..

1 Commits

Author SHA1 Message Date
didericis-claude 1bec3559ab docs(prd): draft PRD for separate agent/bottle selection
Closes #269.
2026-06-25 06:45:58 +00:00
31 changed files with 371 additions and 356 deletions
-9
View File
@@ -1,9 +0,0 @@
[run]
branch = True
source = .
[report]
omit =
bot_bottle/egress_addon.py
bot_bottle/cli/tui.py
tests/*
+1 -7
View File
@@ -39,14 +39,8 @@ jobs:
with:
python-version: "3.12"
- name: Install dev requirements
run: python3 -m pip install -r requirements-dev.txt
- name: Run unit tests
run: python3 -m coverage run -m unittest discover -t . -s tests/unit -v
- name: Report unit coverage
run: python3 -m coverage report -m
run: python3 -m unittest discover -t . -s tests/unit -v
integration:
runs-on: ubuntu-latest
-1
View File
@@ -22,4 +22,3 @@ venv/
.pytest_cache/
.mypy_cache/
.ruff_cache/
.coverage
+2 -1
View File
@@ -61,6 +61,7 @@ class AgentProviderRuntime:
prompt_mode: PromptMode
bypass_args: tuple[str, ...]
resume_args: tuple[str, ...]
remote_control_args: tuple[str, ...]
@dataclass(frozen=True)
@@ -390,7 +391,7 @@ def prompt_args(
if prompt_mode == "append_file":
return ["--append-system-prompt-file", prompt_path]
if prompt_mode == "read_prompt_file":
if argv and ("resume" in argv or "remote-control" in argv):
if argv and "resume" in argv:
return []
return [f"Read and follow the instructions in {prompt_path}."]
if prompt_mode == "print_read_prompt_file":
+2 -1
View File
@@ -109,8 +109,9 @@ class BottlePlan(ABC):
def workspace_plan(self) -> WorkspacePlan:
return workspace_plan(self.spec, guest_home=self.guest_home)
def print(self) -> None:
def print(self, *, remote_control: bool) -> None:
"""Render the y/N preflight summary to stderr."""
del remote_control
spec = self.spec
manifest = self.manifest
agent = manifest.agent
+2 -6
View File
@@ -11,7 +11,7 @@ from pathlib import Path
from ..bottle_state import egress_state_dir
from ..egress import EGRESS_ROUTES_FILENAME
from ..egress_addon_core import LOG_OFF, load_config
from ..egress_addon_core import load_routes
class EgressApplyError(RuntimeError):
@@ -33,15 +33,11 @@ class EgressApplicator(ABC):
@staticmethod
def validate_routes_content(content: str) -> None:
try:
config = load_config(content)
load_routes(content)
except ValueError as e:
raise EgressApplyError(
f"proposed routes.yaml is not valid: {e}"
) from e
if config.log != LOG_OFF:
raise EgressApplyError(
"proposed routes.yaml must not change egress logging"
)
@staticmethod
def _routes_path(slug: str) -> Path:
@@ -68,11 +68,6 @@ def build_image(ref: str, context: str, *, dockerfile: str = "") -> None:
_ensure_builder_dns()
args = [_CONTAINER, "build", "-t", ref, "--dns", dns_server()]
if dockerfile:
# `container build` resolves -f relative to the current working
# directory, not the build context. Anchor a relative Dockerfile to
# the context so builds work from any cwd.
if not os.path.isabs(dockerfile):
dockerfile = os.path.join(context, dockerfile)
args.extend(["-f", dockerfile])
args.append(context)
subprocess.run(args, check=True)
+2
View File
@@ -28,6 +28,7 @@ from .start import _launch_bottle
def cmd_resume(argv: list[str]) -> int:
parser = argparse.ArgumentParser(prog=f"{PROG} resume", add_help=True)
parser.add_argument("--dry-run", action="store_true")
parser.add_argument("--remote-control", action="store_true")
parser.add_argument(
"identity",
help="bottle identity from a prior `start` (see its session-end output)",
@@ -55,5 +56,6 @@ def cmd_resume(argv: list[str]) -> int:
return _launch_bottle(
spec,
dry_run=args.dry_run,
remote_control=args.remote_control,
backend_name=backend_name,
)
+10 -4
View File
@@ -42,6 +42,7 @@ def cmd_start(argv: list[str]) -> int:
parser = argparse.ArgumentParser(prog=f"{PROG} start", add_help=True)
parser.add_argument("--dry-run", action="store_true")
parser.add_argument("--cwd", action="store_true", help="copy host cwd into the running bottle")
parser.add_argument("--remote-control", action="store_true")
parser.add_argument(
"--backend",
choices=known_backend_names(),
@@ -88,6 +89,7 @@ def cmd_start(argv: list[str]) -> int:
return _launch_bottle(
spec,
dry_run=dry_run,
remote_control=args.remote_control,
backend_name=backend_name,
)
@@ -132,7 +134,7 @@ def prepare_with_preflight(
def attach_agent(
bottle: Bottle, *, resume: bool = False,
bottle: Bottle, *, remote_control: bool = False, resume: bool = False,
agent_provider_template: str = "claude",
startup_args: tuple[str, ...] = (),
) -> int:
@@ -151,6 +153,8 @@ def attach_agent(
"(Ctrl-D or 'exit' to leave; container will be removed)"
)
agent_args = list(runtime.bypass_args)
if remote_control:
agent_args.extend(runtime.remote_control_args)
agent_args.extend(startup_args)
if resume:
agent_args.extend(runtime.resume_args)
@@ -214,9 +218,9 @@ def _text_prompt_yes() -> bool:
return reply in ("y", "Y", "yes", "YES")
def _text_render_preflight():
def _text_render_preflight(*, remote_control: bool):
def _render(plan: DockerBottlePlan) -> None:
plan.print()
plan.print(remote_control=remote_control)
return _render
@@ -224,6 +228,7 @@ def _launch_bottle(
spec: BottleSpec,
*,
dry_run: bool,
remote_control: bool,
backend_name: str | None = None,
) -> int:
"""Shared launch core for `start` and `resume`. Builds the plan,
@@ -235,7 +240,7 @@ def _launch_bottle(
plan, identity = prepare_with_preflight(
spec,
stage_dir=stage_dir,
render_preflight=_text_render_preflight(),
render_preflight=_text_render_preflight(remote_control=remote_control),
prompt_yes=_text_prompt_yes,
dry_run=dry_run,
backend_name=backend_name,
@@ -248,6 +253,7 @@ def _launch_bottle(
agent_provider_template = getattr(plan, "agent_provider_template", "claude")
exit_code = attach_agent(
bottle,
remote_control=remote_control,
agent_provider_template=agent_provider_template,
startup_args=plan.agent_provision.startup_args,
)
+36 -9
View File
@@ -2,8 +2,9 @@
act on them (approve / modify / reject).
Curses-based TUI; modify-then-approve shells out to $EDITOR. The
Egress proposals are queued for operator review as full routes.yaml
updates.
approval handler wires to PRD 0016 (capability-block), which rebuilds
the bottle Dockerfile. Egress proposals are queued for operator review
as full routes.yaml updates.
"""
from __future__ import annotations
@@ -21,6 +22,10 @@ from pathlib import Path
from .. import supervise as _supervise
from ..bottle_state import read_metadata
# from ..backend.docker.capability_apply import (
# CapabilityApplyError,
# apply_capability_change,
# )
from ..backend.docker.egress_apply import (
EgressApplyError,
applicator as _docker_applicator,
@@ -33,6 +38,10 @@ from ..backend.smolmachines.egress_apply import (
)
from ..log import Die, error, info
class CapabilityApplyError(RuntimeError):
"""Placeholder while capability_apply is disabled."""
from ..supervise import (
COMPONENT_FOR_TOOL,
AuditEntry,
@@ -41,10 +50,12 @@ from ..supervise import (
STATUS_APPROVED,
STATUS_MODIFIED,
STATUS_REJECTED,
TOOL_CAPABILITY_BLOCK,
TOOL_EGRESS_ALLOW,
TOOL_EGRESS_BLOCK,
TOOL_GITLEAKS_ALLOW,
TOOL_EGRESS_TOKEN_ALLOW,
archive_proposal,
list_pending_proposals,
render_diff,
write_audit_entry,
@@ -72,7 +83,7 @@ class QueuedProposal:
# Errors any remediation engine may raise. Caught by the TUI key
# handlers and surfaced in the status line so a failed apply keeps
# the proposal pending rather than crashing curses.
ApplyError = (EgressApplyError,)
ApplyError = (CapabilityApplyError, EgressApplyError)
def apply_routes_change(slug: str, content: str) -> tuple[str, str]:
@@ -132,6 +143,8 @@ def _detail_lines(
def _suffix_for_tool(tool: str) -> str:
if tool == TOOL_CAPABILITY_BLOCK:
return ".dockerfile"
if tool in (TOOL_EGRESS_ALLOW, TOOL_EGRESS_BLOCK):
return ".yaml"
if tool in (TOOL_GITLEAKS_ALLOW, TOOL_EGRESS_TOKEN_ALLOW):
@@ -153,6 +166,17 @@ def approve(
file_to_apply = final_file if final_file is not None else qp.proposal.proposed_file
diff_before, diff_after = "", ""
# if qp.proposal.tool == TOOL_CAPABILITY_BLOCK:
# _meta = read_metadata(qp.proposal.bottle_slug)
# if _meta is not None and not _meta.compose_project:
# raise CapabilityApplyError(
# "capability-block remediation is not supported for smolmachines "
# "bottles. Reject this proposal or handle the capability change "
# "manually, then restart the bottle."
# )
# diff_before, diff_after = apply_capability_change(
# qp.proposal.bottle_slug, file_to_apply,
# )
if qp.proposal.tool in (TOOL_EGRESS_ALLOW, TOOL_EGRESS_BLOCK):
diff_before, diff_after = apply_routes_change(
qp.proposal.bottle_slug,
@@ -170,6 +194,9 @@ def approve(
qp, action=status, notes=notes,
diff_before=diff_before, diff_after=diff_after,
)
if qp.proposal.tool == TOOL_CAPABILITY_BLOCK:
archive_proposal(qp.queue_dir, qp.proposal.id)
def reject(qp: QueuedProposal, *, reason: str) -> None:
"""Write a rejection response and an audit entry."""
@@ -319,7 +346,7 @@ def _list_once() -> int:
return 0
def _try_init_green() -> int: # pragma: no cover
def _try_init_green() -> int:
"""Initialise a green color pair and return its attr, or 0."""
try:
curses.start_color()
@@ -330,7 +357,7 @@ def _try_init_green() -> int: # pragma: no cover
return 0
def _main_loop(stdscr: "curses._CursesWindow") -> None: # type: ignore # pragma: no cover
def _main_loop(stdscr: "curses._CursesWindow") -> None: # type: ignore
curses.curs_set(0)
stdscr.timeout(_REFRESH_INTERVAL_MS)
green_attr = _try_init_green()
@@ -420,7 +447,7 @@ def _render(
status_line: str,
*,
green_attr: int = 0, # noqa: F841 — unused, but required by interface
) -> None: # pragma: no cover
) -> None:
stdscr.erase()
h, w = stdscr.getmaxyx()
header = f"bot-bottle supervise ({len(pending)} pending)"
@@ -471,7 +498,7 @@ def _detail_view(
qp: QueuedProposal,
*,
green_attr: int = 0,
) -> None: # pragma: no cover
) -> None:
"""Render the full proposal. Scrollable. Press q to return."""
lines = _detail_lines(qp, green_attr=green_attr)
offset = 0
@@ -523,7 +550,7 @@ def _detail_view(
return
def _modify(stdscr: "curses._CursesWindow", qp: QueuedProposal) -> str | None: # type: ignore # pragma: no cover
def _modify(stdscr: "curses._CursesWindow", qp: QueuedProposal) -> str | None: # type: ignore
"""Suspend curses, open $EDITOR on the proposed file, return edited content."""
suffix = _suffix_for_tool(qp.proposal.tool)
curses.endwin()
@@ -534,7 +561,7 @@ def _modify(stdscr: "curses._CursesWindow", qp: QueuedProposal) -> str | None:
return edited
def _prompt(stdscr: "curses._CursesWindow", label: str) -> str: # type: ignore # pragma: no cover
def _prompt(stdscr: "curses._CursesWindow", label: str) -> str: # type: ignore
"""One-line input at the bottom of the screen."""
curses.curs_set(1)
h, _ = stdscr.getmaxyx()
@@ -91,6 +91,7 @@ _RUNTIME = AgentProviderRuntime(
prompt_mode="append_file",
bypass_args=("--dangerously-skip-permissions",),
resume_args=("--continue",),
remote_control_args=("--remote-control",),
)
+6 -9
View File
@@ -1,12 +1,12 @@
# bot-bottle Codex provider image.
#
# Mirrors the default Claude image shape: Node LTS, git/network tooling,
# non-root node user, and the provider CLI installed for that user.
# non-root node user, and the provider CLI installed globally.
FROM node:22-slim
RUN apt-get update \
&& apt-get install -y --no-install-recommends git ca-certificates curl procps \
&& apt-get install -y --no-install-recommends git ca-certificates curl \
&& rm -rf /var/lib/apt/lists/*
# App-specific deps. Python isn't required by codex itself
@@ -17,15 +17,12 @@ RUN apt-get update \
&& apt-get install -y --no-install-recommends python3 python3-pip python3-venv \
&& rm -rf /var/lib/apt/lists/*
RUN npm install -g --no-fund --no-audit @openai/codex@0.136.0 \
&& npm cache clean --force
USER node
WORKDIR /home/node
ENV PATH="/home/node/.local/bin:${PATH}"
# Remote-control support requires the standalone Codex install layout
# under ~/.codex/packages/standalone/current. The npm package can run
# the TUI, but remote-control commands expect this installer-owned path.
RUN mkdir -p /home/node/.codex \
&& curl -fsSL https://chatgpt.com/codex/install.sh | sh
RUN mkdir -p /home/node/.codex
CMD ["codex"]
@@ -55,6 +55,7 @@ _RUNTIME = AgentProviderRuntime(
prompt_mode="read_prompt_file",
bypass_args=("--dangerously-bypass-approvals-and-sandbox",),
resume_args=("resume", "--last"),
remote_control_args=(),
)
+1
View File
@@ -166,6 +166,7 @@ _RUNTIME = AgentProviderRuntime(
prompt_mode="append_system_prompt",
bypass_args=(),
resume_args=(),
remote_control_args=(),
)
+10
View File
@@ -439,6 +439,15 @@ def route_to_yaml_dict(r: Route) -> dict[str, object]:
return d
def load_routes(text: str) -> tuple[Route, ...]:
"""Parse YAML text → routes."""
try:
payload = parse_yaml_subset(text)
except YamlSubsetError as e:
raise ValueError(f"routes payload: invalid YAML: {e}") from e
return parse_routes(payload)
def parse_config(payload: object) -> "Config":
"""Parse a full egress config payload (top-level log level + routes)."""
if not isinstance(payload, dict):
@@ -853,6 +862,7 @@ __all__ = [
"is_git_push_request",
"is_git_fetch_request",
"load_config",
"load_routes",
"match_route",
"outbound_scan_headers",
"parse_config",
+3 -8
View File
@@ -47,11 +47,11 @@ from pathlib import Path
try:
# Same-directory imports inside the bundle container; these files are
# COPYed flat under /app by Dockerfile.sidecars.
from egress_addon_core import LOG_OFF, load_config
from egress_addon_core import load_routes
import supervise as _sv
except ModuleNotFoundError:
# Package imports for host-side tests and tooling.
from .egress_addon_core import LOG_OFF, load_config
from .egress_addon_core import load_routes
from . import supervise as _sv
@@ -297,17 +297,12 @@ def validate_proposed_file(tool: str, content: str) -> None:
pass
elif tool in (_sv.TOOL_EGRESS_ALLOW, _sv.TOOL_EGRESS_BLOCK):
try:
config = load_config(content)
load_routes(content)
except ValueError as e:
raise _RpcError(
ERR_INVALID_PARAMS,
f"{tool}: proposed routes.yaml is not valid: {e}",
) from e
if config.log != LOG_OFF:
raise _RpcError(
ERR_INVALID_PARAMS,
f"{tool}: proposed routes.yaml must not change egress logging",
)
else:
raise _RpcError(ERR_INVALID_PARAMS, f"unknown tool {tool!r}")
@@ -0,0 +1,216 @@
# PRD prd-new: Separate agent and bottle selection
- **Status:** Draft
- **Author:** claude
- **Created:** 2026-06-25
- **Issue:** #269
## Summary
Agents and bottles are two separate concerns: agents carry a system prompt and
skills; bottles carry infrastructure configuration (egress, git-gate, env,
agent provider). Today an agent's manifest file hard-codes a single `bottle:`
reference, which prevents the same agent prompt from being reused across
projects that need different bottle configurations. This PRD decouples them: at
launch time, after choosing the agent, the operator picks an ordered list of
bottles via a multi-select picker. The selected bottles are merged in order
(later entries override earlier ones) to produce the effective bottle for the
session.
## Problem
The current `bottle: <name>` field on an agent manifest file binds the agent
permanently to one bottle. To use the same system prompt with a different bottle
(e.g. `claude-implementer` at home vs. at a client site that needs a different
egress policy), the operator must duplicate the agent file and change the
`bottle:` field. Duplicate agent files drift out of sync.
## Goals / Success Criteria
1. `bottle:` in an agent's frontmatter becomes optional. Existing manifests with
`bottle:` continue to work unchanged (backward compat).
2. After selecting an agent (via the existing single-select picker), a new
multi-select bottle picker appears showing all available bottles.
3. The multi-select picker pre-populates with the agent's `bottle:` value when
present.
4. Confirming with one or more bottles selected uses those bottles, merged in
selection order, as the effective bottle for the session.
5. Confirming with an empty selection falls back to the agent's `bottle:` field.
If neither is set, a ManifestError is raised pointing the operator at the fix.
6. The ordered bottle list is stored in launch metadata so `./cli.py resume`
uses the same bottles.
7. The preflight summary (`y/N` screen) shows the effective bottle name(s).
8. The multi-select picker supports incremental filtering, Space/Enter to toggle
selection, an ordered "Selected: ..." summary line, Ctrl-D to confirm, and
Esc/q to cancel the whole start operation.
9. Unit tests cover: multi-select widget (filter, toggle, confirm, cancel),
the `cmd_start` bottle-picker step, and the manifest `load_for_agent`
runtime-bottle-merge path.
## Non-goals
- Reordering the selection list from within the picker (order = insertion order;
drag-and-drop is out of scope).
- Storing bottle selection history / MRU.
- Changes to `./cli.py edit`, `./cli.py list`, or `./cli.py info`.
- Removing the `bottle:` key from the agent schema (it stays, now optional).
## Design
### `bot_bottle/cli/tui.py``filter_multiselect`
```python
def filter_multiselect(
items: list[str],
*,
title: str = "",
initial: list[str] | None = None,
tty_path: str = "/dev/tty",
) -> list[str] | None:
"""Multi-select variant of filter_select.
Returns the ordered list of selected items, or None on cancel.
Press Space/Enter to toggle the item under the cursor.
Press Ctrl-D to confirm. Press Esc/q to cancel.
"""
```
Layout:
```
Select bottles
Filter: _
─────────────────────────────────────────
> [*] claude
[ ] dev
[ ] codex
─────────────────────────────────────────
Selected (in order): claude
─────────────────────────────────────────
[↑↓/jk] move [Space] toggle [Ctrl-D] done [Esc] cancel
```
`initial` pre-populates the ordered selection. `None` means no pre-selection.
Items added are appended in insertion order; items removed leave the remaining
order unchanged.
### `bot_bottle/manifest_schema.py` — optional `bottle:`
`bottle` moves from `AGENT_KEYS_REQUIRED` to `AGENT_KEYS_OPTIONAL`.
### `bot_bottle/manifest_agent.py` — optional `bottle:`
`ManifestAgent.bottle` changes from `str` (required) to `str = ""`.
`from_dict` no longer requires the key to be present; the bottle-exists
validation is skipped when the key is absent.
### `bot_bottle/manifest_loader.py``scan_bottle_names`
```python
def scan_bottle_names(bottles_dir: Path) -> list[str]:
"""Scan <bottles_dir>/*.md and return sorted bottle names."""
```
### `bot_bottle/manifest.py``ManifestIndex` changes
**`all_bottle_names` property** — analogous to `all_agent_names`; scans
`home_md / "bottles"` in lazy mode, returns `sorted(self.bottles.keys())` in
eager mode.
**`load_for_agent(agent_name, bottle_names: tuple[str, ...] = ())`** — new
`bottle_names` parameter. When non-empty, the listed bottles are resolved and
merged in order (index 0 is the base; each subsequent bottle is applied on top
using the same field-merge rules as `extends:`). The result replaces the bottle
that `agent.bottle` would have provided. When empty, falls back to `agent.bottle`.
Raises ManifestError if neither `bottle_names` nor `agent.bottle` is set.
### `bot_bottle/manifest_extends.py``merge_bottles_runtime`
```python
def merge_bottles_runtime(bottles: list[ManifestBottle]) -> ManifestBottle:
"""Merge an ordered list of pre-resolved ManifestBottle objects.
Index 0 is the base; each subsequent entry overrides the previous using
the same rules as the file-based extends machinery:
- env: dict merge, later wins
- git_user: per-field overlay, later wins on non-empty
- git (repos): union by name, later wins per-name
- egress.routes: concatenate
- agent_provider, supervise: later bottle's value replaces earlier
"""
```
This function operates on already-parsed `ManifestBottle` objects, so it does
not need to touch the raw-dict path.
### `bot_bottle/backend/__init__.py``BottleSpec` + `_validate`
`BottleSpec` gains `bottle_names: tuple[str, ...] = ()`.
`BottleBackend._validate` passes `spec.bottle_names` to `load_for_agent`:
```python
manifest = spec.manifest.load_for_agent(spec.agent_name, spec.bottle_names)
```
The preflight print updates `info(f"bottle: {agent.bottle}")` to display the
effective bottle name(s). When `spec.bottle_names` is non-empty those are
shown; when empty and `agent.bottle` is set, the agent's `bottle:` is shown.
### `bot_bottle/bottle_state.py` — persist bottle names
`BottleMetadata` gains `bottle_names: tuple[str, ...] = ()`. `read_metadata`
reads this from JSON (default `()`). `write_launch_metadata` passes
`spec.bottle_names` through.
### `bot_bottle/cli/start.py` — bottle multiselect step
After agent selection, before the name/color modal:
```python
available_bottle_names = manifest.all_bottle_names
# Peek at agent's bottle default for pre-population
initial_bottle = _peek_agent_bottle(manifest, agent_name)
initial = [initial_bottle] if initial_bottle else []
bottle_names_list = tui.filter_multiselect(
available_bottle_names,
title="Select bottles",
initial=initial,
)
if bottle_names_list is None:
return 0 # user cancelled
bottle_names = tuple(bottle_names_list)
```
`_peek_agent_bottle` reads the agent file's frontmatter without full parsing,
returning the `bottle:` value or `""` when absent.
`BottleSpec` is built with `bottle_names=bottle_names`.
### `bot_bottle/cli/resume.py` — bottle names from metadata
```python
spec = BottleSpec(
...
bottle_names=tuple(metadata.bottle_names),
)
```
## Implementation chunks
1. **Schema + model**`manifest_schema.py`, `manifest_agent.py` (optional
`bottle:`), `manifest_loader.py` (`scan_bottle_names`), `manifest.py`
(`all_bottle_names`, `load_for_agent` signature), `manifest_extends.py`
(`merge_bottles_runtime`), `bottle_state.py` (`bottle_names` field),
`resolve_common.py` (thread through).
2. **Backend**`BottleSpec.bottle_names`, `_validate`, preflight print.
3. **TUI**`filter_multiselect` in `tui.py` + unit tests.
4. **CLI wiring**`start.py` bottle picker step, `resume.py` metadata load.
5. **Tests**`test_cli_start_selector.py` bottle-picker cases,
`test_manifest_agent.py` optional-bottle cases, new
`test_manifest_bottle_merge.py` for `merge_bottles_runtime`.
## Open questions
None.
-1
View File
@@ -4,4 +4,3 @@
pylint>=3.0.0
pyright>=1.1.300
coverage>=7.0.0
+4 -7
View File
@@ -92,9 +92,9 @@ class TestSandboxEscape(unittest.TestCase):
"on PATH: curl -sSL https://smolmachines.com/install.sh | sh"
)
# Throwaway static key for the git-gate fixture. It need not
# be a real SSH key: test 5 reaches gitleaks before any SSH
# attempt anyway.
# Throwaway "identity file" for the git-gate's `identity` field.
# It need not be a real SSH key: test 5 reaches gitleaks before
# any SSH attempt anyway.
fd, kp = tempfile.mkstemp(prefix="sandbox-test-key.")
os.close(fd)
cls._key_path = Path(kp)
@@ -123,10 +123,7 @@ class TestSandboxEscape(unittest.TestCase):
"git-gate": {"repos": {
"throwaway": {
"url": "ssh://git@unreachable.invalid:22/throwaway.git",
"key": {
"provider": "static",
"path": str(cls._key_path),
},
"identity": str(cls._key_path),
},
}},
},
@@ -198,7 +198,6 @@ class TestSmolmachinesLaunch(unittest.TestCase):
# connect fails, which is the property chunk 3 will
# preserve once egress is actually running.
r = self.bottle.exec(
"env -u HTTPS_PROXY -u HTTP_PROXY -u https_proxy -u http_proxy "
f"curl -s --show-error --max-time 3 http://{self.plan.bundle_ip}:9099 "
"2>&1 || true"
)
-21
View File
@@ -102,27 +102,6 @@ class TestAttachAgent(unittest.TestCase):
bottle.argv,
)
def test_remote_control_is_provider_startup_arg(self):
class Bottle:
argv: list[str] = []
def exec_agent(self, argv: list[str], *, tty: bool = True) -> int:
self.argv = list(argv)
return 0
bottle = Bottle()
exit_code = start_mod.attach_agent(
bottle, # type: ignore[arg-type]
agent_provider_template="codex",
startup_args=("remote-control",),
)
self.assertEqual(0, exit_code)
self.assertEqual(
["--dangerously-bypass-approvals-and-sandbox", "remote-control"],
bottle.argv,
)
if __name__ == "__main__":
unittest.main()
@@ -29,9 +29,6 @@ from bot_bottle.supervise import SupervisePlan
_URL = "http://supervise:9100/"
_CODEX_DOCKERFILE = (
Path(__file__).resolve().parents[2] / "bot_bottle/contrib/codex/Dockerfile"
)
def _make_bottle(exec_result: ExecResult | None = None) -> MagicMock:
@@ -279,12 +276,6 @@ class TestCodexProvision(unittest.TestCase):
)
class TestCodexDockerfile(unittest.TestCase):
def test_installs_procps_for_remote_control_pid_management(self):
dockerfile = _CODEX_DOCKERFILE.read_text()
self.assertIn("procps", dockerfile)
class TestCodexSuperviseMcp(unittest.TestCase):
def test_noop_when_supervise_disabled(self):
bottle = _make_bottle()
-10
View File
@@ -136,16 +136,6 @@ class TestClaudeArgv(unittest.TestCase):
argv,
)
def test_codex_remote_control_startup_arg_does_not_receive_initial_prompt(self):
argv = _codex_bottle("/home/node/.bot-bottle-prompt.txt").agent_argv(
["--dangerously-bypass-approvals-and-sandbox", "remote-control"],
)
self.assertEqual(
["docker", "exec", "-it", "bot-bottle-dev-abc", "codex",
"--dangerously-bypass-approvals-and-sandbox", "remote-control"],
argv,
)
def test_codex_resume_does_not_append_initial_prompt(self):
argv = _codex_bottle("/home/node/.bot-bottle-prompt.txt").agent_argv(
["--dangerously-bypass-approvals-and-sandbox", "resume", "--last"],
@@ -31,6 +31,7 @@ class _Provider(AgentProvider):
return AgentProviderRuntime(
template="test", command="test", image="",
prompt_mode="append_file", bypass_args=(), resume_args=(),
remote_control_args=(),
)
def provision_plan(self, **kwargs): # type: ignore[override]
raise NotImplementedError
+8 -8
View File
@@ -322,7 +322,7 @@ class TestRenderRoutes(unittest.TestCase):
self.assertEqual([], parse_yaml_subset(rendered)["routes"])
def test_round_trip_through_addon_core(self):
from bot_bottle.egress_addon_core import load_config
from bot_bottle.egress_addon_core import load_routes
b = _bottle([
{"host": "api.github.com",
"auth": {"scheme": "Bearer", "token_ref": "GH_PAT"},
@@ -333,7 +333,7 @@ class TestRenderRoutes(unittest.TestCase):
{"host": "api.anthropic.com"},
])
routes = egress_routes_for_bottle(b)
addon_routes = load_config(egress_render_routes(routes)).routes
addon_routes = load_routes(egress_render_routes(routes))
self.assertEqual(3, len(addon_routes))
self.assertEqual("Bearer", addon_routes[0].auth_scheme)
self.assertEqual("EGRESS_TOKEN_0", addon_routes[0].token_env)
@@ -341,26 +341,26 @@ class TestRenderRoutes(unittest.TestCase):
self.assertEqual("", addon_routes[2].auth_scheme)
def test_dlp_round_trips(self):
from bot_bottle.egress_addon_core import load_config
from bot_bottle.egress_addon_core import load_routes
b = _bottle([{"host": "x.example", "dlp": {
"outbound_detectors": ["token_patterns"],
"inbound_detectors": False,
}}])
routes = egress_routes_for_bottle(b)
rendered = egress_render_routes(routes)
addon_routes = load_config(rendered).routes
addon_routes = load_routes(rendered)
self.assertEqual(("token_patterns",), addon_routes[0].outbound_detectors)
self.assertEqual((), addon_routes[0].inbound_detectors)
def test_outbound_on_match_round_trips(self):
from bot_bottle.egress_addon_core import load_config
from bot_bottle.egress_addon_core import load_routes
b = _bottle([{"host": "logs.example", "dlp": {
"outbound_on_match": "redact",
}}])
routes = egress_routes_for_bottle(b)
rendered = egress_render_routes(routes)
self.assertIn('outbound_on_match: "redact"', rendered)
addon_routes = load_config(rendered).routes
addon_routes = load_routes(rendered)
self.assertEqual("redact", addon_routes[0].outbound_on_match)
def test_outbound_on_match_default_omitted_from_render(self):
@@ -370,12 +370,12 @@ class TestRenderRoutes(unittest.TestCase):
self.assertNotIn("outbound_on_match", rendered)
def test_git_fetch_policy_round_trips(self):
from bot_bottle.egress_addon_core import load_config
from bot_bottle.egress_addon_core import load_routes
b = _bottle([{"host": "github.com", "git": {"fetch": True}}])
routes = egress_routes_for_bottle(b)
rendered = egress_render_routes(routes)
self.assertEqual({"fetch": True}, self._parsed(routes)[0]["git"])
addon_routes = load_config(rendered).routes
addon_routes = load_routes(rendered)
self.assertTrue(addon_routes[0].git_fetch)
def test_log_zero_omitted_from_render(self):
+42 -27
View File
@@ -32,6 +32,7 @@ from bot_bottle.egress_addon_core import (
is_git_fetch_request,
is_git_push_request,
load_config,
load_routes,
match_route,
outbound_scan_headers,
parse_config,
@@ -288,6 +289,47 @@ class TestParseDlp(unittest.TestCase):
}]})
# --- load_routes ---------------------------------------------------------
class TestLoadRoutes(unittest.TestCase):
def test_yaml_text_round_trip(self):
routes = load_routes(
'routes:\n'
' - host: "api.example"\n'
)
self.assertEqual(1, len(routes))
self.assertEqual("api.example", routes[0].host)
def test_full_route_shape_parses(self):
routes = load_routes(
'routes:\n'
' - host: "api.example"\n'
' auth_scheme: "Bearer"\n'
' token_env: "EGRESS_TOKEN_0"\n'
' matches:\n'
' - paths:\n'
' - value: "/v1/"\n'
' - type: "exact"\n'
' value: "/messages"\n'
)
self.assertEqual(1, len(routes))
r = routes[0]
self.assertEqual("api.example", r.host)
self.assertEqual("Bearer", r.auth_scheme)
self.assertEqual("EGRESS_TOKEN_0", r.token_env)
self.assertEqual(1, len(r.matches))
self.assertEqual(2, len(r.matches[0].paths))
def test_empty_routes_list(self):
routes = load_routes("routes: []\n")
self.assertEqual((), routes)
def test_invalid_yaml_raises_value_error(self):
with self.assertRaises(ValueError):
load_routes("routes:\n\t- host: x\n")
# --- load_config / parse_config ------------------------------------------
@@ -336,33 +378,6 @@ class TestLoadConfig(unittest.TestCase):
with self.assertRaises(ValueError):
parse_config("not a dict")
def test_empty_routes_list(self):
cfg = load_config("routes: []\n")
self.assertEqual((), cfg.routes)
def test_full_route_shape_parses(self):
cfg = load_config(
'routes:\n'
' - host: "api.example"\n'
' auth_scheme: "Bearer"\n'
' token_env: "EGRESS_TOKEN_0"\n'
' matches:\n'
' - paths:\n'
' - value: "/v1/"\n'
' - type: "exact"\n'
' value: "/messages"\n'
)
r = cfg.routes[0]
self.assertEqual("api.example", r.host)
self.assertEqual("Bearer", r.auth_scheme)
self.assertEqual("EGRESS_TOKEN_0", r.token_env)
self.assertEqual(1, len(r.matches))
self.assertEqual(2, len(r.matches[0].paths))
def test_invalid_yaml_raises_value_error(self):
with self.assertRaises(ValueError):
load_config("routes:\n\t- host: x\n")
# --- evaluate_matches ---------------------------------------------------
-9
View File
@@ -54,15 +54,6 @@ class TestValidateRoutesContent(unittest.TestCase):
' auth_scheme: "Bearer"\n'
)
def test_rejects_log_full(self):
with self.assertRaises(EgressApplyError) as cm:
applicator.validate_routes_content(
'log: 2\n'
'routes:\n'
' - host: "x.example"\n'
)
self.assertIn("must not change egress logging", str(cm.exception))
class TestApplyRoutesChange(unittest.TestCase):
def setUp(self):
-27
View File
@@ -73,33 +73,6 @@ resolver #2
)
self.assertTrue(run.call_args_list[-1].kwargs["check"])
def test_build_image_anchors_relative_dockerfile_to_context(self):
status = util.subprocess.CompletedProcess(
args=[],
returncode=0,
stdout=(
'[{"status":{"state":"running"},'
'"configuration":{"dns":{"nameservers":["9.9.9.9"]}}}]'
),
stderr="",
)
with patch.object(util.subprocess, "run", return_value=status) as run, \
patch.object(util.os, "environ", {
"BOT_BOTTLE_MACOS_CONTAINER_DNS": "9.9.9.9",
}):
util.build_image(
"bot-bottle-sidecars:latest",
"/repo",
dockerfile="Dockerfile.sidecars",
)
self.assertEqual(
[
"container", "build", "-t", "bot-bottle-sidecars:latest",
"--dns", "9.9.9.9", "-f", "/repo/Dockerfile.sidecars", "/repo",
],
run.call_args_list[-1].args[0],
)
def test_commit_container_execs_tar_and_builds_image(self):
# stderr is bytes because subprocess.run uses stderr=PIPE without text=True
completed = util.subprocess.CompletedProcess(
+1 -1
View File
@@ -130,7 +130,7 @@ def _capture_print(plan: DockerBottlePlan | SmolmachinesBottlePlan) -> list[str]
orig = sys.stderr
sys.stderr = buf
try:
plan.print()
plan.print(remote_control=False)
finally:
sys.stderr = orig
return buf.getvalue().splitlines()
@@ -42,6 +42,7 @@ class _Provider(AgentProvider):
return AgentProviderRuntime(
template="test", command="test", image="",
prompt_mode="append_file", bypass_args=(), resume_args=(),
remote_control_args=(),
)
def provision_plan(self, **kwargs): # type: ignore[override]
raise NotImplementedError
+21 -175
View File
@@ -20,7 +20,6 @@ import supervise as _sv # noqa: E402 # type: ignore
from bot_bottle import supervise_server # noqa: E402
from bot_bottle.supervise_server import (
ERR_INTERNAL,
ERR_INVALID_PARAMS,
ERR_INVALID_REQUEST,
ERR_METHOD_NOT_FOUND,
@@ -30,9 +29,7 @@ from bot_bottle.supervise_server import (
PROPOSED_FILE_FIELD,
ServerConfig,
TOOL_DEFINITIONS,
_RpcClientError,
_RpcError,
_RpcInternalError,
_response_timeout_from_env,
format_response_text,
handle_initialize,
@@ -50,15 +47,15 @@ from bot_bottle.supervise_server import (
class TestValidation(unittest.TestCase):
def test_capability_block_accepts_anything_nonempty(self):
validate_proposed_file(
_sv.TOOL_CAPABILITY_BLOCK,
"FROM python:3.13\nRUN apk add git\n",
)
def test_empty_proposed_file_rejected_for_tools_with_file_field(self):
with self.assertRaises(_RpcError):
validate_proposed_file(_sv.TOOL_EGRESS_ALLOW, " \n\t")
def test_capability_block_rejected_as_unknown_tool(self):
with self.assertRaises(_RpcError) as cm:
validate_proposed_file("capability-block", "FROM python:3.13\n")
self.assertEqual(ERR_INVALID_PARAMS, cm.exception.code)
self.assertIn("unknown tool", cm.exception.message)
validate_proposed_file(_sv.TOOL_CAPABILITY_BLOCK, " \n\t")
def test_egress_routes_yaml_is_validated(self):
validate_proposed_file(
@@ -70,74 +67,6 @@ class TestValidation(unittest.TestCase):
with self.assertRaises(_RpcError):
validate_proposed_file(_sv.TOOL_EGRESS_BLOCK, "routes: nope\n")
def test_egress_routes_yaml_rejects_log_full(self):
with self.assertRaises(_RpcError) as cm:
validate_proposed_file(
_sv.TOOL_EGRESS_ALLOW,
"log: 2\nroutes:\n - host: example.com\n",
)
self.assertEqual(ERR_INVALID_PARAMS, cm.exception.code)
self.assertIn("must not change egress logging", cm.exception.message)
# --- Error taxonomy --------------------------------------------------------
class TestRpcErrorTaxonomy(unittest.TestCase):
def test_rpc_client_error_is_rpc_error(self):
e = _RpcClientError(ERR_INVALID_PARAMS, "bad param")
self.assertIsInstance(e, _RpcError)
self.assertEqual(ERR_INVALID_PARAMS, e.code)
self.assertEqual("bad param", e.message)
def test_rpc_internal_error_is_rpc_error(self):
e = _RpcInternalError("disk full")
self.assertIsInstance(e, _RpcError)
self.assertEqual(ERR_INTERNAL, e.code)
self.assertEqual("disk full", e.message)
def test_rpc_internal_error_preserves_cause(self):
cause = OSError("no space left on device")
try:
raise _RpcInternalError("failed to write") from cause
except _RpcInternalError as e:
self.assertIs(cause, e.__cause__)
def test_parse_error_is_client_error(self):
with self.assertRaises(_RpcClientError):
parse_jsonrpc(b"{bad json")
def test_validation_error_is_client_error(self):
with self.assertRaises(_RpcClientError):
validate_proposed_file(_sv.TOOL_EGRESS_ALLOW, "routes: nope\n")
def test_unknown_tool_in_tools_call_is_client_error(self):
config = ServerConfig(bottle_slug="dev", queue_dir=Path("/unused"))
with self.assertRaises(_RpcClientError) as cm:
handle_tools_call({"name": "no-such-tool", "arguments": {}}, config)
self.assertEqual(ERR_INVALID_PARAMS, cm.exception.code)
class TestRpcInternalErrorOnIoFailure(unittest.TestCase):
def test_write_proposal_os_error_raises_internal(self):
config = ServerConfig(
bottle_slug="dev",
queue_dir=Path("/dev/null/cannot-exist"),
)
with self.assertRaises(_RpcInternalError) as cm:
handle_tools_call(
{
"name": _sv.TOOL_EGRESS_ALLOW,
"arguments": {
"routes_yaml": "routes:\n - host: example.com\n",
"justification": "x",
},
},
config,
)
self.assertEqual(ERR_INTERNAL, cm.exception.code)
self.assertIsNotNone(cm.exception.__cause__)
# --- JSON-RPC parsing ------------------------------------------------------
@@ -219,6 +148,7 @@ class TestHandleToolsList(unittest.TestCase):
self.assertEqual(
sorted([
_sv.TOOL_EGRESS_ALLOW,
_sv.TOOL_CAPABILITY_BLOCK,
_sv.TOOL_EGRESS_BLOCK,
_sv.TOOL_LIST_EGRESS_ROUTES,
]),
@@ -294,10 +224,10 @@ class TestHandleToolsCall(unittest.TestCase):
try:
result = handle_tools_call(
{
"name": _sv.TOOL_EGRESS_BLOCK,
"name": _sv.TOOL_CAPABILITY_BLOCK,
"arguments": {
"routes_yaml": "routes:\n - host: example.com\n",
"justification": "need example.com",
"dockerfile": "FROM python:3.13\n",
"justification": "need git",
},
},
self.config,
@@ -334,9 +264,9 @@ class TestHandleToolsCall(unittest.TestCase):
try:
result = handle_tools_call(
{
"name": _sv.TOOL_EGRESS_ALLOW,
"name": _sv.TOOL_CAPABILITY_BLOCK,
"arguments": {
"routes_yaml": "routes:\n - host: example.com\n",
"dockerfile": "FROM python:3.13\n",
"justification": "needed for tests",
},
},
@@ -358,52 +288,20 @@ class TestHandleToolsCall(unittest.TestCase):
with self.assertRaises(_RpcError):
handle_tools_call(
{
"name": _sv.TOOL_EGRESS_ALLOW,
"arguments": {"routes_yaml": "routes:\n - host: example.com\n"},
"name": _sv.TOOL_CAPABILITY_BLOCK,
"arguments": {"dockerfile": "FROM python:3.13\n"},
},
self.config,
)
def test_missing_name_raises(self):
with self.assertRaises(_RpcError) as cm:
handle_tools_call({"arguments": {}}, self.config)
self.assertEqual(ERR_INVALID_PARAMS, cm.exception.code)
def test_arguments_must_be_object(self):
with self.assertRaises(_RpcError) as cm:
handle_tools_call(
{
"name": _sv.TOOL_EGRESS_ALLOW,
"arguments": [],
},
self.config,
)
self.assertEqual(ERR_INVALID_PARAMS, cm.exception.code)
self.assertIn("must be an object", cm.exception.message)
def test_capability_block_call_raises_unknown_tool(self):
with self.assertRaises(_RpcError) as cm:
handle_tools_call(
{
"name": "capability-block",
"arguments": {
"dockerfile": "FROM python:3.13\n",
"justification": "need git",
},
},
self.config,
)
self.assertEqual(ERR_INVALID_PARAMS, cm.exception.code)
self.assertIn("unknown tool", cm.exception.message)
def test_archives_proposal_after_response(self):
responder = self._respond_when_proposal_appears(_sv.STATUS_APPROVED)
try:
handle_tools_call(
{
"name": _sv.TOOL_EGRESS_ALLOW,
"name": _sv.TOOL_CAPABILITY_BLOCK,
"arguments": {
"routes_yaml": "routes:\n - host: example.com\n",
"dockerfile": "FROM python:3.13\n",
"justification": "x",
},
},
@@ -425,10 +323,10 @@ class TestHandleToolsCall(unittest.TestCase):
)
result = handle_tools_call(
{
"name": _sv.TOOL_EGRESS_ALLOW,
"name": _sv.TOOL_CAPABILITY_BLOCK,
"arguments": {
"routes_yaml": "routes:\n - host: example.com\n",
"justification": "need egress",
"dockerfile": "FROM python:3.13\n",
"justification": "need a capability",
},
},
config,
@@ -443,31 +341,6 @@ class TestHandleToolsCall(unittest.TestCase):
class TestHandleListEgressRoutes(unittest.TestCase):
def test_success_returns_body_text(self):
class _Resp:
def __enter__(self):
return self
def __exit__(self, exc_type: type[BaseException] | None, exc: BaseException | None, tb: object) -> bool:
return False
def read(self):
return b"[{\"host\": \"example.com\"}]"
class _Opener:
def open(self, *args, **kwargs): # noqa: ANN001, ANN002, ANN003 # type: ignore
return _Resp()
with patch.object(supervise_server.urllib.request, "build_opener", return_value=_Opener()):
result = handle_list_egress_routes(
{},
ServerConfig(bottle_slug="dev", queue_dir=Path("/unused")),
)
self.assertFalse(result["isError"]) # type: ignore[index]
text = result["content"][0]["text"] # type: ignore[index]
self.assertIn("example.com", text)
def test_url_error_returns_tool_error(self):
class _Opener:
def open(self, *args, **kwargs): # noqa: ANN001, ANN002, ANN003 # type: ignore
@@ -527,13 +400,6 @@ class TestFormatResponseText(unittest.TestCase):
self.assertIn("the operator modified", text.lower())
class TestFormatPendingResponseText(unittest.TestCase):
def test_formats_timeout_message(self):
text = supervise_server.format_pending_response_text(12.5)
self.assertIn("status: pending", text)
self.assertIn("12.5s", text)
# --- End-to-end HTTP sanity ------------------------------------------------
@@ -584,7 +450,7 @@ class TestHttpEndToEnd(unittest.TestCase):
self.assertEqual("2.0", result["jsonrpc"])
self.assertEqual(1, result["id"])
names = [t["name"] for t in result["result"]["tools"]] # type: ignore[index]
self.assertNotIn("capability-block", names)
self.assertIn(_sv.TOOL_CAPABILITY_BLOCK, names)
self.assertIn(_sv.TOOL_EGRESS_ALLOW, names)
self.assertIn(_sv.TOOL_EGRESS_BLOCK, names)
@@ -594,26 +460,6 @@ class TestHttpEndToEnd(unittest.TestCase):
)
self.assertEqual(ERR_METHOD_NOT_FOUND, result["error"]["code"]) # type: ignore[index]
def test_internal_error_returns_err_internal_over_http(self):
with patch.object(
supervise_server._sv, "write_proposal",
side_effect=OSError("disk full"),
):
result = self._post_jsonrpc({
"jsonrpc": "2.0",
"id": 99,
"method": "tools/call",
"params": {
"name": _sv.TOOL_EGRESS_ALLOW,
"arguments": {
"routes_yaml": "routes:\n - host: example.com\n",
"justification": "x",
},
},
})
self.assertIn("error", result)
self.assertEqual(ERR_INTERNAL, result["error"]["code"]) # type: ignore[index]
def test_health_endpoint(self):
conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=5)
try: