PRD: Add built-in Pi agent provider #222
@@ -38,13 +38,19 @@ 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",
|
||||
"append_system_prompt",
|
||||
]
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
@@ -108,6 +114,7 @@ class AgentProvisionPlan:
|
||||
prompt_file: Path
|
||||
guest_env: dict[str, str]
|
||||
has_prompt: bool = False
|
||||
startup_args: tuple[str, ...] = ()
|
||||
env_vars: dict[str, str] = field(default_factory=dict)
|
||||
dirs: tuple[AgentProvisionDir, ...] = ()
|
||||
files: tuple[AgentProvisionFile, ...] = ()
|
||||
@@ -163,6 +170,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 +327,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 +351,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 +367,7 @@ def build_agent_provision_plan(
|
||||
trusted_project_path=trusted_project_path,
|
||||
label=label,
|
||||
color=color,
|
||||
provider_settings=provider_settings,
|
||||
)
|
||||
|
||||
|
||||
@@ -372,4 +385,8 @@ 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}."]
|
||||
if prompt_mode == "append_system_prompt":
|
||||
return ["--append-system-prompt", prompt_path]
|
||||
raise ValueError(f"unknown provider prompt mode: {prompt_mode}")
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -23,6 +23,7 @@ class DockerBottle(Bottle):
|
||||
*,
|
||||
agent_command: str = "claude",
|
||||
agent_prompt_mode: PromptMode = "append_file",
|
||||
agent_provider_template: str = "claude",
|
||||
terminal_title: str = "",
|
||||
terminal_color: str = "",
|
||||
):
|
||||
@@ -33,9 +34,7 @@ class DockerBottle(Bottle):
|
||||
self.agent_command = agent_command
|
||||
self.terminal_title = terminal_title
|
||||
self.terminal_color = terminal_color
|
||||
self.agent_provider_template = (
|
||||
"codex" if agent_command == "codex" else "claude"
|
||||
)
|
||||
self.agent_provider_template = agent_provider_template
|
||||
self._closed = False
|
||||
|
||||
def agent_argv(
|
||||
|
||||
@@ -175,6 +175,7 @@ def launch(
|
||||
None,
|
||||
agent_command=plan.agent_command,
|
||||
agent_prompt_mode=plan.agent_prompt_mode,
|
||||
agent_provider_template=plan.agent_provider_template,
|
||||
terminal_title=plan.spec.label or plan.spec.agent_name,
|
||||
terminal_color=plan.spec.color,
|
||||
)
|
||||
|
||||
@@ -69,6 +69,7 @@ class SmolmachinesBottle(Bottle):
|
||||
guest_env: Mapping[str, str] | None = None,
|
||||
agent_command: str = "claude",
|
||||
agent_prompt_mode: PromptMode = "append_file",
|
||||
agent_provider_template: str = "claude",
|
||||
terminal_title: str = "",
|
||||
terminal_color: str = "",
|
||||
) -> None:
|
||||
@@ -86,9 +87,7 @@ class SmolmachinesBottle(Bottle):
|
||||
self.agent_command = agent_command
|
||||
self.terminal_title = terminal_title
|
||||
self.terminal_color = terminal_color
|
||||
self.agent_provider_template = (
|
||||
"codex" if agent_command == "codex" else "claude"
|
||||
)
|
||||
self.agent_provider_template = agent_provider_template
|
||||
|
||||
def agent_argv(
|
||||
self, argv: list[str], *, tty: bool = True,
|
||||
|
||||
@@ -103,6 +103,7 @@ def launch(
|
||||
guest_env=plan.guest_env,
|
||||
agent_command=plan.agent_command,
|
||||
agent_prompt_mode=plan.agent_prompt_mode,
|
||||
agent_provider_template=plan.agent_provider_template,
|
||||
terminal_title=plan.spec.label or plan.spec.agent_name,
|
||||
terminal_color=plan.spec.color,
|
||||
)
|
||||
|
||||
@@ -133,6 +133,7 @@ def prepare_with_preflight(
|
||||
def attach_agent(
|
||||
bottle: Bottle, *, remote_control: bool = False, resume: bool = False,
|
||||
agent_provider_template: str = "claude",
|
||||
startup_args: tuple[str, ...] = (),
|
||||
) -> int:
|
||||
"""Run the selected provider CLI inside `bottle` as an
|
||||
interactive session. Blocks until the session ends; returns the
|
||||
@@ -151,6 +152,7 @@ def attach_agent(
|
||||
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)
|
||||
return bottle.exec_agent(agent_args, tty=True)
|
||||
@@ -235,6 +237,7 @@ def _launch_bottle(
|
||||
bottle,
|
||||
remote_control=remote_control,
|
||||
agent_provider_template=agent_provider_template,
|
||||
startup_args=plan.agent_provision.startup_args,
|
||||
)
|
||||
info(
|
||||
f"session ended (exit {exit_code}); "
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -0,0 +1,29 @@
|
||||
# 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 \
|
||||
fd-find \
|
||||
ripgrep \
|
||||
&& ln -s /usr/bin/fdfind /usr/local/bin/fd \
|
||||
&& 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"]
|
||||
@@ -0,0 +1 @@
|
||||
"""Pi agent provider package."""
|
||||
@@ -0,0 +1,256 @@
|
||||
"""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"
|
||||
_DEFAULT_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, str, list[str], str]:
|
||||
provider_name = str(
|
||||
_settings_value(settings, "provider", _DEFAULT_PROVIDER_NAME)
|
||||
)
|
||||
base_url = str(_settings_value(settings, "base_url", _DEFAULT_BASE_URL))
|
||||
api = str(_settings_value(settings, "api", "openai-completions"))
|
||||
api_key = settings.get("api_key")
|
||||
api_key_env = str(settings.get("api_key_env", ""))
|
||||
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)
|
||||
)
|
||||
provider: dict[str, object] = {
|
||||
"baseUrl": base_url,
|
||||
"api": api,
|
||||
"compat": {
|
||||
"supportsDeveloperRole": supports_developer_role,
|
||||
"supportsReasoningEffort": supports_reasoning_effort,
|
||||
},
|
||||
"models": [{"id": model} for model in models],
|
||||
}
|
||||
if api_key is not None:
|
||||
provider["apiKey"] = str(api_key)
|
||||
elif api_key_env:
|
||||
provider["apiKey"] = "egress-placeholder"
|
||||
elif provider_name == _DEFAULT_PROVIDER_NAME:
|
||||
provider["apiKey"] = "ollama"
|
||||
payload: dict[str, object] = {
|
||||
"providers": {
|
||||
provider_name: provider,
|
||||
}
|
||||
}
|
||||
return payload, base_url, api_key_env, models, provider_name
|
||||
|
||||
|
||||
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="append_system_prompt",
|
||||
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, api_key_env, models, provider_name = (
|
||||
_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())
|
||||
auth_scheme = "Bearer" if api_key_env else ""
|
||||
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,
|
||||
startup_args=(
|
||||
"--models",
|
||||
",".join(f"{provider_name}/{model}" for model in models),
|
||||
),
|
||||
dirs=(AgentProvisionDir(f"{guest_home}/.pi/agent"),),
|
||||
files=(AgentProvisionFile(models_file, _models_path(guest_home)),),
|
||||
egress_routes=(EgressRoute(
|
||||
host=_route_host(base_url),
|
||||
auth_scheme=auth_scheme,
|
||||
token_ref=api_key_env,
|
||||
),),
|
||||
)
|
||||
|
||||
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}")
|
||||
@@ -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,67 @@ 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 = {
|
||||
"provider",
|
||||
"base_url",
|
||||
"api",
|
||||
"api_key",
|
||||
"api_key_env",
|
||||
"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 ("provider", "base_url", "api", "api_key", "api_key_env"):
|
||||
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"
|
||||
)
|
||||
if settings.get("api_key") is not None and settings.get("api_key_env") is not None:
|
||||
raise ManifestError(
|
||||
f"bottle '{bottle_name}' agent_provider.settings may set either "
|
||||
"api_key or api_key_env, not both"
|
||||
)
|
||||
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)
|
||||
|
||||
@@ -0,0 +1,115 @@
|
||||
# PRD 0058: Add built-in Pi agent provider
|
||||
|
||||
- **Status:** Active
|
||||
- **Author:** codex
|
||||
- **Created:** 2026-06-09
|
||||
- **Issue:** #221
|
||||
|
||||
## Summary
|
||||
|
||||
Add `pi` as a built-in `agent_provider.template`. The provider runs the Pi
|
||||
coding-agent CLI, provisions its agent config under `~/.pi/agent`, and writes a
|
||||
provider settings file that targets an unauthenticated Ollama-compatible server.
|
||||
|
||||
The default settings assume an Ollama server at `http://ollama:11434/v1`, using
|
||||
the `openai-completions` API with a dummy API key because Ollama ignores it.
|
||||
Users can override the provider id, base URL, model list, API key, API-key env
|
||||
reference, API type, and compatibility flags through a new
|
||||
`agent_provider.settings` object.
|
||||
|
||||
## Problem
|
||||
|
||||
bot-bottle currently ships Claude and Codex as built-in agent providers. Pi is a
|
||||
useful third harness, but using it today requires a custom provider plugin and a
|
||||
custom image. That repeats boilerplate for prompt copying, skill copying,
|
||||
provider config, and runtime registration.
|
||||
|
||||
Pi's local-model path is also easy to misconfigure: its custom-model docs require
|
||||
`~/.pi/agent/models.json`, an API entry, at least one model id, and a dummy
|
||||
`apiKey` for Ollama even though the server does not authenticate. bot-bottle
|
||||
should generate that shape consistently.
|
||||
|
||||
## Goals / Success Criteria
|
||||
|
||||
- `agent_provider.template: pi` is accepted as a built-in provider.
|
||||
- `bot_bottle/contrib/pi/` provides a Pi image and `PiAgentProvider`.
|
||||
- Pi receives the bot-bottle prompt at `~/.bot-bottle-prompt.txt` and starts in
|
||||
print-mode prompt delivery like Codex.
|
||||
- Pi skills are copied into `~/.pi/agent/skills/<name>/`.
|
||||
- Pi provider settings are configurable from the bottle manifest via
|
||||
`agent_provider.settings`.
|
||||
- The default Pi provider settings configure an unauthenticated Ollama-compatible
|
||||
server.
|
||||
- Unit tests cover manifest parsing, runtime selection, plan generation, prompt,
|
||||
skills, and provider provisioning.
|
||||
|
||||
## Non-goals
|
||||
|
||||
- Managing or launching an Ollama server.
|
||||
- Authenticating to Ollama or any remote Pi provider.
|
||||
- Forwarding host Pi credentials.
|
||||
- Implementing Pi extensions or MCP registration.
|
||||
- Changing Claude or Codex provider behavior.
|
||||
|
||||
## Design
|
||||
|
||||
### Manifest
|
||||
|
||||
Extend `agent_provider` with an optional `settings` object. It is currently only
|
||||
supported for built-in `pi`.
|
||||
|
||||
Supported keys:
|
||||
|
||||
- `base_url`: string, defaults to `http://ollama:11434/v1`
|
||||
- `provider`: string, defaults to `ollama`
|
||||
- `api`: string, defaults to `openai-completions`
|
||||
- `api_key`: string, defaults to `ollama`
|
||||
- `api_key_env`: string, optional host env var name for egress auth injection
|
||||
- `models`: non-empty array of strings, defaults to `["qwen2.5-coder:7b"]`
|
||||
- `supports_developer_role`: boolean, defaults to `false`
|
||||
- `supports_reasoning_effort`: boolean, defaults to `false`
|
||||
|
||||
The snake-case manifest keys are converted into Pi's JSON field names:
|
||||
`baseUrl`, `apiKey`, `supportsDeveloperRole`, and
|
||||
`supportsReasoningEffort`.
|
||||
|
||||
`api_key` and `api_key_env` are mutually exclusive. When targeting a hosted
|
||||
provider through bot-bottle's egress sidecar, omit `api_key` and set
|
||||
`api_key_env` to the host env var that holds the API key. The generated
|
||||
`models.json` receives only an `egress-placeholder` API key, and the egress
|
||||
route injects the real `Authorization` header from the sidecar env. For example,
|
||||
OpenRouter can use provider id `openrouter` with
|
||||
`api_key_env: OPENROUTER_API_KEY`, keeping the key out of the agent env and
|
||||
`models.json`.
|
||||
|
||||
### Provider
|
||||
|
||||
`PiAgentProvider.provision_plan` writes `models.json` into the per-launch state
|
||||
directory and returns an `AgentProvisionPlan` that copies it to
|
||||
`~/.pi/agent/models.json`. The provider also declares an unauthenticated egress
|
||||
route for the configured base URL host so the egress layer can allow the Ollama
|
||||
endpoint.
|
||||
|
||||
The Pi runtime uses:
|
||||
|
||||
- `command="pi"`
|
||||
- `prompt_mode="append_system_prompt"`
|
||||
- `image="bot-bottle-pi:latest"`
|
||||
- `bypass_args=()`
|
||||
- `resume_args=()`
|
||||
- `remote_control_args=()`
|
||||
|
||||
The Dockerfile installs `@earendil-works/pi-coding-agent` globally from npm and
|
||||
keeps the same Debian/node base shape as the existing provider images.
|
||||
|
||||
### Supervise MCP
|
||||
|
||||
Pi does not have built-in MCP support in the current public docs, so
|
||||
`provision_supervise_mcp` is a no-op. This keeps Pi bottles launchable with
|
||||
`supervise: true` while preserving the explicit non-goal of implementing Pi
|
||||
extensions.
|
||||
|
||||
## Merge rule(s)
|
||||
|
||||
This PR can merge when the focused unit tests pass and the PRD status is flipped
|
||||
from Draft to Active in the final implementation commit.
|
||||
@@ -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,110 @@ 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("append_system_prompt", 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),
|
||||
)
|
||||
self.assertEqual(("--models", "ollama/qwen2.5-coder:7b"), plan.startup_args)
|
||||
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_plan_can_target_openrouter_with_egress_injected_api_key(self):
|
||||
with tempfile.TemporaryDirectory(prefix="bb-provider.") as tmp:
|
||||
plan = build_agent_provision_plan(
|
||||
template="pi",
|
||||
dockerfile="",
|
||||
state_dir=Path(tmp),
|
||||
instance_name="bot-bottle-test",
|
||||
prompt_file=Path(tmp) / "prompt.txt",
|
||||
provider_settings={
|
||||
"provider": "openrouter",
|
||||
"base_url": "https://openrouter.ai/api/v1",
|
||||
"api": "openai-completions",
|
||||
"api_key_env": "OPENROUTER_API_KEY",
|
||||
"models": ["google/gemma-4-26b-a4b-it:free"],
|
||||
"supports_reasoning_effort": True,
|
||||
},
|
||||
)
|
||||
models = json.loads(Path(tmp, "pi-models.json").read_text())
|
||||
provider = models["providers"]["openrouter"]
|
||||
self.assertEqual("https://openrouter.ai/api/v1", provider["baseUrl"])
|
||||
self.assertEqual("openai-completions", provider["api"])
|
||||
self.assertEqual("egress-placeholder", provider["apiKey"])
|
||||
self.assertEqual(
|
||||
[{"id": "google/gemma-4-26b-a4b-it:free"}],
|
||||
provider["models"],
|
||||
)
|
||||
self.assertEqual(
|
||||
("--models", "openrouter/google/gemma-4-26b-a4b-it:free"),
|
||||
plan.startup_args,
|
||||
)
|
||||
self.assertEqual("openrouter.ai", plan.egress_routes[0].host)
|
||||
self.assertEqual("Bearer", plan.egress_routes[0].auth_scheme)
|
||||
self.assertEqual("OPENROUTER_API_KEY", plan.egress_routes[0].token_ref)
|
||||
self.assertNotIn("OPENROUTER_API_KEY", plan.guest_env)
|
||||
self.assertTrue(provider["compat"]["supportsReasoningEffort"])
|
||||
|
||||
def test_pi_prompt_mode_appends_system_prompt_interactively(self):
|
||||
self.assertEqual(
|
||||
["--append-system-prompt", "/home/node/.bot-bottle-prompt.txt"],
|
||||
prompt_args("append_system_prompt", "/home/node/.bot-bottle-prompt.txt"),
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
@@ -80,5 +80,28 @@ class TestSettleState(_FakeHomeMixin, unittest.TestCase):
|
||||
start_mod.settle_state("") # should not raise
|
||||
|
||||
|
||||
class TestAttachAgent(unittest.TestCase):
|
||||
def test_passes_provider_startup_args(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="pi",
|
||||
startup_args=("--models", "openrouter/google/gemma"),
|
||||
)
|
||||
|
||||
self.assertEqual(0, exit_code)
|
||||
self.assertEqual(
|
||||
["--models", "openrouter/google/gemma"],
|
||||
bottle.argv,
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
@@ -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="append_system_prompt",
|
||||
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="append_system_prompt",
|
||||
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="append_system_prompt",
|
||||
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()
|
||||
@@ -31,6 +31,16 @@ def _codex_bottle(prompt_path: str | None = None) -> DockerBottle:
|
||||
)
|
||||
|
||||
|
||||
def _pi_bottle(prompt_path: str | None = None) -> DockerBottle:
|
||||
return DockerBottle(
|
||||
container="bot-bottle-dev-abc",
|
||||
teardown=lambda: None,
|
||||
prompt_path_in_container=prompt_path,
|
||||
agent_command="pi",
|
||||
agent_prompt_mode="append_system_prompt",
|
||||
)
|
||||
|
||||
|
||||
class TestClaudeArgv(unittest.TestCase):
|
||||
def test_minimal_argv_no_prompt(self):
|
||||
argv = _bottle().agent_argv([])
|
||||
@@ -117,6 +127,15 @@ class TestClaudeArgv(unittest.TestCase):
|
||||
argv,
|
||||
)
|
||||
|
||||
def test_pi_provider_appends_system_prompt_without_print_mode(self):
|
||||
argv = _pi_bottle("/home/node/.bot-bottle-prompt.txt").agent_argv([])
|
||||
self.assertEqual(
|
||||
["docker", "exec", "-it", "bot-bottle-dev-abc", "pi",
|
||||
"--append-system-prompt", "/home/node/.bot-bottle-prompt.txt"],
|
||||
argv,
|
||||
)
|
||||
self.assertNotIn("-p", argv)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
@@ -111,6 +111,76 @@ class TestAgentProviderHostCredentials(unittest.TestCase):
|
||||
"auth_token": "SOME_TOKEN",
|
||||
})
|
||||
|
||||
def test_settings_allowed_for_pi(self):
|
||||
b = _provider_config_bottle({
|
||||
"template": "pi",
|
||||
"settings": {
|
||||
"provider": "ollama",
|
||||
"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(
|
||||
{
|
||||
"provider": "ollama",
|
||||
"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_allowed_for_pi_api_key_env(self):
|
||||
b = _provider_config_bottle({
|
||||
"template": "pi",
|
||||
"settings": {
|
||||
"provider": "openrouter",
|
||||
"base_url": "https://openrouter.ai/api/v1",
|
||||
"api": "openai-completions",
|
||||
"api_key_env": "OPENROUTER_API_KEY",
|
||||
"models": ["google/gemma-4-26b-a4b-it:free"],
|
||||
},
|
||||
})
|
||||
self.assertEqual("OPENROUTER_API_KEY", b.agent_provider.settings["api_key_env"])
|
||||
|
||||
def test_settings_rejects_api_key_and_api_key_env_together(self):
|
||||
with self.assertRaises(ManifestError):
|
||||
_provider_config_bottle({
|
||||
"template": "pi",
|
||||
"settings": {
|
||||
"api_key": "literal",
|
||||
"api_key_env": "OPENROUTER_API_KEY",
|
||||
},
|
||||
})
|
||||
|
||||
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):
|
||||
|
||||
@@ -28,6 +28,15 @@ def _bottle(prompt_path: str | None = None, **env: str) -> SmolmachinesBottle:
|
||||
)
|
||||
|
||||
|
||||
def _pi_bottle(prompt_path: str | None = None) -> SmolmachinesBottle:
|
||||
return SmolmachinesBottle(
|
||||
"bot-bottle-dev-abc",
|
||||
prompt_path=prompt_path,
|
||||
agent_command="pi",
|
||||
agent_prompt_mode="append_system_prompt",
|
||||
)
|
||||
|
||||
|
||||
def _unwrap(argv: list[str]) -> list[str]:
|
||||
"""Strip the pty_resize wrapper from the front of a TTY-mode
|
||||
argv, return the inner smolvm argv. Mirrors what the kernel
|
||||
@@ -122,6 +131,16 @@ class TestClaudeArgvWrapped(unittest.TestCase):
|
||||
argv[agent_idx - 7:agent_idx - 2],
|
||||
)
|
||||
|
||||
def test_pi_provider_appends_system_prompt_without_print_mode(self):
|
||||
argv = _unwrap(
|
||||
_pi_bottle("/home/node/.bot-bottle-prompt.txt").agent_argv([])
|
||||
)
|
||||
self.assertEqual(
|
||||
["pi", "--append-system-prompt", "/home/node/.bot-bottle-prompt.txt"],
|
||||
argv[argv.index("pi"):],
|
||||
)
|
||||
self.assertNotIn("-p", argv)
|
||||
|
||||
|
||||
class TestClaudeArgvNoTTY(unittest.TestCase):
|
||||
"""`tty=False` paths skip the pty_resize wrapper — there's no
|
||||
|
||||
Reference in New Issue
Block a user