Compare commits

...

4 Commits

Author SHA1 Message Date
Quality Badge Bot 5970b785aa chore: update quality badges
- Coverage: 83%
- Core coverage: 95%

[skip ci]
2026-07-01 16:51:08 +00:00
didericis 2f5cf81cf5 fix(git-gate): defer dynamic key provisioning
lint / lint (push) Successful in 1m59s
test / unit (push) Successful in 49s
test / integration (push) Successful in 23s
test / coverage (push) Successful in 1m0s
Update Quality Badges / update-badges (push) Successful in 53s
2026-07-01 12:45:46 -04:00
didericis 4a1e667306 fix(git-gate): inline GIT_GATE_TIMEOUT_SECS to fix git-http ImportError
lint / lint (push) Successful in 1m56s
test / unit (push) Successful in 48s
test / integration (push) Successful in 20s
test / coverage (push) Successful in 1m2s
Update Quality Badges / update-badges (push) Successful in 52s
git_http_backend.py is copied flat into the sidecar bundle image as a
standalone script, not as part of the bot_bottle package, and
git_gate.py/git_gate_render.py are never copied in. Its relative
import of GIT_GATE_TIMEOUT_SECS crashed the git-http daemon (port
9420) on every startup, silently leaving the smart-HTTP git-gate
transport down while the other sidecar daemons stayed up.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-07-01 11:53:26 -04:00
didericis b93fe58523 feat(cli): add headless launch mode for orchestrators
test / unit (pull_request) Successful in 47s
test / integration (pull_request) Successful in 16s
test / coverage (pull_request) Successful in 1m4s
lint / lint (push) Successful in 2m0s
test / unit (push) Successful in 48s
test / integration (push) Successful in 18s
test / coverage (push) Successful in 57s
Update Quality Badges / update-badges (push) Successful in 57s
`--headless` is a non-interactive launch path for `cli.py start`:
agent, bottles, label, and color come from flags + manifest defaults
with no TUI selectors and no y/N preflight (auto-confirmed via a new
`assume_yes` param threaded into the shared `_launch_bottle` core).

- `--bottle` (repeatable) defaults to the agent's own `bottle:`;
  `--label` defaults to the agent name and auto-uniquifies on slug
  collision; `--color` defaults to none.
- `--prompt TEXT` is required in headless mode and is delivered to the
  agent via a new `headless_prompt(prompt)` method on `AgentProvider`,
  implemented for claude (`-p`), codex (positional), and pi (`-p`).
- The agent still execs on inherited stdio/PTY, so whatever allocates
  the PTY drives the live session; only the launch chrome is headless.
- `--headless --dry-run` previews the resolved plan without launching.

Adds unit coverage in tests/unit/test_cli_start_headless.py and
headless_prompt tests for each provider. Also stubs headless_prompt on
the in-test AgentProvider subclasses so the unit suite collects cleanly.

