fix(pi): stream stdout to disk + NDJSON parsing to bound memory

src/adapters/pi.ts

@@ -9,7 +9,7 @@ import type {
 } from "../core/types.ts"
 import { createLogger } from "../core/logger.ts"
 import { getAdapterRepoDir, getAdapterSettings } from "../core/config.ts"
-import { envForRoute, resolveRoute, validateModelIdForRoute } from "../providers/registry.ts"
+import { envForRoute, resolveRoute, resolveBackendModel, validateModelIdForRoute } from "../providers/registry.ts"
 import { runSubprocess } from "../core/subprocess.ts"
 import { subprocessVerdict } from "./subprocess-verdict.ts"
 import { TASK_FILE_DEFAULTS } from "../core/ui-defaults.ts"
@@ -21,10 +21,11 @@ import {
   type Sandbox,
 } from "../core/adapter-sandbox.ts"
 import {
-  parsePiNDJSON,
-  piEventsToRunRecord,
+  piBuildRunRecordFromNDJSON,
+  piBuildRunRecordFromFile,
   toPiModel,
   renderPiBaseUrlOverride,
+  renderPiModelRegistration,
 } from "../core/pi-runtime.ts"
 
 const log = createLogger("pi")
@@ -186,7 +187,16 @@ export class PiAdapter implements AgentAdapter {
       // is needed to redirect the endpoint. Auth flows in via env vars derived
       // from the route — no auth.json needed.
       const route = resolveRoute(config.model)
-      const doc = renderPiBaseUrlOverride(route)
+      // For openai-compatible routes (DeepSeek, vLLM, any OpenAI proxy), the
+      // baseUrl-only override is NOT enough: pi's `openai` provider defaults
+      // custom models to `openai-responses` (POST {baseUrl}/responses), which
+      // non-OpenAI backends don't implement -> 404. Register the model
+      // explicitly with `api: openai-completions` so pi uses /chat/completions.
+      // Matches the headless library driver's behavior (see pi-runtime.ts).
+      const modelId = resolveBackendModel(config.model)
+      const doc = route.kind === "openai-compatible"
+        ? renderPiModelRegistration(route, modelId)
+        : renderPiBaseUrlOverride(route)
       if (doc) await Bun.write(path.join(root, "models.json"), doc)
     }
 
