didericis-claude
4d51b1aa02
tracker-policy-pr / check-pr (pull_request) Successful in 9s
lint / lint (push) Successful in 2m52s
test / integration-docker (pull_request) Successful in 14s
test / unit (pull_request) Successful in 39s
test / integration-firecracker (pull_request) Successful in 3m45s
test / coverage (pull_request) Successful in 1m0s
test / publish-infra (pull_request) Has been skipped
The previous fix interpolated the real per-bottle identity token into the `warn()` message shown when `claude mcp add supervise` fails, so a registration failure would leak the credential into host terminal output and any collected launch logs (#476 review, PR #471 comment). Render a `<bottle-identity-token>` placeholder instead. The recovery hint still shows the required `--header x-bot-bottle-identity: …` shape, and the operator substitutes the value from inside the bottle (it rides in the agent's HTTPS_PROXY credentials) — so the token never reaches the host log. The token truthiness still gates whether the header is shown at all. Test updated to assert the header name and placeholder are present and the raw token is absent. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
394 lines
15 KiB
Python
394 lines
15 KiB
Python
"""Unit: ClaudeAgentProvider provisioning (PRD 0050, contrib/claude).
|
|
|
|
Each provider owns its own in-guest provisioning end-to-end —
|
|
skills copy, prompt copy, declarative dirs/files/pre_copy/verify
|
|
apply, and supervise MCP registration. The Claude / Codex paths
|
|
intentionally don't share a helper module: harness changes on
|
|
either side are expected to diverge the implementations."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from bot_bottle.agent_provider import (
|
|
AgentProvisionCommand,
|
|
AgentProvisionFile,
|
|
AgentProvisionPlan,
|
|
)
|
|
from bot_bottle.backend import Bottle, BottleSpec, ExecResult
|
|
from bot_bottle.backend.docker.bottle_plan import DockerBottlePlan
|
|
from bot_bottle.contrib.claude.agent_provider import ClaudeAgentProvider
|
|
from bot_bottle.egress import EgressPlan
|
|
from bot_bottle.git_gate import GitGatePlan
|
|
from bot_bottle.manifest import ManifestIndex
|
|
from bot_bottle.supervisor.plan import SupervisePlan
|
|
|
|
|
|
_URL = "http://supervise:9100/"
|
|
|
|
|
|
def _make_bottle(exec_result: ExecResult | None = None) -> MagicMock:
|
|
bottle = MagicMock(spec=Bottle)
|
|
bottle.name = "bot-bottle-demo-abc12"
|
|
bottle.exec.return_value = (
|
|
exec_result if exec_result is not None
|
|
else ExecResult(returncode=0, stdout="", stderr="")
|
|
)
|
|
return bottle
|
|
|
|
|
|
def _exec_scripts(bottle: MagicMock) -> list[str]:
|
|
return [c.args[0] for c in bottle.exec.call_args_list]
|
|
|
|
|
|
def _plan(
|
|
*,
|
|
agent_prompt: str = "",
|
|
skills: list[str] | None = None,
|
|
agent_provision: AgentProvisionPlan | None = None,
|
|
supervise: bool = False,
|
|
identity_token: str = "",
|
|
) -> DockerBottlePlan:
|
|
bottle_json: dict = {"agent_provider": {"template": "claude"}} # type: ignore
|
|
if supervise:
|
|
bottle_json["supervise"] = True
|
|
index = ManifestIndex.from_json_obj({
|
|
"bottles": {"dev": bottle_json},
|
|
"agents": {
|
|
"demo": {
|
|
"skills": list(skills or []),
|
|
"prompt": agent_prompt,
|
|
"bottle": "dev",
|
|
},
|
|
},
|
|
})
|
|
manifest = index.load_for_agent("demo")
|
|
spec = BottleSpec(
|
|
manifest=index, agent_name="demo",
|
|
copy_cwd=False, user_cwd="/tmp/x",
|
|
)
|
|
supervise_plan = None
|
|
if supervise:
|
|
supervise_plan = SupervisePlan(
|
|
slug="demo-abc12",
|
|
db_path=Path("/tmp/bot-bottle.db"),
|
|
)
|
|
return DockerBottlePlan(
|
|
spec=spec,
|
|
manifest=manifest,
|
|
stage_dir=Path("/tmp/stage"),
|
|
slug="demo-abc12",
|
|
forwarded_env={},
|
|
git_gate_plan=GitGatePlan(
|
|
slug="demo-abc12",
|
|
entrypoint_script=Path("/tmp/git-gate-entrypoint.sh"),
|
|
hook_script=Path("/tmp/git-gate-hook"),
|
|
access_hook_script=Path("/tmp/git-gate-access-hook"),
|
|
upstreams=(),
|
|
),
|
|
egress_plan=EgressPlan(
|
|
slug="demo-abc12",
|
|
routes_path=Path("/tmp/routes.yaml"),
|
|
routes=(),
|
|
token_env_map={},
|
|
),
|
|
supervise_plan=supervise_plan,
|
|
use_runsc=False,
|
|
identity_token=identity_token,
|
|
agent_provision=agent_provision or AgentProvisionPlan(
|
|
template="claude", command="claude", prompt_mode="append_file",
|
|
image="bot-bottle-claude:latest", dockerfile="",
|
|
guest_home="/home/node",
|
|
instance_name="bot-bottle-demo-abc12",
|
|
prompt_file=Path("/tmp/state/demo-abc12/agent/prompt.txt"),
|
|
guest_env={},
|
|
),
|
|
)
|
|
|
|
|
|
class TestClaudeProvisionPrompt(unittest.TestCase):
|
|
def test_cp_uses_bottle_cp_in(self):
|
|
bottle = _make_bottle()
|
|
ClaudeAgentProvider().provision_prompt(_plan(), bottle)
|
|
bottle.cp_in.assert_called_once_with(
|
|
"/tmp/state/demo-abc12/agent/prompt.txt",
|
|
"/home/node/.bot-bottle-prompt.txt",
|
|
)
|
|
|
|
def test_returns_path_when_agent_has_prompt(self):
|
|
bottle = _make_bottle()
|
|
r = ClaudeAgentProvider().provision_prompt(
|
|
_plan(agent_prompt="You are helpful."), bottle,
|
|
)
|
|
self.assertEqual("/home/node/.bot-bottle-prompt.txt", r)
|
|
|
|
def test_returns_none_when_agent_has_no_prompt(self):
|
|
bottle = _make_bottle()
|
|
r = ClaudeAgentProvider().provision_prompt(_plan(agent_prompt=""), bottle)
|
|
self.assertIsNone(r)
|
|
bottle.cp_in.assert_called_once()
|
|
|
|
def test_returns_path_when_provider_prompt_exists(self):
|
|
bottle = _make_bottle()
|
|
provision = AgentProvisionPlan(
|
|
template="claude", command="claude", prompt_mode="append_file",
|
|
image="", dockerfile="", guest_home="/home/node",
|
|
instance_name="bot-bottle-demo-abc12",
|
|
prompt_file=Path("/tmp/prompt.txt"),
|
|
guest_env={},
|
|
has_prompt=True,
|
|
)
|
|
r = ClaudeAgentProvider().provision_prompt(
|
|
_plan(agent_prompt="", agent_provision=provision), bottle,
|
|
)
|
|
self.assertEqual("/home/node/.bot-bottle-prompt.txt", r)
|
|
|
|
def test_chowns_to_node_after_copy(self):
|
|
bottle = _make_bottle()
|
|
ClaudeAgentProvider().provision_prompt(_plan(), bottle)
|
|
scripts = _exec_scripts(bottle)
|
|
self.assertTrue(
|
|
any("chown node:node" in s
|
|
and "/home/node/.bot-bottle-prompt.txt" in s
|
|
for s in scripts)
|
|
)
|
|
self.assertTrue(
|
|
any("chmod 600" in s
|
|
and "/home/node/.bot-bottle-prompt.txt" in s
|
|
for s in scripts)
|
|
)
|
|
|
|
|
|
class TestClaudeProvisionSkills(unittest.TestCase):
|
|
def test_noop_when_agent_has_no_skills(self):
|
|
bottle = _make_bottle()
|
|
ClaudeAgentProvider().provision_skills(_plan(skills=[]), bottle)
|
|
bottle.cp_in.assert_not_called()
|
|
bottle.exec.assert_not_called()
|
|
|
|
def test_mkdir_plus_cp_per_skill(self):
|
|
bottle = _make_bottle()
|
|
with patch(
|
|
"bot_bottle.backend.util.host_skill_dir",
|
|
side_effect=lambda n: f"/host/skills/{n}", # type: ignore
|
|
), patch(
|
|
"bot_bottle.contrib.claude.agent_provider.os.path.isdir",
|
|
return_value=True,
|
|
):
|
|
ClaudeAgentProvider().provision_skills(
|
|
_plan(skills=["init-prd", "verify"]), bottle,
|
|
)
|
|
scripts = _exec_scripts(bottle)
|
|
self.assertTrue(
|
|
any("mkdir -p" in s and "/home/node/.claude/skills" in s
|
|
for s in scripts)
|
|
)
|
|
cp_targets = {c.args[1] for c in bottle.cp_in.call_args_list}
|
|
self.assertEqual({
|
|
"/home/node/.claude/skills/init-prd/",
|
|
"/home/node/.claude/skills/verify/",
|
|
}, cp_targets)
|
|
self.assertEqual(
|
|
2, sum(1 for s in scripts if "chown -R node:node" in s),
|
|
)
|
|
|
|
def test_missing_skill_dies(self):
|
|
bottle = _make_bottle()
|
|
with patch(
|
|
"bot_bottle.backend.util.host_skill_dir",
|
|
side_effect=lambda n: f"/host/skills/{n}", # type: ignore
|
|
), patch(
|
|
"bot_bottle.contrib.claude.agent_provider.os.path.isdir",
|
|
return_value=False,
|
|
):
|
|
with self.assertRaises(SystemExit):
|
|
ClaudeAgentProvider().provision_skills(
|
|
_plan(skills=["init-prd"]), bottle,
|
|
)
|
|
|
|
|
|
class TestClaudeProvision(unittest.TestCase):
|
|
"""The declarative dirs/files/pre_copy/verify apply loop for
|
|
the claude.json trust marker."""
|
|
|
|
def test_noop_on_empty_provision_plan(self):
|
|
bottle = _make_bottle()
|
|
ClaudeAgentProvider().provision(_plan(), bottle)
|
|
bottle.cp_in.assert_not_called()
|
|
bottle.exec.assert_not_called()
|
|
|
|
def test_copies_files_and_chowns(self):
|
|
provision = AgentProvisionPlan(
|
|
template="claude", command="claude", prompt_mode="append_file",
|
|
image="", dockerfile="", guest_home="/home/node",
|
|
instance_name="bot-bottle-demo-abc12",
|
|
prompt_file=Path("/tmp/prompt.txt"),
|
|
guest_env={},
|
|
files=(AgentProvisionFile(
|
|
Path("/tmp/claude.json"), "/home/node/.claude.json",
|
|
),),
|
|
)
|
|
bottle = _make_bottle()
|
|
ClaudeAgentProvider().provision(
|
|
_plan(agent_provision=provision), bottle,
|
|
)
|
|
bottle.cp_in.assert_called_once_with(
|
|
"/tmp/claude.json", "/home/node/.claude.json",
|
|
)
|
|
scripts = _exec_scripts(bottle)
|
|
self.assertTrue(
|
|
any("chown" in s and "/home/node/.claude.json" in s for s in scripts)
|
|
)
|
|
self.assertTrue(
|
|
any("chmod" in s and "/home/node/.claude.json" in s for s in scripts)
|
|
)
|
|
|
|
def test_dies_when_file_chown_fails(self):
|
|
provision = AgentProvisionPlan(
|
|
template="claude", command="claude", prompt_mode="append_file",
|
|
image="", dockerfile="", guest_home="/home/node",
|
|
instance_name="bot-bottle-demo-abc12",
|
|
prompt_file=Path("/tmp/prompt.txt"),
|
|
guest_env={},
|
|
files=(AgentProvisionFile(
|
|
Path("/tmp/claude.json"), "/home/node/.claude.json",
|
|
),),
|
|
)
|
|
bottle = _make_bottle(
|
|
exec_result=ExecResult(1, "", "chown: no such file\n"),
|
|
)
|
|
with self.assertRaises(SystemExit):
|
|
ClaudeAgentProvider().provision(
|
|
_plan(agent_provision=provision), bottle,
|
|
)
|
|
|
|
|
|
class TestClaudeUiProvision(unittest.TestCase):
|
|
def test_writes_statusline_and_custom_theme_files(self):
|
|
with tempfile.TemporaryDirectory(prefix="bb-claude-ui.") as tmp:
|
|
state_dir = Path(tmp)
|
|
prompt_file = state_dir / "prompt.txt"
|
|
prompt_file.write_text("Existing instructions.\n")
|
|
plan = ClaudeAgentProvider().provision_plan(
|
|
dockerfile="",
|
|
state_dir=state_dir,
|
|
instance_name="bot-bottle-demo-abc12",
|
|
prompt_file=prompt_file,
|
|
label="research-ui",
|
|
color="blue",
|
|
)
|
|
settings = json.loads((state_dir / "claude-settings.json").read_text())
|
|
statusline = (state_dir / "claude-statusline.sh").read_text()
|
|
theme = json.loads((state_dir / "bot-bottle-research-ui.json").read_text())
|
|
prompt_text = prompt_file.read_text()
|
|
self.assertTrue(plan.has_prompt)
|
|
self.assertEqual("Existing instructions.\n", prompt_text)
|
|
self.assertEqual("command", settings["statusLine"]["type"])
|
|
self.assertEqual("~/.claude/statusline.sh", settings["statusLine"]["command"])
|
|
self.assertEqual("custom:bot-bottle-research-ui", settings["theme"])
|
|
self.assertIn("research-ui", statusline)
|
|
self.assertIn("\x1b[94m", statusline)
|
|
self.assertEqual("dark", theme["base"])
|
|
self.assertEqual("ansi:blueBright", theme["overrides"]["claude"])
|
|
|
|
def test_runs_verify_commands(self):
|
|
provision = AgentProvisionPlan(
|
|
template="claude", command="claude", prompt_mode="append_file",
|
|
image="", dockerfile="", guest_home="/home/node",
|
|
instance_name="bot-bottle-demo-abc12",
|
|
prompt_file=Path("/tmp/prompt.txt"),
|
|
guest_env={},
|
|
verify=(AgentProvisionCommand(
|
|
("/usr/bin/true",), "verify failed",
|
|
),),
|
|
)
|
|
bottle = _make_bottle()
|
|
ClaudeAgentProvider().provision(
|
|
_plan(agent_provision=provision), bottle,
|
|
)
|
|
scripts = _exec_scripts(bottle)
|
|
self.assertTrue(any("/usr/bin/true" in s for s in scripts))
|
|
|
|
|
|
class TestClaudeSuperviseMcp(unittest.TestCase):
|
|
def test_noop_when_supervise_disabled(self):
|
|
bottle = _make_bottle()
|
|
ClaudeAgentProvider().provision_supervise_mcp(
|
|
_plan(supervise=False), bottle, _URL,
|
|
)
|
|
bottle.exec.assert_not_called()
|
|
|
|
def test_runs_claude_mcp_add_as_node(self):
|
|
bottle = _make_bottle()
|
|
ClaudeAgentProvider().provision_supervise_mcp(
|
|
_plan(supervise=True), bottle, _URL,
|
|
)
|
|
bottle.exec.assert_called_once()
|
|
script = bottle.exec.call_args.args[0]
|
|
self.assertEqual("node", bottle.exec.call_args.kwargs.get("user"))
|
|
self.assertIn("claude mcp add", script)
|
|
self.assertIn("--scope user", script)
|
|
self.assertIn("--transport http", script)
|
|
self.assertIn("supervise", script)
|
|
self.assertIn(_URL, script)
|
|
|
|
def test_includes_identity_header_in_normal_registration(self):
|
|
bottle = _make_bottle()
|
|
ClaudeAgentProvider().provision_supervise_mcp(
|
|
_plan(supervise=True, identity_token="tok-abc"), bottle, _URL,
|
|
)
|
|
script = bottle.exec.call_args.args[0]
|
|
self.assertIn("x-bot-bottle-identity", script)
|
|
self.assertIn("tok-abc", script)
|
|
|
|
def test_logs_warning_on_failure_but_does_not_raise(self):
|
|
bottle = _make_bottle(
|
|
exec_result=ExecResult(returncode=1, stdout="", stderr="boom"),
|
|
)
|
|
ClaudeAgentProvider().provision_supervise_mcp(
|
|
_plan(supervise=True), bottle, _URL,
|
|
)
|
|
|
|
def test_fallback_warning_includes_header_but_not_raw_token(self):
|
|
# The recovery hint must show the header is required, but must NOT
|
|
# render the per-bottle identity token into the host log (#476 review).
|
|
bottle = _make_bottle(
|
|
exec_result=ExecResult(returncode=1, stdout="", stderr="boom"),
|
|
)
|
|
with patch("bot_bottle.contrib.claude.agent_provider.warn") as warn_mock:
|
|
ClaudeAgentProvider().provision_supervise_mcp(
|
|
_plan(supervise=True, identity_token="tok-xyz"), bottle, _URL,
|
|
)
|
|
msg = warn_mock.call_args.args[0]
|
|
self.assertIn("x-bot-bottle-identity", msg)
|
|
self.assertNotIn("tok-xyz", msg)
|
|
self.assertIn("<bottle-identity-token>", msg)
|
|
|
|
def test_fallback_warning_omits_header_when_no_token(self):
|
|
bottle = _make_bottle(
|
|
exec_result=ExecResult(returncode=1, stdout="", stderr="boom"),
|
|
)
|
|
with patch("bot_bottle.contrib.claude.agent_provider.warn") as warn_mock:
|
|
ClaudeAgentProvider().provision_supervise_mcp(
|
|
_plan(supervise=True, identity_token=""), bottle, _URL,
|
|
)
|
|
msg = warn_mock.call_args.args[0]
|
|
self.assertNotIn("x-bot-bottle-identity", msg)
|
|
|
|
|
|
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()
|