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

feat: add pi agent provider

Browse Source
This commit is contained in:
didericis (codex)
2026-06-09 08:31:48 +00:00
parent 1f38a96561
commit 4f7cfc0418
11 changed files with 651 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
bot_bottle/agent_provider.py
+11 -2
View File
@@ -38,13 +38,14 @@ if TYPE_CHECKING:
PROVIDER_CLAUDE = "claude"
PROVIDER_CODEX = "codex"
PROVIDER_TEMPLATES = frozenset({PROVIDER_CLAUDE, PROVIDER_CODEX})
PROVIDER_PI = "pi"
PROVIDER_TEMPLATES = frozenset({PROVIDER_CLAUDE, PROVIDER_CODEX, PROVIDER_PI})
# Hosts that egress injects the host ChatGPT bearer on when Codex
# forward_host_credentials is enabled. Pipelock must pass these through
# (no TLS MITM) or its header DLP blocks the injected JWT.
CODEX_HOST_CREDENTIAL_HOSTS = ("api.openai.com", "chatgpt.com")
PromptMode = Literal["append_file", "read_prompt_file"]
PromptMode = Literal["append_file", "read_prompt_file", "print_read_prompt_file"]
@dataclass(frozen=True)
@@ -163,6 +164,7 @@ class AgentProvider(ABC):
trusted_project_path: str = "",
label: str = "",
color: str = "",
provider_settings: dict[str, object] | None = None,
) -> AgentProvisionPlan:
"""Build the declarative AgentProvisionPlan for one launch.
Backends call this during `prepare` and consume the result as
@@ -319,6 +321,9 @@ def get_provider(template: str) -> AgentProvider:
if template == PROVIDER_CODEX:
from .contrib.codex.agent_provider import CodexAgentProvider
return CodexAgentProvider()
if template == PROVIDER_PI:
from .contrib.pi.agent_provider import PiAgentProvider
return PiAgentProvider()
raise ValueError(f"unknown agent provider template: {template!r}")
@@ -340,6 +345,7 @@ def build_agent_provision_plan(
trusted_project_path: str = "",
label: str = "",
color: str = "",
provider_settings: dict[str, object] | None = None,
) -> AgentProvisionPlan:
"""Back-compat shim — `prepare` callers stay the same; the work
now lives on the provider plugin."""
@@ -355,6 +361,7 @@ def build_agent_provision_plan(
trusted_project_path=trusted_project_path,
label=label,
color=color,
provider_settings=provider_settings,
)
@@ -372,4 +379,6 @@ def prompt_args(
if argv and "resume" in argv:
return []
return [f"Read and follow the instructions in {prompt_path}."]
if prompt_mode == "print_read_prompt_file":
return ["-p", f"Read and follow the instructions in {prompt_path}."]
raise ValueError(f"unknown provider prompt mode: {prompt_mode}")
bot_bottle/backend/__init__.py
+1
View File
@@ -327,6 +327,7 @@ class BottleBackend(ABC, Generic[PlanT, CleanupT]):
trusted_project_path=workspace.workdir,
label=spec.label,
color=spec.color,
provider_settings=manifest_agent_provider.settings,
)
agent_provision_plan = merge_provision_env_vars(agent_provision_plan)
egress_plan = prepare_egress(manifest_bottle, slug, agent_provision_plan)
bot_bottle/contrib/claude/agent_provider.py
+2 -1
View File
@@ -135,8 +135,9 @@ class ClaudeAgentProvider(AgentProvider):
trusted_project_path: str = "",
label: str = "",
color: str = "",
provider_settings: dict[str, object] | None = None,
) -> AgentProvisionPlan:
del forward_host_credentials, host_env # Codex-only knobs
del forward_host_credentials, host_env, provider_settings
resolved_guest_env = dict(guest_env or {})
guest_home = self.guest_home
trusted_path = trusted_project_path or guest_home
bot_bottle/contrib/codex/agent_provider.py
+2 -1
View File
@@ -77,8 +77,9 @@ class CodexAgentProvider(AgentProvider):
trusted_project_path: str = "",
label: str = "",
color: str = "",
provider_settings: dict[str, object] | None = None,
) -> AgentProvisionPlan:
del auth_token, label, color # Claude-only / title-only knobs
del auth_token, label, color, provider_settings
resolved_guest_env = dict(guest_env or {})
guest_home = self.guest_home
trusted_path = trusted_project_path or guest_home
bot_bottle/contrib/pi/Dockerfile
+23
View File
@@ -0,0 +1,23 @@
# bot-bottle Pi provider image.
#
# Node LTS, git/network tooling, and the Pi coding-agent CLI installed globally.
FROM node:22-slim
RUN apt-get update \
&& apt-get install -y --no-install-recommends git ca-certificates curl \
&& rm -rf /var/lib/apt/lists/*
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 --ignore-scripts --no-fund --no-audit @earendil-works/pi-coding-agent \
&& npm cache clean --force
USER node
WORKDIR /home/node
RUN mkdir -p /home/node/.pi/agent
CMD ["pi"]
bot_bottle/contrib/pi/__init__.py
+1
View File
@@ -0,0 +1 @@
"""Pi agent provider package."""
bot_bottle/contrib/pi/agent_provider.py
+233
View File
@@ -0,0 +1,233 @@
"""Pi agent provider plugin (PRD 0058, contrib).
Pi uses ~/.pi/agent/models.json for custom provider/model settings.
This provider writes an Ollama-compatible default configuration and
lets bottles override the model endpoint and model ids via
agent_provider.settings.
"""
from __future__ import annotations
import json
import os
import shlex
from pathlib import Path
from typing import TYPE_CHECKING
from urllib.parse import urlparse
from ...agent_provider import (
AgentProvider,
AgentProviderRuntime,
AgentProvisionDir,
AgentProvisionFile,
AgentProvisionPlan,
)
from ...egress import EgressRoute
from ...log import die, info
if TYPE_CHECKING:
from ...backend import Bottle, BottlePlan
_DEFAULT_BASE_URL = "http://ollama:11434/v1"
_DEFAULT_MODEL = "qwen2.5-coder:7b"
_PROVIDER_NAME = "ollama"
def _skills_dir(guest_home: str) -> str:
return f"{guest_home}/.pi/agent/skills"
def _prompt_path(guest_home: str) -> str:
return f"{guest_home}/.bot-bottle-prompt.txt"
def _models_path(guest_home: str) -> str:
return f"{guest_home}/.pi/agent/models.json"
def _settings_value(
settings: dict[str, object],
key: str,
default: object,
) -> object:
value = settings.get(key)
return default if value is None else value
def _pi_models_json(settings: dict[str, object]) -> tuple[dict[str, object], str]:
base_url = str(_settings_value(settings, "base_url", _DEFAULT_BASE_URL))
api = str(_settings_value(settings, "api", "openai-completions"))
api_key = str(_settings_value(settings, "api_key", "ollama"))
models_raw = _settings_value(settings, "models", [_DEFAULT_MODEL])
models = [str(model) for model in models_raw] # type: ignore[union-attr]
supports_developer_role = bool(
_settings_value(settings, "supports_developer_role", False)
)
supports_reasoning_effort = bool(
_settings_value(settings, "supports_reasoning_effort", False)
)
payload: dict[str, object] = {
"providers": {
_PROVIDER_NAME: {
"baseUrl": base_url,
"api": api,
"apiKey": api_key,
"compat": {
"supportsDeveloperRole": supports_developer_role,
"supportsReasoningEffort": supports_reasoning_effort,
},
"models": [{"id": model} for model in models],
}
}
}
return payload, base_url
def _route_host(base_url: str) -> str:
parsed = urlparse(base_url)
if not parsed.scheme or not parsed.hostname:
die(
"agent provider provisioning: pi settings base_url must be an "
f"absolute URL (was {base_url!r})"
)
return parsed.hostname
_RUNTIME = AgentProviderRuntime(
template="pi",
command="pi",
image="bot-bottle-pi:latest",
prompt_mode="print_read_prompt_file",
bypass_args=(),
resume_args=(),
remote_control_args=(),
)
class PiAgentProvider(AgentProvider):
@property
def runtime(self) -> AgentProviderRuntime:
return _RUNTIME
def provision_plan(
self,
*,
dockerfile: str,
state_dir: Path,
instance_name: str,
prompt_file: Path,
guest_env: dict[str, str] | None = None,
auth_token: str = "",
forward_host_credentials: bool = False,
host_env: dict[str, str] | None = None,
trusted_project_path: str = "",
label: str = "",
color: str = "",
provider_settings: dict[str, object] | None = None,
) -> AgentProvisionPlan:
del auth_token, forward_host_credentials, host_env, trusted_project_path
del label, color
resolved_guest_env = dict(guest_env or {})
guest_home = self.guest_home
settings = dict(provider_settings or {})
models_payload, base_url = _pi_models_json(settings)
models_file = state_dir / "pi-models.json"
models_file.write_text(json.dumps(models_payload, indent=2) + "\n")
models_file.chmod(0o600)
has_prompt = prompt_file.exists() and bool(prompt_file.read_text())
return AgentProvisionPlan(
template=_RUNTIME.template,
command=_RUNTIME.command,
prompt_mode=_RUNTIME.prompt_mode,
image=_RUNTIME.image,
dockerfile=dockerfile,
guest_home=guest_home,
instance_name=instance_name,
prompt_file=prompt_file,
guest_env=resolved_guest_env,
has_prompt=has_prompt,
dirs=(AgentProvisionDir(f"{guest_home}/.pi/agent"),),
files=(AgentProvisionFile(models_file, _models_path(guest_home)),),
egress_routes=(EgressRoute(host=_route_host(base_url)),),
)
def provision_skills(self, plan: "BottlePlan", bottle: "Bottle") -> None:
from ...backend.util import host_skill_dir
agent = plan.spec.manifest.agents[plan.spec.agent_name]
if not agent.skills:
return
skills_dir = _skills_dir(plan.guest_home)
bottle.exec(f"mkdir -p {skills_dir}", user="root")
for name in agent.skills:
src = host_skill_dir(name)
if not os.path.isdir(src):
die(
f"skill {name!r} disappeared from host between "
f"validation and copy at {src}."
)
dst = f"{skills_dir}/{name}"
info(f"copying skill {name} into {bottle.name}:{dst}")
bottle.exec(f"rm -rf {dst} && mkdir -p {dst}", user="root")
bottle.cp_in(f"{src}/.", f"{dst}/")
bottle.exec(f"chown -R node:node {dst}", user="root")
def provision_prompt(self, plan: "BottlePlan", bottle: "Bottle") -> str | None:
prompt_path = _prompt_path(plan.guest_home)
bottle.cp_in(str(plan.prompt_file), prompt_path) # type: ignore
bottle.exec(
f"chown node:node {prompt_path} && chmod 600 {prompt_path}",
user="root",
)
agent = plan.spec.manifest.agents[plan.spec.agent_name]
return prompt_path if plan.agent_provision.has_prompt or agent.prompt else None
def provision(self, plan: "BottlePlan", bottle: "Bottle") -> None:
provision = plan.agent_provision
for d in provision.dirs:
path = shlex.quote(d.guest_path)
_exec(bottle, f"mkdir -p {path}", f"could not create {d.guest_path}")
_exec(
bottle,
f"chown {shlex.quote(d.owner)} {path}",
f"could not chown {d.guest_path}",
)
_exec(
bottle,
f"chmod {shlex.quote(d.mode)} {path}",
f"could not chmod {d.guest_path}",
)
for f in provision.files:
bottle.cp_in(str(f.host_path), f.guest_path)
path = shlex.quote(f.guest_path)
_exec(
bottle,
f"chown {shlex.quote(f.owner)} {path}",
f"could not chown {f.guest_path}",
)
_exec(
bottle,
f"chmod {shlex.quote(f.mode)} {path}",
f"could not chmod {f.guest_path}",
)
def provision_supervise_mcp(
self,
plan: "BottlePlan",
bottle: "Bottle",
supervise_url: str,
) -> None:
del plan, bottle, supervise_url
def _exec(bottle: "Bottle", script: str, error: str) -> None:
result = bottle.exec(script, user="root")
if result.returncode != 0:
detail = (result.stderr or result.stdout).strip()
if detail:
detail = f": {detail}"
die(f"agent provider provisioning: {error}{detail}")
bot_bottle/manifest_agent.py
+70 -3
View File
@@ -2,7 +2,7 @@
from __future__ import annotations
from dataclasses import dataclass
from dataclasses import dataclass, field
from typing import cast
from .agent_provider import PROVIDER_TEMPLATES
@@ -33,15 +33,23 @@ class ManifestAgentProvider:
dockerfile: str = ""
auth_token: str = ""
forward_host_credentials: bool = False
settings: dict[str, object] = field(default_factory=dict)
@classmethod
def from_dict(cls, bottle_name: str, raw: object) -> "ManifestAgentProvider":
d = as_json_object(raw, f"bottle '{bottle_name}' agent_provider")
for k in d:
if k not in {"template", "dockerfile", "auth_token", "forward_host_credentials"}:
if k not in {
"template",
"dockerfile",
"auth_token",
"forward_host_credentials",
"settings",
}:
raise ManifestError(
f"bottle '{bottle_name}' agent_provider has unknown key {k!r}; "
f"allowed: template, dockerfile, auth_token, forward_host_credentials"
"allowed: template, dockerfile, auth_token, "
"forward_host_credentials, settings"
)
template = d.get("template", "claude")
if not isinstance(template, str) or not template:
@@ -89,11 +97,13 @@ class ManifestAgentProvider:
f"bottle '{bottle_name}' agent_provider.forward_host_credentials "
"is currently only supported for template 'codex'"
)
settings = _parse_provider_settings(bottle_name, template, d.get("settings"))
return cls(
template=template,
dockerfile=dockerfile,
auth_token=auth_token,
forward_host_credentials=forward_host_credentials,
settings=settings,
)
@@ -180,3 +190,60 @@ class ManifestAgent:
git_user = ManifestGitUser.from_dict(name, gd["user"])
return cls(bottle=bottle, skills=skills, prompt=prompt, git_user=git_user)
def _parse_provider_settings(
bottle_name: str,
template: str,
raw: object,
) -> dict[str, object]:
if raw is None:
return {}
if template != "pi":
raise ManifestError(
f"bottle '{bottle_name}' agent_provider.settings is only "
"supported for template 'pi'"
)
settings = as_json_object(raw, f"bottle '{bottle_name}' agent_provider.settings")
allowed = {
"base_url",
"api",
"api_key",
"models",
"supports_developer_role",
"supports_reasoning_effort",
}
for key in settings:
if key not in allowed:
raise ManifestError(
f"bottle '{bottle_name}' agent_provider.settings has unknown "
f"key {key!r}; allowed: {', '.join(sorted(allowed))}"
)
for key in ("base_url", "api", "api_key"):
value = settings.get(key)
if value is not None and (not isinstance(value, str) or not value):
raise ManifestError(
f"bottle '{bottle_name}' agent_provider.settings.{key} must "
"be a non-empty string"
)
models = settings.get("models")
if models is not None:
if not isinstance(models, list) or not models:
raise ManifestError(
f"bottle '{bottle_name}' agent_provider.settings.models must "
"be a non-empty array of strings"
)
for i, model in enumerate(models):
if not isinstance(model, str) or not model:
raise ManifestError(
f"bottle '{bottle_name}' agent_provider.settings.models[{i}] "
"must be a non-empty string"
)
for key in ("supports_developer_role", "supports_reasoning_effort"):
value = settings.get(key)
if value is not None and not isinstance(value, bool):
raise ManifestError(
f"bottle '{bottle_name}' agent_provider.settings.{key} must "
f"be a boolean (was {type(value).__name__})"
)
return dict(settings)
tests/unit/test_agent_provider.py
+68
View File
@@ -11,6 +11,7 @@ from pathlib import Path
from bot_bottle.agent_provider import (
CODEX_HOST_CREDENTIAL_HOSTS,
build_agent_provision_plan,
prompt_args,
)
from bot_bottle.egress import CODEX_HOST_CREDENTIAL_TOKEN_REF
@@ -260,6 +261,73 @@ class TestAgentProviderRuntime(unittest.TestCase):
)
self.assertEqual({}, plan.provisioned_env)
def test_pi_plan_writes_default_ollama_models(self):
with tempfile.TemporaryDirectory(prefix="bb-provider.") as tmp:
plan = build_agent_provision_plan(
template="pi",
dockerfile="/tmp/Dockerfile.pi",
state_dir=Path(tmp),
instance_name="bot-bottle-test",
prompt_file=Path(tmp) / "prompt.txt",
)
models = json.loads(Path(tmp, "pi-models.json").read_text())
self.assertEqual("pi", plan.template)
self.assertEqual("pi", plan.command)
self.assertEqual("print_read_prompt_file", plan.prompt_mode)
self.assertEqual("/tmp/Dockerfile.pi", plan.dockerfile)
self.assertEqual("bot-bottle-pi:latest", plan.image)
self.assertEqual(
("/home/node/.pi/agent",),
tuple(d.guest_path for d in plan.dirs),
)
self.assertEqual(
("/home/node/.pi/agent/models.json",),
tuple(f.guest_path for f in plan.files),
)
provider = models["providers"]["ollama"]
self.assertEqual("http://ollama:11434/v1", provider["baseUrl"])
self.assertEqual("openai-completions", provider["api"])
self.assertEqual("ollama", provider["apiKey"])
self.assertEqual([{"id": "qwen2.5-coder:7b"}], provider["models"])
self.assertEqual("ollama", plan.egress_routes[0].host)
self.assertEqual("", plan.egress_routes[0].auth_scheme)
self.assertEqual("", plan.egress_routes[0].token_ref)
def test_pi_plan_uses_provider_settings(self):
with tempfile.TemporaryDirectory(prefix="bb-provider.") as tmp:
build_agent_provision_plan(
template="pi",
dockerfile="",
state_dir=Path(tmp),
instance_name="bot-bottle-test",
prompt_file=Path(tmp) / "prompt.txt",
provider_settings={
"base_url": "http://host.docker.internal:11434/v1",
"api": "openai-responses",
"api_key": "local",
"models": ["gpt-oss:20b", "qwen3:14b"],
"supports_developer_role": True,
"supports_reasoning_effort": True,
},
)
models = json.loads(Path(tmp, "pi-models.json").read_text())
provider = models["providers"]["ollama"]
self.assertEqual("http://host.docker.internal:11434/v1", provider["baseUrl"])
self.assertEqual("openai-responses", provider["api"])
self.assertEqual("local", provider["apiKey"])
self.assertEqual(
[{"id": "gpt-oss:20b"}, {"id": "qwen3:14b"}],
provider["models"],
)
self.assertTrue(provider["compat"]["supportsDeveloperRole"])
self.assertTrue(provider["compat"]["supportsReasoningEffort"])
def test_pi_prompt_mode_uses_print_flag(self):
self.assertEqual(
["-p", "Read and follow the instructions in /home/node/.bot-bottle-prompt.txt."],
prompt_args("print_read_prompt_file", "/home/node/.bot-bottle-prompt.txt"),
)
if __name__ == "__main__":
unittest.main()
tests/unit/test_contrib_pi_provider.py
+195
View File
@@ -0,0 +1,195 @@
"""Unit: PiAgentProvider provisioning (PRD 0058, contrib/pi)."""
from __future__ import annotations
import unittest
from pathlib import Path
from unittest.mock import MagicMock, patch
from bot_bottle.agent_provider import (
AgentProvisionDir,
AgentProvisionFile,
AgentProvisionPlan,
)
from bot_bottle.backend import Bottle, BottleSpec, ExecResult
from bot_bottle.backend.docker.bottle_plan import DockerBottlePlan
from bot_bottle.contrib.pi.agent_provider import PiAgentProvider
from bot_bottle.egress import EgressPlan
from bot_bottle.git_gate import GitGatePlan
from bot_bottle.manifest import Manifest
_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,
) -> DockerBottlePlan:
manifest = Manifest.from_json_obj({
"bottles": {"dev": {"agent_provider": {"template": "pi"}}},
"agents": {
"demo": {
"skills": list(skills or []),
"prompt": agent_prompt,
"bottle": "dev",
},
},
})
spec = BottleSpec(
manifest=manifest, agent_name="demo",
copy_cwd=False, user_cwd="/tmp/x",
)
return DockerBottlePlan(
spec=spec,
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=None,
use_runsc=False,
agent_provision=agent_provision or AgentProvisionPlan(
template="pi", command="pi", prompt_mode="print_read_prompt_file",
image="bot-bottle-pi: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 TestPiProvisionPrompt(unittest.TestCase):
def test_cp_uses_bottle_cp_in_and_chowns(self):
bottle = _make_bottle()
result = PiAgentProvider().provision_prompt(
_plan(agent_prompt="hello"), bottle,
)
self.assertEqual("/home/node/.bot-bottle-prompt.txt", result)
bottle.cp_in.assert_called_once_with(
"/tmp/state/demo-abc12/agent/prompt.txt",
"/home/node/.bot-bottle-prompt.txt",
)
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)
)
def test_returns_none_when_agent_has_no_prompt(self):
bottle = _make_bottle()
result = PiAgentProvider().provision_prompt(_plan(agent_prompt=""), bottle)
self.assertIsNone(result)
bottle.cp_in.assert_called_once()
class TestPiProvisionSkills(unittest.TestCase):
def test_noop_when_agent_has_no_skills(self):
bottle = _make_bottle()
PiAgentProvider().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.pi.agent_provider.os.path.isdir",
return_value=True,
):
PiAgentProvider().provision_skills(_plan(skills=["search"]), bottle)
scripts = _exec_scripts(bottle)
self.assertTrue(
any("mkdir -p" in s and "/home/node/.pi/agent/skills" in s
for s in scripts)
)
bottle.cp_in.assert_called_once()
self.assertEqual(
"/home/node/.pi/agent/skills/search/",
bottle.cp_in.call_args.args[1],
)
class TestPiProvision(unittest.TestCase):
def test_creates_dir_and_copies_models_config(self):
provision = AgentProvisionPlan(
template="pi", command="pi", prompt_mode="print_read_prompt_file",
image="", dockerfile="", guest_home="/home/node",
instance_name="bot-bottle-demo-abc12",
prompt_file=Path("/tmp/prompt.txt"),
guest_env={},
dirs=(AgentProvisionDir("/home/node/.pi/agent"),),
files=(AgentProvisionFile(
Path("/tmp/pi-models.json"),
"/home/node/.pi/agent/models.json",
),),
)
bottle = _make_bottle()
PiAgentProvider().provision(_plan(agent_provision=provision), bottle)
bottle.cp_in.assert_called_once_with(
"/tmp/pi-models.json",
"/home/node/.pi/agent/models.json",
)
scripts = _exec_scripts(bottle)
self.assertTrue(
any("mkdir -p" in s and "/home/node/.pi/agent" in s for s in scripts)
)
self.assertTrue(
any("chown" in s and "/home/node/.pi/agent/models.json" in s
for s in scripts)
)
def test_dies_when_dir_creation_fails(self):
provision = AgentProvisionPlan(
template="pi", command="pi", prompt_mode="print_read_prompt_file",
image="", dockerfile="", guest_home="/home/node",
instance_name="bot-bottle-demo-abc12",
prompt_file=Path("/tmp/prompt.txt"),
guest_env={},
dirs=(AgentProvisionDir("/home/node/.pi/agent"),),
)
bottle = _make_bottle(exec_result=ExecResult(1, "", "mkdir: nope\n"))
with self.assertRaises(SystemExit):
PiAgentProvider().provision(_plan(agent_provision=provision), bottle)
class TestPiSuperviseMcp(unittest.TestCase):
def test_noop(self):
bottle = _make_bottle()
PiAgentProvider().provision_supervise_mcp(_plan(), bottle, _URL)
bottle.exec.assert_not_called()
if __name__ == "__main__":
unittest.main()
tests/unit/test_manifest_egress.py
+45
View File
@@ -111,6 +111,51 @@ class TestAgentProviderHostCredentials(unittest.TestCase):
"auth_token": "SOME_TOKEN",
})
def test_settings_allowed_for_pi(self):
b = _provider_config_bottle({
"template": "pi",
"settings": {
"base_url": "http://ollama:11434/v1",
"api": "openai-completions",
"api_key": "ollama",
"models": ["qwen2.5-coder:7b"],
"supports_developer_role": False,
"supports_reasoning_effort": False,
},
})
self.assertEqual(
{
"base_url": "http://ollama:11434/v1",
"api": "openai-completions",
"api_key": "ollama",
"models": ["qwen2.5-coder:7b"],
"supports_developer_role": False,
"supports_reasoning_effort": False,
},
b.agent_provider.settings,
)
def test_settings_rejected_for_claude(self):
with self.assertRaises(ManifestError):
_provider_config_bottle({
"template": "claude",
"settings": {"models": ["qwen2.5-coder:7b"]},
})
def test_settings_models_must_be_non_empty_string_array(self):
with self.assertRaises(ManifestError):
_provider_config_bottle({
"template": "pi",
"settings": {"models": []},
})
def test_settings_boolean_flags_must_be_boolean(self):
with self.assertRaises(ManifestError):
_provider_config_bottle({
"template": "pi",
"settings": {"supports_developer_role": "no"},
})
class TestMatches(unittest.TestCase):
def test_optional(self):