@@ -247,10 +257,16 @@ export class PiAdapter implements AgentAdapter {
     const envOverlay: Record<string, string> = { ...this.routeEnv }
     if (this.piAgentDir) envOverlay.PI_CODING_AGENT_DIR = this.piAgentDir
 
+    // Stream pi's NDJSON stdout straight to the convLog file instead of
+    // buffering it (agentic transcripts reach 0.3–1.7 GB → OOM). When a
+    // convLog path exists, streaming IS the convLog write; the file parser
+    // then reads it back. No convLog → fall back to the string path.
+    const convLogPath = task.convLog?.filePath
     const { stdout, stderr, exitCode, timedOut } = await runSubprocess(cmd, {
       cwd: task.workDir,
       timeoutMs: task.timeoutMs ?? this.timeoutMs,
       env: envOverlay,
+      stdoutSink: convLogPath,
     })
 
     const durationMs = performance.now() - startMs
@@ -259,18 +275,9 @@ export class PiAdapter implements AgentAdapter {
       log.warn(`pi exited with code ${exitCode}: ${stderr.slice(0, 200)}`)
     }
 
-    if (task.convLog && stdout.trim()) {
-      try {
-        const destDir = path.dirname(task.convLog.filePath)
-        await mkdir(destDir, { recursive: true })
-        await Bun.write(task.convLog.filePath, stdout)
-      } catch (err) {
-        log.warn(`Failed to save pi NDJSON: ${err}`)
-      }
-    }
-
-    const events = parsePiNDJSON(stdout)
-    const builder = piEventsToRunRecord(events)
+    const builder = convLogPath
+      ? await piBuildRunRecordFromFile(convLogPath)
+      : piBuildRunRecordFromNDJSON(stdout)
 
     if (task.skill && skillLoaded === false) {
       const skillSnippet = task.skill.content.replace(/^#.*\n/m, "").trim().slice(0, 60)

src/core/pi-runtime.ts

@@ -111,25 +111,16 @@ export function parsePiNDJSON(output: string): PiEvent[] {
 // events → RunResult (shared by adapter + headless driver)
 // ---------------------------------------------------------------------------
 
-export function piEventsToRunRecord(events: PiEvent[]): RunRecordBuilder {
-  const agentEndEvents = events.filter(
-    (e): e is Extract<PiEvent, { type: "agent_end" }> => e.type === "agent_end",
-  )
-  const lastAgentEnd = agentEndEvents[agentEndEvents.length - 1]
-
-  const messages: PiMessage[] = lastAgentEnd?.messages ? [...lastAgentEnd.messages] : []
-
-  if (messages.length === 0) {
-    const messageEnds = events.filter(
-      (e): e is Extract<PiEvent, { type: "message_end" }> => e.type === "message_end",
-    )
-    for (const me of messageEnds) {
-      if (me.message.role === "assistant" || me.message.role === "toolResult") {
-        messages.push(me.message)
-      }
-    }
-  }
-
+/**
+ * Shared message → RunRecord logic for both the NDJSON streaming path
+ * (`piBuildRunRecordFromNDJSON`) and the full-events path
+ * (`piEventsToRunRecord`, used by the headless driver which already has the
+ * events in memory). Extracted so the two paths cannot drift in behavior.
+ */
+function piMessagesToRunRecord(
+  messages: PiMessage[],
+  lastAgentEnd: Extract<PiEvent, { type: "agent_end" }> | undefined,
+): RunRecordBuilder {
   // Pi dialect: text on a tool-call turn claims the final text; conversation
   // order pairs outputs back via toolResult(), which also records the
   // standalone tool step pi transcripts carry.
@@ -192,6 +183,144 @@ export function piEventsToRunRecord(events: PiEvent[]): RunRecordBuilder {
   return builder
 }
 
+export function piEventsToRunRecord(events: PiEvent[]): RunRecordBuilder {
+  const agentEndEvents = events.filter(
+    (e): e is Extract<PiEvent, { type: "agent_end" }> => e.type === "agent_end",
+  )
+  const lastAgentEnd = agentEndEvents[agentEndEvents.length - 1]
+
+  const messages: PiMessage[] = lastAgentEnd?.messages ? [...lastAgentEnd.messages] : []
+
+  if (messages.length === 0) {
+    const messageEnds = events.filter(
+      (e): e is Extract<PiEvent, { type: "message_end" }> => e.type === "message_end",
+    )
+    for (const me of messageEnds) {
+      if (me.message.role === "assistant" || me.message.role === "toolResult") {
+        messages.push(me.message)
+      }
+    }
+  }
+
+  return piMessagesToRunRecord(messages, lastAgentEnd)
+}
+
+/**
+ * Collects the only pi events the RunRecord builder consumes (agent_end +
+ * message_end). Shared by the string-scanning `piBuildRunRecordFromNDJSON`
+ * and the streaming `piBuildRunRecordFromFile` so both paths apply the same
+ * `includes` pre-filter (the 99.9% noise skip) and the same fallback rules.
+ */
+class PiEventCollector {
+  private lastAgentEnd: Extract<PiEvent, { type: "agent_end" }> | undefined
+  private messageEnds: Extract<PiEvent, { type: "message_end" }>[] = []
+
+  ingestLine(line: string): void {
+    if (!line.trim()) return
+    // Pre-filter: only agent_end / message_end feed the builder. Skipping
+    // JSON.parse for the other ~99.9% of lines avoids creating tens of
+    // thousands of transient objects that would pressure the GC even though
+    // they are never retained.
+    if (
+      !line.includes('"type":"agent_end"') &&
+      !line.includes('"type":"message_end"')
+    ) {
+      return
+    }
+    try {
+      const e = JSON.parse(line) as PiEvent
+      if (e.type === "agent_end") {
+        this.lastAgentEnd = e
+      } else if (e.type === "message_end") {
+        this.messageEnds.push(e)
+      }
+    } catch {
+      log.debug(`Skipping non-JSON line: ${line.slice(0, 100)}`)
+    }
+  }
+
+  build(): RunRecordBuilder {
+    const messages: PiMessage[] = this.lastAgentEnd?.messages
+      ? [...this.lastAgentEnd.messages]
+      : []
+    if (messages.length === 0) {
+      for (const me of this.messageEnds) {
+        if (me.message.role === "assistant" || me.message.role === "toolResult") {
+          messages.push(me.message)
+        }
+      }
+    }
+    return piMessagesToRunRecord(messages, this.lastAgentEnd)
+  }
+}
+
+/**
+ * Stream-parse pi's NDJSON stdout directly into a RunRecord, retaining ONLY
+ * the events the builder actually consumes (agent_end + message_end).
+ *
+ * WHY: pi emits ~30k NDJSON events per long task (message_update / thinking /
+ * *_delta streaming deltas make up ~99.9% of a 0.3–1.7 GB transcript). The
+ * old path — `piEventsToRunRecord(parsePiNDJSON(stdout))` — materialized ALL
+ * of them into a retained `PiEvent[]`, but the builder reads only agent_end /
+ * message_end (~25 events). Holding 30k parsed objects alongside the buffered
+ * stdout string drove peak heap to 10–32 GB and threw
+ * `RangeError: Out of memory` on long crypto tasks (circuit-fibsqrt,
+ * feal-*-cryptanalysis). This function is behaviorally equivalent to the old
+ * path but with O(relevant-events) memory instead of O(total-events).
+ *
+ * `output` is the raw NDJSON string (already buffered by runSubprocess); we
+ * scan it with indexOf instead of `output.split("\n")` so no 30k-element
+ * substring array is materialized, and each line string is releasable as we
+ * advance. A cheap `includes` pre-filter skips JSON.parse for the 99.9% of
+ * lines that are streaming deltas — pi emits compact JSON
+ * (`{"type":"message_end",...}`, no spaces), so the literal match is exact.
+ * False positives are harmless (JSON.parse still classifies the event);
+ * false negatives are impossible for pi's compact format.
+ */
+export function piBuildRunRecordFromNDJSON(output: string): RunRecordBuilder {
+  const collector = new PiEventCollector()
+
+  const len = output.length
+  let lineStart = 0
+  for (let i = 0; i <= len; i++) {
+    if (i !== len && output.charCodeAt(i) !== 10 /* \n */) continue
+    const line = lineStart < i ? output.slice(lineStart, i) : ""
+    lineStart = i + 1
+    collector.ingestLine(line)
+  }
+
+  return collector.build()
+}
+
+/**
+ * Streaming variant of `piBuildRunRecordFromNDJSON`: reads the NDJSON
+ * transcript from disk line-by-line instead of materializing the whole stdout
+ * string in memory. Used together with `runSubprocess({ stdoutSink })` — the
+ * subprocess streams its stdout verbatim to the convLog file, and this
+ * function streams it back out, so peak heap is O(longest-line) rather than
+ * O(transcript). For a 0.3–1.7 GB agentic transcript this is the difference
+ * between a 10–32 GB heap (RangeError: Out of memory) and a few MB.
+ */
+export async function piBuildRunRecordFromFile(filePath: string): Promise<RunRecordBuilder> {
+  const collector = new PiEventCollector()
+  const file = Bun.file(filePath)
+  if (!(await file.exists())) return collector.build()
+
+  const decoder = new TextDecoder()
+  let buf = ""
+  for await (const chunk of file.stream() as ReadableStream<Uint8Array>) {
+    buf += decoder.decode(chunk, { stream: true })
+    let nl: number
+    while ((nl = buf.indexOf("\n")) >= 0) {
+      collector.ingestLine(buf.slice(0, nl))
+      buf = buf.slice(nl + 1)
+    }
+  }
+  if (buf) collector.ingestLine(buf) // trailing line without a final \n
+
+  return collector.build()
+}
+
 // ---------------------------------------------------------------------------
 // Model translation (skvm route → pi provider/model id)
 // ---------------------------------------------------------------------------

src/core/subprocess.ts

@@ -6,12 +6,18 @@
  * (~64 KB on macOS) while the parent blocks on `proc.exited`.
  */
 
+import { mkdir } from "node:fs/promises"
+import path from "node:path"
+
 export interface SubprocessResult {
   exitCode: number
   stdout: string
   stderr: string
   durationMs: number
   timedOut: boolean
+  /** Present (== opts.stdoutSink) when at least one stdout byte was streamed
+   * to disk instead of buffered into `stdout`. When set, `stdout` is empty. */
+  stdoutFile?: string
 }
 
 export interface SubprocessOptions {
@@ -24,6 +30,49 @@ export interface SubprocessOptions {
    * removes that variable from the child's environment.
    */
   env?: Record<string, string | undefined>
+  /**
+   * When set, stream raw stdout bytes verbatim to this file path instead of
+   * buffering them into `result.stdout`. `result.stdout` becomes "" and
+   * `result.stdoutFile` is set to the path only after the first non-empty
+   * chunk. A child that produces no stdout leaves no file or `stdoutFile`
+   * behind (preserves the old `stdout.trim()` guard in adapters).
+   *
+   * WHY: agentic LLM transcripts (pi/opencode/claude-code) reach 0.3–1.7 GB;
+   * buffering that into a single string drives peak heap to 10–32 GB and
+   * throws `RangeError: Out of memory`. Streaming to the convLog file
+   * collapses the dual-use "write to convLog + parse" into one disk write.
+   */
+  stdoutSink?: string
+}
+
+/**
+ * Stream raw stdout bytes verbatim to a file. Lazy-open on the first
+ * non-empty chunk so a child with no stdout leaves no empty file behind.
+ * `finally { writer.end() }` flushes whatever was captured even if the
+ * stream ends early (e.g. the child is killed on timeout).
+ */
+// eslint-disable-next-line @typescript-eslint/no-explicit-any
+async function readStreamToSink(stream: any, sinkPath: string): Promise<boolean> {
+  const reader = stream.getReader()
+  // eslint-disable-next-line @typescript-eslint/no-explicit-any
+  let writer: any
+  let wroteAny = false
+  try {
+    for (;;) {
+      const { done, value } = await reader.read()
+      if (done) break
+      if (!value || value.byteLength === 0) continue
+      if (!writer) {
+        await mkdir(path.dirname(sinkPath), { recursive: true })
+        writer = Bun.file(sinkPath).writer()
+      }
+      writer.write(value)
+      wroteAny = true
+    }
+  } finally {
+    await writer?.end()
+  }
+  return wroteAny
 }
 
 export async function runSubprocess(
@@ -50,12 +99,29 @@ export async function runSubprocess(
     }, opts.timeoutMs)
   }
 
+  // When a sink is requested, stream stdout to disk (bounds heap for giant
+  // transcripts); otherwise buffer it into the result string as before.
+  const sinkPath = opts?.stdoutSink
+  let wroteStdoutToSink = false
+  const stdoutPromise = sinkPath
+    ? readStreamToSink(proc.stdout, sinkPath).then((wroteAny) => {
+        wroteStdoutToSink = wroteAny
+        return ""
+      })
+    : new Response(proc.stdout).text()
   const [exitCode, stdout, stderr] = await Promise.all([
     proc.exited.then((code) => { if (timer) clearTimeout(timer); return code }),
-    new Response(proc.stdout).text(),
+    stdoutPromise,
     new Response(proc.stderr).text(),
   ])
-  return { exitCode, stdout, stderr, durationMs: Date.now() - start, timedOut }
+  return {
+    exitCode,
+    stdout,
+    stderr,
+    durationMs: Date.now() - start,
+    timedOut,
+    ...(wroteStdoutToSink && sinkPath ? { stdoutFile: sinkPath } : {}),
+  }
 }
 
 function mergeEnv(