import { appendFileSync, closeSync, copyFileSync, existsSync, mkdirSync, openSync, readFileSync, readSync, readdirSync, renameSync, statSync, writeFileSync, } from "node:fs"; import { randomBytes, randomUUID } from "node:crypto"; import { dirname, join } from "node:path"; export interface SessionEntry { type: string; id: string; parentId?: string; [key: string]: unknown; } export interface MessageEntry extends SessionEntry { type: "message"; message: { role: "user" | "assistant" | "toolResult"; content: Array<{ type: string; text?: string; [key: string]: unknown }>; }; } export type SeededSubagentSessionMode = "lineage-only" | "fork"; function getForkContentLines(parentSessionFile: string): string[] { const raw = readFileSync(parentSessionFile, "utf8"); const lines = raw.split("\n").filter((line) => line.trim()); let truncateAt = lines.length; for (let i = lines.length - 1; i >= 0; i--) { try { const entry = JSON.parse(lines[i]); if (entry.type === "message" && entry.message?.role === "user") { truncateAt = i; break; } } catch { // ignore malformed lines } } return lines.slice(0, truncateAt).filter((line) => { try { return JSON.parse(line).type !== "session"; } catch { return true; } }); } export function seedSubagentSessionFile(params: { mode: SeededSubagentSessionMode; parentSessionFile: string; childSessionFile: string; childCwd: string; }): void { const header = { type: "session", version: 3, id: randomUUID(), timestamp: new Date().toISOString(), cwd: params.childCwd, parentSession: params.parentSessionFile, }; const contentLines = params.mode === "fork" ? getForkContentLines(params.parentSessionFile) : []; const lines = [JSON.stringify(header), ...contentLines]; mkdirSync(dirname(params.childSessionFile), { recursive: true }); writeFileSync(params.childSessionFile, lines.join("\n") + "\n", "utf8"); } /** * A snapshot of everything needed to reconstruct a subagent's sandbox when its * session is later resumed via `subagent_message({ sessionId })`. * * Written next to the session file as `.loadout.json` at spawn * time. Resume replays this exact snapshot so the reincarnated process gets the * same `--no-extensions` + `--tools` restriction, model, identity, spawn * whitelist, cwd, and config dir it originally ran with — instead of falling * back to pi's default (all global extensions + full toolset). Storing the * resolved loadout (rather than re-deriving from the agent `.md` by name) keeps * resume faithful even if the agent definition is later edited, moved, or * deleted. */ export interface SubagentLoadout { /** Agent profile name (for PI_SUBAGENT_AGENT); null for agentless spawns. */ agent: string | null; /** The `--tools` allowlist string, or null when the spawn was unrestricted. */ toolAllowlist: string | null; /** Model id (without thinking suffix), or null to use the session default. */ model: string | null; /** Thinking level appended to the model as `model:level`, or null. */ thinking: string | null; /** How the identity text was applied: append/replace, or null. */ systemPromptMode: "append" | "replace" | null; /** The system-prompt/identity text, only when it lived in the system prompt. */ identity: string | null; /** Agents this subagent was allowed to spawn (for PI_SUBAGENT_ALLOWED). */ spawnable: string[] | null; /** Whether the agent auto-exits (informational; resume forces autonomous). */ autoExit: boolean; /** Working directory the subagent ran in, or null. */ cwd: string | null; /** PI_CODING_AGENT_DIR the subagent resolved config/extensions from, or null. */ agentDir: string | null; } /** Path of the loadout sidecar written next to a subagent session file. */ export function loadoutSidecarPath(sessionFile: string): string { return `${sessionFile}.loadout.json`; } /** Persist a subagent's resolved sandbox loadout beside its session file. */ export function writeSubagentLoadout(sessionFile: string, loadout: SubagentLoadout): void { try { writeFileSync(loadoutSidecarPath(sessionFile), JSON.stringify(loadout), "utf8"); } catch { // Best-effort: a missing snapshot only means resume will refuse, never that // it launches unrestricted. } } /** Read a subagent's loadout snapshot, or null if absent/unparseable. */ export function readSubagentLoadout(sessionFile: string): SubagentLoadout | null { try { const p = loadoutSidecarPath(sessionFile); if (!existsSync(p)) return null; const parsed = JSON.parse(readFileSync(p, "utf8")); if (!parsed || typeof parsed !== "object") return null; return parsed as SubagentLoadout; } catch { return null; } } // ── Name registry ──────────────────────────────────────────────────────────── // Each spawner session (the top-level pi session, or a worker that spawns its // own children) gets a registry mapping a subagent's display name to the // session file it ran in. Names are unique per spawner session and persist on // disk, so `subagent_message({ name })` can steer a running subagent or resume // a finished one by the same handle — even across a pi restart. The registry // lives in the spawner's own artifact dir, which is directly addressable from // the spawner's session id (no sessions-tree scan, so resume stays fast). export interface NameRegistryEntry { /** Absolute path to the subagent's session .jsonl file. */ sessionFile: string; /** Canonical session header id (kept for display/lineage). */ sessionId: string | null; } export type NameRegistry = Record; /** Path of the name registry for a given spawner session's artifact dir. */ export function nameRegistryPath(artifactDir: string): string { return join(artifactDir, "subagent-registry.json"); } /** Read a spawner session's name registry, or {} if absent/corrupt. */ export function readNameRegistry(artifactDir: string): NameRegistry { try { const p = nameRegistryPath(artifactDir); if (!existsSync(p)) return {}; const parsed = JSON.parse(readFileSync(p, "utf8")); if (!parsed || typeof parsed !== "object" || Array.isArray(parsed)) return {}; return parsed as NameRegistry; } catch { return {}; } } /** * Register (or overwrite) a name → session mapping for a spawner session. * Writes atomically (temp file + rename) so a concurrent reader never sees a * partial registry. */ export function registerName( artifactDir: string, name: string, entry: NameRegistryEntry, ): void { try { mkdirSync(artifactDir, { recursive: true }); const registry = readNameRegistry(artifactDir); registry[name] = entry; const p = nameRegistryPath(artifactDir); const tmp = `${p}.tmp-${process.pid}-${Math.random().toString(16).slice(2, 8)}`; writeFileSync(tmp, JSON.stringify(registry, null, 2), "utf8"); renameSync(tmp, p); } catch { // Best-effort: a failed registration only means resume-by-name won't find // this subagent later; it never breaks the spawn itself. } } /** Resolve a name to its registry entry within a spawner session, or null. */ export function resolveNameInRegistry( artifactDir: string, name: string, ): NameRegistryEntry | null { const entry = readNameRegistry(artifactDir)[name]; return entry && typeof entry.sessionFile === "string" ? entry : null; } function readEntries(sessionFile: string): SessionEntry[] { const raw = readFileSync(sessionFile, "utf8"); return raw .split("\n") .filter((line) => line.trim()) .map((line) => JSON.parse(line) as SessionEntry); } /** * Return the id of the last entry in the session file (current branch point / leaf). */ export function getLeafId(sessionFile: string): string | null { const entries = readEntries(sessionFile); return entries.length > 0 ? entries[entries.length - 1].id : null; } /** * Read the canonical session id from a session file's header. * * pi's `--session ` flag resolves against this header `id` (exact match, * then prefix), NOT the filename — so this is the value to hand back to the * orchestrator for follow-ups. */ /** * Read only the first line of a file without loading the whole thing into * memory. Session files grow to many MB, but the header we need is always the * first JSON line, so reading a small prefix keeps header lookups cheap — this * is what makes scanning a large session tree fast enough to avoid blocking the * event loop. Returns the first line (sans trailing newline), or null. */ function readFirstLine(path: string, maxBytes = 65536): string | null { let fd: number | undefined; try { fd = openSync(path, "r"); const buf = Buffer.allocUnsafe(maxBytes); const bytes = readSync(fd, buf, 0, maxBytes, 0); if (bytes <= 0) return null; const nl = buf.indexOf(0x0a); // '\n' const end = nl === -1 || nl >= bytes ? bytes : nl; return buf.toString("utf8", 0, end); } catch { return null; } finally { if (fd !== undefined) { try { closeSync(fd); } catch { /* ignore */ } } } } export function getSessionId(sessionFile: string): string | null { return readHeaderId(sessionFile); } function readHeaderId(sessionFile: string): string | null { const firstLine = readFirstLine(sessionFile)?.trim(); if (!firstLine) return null; try { const entry = JSON.parse(firstLine) as { type?: string; id?: string }; return entry.type === "session" && typeof entry.id === "string" ? entry.id : null; } catch { return null; } } /** * Resolve a session id (or id prefix) to a session file path by scanning every * `*.jsonl` under `sessionsRoot` and matching the header `id`. Mirrors pi's own * resolution order: exact match first, then prefix match. Most recently * modified file wins on ties. Returns null when nothing matches. */ /** * In-process index of session id → session file, per sessions root. * * Resolving a session id naively walks every `.jsonl` under the sessions tree * and reads each header. With a few thousand sessions that is thousands of * synchronous open/read/stat syscalls — on the extension host's single thread * that blocks the entire terminal UI for many seconds (measured ~67s on a * 2010-file tree). To avoid that, we build the index once per root and cache * it; subsequent lookups are O(1). The cache is validated cheaply (a directory * listing plus statSync-only mtime checks) on every call, so new sessions are * picked up without re-reading unchanged headers and without ever freezing the * UI again. */ interface SessionIndex { idToFile: Map; /** file path → mtime when indexed (staleness detection). */ files: Map; /** top-level dir signature used to detect newly added cwd dirs. */ topSig: string; } const sessionIndexCache = new Map(); function topLevelSignature(root: string): string { const parts: string[] = []; let entries: import("node:fs").Dirent[]; try { entries = readdirSync(root, { withFileTypes: true }); } catch { return ""; } for (const e of entries) { const full = join(root, e.name); if (e.isDirectory()) { let m = 0; try { m = statSync(full).mtimeMs; } catch { /* ignore */ } parts.push(`d:${e.name}:${m}`); } else if (e.isFile() && e.name.endsWith(".jsonl")) { parts.push(`f:${e.name}`); } } parts.sort(); return parts.join("|"); } /** Recursively index new/changed .jsonl files under dir into idx. */ function indexDir(dir: string, idx: SessionIndex): void { let entries: import("node:fs").Dirent[]; try { entries = readdirSync(dir, { withFileTypes: true }); } catch { return; } for (const entry of entries) { const full = join(dir, entry.name); if (entry.isDirectory()) { indexDir(full, idx); } else if (entry.isFile() && entry.name.endsWith(".jsonl")) { let mtime = 0; try { mtime = statSync(full).mtimeMs; } catch { continue; } const known = idx.files.get(full); if (known !== undefined && known === mtime) continue; // unchanged const id = readHeaderId(full); // only read headers for new/changed files idx.files.set(full, mtime); if (!id) continue; const prev = idx.idToFile.get(id); if (!prev || mtime >= prev.mtime) { idx.idToFile.set(id, { path: full, mtime }); } } } } function getSessionIndex(sessionsRoot: string): SessionIndex { let idx = sessionIndexCache.get(sessionsRoot); const sig = topLevelSignature(sessionsRoot); if (!idx) { idx = { idToFile: new Map(), files: new Map(), topSig: sig }; sessionIndexCache.set(sessionsRoot, idx); indexDir(sessionsRoot, idx); // first build: full scan, once per process } else if (idx.topSig !== sig) { idx.topSig = sig; indexDir(sessionsRoot, idx); // a cwd dir was added/changed: incremental rescan } else { indexDir(sessionsRoot, idx); // cheap: stats files, reads only new/changed headers } return idx; } export function resolveSessionFileById(sessionId: string, sessionsRoot: string): string | null { if (!sessionId || !existsSync(sessionsRoot)) return null; const idx = getSessionIndex(sessionsRoot); return lookupSessionIndex(idx, sessionId); } function lookupSessionIndex( idx: { idToFile: Map }, sessionId: string, ): string | null { // Exact match first. const exact = idx.idToFile.get(sessionId); if (exact && existsSync(exact.path)) return exact.path; // Prefix match: most recently modified wins (ids are unique in practice, so // this is only a convenience for hand-typed short prefixes). let best: { path: string; mtime: number } | null = null; for (const [id, rec] of idx.idToFile) { if (!id.startsWith(sessionId)) continue; if (!existsSync(rec.path)) continue; if (!best || rec.mtime > best.mtime) best = rec; } return best ? best.path : null; } /** * Async variant used by the interactive resume path. Index building/refresh is * synchronous I/O, which can take many seconds on a cold OS page cache with a * few thousand sessions; running it synchronously would block the extension * host's single thread and freeze the terminal UI. Deferring to a macrotask * keeps the event loop responsive. The heavy work only happens on the first * resolution per process (and incrementally thereafter); warm lookups are ~50ms. */ export async function resolveSessionFileByIdAsync( sessionId: string, sessionsRoot: string, ): Promise { if (!sessionId || !existsSync(sessionsRoot)) return null; // Let the event loop breathe (and the UI repaint) before the sync scan. await new Promise((r) => setImmediate(r)); const idx = getSessionIndex(sessionsRoot); return lookupSessionIndex(idx, sessionId); } /** Test hook: drop the cached session index so tests start clean. */ export function resetSessionIndexCache(): void { sessionIndexCache.clear(); } /** * Return entries added after `afterLine` (1-indexed count of existing entries). */ /** * Count the number of entry lines in a session file without parsing each line * into an object. Used by the resume path, which only needs the *count* of * pre-existing entries (so it can later slice out the new ones). Parsing every * line of a large resumed transcript synchronously at resume time would block * the UI; counting newlines is dramatically cheaper. */ export function countSessionEntryLines(sessionFile: string): number { try { const raw = readFileSync(sessionFile, "utf8"); // Count non-blank lines, mirroring getNewEntries' `.filter(line => line.trim())` // but skipping the per-line JSON.parse that makes resume slow on big files. let count = 0; for (const line of raw.split("\n")) { if (line.trim()) count++; } return count; } catch { return 0; } } export function getNewEntries(sessionFile: string, afterLine: number): SessionEntry[] { const raw = readFileSync(sessionFile, "utf8"); const lines = raw.split("\n").filter((line) => line.trim()); return lines.slice(afterLine).map((line) => JSON.parse(line) as SessionEntry); } /** * Find the last assistant message text in a list of entries. * * Falls back to the `errorMessage` field when the last assistant message has * `stopReason: "error"` and no usable text content — this happens when * auto-retry exhausts on a provider overload / rate limit / server error, and * without this fallback the parent would silently see a stale earlier message. */ export function findLastAssistantMessage(entries: SessionEntry[]): string | null { for (let i = entries.length - 1; i >= 0; i--) { const entry = entries[i]; if (entry.type !== "message") continue; const msg = entry as MessageEntry; if (msg.message.role !== "assistant") continue; const texts = msg.message.content .filter( (block) => block.type === "text" && typeof block.text === "string" && block.text.trim() !== "", ) .map((block) => block.text as string); if (texts.length > 0 && texts.join("").trim()) return texts.join("\n"); const stopReason = (msg.message as { stopReason?: unknown }).stopReason; const errorMessage = (msg.message as { errorMessage?: unknown }).errorMessage; if ( stopReason === "error" && typeof errorMessage === "string" && errorMessage.trim() !== "" ) { return `Subagent error: ${errorMessage.trim()}`; } } return null; } /** * Append a branch_summary entry to the session file. * Returns the new entry's id. */ export function appendBranchSummary( sessionFile: string, branchPointId: string, fromId: string | null, summary: string, ): string { const id = randomBytes(4).toString("hex"); const entry = { type: "branch_summary", id, parentId: branchPointId, timestamp: new Date().toISOString(), fromId: fromId ?? branchPointId, summary, }; appendFileSync(sessionFile, JSON.stringify(entry) + "\n", "utf8"); return id; } /** * Copy the session file to destDir for parallel worker isolation. * Returns the path of the copy. */ export function copySessionFile(sessionFile: string, destDir: string): string { const id = randomBytes(4).toString("hex"); const dest = join(destDir, `subagent-${id}.jsonl`); copyFileSync(sessionFile, dest); return dest; } /** * Read new entries from sourceFile (after afterLine), append them to targetFile. * Returns the appended entries. */ export function mergeNewEntries( sourceFile: string, targetFile: string, afterLine: number, ): SessionEntry[] { const entries = getNewEntries(sourceFile, afterLine); for (const entry of entries) { appendFileSync(targetFile, JSON.stringify(entry) + "\n", "utf8"); } return entries; } export interface SessionStats { model: string | null; toolCount: number; /** Cumulative token usage across all assistant turns. */ inputTokens: number; outputTokens: number; cacheReadTokens: number; cacheWriteTokens: number; /** Current context size: the last assistant turn's totalTokens. */ contextTokens: number; /** Cumulative cost in USD across all assistant turns. */ cost: number; } /** * Parse a completed subagent session JSONL into aggregate stats for display: * model, tool-call count, cumulative token usage + cost, and current context * size. Cumulative usage fields are summed across every assistant turn; the * context size is taken from the last assistant turn's `totalTokens` (the live * context window occupancy). Returns null if the file can't be read. */ export function summarizeSessionStats(sessionFile: string): SessionStats | null { let entries: SessionEntry[]; try { entries = readEntries(sessionFile); } catch { return null; } const stats: SessionStats = { model: null, toolCount: 0, inputTokens: 0, outputTokens: 0, cacheReadTokens: 0, cacheWriteTokens: 0, contextTokens: 0, cost: 0, }; for (const entry of entries) { if (entry.type === "model_change") { const modelId = (entry as { modelId?: unknown }).modelId; if (typeof modelId === "string" && modelId) stats.model = modelId; continue; } if (entry.type !== "message") continue; const msg = (entry as MessageEntry).message; if (msg.role !== "assistant") continue; const model = (msg as { model?: unknown }).model; if (typeof model === "string" && model) stats.model = model; for (const block of msg.content) { if (block.type === "toolCall") stats.toolCount++; } const usage = (msg as { usage?: Record }).usage; if (usage && typeof usage === "object") { const num = (v: unknown): number => (typeof v === "number" && Number.isFinite(v) ? v : 0); stats.inputTokens += num(usage.input); stats.outputTokens += num(usage.output); stats.cacheReadTokens += num(usage.cacheRead); stats.cacheWriteTokens += num(usage.cacheWrite); const total = num(usage.totalTokens); if (total > 0) stats.contextTokens = total; const cost = usage.cost; if (cost && typeof cost === "object") stats.cost += num((cost as Record).total); } } return stats; }