Agent stream

Why it existsCopy Anchor Link

A delegated task's harness is launched with a prompt on stdin and nothing else agreed. That poverty is what lets the command hold Claude Code, Codex, or a shell script. Streaming the session needs strictly more than that, so the extra knowledge lives here, in one adapter per dialect, and nowhere else.

Two rules everything else depends on:

1. The vocabulary is what gets stored, never the harness's own shape. A CLI's JSON output is an output format, not an API; persisting it would date the history the first time the vendor reshuffles a field. Adapters translate into a small set of event types, and anything unmapped becomes raw rather than being dropped.
2. A malformed line is never fatal. This parses third-party output on a read path. Garbage in a line costs that line, not the task.

This module is here for the reason that inverts the rest of the package: nothing has forked it yet. Every other module was extracted after two copies had drifted. Putting this one in the shared package now costs nothing and means the second product never writes its own.

The event vocabulary, field by field, is documented in Formats → Agent events.

import {
  applyProtocol,
  splitCompleteLines,
  parseLines,
  summarizeUsage,
  type AgentProtocol,
} from "@spunto/build/agent-stream"

ProtocolsCopy Anchor Link

AGENT_PROTOCOLS / AgentProtocolCopy Anchor Link

const AGENT_PROTOCOLS = ["none", "claude-json", "claude-stream", "jsonl"] as const
type AgentProtocol = (typeof AGENT_PROTOCOLS)[number]

applyProtocol(command, protocol)Copy Anchor Link

function applyProtocol(command: string, protocol: AgentProtocol): string

The command actually run, given the project's harness command and its protocol. The platform appends the flags rather than asking the project to, because getting them wrong is silent: a missing --verbose makes Claude Code refuse stream-json outright, and a missing --output-format produces prose that parses into nothing.

Left untouched for none / jsonl, and when the command already contains --output-format — a project that wants different flags keeps full control by saying so.

applyProtocol("claude -p", "claude-json")
// → "claude -p --output-format stream-json --verbose"
applyProtocol("claude -p", "claude-stream")
// → "claude -p --output-format stream-json --verbose --input-format stream-json"
applyProtocol("my-agent", "jsonl")
// → "my-agent"

isInteractive(protocol)Copy Anchor Link

function isInteractive(protocol: AgentProtocol): boolean

Does this protocol keep the harness alive, listening on stdin, between turns? Only claude-stream.

userTurnLine(protocol, text)Copy Anchor Link

function userTurnLine(protocol: AgentProtocol, text: string): string

One turn of the conversation, as the harness expects it on stdin. For claude-stream, a {"type":"user","message":{…}} JSON line; for every other protocol, the text itself. Always newline-terminated, to be written as a single write — under PIPE_BUF that is atomic, which is what stops two messages from interleaving into one unparseable line.

stdin.write(userTurnLine("claude-stream", "Now add tests for it."))

interruptLine(protocol, requestId)Copy Anchor Link

function interruptLine(protocol: AgentProtocol, requestId: string): string | null

Stops what the harness is doing without killing it: a control_request / interrupt line for claude-stream, null for every other protocol. Measured behaviour: Claude Code answers with a success control_response, ends the turn with result/error_during_execution, and stays alive — so the next message is a normal turn, not a resume.

Parsing a streamCopy Anchor Link

TASK_EVENT_TYPES / TaskEventType / NormalizedEventCopy Anchor Link

const TASK_EVENT_TYPES = [
  "session.started", "session.title", "message", "thinking", "tool.call", "tool.result",
  "plan", "usage", "session.ended", "session.background", "raw",
] as const
type TaskEventType = (typeof TASK_EVENT_TYPES)[number]
type NormalizedEvent = { type: TaskEventType; payload: Record<string, unknown> }

The normalised vocabulary. Deliberately short — it has to be translatable from several harnesses, so anything specific to one of them does not belong in it.

parseStreamLine(protocol, line)Copy Anchor Link

function parseStreamLine(protocol: AgentProtocol, line: string): NormalizedEvent[]

Parses one line into zero or more events. Zero is a normal answer (a blank line, a JSON value that is not an object). A non-JSON line becomes a raw event with its text — a harness that also prints a banner to stdout is common, and that must not be invisible either.