Closes #315.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01WL77TgFxKbs3cidGMG9dz7
2026-06-30 15:08:14 -04:00
19 changed files with 503 additions and 25 deletions
+2 -2
View File
@@ -5,8 +5,8 @@
# 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)
[![coverage](https://img.shields.io/badge/coverage-84%25-brightgreen)](https://coverage.readthedocs.io/)
[![core coverage](https://img.shields.io/badge/core%20coverage-96%25-brightgreen)](https://gitea.dideric.is/didericis/bot-bottle/src/branch/main/docs/decisions/0004-coverage-policy.md)
[![coverage](https://img.shields.io/badge/coverage-83%25-brightgreen)](https://coverage.readthedocs.io/)
[![core coverage](https://img.shields.io/badge/core%20coverage-95%25-brightgreen)](https://gitea.dideric.is/didericis/bot-bottle/src/branch/main/docs/decisions/0004-coverage-policy.md)
**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.
+9
View File
@@ -209,6 +209,15 @@ class AgentProvider(ABC):
the supervise sidecar is reachable. No-op when
`plan.supervise_plan is None`."""
@abstractmethod
def headless_prompt(self, prompt: str) -> list[str]:
"""Return the agent CLI args that deliver `prompt` as the
initial task in a non-interactive (headless) session.
Called only when ``--prompt`` is passed to
``./cli.py start --headless``; the returned args are appended
after the provider's ``bypass_args`` and ``startup_args``."""
def provision_ca(self, bottle: "Bottle", plan: "BottlePlan") -> None:
"""Install the egress MITM CA into the agent's trust store.
+9 -1
View File
@@ -37,7 +37,10 @@ from pathlib import Path
from typing import Callable, Generator
from ...egress import egress_resolve_token_values
from ...git_gate import revoke_git_gate_provisioned_keys
from ...git_gate import (
provision_git_gate_dynamic_keys,
revoke_git_gate_provisioned_keys,
)
from ...log import info, warn
from . import network as network_mod
from . import util as docker_mod
@@ -118,6 +121,11 @@ def launch(
git_gate_plan = plan.git_gate_plan
if git_gate_plan.upstreams:
git_gate_plan = provision_git_gate_dynamic_keys(
plan.manifest.bottle,
git_gate_plan,
git_gate_state_dir(plan.slug),
)
git_gate_plan = dataclasses.replace(
git_gate_plan,
internal_network=internal_network,
+19 -1
View File
@@ -28,7 +28,10 @@ from ...egress import (
egress_resolve_token_values,
egress_sidecar_env_entries,
)
from ...git_gate import revoke_git_gate_provisioned_keys
from ...git_gate import (
provision_git_gate_dynamic_keys,
revoke_git_gate_provisioned_keys,
)
from ...log import die, info, warn
from ...supervise import QUEUE_DIR_IN_CONTAINER, SUPERVISE_PORT
from ...util import expand_tilde
@@ -98,6 +101,8 @@ def launch(
egress_network = egress_network_name(plan.slug)
_create_networks(internal_network, egress_network, stack)
plan = _provision_git_gate_keys(plan)
sidecar_name = sidecar_container_name(plan.slug)
container_mod.force_remove_container(sidecar_name)
_start_sidecar_bundle(plan, sidecar_name, internal_network, egress_network)
@@ -241,6 +246,19 @@ def _stamp_agent_urls(
)
def _provision_git_gate_keys(
plan: MacosContainerBottlePlan,
) -> MacosContainerBottlePlan:
if not plan.git_gate_plan.upstreams:
return plan
git_gate_plan = provision_git_gate_dynamic_keys(
plan.manifest.bottle,
plan.git_gate_plan,
git_gate_state_dir(plan.slug),
)
return dataclasses.replace(plan, git_gate_plan=git_gate_plan)
def _stage_git_gate(plan: MacosContainerBottlePlan, sidecar_name: str) -> None:
gp = plan.git_gate_plan
if not gp.upstreams:
+18 -1
View File
@@ -41,7 +41,10 @@ from ..docker.git_gate import (
GIT_GATE_ENTRYPOINT_IN_CONTAINER,
GIT_GATE_HOOK_IN_CONTAINER,
)
from ...git_gate import revoke_git_gate_provisioned_keys
from ...git_gate import (
provision_git_gate_dynamic_keys,
revoke_git_gate_provisioned_keys,
)
from ...log import info, warn
from ...bottle_state import (
egress_state_dir,
@@ -174,6 +177,7 @@ def _start_bundle(
) -> SmolmachinesBottlePlan:
"""Build the BundleLaunchSpec, resolve token env, start the
sidecar bundle container, and register teardown."""
plan = _provision_git_gate_keys(plan)
bundle_spec = _bundle_launch_spec(plan, network, loopback_ip)
token_env = _resolve_token_env(plan, dict(os.environ))
_bundle.ensure_bundle_image(bundle_spec.image)
@@ -182,6 +186,19 @@ def _start_bundle(
return plan
def _provision_git_gate_keys(
plan: SmolmachinesBottlePlan,
) -> SmolmachinesBottlePlan:
if not plan.git_gate_plan.upstreams:
return plan
git_gate_plan = provision_git_gate_dynamic_keys(
plan.manifest.bottle,
plan.git_gate_plan,
git_gate_state_dir(plan.slug),
)
return dataclasses.replace(plan, git_gate_plan=git_gate_plan)
def _discover_urls(
plan: SmolmachinesBottlePlan,
loopback_ip: str,
+142 -7
View File
@@ -2,6 +2,11 @@
interactive claude-code session. The container is torn down when the
session ends.
`--headless` selects a non-interactive launch (agent/bottles/label from
flags, no TUI selectors, no y/N prompt) for orchestrators,
CI, and webhook dispatch. The agent still execs on the inherited
stdio/PTY, so an orchestrator that allocates the PTY drives the session.
The launch core is shared with `cli.py resume <identity>` through
the private orchestrator `_launch_bottle`.
"""
@@ -16,7 +21,7 @@ import tempfile
from pathlib import Path
from typing import Callable
from ..agent_provider import runtime_for
from ..agent_provider import get_provider, runtime_for
from ..backend import (
Bottle,
BottleSpec,
@@ -31,7 +36,7 @@ from ..bottle_state import (
is_preserved,
mark_preserved,
)
from ..log import info
from ..log import info, die
from ..manifest import Manifest, ManifestIndex
from ._common import PROG, USER_CWD, read_tty_line
from . import tui
@@ -50,6 +55,39 @@ def cmd_start(argv: list[str]) -> int:
"or host auto-selection). Overrides the env var when set."
),
)
parser.add_argument(
"--headless",
action="store_true",
help=(
"non-interactive launch: take agent/bottles/label from flags, "
"skip all prompts. For orchestrators, CI, and webhooks."
),
)
parser.add_argument(
"--bottle",
action="append",
default=None,
metavar="NAME",
help=(
"bottle to compose, repeatable (order = merge order). In "
"--headless, defaults to the agent's own bottle when omitted."
),
)
parser.add_argument(
"--label",
default=None,
help="bottle label / terminal title (--headless default: agent name)",
)
parser.add_argument(
"--color",
default=None,
help="bottle color, one of the 16 ANSI color names (--headless default: none)",
)
parser.add_argument(
"--prompt",
default=None,
help="initial task prompt delivered to the agent (required with --headless)",
)
parser.add_argument(
"name",
nargs="?",
@@ -61,6 +99,12 @@ def cmd_start(argv: list[str]) -> int:
dry_run = args.dry_run or os.environ.get("BOT_BOTTLE_DRY_RUN") == "1"
manifest = ManifestIndex.resolve(USER_CWD)
backend_name: str | None = args.backend
if args.headless:
return _start_headless(
manifest, args, dry_run=dry_run, backend_name=backend_name
)
agent_name: str | None = args.name
if agent_name is None:
@@ -71,8 +115,6 @@ def cmd_start(argv: list[str]) -> int:
if agent_name is None:
return 0
backend_name: str | None = args.backend
# Bottle multiselect: always show after agent selection so operators
# can compose bottles at launch time without editing agent manifests.
available_bottles = manifest.all_bottle_names
@@ -109,6 +151,83 @@ def cmd_start(argv: list[str]) -> int:
)
# --- Headless launch -----------------------------------------------------
def _start_headless(
manifest: ManifestIndex,
args: argparse.Namespace,
*,
dry_run: bool,
backend_name: str | None,
) -> int:
"""Non-interactive launch path for orchestrators / CI / webhooks.
Resolves agent, bottles, label, and color from flags + manifest
defaults instead of the TUI selectors, and auto-confirms the
preflight. Otherwise runs the same launch core as the interactive
path, so the agent still execs on the inherited stdio/PTY an
orchestrator allocates that PTY and relays it to its
desktop/mobile clients."""
agent_name = args.name
if not agent_name:
die("--headless requires an agent name: ./cli.py start <agent> --headless")
manifest.require_agent(agent_name) # raises ManifestError if unknown
prompt = args.prompt
if not prompt:
die(
"--headless requires --prompt: "
"./cli.py start <agent> --headless --prompt 'Do the thing'"
)
if args.bottle:
bottle_names: tuple[str, ...] = tuple(args.bottle)
else:
default_bottle = _peek_agent_bottle(manifest, agent_name)
if not default_bottle:
die(
f"--headless: agent '{agent_name}' has no default bottle; "
f"pass one or more --bottle NAME"
)
bottle_names = (default_bottle,)
label = _uniquify_label_headless(args.label or agent_name)
spec = BottleSpec(
manifest=manifest,
agent_name=agent_name,
copy_cwd=args.cwd,
user_cwd=USER_CWD,
label=label,
color=args.color or "",
bottle_names=bottle_names,
)
return _launch_bottle(
spec,
dry_run=dry_run,
backend_name=backend_name,
assume_yes=True,
headless_prompt_text=prompt,
)
def _uniquify_label_headless(label: str) -> str:
"""Non-interactive analog of `_resolve_unique_label`: if the label's
slug collides with a running bottle, append -2, -3, until free,
logging the chosen label. Orchestrators fire-and-forget many bottles,
so silently picking a free name beats erroring on every collision."""
active_slugs = {a.slug for a in enumerate_active_agents()}
if docker_mod.slugify(label) not in active_slugs:
return label
n = 2
while docker_mod.slugify(f"{label}-{n}") in active_slugs:
n += 1
chosen = f"{label}-{n}"
info(f"label '{label}' already in use; using '{chosen}'")
return chosen
# --- Launch helpers ------------------------------------------------------
@@ -376,10 +495,19 @@ def _launch_bottle(
*,
dry_run: bool,
backend_name: str | None = None,
assume_yes: bool = False,
headless_prompt_text: str = "",
) -> int:
"""Shared launch core for `start` and `resume`. Builds the plan,
prints / dry-runs / prompts as appropriate, brings the bottle up,
attaches claude, and prints the resume hint on session end."""
attaches claude, and prints the resume hint on session end.
`assume_yes` skips the interactive y/N confirmation (headless /
orchestrator launches), where there is no human at the prompt.
`headless_prompt_text` is passed to the provider's `headless_prompt`
method and the resulting args are appended to startup_args so the
agent receives the initial task without interactive input."""
stage_dir = Path(tempfile.mkdtemp(prefix="bot-bottle-stage."))
identity = ""
try:
@@ -387,7 +515,7 @@ def _launch_bottle(
spec,
stage_dir=stage_dir,
render_preflight=_text_render_preflight(),
prompt_yes=_text_prompt_yes,
prompt_yes=(lambda: True) if assume_yes else _text_prompt_yes,
dry_run=dry_run,
backend_name=backend_name,
)
@@ -397,10 +525,17 @@ def _launch_bottle(
backend = get_bottle_backend(backend_name)
with backend.launch(plan) as bottle:
agent_provider_template = getattr(plan, "agent_provider_template", "claude")
extra_args: tuple[str, ...] = ()
if headless_prompt_text:
extra_args = tuple(
get_provider(agent_provider_template).headless_prompt(
headless_prompt_text
)
)
exit_code = attach_agent(
bottle,
agent_provider_template=agent_provider_template,
startup_args=plan.agent_provision.startup_args,
startup_args=plan.agent_provision.startup_args + extra_args,
)
info(
f"session ended (exit {exit_code}); "
@@ -313,6 +313,9 @@ class ClaudeAgentProvider(AgentProvider):
f"claude mcp add --scope user --transport http supervise {supervise_url}"
)
def headless_prompt(self, prompt: str) -> list[str]:
return ["-p", prompt]
def _exec(bottle: "Bottle", script: str, error: str) -> None:
result = bottle.exec(script, user="root")
@@ -279,6 +279,9 @@ class CodexAgentProvider(AgentProvider):
f"codex mcp add supervise --url {shlex.quote(supervise_url)}"
)
def headless_prompt(self, prompt: str) -> list[str]:
return [prompt]
def _exec(bottle: "Bottle", script: str, error: str) -> None:
result = bottle.exec(script, user="root")
+3
View File
@@ -315,6 +315,9 @@ class PiAgentProvider(AgentProvider):
) -> None:
del plan, bottle, supervise_url
def headless_prompt(self, prompt: str) -> list[str]:
return ["-p", prompt]
def _exec(bottle: "Bottle", script: str, error: str) -> None:
result = bottle.exec(script, user="root")
+6 -11
View File
@@ -30,7 +30,6 @@ backend-specific and lives on concrete subclasses (see
from __future__ import annotations
import dataclasses
from abc import ABC
from dataclasses import dataclass
from pathlib import Path
@@ -53,6 +52,7 @@ from .git_gate_render import (
_gitconfig_validate_value,
)
from .git_gate_provision import (
provision_git_gate_dynamic_keys,
revoke_git_gate_provisioned_keys,
_provision_dynamic_key,
_resolve_identity_file,
@@ -93,20 +93,14 @@ class GitGate(ABC):
entrypoint, pre-receive hook, and access-hook scripts (mode
600) under `stage_dir`. Pure host-side, no docker subprocess.
For `gitea` key entries, also generates and registers
a fresh deploy key via the forge API and writes the private key
+ key ID to `stage_dir`.
For `gitea` key entries, the returned upstream intentionally
has an empty identity file. Backend launch fills that in after
the operator confirms the preflight.
Returned plan is incomplete: the launch step must fill
`internal_network` / `egress_network` via `dataclasses.replace`
before passing the plan to `.start`."""
upstreams_list = list(git_gate_upstreams_for_bottle(bottle))
for i, entry in enumerate(bottle.git):
upstreams_list[i] = dataclasses.replace(
upstreams_list[i],
identity_file=_resolve_identity_file(entry, slug, stage_dir),
)
upstreams = tuple(upstreams_list)
upstreams = git_gate_upstreams_for_bottle(bottle)
entrypoint = stage_dir / "git_gate_entrypoint.sh"
entrypoint.write_text(git_gate_render_entrypoint(upstreams))
entrypoint.chmod(0o600)
@@ -162,6 +156,7 @@ __all__ = [
"git_gate_render_entrypoint",
"git_gate_render_hook",
"git_gate_render_access_hook",
"provision_git_gate_dynamic_keys",
"revoke_git_gate_provisioned_keys",
"_gitconfig_validate_value",
"_provision_dynamic_key",
+43
View File
@@ -9,10 +9,16 @@ imported (`deploy_key_provisioner`) to keep its cost off the host path.
from __future__ import annotations
import os
import dataclasses
from pathlib import Path
from typing import TYPE_CHECKING
from .log import info
from .manifest import ManifestBottle, ManifestGitEntry
from .git_gate_render import GitGateUpstream
if TYPE_CHECKING:
from .git_gate import GitGatePlan
def _provision_dynamic_key(
entry: ManifestGitEntry,
@@ -95,8 +101,45 @@ def _resolve_identity_file(entry: ManifestGitEntry, slug: str, stage_dir: Path)
return entry.IdentityFile
def provision_git_gate_dynamic_keys(
bottle: ManifestBottle,
plan: "GitGatePlan",
stage_dir: Path,
) -> "GitGatePlan":
"""Provision dynamic git-gate keys and return an updated plan.
This runs during backend launch, after the operator confirms the
preflight. Plan preparation intentionally stays side-effect-light:
dry-runs and aborted launches must not create remote deploy keys.
"""
if not plan.upstreams:
return plan
upstreams_by_name: dict[str, GitGateUpstream] = {
upstream.name: upstream for upstream in plan.upstreams
}
updated: list[GitGateUpstream] = []
for entry in bottle.git:
upstream = upstreams_by_name.get(entry.Name)
if upstream is None:
continue
if entry.Key.provider == "gitea":
identity_file = _provision_dynamic_key(entry, plan.slug, stage_dir)
upstream = dataclasses.replace(upstream, identity_file=identity_file)
updated.append(upstream)
if len(updated) != len(plan.upstreams):
updated_names = {u.name for u in updated}
for upstream in plan.upstreams:
if upstream.name not in updated_names:
updated.append(upstream)
return dataclasses.replace(plan, upstreams=tuple(updated))
__all__ = [
"revoke_git_gate_provisioned_keys",
"provision_git_gate_dynamic_keys",
"_provision_dynamic_key",
"_resolve_identity_file",
]
+7 -2
View File
@@ -16,11 +16,16 @@ from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.parse import urlsplit
from .git_gate import GIT_GATE_TIMEOUT_SECS
DEFAULT_PORT = 9420
# Mirrors git_gate_render.GIT_GATE_TIMEOUT_SECS. Duplicated rather than
# imported: this module ships as a flat top-level sibling in the sidecar
# bundle image (see Dockerfile.sidecars), not as part of the bot_bottle
# package, so `bot_bottle.git_gate` and its dependency chain aren't
# available at runtime.
GIT_GATE_TIMEOUT_SECS = 15
# Bound memory use while still allowing ordinary git push packfiles.
MAX_BODY_BYTES = 100 * 1024 * 1024
+188
View File
@@ -0,0 +1,188 @@
"""Unit: `cli.py start --headless` non-interactive launch path.
Headless is the keystone for orchestrators, CI, and webhook
dispatch: agent/bottles/label come from flags + manifest defaults, no
TUI selectors fire, and the preflight y/N is auto-confirmed
(`assume_yes=True`). All actual launch work is stubbed so no container
is created.
"""
from __future__ import annotations
import os
import unittest
from unittest.mock import MagicMock, patch
import bot_bottle.cli.start as start_mod
import bot_bottle.cli.tui as tui_mod
from bot_bottle.backend import ActiveAgent
from bot_bottle.log import Die
from bot_bottle.manifest import ManifestError
def _make_manifest(
agent_names: list[str],
bottle_names: list[str] | None = None,
agent_bottle: str = "",
):
manifest = MagicMock()
manifest.agents = {name: MagicMock(bottle=agent_bottle) for name in agent_names}
manifest.all_agent_names = sorted(agent_names)
manifest.all_bottle_names = sorted(bottle_names or [])
manifest.home_md = None # eager mode so _peek_agent_bottle uses agents dict
manifest.require_agent = MagicMock(return_value=None)
return manifest
def _active_agent(slug: str) -> ActiveAgent:
return ActiveAgent(
backend_name="docker",
slug=slug,
agent_name="demo",
started_at="2026-01-01T00:00:00+00:00",
services=(),
)
class TestCmdStartHeadless(unittest.TestCase):
"""Drive `cmd_start --headless` with launch + TUI stubbed out."""
def setUp(self):
self._manifest = _make_manifest(
["researcher", "implementer"], ["claude", "dev"], agent_bottle="claude"
)
patch(
"bot_bottle.cli.start.ManifestIndex.resolve",
return_value=self._manifest,
).start()
self._launch_mock = patch(
"bot_bottle.cli.start._launch_bottle", return_value=0
).start()
# No bottles running by default → no label collision.
patch(
"bot_bottle.cli.start.enumerate_active_agents", return_value=[]
).start()
# If any TUI picker fires in headless mode, that's a bug.
self._agent_picker = patch.object(tui_mod, "filter_select").start()
self._bottle_picker = patch.object(tui_mod, "filter_multiselect").start()
self._modal = patch.object(tui_mod, "name_color_modal").start()
patch.dict(os.environ, {}, clear=False).start()
os.environ.pop("BOT_BOTTLE_BACKEND", None)
self.addCleanup(patch.stopall)
def _spec(self):
self._launch_mock.assert_called_once()
return self._launch_mock.call_args[0][0]
# -- no TUI in headless --------------------------------------------
def test_headless_fires_no_pickers(self):
rc = start_mod.cmd_start(
["--headless", "researcher", "--bottle", "claude", "--prompt", "Do it"]
)
self.assertEqual(0, rc)
self._agent_picker.assert_not_called()
self._bottle_picker.assert_not_called()
self._modal.assert_not_called()
def test_headless_assume_yes_forwarded(self):
start_mod.cmd_start(
["--headless", "researcher", "--bottle", "claude", "--prompt", "Do it"]
)
self.assertTrue(self._launch_mock.call_args[1]["assume_yes"])
# -- prompt --------------------------------------------------------
def test_headless_without_prompt_dies(self):
with self.assertRaises(Die):
start_mod.cmd_start(["--headless", "researcher", "--bottle", "claude"])
self._launch_mock.assert_not_called()
def test_headless_prompt_forwarded_to_launch(self):
start_mod.cmd_start(
["--headless", "researcher", "--bottle", "claude",
"--prompt", "Implement issue #42"]
)
self.assertEqual(
"Implement issue #42",
self._launch_mock.call_args[1]["headless_prompt_text"],
)
# -- bottle resolution ---------------------------------------------
def test_explicit_bottles_forwarded_in_order(self):
start_mod.cmd_start(
["--headless", "researcher", "--bottle", "dev", "--bottle", "claude",
"--prompt", "Do it"]
)
self.assertEqual(("dev", "claude"), self._spec().bottle_names)
def test_omitted_bottle_falls_back_to_agent_default(self):
start_mod.cmd_start(["--headless", "implementer", "--prompt", "Do it"])
self.assertEqual(("claude",), self._spec().bottle_names)
def test_no_bottle_and_no_default_dies(self):
manifest = _make_manifest(["researcher"], ["claude"], agent_bottle="")
with patch(
"bot_bottle.cli.start.ManifestIndex.resolve", return_value=manifest
):
with self.assertRaises(Die):
start_mod.cmd_start(
["--headless", "researcher", "--prompt", "Do it"]
)
self._launch_mock.assert_not_called()
# -- agent resolution ----------------------------------------------
def test_missing_agent_name_dies(self):
with self.assertRaises(Die):
start_mod.cmd_start(["--headless"])
self._launch_mock.assert_not_called()
def test_unknown_agent_raises_manifest_error(self):
self._manifest.require_agent.side_effect = ManifestError("agent 'x' not defined")
with self.assertRaises(ManifestError):
start_mod.cmd_start(
["--headless", "x", "--bottle", "claude", "--prompt", "Do it"]
)
self._launch_mock.assert_not_called()
# -- label / color -------------------------------------------------
def test_label_defaults_to_agent_name(self):
start_mod.cmd_start(
["--headless", "researcher", "--bottle", "claude", "--prompt", "Do it"]
)
self.assertEqual("researcher", self._spec().label)
def test_explicit_label_and_color_forwarded(self):
start_mod.cmd_start(
["--headless", "researcher", "--bottle", "claude",
"--label", "nightly", "--color", "green", "--prompt", "Do it"]
)
spec = self._spec()
self.assertEqual("nightly", spec.label)
self.assertEqual("green", spec.color)
def test_label_collision_uniquifies(self):
with patch(
"bot_bottle.cli.start.enumerate_active_agents",
return_value=[_active_agent("researcher")],
):
start_mod.cmd_start(
["--headless", "researcher", "--bottle", "claude", "--prompt", "Do it"]
)
self.assertEqual("researcher-2", self._spec().label)
# -- backend wiring ------------------------------------------------
def test_backend_flag_forwarded(self):
start_mod.cmd_start(
["--headless", "--backend=docker", "researcher", "--bottle", "claude",
"--prompt", "Do it"]
)
self.assertEqual("docker", self._launch_mock.call_args[1]["backend_name"])
if __name__ == "__main__":
unittest.main()
@@ -343,5 +343,14 @@ class TestClaudeSuperviseMcp(unittest.TestCase):
)
class TestClaudeHeadlessPrompt(unittest.TestCase):
def test_returns_p_flag_and_prompt(self):
self.assertEqual(["-p", "Do the task"], ClaudeAgentProvider().headless_prompt("Do the task"))
def test_preserves_prompt_text_verbatim(self):
text = "Fix issue #42: the widget breaks on empty input"
self.assertEqual(["-p", text], ClaudeAgentProvider().headless_prompt(text))
if __name__ == "__main__":
unittest.main()
@@ -314,5 +314,14 @@ class TestCodexSuperviseMcp(unittest.TestCase):
)
class TestCodexHeadlessPrompt(unittest.TestCase):
def test_returns_prompt_as_positional_arg(self):
self.assertEqual(["Do the task"], CodexAgentProvider().headless_prompt("Do the task"))
def test_preserves_prompt_text_verbatim(self):
text = "Fix issue #42: the widget breaks on empty input"
self.assertEqual([text], CodexAgentProvider().headless_prompt(text))
if __name__ == "__main__":
unittest.main()
+9
View File
@@ -223,5 +223,14 @@ class TestPiDockerfile(unittest.TestCase):
self.assertIn("chmod 1777 /tmp /var/tmp", dockerfile)
class TestPiHeadlessPrompt(unittest.TestCase):
def test_returns_p_flag_and_prompt(self):
self.assertEqual(["-p", "Do the task"], PiAgentProvider().headless_prompt("Do the task"))
def test_preserves_prompt_text_verbatim(self):
text = "Fix issue #42: the widget breaks on empty input"
self.assertEqual(["-p", text], PiAgentProvider().headless_prompt(text))
if __name__ == "__main__":
unittest.main()
@@ -38,6 +38,7 @@ class _Provider(AgentProvider):
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]
def headless_prompt(self, prompt): return [] # type: ignore[override]
_PROVIDER = _Provider()
+22
View File
@@ -14,6 +14,7 @@ from bot_bottle.git_gate import (
git_gate_render_access_hook,
git_gate_render_entrypoint,
git_gate_render_hook,
provision_git_gate_dynamic_keys,
revoke_git_gate_provisioned_keys,
_resolve_identity_file,
git_gate_upstreams_for_bottle,
@@ -371,6 +372,27 @@ class TestDynamicKeyProvisioning(unittest.TestCase):
self.assertEqual("/tmp/provisioned-key", _resolve_identity_file(entry, "demo", self.stage))
mock_provision.assert_called_once()
def test_prepare_defers_gitea_key_provisioning(self):
bottle = self._gitea_manifest().bottles["dev"]
with patch("bot_bottle.git_gate_provision._provision_dynamic_key") as mock_provision:
plan = _StubGate().prepare(bottle, "demo", self.stage)
mock_provision.assert_not_called()
self.assertEqual("", plan.upstreams[0].identity_file)
def test_launch_time_helper_provisions_gitea_keys(self):
bottle = self._gitea_manifest().bottles["dev"]
plan = _StubGate().prepare(bottle, "demo", self.stage)
with patch(
"bot_bottle.git_gate_provision._provision_dynamic_key",
return_value="/tmp/provisioned-key",
) as mock_provision:
updated = provision_git_gate_dynamic_keys(bottle, plan, self.stage)
mock_provision.assert_called_once_with(bottle.git[0], "demo", self.stage)
self.assertEqual("/tmp/provisioned-key", updated.upstreams[0].identity_file)
def test_revoke_skips_non_gitea_and_missing_id_file(self):
revoke_git_gate_provisioned_keys(fixture_with_git().bottles["dev"], self.stage)
@@ -49,6 +49,7 @@ class _Provider(AgentProvider):
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]
def headless_prompt(self, prompt): return [] # type: ignore[override]
_PROVIDER = _Provider()