mirror of
https://github.com/0xWheatyz/handler.git
synced 2026-08-29 19:21:40 +00:00
Add the pi harness: lightweight local-model agents with full gate parity
Model backend rows gain a harness column (claude | pi). A pi-harness row runs the agent through the pi coding agent instead of the claude binary — pi speaks the OpenAI Completions API natively, so a bare vLLM/llama.cpp/Ollama endpoint needs no LiteLLM/claude-code-router translation proxy, and the loop is far lighter for slow local token throughput. The Claude subscription and existing claude-harness backends are untouched. Parity comes from generated per-agent artifacts under ~/.handler-pi (outside the repo tree, so the clean-tree gate never trips): models.json + settings.json render the row as a pi provider pinned as the default model; a bundled bridge extension (pi_bridge.ts) adapts pi's events to the exact stdin/stdout contract of `python -m handler.hooks` — the Stop/completion gate re-prompts pi with blockers via a follow-up message, git push runs the test/build/approval gates and denies on failure, questions defer through an ask_operator tool into the normal answer/resume flow, and memory recall is injected at session start. The memory tools are registered natively (pi has no MCP), shelling to a new `python -m handler.mcpserver --call <tool>` seam that reuses the MCP server's implementations. Skills reuse the same ~/.claude/skills sync (pi implements the same SKILL.md standard) plus the repo's committed .claude/skills. Sessions are single JSONL files pre-assigned via --session, so cross-worker resume archives/materializes exactly like claude's; the prompt travels on stdin (pi has no -- separator). The supervisor normalizes pi's event stream on the fly: assistant message_end feeds last_output, the final agent_end becomes the run result. The whole chain was validated live against pi 0.84.1 with a stub OpenAI endpoint: memory injection, push-gate denial (including the protected- branch approval gate), stop-gate block loop, and ask_operator pause all ran end to end through the real hooks and DB. Also: harness selector in the dashboard Models form, pi baked into the control image (NodeSource 22 for pi's node >= 22.19 floor), PI_BIN override, docs in docs/local-models.md, fake_pi fixture + 12 tests (361 total green). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01KdGv3u3DfTsP1S188KDhVH
This commit is contained in:
Claude
@@ -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
|
||||
|
||||
+8
-4
@@ -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 \
|
||||
|
||||
@@ -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
|
||||
|
||||
+81
-4
@@ -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://<host>: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://<host>: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 <tool>` 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 <path>`), 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,
|
||||
|
||||
@@ -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() {
|
||||
<>
|
||||
<div className="faint" style={{ fontSize: "var(--text-sm)", marginBottom: 14 }}>
|
||||
Alternative model backends the spawn dropdown offers next to the Claude
|
||||
subscription — the same <span className="mono">claude</span> binary pointed at a
|
||||
different endpoint via <span className="mono">ANTHROPIC_BASE_URL</span>, so
|
||||
skills, connectors, hooks, and gates apply unchanged. The endpoint must speak the{" "}
|
||||
<b>Anthropic Messages API including tool use</b> — 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{" "}
|
||||
<span className="mono">docs/local-models.md</span> for working Qwen-Coder stacks.
|
||||
subscription. The <b>claude</b> harness is the same{" "}
|
||||
<span className="mono">claude</span> binary pointed at a different endpoint via{" "}
|
||||
<span className="mono">ANTHROPIC_BASE_URL</span> — the endpoint must speak the{" "}
|
||||
<b>Anthropic Messages API including tool use</b>, so front a bare
|
||||
OpenAI-compatible server with LiteLLM or claude-code-router. The <b>pi</b>{" "}
|
||||
harness runs the lightweight pi coding agent instead, which speaks{" "}
|
||||
<b>OpenAI-compatible endpoints natively</b> (vLLM, llama.cpp, Ollama — no
|
||||
translation proxy) — hooks, gates, memory, and skills still apply via
|
||||
handler's bridge. See <span className="mono">docs/local-models.md</span>.
|
||||
</div>
|
||||
<Card>
|
||||
<div className="card-head" style={{ marginBottom: 14 }}>
|
||||
@@ -594,6 +604,12 @@ function ModelsPanel() {
|
||||
onChange={(v) => setForm({ ...form, small_fast_model: v })}
|
||||
placeholder="qwen3-1.7b"
|
||||
/>
|
||||
<Select
|
||||
label="Harness (which agent binary runs against this endpoint)"
|
||||
value={form.harness}
|
||||
onChange={(v) => setForm({ ...form, harness: v as "claude" | "pi" })}
|
||||
options={HARNESS_OPTS}
|
||||
/>
|
||||
<Input
|
||||
label={editingId != null ? "API key (blank = keep stored key)" : "API key (optional)"}
|
||||
value={form.api_key}
|
||||
@@ -635,6 +651,7 @@ function ModelsPanel() {
|
||||
{m.name}
|
||||
</span>
|
||||
<div className="hstack">
|
||||
{m.harness === "pi" && <Badge tone="info">pi harness</Badge>}
|
||||
{m.has_api_key && <Badge tone="info">key stored</Badge>}
|
||||
<Badge tone={m.enabled ? "success" : "neutral"}>
|
||||
{m.enabled ? "enabled" : "disabled"}
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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<string, string> | null;
|
||||
enabled: boolean;
|
||||
has_api_key: boolean;
|
||||
|
||||
@@ -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,
|
||||
)
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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"
|
||||
|
||||
+118
-27
@@ -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
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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 <event>` — 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 <tool>` — 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<string, unknown>, 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, unknown>): 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<string, unknown>) {
|
||||
const text = callMemory(tool.name, params ?? {});
|
||||
return { content: [{ type: "text", text }], details: {} };
|
||||
},
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -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 <dir>/sessions/<uuid>.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)),
|
||||
]
|
||||
@@ -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"
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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()),
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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")
|
||||
|
||||
@@ -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 <tool>`` 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())
|
||||
|
||||
@@ -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")
|
||||
+3
-2
@@ -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
|
||||
|
||||
|
||||
+126
@@ -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 <bridge> --session <path>``), 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 <path>\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())
|
||||
@@ -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"
|
||||
Reference in New Issue
Block a user