For the Claude dialects: system/init → session.started; assistant/user content blocks → message, thinking, tool.call, tool.result; a TodoWrite tool call additionally yields a plan; each assistant message with non-zero usage yields a usage with scope: "call" (and contextTokens = input + cacheRead + cacheCreation); result → a usage with scope: "session" plus session.ended; system/background_tasks_changed → session.background; anything else → raw. For jsonl, an object whose type is in the vocabulary passes through with the rest of the object as payload; anything else becomes raw.

parseStreamLine("jsonl", '{"type":"message","role":"assistant","text":"Done."}')
// → [{ type: "message", payload: { role: "assistant", text: "Done." } }]

SourcedEvent and parseLines(protocol, lines, budget)Copy Anchor Link

type SourcedEvent = NormalizedEvent & { source: string }
function parseLines(protocol: AgentProtocol, lines: string[], budget: number): SourcedEvent[]

Parses whole lines, keeping each event attached to the (trimmed) line it came from. That pairing is what makes normalisation reversible: type/payload is our reading of the line, source is the line — get the reading wrong and the line is still there to read again. budget caps how many events one pass may produce, so a runaway session cannot blow past a per-task ceiling in one chunk. An adapter that throws costs its line (with a console.warn), not the pass.

splitCompleteLines(chunk)Copy Anchor Link

function splitCompleteLines(chunk: Uint8Array): { lines: string[]; consumed: number }

Splits a chunk of bytes into whole lines and says how many bytes that consumed. The cursor only ever advances past a complete line; a read that lands mid-line leaves the tail for the next one. Without this, a 3 KB JSON event straddling two reads would be parsed as two invalid halves. Takes a Uint8Array so a Node Buffer passes straight through and the package stays runnable anywhere.

let offset = 0
async function poll(taskId: string) {
  const chunk = await readFrom(worker, logPath, offset) // bytes from `offset`
  const { lines, consumed } = splitCompleteLines(chunk)
  offset += consumed
  const events = parseLines(task.protocol, lines, 500)
  await saveEvents(taskId, events) // store type, payload and source
}

Session titlesCopy Anchor Link

Claude Code names its own conversations, but does not put that name on stdout: it writes it to the session transcript as {"type":"ai-title","aiTitle":…} lines. So the only way to get it is to read it out of band, in the worker — hence a shell probe plus a parser. A jsonl harness says its name in band, as a session.title event, and pays nothing.

readsTitleOutOfBand(protocol)Copy Anchor Link

function readsTitleOutOfBand(protocol: AgentProtocol): boolean

True for claude-json and claude-stream. Ask it before looking up a session id.

sessionTitleProbe(protocol, sessionId)Copy Anchor Link

function sessionTitleProbe(protocol: AgentProtocol, sessionId: string): string | null

A shell to run in the worker that prints the last few ai-title lines of the session's transcript (found by file name under ${CLAUDE_CONFIG_DIR:-$HOME/.claude}/projects). null for a dialect with no out-of-band title, or for a session id outside [A-Za-z0-9._-] (1–128 chars) — the id comes from the harness's output and is about to be interpolated into a shell, so anything else is refused. Finding nothing (for example, a harness run with --no-session-persistence) is a normal answer.

parseSessionTitle(protocol, output)Copy Anchor Link

function parseSessionTitle(protocol: AgentProtocol, output: string): { title: string; source: string } | null

