didericis/bot-bottle
1
0
Fork 0
Code Issues 7 Pull Requests 15 Actions Packages Projects Releases Wiki Activity

refactor(preflight): compact y/N summary #31

Merged
didericis merged 2 commits from compact-preflight-summary into main 2026-05-25 20:59:48 -04:00
Conversation 0 Commits 2 Files Changed 6
+39 -380
6 changed files with 39 additions and 380 deletions
Download Patch File Download Diff File Expand all files Collapse all files
claude_bottle/backend/__init__.py
-7
View File
@@ -73,13 +73,6 @@ class BottlePlan(ABC):
def print(self, *, remote_control: bool) -> None:
"""Render the y/N preflight summary to stderr."""
@abstractmethod
def to_dict(self, *, remote_control: bool) -> dict[str, object]:
"""Return the plan as a JSON-serializable dict for machine
consumption (used by `start --dry-run --format=json`). The key
set is part of the CLI's user-facing contract — adding fields
is fine, renaming or removing is a breaking change."""
@dataclass(frozen=True)
class BottleCleanupPlan(ABC):
claude_bottle/backend/docker/bottle_plan.py
+38 -146
View File
@@ -14,24 +14,11 @@ from pathlib import Path
from ...egress_proxy import EgressProxyPlan
from ...git_gate import GitGatePlan
from ...log import info
from ...manifest import Agent, Bottle
from ...pipelock import PipelockProxyPlan, pipelock_effective_allowlist
from ...pipelock import PipelockProxyPlan
from ...supervise import SupervisePlan
from .. import BottlePlan
@dataclass(frozen=True)
class _PlanView:
"""Fields that both `print` and `to_dict` need but the plan
doesn't store directly. Cheap to compute; computed once per call."""
agent: Agent
bottle: Bottle
env_names: list[str]
git_names: list[str]
prompt_first_line: str
@dataclass(frozen=True)
class DockerBottlePlan(BottlePlan):
"""Docker-specific resolved fields produced by
@@ -65,150 +52,55 @@ class DockerBottlePlan(BottlePlan):
allowlist_summary: str
use_runsc: bool
def _view(self) -> _PlanView:
def print(self, *, remote_control: bool) -> None:
"""Render the y/N preflight summary to stderr — compact form
intended to fit on screen without scrolling. The full
structured shape (image, container, runtime, etc.) lives on
this dataclass for tooling that wants to introspect it."""
del remote_control # not surfaced in the compact summary
spec = self.spec
manifest = spec.manifest
agent = manifest.agents[spec.agent_name]
bottle = manifest.bottle_for(spec.agent_name)
# The agent sees the union of literal env names (rendered into
# --env-file) and forwarded env names (`-e NAME` with the value
# arriving via subprocess env). The forwarded set holds the
# OAuth token (CLAUDE_CODE_OAUTH_TOKEN) and any host-env
# interpolations from the manifest; egress-proxy holds upstream
# tokens in its own environ, so no token forwarding from the
# agent to the proxy is needed.
# --env-file) and forwarded env names (`-e NAME` with the
# value arriving via subprocess env). The forwarded set holds
# the OAuth token (CLAUDE_CODE_OAUTH_TOKEN) and any host-env
# interpolations from the manifest; egress-proxy holds
# upstream tokens in its own environ, so no token forwarding
# from the agent to the proxy is needed.
env_names = sorted(set(bottle.env.keys()) | set(self.forwarded_env.keys()))
return _PlanView(
agent=agent,
bottle=bottle,
env_names=env_names,
git_names=[e.Name for e in bottle.git],
prompt_first_line=agent.prompt.splitlines()[0] if agent.prompt else "",
)
def print(self, *, remote_control: bool) -> None:
"""Render the y/N preflight summary to stderr. Pure presentation."""
v = self._view()
spec = self.spec
runtime_label = "runsc (gVisor)" if self.use_runsc else "runc (default)"
def _multi(label: str, values: list[str]) -> None:
"""Print a label with N continuation-indented values. Used
for env / skills / git-gate / egress-proxy where one item
per line keeps the summary scannable."""
if not values:
info(f"{label}: (none)")
return
info(f"{label}: {values[0]}")
indent = " " * (len(label) + 2)
for v_ in values[1:]:
info(f"{indent}{v_}")
print(file=sys.stderr)
info(f"agent : {spec.agent_name}")
info(f"image : {self.image}")
if self.dockerfile_path:
info(
f"dockerfile : {self.dockerfile_path} "
f"(per-bottle override from PRD 0016 capability rebuild)"
)
if self.derived_image:
info(
f"cwd : {spec.user_cwd} -> /home/node/workspace "
f"(derived: {self.derived_image})"
)
info(f"container : {self.container_name}")
info(f"stage dir : {self.stage_dir}")
info("env (names only): " + (", ".join(v.env_names) if v.env_names else "(none)"))
info("skills : " + (" ".join(v.agent.skills) if v.agent.skills else "(none)"))
info(f"docker runtime : {runtime_label}")
info(f"bottle : {v.agent.bottle}")
if v.git_names:
info(f" git remotes : {', '.join(v.git_names)}")
git_lines = [
f"{u.name} -> {u.upstream_host}:{u.upstream_port} "
f"(gitleaks-scanned)"
for u in self.git_gate_plan.upstreams
]
info(f" git gate : {'; '.join(git_lines)}")
else:
info(" git remotes : (none)")
_multi("env ", env_names)
_multi("skills ", list(agent.skills))
info(f"bottle : {agent.bottle}")
git_lines = [
f"{u.upstream_host}:{u.upstream_port}"
for u in self.git_gate_plan.upstreams
]
if git_lines:
_multi(" git gate ", git_lines)
if self.egress_proxy_plan.routes:
lines = []
egress_lines = []
for r in self.egress_proxy_plan.routes:
paths = (
" " + ",".join(r.path_allowlist) if r.path_allowlist else ""
)
auth = f" [auth:{r.auth_scheme}]" if r.auth_scheme else ""
lines.append(f"{r.host}{auth}{paths}")
refs = sorted({r.token_ref for r in self.egress_proxy_plan.routes if r.token_ref})
tokens_part = (
f"; tokens: {', '.join(refs)}" if refs else ""
)
info(f" egress-proxy : {len(lines)} route(s){tokens_part}")
for line in lines:
info(f" {line}")
else:
info(" egress-proxy : (none)")
info(f" egress : {self.allowlist_summary}")
info(" tls intercept : egress-proxy (per-bottle ephemeral CA, generated at launch)")
if self.supervise_plan is not None:
info(
f" supervise : enabled; queue at {self.supervise_plan.queue_dir}"
)
else:
info(" supervise : disabled (set bottle.supervise=true to enable)")
info(
f"prompt : {len(v.agent.prompt)} chars; "
f"first line: {v.prompt_first_line or '(empty)'}"
)
info("remote-control : " + ("enabled" if remote_control else "disabled"))
egress_lines.append(f"{r.host}{auth}")
_multi(" egress-proxy ", egress_lines)
print(file=sys.stderr)
def to_dict(self, *, remote_control: bool) -> dict[str, object]:
v = self._view()
hosts = pipelock_effective_allowlist(v.bottle)
return {
"agent": self.spec.agent_name,
"bottle": v.agent.bottle,
"container_name": self.container_name,
"image": self.image,
"derived_image": self.derived_image,
"stage_dir": str(self.stage_dir),
"runtime": "runsc" if self.use_runsc else "runc",
"env_names": v.env_names,
"skills": list(v.agent.skills),
"git_remotes": v.git_names,
"git_gate": [
{
"name": u.name,
"upstream": f"{u.upstream_host}:{u.upstream_port}",
"upstream_url": u.upstream_url,
"known_host_key_pinned": bool(u.known_host_key),
}
for u in self.git_gate_plan.upstreams
],
"egress_proxy": [
{
"host": r.host,
"path_allowlist": list(r.path_allowlist),
"auth_scheme": r.auth_scheme,
"token_ref": r.token_ref,
}
for r in self.egress_proxy_plan.routes
],
"egress": {
"host_count": len(hosts),
"hosts": hosts,
# PRD 0017: TLS interception moved from pipelock to
# egress-proxy. ca_fingerprint is always null at
# dry-run because the CA doesn't exist yet — real
# launches print the fingerprint to stderr from
# provision_ca. Reserved field for forward-compat.
"tls_interception": {
"enabled": True,
"ca_fingerprint": None,
},
},
"supervise": {
"enabled": self.supervise_plan is not None,
"queue_dir": (
str(self.supervise_plan.queue_dir)
if self.supervise_plan is not None
else None
),
},
"prompt": {
"length": len(v.agent.prompt),
"first_line": v.prompt_first_line,
},
"remote_control": remote_control,
}
claude_bottle/cli/resume.py
-7
View File
@@ -29,12 +29,6 @@ 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(
"--format",
choices=("text", "json"),
default="text",
help="preflight output format; --format=json requires --dry-run",
)
parser.add_argument(
"identity",
help="bottle identity from a prior `start` (see its session-end output)",
@@ -61,6 +55,5 @@ def cmd_resume(argv: list[str]) -> int:
return _launch_bottle(
spec,
dry_run=args.dry_run,
output_format=args.format,
remote_control=args.remote_control,
)
claude_bottle/cli/start.py
+1 -17
View File
@@ -9,7 +9,6 @@ _launch_bottle below.
from __future__ import annotations
import argparse
import json
import os
import shutil
import sys
@@ -23,7 +22,7 @@ from ..backend.docker.bottle_state import (
mark_preserved,
)
from ..backend.docker.capability_apply import snapshot_transcript
from ..log import die, info
from ..log import info
from ..manifest import Manifest
from ._common import PROG, USER_CWD, read_tty_line
@@ -33,18 +32,10 @@ def cmd_start(argv: list[str]) -> int:
parser.add_argument("--dry-run", action="store_true")
parser.add_argument("--cwd", action="store_true", help="copy host cwd into a derived image")
parser.add_argument("--remote-control", action="store_true")
parser.add_argument(
"--format",
choices=("text", "json"),
default="text",
help="preflight output format; --format=json requires --dry-run",
)
parser.add_argument("name", help="agent name defined in claude-bottle.json")
args = parser.parse_args(argv)
dry_run = args.dry_run or os.environ.get("CLAUDE_BOTTLE_DRY_RUN") == "1"
if args.format == "json" and not dry_run:
die("--format=json requires --dry-run")
manifest = Manifest.resolve(USER_CWD)
spec = BottleSpec(
@@ -56,7 +47,6 @@ def cmd_start(argv: list[str]) -> int:
return _launch_bottle(
spec,
dry_run=dry_run,
output_format=args.format,
remote_control=args.remote_control,
)
@@ -65,7 +55,6 @@ def _launch_bottle(
spec: BottleSpec,
*,
dry_run: bool,
output_format: str,
remote_control: bool,
) -> int:
"""Shared launch core for `start` and `resume`. Builds the plan,
@@ -76,11 +65,6 @@ def _launch_bottle(
backend = get_bottle_backend()
plan = backend.prepare(spec, stage_dir=stage_dir)
if output_format == "json":
json.dump(plan.to_dict(remote_control=remote_control), sys.stdout, indent=2)
sys.stdout.write("\n")
return 0
plan.print(remote_control=remote_control)
if dry_run:
tests/integration/test_dry_run_plan.py
-149
View File
@@ -1,149 +0,0 @@
"""Integration: cli.py start --dry-run --format=json renders a stable
machine-readable plan and creates zero Docker resources. The shape of
the JSON document is part of the CLI's user-facing contract."""
import json
import os
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path
from tests._docker import skip_unless_docker
REPO_ROOT = Path(__file__).resolve().parent.parent.parent
@skip_unless_docker()
class TestDryRunPlan(unittest.TestCase):
def test_dry_run_emits_structured_plan(self):
work_dir = Path(tempfile.mkdtemp())
try:
# PRD 0011 layout: per-file MD under .claude-bottle/.
# work_dir doubles as $HOME and as cwd for this test.
cb = work_dir / ".claude-bottle"
(cb / "bottles").mkdir(parents=True)
(cb / "agents").mkdir(parents=True)
(cb / "bottles" / "dev.md").write_text(
"---\n"
"egress:\n"
" allowlist:\n"
" - example.org\n"
"---\n"
)
(cb / "agents" / "demo.md").write_text(
"---\n"
"bottle: dev\n"
"---\n"
)
# Under act_runner with a host-mounted docker socket, the
# `docker network ls` / `docker ps -a` calls from inside the
# job container exit non-zero (see docs/ci.md for the same
# topology issue affecting other integration tests). Skip
# the side-effects guard there; locally the check still
# catches accidental docker resource creation by the dry
# run.
check_side_effects = os.environ.get("GITEA_ACTIONS") != "true"
nets_before = self._count_claude_bottle_networks() if check_side_effects else 0
ctrs_before = self._count_claude_bottle_containers() if check_side_effects else 0
env = os.environ.copy()
env["HOME"] = str(work_dir)
env.pop("CLAUDE_BOTTLE_DRY_RUN", None)
# The HOME override above isolates the manifest under test
# from the dev's real ~/claude-bottle.json. On Docker Desktop
# that same override breaks docker CLI endpoint resolution,
# since the active context lives in $HOME/.docker/config.json
# and the per-user socket sits under $HOME/.docker/run/.
# Pin DOCKER_HOST to the parent's resolved endpoint so the
# subprocess reaches the same daemon regardless of $HOME.
endpoint = subprocess.run(
["docker", "context", "inspect",
"--format", "{{.Endpoints.docker.Host}}"],
capture_output=True, text=True, check=True,
).stdout.strip()
if endpoint:
env["DOCKER_HOST"] = endpoint
result = subprocess.run(
[
sys.executable, str(REPO_ROOT / "cli.py"),
"start", "--dry-run", "--format", "json", "demo",
],
cwd=work_dir,
env=env,
capture_output=True,
text=True,
check=False,
)
self.assertEqual(
0, result.returncode,
f"start --dry-run failed: stderr={result.stderr}",
)
plan = json.loads(result.stdout)
self.assertEqual("demo", plan["agent"])
self.assertEqual("dev", plan["bottle"])
self.assertEqual("runc", plan["runtime"],
"runsc isn't available on the CI runner")
self.assertEqual([], plan["skills"])
self.assertEqual([], plan["git_remotes"])
self.assertEqual([], plan["git_gate"])
self.assertEqual(False, plan["remote_control"])
self.assertEqual(0, plan["prompt"]["length"])
# User-declared host + a baked default both present; the union
# is sorted and deduplicated.
hosts = plan["egress"]["hosts"]
self.assertIn("example.org", hosts)
self.assertIn("api.anthropic.com", hosts)
self.assertEqual(plan["egress"]["host_count"], len(hosts))
self.assertEqual(sorted(set(hosts)), hosts,
"hosts must be sorted and deduplicated")
# PRD 0006: TLS interception is on for every launched
# bottle. Fingerprint is null at dry-run (no CA exists
# yet); real launches log it from provision_ca.
self.assertEqual(
{"enabled": True, "ca_fingerprint": None},
plan["egress"]["tls_interception"],
)
# No Docker side effects (see the GITEA_ACTIONS skip note
# above — this guard runs locally only).
if check_side_effects:
self.assertEqual(nets_before, self._count_claude_bottle_networks(),
"no networks created")
self.assertEqual(ctrs_before, self._count_claude_bottle_containers(),
"no containers created")
finally:
import shutil
shutil.rmtree(work_dir, ignore_errors=True)
def _count_claude_bottle_networks(self) -> int:
return self._count_with_prefix(
["docker", "network", "ls", "--format", "{{.Name}}"], "claude-bottle"
)
def _count_claude_bottle_containers(self) -> int:
return self._count_with_prefix(
["docker", "ps", "-a", "--format", "{{.Names}}"], "claude-bottle"
)
def _count_with_prefix(self, cmd: list[str], prefix: str) -> int:
# capture_output + explicit returncode check so a docker
# failure surfaces its stderr in the test report instead of
# the bare CalledProcessError we used to get.
result = subprocess.run(cmd, capture_output=True, text=True, check=False)
if result.returncode != 0:
self.fail(
f"{' '.join(cmd)!r} failed (exit {result.returncode}): "
f"stderr={result.stderr.strip()!r}"
)
return sum(1 for n in result.stdout.splitlines() if n.startswith(prefix))
if __name__ == "__main__":
unittest.main()
tests/unit/test_cli_start_format.py
-54
View File
@@ -1,54 +0,0 @@
"""Unit: argparse-level CLI checks for `start --format`.
Lives in tests/unit/ because nothing here touches Docker — the CLI
exits at argument-validation time before any backend code runs.
"""
import json
import os
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path
REPO_ROOT = Path(__file__).resolve().parent.parent.parent
class TestStartFormatFlag(unittest.TestCase):
def test_json_format_requires_dry_run(self):
"""Emitting JSON in a real run would race the y/N prompt; the
CLI must reject the combination with a message that names the
offending flag."""
work_dir = Path(tempfile.mkdtemp())
try:
(work_dir / "claude-bottle.json").write_text(json.dumps({
"bottles": {"dev": {}},
"agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}},
}))
env = os.environ.copy()
env["HOME"] = str(work_dir)
env.pop("CLAUDE_BOTTLE_DRY_RUN", None)
result = subprocess.run(
[
sys.executable, str(REPO_ROOT / "cli.py"),
"start", "--format", "json", "demo",
],
cwd=work_dir,
env=env,
capture_output=True,
text=True,
check=False,
)
self.assertNotEqual(0, result.returncode)
self.assertIn(
"--format=json requires --dry-run", result.stderr,
f"expected the flag-conflict message; got stderr={result.stderr!r}",
)
finally:
import shutil
shutil.rmtree(work_dir, ignore_errors=True)
if __name__ == "__main__":
unittest.main()