#!/usr/bin/env npx tsx
/**
* snappy-dispatch/api.ts — Fire-and-return sub-agent dispatch.
*
* Orchestrator (Claude Sonnet / Opus) shells out here to get work done by a
* cheaper model. The callee runs `pi` (@mariozechner/pi-coding-agent) with a
* chosen provider/model, captures stdout, logs cost+latency to ndjson, returns.
*
* Usage (TS):
* import { dispatch } from "./api.ts";
* const out = await dispatch({
* prompt: "count files in /tmp",
* tools: ["bash", "ls"],
* model: "haiku", // or "gemini", "sonnet", or a full "provider/model-id"
* });
*
* Usage (CLI):
* npx tsx api.ts "count files in /tmp"
* npx tsx api.ts --model gemini --tools bash "..."
*
* Defaults: haiku-4-5, tools=read,bash,grep,ls, timeout=120s, ephemeral session.
*/
import { spawnSync } from "child_process";
import { appendFileSync, mkdirSync, realpathSync } from "fs";
import { dirname } from "path";
import { env } from "../snappy-settings/load.ts";
import { defaultProvider } from "../snappy-settings/providers-choice.ts";
import { refusalTable } from "../snappy-settings/refusal-codes.ts";
// Backend: "pi" → pi-coding-agent + OpenRouter/Anthropic API (per-token billing).
// "claude-code" → local `claude -p` CLI, authenticated against the
// Claude Code subscription on this machine (no per-token cost,
// counts against the subscription quota).
export type DispatchBackend = "pi" | "claude-code";
const LOG_PATH = `${process.env.HOME}/.claude/logs/dispatches.ndjson`;
// Shortcut aliases → (provider, full model id).
// Add new ones here as we validate them.
const ALIASES: Record<string, { provider: string; model: string; keyEnv: string }> = {
haiku: { provider: "anthropic", model: "claude-haiku-4-5", keyEnv: "ANTHROPIC_API_KEY" },
sonnet: { provider: "anthropic", model: "claude-sonnet-4-6", keyEnv: "ANTHROPIC_API_KEY" },
gemini: { provider: "openrouter", model: "google/gemini-2.5-flash", keyEnv: "OPENROUTER_API_KEY" },
"gemini-pro": { provider: "openrouter", model: "google/gemini-2.5-pro", keyEnv: "OPENROUTER_API_KEY" },
llama: { provider: "openrouter", model: "meta-llama/llama-3.3-70b-instruct", keyEnv: "OPENROUTER_API_KEY" },
qwen: { provider: "openrouter", model: "qwen/qwen-2.5-72b-instruct", keyEnv: "OPENROUTER_API_KEY" },
deepseek: { provider: "openrouter", model: "deepseek/deepseek-chat", keyEnv: "OPENROUTER_API_KEY" },
// claude-code: marker alias for the local `claude -p` CLI backend.
// Resolved specially in dispatch() — does not route through pi.
"claude-code": { provider: "claude-code", model: "sonnet", keyEnv: "" },
cc: { provider: "claude-code", model: "sonnet", keyEnv: "" },
};
export interface DispatchOptions {
prompt: string;
model?: string; // alias ("haiku") or full "provider/id"
tools?: string[]; // default ["read","bash","grep","ls"]
timeoutMs?: number; // default 120_000
systemPrompt?: string; // optional override
cwd?: string; // working dir for the dispatched agent
}
export interface DispatchResult {
ok: boolean;
output: string; // final stdout (model's textual answer)
stderr: string;
durationMs: number;
provider: string;
model: string;
exitCode: number;
error?: string;
// Optional usage fields. Populated by pi `--mode json` parser when available;
// populated for claude-code backend when `~/.config/claude/session-usage.json`
// is present or `ccusage` is on PATH. Undefined when neither path produces a
// value — never fail the dispatch over missing usage data.
inputTokens?: number;
outputTokens?: number;
cacheReadTokens?: number;
cacheWriteTokens?: number;
costUsd?: number;
}
/**
* ── WHAT AN UNASKED DISPATCH SPENDS: HIS DEFAULT, WHERE THERE IS A CHEAP ROAD ─
*
* ⟨owner, 2026-09-09 10:25⟩ he picks a Default provider on the bar's Providers
* tab. This hand picks a MODEL, not a provider, so it honours that choice the
* only way that keeps its own contract: the CHEAPEST alias of the provider he
* named, and only when the key for it is actually held. There is no effort here
* — pi takes no reasoning-effort argument, so nothing of his level is dropped.
*
* `chatgpt`, `openai-api` and `gemini` map to NOTHING: this hand has no OpenAI
* or Google alias at all (see ALIASES above), and inventing one would spend a
* provider on a model nobody validated for grunt work. They keep `gemini`
* (OpenRouter Flash), which is what an unasked dispatch has always spent.
*/
const DISPATCH_ALIAS_FOR_CHOICE: Record<string, string> = {
openrouter: "gemini",
claude: "haiku",
};
/** The alias an unasked dispatch spends. Exported so a test can read it. */
export function defaultAlias(): string {
const alias = DISPATCH_ALIAS_FOR_CHOICE[defaultProvider() ?? ""];
if (alias === undefined) return "gemini";
const keyEnv = ALIASES[alias]?.keyEnv;
// A provider whose key this machine does not hold is not a default, it is a
// failed run: fall back to the shipping alias rather than refuse the errand.
// `env(key, false)` — a missing key here is a fact to branch on, not a throw.
return keyEnv && env(keyEnv, false) ? alias : "gemini";
}
function resolveModel(model: string | undefined): { provider: string; model: string; keyEnv: string } {
const m = (model || defaultAlias()).trim();
if (ALIASES[m]) return ALIASES[m];
// Allow full "provider/model-id" form
if (m.includes("/")) {
const [prefix, ...rest] = m.split("/");
const tail = rest.join("/");
// Explicit `openrouter/<anything>` forces routing through openrouter.
// Use this when the native key is exhausted (e.g. OpenAI quota) but
// we still have openrouter credit.
if (prefix === "openrouter") return { provider: "openrouter", model: tail, keyEnv: "OPENROUTER_API_KEY" };
// Native routing for first-party providers when we have direct keys.
if (prefix === "anthropic") return { provider: "anthropic", model: tail, keyEnv: "ANTHROPIC_API_KEY" };
if (prefix === "openai") return { provider: "openai", model: tail, keyEnv: "OPENAI_API_KEY" };
if (prefix === "google") return { provider: "google", model: tail, keyEnv: "GEMINI_API_KEY" };
// Everything else rides openrouter with the full `provider/model` as the id.
return { provider: "openrouter", model: m, keyEnv: "OPENROUTER_API_KEY" };
}
throw new Error(`Unknown model alias "${m}". Known: ${Object.keys(ALIASES).join(", ")} or "provider/model-id".`);
}
// Known transient failures from pi + OpenRouter SSE streaming. Seen in the
// wild (2026-04-14): SSE stream occasionally drops a JSON frame under load
// when skill context is loaded. The run is recoverable on retry.
const TRANSIENT_MARKERS = [
"JSON error injected into SSE stream",
"Provider finish_reason: error",
"fetch failed",
"socket hang up",
"Network connection lost",
"ECONNRESET",
"ETIMEDOUT",
];
function isTransient(output: string, stderr: string, exitCode: number): boolean {
if (exitCode === 0) return false;
const hay = `${output}\n${stderr}`;
return TRANSIENT_MARKERS.some(m => hay.includes(m));
}
// pi `--mode json` emits a stream of NDJSON frames. The final `agent_end` frame
// holds the full message list with the final `usage` object populated. Parse
// the stream, return clean assistant text + usage. Returns nulls when parsing
// fails so the caller can fall back to raw stdout.
type PiUsage = {
inputTokens?: number;
outputTokens?: number;
cacheReadTokens?: number;
cacheWriteTokens?: number;
costUsd?: number;
};
function parsePiJsonStream(stdout: string): { text: string | null; usage: PiUsage } {
const usage: PiUsage = {};
let text: string | null = null;
let lastAgentEnd: any = null;
for (const line of stdout.split("\n")) {
const trimmed = line.trim();
if (!trimmed.startsWith("{")) continue;
try {
const frame = JSON.parse(trimmed);
if (frame?.type === "agent_end") lastAgentEnd = frame;
} catch { /* non-frame line (warning, error message) — skip */ }
}
if (!lastAgentEnd?.messages?.length) return { text: null, usage };
// Walk messages backwards to find the final assistant message
const messages = lastAgentEnd.messages as any[];
for (let i = messages.length - 1; i >= 0; i--) {
const msg = messages[i];
if (msg?.role !== "assistant") continue;
const blocks = msg.content || [];
text = blocks
.filter((b: any) => b?.type === "text" && typeof b.text === "string")
.map((b: any) => b.text)
.join("");
const u = msg.usage;
if (u && typeof u === "object") {
if (typeof u.input === "number") usage.inputTokens = u.input;
if (typeof u.output === "number") usage.outputTokens = u.output;
if (typeof u.cacheRead === "number") usage.cacheReadTokens = u.cacheRead;
if (typeof u.cacheWrite === "number") usage.cacheWriteTokens = u.cacheWrite;
if (u.cost && typeof u.cost.total === "number") usage.costUsd = u.cost.total;
}
break;
}
return { text, usage };
}
// claude-code backend has no per-call API for usage. Best-effort: shell out to
// `ccusage` if present (community OSS tool that reads ~/.claude session db),
// else read `~/.config/claude/session-usage.json` if it exists. Return undefined
// fields when neither source is available — usage stays optional.
function getClaudeCodeUsage(): PiUsage {
const usage: PiUsage = {};
// Prefer ccusage if installed
const ccusage = spawnSync("ccusage", ["--latest", "--json"], {
encoding: "utf-8",
timeout: 5_000,
});
if (!ccusage.error && ccusage.status === 0 && ccusage.stdout) {
try {
const j = JSON.parse(ccusage.stdout);
if (typeof j?.input_tokens === "number") usage.inputTokens = j.input_tokens;
if (typeof j?.output_tokens === "number") usage.outputTokens = j.output_tokens;
if (typeof j?.cache_read_input_tokens === "number") usage.cacheReadTokens = j.cache_read_input_tokens;
if (typeof j?.cache_creation_input_tokens === "number") usage.cacheWriteTokens = j.cache_creation_input_tokens;
if (typeof j?.total_cost_usd === "number") usage.costUsd = j.total_cost_usd;
return usage;
} catch { /* fall through */ }
}
// Fallback: ~/.config/claude/session-usage.json (older convention)
try {
const path = `${process.env.HOME}/.config/claude/session-usage.json`;
const raw = require("fs").readFileSync(path, "utf-8");
const j = JSON.parse(raw);
if (typeof j?.input_tokens === "number") usage.inputTokens = j.input_tokens;
if (typeof j?.output_tokens === "number") usage.outputTokens = j.output_tokens;
if (typeof j?.cost_usd === "number") usage.costUsd = j.cost_usd;
} catch { /* file missing — usage stays empty, never fail dispatch */ }
return usage;
}
async function dispatchClaudeCode(opts: DispatchOptions, resolvedModel: string): Promise<DispatchResult> {
const started = Date.now();
const timeoutMs = opts.timeoutMs ?? 120_000;
const args = ["-p", "--model", resolvedModel || "sonnet"];
if (opts.systemPrompt) args.push("--append-system-prompt", opts.systemPrompt);
const r = spawnSync("claude", args, {
input: opts.prompt,
encoding: "utf-8",
cwd: opts.cwd,
env: { ...process.env },
timeout: timeoutMs,
maxBuffer: 10 * 1024 * 1024,
});
const durationMs = Date.now() - started;
const output = (r.stdout || "").trim();
const stderr = (r.stderr || "").trim();
const exitCode = r.status ?? -1;
const timedOut = r.signal === "SIGTERM";
const ok = !r.error && !timedOut && exitCode === 0 && output.length > 0;
const ccUsage = getClaudeCodeUsage();
const result: DispatchResult = {
ok,
output,
stderr,
durationMs,
provider: "claude-code",
model: resolvedModel || "sonnet",
exitCode,
error: r.error?.message || (timedOut ? "timeout" : undefined),
...ccUsage,
};
try {
mkdirSync(dirname(LOG_PATH), { recursive: true });
const { output: _fullOutput, ...logMeta } = result; // full output stays in the return value, not the log
appendFileSync(LOG_PATH, JSON.stringify({
ts: new Date().toISOString(),
promptPreview: opts.prompt.slice(0, 200),
backend: "claude-code",
...logMeta,
outputPreview: output.slice(0, 500),
}) + "\n");
} catch { /* */ }
return result;
}
export async function dispatch(opts: DispatchOptions): Promise<DispatchResult> {
const started = Date.now();
const { provider, model, keyEnv } = resolveModel(opts.model);
// Route to the claude-code CLI backend instead of pi when asked.
if (provider === "claude-code") {
return dispatchClaudeCode(opts, model);
}
const tools = (opts.tools && opts.tools.length ? opts.tools : ["read", "bash", "grep", "ls"]).join(",");
const timeoutMs = opts.timeoutMs ?? 120_000;
// Export the right API key for pi. pi auto-reads env vars by provider.
const apiKey = env(keyEnv);
const spawnEnv: NodeJS.ProcessEnv = {
...process.env,
[keyEnv]: apiKey,
};
const args = [
"--print",
"--no-session",
"--mode", "json",
"--provider", provider,
"--model", model,
"--tools", tools,
];
if (opts.systemPrompt) {
args.push("--append-system-prompt", opts.systemPrompt);
}
args.push(opts.prompt);
const runOnce = () => spawnSync("pi", args, {
encoding: "utf-8",
cwd: opts.cwd,
env: spawnEnv,
timeout: timeoutMs,
maxBuffer: 10 * 1024 * 1024,
});
// Up to 2 retries on transient SSE/network errors. Kernel-loaded dispatches
// ship a bigger context and hit OpenRouter SSE flakes more often; silent
// retries keep callers from having to know.
let r = runOnce();
let attempts = 1;
while (attempts < 3 && isTransient((r.stdout || "").trim(), (r.stderr || "").trim(), r.status ?? -1)) {
attempts++;
r = runOnce();
}
const durationMs = Date.now() - started;
const rawStdout = (r.stdout || "").trim();
const stderr = (r.stderr || "").trim();
const exitCode = r.status ?? -1;
const timedOut = r.signal === "SIGTERM";
// Parse pi's --mode json stream. Falls back to raw stdout if parsing fails
// (malformed frames, network error before any frame, etc).
const parsed = parsePiJsonStream(rawStdout);
const output = parsed.text != null ? parsed.text.trim() : rawStdout;
const ok = !r.error && !timedOut && exitCode === 0 && output.length > 0;
const result: DispatchResult = {
ok,
output,
stderr,
durationMs,
provider,
model,
exitCode,
error: r.error?.message || (timedOut ? "timeout" : undefined),
...parsed.usage,
};
// ndjson log — one line per dispatch, zero coupling to any log sink
try {
mkdirSync(dirname(LOG_PATH), { recursive: true });
const { output: _fullOutput, ...logMeta } = result; // full output stays in the return value, not the log
appendFileSync(LOG_PATH, JSON.stringify({
ts: new Date().toISOString(),
promptPreview: opts.prompt.slice(0, 200),
tools,
backend: "pi",
...logMeta,
outputPreview: output.slice(0, 500),
}) + "\n");
} catch { /* non-fatal — logging must never break dispatch */ }
return result;
}
// --- CLI ---
/** WHAT THIS HAND ANSWERS, and what each verb does to the world.
* Derived from this file's own CLI dispatch by
* `snappy-hands/contract-derive.ts` — a verb the code does not implement is
* never declared here. Snappy's daemon reads it (`api.ts contract`) to
* validate every call, build the argument words in order, decide whether the
* act runs now or stages for the owner, and hand the child exactly the
* environment keys named in `requires` — never a value, never anything else.
*/
export const HAND_CONTRACT = {
skill: "snappy-dispatch",
description: "The cheap-labor lever. Dispatch mechanical, verifiable work from the Sonnet/Opus orchestrator to cheap sub-agents (Haiku, Gemini, Llama, etc.) running locally via pi-coding-agent. Triggers on mentions of dispatch, cheap labor, sub-agent fan-out, pi-coding-agent, Haiku/Gemini/Llama grunt work.",
managed: false,
requires: [] as string[],
refusals: refusalTable("unknown_verb"),
verbs: {
},
} as const;
if (import.meta.url === `file://${realpathSync(process.argv[1])}` && process.argv[2] === "contract") {
console.log(JSON.stringify(HAND_CONTRACT, null, 2));
process.exit(0);
}
if (import.meta.url === `file://${realpathSync(process.argv[1])}`) {
const args = process.argv.slice(2);
let model: string | undefined;
let tools: string[] | undefined;
let timeoutMs: number | undefined;
const prompt: string[] = [];
for (let i = 0; i < args.length; i++) {
const a = args[i];
if (a === "--model") { model = args[++i]; continue; }
if (a === "--tools") { tools = args[++i].split(","); continue; }
if (a === "--timeout") { timeoutMs = Number(args[++i]) * 1000; continue; }
if (a === "--help" || a === "-h") {
console.log(`snappy-dispatch — sub-agent dispatch CLI
npx tsx api.ts [--model alias] [--tools csv] [--timeout seconds] "prompt..."
Model aliases: ${Object.keys(ALIASES).join(", ")}
or pass a full "provider/model-id" string.
Default: model=haiku, tools=read,bash,grep,ls, timeout=120`);
process.exit(0);
}
prompt.push(a);
}
if (prompt.length === 0) {
console.error("Usage: npx tsx api.ts [flags] \"prompt...\"");
process.exit(1);
}
dispatch({ prompt: prompt.join(" "), model, tools, timeoutMs })
.then(r => {
if (r.ok) {
process.stdout.write(r.output + "\n");
process.exit(0);
} else {
console.error(`[dispatch] failed: exit=${r.exitCode} err=${r.error || "-"}`);
if (r.stderr) console.error(r.stderr);
process.exit(r.exitCode || 1);
}
})
.catch(e => { console.error(e); process.exit(1); });
}
#!/usr/bin/env npx tsx
/**
* snappy-dispatch/api.ts — Fire-and-return sub-agent dispatch.
*
* Orchestrator (Claude Sonnet / Opus) shells out here to get work done by a
* cheaper model. The callee runs `pi` (@mariozechner/pi-coding-agent) with a
* chosen provider/model, captures stdout, logs cost+latency to ndjson, returns.
*
* Usage (TS):
* import { dispatch } from "./api.ts";
* const out = await dispatch({
* prompt: "count files in /tmp",
* tools: ["bash", "ls"],
* model: "haiku", // or "gemini", "sonnet", or a full "provider/model-id"
* });
*
* Usage (CLI):
* npx tsx api.ts "count files in /tmp"
* npx tsx api.ts --model gemini --tools bash "..."
*
* Defaults: haiku-4-5, tools=read,bash,grep,ls, timeout=120s, ephemeral session.
*/
import { spawnSync } from "child_process";
import { appendFileSync, mkdirSync, realpathSync } from "fs";
import { dirname } from "path";
import { env } from "../snappy-settings/load.ts";
import { defaultProvider } from "../snappy-settings/providers-choice.ts";
import { refusalTable } from "../snappy-settings/refusal-codes.ts";
// Backend: "pi" → pi-coding-agent + OpenRouter/Anthropic API (per-token billing).
// "claude-code" → local `claude -p` CLI, authenticated against the
// Claude Code subscription on this machine (no per-token cost,
// counts against the subscription quota).
export type DispatchBackend = "pi" | "claude-code";
const LOG_PATH = `${process.env.HOME}/.claude/logs/dispatches.ndjson`;
// Shortcut aliases → (provider, full model id).
// Add new ones here as we validate them.
const ALIASES: Record<string, { provider: string; model: string; keyEnv: string }> = {
haiku: { provider: "anthropic", model: "claude-haiku-4-5", keyEnv: "ANTHROPIC_API_KEY" },
sonnet: { provider: "anthropic", model: "claude-sonnet-4-6", keyEnv: "ANTHROPIC_API_KEY" },
gemini: { provider: "openrouter", model: "google/gemini-2.5-flash", keyEnv: "OPENROUTER_API_KEY" },
"gemini-pro": { provider: "openrouter", model: "google/gemini-2.5-pro", keyEnv: "OPENROUTER_API_KEY" },
llama: { provider: "openrouter", model: "meta-llama/llama-3.3-70b-instruct", keyEnv: "OPENROUTER_API_KEY" },
qwen: { provider: "openrouter", model: "qwen/qwen-2.5-72b-instruct", keyEnv: "OPENROUTER_API_KEY" },
deepseek: { provider: "openrouter", model: "deepseek/deepseek-chat", keyEnv: "OPENROUTER_API_KEY" },
// claude-code: marker alias for the local `claude -p` CLI backend.
// Resolved specially in dispatch() — does not route through pi.
"claude-code": { provider: "claude-code", model: "sonnet", keyEnv: "" },
cc: { provider: "claude-code", model: "sonnet", keyEnv: "" },
};
export interface DispatchOptions {
prompt: string;
model?: string; // alias ("haiku") or full "provider/id"
tools?: string[]; // default ["read","bash","grep","ls"]
timeoutMs?: number; // default 120_000
systemPrompt?: string; // optional override
cwd?: string; // working dir for the dispatched agent
}
export interface DispatchResult {
ok: boolean;
output: string; // final stdout (model's textual answer)
stderr: string;
durationMs: number;
provider: string;
model: string;
exitCode: number;
error?: string;
// Optional usage fields. Populated by pi `--mode json` parser when available;
// populated for claude-code backend when `~/.config/claude/session-usage.json`
// is present or `ccusage` is on PATH. Undefined when neither path produces a
// value — never fail the dispatch over missing usage data.
inputTokens?: number;
outputTokens?: number;
cacheReadTokens?: number;
cacheWriteTokens?: number;
costUsd?: number;
}
/**
* ── WHAT AN UNASKED DISPATCH SPENDS: HIS DEFAULT, WHERE THERE IS A CHEAP ROAD ─
*
* ⟨owner, 2026-09-09 10:25⟩ he picks a Default provider on the bar's Providers
* tab. This hand picks a MODEL, not a provider, so it honours that choice the
* only way that keeps its own contract: the CHEAPEST alias of the provider he
* named, and only when the key for it is actually held. There is no effort here
* — pi takes no reasoning-effort argument, so nothing of his level is dropped.
*
* `chatgpt`, `openai-api` and `gemini` map to NOTHING: this hand has no OpenAI
* or Google alias at all (see ALIASES above), and inventing one would spend a
* provider on a model nobody validated for grunt work. They keep `gemini`
* (OpenRouter Flash), which is what an unasked dispatch has always spent.
*/
const DISPATCH_ALIAS_FOR_CHOICE: Record<string, string> = {
openrouter: "gemini",
claude: "haiku",
};
/** The alias an unasked dispatch spends. Exported so a test can read it. */
export function defaultAlias(): string {
const alias = DISPATCH_ALIAS_FOR_CHOICE[defaultProvider() ?? ""];
if (alias === undefined) return "gemini";
const keyEnv = ALIASES[alias]?.keyEnv;
// A provider whose key this machine does not hold is not a default, it is a
// failed run: fall back to the shipping alias rather than refuse the errand.
// `env(key, false)` — a missing key here is a fact to branch on, not a throw.
return keyEnv && env(keyEnv, false) ? alias : "gemini";
}
function resolveModel(model: string | undefined): { provider: string; model: string; keyEnv: string } {
const m = (model || defaultAlias()).trim();
if (ALIASES[m]) return ALIASES[m];
// Allow full "provider/model-id" form
if (m.includes("/")) {
const [prefix, ...rest] = m.split("/");
const tail = rest.join("/");
// Explicit `openrouter/<anything>` forces routing through openrouter.
// Use this when the native key is exhausted (e.g. OpenAI quota) but
// we still have openrouter credit.
if (prefix === "openrouter") return { provider: "openrouter", model: tail, keyEnv: "OPENROUTER_API_KEY" };
// Native routing for first-party providers when we have direct keys.
if (prefix === "anthropic") return { provider: "anthropic", model: tail, keyEnv: "ANTHROPIC_API_KEY" };
if (prefix === "openai") return { provider: "openai", model: tail, keyEnv: "OPENAI_API_KEY" };
if (prefix === "google") return { provider: "google", model: tail, keyEnv: "GEMINI_API_KEY" };
// Everything else rides openrouter with the full `provider/model` as the id.
return { provider: "openrouter", model: m, keyEnv: "OPENROUTER_API_KEY" };
}
throw new Error(`Unknown model alias "${m}". Known: ${Object.keys(ALIASES).join(", ")} or "provider/model-id".`);
}
// Known transient failures from pi + OpenRouter SSE streaming. Seen in the
// wild (2026-04-14): SSE stream occasionally drops a JSON frame under load
// when skill context is loaded. The run is recoverable on retry.
const TRANSIENT_MARKERS = [
"JSON error injected into SSE stream",
"Provider finish_reason: error",
"fetch failed",
"socket hang up",
"Network connection lost",
"ECONNRESET",
"ETIMEDOUT",
];
function isTransient(output: string, stderr: string, exitCode: number): boolean {
if (exitCode === 0) return false;
const hay = `${output}\n${stderr}`;
return TRANSIENT_MARKERS.some(m => hay.includes(m));
}
// pi `--mode json` emits a stream of NDJSON frames. The final `agent_end` frame
// holds the full message list with the final `usage` object populated. Parse
// the stream, return clean assistant text + usage. Returns nulls when parsing
// fails so the caller can fall back to raw stdout.
type PiUsage = {
inputTokens?: number;
outputTokens?: number;
cacheReadTokens?: number;
cacheWriteTokens?: number;
costUsd?: number;
};
function parsePiJsonStream(stdout: string): { text: string | null; usage: PiUsage } {
const usage: PiUsage = {};
let text: string | null = null;
let lastAgentEnd: any = null;
for (const line of stdout.split("\n")) {
const trimmed = line.trim();
if (!trimmed.startsWith("{")) continue;
try {
const frame = JSON.parse(trimmed);
if (frame?.type === "agent_end") lastAgentEnd = frame;
} catch { /* non-frame line (warning, error message) — skip */ }
}
if (!lastAgentEnd?.messages?.length) return { text: null, usage };
// Walk messages backwards to find the final assistant message
const messages = lastAgentEnd.messages as any[];
for (let i = messages.length - 1; i >= 0; i--) {
const msg = messages[i];
if (msg?.role !== "assistant") continue;
const blocks = msg.content || [];
text = blocks
.filter((b: any) => b?.type === "text" && typeof b.text === "string")
.map((b: any) => b.text)
.join("");
const u = msg.usage;
if (u && typeof u === "object") {
if (typeof u.input === "number") usage.inputTokens = u.input;
if (typeof u.output === "number") usage.outputTokens = u.output;
if (typeof u.cacheRead === "number") usage.cacheReadTokens = u.cacheRead;
if (typeof u.cacheWrite === "number") usage.cacheWriteTokens = u.cacheWrite;
if (u.cost && typeof u.cost.total === "number") usage.costUsd = u.cost.total;
}
break;
}
return { text, usage };
}
// claude-code backend has no per-call API for usage. Best-effort: shell out to
// `ccusage` if present (community OSS tool that reads ~/.claude session db),
// else read `~/.config/claude/session-usage.json` if it exists. Return undefined
// fields when neither source is available — usage stays optional.
function getClaudeCodeUsage(): PiUsage {
const usage: PiUsage = {};
// Prefer ccusage if installed
const ccusage = spawnSync("ccusage", ["--latest", "--json"], {
encoding: "utf-8",
timeout: 5_000,
});
if (!ccusage.error && ccusage.status === 0 && ccusage.stdout) {
try {
const j = JSON.parse(ccusage.stdout);
if (typeof j?.input_tokens === "number") usage.inputTokens = j.input_tokens;
if (typeof j?.output_tokens === "number") usage.outputTokens = j.output_tokens;
if (typeof j?.cache_read_input_tokens === "number") usage.cacheReadTokens = j.cache_read_input_tokens;
if (typeof j?.cache_creation_input_tokens === "number") usage.cacheWriteTokens = j.cache_creation_input_tokens;
if (typeof j?.total_cost_usd === "number") usage.costUsd = j.total_cost_usd;
return usage;
} catch { /* fall through */ }
}
// Fallback: ~/.config/claude/session-usage.json (older convention)
try {
const path = `${process.env.HOME}/.config/claude/session-usage.json`;
const raw = require("fs").readFileSync(path, "utf-8");
const j = JSON.parse(raw);
if (typeof j?.input_tokens === "number") usage.inputTokens = j.input_tokens;
if (typeof j?.output_tokens === "number") usage.outputTokens = j.output_tokens;
if (typeof j?.cost_usd === "number") usage.costUsd = j.cost_usd;
} catch { /* file missing — usage stays empty, never fail dispatch */ }
return usage;
}
async function dispatchClaudeCode(opts: DispatchOptions, resolvedModel: string): Promise<DispatchResult> {
const started = Date.now();
const timeoutMs = opts.timeoutMs ?? 120_000;
const args = ["-p", "--model", resolvedModel || "sonnet"];
if (opts.systemPrompt) args.push("--append-system-prompt", opts.systemPrompt);
const r = spawnSync("claude", args, {
input: opts.prompt,
encoding: "utf-8",
cwd: opts.cwd,
env: { ...process.env },
timeout: timeoutMs,
maxBuffer: 10 * 1024 * 1024,
});
const durationMs = Date.now() - started;
const output = (r.stdout || "").trim();
const stderr = (r.stderr || "").trim();
const exitCode = r.status ?? -1;
const timedOut = r.signal === "SIGTERM";
const ok = !r.error && !timedOut && exitCode === 0 && output.length > 0;
const ccUsage = getClaudeCodeUsage();
const result: DispatchResult = {
ok,
output,
stderr,
durationMs,
provider: "claude-code",
model: resolvedModel || "sonnet",
exitCode,
error: r.error?.message || (timedOut ? "timeout" : undefined),
...ccUsage,
};
try {
mkdirSync(dirname(LOG_PATH), { recursive: true });
const { output: _fullOutput, ...logMeta } = result; // full output stays in the return value, not the log
appendFileSync(LOG_PATH, JSON.stringify({
ts: new Date().toISOString(),
promptPreview: opts.prompt.slice(0, 200),
backend: "claude-code",
...logMeta,
outputPreview: output.slice(0, 500),
}) + "\n");
} catch { /* */ }
return result;
}
export async function dispatch(opts: DispatchOptions): Promise<DispatchResult> {
const started = Date.now();
const { provider, model, keyEnv } = resolveModel(opts.model);
// Route to the claude-code CLI backend instead of pi when asked.
if (provider === "claude-code") {
return dispatchClaudeCode(opts, model);
}
const tools = (opts.tools && opts.tools.length ? opts.tools : ["read", "bash", "grep", "ls"]).join(",");
const timeoutMs = opts.timeoutMs ?? 120_000;
// Export the right API key for pi. pi auto-reads env vars by provider.
const apiKey = env(keyEnv);
const spawnEnv: NodeJS.ProcessEnv = {
...process.env,
[keyEnv]: apiKey,
};
const args = [
"--print",
"--no-session",
"--mode", "json",
"--provider", provider,
"--model", model,
"--tools", tools,
];
if (opts.systemPrompt) {
args.push("--append-system-prompt", opts.systemPrompt);
}
args.push(opts.prompt);
const runOnce = () => spawnSync("pi", args, {
encoding: "utf-8",
cwd: opts.cwd,
env: spawnEnv,
timeout: timeoutMs,
maxBuffer: 10 * 1024 * 1024,
});
// Up to 2 retries on transient SSE/network errors. Kernel-loaded dispatches
// ship a bigger context and hit OpenRouter SSE flakes more often; silent
// retries keep callers from having to know.
let r = runOnce();
let attempts = 1;
while (attempts < 3 && isTransient((r.stdout || "").trim(), (r.stderr || "").trim(), r.status ?? -1)) {
attempts++;
r = runOnce();
}
const durationMs = Date.now() - started;
const rawStdout = (r.stdout || "").trim();
const stderr = (r.stderr || "").trim();
const exitCode = r.status ?? -1;
const timedOut = r.signal === "SIGTERM";
// Parse pi's --mode json stream. Falls back to raw stdout if parsing fails
// (malformed frames, network error before any frame, etc).
const parsed = parsePiJsonStream(rawStdout);
const output = parsed.text != null ? parsed.text.trim() : rawStdout;
const ok = !r.error && !timedOut && exitCode === 0 && output.length > 0;
const result: DispatchResult = {
ok,
output,
stderr,
durationMs,
provider,
model,
exitCode,
error: r.error?.message || (timedOut ? "timeout" : undefined),
...parsed.usage,
};
// ndjson log — one line per dispatch, zero coupling to any log sink
try {
mkdirSync(dirname(LOG_PATH), { recursive: true });
const { output: _fullOutput, ...logMeta } = result; // full output stays in the return value, not the log
appendFileSync(LOG_PATH, JSON.stringify({
ts: new Date().toISOString(),
promptPreview: opts.prompt.slice(0, 200),
tools,
backend: "pi",
...logMeta,
outputPreview: output.slice(0, 500),
}) + "\n");
} catch { /* non-fatal — logging must never break dispatch */ }
return result;
}
// --- CLI ---
/** WHAT THIS HAND ANSWERS, and what each verb does to the world.
* Derived from this file's own CLI dispatch by
* `snappy-hands/contract-derive.ts` — a verb the code does not implement is
* never declared here. Snappy's daemon reads it (`api.ts contract`) to
* validate every call, build the argument words in order, decide whether the
* act runs now or stages for the owner, and hand the child exactly the
* environment keys named in `requires` — never a value, never anything else.
*/
export const HAND_CONTRACT = {
skill: "snappy-dispatch",
description: "The cheap-labor lever. Dispatch mechanical, verifiable work from the Sonnet/Opus orchestrator to cheap sub-agents (Haiku, Gemini, Llama, etc.) running locally via pi-coding-agent. Triggers on mentions of dispatch, cheap labor, sub-agent fan-out, pi-coding-agent, Haiku/Gemini/Llama grunt work.",
managed: false,
requires: [] as string[],
refusals: refusalTable("unknown_verb"),
verbs: {
},
} as const;
if (import.meta.url === `file://${realpathSync(process.argv[1])}` && process.argv[2] === "contract") {
console.log(JSON.stringify(HAND_CONTRACT, null, 2));
process.exit(0);
}
if (import.meta.url === `file://${realpathSync(process.argv[1])}`) {
const args = process.argv.slice(2);
let model: string | undefined;
let tools: string[] | undefined;
let timeoutMs: number | undefined;
const prompt: string[] = [];
for (let i = 0; i < args.length; i++) {
const a = args[i];
if (a === "--model") { model = args[++i]; continue; }
if (a === "--tools") { tools = args[++i].split(","); continue; }
if (a === "--timeout") { timeoutMs = Number(args[++i]) * 1000; continue; }
if (a === "--help" || a === "-h") {
console.log(`snappy-dispatch — sub-agent dispatch CLI
npx tsx api.ts [--model alias] [--tools csv] [--timeout seconds] "prompt..."
Model aliases: ${Object.keys(ALIASES).join(", ")}
or pass a full "provider/model-id" string.
Default: model=haiku, tools=read,bash,grep,ls, timeout=120`);
process.exit(0);
}
prompt.push(a);
}
if (prompt.length === 0) {
console.error("Usage: npx tsx api.ts [flags] \"prompt...\"");
process.exit(1);
}
dispatch({ prompt: prompt.join(" "), model, tools, timeoutMs })
.then(r => {
if (r.ok) {
process.stdout.write(r.output + "\n");
process.exit(0);
} else {
console.error(`[dispatch] failed: exit=${r.exitCode} err=${r.error || "-"}`);
if (r.stderr) console.error(r.stderr);
process.exit(r.exitCode || 1);
}
})
.catch(e => { console.error(e); process.exit(1); });
}