The title out of what the probe printed: the newest line that is really an ai-title record wins (the probe's grep is a substring match on a file the agent itself can write into).

if (readsTitleOutOfBand(protocol)) {
  const probe = sessionTitleProbe(protocol, sessionId)
  if (probe) {
    const found = parseSessionTitle(protocol, await exec(worker, probe))
    if (found) await setTitle(taskId, found.title)
  }
}

cleanTitle(value)Copy Anchor Link

function cleanTitle(value: unknown): string | null

One line, whitespace runs collapsed, trimmed, at most 200 characters; null for a non-string or an empty result.

UsageCopy Anchor Link

UsageInput and summarizeUsage(input)Copy Anchor Link

type UsageInput = {
  calls: { count: number; inputTokens: number; outputTokens: number; cacheReadTokens: number; cacheCreationTokens: number }
  latestCall: Record<string, unknown> | null    // most recent scope: "call" payload
  latestSession: Record<string, unknown> | null // most recent scope: "session" payload
  started: Record<string, unknown> | null       // most recent session.started payload
}
function summarizeUsage(input: UsageInput): SessionUsage | null

Folds a task's usage events into the one recap a reader wants. Which number comes from where is the whole content of the function:

• tokens are summed from per-call events, because a turn in flight has not produced a harness total yet — and "0 tokens" under a session that has worked for ten minutes is never true. The harness total is the fallback for a dialect that only reports at the end;
• cost, turns and duration only ever come from the harness: pricing a call needs a rate card, and a platform that made one up would be quoting a number it cannot stand behind;
• context is the latest call's prompt size, against the window the session was opened with.

Returns null when nothing was measured at all — nothing measured is not zero measured.

SessionUsageCopy Anchor Link

Context windowsCopy Anchor Link

A session's context window has two sources of very different quality: a catalogue in this package (free, synchronous, and wrong on the day a model ships) and the Models API (GET /v1/models/{id} → max_input_tokens, the model's own answer). Neither guesses: a model in neither source is worth null, and a UI then shows a token count with no percentage — a wrong bar is worse than no bar.

apiModelId(model)Copy Anchor Link

function apiModelId(model: string): string

The API id inside whatever a harness wrote, so one catalogue covers every spelling: lowercases, drops Claude Code's [1m] suffix, a Bedrock anthropic. prefix (optionally behind a region), a Vertex @snapshot, a Bedrock -v1:0 suffix and a trailing -YYYYMMDD date.

apiModelId("claude-opus-5[1m]")                       // → "claude-opus-5"
apiModelId("us.anthropic.claude-sonnet-4-5-20250929-v1:0") // → "claude-sonnet-4-5"
apiModelId("claude-3-5-haiku@20241022")               // → "claude-3-5-haiku"

contextWindowOf(model)Copy Anchor Link

function contextWindowOf(model: string | null | undefined): number | null

From the catalogue alone — no credential, no network. 1_000_000 for any model string carrying the [1m] suffix; otherwise the catalogued window of its apiModelId, or null.

contextWindowOf("claude-opus-5")    // → 1000000
contextWindowOf("claude-haiku-4-5") // → 200000
contextWindowOf("some-future-model") // → null

ModelCredentialCopy Anchor Link

type ModelCredential = { kind: "api-key" | "oauth"; value: string }

How a session authenticates against Anthropic. A tagged union, not a string, because the two kinds use different headers and sending a token on the wrong one fails quietly as a 401 that looks like a revoked key: api-key (ANTHROPIC_API_KEY) goes on x-api-key; oauth (a subscription token such as CLAUDE_CODE_OAUTH_TOKEN) goes on Authorization: Bearer plus the oauth-2025-04-20 beta header.

liveContextWindow(model, credential)Copy Anchor Link

function liveContextWindow(model: string, credential: ModelCredential): Promise<number | null>

Asks the Models API (4-second timeout). Never throws: every failure mode — no network, revoked key, unknown model, insufficient scope, 429 — resolves to null, because a session must not stop streaming over a catalogue lookup. Memoised per process: a hit for the life of the process, a miss retried after 10 minutes.

resolveContextWindow(model, credential)Copy Anchor Link

function resolveContextWindow(
  model: string | null | undefined,
  credential: ModelCredential | null | undefined,
): Promise<number | null>

The window to record for a session, best source first: the [1m] suffix wins outright (it is a statement about this session); then the Models API when a credential is given; then the catalogue — including when there is no credential at all, which is ordinary, not an error.

const started = events.find((e) => e.type === "session.started")
if (started) {
  const key = secrets.ANTHROPIC_API_KEY
  started.payload.contextWindow = await resolveContextWindow(
    started.payload.model as string | null,
    key ? { kind: "api-key", value: key } : null,
  )
}

resetModelWindowCache()Copy Anchor Link

function resetModelWindowCache(): void

Clears the process-wide memo of liveContextWindow. For tests only.