diff --git a/cli-tool/components/mods/integrations/pi-agent-for-claude/.claude-plugin/plugin.json b/cli-tool/components/mods/integrations/pi-agent-for-claude/.claude-plugin/plugin.json new file mode 100644 index 00000000..973356c3 --- /dev/null +++ b/cli-tool/components/mods/integrations/pi-agent-for-claude/.claude-plugin/plugin.json @@ -0,0 +1,19 @@ +{ + "name": "pi-agent-for-claude", + "version": "0.1.0", + "description": "pi agent for claude: a `pi` subagent type whose model steps run the pi CLI, so pi subagents and agent-team teammates live in Claude Code's own agent UI, streamed live, on any model pi can reach.", + "author": { + "name": "Fazal Ali" + }, + "homepage": "https://github.com/FazalAAli/pi-agent-for-claude", + "repository": "https://github.com/FazalAAli/pi-agent-for-claude", + "license": "MIT", + "keywords": [ + "pi", + "subagent", + "agent-teams", + "multi-model", + "openrouter", + "function-hooks" + ] +} diff --git a/cli-tool/components/mods/integrations/pi-agent-for-claude/LICENSE b/cli-tool/components/mods/integrations/pi-agent-for-claude/LICENSE new file mode 100644 index 00000000..deb65246 --- /dev/null +++ b/cli-tool/components/mods/integrations/pi-agent-for-claude/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2026 Fazal Ali + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/cli-tool/components/mods/integrations/pi-agent-for-claude/README.md b/cli-tool/components/mods/integrations/pi-agent-for-claude/README.md new file mode 100644 index 00000000..f512b773 --- /dev/null +++ b/cli-tool/components/mods/integrations/pi-agent-for-claude/README.md @@ -0,0 +1,80 @@ +# pi-agent-for-claude + +> Vendored from [FazalAAli/pi-agent-for-claude](https://github.com/FazalAAli/pi-agent-for-claude) (MIT, by Fazal Ali). Install with `npx claude-code-templates@latest --mod integrations/pi-agent-for-claude`, or from the author's marketplace as described upstream. Report issues upstream. + +Runs [pi](https://pi.dev) as a native Claude Code subagent. It shows in the +agent list, streams live, takes follow-ups, and opens in a tmux pane as a +teammate, while the model behind it is anything pi can reach. + +``` +> Have pi run with model kimi k2 and summarize this repo. + +⏺ pi-agent-for-claude:pi(Summarize repo) + ⎿ ▸ bash: {"command":"ls -la && git log --oneline -10"} + ▸ bash: {"command":"cat package.json"} + This repo is an Expo/React Native calling app with a Go admin service... + + — answered by pi, openrouter/moonshotai/kimi-k2 +``` + +Claude Code starts a real subagent. The mod replaces only its model requests +(`turn.step`) with a detached `pi --mode json` run and streams pi's text, +thinking and tool calls. Hook calls get 10 s, so long runs span several steps +chained through a no-op `pi_progress` tool. pi's final answer goes back as +text, `SubagentHandback`, or `SendMessage` for teammates, with pi's real token +usage. + +## Requirements + +- Claude Code 2.1.275+ (the first build that dispatches `agent.spawn` and `turn.step`) +- [pi](https://pi.dev), logged in to at least one provider +- Node.js, macOS or Linux +- tmux, for teammate panes + +## Usage + +Ask Claude for pi in plain language. It picks the `pi-agent-for-claude:pi` agent type. + +- **Spawn:** "Have pi review the error handling in src/api." +- **Pick a model:** "Have pi run with model kimi k2 and …". Claude adds a `pi-model: ` line, passed to `pi --model`. Without it pi uses its default. `pi --list-models ` lists models. +- **Follow up:** messages to a running pi agent continue its pi session and model. +- **Teammate pane:** name it. "Spawn a pi agent named piper to …" + +Answers end with `— answered by pi, `, read from pi's own +events. Trust it over what the model says about itself. + +## Security + +pi runs its own read/bash/edit/write tools, outside Claude Code's permission +prompts, tool policy and safety classifier. A pi agent can change files +without asking. Restrict pi in its own config if that matters. + +## Limitations + +- **No hooks, no pi.** With function hooks off, a "pi" agent runs as Claude Haiku and only replies that pi is not active. For teammates, the flag must be in settings `env`. +- **Relies on undocumented engine behaviour:** the 10 s hook budget, subagent transcript files under `~/.claude/projects/`, exact engine message text, and `ps` to detect teammates. +- **Cost figures** for pi agents are pi's real token counts priced as Claude tokens. pi's provider bills them; check its billing. +- **Teammate shutdown** requests go to pi as plain messages. Close the pane yourself. + +## Install + +```sh +npx claude-code-templates@latest --mod integrations/pi-agent-for-claude +``` + +Then add the flags to `~/.claude/settings.json`, since shell variables alone +don't reach teammate panes, and restart (inside tmux for teammate panes): + +```json +{ + "env": { + "CLAUDE_CODE_ENABLE_FUNCTION_HOOKS": "1", + "CLAUDE_CODE_EXPERIMENTAL_AGENT_TEAMS": "1" + }, + "teammateMode": "tmux" +} +``` + +It is written to `.claude/skills/pi-agent-for-claude/`, which Claude Code auto-loads as `pi-agent-for-claude@skills-dir` once the workspace is trusted. For one session with hot reload: `CLAUDE_CODE_ENABLE_FUNCTION_HOOKS=1 claude --plugin-dir .claude/skills/pi-agent-for-claude`. `claude plugin validate .claude/skills/pi-agent-for-claude` prints every event it hooks and every `$` call it makes. + +**Early access.** Mods need Claude Code 2.1.259+ with `CLAUDE_CODE_ENABLE_FUNCTION_HOOKS=1`, and this one needs 2.1.275+; the `$` API may change between releases. Typed against Anthropic's declarations: https://github.com/anthropics/claude-code/tree/main/mods diff --git a/cli-tool/components/mods/integrations/pi-agent-for-claude/agents/pi.md b/cli-tool/components/mods/integrations/pi-agent-for-claude/agents/pi.md new file mode 100644 index 00000000..277f5a7e --- /dev/null +++ b/cli-tool/components/mods/integrations/pi-agent-for-claude/agents/pi.md @@ -0,0 +1,16 @@ +--- +name: pi +description: Delegates to the pi CLI in an isolated pi session. Use for work you want done by pi's model and tools rather than Claude's - a second opinion from another provider, or a cheap bulk pass. Runs pi's own read/bash/edit/write tools, so it can change files. To choose pi's model, make the prompt's first line `pi-model: ` (e.g. `pi-model: openrouter/moonshotai/kimi-k2`); the pattern is anything `pi --model` accepts, `pi --list-models ` lists them, and the provider/id form avoids ambiguity. Without that line pi uses its own default model. Every genuine answer ends with a line `— answered by pi, `. An answer WITHOUT that line did not come from pi - the pi-agent-for-claude module is not running and a Claude model answered instead; do not present it as pi's work, and tell the user to enable function hooks (CLAUDE_CODE_ENABLE_FUNCTION_HOOKS=1 in the env block of ~/.claude/settings.json) and restart. +model: haiku +--- + +If you are reading this, you are NOT pi: the pi-agent-for-claude hooks module +did not load, so this agent fell back to a Claude model. (While the module +runs, these instructions never reach any model - pi gets only the task.) + +Do not attempt the task and do not use any tools. Reply with exactly this, +and nothing else: + +pi agent for claude is not active, so pi did not run this task. Enable +function hooks (set CLAUDE_CODE_ENABLE_FUNCTION_HOOKS=1 in the env block of +~/.claude/settings.json, so teammates get it too) and restart Claude Code. diff --git a/cli-tool/components/mods/integrations/pi-agent-for-claude/hooks/hooks.json b/cli-tool/components/mods/integrations/pi-agent-for-claude/hooks/hooks.json new file mode 100644 index 00000000..e2df4981 --- /dev/null +++ b/cli-tool/components/mods/integrations/pi-agent-for-claude/hooks/hooks.json @@ -0,0 +1,4 @@ +{ + "description": "Answers an Agent call on the pi subagent type from the pi CLI, recorded as the Agent tool's own result", + "modules": ["./register.ts"] +} diff --git a/cli-tool/components/mods/integrations/pi-agent-for-claude/hooks/register.ts b/cli-tool/components/mods/integrations/pi-agent-for-claude/hooks/register.ts new file mode 100644 index 00000000..b8f3eef6 --- /dev/null +++ b/cli-tool/components/mods/integrations/pi-agent-for-claude/hooks/register.ts @@ -0,0 +1,607 @@ +import type { EngineInterface, On, TurnStepChunk, TurnStepResult, TurnUsage } from 'claude-code' + +/** How often a running pi's event stream is read for new events. */ +const STREAM_POLL_MS = 150 + +/** + * How long one step streams pi before handing off to the next. The engine + * gives a hook call 10 s and, past it, finishes the step with the real model; + * pi runs detached, so a longer run spans several steps, chained through + * PROGRESS_TOOL. + */ +const STEP_BUDGET_MS = 7_000 + +/** + * The no-op tool a step ends on while pi is still running: its call makes the + * engine run another step, which carries on streaming the same pi run. + */ +const PROGRESS_TOOL = 'pi_progress' + +/** How long a follow-up step waits for its message to reach the transcript. */ +const FOLLOW_UP_WAIT_MS = 5_000 +const POLL_MS = 200 + +/** + * The tool a subagent delivers its final report through, where the session + * runs the hand-back contract (`CLAUDE_CODE_SENDMESSAGE_HANDBACK`). A report + * left as plain text is bounced back with `[handback-send-enforce]`. + */ +const HANDBACK_TOOL = 'SubagentHandback' + +/** The tool a teammate answers another agent with. */ +const SEND_TOOL = 'SendMessage' + +/** + * User messages the engine writes into a subagent's transcript itself. They + * are not the person's follow-ups, so pi is never sent them. + */ +const ENGINE_PREFIXES = [ + '', + '[handback-send-enforce]', + '[Your previous response had no visible output', +] + +/** + * Registers the `pi` subagent type. + * + * The spawn runs as it always does, so the engine starts a real subagent loop + * with its own id, transcript and entry in `$.agent.list()`; only the model + * request inside that loop is replaced, by a run of the pi CLI. Each loop gets + * a pi session named after its agent id, so a follow-up continues the same pi + * conversation. + * + * Which model pi runs and which tools it may use are pi's own settings, not + * this mod's: it shells out to `pi` and lets pi's config decide. + * + * @param on the engine's registrar + */ +export function register(on: On) { + const seeds = new Map() + const consumed = new Map() + // Loops whose last step delivered pi's answer through a tool call, with the + // text the step after it ends on. + const delivered = new Map() + // This process's teammate key when it runs as a pi teammate (`claude + // --agent-type pi-agent-for-claude:pi` in its own pane): then its main loop is pi's. + let teammate: string | undefined + // PROGRESS_TOOL's full name as the engine registered it, `mcp____…`. + let progressTool = '' + const lastAnswers = new Map() + const runs = new Map() + // The pi model each loop asked for with a `pi-model:` line; it holds for + // that loop's later messages until another such line names a new one. + const models = new Map() + + on('session.start', async ($, e, next) => { + const started = await next(e) + teammate = await teammateOf($) + ;({ tool: progressTool } = await $.tool.register({ + name: PROGRESS_TOOL, + description: 'Internal to pi subagents: marks a pi run still in progress. Never call it.', + })) + return started + }) + + on('tool.call', async ($, e, next) => { + if (e.tool !== progressTool) return next(e) + const key = e.agentId ?? teammate + if (key === undefined || !runs.has(key)) { + return { deny: `${PROGRESS_TOOL} is internal to pi subagents.` } + } + return { result: 'pi is still working.' } + }) + + on('agent.spawn', async ($, e, next) => { + const started = await next(e) + if (isPi(e.subagentType) && started.agentId) { + seeds.set(started.agentId, e.prompt) + } + return started + }) + + on('turn.step', async function* ($, e, next) { + // A pi subagent's loop, keyed by its id, or this process's main loop when + // it is a pi teammate; any other loop is left to the model. + const seed = e.agentId === undefined ? undefined : seeds.get(e.agentId) + const agentId = e.agentId === undefined ? teammate : seed === undefined ? undefined : e.agentId + if (agentId === undefined) { + return yield* next(e) + } + const deadline = (await $.clock.now()) + STEP_BUDGET_MS + + // The step after a delivery is that tool's result coming back: the answer + // is delivered, so the loop ends here without asking pi again. It ends on + // visible text, since an empty response is bounced back as one. + const done = delivered.get(agentId) + if (done !== undefined) { + delivered.delete(agentId) + return yield* respond(e, [ + { kind: 'text', index: 0, text: done }, + { kind: 'stop', stopReason: 'end_turn', usage: null }, + ], done, []) + } + + let run = runs.get(agentId) + if (run === undefined) { + const already = consumed.get(agentId) ?? 0 + const { prompt, count, handback, bounced, replyTo } = seed === undefined + ? { ...(await teammatePromptOf($, already)), handback: false, bounced: false } + : { ...(await promptOf($, agentId, seed, already)), replyTo: undefined } + consumed.set(agentId, count) + if (bounced || prompt === undefined) { + // A bounce is the engine refusing a plain-text report: hand back the + // answer pi already gave rather than asking it again. + const text = bounced + ? lastAnswers.get(agentId) ?? '' + : "pi agent: no new message reached this agent's transcript." + return yield* finish(e, agentId, text, text, 1, handback || bounced, undefined, [{ kind: 'text', index: 0, text }]) + } + const asked = modelOf(prompt) + if (asked.model) models.set(agentId, asked.model) + run = await start($, agentId, asked.prompt, count, handback, models.get(agentId)) + run.replyTo = replyTo + runs.set(agentId, run) + } + + const { shown, blocks, lastKind } = yield* pump($, run, deadline, next.signal) + if (!run.ended) { + const input = {} + return yield* respond(e, [ + { kind: 'tool', index: blocks, id: `toolu_pi_${agentId}_${run.id}_${run.steps++}`, name: progressTool }, + { kind: 'input', index: blocks, json: JSON.stringify(input) }, + { kind: 'stop', stopReason: 'tool_use', usage: takeUsage(run) }, + ], shown, [{ name: progressTool, input }]) + } + + runs.delete(agentId) + // The hand-back reminder can reach the transcript after the run's first + // step read it; checking again now saves a bounce. + const handback = seed !== undefined && (run.handback || (await transcriptOf($, agentId)).handback) + const answer = run.report.trim() || run.shown + // Named from pi's own events: a model asked what it is often answers + // wrongly, and the parent has nothing else to check it by. + const report = run.model ? `${answer}\n\n— answered by pi, ${run.model}` : answer + // A run spanning several steps streamed its answer across them; the last + // step carries the whole of it, since that step is what the parent reads. + // The parent reads the last text block, so it must hold the whole answer. + // A run this step only finished gets the answer again as its own block; one + // whose answer streamed whole here gets just the model line, onto that + // same block, not alone in a new one. + const whole = run.steps > 0 && !shown.includes(answer) + const extra = whole ? report : report.slice(answer.length) + const into = !whole && lastKind === 'text' ? blocks - 1 : blocks + const tail: TurnStepChunk[] = extra ? [{ kind: 'text', index: into, text: extra }] : [] + return yield* finish(e, agentId, shown + extra, report, into + (extra ? 1 : 0), handback, run.replyTo, tail, takeUsage(run)) + }) + + /** + * Ends a step on pi's answer: plain text, or text then the tool call that + * delivers it - HANDBACK_TOOL where the session runs the hand-back contract, + * or, in a pi teammate, SendMessage to whoever sent the message pi answered. + */ + async function* finish( + e: { turnId: string; index: number }, + agentId: string, + shown: string, + report: string, + blocks: number, + handback: boolean, + replyTo: string | undefined, + chunks: TurnStepChunk[], + usage: TurnUsage | null = null, + ) { + lastAnswers.set(agentId, report) + const delivery = handback + ? { name: HANDBACK_TOOL, input: { message: report }, done: 'Report delivered.' } + : replyTo !== undefined + ? { name: SEND_TOOL, input: { to: replyTo, message: report }, done: `Sent to ${replyTo}.` } + : undefined + if (delivery === undefined) { + return yield* respond(e, [...chunks, { kind: 'stop', stopReason: 'end_turn', usage }], shown, []) + } + delivered.set(agentId, delivery.done) + const id = `toolu_pi_${agentId.replace(/[^A-Za-z0-9_]/g, '_')}_${delivery.name}_${consumed.get(agentId) ?? 0}` + return yield* respond(e, [ + ...chunks, + { kind: 'tool', index: blocks, id, name: delivery.name }, + { kind: 'input', index: blocks, json: JSON.stringify(delivery.input) }, + { kind: 'stop', stopReason: 'tool_use', usage }, + ], shown, [{ name: delivery.name, input: delivery.input }]) + } +} + +/** + * The tokens a run used since the last step reported them, as this step's + * usage under the model pi named; null before pi named one. Resets the count. + */ +export function takeUsage(run: Run | undefined): TurnUsage | null { + if (!run?.model) return null + const { i, o, cr, cw } = run.usage + run.usage = { i: 0, o: 0, cr: 0, cw: 0 } + return { + model: run.model, + input_tokens: i, + output_tokens: o, + cache_read_input_tokens: cr, + cache_creation_input_tokens: cw, + } +} + +/** + * This process's teammate key, when it is a pi teammate: a teammate runs as + * its own `claude --agent-id --agent-type ` process, and a hooks + * module has no API for its process's arguments, so this reads them with `ps`. + * The key names the teammate's pi session; undefined in any other process. + */ +export async function teammateOf($: EngineInterface) { + const { stdout } = await $.process.run(['sh', '-c', 'ps -o command= -p "$PPID"']) + const type = stdout.match(/--agent-type\s+(\S+)/)?.[1] + const id = stdout.match(/--agent-id\s+(\S+)/)?.[1] + if (type === undefined || id === undefined || !isPi(type)) return undefined + return id.replace(/[^A-Za-z0-9_-]/g, '-') +} + +/** + * Waits for a message past the first `already` in a pi teammate's own + * transcript - its main loop, so `$.session.messages()` reads it - and returns + * the newest as pi's prompt, with who sent it when another agent did. + * + * A message from another agent arrives wrapped as `…`; one typed into the pane is + * plain, and is answered in the pane rather than sent anywhere. + */ +export async function teammatePromptOf($: EngineInterface, already: number) { + for (let waited = 0; ; waited += POLL_MS) { + const texts = (await $.session.messages()) + .filter((m) => m.role === 'user') + .map((m) => m.text.trim()) + .filter((text) => text && !ENGINE_PREFIXES.some((prefix) => text.startsWith(prefix))) + if (texts.length > already) { + const text = texts.at(-1) as string + const replyTo = text.match(/]*>/g, 'Message from $1:\n') + .replace(/<\/teammate-message>/g, '') + .trim() + return { prompt, count: texts.length, replyTo } + } + if (waited >= FOLLOW_UP_WAIT_MS) return { prompt: undefined, count: already, replyTo: undefined } + await $.clock.sleep(POLL_MS) + } +} + +/** + * Takes a `pi-model: ` line out of a prompt: the way a caller picks + * the model pi runs, since the Agent tool's own `model` names Claude models + * only. The pattern is anything `pi --model` takes (`openrouter/`, a + * fuzzy `*sonnet*`); only the first such line counts. + */ +export function modelOf(text: string): { prompt: string; model?: string } { + const line = text.match(/^[ \t]*pi-model:[ \t]*([A-Za-z0-9_.:\/*~@+-]+)[ \t]*$/im) + if (!line) return { prompt: text } + return { prompt: text.replace(line[0], '').trim(), model: line[1] } +} + +/** Whether a spawn names this plugin's type, plain or plugin-qualified. */ +export function isPi(subagentType: string) { + return subagentType === 'pi' || subagentType.endsWith(':pi') +} + +/** + * Yields a step's chunks and returns the result the engine expects of it. + * + * @param e the step being answered + * @param chunks what the person watches stream, in order + * @param answer the step's visible text + * @param toolUses the tool calls the chunks made + */ +export async function* respond( + e: { turnId: string; index: number }, + chunks: TurnStepChunk[], + answer: string, + toolUses: TurnStepResult['toolUses'], +): AsyncGenerator { + for (const chunk of chunks) yield chunk + const stop = chunks.at(-1) + const stopReason = stop?.kind === 'stop' ? stop.stopReason : 'end_turn' + const usage = stop?.kind === 'stop' ? stop.usage : null + return { turnId: e.turnId, index: e.index, answer, toolUses, stopReason, usage } +} + +/** + * Rewrites pi's `--mode json` event stream as one small JSON line per thing a + * step shows: `{k:"msg"}` when an assistant message starts, `{k:"text"| + * "thinking", t}` per delta, `{k:"tool", t}` per tool pi starts, `{k:"usage", + * i, o, cr, cw}` per finished assistant message, `{k:"end"}`. `msg` carries + * the `provider/model` pi sent the request to. + * + * pi's own events carry whole tool outputs and transcripts, and + * `$.process.run` cuts a child's stdout at its output limit, so the step + * never reads them raw. + */ +const FILTER = String.raw` +const rl = require('readline').createInterface({ input: process.stdin }) +const out = (o) => process.stdout.write(JSON.stringify(o) + '\n') +rl.on('line', (line) => { + let e + try { e = JSON.parse(line) } catch { return } + const d = e.type === 'message_update' ? e.assistantMessageEvent : undefined + if (e.type === 'message_start' && e.message?.role === 'assistant') out({ k: 'msg', model: e.message.provider + '/' + e.message.model }) + else if (e.type === 'message_end' && e.message?.role === 'assistant') { + const u = e.message.usage ?? {} + out({ k: 'usage', i: u.input ?? 0, o: u.output ?? 0, cr: u.cacheRead ?? 0, cw: u.cacheWrite ?? 0 }) + } + else if (d?.type === 'text_delta') out({ k: 'text', t: d.delta }) + else if (d?.type === 'thinking_delta') out({ k: 'thinking', t: d.delta }) + else if (e.type === 'tool_execution_start') { + const a = JSON.stringify(e.args ?? {}) + out({ k: 'tool', t: '\n▸ ' + e.toolName + ': ' + (a.length > 100 ? a.slice(0, 100) + '…' : a) + '\n' }) + } else if (e.type === 'agent_end') out({ k: 'end' }) +}) +` + +/** One pi run, as it streams across the steps it spans. */ +export type Run = { + /** The detached pipeline's pid. */ + pid: string + /** pi's session, `pi-`, which also names its process for a kill. */ + session: string + /** The file FILTER writes pi's events to. */ + events: string + /** This run's number for the agent, which keeps tool call ids unique. */ + id: number + /** How many event lines have been read. */ + read: number + /** Every piece of text shown so far, across steps. */ + shown: string + /** The text of pi's latest assistant message: its answer once it ends. */ + report: string + /** Whether pi has ended. */ + ended: boolean + /** Whether the answer must be handed back through HANDBACK_TOOL. */ + handback: boolean + /** In a pi teammate, who sent the message pi is answering. */ + replyTo?: string + /** How many steps have handed off to the next through PROGRESS_TOOL. */ + steps: number + /** The `provider/model` pi reported for its latest message. */ + model: string + /** Tokens pi used since the last step reported them. */ + usage: { i: number; o: number; cr: number; cw: number } +} + +/** + * Starts pi detached in `agentId`'s pi session, piped through FILTER into a + * file the steps read: `$.process.run` only returns once its child exits, so + * it cannot stream a child it waits on. + * + * @param prompt what to send pi + * @param id this run's number for the agent + * @param handback whether pi's answer must be handed back + * @param model a pi `--model` pattern; absent, pi's own default + */ +export async function start( + $: EngineInterface, + agentId: string, + prompt: string, + id: number, + handback: boolean, + model: string | undefined, +): Promise { + const session = `pi-${agentId}` + const events = `/tmp/pi-agent-for-claude-${agentId}-${id}.jsonl` + const cwd = await $.session.cwd() + const started = await $.process.run([ + 'sh', + '-c', + 'nohup sh -c \'pi -p --mode json --session-id "$1" $5 -- "$2" "$3.err" | node -e "$4" >"$3"\' sh "$1" "$2" "$3" "$4" "$5" >/dev/null 2>&1 & echo $!', + 'sh', + session, + prompt, + events, + FILTER, + // Unquoted in the script so an absent model adds no argument; a pattern + // is one word (`openrouter/moonshotai/kimi-k2`), which modelOf enforces. + model ? `--model ${model}` : '', + ], { cwd }) + return { + pid: started.stdout.trim(), + session, + events, + id, + read: 0, + shown: '', + report: '', + ended: false, + handback, + steps: 0, + model: '', + usage: { i: 0, o: 0, cr: 0, cw: 0 }, + } +} + +/** + * Streams a pi run's new output as this step's chunks, until pi ends or the + * step's `deadline` passes: pi's text as text, its thinking as thinking, and + * each tool it starts as a one-line `▸ tool: args` note in the text. + * + * Never throws: a throw here would hand the step to the model beneath, a + * Claude subagent with Claude's tools, so a failure ends the run as text. An + * aborted step kills pi, since a detached pi would outlive it. + * + * ponytail: no run time limit; pi runs until it ends or a step is aborted. A + * subagent stopped between two steps leaves its pi to finish on its own. + * + * @param deadline when this step must stop streaming, in `$.clock` ms + * @param signal the step's abort signal + * @returns the text this step showed, how many content blocks it used, and + * the kind of the last one + */ +export async function* pump( + $: EngineInterface, + run: Run, + deadline: number, + signal: AbortSignal, +): AsyncGenerator { + let shown = '' + let index = -1 + let kind: 'text' | 'thinking' | undefined + let failure = '' + let finished = false + + try { + while (!run.ended && !signal.aborted && (await $.clock.now()) < deadline) { + const { stdout } = await $.process.run([ + 'sh', + '-c', + // Liveness first: once pi is gone the tail after it has every event. + 'kill -0 "$3" 2>/dev/null; alive=$?; tail -n +"$2" "$1" 2>/dev/null; [ $alive = 0 ] || printf "\\n{\\"k\\":\\"exited\\"}\\n"', + 'sh', + run.events, + String(run.read + 1), + run.pid, + ]) + const lines = stdout.split('\n') + lines.pop() // the line still being written, or '' after a complete one + for (const line of lines) { + run.read += 1 + let event: { k: string; t?: string; model?: string; i?: number; o?: number; cr?: number; cw?: number } + try { + event = JSON.parse(line) + } catch { + continue + } + if (event.k === 'end' || event.k === 'exited') { + run.ended = true + continue + } + if (event.k === 'msg') { + run.report = '' + if (event.model) run.model = event.model + continue + } + if (event.k === 'usage') { + run.usage.i += event.i ?? 0 + run.usage.o += event.o ?? 0 + run.usage.cr += event.cr ?? 0 + run.usage.cw += event.cw ?? 0 + continue + } + const pieceKind = event.k === 'thinking' ? 'thinking' : 'text' + const text = event.t ?? '' + if (pieceKind !== kind) { + kind = pieceKind + index += 1 + } + if (event.k === 'text') run.report += text + if (pieceKind === 'text') shown += text + yield { kind: pieceKind, index, text } + } + if (!run.ended) await $.clock.sleep(STREAM_POLL_MS) + } + finished = true + } catch (err) { + failure = `pi agent: the pi run failed: ${String(err)}` + finished = true + } finally { + // Closed by the engine, or aborted: pi must not outlive the subagent. + if (!run.ended && (!finished || signal.aborted)) { + run.ended = true + await $.process.run(['sh', '-c', 'pkill -f -- "--session-id $1 "; rm -f "$2" "$2.err"', 'sh', run.session, run.events]) + .catch(() => undefined) + } + } + if (failure) run.ended = true + + if (run.ended) { + if (!failure && !(run.shown + shown).trim()) { + const { stdout: err } = await $.process.run(['sh', '-c', 'cat "$1.err" 2>/dev/null', 'sh', run.events]) + .catch(() => ({ stdout: '' })) + // pi colours its errors; the escapes would show as noise in a transcript. + const plain = err.replace(/\u001b\[[0-9;]*m/g, '').trim() + failure = `pi ended with no answer.\n${plain || '(no stderr)'}` + } + if (failure) { + shown += failure + run.report = failure + index += 1 + yield { kind: 'text', index, text: failure } + } + await $.process.run(['rm', '-f', run.events, `${run.events}.err`]).catch(() => undefined) + } + run.shown += shown + return { shown, blocks: index + 1, lastKind: kind } +} + +/** + * Decides what to send pi this step, and whether its answer must be handed + * back through HANDBACK_TOOL. + * + * The first step sends the spawn's prompt. A later one waits for a person's + * message past the first `already` in `agentId`'s own transcript and sends the + * newest. A follow-up (SendMessage, or typed into the agent's view) is only + * found in `subagents/agent-.jsonl` - `$.session.messages()` reads the main + * loop's - and the engine can start the step before it writes the message + * there, so this polls briefly. + * + * ponytail: finds the file by name under ~/.claude/projects; the id is unique, + * but it leans on Claude Code's on-disk layout, not an API. Polls for up to + * FOLLOW_UP_WAIT_MS; raise it if a slow disk drops follow-ups. + * + * @param seed the prompt the spawn was given + * @param already how many of the person's messages pi has been sent + */ +export async function promptOf($: EngineInterface, agentId: string, seed: string, already: number) { + for (let waited = 0; ; waited += POLL_MS) { + const { texts, handback, bounced } = await transcriptOf($, agentId) + if (already === 0) return { prompt: seed, count: Math.max(texts.length, 1), handback, bounced: false } + if (texts.length > already) return { prompt: texts.at(-1), count: texts.length, handback, bounced: false } + if (bounced) return { prompt: undefined, count: already, handback: true, bounced } + if (waited >= FOLLOW_UP_WAIT_MS) return { prompt: undefined, count: already, handback, bounced } + await $.clock.sleep(POLL_MS) + } +} + +/** + * Reads `agentId`'s transcript: the text of every message the person (or the + * spawn) sent it, oldest first; whether the engine asked it to hand its + * report back through HANDBACK_TOOL; and whether the newest message is the + * engine bouncing a report that was not. + */ +export async function transcriptOf($: EngineInterface, agentId: string) { + const { stdout } = await $.process.run([ + 'sh', + '-c', + 'f=$(find "$HOME/.claude/projects" -name "agent-$1.jsonl" -print -quit) && [ -n "$f" ] && cat "$f"', + 'sh', + agentId, + ]) + const all = stdout + .split('\n') + .flatMap((line) => { + try { + return [JSON.parse(line)] + } catch { + return [] // the line Claude Code is still writing + } + }) + .filter((entry) => entry.type === 'user') + .map((entry) => textOf(entry.message?.content)) + .filter(Boolean) + return { + texts: all.filter((text) => !ENGINE_PREFIXES.some((prefix) => text.startsWith(prefix))), + handback: all.some((text) => text.includes(HANDBACK_TOOL)), + bounced: all.at(-1)?.startsWith('[handback-send-enforce]') ?? false, + } +} + +/** Joins a message's text: a plain string, or the text blocks of a block list. */ +export function textOf(content: unknown): string { + if (typeof content === 'string') return content.trim() + if (!Array.isArray(content)) return '' + return content + .filter((block) => block?.type === 'text') + .map((block) => block.text) + .join('\n') + .trim() +}