diff --git a/.env.example b/.env.example index 8dcc234..fc11d91 100644 --- a/.env.example +++ b/.env.example @@ -34,6 +34,7 @@ PROJECTS_ROOT=/var/lib/handler/projects # Binary overrides (defaults shown). Point at fakes in tests/CI. # CLAUDE_BIN=claude +# PI_BIN=pi # MISE_BIN=mise # TMUX_BIN=tmux # FORGE_BIN=forge diff --git a/Dockerfile.control b/Dockerfile.control index ec4b436..8b4c938 100644 --- a/Dockerfile.control +++ b/Dockerfile.control @@ -39,11 +39,14 @@ FROM python:3.11-slim # tmux — one detached session per agent (and per login attempt) # node + claude — the Claude Code CLI the agents *are*, and the /login flow the # dashboard drives (see control/login.py) +# pi — the lightweight pi coding agent (harness='pi' model backends, +# see control/pi_harness.py); speaks OpenAI-compatible endpoints +# natively, so local vLLM/llama.cpp backends need no proxy # mise — the per-project task runner the test/build gates invoke # forge — the cross-forge CLI the CI poller reads run status from -# Node comes from NodeSource (>=18 is required by Claude Code); mise from its official apt -# repo; forge from the build stage above. Installed under /usr/{bin,local/bin} — outside the -# /var/lib/handler VOLUME — so the volume mount never masks them at runtime. The image is +# Node comes from NodeSource (>=18 for Claude Code, >=22.19 for pi); mise from its official +# apt repo; forge from the build stage above. Installed under /usr/{bin,local/bin} — outside +# the /var/lib/handler VOLUME — so the volume mount never masks them at runtime. The image is # built for amd64 and arm64: NodeSource + forge detect the arch, and the mise apt source is # pinned to `dpkg --print-architecture` (the image's own arch) so the arm64 build pulls the # arm64 package, not an amd64 one. @@ -51,9 +54,10 @@ RUN apt-get update \ && apt-get install -y --no-install-recommends \ git tmux openssh-client curl ca-certificates gnupg \ && install -dm 755 /etc/apt/keyrings \ - && curl -fsSL https://deb.nodesource.com/setup_20.x | bash - \ + && curl -fsSL https://deb.nodesource.com/setup_22.x | bash - \ && apt-get install -y --no-install-recommends nodejs \ && npm install -g @anthropic-ai/claude-code \ + && npm install -g --ignore-scripts @earendil-works/pi-coding-agent \ && npm cache clean --force \ && curl -fsSL https://mise.jdx.dev/gpg-key.pub \ | gpg --dearmor -o /etc/apt/keyrings/mise-archive-keyring.gpg \ diff --git a/README.md b/README.md index 2b5edb2..0840fb8 100644 --- a/README.md +++ b/README.md @@ -132,7 +132,7 @@ Configuration is entirely environment-driven (see [`.env.example`](.env.example) | `WEBHOOK_URL` | Generic target for the `Notification` hook (ntfy, Slack, …) | unset → no-op | | `HANDLER_SECRET_KEY` | Fernet key encrypting git-server tokens + SSH keys at rest (set the same value on API and control) | unset → secret store disabled | | `PROJECTS_ROOT` | Base dir for per-project roots / worktrees / auto-clones | `./projects` | -| `CLAUDE_BIN` / `MISE_BIN` / `TMUX_BIN` / `FORGE_BIN` / `GIT_BIN` | Binary overrides | `claude` / `mise` / `tmux` / `forge` / `git` | +| `CLAUDE_BIN` / `PI_BIN` / `MISE_BIN` / `TMUX_BIN` / `FORGE_BIN` / `GIT_BIN` | Binary overrides | `claude` / `pi` / `mise` / `tmux` / `forge` / `git` | | `FORGE_VERSION` | Pinned forge version verified at spawn (Phase 2) | unset → skip check | | `PROTECTED_BRANCHES` | Branches a direct push needs an approval to reach (Phase 2) | `main,master` | @@ -257,14 +257,18 @@ What the dashboard can now do (all state-changing actions require `ADMIN_TOKEN`) - **Claude** — the management page for the Claude Code install agents run on. The account login lives here (see below), plus web-managed **model backends**, **skills**, **MCP connectors**, **plugins**, and **permission overrides**. Model backends are - Anthropic-API-compatible endpoints (a local Qwen/Llama behind LiteLLM or - claude-code-router, an LLM gateway) offered in the spawn form's **Model** dropdown next - to the Claude subscription: the same `claude` binary is pointed at the endpoint via - `ANTHROPIC_BASE_URL`/`ANTHROPIC_MODEL` env at launch, so hooks, skills, connectors, and - gates apply unchanged, and the agent stays pinned to its backend across resumes. API - keys are stored encrypted (`HANDLER_SECRET_KEY`) and never returned. See - [`docs/local-models.md`](docs/local-models.md) for working local stacks (and why bare - OpenAI-compatible servers break tool calling). These are plain DB rows the control container + alternative endpoints offered in the spawn form's **Model** dropdown next to the + Claude subscription, and each picks a **harness**: `claude` (the same `claude` binary + pointed at an Anthropic-API-compatible endpoint — a local model behind LiteLLM or + claude-code-router, an LLM gateway — via `ANTHROPIC_BASE_URL`/`ANTHROPIC_MODEL` env at + launch) or `pi` (the lightweight [pi coding agent](https://github.com/badlogic/pi-mono), + which speaks bare OpenAI-compatible endpoints — vLLM, llama.cpp, Ollama — natively, no + translation proxy, with handler's hooks/gates/memory/skills bridged in via a bundled pi + extension). Either way hooks, skills, and gates apply, and the agent stays pinned to + its backend across resumes. API keys are stored encrypted (`HANDLER_SECRET_KEY`) and + never returned. See [`docs/local-models.md`](docs/local-models.md) for working local + stacks (and why bare OpenAI-compatible servers break tool calling *on the claude + harness*). These are plain DB rows the control container applies at every launch: skills sync to each worker's user-level `~/.claude/skills` (marker-file managed, so hand-installed skills survive), enabled connectors become the run's `--mcp-config` file (nothing lands in the repo tree), and plugins/permissions fold diff --git a/docs/local-models.md b/docs/local-models.md index 7afc7db..0f43239 100644 --- a/docs/local-models.md +++ b/docs/local-models.md @@ -1,9 +1,25 @@ # Local model backends (Qwen-Coder & friends) -Handler can run agents on locally-hosted models without changing anything about how an -agent works: it is still the same `claude` binary with the same generated -`settings.json`, hooks, skills, MCP connectors, plugins, and permission gates. The only -thing a **model backend** changes is the environment of that one agent's process: +Handler runs agents on locally-hosted models through a **model backend** row, and every +backend picks one of two **harnesses**: + +| Harness | Binary | Endpoint it needs | When to pick it | +|---|---|---|---| +| `claude` (default) | Claude Code | **Anthropic Messages API** incl. tool use — put LiteLLM / claude-code-router in front of a local server | You want the exact Claude Code toolchain (MCP connectors, plugins, permission modes) | +| `pi` | [pi coding agent](https://github.com/badlogic/pi-mono) | **bare OpenAI-compatible** (`/v1/chat/completions`) — vLLM, llama.cpp, Ollama directly, no proxy | You want the lightest loop for slow local token throughput | + +Both harnesses keep handler's contract intact: the same hooks (test/completion gate, +push gate, approval gate), the same checkmark/log streaming, the same memory layer, the +same skills, kill/resume, and schedules. The Claude subscription (no backend selected) +always launches `claude` — pi is only ever used when you point an agent at a backend row +that says so. + +## The claude harness + +Nothing about how an agent works changes: it is still the same `claude` binary with the +same generated `settings.json`, hooks, skills, MCP connectors, plugins, and permission +gates. The only thing the backend changes is the environment of that one agent's +process: | Variable | From | |---|---| @@ -88,6 +104,67 @@ Then register the backend in Handler: base URL `http://:4000`, model [claude-code-router](https://github.com/musistudio/claude-code-router) in front as the Anthropic translator. +## The pi harness + +Set **Harness: pi** on the backend row (or `"harness": "pi"` via `POST /claude/models`) +and point `base_url` straight at the OpenAI-compatible endpoint — no LiteLLM, no +claude-code-router: + +```bash +# vLLM with the Qwen tool parser is all you need: +vllm serve Qwen/Qwen3-Coder-30B-A3B-Instruct \ + --enable-auto-tool-choice --tool-call-parser qwen3_coder --port 8000 +``` + +Backend row: base URL `http://:8000/v1`, model +`Qwen/Qwen3-Coder-30B-A3B-Instruct`, harness `pi`, API key optional (pi requires *some* +credential, so handler injects a placeholder when none is stored). The same +tool-parser/template caveats apply as ever — the model's tool calls must come back as +structured `tool_calls`, so use vLLM's parser flags, `--jinja` on `llama-server`, or an +Ollama model whose template declares `.Tools`. + +### What the control layer generates + +At every launch (spawn *and* resume) the backend row is materialized into a per-agent +`PI_CODING_AGENT_DIR` under `~/.handler-pi/` — outside the repo tree, so the clean-tree +completion gate never sees generated files: + +- **`models.json` + `settings.json`** — the row as a pi provider (`openai-completions` + by default) pinned as the default model. Row `env` keys `PI_PROVIDER_API`, + `PI_CONTEXT_WINDOW`, and `PI_MAX_TOKENS` tune it; everything else in the env map + passes through to the process. +- **`extensions/handler-bridge.ts`** — the bundled bridge extension that adapts pi's + events to the same `python -m handler.hooks` contract claude uses. The gates are the + *same tested Python code*: the Stop/completion gate re-prompts pi with the blockers, + `git push` runs the test + image-build + protected-branch approval gates and denies on + failure, and questions go through an `ask_operator` tool that pauses the agent for the + normal answer/resume flow. Memory recall is injected at session start, and the memory + tools (`memory_search/get/save/link`) are registered directly — pi has no MCP by + design, so the bridge shells to `python -m handler.mcpserver --call ` instead. +- **`APPEND_SYSTEM.md`** — the handler conventions (completion contract, ask_operator, + memory usage) appended to pi's system prompt. + +Skills work unchanged: pi implements the same SKILL.md standard as Claude Code, and the +generated `settings.json` points pi's discovery at the web-managed `~/.claude/skills` +sync plus the repo's committed `.claude/skills` (the forge role skills). pi also reads +`AGENTS.md` / `CLAUDE.md` context files natively. + +Sessions are single JSONL files pre-assigned by handler (`--session `), so +cross-worker resume works exactly like claude's: archived to the DB, materialized by +whichever worker claims the resume, continued by launching pi again on the same file. + +### What differs from the claude harness + +- **MCP connectors and plugins don't apply** — pi has no MCP client or plugin system. + The bundled memory server is bridged as native tools; other connectors are + claude-harness-only for now. +- **Permission modes don't apply** — pi has no permission system. The hard gates + (PreToolUse-equivalent blocking, Stop gate) are enforced by the bridge, which is the + layer handler actually relies on for claude too. +- **`--max-budget-usd` doesn't apply** — local tokens are free; pi has no budget flag. +- The `pi` binary must be on the worker's PATH (the control image bakes it in; + `PI_BIN` overrides, same as `CLAUDE_BIN`). + ## Expectations and tips for small models - **Keep the harness light.** Handler's agents run tool-heavy (hooks, MCP connectors, diff --git a/frontend/components/sections/ClaudeSection.tsx b/frontend/components/sections/ClaudeSection.tsx index 1ab29ea..ec0f6a9 100644 --- a/frontend/components/sections/ClaudeSection.tsx +++ b/frontend/components/sections/ClaudeSection.tsx @@ -506,11 +506,17 @@ const emptyModel = { base_url: "", model: "", small_fast_model: "", + harness: "claude" as "claude" | "pi", api_key: "", env: "", enabled: true, }; +const HARNESS_OPTS = [ + { value: "claude", label: "claude — Anthropic-compatible endpoint (LiteLLM/gateway)" }, + { value: "pi", label: "pi — bare OpenAI-compatible endpoint (vLLM/llama.cpp/Ollama)" }, +]; + function ModelsPanel() { const s = useDashboard(); const [form, setForm] = useState(emptyModel); @@ -527,6 +533,7 @@ function ModelsPanel() { base_url: form.base_url.trim(), model: form.model.trim(), small_fast_model: form.small_fast_model.trim() || null, + harness: form.harness, api_key: form.api_key.trim() || null, env: parseKeyValues(form.env), enabled: form.enabled, @@ -544,6 +551,7 @@ function ModelsPanel() { base_url: m.base_url, model: m.model, small_fast_model: m.small_fast_model ?? "", + harness: m.harness ?? "claude", api_key: "", // write-only; blank = keep the stored key env: formatKeyValues(m.env), enabled: m.enabled, @@ -555,13 +563,15 @@ function ModelsPanel() { <>
Alternative model backends the spawn dropdown offers next to the Claude - subscription — the same claude binary pointed at a - different endpoint via ANTHROPIC_BASE_URL, so - skills, connectors, hooks, and gates apply unchanged. The endpoint must speak the{" "} - Anthropic Messages API including tool use — a bare OpenAI-compatible - server (Ollama, llama.cpp, LM Studio) breaks tool calling; front it with LiteLLM - or claude-code-router and enable the backend's native tool parser. See{" "} - docs/local-models.md for working Qwen-Coder stacks. + subscription. The claude harness is the same{" "} + claude binary pointed at a different endpoint via{" "} + ANTHROPIC_BASE_URL — the endpoint must speak the{" "} + Anthropic Messages API including tool use, so front a bare + OpenAI-compatible server with LiteLLM or claude-code-router. The pi{" "} + harness runs the lightweight pi coding agent instead, which speaks{" "} + OpenAI-compatible endpoints natively (vLLM, llama.cpp, Ollama — no + translation proxy) — hooks, gates, memory, and skills still apply via + handler's bridge. See docs/local-models.md.
@@ -594,6 +604,12 @@ function ModelsPanel() { onChange={(v) => setForm({ ...form, small_fast_model: v })} placeholder="qwen3-1.7b" /> +
+ {m.harness === "pi" && pi harness} {m.has_api_key && key stored} {m.enabled ? "enabled" : "disabled"} diff --git a/frontend/components/store.tsx b/frontend/components/store.tsx index 4fa48d6..c493202 100644 --- a/frontend/components/store.tsx +++ b/frontend/components/store.tsx @@ -201,6 +201,7 @@ export interface ModelBody { base_url: string; model: string; small_fast_model: string | null; + harness: "claude" | "pi"; /* Write-only: encrypted at rest server-side, never echoed back. Null = no change. */ api_key: string | null; clear_api_key?: boolean; diff --git a/frontend/lib/api.ts b/frontend/lib/api.ts index 3311dba..3a7a006 100644 --- a/frontend/lib/api.ts +++ b/frontend/lib/api.ts @@ -193,6 +193,9 @@ export interface ClaudeModel { base_url: string; model: string; small_fast_model?: string | null; + /* Which agent binary runs against this backend: "claude" (Anthropic-compatible + endpoint required) or "pi" (bare OpenAI-compatible endpoint, lightweight). */ + harness?: "claude" | "pi"; env?: Record | null; enabled: boolean; has_api_key: boolean; diff --git a/src/handler/api/routes/claude.py b/src/handler/api/routes/claude.py index fada5a4..c16b78b 100644 --- a/src/handler/api/routes/claude.py +++ b/src/handler/api/routes/claude.py @@ -304,6 +304,7 @@ def create_model(body: ClaudeModelIn, conn: Connection = Depends(db_conn)) -> di body.model, api_key_enc=api_key_enc, small_fast_model=body.small_fast_model, + harness=body.harness, env=body.env, enabled=body.enabled, ) diff --git a/src/handler/api/schemas.py b/src/handler/api/schemas.py index 975ff4d..f8b9594 100644 --- a/src/handler/api/schemas.py +++ b/src/handler/api/schemas.py @@ -14,6 +14,9 @@ from pydantic import BaseModel, ConfigDict, Field, field_validator, model_valida Role = Literal["junior", "senior", "deploy"] ForgeType = Literal["github", "gitlab", "gitea", "forgejo", "bitbucket"] +# Agent harness a model backend launches (claude_models.harness). +Harness = Literal["claude", "pi"] + # credential_ref schemes an operator may set over the web. ``cmd:`` is intentionally # excluded — it would run an arbitrary command in the control container at spawn — so the # API rejects it even though the CLI/DB path still allows it. @@ -532,6 +535,10 @@ class ClaudeModelIn(BaseModel): base_url: str = Field(min_length=1) model: str = Field(min_length=1) small_fast_model: str | None = None + # Which agent binary runs against this backend. "claude" needs an + # Anthropic-API-compatible endpoint; "pi" speaks OpenAI-compatible endpoints + # (vLLM, llama.cpp, Ollama) natively — no translation proxy. + harness: Harness = "claude" api_key: str | None = None env: dict[str, str] = Field(default_factory=dict) enabled: bool = True @@ -550,6 +557,7 @@ class ClaudeModelUpdateIn(BaseModel): base_url: str | None = None model: str | None = Field(default=None, min_length=1) small_fast_model: str | None = None + harness: Harness | None = None api_key: str | None = None clear_api_key: bool = False env: dict[str, str] | None = None @@ -574,6 +582,7 @@ class ClaudeModelOut(BaseModel): base_url: str model: str small_fast_model: str | None = None + harness: Harness = "claude" env: dict[str, str] | None = None enabled: bool # The key never leaves the server; this says whether one is stored. diff --git a/src/handler/config.py b/src/handler/config.py index a2ab170..263df30 100644 --- a/src/handler/config.py +++ b/src/handler/config.py @@ -46,6 +46,7 @@ class Settings(BaseSettings): # Binary overrides so tests/CI can point at fakes. claude_bin: str = "claude" + pi_bin: str = "pi" mise_bin: str = "mise" tmux_bin: str = "tmux" forge_bin: str = "forge" diff --git a/src/handler/control/headless.py b/src/handler/control/headless.py index 9bb68e3..ba0c78c 100644 --- a/src/handler/control/headless.py +++ b/src/handler/control/headless.py @@ -1,4 +1,6 @@ -"""The headless runner: worker-owned ``claude -p`` subprocesses streaming to the DB. +"""The headless runner: worker-owned agent subprocesses streaming to the DB — +``claude -p --output-format stream-json`` by default, ``pi -p --mode json`` for agents +pinned to a pi-harness model backend (see ``control.pi_harness``). This is the tmux replacement seam for agent *runs* (tmux stays only for the interactive ``/login`` flow). Each invocation is ``claude -p --output-format stream-json`` with a @@ -34,7 +36,7 @@ from pathlib import Path from ..config import get_settings from ..db import repository as repo from ..db.engine import connection -from . import claude_gen +from . import claude_gen, pi_harness # Stream types we recognize from ``--output-format stream-json``; anything else (or an # unparseable line) is stored as-is so no output is ever dropped. ``worker`` is our own: @@ -144,20 +146,59 @@ def assistant_text(payload: dict) -> str | None: return text or None -def archive_session(working_dir: str, session_id: str, max_bytes: int | None = None) -> bytes | None: - """Tar.gz the session transcript + sidecar dir, or None when nothing exists yet or - the result would exceed ``max_bytes`` (the caller records a worker event; the run - itself is unaffected — only cross-worker resume degrades).""" - base = session_dir(working_dir) - jsonl = base / f"{session_id}.jsonl" - sidecar = base / session_id - if not jsonl.exists() and not sidecar.exists(): +def pi_assistant_text(payload: dict) -> str | None: + """The assistant text of a pi ``message_end`` event, or None for non-assistant + messages (pi emits ``message_end`` for user/custom messages too — claude's + ``assistant`` events never carry another role, so this guard is pi-only).""" + message = payload.get("message") + if not isinstance(message, dict) or message.get("role") != "assistant": + return None + return assistant_text(payload) + + +def pi_result_payload(agent_end: dict) -> dict: + """A pi ``agent_end`` event normalized into the claude-``result``-shaped payload the + run row stores: enough for the ``is_error`` verdict and the UI. pi emits one + ``agent_end`` per low-level run (auto-retries and stop-gate continuations each get + their own); the supervisor overwrites on each, so the final one wins — exactly the + verdict of the run's last words.""" + messages = agent_end.get("messages") + stop_reason = None + if isinstance(messages, list): + for message in reversed(messages): + if isinstance(message, dict) and message.get("role") == "assistant": + stop_reason = message.get("stopReason") + break + return { + "harness": "pi", + "is_error": stop_reason in ("error", "aborted"), + "stop_reason": stop_reason, + "will_retry": bool(agent_end.get("willRetry")), + } + + +def archive_session( + working_dir: str, session_id: str, max_bytes: int | None = None, harness: str = "claude" +) -> bytes | None: + """Tar.gz the session transcript (+ claude's sidecar dir), or None when nothing + exists yet or the result would exceed ``max_bytes`` (the caller records a worker + event; the run itself is unaffected — only cross-worker resume degrades). A pi + session is a single JSONL file at the path handler pre-assigned via ``--session``.""" + if harness == "pi": + base = pi_harness.sessions_dir(working_dir) + jsonl = pi_harness.session_file(working_dir, session_id) + sidecar = None + else: + base = session_dir(working_dir) + jsonl = base / f"{session_id}.jsonl" + sidecar = base / session_id + if not jsonl.exists() and (sidecar is None or not sidecar.exists()): return None buf = io.BytesIO() with tarfile.open(fileobj=buf, mode="w:gz") as tar: if jsonl.exists(): tar.add(jsonl, arcname=jsonl.name) - if sidecar.is_dir(): + if sidecar is not None and sidecar.is_dir(): tar.add(sidecar, arcname=sidecar.name) data = buf.getvalue() limit = max_bytes if max_bytes is not None else get_settings().session_archive_max_bytes @@ -166,13 +207,16 @@ def archive_session(working_dir: str, session_id: str, max_bytes: int | None = N return data -def materialize_session(working_dir: str, archive: bytes) -> None: - """Unpack a session archive where claude will look for it on ``--resume``. +def materialize_session(working_dir: str, archive: bytes, harness: str = "claude") -> None: + """Unpack a session archive where the harness will look for it on resume. ``filter="data"`` rejects path traversal and special members — the archive came from our own DB, but a defense-in-depth default costs nothing. """ - base = session_dir(working_dir) + if harness == "pi": + base = pi_harness.sessions_dir(working_dir) + else: + base = session_dir(working_dir) base.mkdir(parents=True, exist_ok=True) with tarfile.open(fileobj=io.BytesIO(archive), mode="r:gz") as tar: tar.extractall(base, filter="data") @@ -198,6 +242,8 @@ class RunSupervisor: cancel_poll: float = 5.0, archive_interval: float = 60.0, on_exit=None, + harness: str = "claude", + prompt_stdin: str | None = None, ) -> None: self.agent = agent self.run = run @@ -207,6 +253,10 @@ class RunSupervisor: self.cancel_poll = cancel_poll self.archive_interval = archive_interval self.on_exit = on_exit # worker's slot-release callback + self.harness = harness + # pi takes the prompt on stdin (it has no ``--`` argv separator); claude runs + # keep stdin closed as before. + self.prompt_stdin = prompt_stdin self._seq = 0 self._seq_lock = threading.Lock() # reader thread + supervisor both emit events self._result_payload: dict | None = None @@ -241,7 +291,18 @@ class RunSupervisor: etype, payload = parse_stream_line(line) try: self._insert_event(etype, payload) - if etype == "result": + if self.harness == "pi": + # pi's stream is normalized on the fly: the final ``agent_end`` + # plays claude's ``result`` role, assistant ``message_end`` events + # feed last_output. Raw events are stored as-is either way. + if etype == "agent_end": + self._result_payload = pi_result_payload(payload) + elif etype == "message_end": + text = pi_assistant_text(payload) + if text: + with connection() as conn: + repo.update_agent_output(conn, self.agent["id"], text) + elif etype == "result": self._result_payload = payload elif etype == "assistant": text = assistant_text(payload) @@ -253,7 +314,7 @@ class RunSupervisor: def _upload_archive(self) -> None: try: - data = archive_session(self.cwd, self.run["session_id"]) + data = archive_session(self.cwd, self.run["session_id"], harness=self.harness) if data is None: return with connection() as conn: @@ -278,20 +339,27 @@ class RunSupervisor: def _supervise(self) -> None: stderr_file = tempfile.TemporaryFile() + pipe_prompt = self.prompt_stdin is not None try: proc = subprocess.Popen( self.argv, cwd=self.cwd, env={**os.environ, **self.env}, - stdin=subprocess.DEVNULL, + stdin=subprocess.PIPE if pipe_prompt else subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=stderr_file, text=True, ) except OSError as exc: stderr_file.close() - self._settle(exit_code=None, stderr_tail=f"failed to launch claude: {exc}") + self._settle(exit_code=None, stderr_tail=f"failed to launch {self.harness}: {exc}") return + if pipe_prompt: + try: + proc.stdin.write(self.prompt_stdin) + proc.stdin.close() + except (OSError, ValueError): + pass # a dead-on-arrival process is settled by the wait loop below reader = threading.Thread( target=self._pump_stdout, args=(proc.stdout,), daemon=True ) @@ -381,33 +449,56 @@ def launch( env: dict[str, str], worker_id: str, on_exit=None, + harness: str = "claude", ) -> dict: """Start a headless run for ``agent`` and return its ``agent_runs`` row. ``kind`` is ``spawn`` (fresh session, new UUID) or ``resume`` (materialize the stored archive, continue the agent's existing session). Fire-and-forget from the caller's perspective — the returned run row is already ``running`` and a daemon supervisor - owns the process from here. + owns the process from here. ``harness`` selects the binary: ``claude`` (default) or + ``pi`` (see ``control.pi_harness``) — the supervision, event log, archive, and + status reconciliation are identical either way. """ working_dir = agent["working_dir"] - # The generated connectors file (control.claude_gen) rides along when present; its - # presence on disk is the contract, so the launch seam's signature stays stable. - mcp_config = claude_gen.mcp_config_path(working_dir) - if not os.path.exists(mcp_config): - mcp_config = None + prompt_stdin: str | None = None if kind == "spawn": session_id = str(uuid.uuid4()) - argv = build_spawn_argv(prompt, settings_path, session_id, mcp_config) else: session_id = agent.get("session_id") if not session_id: raise ValueError(f"agent '{agent['name']}' has no session to resume") - argv = build_resume_argv(session_id, prompt, settings_path, mcp_config) + if harness == "pi": + # Spawn and resume are the same invocation: the pre-assigned --session file is + # created on first use and appended to afterwards. The prompt travels on stdin. + argv = pi_harness.build_argv(session_id, working_dir) + prompt_stdin = prompt + # The bridge extension reports hook events under the run's session UUID (the + # transcript basename), same identity contract as claude's hook stdin. + env = {**env, "HANDLER_SESSION_ID": session_id} + else: + # The generated connectors file (control.claude_gen) rides along when present; + # its presence on disk is the contract, so the launch seam's signature stays + # stable. (pi has no MCP — its bridge registers the memory tools directly.) + mcp_config = claude_gen.mcp_config_path(working_dir) + if not os.path.exists(mcp_config): + mcp_config = None + if kind == "spawn": + argv = build_spawn_argv(prompt, settings_path, session_id, mcp_config) + else: + argv = build_resume_argv(session_id, prompt, settings_path, mcp_config) with connection() as conn: run = repo.create_run(conn, agent["id"], session_id, worker_id, kind) repo.set_agent_session(conn, agent["id"], session_id, worker_id) supervisor = RunSupervisor( - agent, run, argv, cwd=working_dir, env=env, on_exit=on_exit + agent, + run, + argv, + cwd=working_dir, + env=env, + on_exit=on_exit, + harness=harness, + prompt_stdin=prompt_stdin, ) supervisor.start() return run diff --git a/src/handler/control/models.py b/src/handler/control/models.py index 6c74ca1..041b9f8 100644 --- a/src/handler/control/models.py +++ b/src/handler/control/models.py @@ -9,11 +9,16 @@ binary at the endpoint, ``ANTHROPIC_MODEL`` / ``ANTHROPIC_SMALL_FAST_MODEL`` nam it serves, and ``ANTHROPIC_AUTH_TOKEN`` carries the endpoint's key (decrypted here, in the control container, from the encrypted column — the API never returns it). -The endpoint must speak the Anthropic Messages API *including tool use*. A bare -OpenAI-compatible server (Ollama, llama.cpp, LM Studio, vLLM) is not enough on its own — -that mismatch is exactly the "tool calling not working" failure with Qwen-Coder — so put -a translating proxy in front and enable the backend's native tool parser; see -``docs/local-models.md`` for working stacks. +For the default ``harness='claude'`` the endpoint must speak the Anthropic Messages API +*including tool use*. A bare OpenAI-compatible server (Ollama, llama.cpp, LM Studio, +vLLM) is not enough on its own — that mismatch is exactly the "tool calling not working" +failure with Qwen-Coder — so put a translating proxy in front and enable the backend's +native tool parser; see ``docs/local-models.md`` for working stacks. + +``harness='pi'`` rows skip all of that: the lightweight pi coding agent speaks the +OpenAI Completions API natively, so the row's ``base_url`` is the bare local endpoint +and there is no ``ANTHROPIC_*`` env at all — the row renders into a pi provider config +instead (see ``control.pi_harness``). """ from __future__ import annotations @@ -35,10 +40,11 @@ class ModelError(Exception): """Raised when a selected model backend cannot be resolved into an environment.""" -def resolve_model_env( +def resolve_model( conn, model_id: int | None, *, require_enabled: bool = False -) -> dict[str, str]: - """The env overrides for ``model_id``, or ``{}`` for None (the Claude subscription). +) -> tuple[dict, str] | None: + """The backend row + its decrypted API key (a placeholder when none is stored), or + ``None`` for ``model_id=None`` (the Claude subscription). ``require_enabled`` is the spawn path (a disabled backend must not take new agents); resumes pass False so an agent already pinned to a since-disabled backend can still @@ -47,7 +53,7 @@ def resolve_model_env( this agent not to use. """ if model_id is None: - return {} + return None row = repo.get_claude_model(conn, model_id) if row is None: raise ModelError( @@ -65,6 +71,22 @@ def resolve_model_env( ) from exc else: key = _PLACEHOLDER_KEY + return row, key + + +def harness_of(resolved: tuple[dict, str] | None) -> str: + """Which agent binary a resolved backend launches; the subscription is claude.""" + if resolved is None: + return "claude" + return resolved[0].get("harness") or "claude" + + +def claude_env(resolved: tuple[dict, str] | None) -> dict[str, str]: + """The ``ANTHROPIC_*`` env for a claude-harness backend, ``{}`` for None (the + subscription needs no overrides).""" + if resolved is None: + return {} + row, key = resolved env = { **_LOCAL_DEFAULTS, "ANTHROPIC_BASE_URL": row["base_url"], @@ -77,3 +99,15 @@ def resolve_model_env( for k, v in (row.get("env") or {}).items(): env[str(k)] = str(v) return env + + +def resolve_model_env( + conn, model_id: int | None, *, require_enabled: bool = False +) -> dict[str, str]: + """The claude-harness env overrides for ``model_id`` (``{}`` for the subscription). + Kept as the validation seam ``spawn`` fail-fasts through; pi-harness rows resolve + fine here too (spawn only checks resolvability), their env just isn't this one.""" + resolved = resolve_model(conn, model_id, require_enabled=require_enabled) + if harness_of(resolved) != "claude": + return {} + return claude_env(resolved) diff --git a/src/handler/control/pi_bridge.ts b/src/handler/control/pi_bridge.ts new file mode 100644 index 0000000..f5f41eb --- /dev/null +++ b/src/handler/control/pi_bridge.ts @@ -0,0 +1,249 @@ +/** + * handler-bridge — the pi extension that wires a pi agent into handler's hooks. + * + * Installed by `handler.control.pi_harness` into the per-agent PI_CODING_AGENT_DIR and + * loaded with `-e`, so it runs on every pi-harness launch. It adapts pi's extension + * events to the exact stdin/stdout contract of `python -m handler.hooks ` — all + * gate logic (test gate, push gate, approval gate, question deferral, memory recall) + * stays in the tested Python modules; this file only translates: + * + * - before_agent_start → hooks session_start (memory recall, injected once as context) + * - tool_call "bash" → hooks pre_tool_use (git-push / merge-deploy gates; can block) + * - ask_operator tool → hooks pre_tool_use as AskUserQuestion (defer + pause), then + * terminates the run so the async answer/resume flow takes over + * - agent_settled → hooks stop (the completion gate; a block is fed back as a + * follow-up user message, continuing the run like claude's + * Stop-hook re-invoke; stop_hook_active guards the loop) + * - session_shutdown → hooks session_end + * + * It also registers the memory tools (memory_search/get/save/link) that claude agents + * reach over MCP, by shelling to `python -m handler.mcpserver --call ` — pi has + * no MCP by design, and a subprocess inheriting the agent env is the same trust model + * the MCP server used anyway. + * + * Identity and configuration arrive via the spawn environment, exactly like hooks: + * HANDLER_AGENT_ID / HANDLER_PROJECT_ID / HANDLER_AGENT_NAME / DATABASE_URL, plus + * HANDLER_PYTHON (the control layer's interpreter) and HANDLER_SESSION_ID (the run's + * session UUID, which is also the session file's basename). + */ +import { spawnSync } from "node:child_process"; + +import type { ExtensionAPI } from "@earendil-works/pi-coding-agent"; +import { Type } from "typebox"; + +const PYTHON = process.env.HANDLER_PYTHON || "python3"; +const SESSION_ID = process.env.HANDLER_SESSION_ID || ""; +// The stop hook runs the project's own test suite; give it real time. +const STOP_HOOK_TIMEOUT_MS = 30 * 60 * 1000; +const HOOK_TIMEOUT_MS = 10 * 60 * 1000; + +function runHook(event: string, payload: Record, timeoutMs = HOOK_TIMEOUT_MS): any { + const res = spawnSync(PYTHON, ["-m", "handler.hooks", event], { + input: JSON.stringify({ session_id: SESSION_ID, cwd: process.cwd(), ...payload }), + encoding: "utf8", + timeout: timeoutMs, + }); + if (res.error || res.status !== 0) { + const detail = res.error ? String(res.error) : (res.stderr || "").slice(-2000); + process.stderr.write(`handler-bridge: hook ${event} failed: ${detail}\n`); + return null; + } + const line = (res.stdout || "").trim().split("\n").filter(Boolean).pop(); + if (!line) return {}; + try { + return JSON.parse(line); + } catch { + return {}; + } +} + +function callMemory(tool: string, args: Record): string { + const res = spawnSync(PYTHON, ["-m", "handler.mcpserver", "--call", tool], { + input: JSON.stringify(args ?? {}), + encoding: "utf8", + timeout: 60_000, + }); + if (res.error || res.status !== 0) { + const detail = res.error ? String(res.error) : (res.stderr || "").slice(-2000); + throw new Error(`${tool} failed: ${detail}`); + } + return (res.stdout || "").trim() || "{}"; +} + +function permissionDeny(out: any): string | null { + const spec = out?.hookSpecificOutput; + if (spec?.permissionDecision === "deny") { + return String(spec.permissionDecisionReason || "denied by handler"); + } + return null; +} + +export default function (pi: ExtensionAPI) { + let contextInjected = false; + let askDeferred = false; + let stopRounds = 0; + let lastAssistantText = ""; + + // ---- memory recall at session start (claude's SessionStart hook) ---------------- + pi.on("before_agent_start", async () => { + if (contextInjected) return; + contextInjected = true; + const out = runHook("session_start", {}); + const text = out?.hookSpecificOutput?.additionalContext; + if (typeof text === "string" && text.trim()) { + return { + message: { customType: "handler-context", content: text, display: true }, + }; + } + }); + + // ---- PreToolUse gates ------------------------------------------------------------ + pi.on("tool_call", async (event) => { + if (event.toolName === "bash") { + const command = String((event.input as any)?.command ?? ""); + // Cheap pre-filter mirroring the hook's own matchers: only push/merge/deploy + // commands pay for a hook subprocess; everything else runs untouched. + if (!/\bgit\s+push\b|\bforge\s+(?:pr\s+)?merge\b|\bmise\s+run\s+deploy\b/.test(command)) return; + const out = runHook( + "pre_tool_use", + { tool_name: "Bash", tool_input: { command } }, + STOP_HOOK_TIMEOUT_MS, // the push gate runs tests + an image build + ); + const reason = permissionDeny(out); + if (reason) return { block: true, reason }; + return; + } + if (event.toolName === "ask_operator") { + const question = String((event.input as any)?.question ?? "").trim(); + const out = runHook("pre_tool_use", { + tool_name: "AskUserQuestion", + tool_input: { questions: [{ question }] }, + }); + askDeferred = true; + runHook("notification", { message: `question for the operator: ${question}`.slice(0, 500) }); + const reason = + permissionDeny(out) || + "Question deferred to the operator; the run pauses here and resumes with the answer."; + // terminate: the operator answers asynchronously (answer + resume); there is + // nothing useful this process can do while it waits. + return { block: true, reason, terminate: true }; + } + }); + + // ---- the completion gate (claude's Stop hook) ------------------------------------- + pi.on("message_end", async (event) => { + const m: any = (event as any).message; + if (m?.role === "assistant" && Array.isArray(m.content)) { + const text = m.content + .filter((b: any) => b?.type === "text" && b.text) + .map((b: any) => b.text) + .join("\n") + .trim(); + if (text) lastAssistantText = text; + } + }); + + pi.on("agent_settled", async () => { + if (askDeferred) return; // paused_for_input is already the checkpoint of record + const out = runHook( + "stop", + { + stop_hook_active: stopRounds > 0, + final_assistant_text: lastAssistantText || null, + }, + STOP_HOOK_TIMEOUT_MS, + ); + if (out?.decision === "block" && out.reason) { + stopRounds += 1; + pi.sendUserMessage(String(out.reason), { deliverAs: "followUp" }); + } + }); + + pi.on("session_shutdown", async () => { + if (askDeferred) return; + runHook("session_end", { reason: "session ended" }); + }); + + // ---- the operator question tool ---------------------------------------------------- + pi.registerTool({ + name: "ask_operator", + label: "Ask operator", + description: + "Ask the human operator a question you cannot answer yourself (a decision, a " + + "credential, an ambiguity in the task). The run pauses; the operator's answer " + + "arrives when the session is resumed. Ask only when genuinely blocked.", + parameters: Type.Object({ + question: Type.String({ description: "The question for the operator" }), + }), + async execute() { + // Normally unreachable: the tool_call handler defers the question and + // terminates the run before execution. + return { + content: [{ type: "text", text: "Question recorded; the run will pause for the operator." }], + details: {}, + }; + }, + }); + + // ---- memory tools (the handler-memory MCP server's surface, sans MCP) -------------- + const memoryTools: Array<{ name: string; label: string; description: string; parameters: any }> = [ + { + name: "memory_search", + label: "Memory search", + description: + "Search the team memory store (notes left by earlier agent runs and the operator) " + + "for your project plus global notes. Every whitespace-separated term must match " + + "the title, body, or kind (case-insensitive). An empty query returns the most " + + "recent notes. Use this BEFORE re-deriving how something works.", + parameters: Type.Object({ + query: Type.Optional(Type.String({ description: "Search terms; empty = recent notes" })), + limit: Type.Optional(Type.Integer({ minimum: 1, maximum: 50 })), + }), + }, + { + name: "memory_get", + label: "Memory get", + description: "Fetch one memory note in full, including its links to other notes.", + parameters: Type.Object({ + note_id: Type.Integer({ description: "The note id" }), + }), + }, + { + name: "memory_save", + label: "Memory save", + description: + "Save durable knowledge for future agent runs: a fact, a decision and its " + + "rationale, a gotcha, or a runbook. Write for a reader with no context from this " + + "session. Pass note_id to update; pass global=true only for cross-project knowledge.", + parameters: Type.Object({ + title: Type.String({ description: "Short, searchable headline" }), + body: Type.String({ description: "The knowledge itself, markdown ok" }), + kind: Type.Optional(Type.String({ description: "fact | decision | gotcha | runbook" })), + tags: Type.Optional(Type.Array(Type.String())), + note_id: Type.Optional(Type.Integer({ description: "Update this note instead" })), + global: Type.Optional(Type.Boolean({ description: "Store unscoped (all projects)" })), + }), + }, + { + name: "memory_link", + label: "Memory link", + description: + "Connect two memory notes so the knowledge graph shows how they relate. " + + "Idempotent: repeating an existing link is fine.", + parameters: Type.Object({ + src_note_id: Type.Integer(), + dst_note_id: Type.Integer(), + relation: Type.Optional(Type.String({ description: "default: relates_to" })), + }), + }, + ]; + for (const tool of memoryTools) { + pi.registerTool({ + ...tool, + async execute(_toolCallId: string, params: Record) { + const text = callMemory(tool.name, params ?? {}); + return { content: [{ type: "text", text }], details: {} }; + }, + }); + } +} diff --git a/src/handler/control/pi_harness.py b/src/handler/control/pi_harness.py new file mode 100644 index 0000000..4dbf1c6 --- /dev/null +++ b/src/handler/control/pi_harness.py @@ -0,0 +1,214 @@ +"""The pi harness: run an agent through the lightweight `pi` coding agent instead of +``claude``, keeping every handler behavior (hooks, gates, memory, skills, resume). + +Why it exists: ``claude`` is a heavy loop for a local 30B — and it only speaks the +Anthropic Messages API, so a local vLLM/llama.cpp/Ollama endpoint needs a translating +proxy (LiteLLM, claude-code-router) in front of it just to make tool calling work. pi +speaks the OpenAI Completions API natively and carries a fraction of the harness +overhead, so a ``claude_models`` row with ``harness='pi'`` points pi straight at the +bare endpoint. Agents without a model backend (the Claude subscription) are untouched. + +The launch is the same shape as claude's: ``pi -p --mode json`` streaming JSON events +on stdout, supervised by :class:`handler.control.headless.RunSupervisor`. Parity with +the claude harness comes from three generated artifacts, all under a per-working-dir +``PI_CODING_AGENT_DIR`` (outside the repo tree, so no gate ever sees them as dirt): + +- ``models.json`` + ``settings.json`` — the backend row rendered as a pi provider + (``openai-completions`` by default) and pinned as the default model; ``settings.json`` + also points pi's skills discovery at the same ``~/.claude/skills`` dir the web-managed + skill sync maintains (pi implements the same SKILL.md standard), plus the repo's + committed ``.claude/skills`` (the forge role skills). +- ``extensions/handler-bridge.ts`` — the bundled extension (``pi_bridge.ts``) adapting + pi's events to ``python -m handler.hooks``: the Stop/test gate, the git-push and + merge/deploy gates, AskUserQuestion deferral (as an ``ask_operator`` tool), memory + recall at session start, and the memory tools that claude reaches over MCP. +- ``APPEND_SYSTEM.md`` — the handler conventions appended to pi's system prompt. + +Sessions: the launch passes ``--session /sessions/.jsonl`` explicitly, so the +transcript is a single file at a path handler chose — pre-assignable like claude's +``--session-id``, trivially archivable for cross-worker resume, and resuming is just +launching again with the same path (verified: pi creates the file on first use and +appends on subsequent runs). + +The task prompt travels via **stdin**, not argv: pi merges piped stdin into the prompt +and has no ``--`` separator, so argv delivery would misparse a task starting with ``-``. +""" + +from __future__ import annotations + +import json +import os +import sys +from importlib import resources +from pathlib import Path + +from ..config import get_settings + +PROVIDER_NAME = "handler" +BRIDGE_FILENAME = "handler-bridge.ts" + +# Sensible local-endpoint defaults for pi model entries; overridable per row via the +# env map keys below (the same escape hatch claude rows use for endpoint quirks). +_DEFAULT_CONTEXT_WINDOW = 128_000 +_DEFAULT_MAX_TOKENS = 16_384 + +# Row ``env`` keys the pi harness interprets itself (everything else passes through to +# the process environment unchanged): +# PI_PROVIDER_API — pi api dialect (default "openai-completions"; also accepts +# "anthropic-messages", "openai-responses", …) +# PI_CONTEXT_WINDOW — advertised context window for the model entries +# PI_MAX_TOKENS — max output tokens for the model entries +_CONFIG_KEYS = {"PI_PROVIDER_API", "PI_CONTEXT_WINDOW", "PI_MAX_TOKENS"} + +_SYSTEM_APPEND = """\ +## Handler agent contract + +You are an unattended background agent supervised by handler. No human watches this +terminal, and plain-text questions go nowhere. + +- **Questions**: when genuinely blocked on a decision only the operator can make, call + the `ask_operator` tool. The run pauses; the operator's answer arrives when the + session resumes. Never invent credentials or guess at destructive choices. +- **Completion gate**: you are only done when the project's test task (`mise run test`) + passes AND your work is committed AND pushed. Ending the session runs this gate; a + failure sends the blockers back to you — fix them rather than re-explaining. +- **Pushing**: `git push` is gated the same way (tests, then a throwaway image build + when the project defines one). A denied push tells you exactly why. +- **Team memory**: search it with `memory_search` before re-deriving how something + works — earlier runs may have written it down. Save durable findings (facts, + decisions, gotchas, runbooks) with `memory_save`; connect related notes with + `memory_link`. +""" + + +def _munged(working_dir: str) -> str: + """Same path-munging claude uses for its per-cwd session dirs (see + ``headless.munged_project_dir``): stable, filesystem-safe, layout-invariant across + workers that share ``PROJECTS_ROOT``.""" + return working_dir.replace("/", "-").replace(".", "-") + + +def pi_dir(working_dir: str) -> Path: + """The per-working-dir ``PI_CODING_AGENT_DIR`` — pi's whole config universe for this + agent (models, settings, extensions, sessions). Under ``$HOME`` and not the repo + tree, so the clean-tree completion gate never trips on generated files.""" + return Path(os.path.expanduser("~")) / ".handler-pi" / _munged(working_dir) + + +def sessions_dir(working_dir: str) -> Path: + return pi_dir(working_dir) / "sessions" + + +def session_file(working_dir: str, session_id: str) -> Path: + """The transcript path handler pre-assigns via ``--session`` (pi creates it).""" + return sessions_dir(working_dir) / f"{session_id}.jsonl" + + +def bridge_path(working_dir: str) -> Path: + return pi_dir(working_dir) / "extensions" / BRIDGE_FILENAME + + +def _model_entries(row: dict) -> list[dict]: + env = row.get("env") or {} + try: + context_window = int(env.get("PI_CONTEXT_WINDOW", _DEFAULT_CONTEXT_WINDOW)) + except (TypeError, ValueError): + context_window = _DEFAULT_CONTEXT_WINDOW + try: + max_tokens = int(env.get("PI_MAX_TOKENS", _DEFAULT_MAX_TOKENS)) + except (TypeError, ValueError): + max_tokens = _DEFAULT_MAX_TOKENS + ids: list[str] = [row["model"]] + small = row.get("small_fast_model") + if small and small not in ids: + ids.append(small) + return [ + { + "id": model_id, + "name": model_id, + "reasoning": False, + "input": ["text"], + "cost": {"input": 0, "output": 0, "cacheRead": 0, "cacheWrite": 0}, + "contextWindow": context_window, + "maxTokens": max_tokens, + } + for model_id in ids + ] + + +def write_config(working_dir: str, row: dict, api_key: str) -> Path: + """Materialize the pi config dir for one launch (spawn and resume both, like + ``claude_gen.apply``): provider + default model from the backend row, the bridge + extension, the skills pointers, and the system-prompt append. Regenerated every + launch so a row edit reaches the next run.""" + base = pi_dir(working_dir) + (base / "extensions").mkdir(parents=True, exist_ok=True) + sessions_dir(working_dir).mkdir(parents=True, exist_ok=True) + + env = row.get("env") or {} + provider = { + "name": PROVIDER_NAME, + "baseUrl": row["base_url"], + "api": env.get("PI_PROVIDER_API") or "openai-completions", + # pi hides models until the provider has *some* credential; local endpoints + # accept anything, and the placeholder never leaks a real secret. + "apiKey": api_key, + "models": _model_entries(row), + } + with open(base / "models.json", "w") as fh: + json.dump({"providers": {PROVIDER_NAME: provider}}, fh, indent=2) + + settings = { + "defaultProvider": PROVIDER_NAME, + "defaultModel": row["model"], + # pi implements the same SKILL.md standard as claude: reuse the web-managed + # sync's user-level dir, plus the repo's committed skills (forge roles). + "skills": [ + os.path.join(os.path.expanduser("~"), ".claude", "skills"), + os.path.join(working_dir, ".claude", "skills"), + ], + } + with open(base / "settings.json", "w") as fh: + json.dump(settings, fh, indent=2) + + bridge_src = resources.files("handler.control").joinpath("pi_bridge.ts").read_text() + with open(bridge_path(working_dir), "w") as fh: + fh.write(bridge_src) + + with open(base / "APPEND_SYSTEM.md", "w") as fh: + fh.write(_SYSTEM_APPEND) + return base + + +def agent_env(working_dir: str, row: dict) -> dict[str, str]: + """The env overrides a pi-harness agent launches with (the pi analog of the + ``ANTHROPIC_*`` set): pi's config dir, offline startup (a local endpoint serves no + update checks), and the interpreter the bridge shells hooks out to. Row env extras + win over everything, minus the keys the config writer already consumed.""" + env = { + "PI_CODING_AGENT_DIR": str(pi_dir(working_dir)), + "PI_OFFLINE": "1", + "HANDLER_PYTHON": sys.executable, + } + for k, v in (row.get("env") or {}).items(): + if str(k) not in _CONFIG_KEYS: + env[str(k)] = str(v) + return env + + +def build_argv(session_id: str, working_dir: str) -> list[str]: + """The headless pi invocation — identical for spawn and resume, because the + pre-assigned ``--session`` file either doesn't exist yet (spawn: pi creates it) or + carries the history (resume: pi appends). The prompt is NOT here: it is piped to + stdin by the supervisor (pi has no ``--`` separator, so argv can't safely carry an + arbitrary task). ``--no-extensions`` disables discovery — the bridge is the one + extension, loaded explicitly — so a managed repo can't inject code into the run.""" + s = get_settings() + return [ + s.pi_bin, + "-p", + "--mode", "json", + "--no-extensions", + "-e", str(bridge_path(working_dir)), + "--session", str(session_file(working_dir, session_id)), + ] diff --git a/src/handler/control/spawn.py b/src/handler/control/spawn.py index aa76885..872a050 100644 --- a/src/handler/control/spawn.py +++ b/src/handler/control/spawn.py @@ -22,6 +22,7 @@ from . import ( headless, mise, models, + pi_harness, reposync, settings_gen, worktree, @@ -139,7 +140,7 @@ def spawn( # Same fail-fast contract as credentials: a selected model backend that is # missing, disabled, or undecryptable refuses the spawn before any row exists. try: - models.resolve_model_env(conn, model_id, require_enabled=True) + models.resolve_model(conn, model_id, require_enabled=True) except models.ModelError as exc: raise SpawnError(str(exc)) from exc @@ -155,9 +156,10 @@ def spawn( settings_path = settings_gen.write_settings(working_dir) # Materialize the web-managed Claude config (MCP connectors + user-level skills) - # so this launch picks up what the operator configured in the dashboard. + # so this launch picks up what the operator configured in the dashboard. The skills + # half also feeds pi-harness agents (their settings.json points at the same dir). claude_gen.apply(working_dir) - env = _agent_env(project, agent, token, role=role, mise_init=mise_init) + env, harness = _agent_env(project, agent, token, role=role, mise_init=mise_init) # Verify the pinned forge version, if one is configured. Non-fatal: a version drift # is recorded as a warning rather than blocking the spawn, since not every agent @@ -171,6 +173,7 @@ def spawn( settings_path=settings_path, env=env, worker_id=worker_id or f"cli-{os.getpid()}", + harness=harness, ) agent = {**agent, "forge_note": forge_note, "sync_note": sync_note} return agent @@ -183,10 +186,11 @@ def _agent_env( *, role: str | None = None, mise_init: bool = False, -) -> dict[str, str]: - """The environment an agent process (and therefore its hooks) runs with: identity, - ``DATABASE_URL``, and resolved forge/git credentials. Shared by spawn and the - headless resume path (a resume is a brand-new process needing the same env).""" +) -> tuple[dict[str, str], str]: + """The environment an agent process (and therefore its hooks) runs with — identity, + ``DATABASE_URL``, and resolved forge/git credentials — plus which harness launches + it. Shared by spawn and the headless resume path (a resume is a brand-new process + needing the same env).""" env = { "HANDLER_PROJECT_ID": project["id"], "HANDLER_AGENT_NAME": agent["name"], @@ -202,13 +206,22 @@ def _agent_env( # A short read connection lets credential/host resolution consult the forge_hosts # registry (falling back to the built-in host map when a host has no row). with connection() as conn: - # Agent pinned to a model backend: point the claude binary at it. Resumes are a + # Agent pinned to a model backend: point its harness at it. Resumes are a # brand-new process, so this is what keeps an agent on the backend it started on # (a since-disabled backend may still finish; a deleted one fails loudly). try: - env.update(models.resolve_model_env(conn, agent.get("model_id"))) + resolved = models.resolve_model(conn, agent.get("model_id")) except models.ModelError as exc: raise SpawnError(str(exc)) from exc + harness = models.harness_of(resolved) + if harness == "pi": + row, key = resolved + # The pi config dir is regenerated per launch, same contract as the claude + # settings/skills materialization — a row edit reaches the next run. + pi_harness.write_config(agent["working_dir"], row, key) + env.update(pi_harness.agent_env(agent["working_dir"], row)) + else: + env.update(models.claude_env(resolved)) env.update(credentials.credential_env(token, project.get("git_remote"), conn)) if token: _install_git_credentials(agent["working_dir"], project.get("git_remote"), conn) @@ -217,7 +230,7 @@ def _agent_env( env.update(reposync.ssh_env(project.get("git_remote"), conn)) except reposync.SyncError as exc: raise SpawnError(str(exc)) from exc - return env + return env, harness def _check_forge_version(working_dir: str) -> str | None: @@ -278,20 +291,23 @@ def resume(agent: dict, answer: str, worker_id: str | None = None) -> tuple[bool token = credentials.resolve_for_project(project, conn) except credentials.CredentialError as exc: return False, str(exc) - env = _agent_env(project, agent, token) + env, harness = _agent_env(project, agent, token) if not agent.get("session_id"): # Pre-headless agent row (or a spawn that never launched): nothing to --resume. - return _resume_reinjected(agent, answer, settings_path, env, worker_id) + return _resume_reinjected(agent, answer, settings_path, env, worker_id, harness) - transcript = headless.session_dir(working_dir) / f"{agent['session_id']}.jsonl" + if harness == "pi": + transcript = pi_harness.session_file(working_dir, agent["session_id"]) + else: + transcript = headless.session_dir(working_dir) / f"{agent['session_id']}.jsonl" if archive is not None: try: - headless.materialize_session(working_dir, bytes(archive["archive"])) + headless.materialize_session(working_dir, bytes(archive["archive"]), harness=harness) except (OSError, ValueError) as exc: return False, f"could not materialize session archive: {exc}" elif not transcript.exists(): - return _resume_reinjected(agent, answer, settings_path, env, worker_id) + return _resume_reinjected(agent, answer, settings_path, env, worker_id, harness) try: run = headless.launch( @@ -301,6 +317,7 @@ def resume(agent: dict, answer: str, worker_id: str | None = None) -> tuple[bool settings_path=settings_path, env=env, worker_id=worker_id, + harness=harness, ) except repo.RunConflictError: # Another worker won the race for this resume (two queued resume commands, or a @@ -311,7 +328,12 @@ def resume(agent: dict, answer: str, worker_id: str | None = None) -> tuple[bool def _resume_reinjected( - agent: dict, answer: str, settings_path: str, env: dict, worker_id: str + agent: dict, + answer: str, + settings_path: str, + env: dict, + worker_id: str, + harness: str = "claude", ) -> tuple[bool, str]: """Degraded resume: no transcript anywhere, so start a fresh session with the context rebuilt from the DB. Continuity is approximate — say so in the event log.""" @@ -341,6 +363,7 @@ def _resume_reinjected( settings_path=settings_path, env=env, worker_id=worker_id, + harness=harness, ) except repo.RunConflictError: return False, "another worker is already running this agent's session" diff --git a/src/handler/db/repository.py b/src/handler/db/repository.py index 05267b2..2f4853e 100644 --- a/src/handler/db/repository.py +++ b/src/handler/db/repository.py @@ -1198,6 +1198,7 @@ def create_claude_model( model: str, api_key_enc: str | None = None, small_fast_model: str | None = None, + harness: str = "claude", env: dict | None = None, enabled: bool = True, ) -> dict: @@ -1208,6 +1209,7 @@ def create_claude_model( api_key_enc=api_key_enc, model=model, small_fast_model=small_fast_model, + harness=harness, env=env, enabled=enabled, created_at=_now(), @@ -1217,7 +1219,16 @@ def create_claude_model( def update_claude_model(conn: Connection, model_id: int, **fields: Any) -> dict | None: - allowed = {"name", "base_url", "api_key_enc", "model", "small_fast_model", "env", "enabled"} + allowed = { + "name", + "base_url", + "api_key_enc", + "model", + "small_fast_model", + "harness", + "env", + "enabled", + } values = {k: v for k, v in fields.items() if k in allowed} if values: conn.execute( diff --git a/src/handler/db/tables.py b/src/handler/db/tables.py index d61fb33..64c3882 100644 --- a/src/handler/db/tables.py +++ b/src/handler/db/tables.py @@ -424,6 +424,10 @@ claude_models = Table( Column("api_key_enc", String), # encrypted (HANDLER_SECRET_KEY); never returned by the API Column("model", String, nullable=False), # ANTHROPIC_MODEL — the id the endpoint serves Column("small_fast_model", String), # ANTHROPIC_SMALL_FAST_MODEL; falls back to ``model`` + # Which agent binary this backend launches: "claude" (default — the claude binary + # pointed at an Anthropic-API-compatible endpoint) or "pi" (the lightweight pi coding + # agent, which speaks OpenAI-compatible endpoints natively — no translation proxy). + Column("harness", String, nullable=False, server_default="claude"), Column("env", PortableJSON), # extra env overrides (timeouts, max tokens, …), merged last Column("enabled", Boolean, nullable=False, server_default="1"), Column("created_at", PortableTimestamp, nullable=False, server_default=func.now()), diff --git a/src/handler/hooks/checkpoint.py b/src/handler/hooks/checkpoint.py index 9004b9f..7e09620 100644 --- a/src/handler/hooks/checkpoint.py +++ b/src/handler/hooks/checkpoint.py @@ -169,8 +169,11 @@ def handle_stop(conn: Connection, ident: Identity, hook_input: HookInput) -> dic ) # A done agent's checkmark carries its own closing message — the substance the # dashboard shows; a blocked one carries the blockers (the next turn will re-capture - # the narrative once the gate clears). - final_text = _final_assistant_text(hook_input.transcript_path) + # the narrative once the gate clears). A harness that passes the text directly (pi) + # wins over the claude-transcript parse. + final_text = hook_input.final_assistant_text or _final_assistant_text( + hook_input.transcript_path + ) where_it_stopped = final_text[:4000] if not blockers and final_text else summary log_id = repo.insert_log_entry( diff --git a/src/handler/hooks/context.py b/src/handler/hooks/context.py index 5b03ad4..aaa5bc5 100644 --- a/src/handler/hooks/context.py +++ b/src/handler/hooks/context.py @@ -52,6 +52,16 @@ class HookInput: def stop_hook_active(self) -> bool: return bool(self.raw.get("stop_hook_active")) + @property + def final_assistant_text(self) -> str | None: + """The agent's closing message, when the harness passes it directly (the pi + bridge does — pi session files aren't claude-transcript-shaped, so the + transcript fallback can't parse them).""" + value = self.raw.get("final_assistant_text") + if isinstance(value, str) and value.strip(): + return value + return None + @property def reason(self) -> str | None: return self.raw.get("reason") diff --git a/src/handler/mcpserver/__main__.py b/src/handler/mcpserver/__main__.py index 08d6f4d..f3834be 100644 --- a/src/handler/mcpserver/__main__.py +++ b/src/handler/mcpserver/__main__.py @@ -1,10 +1,50 @@ -"""``python -m handler.mcpserver`` — run the bundled handler-memory MCP server.""" +"""``python -m handler.mcpserver`` — run the bundled handler-memory MCP server. + +``--call `` runs one memory tool directly instead: JSON arguments on stdin, JSON +result on stdout, exit 1 with the error on stderr for a failed call. This is the seam +the pi harness's bridge extension uses (pi has no MCP by design), sharing the exact +tool implementations — and the same identity-from-environment contract — the MCP +server dispatches to. +""" from __future__ import annotations +import json +import os import sys -from . import serve +from . import MemoryServer, serve + + +def call_tool(name: str, stdin=None, stdout=None) -> int: + stdin = stdin or sys.stdin + stdout = stdout or sys.stdout + raw = stdin.read() + try: + args = json.loads(raw) if raw.strip() else {} + except ValueError: + print("invalid JSON arguments on stdin", file=sys.stderr) + return 2 + agent_id_raw = os.environ.get("HANDLER_AGENT_ID") + server = MemoryServer( + agent_id=int(agent_id_raw) if agent_id_raw else None, + project_id=os.environ.get("HANDLER_PROJECT_ID") or None, + ) + try: + payload = server.call_tool(name, args if isinstance(args, dict) else {}) + except Exception as exc: # noqa: BLE001 - the caller renders this as a tool error + print(f"error: {exc}", file=sys.stderr) + return 1 + print(json.dumps(payload, ensure_ascii=False), file=stdout, flush=True) + return 0 + + +def main(argv: list[str] | None = None) -> int: + argv = sys.argv[1:] if argv is None else argv + if len(argv) >= 2 and argv[0] == "--call": + return call_tool(argv[1]) + return serve() + if __name__ == "__main__": - sys.exit(serve()) + sys.exit(main()) diff --git a/src/handler/migrations/versions/0015_model_harness.py b/src/handler/migrations/versions/0015_model_harness.py new file mode 100644 index 0000000..ca98256 --- /dev/null +++ b/src/handler/migrations/versions/0015_model_harness.py @@ -0,0 +1,37 @@ +"""model backends: harness selection (claude | pi) + +Revision ID: 0015_model_harness +Revises: 0014_agent_memory +Create Date: 2026-08-12 + +Adds ``claude_models.harness`` — which agent binary a backend row launches. ``claude`` +(the default, and what every existing row becomes) keeps the current behavior: the +``claude`` binary pointed at an Anthropic-API-compatible endpoint. ``pi`` launches the +same agent run through the lightweight `pi` coding agent instead, which speaks the +OpenAI Completions API natively — so a local vLLM/llama.cpp/Ollama endpoint needs no +Anthropic-translation proxy in front of it. +""" + +from __future__ import annotations + +from collections.abc import Sequence + +import sqlalchemy as sa +from alembic import op + +revision: str = "0015_model_harness" +down_revision: str | None = "0014_agent_memory" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + op.add_column( + "claude_models", + sa.Column("harness", sa.String(), nullable=False, server_default="claude"), + ) + + +def downgrade() -> None: + with op.batch_alter_table("claude_models", schema=None) as batch_op: + batch_op.drop_column("harness") diff --git a/tests/conftest.py b/tests/conftest.py index d931253..3ad54bb 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -135,14 +135,15 @@ def fake_launch(monkeypatch): calls: list[dict] = [] - def launch(agent, *, kind, prompt, settings_path, env, worker_id, on_exit=None): + def launch(agent, *, kind, prompt, settings_path, env, worker_id, on_exit=None, + harness="claude"): session_id = agent.get("session_id") if kind == "resume" else f"fake-sid-{len(calls) + 1}" with connection() as conn: run = repo.create_run(conn, agent["id"], session_id, worker_id, kind) repo.set_agent_session(conn, agent["id"], session_id, worker_id) calls.append( {"agent": agent, "kind": kind, "prompt": prompt, "settings_path": settings_path, - "env": env, "worker_id": worker_id, "run": run} + "env": env, "worker_id": worker_id, "run": run, "harness": harness} ) return run diff --git a/tests/fixtures/fake_pi.py b/tests/fixtures/fake_pi.py new file mode 100755 index 0000000..4d9c8e2 --- /dev/null +++ b/tests/fixtures/fake_pi.py @@ -0,0 +1,126 @@ +#!/usr/bin/env python3 +"""A stand-in ``pi`` binary for pi-harness headless-runner tests. + +Wired in via the ``pi_bin`` setting (the same seam ``fake_claude.py`` uses for +``claude_bin``). Parses the real pi-harness argv (``-p --mode json --no-extensions +-e --session ``), reads the prompt from **stdin** (pi has no ``--`` +separator, so that is how the supervisor delivers it), emits a scripted ``--mode json`` +event stream on stdout, and appends to the genuine ``--session`` transcript file so +archive/materialize/resume paths exercise the real single-file layout. Behavior is +selected with ``FAKE_PI_MODE``: + +- ``success`` (default): session header + user/assistant message_end + agent_end + + agent_settled, exit 0. +- ``error``: header + one garbage line + agent_end whose assistant stopReason is + ``error``, then exit 1 (pi's exit code for an errored final message). +- ``hang``: header, then sleep forever (kill/cancel tests SIGTERM it). + +``FAKE_PI_EXPECT_HISTORY=1`` makes a run fail loudly (exit 3) when the ``--session`` +file does not already exist — the cross-worker resume tests use it to prove the +archive really was materialized where pi would look. +""" + +from __future__ import annotations + +import json +import os +import signal +import sys +import time +from pathlib import Path + + +def _parse_argv(argv: list[str]) -> dict: + opts = { + "print": False, + "mode": None, + "no_extensions": False, + "extension": None, + "session": None, + } + i = 0 + while i < len(argv): + arg = argv[i] + if arg == "-p": + opts["print"] = True + elif arg == "--mode": + i += 1 + opts["mode"] = argv[i] + elif arg == "--no-extensions": + opts["no_extensions"] = True + elif arg == "-e": + i += 1 + opts["extension"] = argv[i] + elif arg == "--session": + i += 1 + opts["session"] = argv[i] + i += 1 + return opts + + +def _emit(event: dict) -> None: + sys.stdout.write(json.dumps(event) + "\n") + sys.stdout.flush() + + +def _assistant(text: str) -> dict: + return { + "role": "assistant", + "content": [{"type": "text", "text": text}], + "stopReason": "stop", + } + + +def main() -> int: + signal.signal(signal.SIGTERM, signal.SIG_DFL) + mode = os.environ.get("FAKE_PI_MODE", "success") + opts = _parse_argv(sys.argv[1:]) + if not opts["print"] or opts["mode"] != "json" or not opts["session"]: + sys.stderr.write("fake_pi: expected -p --mode json --session \n") + return 64 + if not opts["no_extensions"] or not opts["extension"]: + sys.stderr.write("fake_pi: expected --no-extensions with an explicit -e bridge\n") + return 64 + + prompt = sys.stdin.read().strip() + session_path = Path(opts["session"]) + is_resume = session_path.exists() + if os.environ.get("FAKE_PI_EXPECT_HISTORY") and not is_resume: + sys.stderr.write(f"fake_pi: session file missing at {session_path}\n") + return 3 + + _emit({"type": "session", "version": 3, "id": "internal-uuid", "cwd": os.getcwd()}) + if mode == "hang": + time.sleep(3600) + return 0 + + user_msg = {"role": "user", "content": [{"type": "text", "text": prompt}]} + _emit({"type": "agent_start"}) + _emit({"type": "message_end", "message": user_msg}) + if mode == "error": + sys.stdout.write("this is not json\n") + sys.stdout.flush() + errored = { + "role": "assistant", + "content": [], + "stopReason": "error", + "errorMessage": "Connection error.", + } + _emit({"type": "agent_end", "messages": [user_msg, errored], "willRetry": False}) + return 1 + + assistant = _assistant(f"working on: {prompt}") + _emit({"type": "message_end", "message": assistant}) + + session_path.parent.mkdir(parents=True, exist_ok=True) + with session_path.open("a") as fh: + fh.write(json.dumps({"type": "message", "message": user_msg}) + "\n") + fh.write(json.dumps({"type": "message", "message": assistant}) + "\n") + + _emit({"type": "agent_end", "messages": [user_msg, assistant], "willRetry": False}) + _emit({"type": "agent_settled"}) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/test_pi_harness.py b/tests/test_pi_harness.py new file mode 100644 index 0000000..250515b --- /dev/null +++ b/tests/test_pi_harness.py @@ -0,0 +1,372 @@ +"""The pi harness: config generation, the launch/resume paths against the fake ``pi`` +binary, event normalization, and the API/spawn plumbing that selects it. + +Same testing philosophy as the claude runner (``test_headless_run``): real +subprocesses, real threads, real SQLite — ``fake_pi.py`` stands in for the binary via +the ``pi_bin`` setting and emits genuine ``--mode json`` events, so what lands in the +DB is exactly what the API/UI will read. The bridge extension itself is TypeScript and +runs inside real pi, so here it is asserted as an artifact (installed, wired into +argv); its hook contract is the same ``python -m handler.hooks`` surface the hook tests +already cover. +""" + +from __future__ import annotations + +import json +import time +from pathlib import Path + +import pytest + +from handler.control import headless, models, pi_harness, spawn +from handler.db import repository as repo +from handler.db.engine import get_engine + +REPO_ROOT = Path(__file__).resolve().parents[1] +FAKE_PI = str(REPO_ROOT / "tests" / "fixtures" / "fake_pi.py") + + +@pytest.fixture +def pi_env(env, monkeypatch): + from handler import config + + monkeypatch.setenv("PI_BIN", FAKE_PI) + config.get_settings.cache_clear() + yield env + config.get_settings.cache_clear() + + +def _pi_row(**overrides): + row = { + "name": "qwen-local", + "base_url": "http://127.0.0.1:8000/v1", + "model": "qwen3-coder-30b", + "small_fast_model": "qwen3-1.7b", + "harness": "pi", + "env": {}, + } + row.update(overrides) + return row + + +def _make_agent(tmp_path, name="p1", model_id=None): + working_dir = tmp_path / "projects" / "p" / name + working_dir.mkdir(parents=True) + with get_engine().begin() as conn: + if repo.get_project(conn, "p") is None: + repo.create_project(conn, "p", str(tmp_path / "projects" / "p")) + agent = repo.create_agent(conn, "p", name, str(working_dir), model_id=model_id) + return agent + + +def _wait_for(predicate, timeout=20.0): + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + result = predicate() + if result: + return result + time.sleep(0.1) + return None + + +def _finished_run(run_id): + def check(): + with get_engine().begin() as conn: + run = repo.get_run(conn, run_id) + return run if run["status"] != "running" else None + + return check + + +# --- config generation ------------------------------------------------------------------- + + +def test_write_config_renders_provider_and_bridge(pi_env, tmp_path): + wd = str(tmp_path / "wd") + base = pi_harness.write_config(wd, _pi_row(), "sk-local-123") + + provider = json.loads((base / "models.json").read_text())["providers"]["handler"] + assert provider["baseUrl"] == "http://127.0.0.1:8000/v1" + assert provider["api"] == "openai-completions" + assert provider["apiKey"] == "sk-local-123" + assert [m["id"] for m in provider["models"]] == ["qwen3-coder-30b", "qwen3-1.7b"] + + settings = json.loads((base / "settings.json").read_text()) + assert settings["defaultProvider"] == "handler" + assert settings["defaultModel"] == "qwen3-coder-30b" + # Skills parity: the web-managed sync's user dir plus the repo's committed skills. + assert any(s.endswith(".claude/skills") for s in settings["skills"]) + assert any(s.startswith(wd) for s in settings["skills"]) + + bridge = pi_harness.bridge_path(wd) + assert bridge.exists() + text = bridge.read_text() + # The bridge is the hooks adapter — it must shell to the hook dispatcher and the + # memory tool CLI, and register the question-deferral tool. + assert "handler.hooks" in text + assert "handler.mcpserver" in text + assert "ask_operator" in text + assert (base / "APPEND_SYSTEM.md").read_text().strip() + + +def test_write_config_row_env_tunes_provider(pi_env, tmp_path): + wd = str(tmp_path / "wd") + row = _pi_row( + env={ + "PI_PROVIDER_API": "anthropic-messages", + "PI_CONTEXT_WINDOW": "32000", + "PI_MAX_TOKENS": "4096", + "SOME_VAR": "yes", + } + ) + base = pi_harness.write_config(wd, row, "k") + provider = json.loads((base / "models.json").read_text())["providers"]["handler"] + assert provider["api"] == "anthropic-messages" + assert provider["models"][0]["contextWindow"] == 32000 + assert provider["models"][0]["maxTokens"] == 4096 + + env = pi_harness.agent_env(wd, row) + # Config-only keys are consumed by the writer, not leaked into the process env; + # everything else passes through, and the defaults are present. + assert "PI_PROVIDER_API" not in env + assert env["SOME_VAR"] == "yes" + assert env["PI_OFFLINE"] == "1" + assert env["PI_CODING_AGENT_DIR"] == str(pi_harness.pi_dir(wd)) + assert env["HANDLER_PYTHON"] + + +def test_build_argv_wires_bridge_and_session(pi_env, tmp_path): + wd = str(tmp_path / "wd") + argv = pi_harness.build_argv("sid-1", wd) + assert argv[1:5] == ["-p", "--mode", "json", "--no-extensions"] + assert argv[argv.index("-e") + 1] == str(pi_harness.bridge_path(wd)) + assert argv[argv.index("--session") + 1] == str(pi_harness.session_file(wd, "sid-1")) + # No prompt in argv: pi has no ``--`` separator, so the task travels on stdin. + assert argv[-1] == str(pi_harness.session_file(wd, "sid-1")) + + +# --- model resolution -------------------------------------------------------------------- + + +def test_resolve_model_and_harness(conn): + row = repo.create_claude_model( + conn, "qwen-pi", "http://127.0.0.1:8000/v1", "qwen3", harness="pi" + ) + resolved = models.resolve_model(conn, row["id"]) + assert models.harness_of(resolved) == "pi" + assert models.harness_of(None) == "claude" + # A pi row produces no ANTHROPIC_* env — its config is files, not env. + assert models.resolve_model_env(conn, row["id"]) == {} + # claude rows are unchanged. + claude_row = repo.create_claude_model(conn, "qwen-claude", "http://llm:4000", "qwen3") + env = models.resolve_model_env(conn, claude_row["id"]) + assert env["ANTHROPIC_BASE_URL"] == "http://llm:4000" + + +# --- headless runs against the fake pi binary --------------------------------------------- + + +def test_pi_spawn_streams_events_and_completes(pi_env, tmp_path): + agent = _make_agent(tmp_path) + pi_harness.write_config(agent["working_dir"], _pi_row(), "k") + run = headless.launch( + agent, kind="spawn", prompt="build the thing", + settings_path=str(tmp_path / "s.json"), env={}, worker_id="w1", harness="pi", + ) + finished = _wait_for(_finished_run(run["id"])) + assert finished is not None, "run never finished" + assert finished["status"] == "completed" + assert finished["exit_code"] == 0 + # agent_end normalized into the result the run row stores. + assert finished["result"]["is_error"] is False + assert finished["result"]["harness"] == "pi" + + with get_engine().begin() as conn: + events = repo.list_agent_events(conn, agent["id"]) + updated = repo.get_agent_by_id(conn, agent["id"]) + archive = repo.get_session_archive(conn, agent["id"]) + + types = [e["type"] for e in events] + assert "session" in types and "agent_end" in types and "message_end" in types + # last_output comes from assistant message_end events (user ones don't count). + assert updated["last_output"] == "working on: build the thing" + assert updated["status"] == "blocked" # no hooks ran in the fake — not done + assert updated["session_id"] == run["session_id"] + # The single-file pi session was archived for cross-worker resume. + assert archive is not None + assert pi_harness.session_file(agent["working_dir"], run["session_id"]).exists() + + +def test_pi_failed_run_records_error_result(pi_env, tmp_path, monkeypatch): + monkeypatch.setenv("FAKE_PI_MODE", "error") + agent = _make_agent(tmp_path, "p-err") + run = headless.launch( + agent, kind="spawn", prompt="boom", + settings_path=str(tmp_path / "s.json"), env={}, worker_id="w1", harness="pi", + ) + finished = _wait_for(_finished_run(run["id"])) + assert finished["status"] == "failed" + assert finished["exit_code"] == 1 + assert finished["result"]["is_error"] is True + + with get_engine().begin() as conn: + events = repo.list_agent_events(conn, agent["id"]) + assert repo.get_agent_by_id(conn, agent["id"])["status"] == "blocked" + raw = next(e for e in events if e["type"] == "raw") + assert "this is not json" in raw["payload"]["line"] + + +def test_pi_cancel_terminates_hanging_run(pi_env, tmp_path, monkeypatch): + monkeypatch.setenv("FAKE_PI_MODE", "hang") + agent = _make_agent(tmp_path, "p-hang") + run = headless.launch( + agent, kind="spawn", prompt="hang", + settings_path=str(tmp_path / "s.json"), env={}, worker_id="w1", harness="pi", + ) + _wait_for(lambda: _events_count(agent["id"]) >= 1) + with get_engine().begin() as conn: + assert repo.request_run_cancel(conn, run["id"]) is True + finished = _wait_for(_finished_run(run["id"]), timeout=30.0) + assert finished is not None and finished["status"] == "canceled" + + +def _events_count(agent_id): + with get_engine().begin() as conn: + return len(repo.list_agent_events(conn, agent_id)) + + +def test_pi_cross_worker_resume_materializes_single_file(pi_env, tmp_path, monkeypatch): + """Worker B resumes a pi session it never ran, from the DB archive alone — the pi + analog of the claude linchpin test, on the single-file session layout.""" + with get_engine().begin() as conn: + model = repo.create_claude_model( + conn, "qwen-pi", "http://127.0.0.1:8000/v1", "qwen3", harness="pi" + ) + agent = _make_agent(tmp_path, "p-resume", model_id=model["id"]) + pi_harness.write_config(agent["working_dir"], _pi_row(), "k") + run = headless.launch( + agent, kind="spawn", prompt="first pass", + settings_path=str(tmp_path / "s.json"), env={}, worker_id="worker-a", harness="pi", + ) + assert _wait_for(_finished_run(run["id"]))["status"] == "completed" + + # "Worker B": a clean HOME — no pi config, no session file. + other_home = tmp_path / "worker-b-home" + other_home.mkdir() + monkeypatch.setenv("HOME", str(other_home)) + # The fake proves materialization: it exits 3 when the session file is absent. + monkeypatch.setenv("FAKE_PI_EXPECT_HISTORY", "1") + + with get_engine().begin() as conn: + agent = repo.get_agent_by_id(conn, agent["id"]) + ok, detail = spawn.resume(agent, "the operator's answer", worker_id="worker-b") + assert ok, detail + + with get_engine().begin() as conn: + resumed = repo.get_latest_run(conn, agent["id"]) + assert resumed["kind"] == "resume" + finished = _wait_for(_finished_run(resumed["id"])) + assert finished["status"] == "completed", f"exit={finished['exit_code']}" + assert finished["session_id"] == run["session_id"] # same session, continued + # Resume regenerated the pi config under worker B's HOME (row edits reach resumes). + assert pi_harness.pi_dir(agent["working_dir"]).exists() + + +# --- spawn plumbing ------------------------------------------------------------------------ + + +def test_spawn_with_pi_model_launches_pi_harness(env, fake_launch, tmp_path): + root = tmp_path / "projects" / "proj" + root.mkdir(parents=True) + (root / ".mise.toml").write_text("[tasks.test]\nrun = 'pytest'\n") + with get_engine().begin() as conn: + repo.create_project(conn, "proj", str(root)) + model = repo.create_claude_model( + conn, "qwen-pi", "http://127.0.0.1:8000/v1", "qwen3", harness="pi" + ) + + agent = spawn.spawn("proj", "worker", task="do it", model_id=model["id"]) + call = fake_launch[0] + assert call["harness"] == "pi" + assert call["env"]["PI_CODING_AGENT_DIR"] == str(pi_harness.pi_dir(agent["working_dir"])) + assert "ANTHROPIC_BASE_URL" not in call["env"] + # The config artifacts were materialized before launch. + assert pi_harness.bridge_path(agent["working_dir"]).exists() + provider = json.loads( + (pi_harness.pi_dir(agent["working_dir"]) / "models.json").read_text() + )["providers"]["handler"] + assert provider["baseUrl"] == "http://127.0.0.1:8000/v1" + + +def test_spawn_with_claude_model_still_launches_claude(env, fake_launch, tmp_path): + root = tmp_path / "projects" / "proj2" + root.mkdir(parents=True) + (root / ".mise.toml").write_text("[tasks.test]\nrun = 'pytest'\n") + with get_engine().begin() as conn: + repo.create_project(conn, "proj2", str(root)) + model = repo.create_claude_model(conn, "qwen-claude", "http://llm:4000", "qwen3") + + spawn.spawn("proj2", "worker", task="do it", model_id=model["id"]) + call = fake_launch[0] + assert call["harness"] == "claude" + assert call["env"]["ANTHROPIC_BASE_URL"] == "http://llm:4000" + assert "PI_CODING_AGENT_DIR" not in call["env"] + + +# --- API ------------------------------------------------------------------------------------- + + +def test_model_api_harness_round_trip(client, auth): + r = client.post( + "/claude/models", + json={ + "name": "qwen-pi", + "base_url": "http://127.0.0.1:8000/v1", + "model": "qwen3", + "harness": "pi", + }, + headers=auth, + ) + assert r.status_code == 201 + assert r.json()["harness"] == "pi" + # Default stays claude, and junk is rejected with a clean 422. + r = client.post( + "/claude/models", + json={"name": "plain", "base_url": "http://llm:4000", "model": "m"}, + headers=auth, + ) + assert r.json()["harness"] == "claude" + r = client.post( + "/claude/models", + json={"name": "bad", "base_url": "http://x", "model": "m", "harness": "aider"}, + headers=auth, + ) + assert r.status_code == 422 + + model_id = client.get("/claude/models", headers=auth).json()[0]["id"] + r = client.patch(f"/claude/models/{model_id}", json={"harness": "pi"}, headers=auth) + assert r.status_code == 200 and r.json()["harness"] == "pi" + + +# --- hook input: harness-provided final text -------------------------------------------------- + + +def test_stop_checkpoint_prefers_harness_final_text(env, monkeypatch, tmp_path): + """The pi bridge passes the closing message directly (pi session files aren't + claude-transcript-shaped); the checkpoint must prefer it over the transcript parse.""" + from handler.hooks import checkpoint, verify + from handler.hooks.context import HookInput, Identity + + monkeypatch.setattr(verify, "run_test", lambda cwd: (True, "ok")) + agent = _make_agent(tmp_path, "hooked") + ident = Identity(agent["id"], "p", "hooked", working_dir=agent["working_dir"]) + hook_input = HookInput( + raw={"session_id": "s1", "final_assistant_text": "shipped the feature"}, + event="stop", + ) + with get_engine().begin() as conn: + result = checkpoint.handle_stop(conn, ident, hook_input) + cm = repo.get_checkmark(conn, agent["id"]) + assert result == {} + assert cm["status"] == "done" + assert cm["where_it_stopped"] == "shipped the feature"