Streaming

Watch a run as it works: assistant text as it is produced, tool calls as they start and finish, reports as they are pushed, and the settled outcome.

The SDK surface

const stream = client.nodes.stream({ prompt: 'Fix the failing test', cwd }, { headers: { 'x-trace': 'run-9' } });

stream.on('node.output_text.delta', (event) => process.stdout.write(event.delta));

for await (const event of stream) {
  // the same events, typed by `type`
}

const outcome = await stream.finalOutcome(); // resolves on node.settled
stream.abort(); // ends the HTTP response; the node keeps running

| Member | Behaviour | |---|---| | client.nodes.stream(params, options?) | Creates the node, then opens its event stream. Returns synchronously; stream.node is a Promise<NodeDetailDTO> for the created node. The request options apply to creation and to opening the stream, except timeout, because a stream has no wall-clock timeout. | | client.nodes.events(id, options?) | Validates and retrieves an existing node, then streams it. options is NodeEventsOptions: after plus request headers, signal, and maxRetries; it deliberately has no timeout. | | .on(type, listener) / .off(type, listener) | Typed event emitter; each returns the NodeStream. | | [Symbol.asyncIterator]() | Yields the same NodeEventDTO union. | | .finalOutcome() | Returns Promise<NodeOutcomeDTO>, resolving on node.settled and rejecting on a terminal error event or a connection failure. | | .abort() | Aborts the underlying fetch. An in-flight iterator and finalOutcome() reject with APIUserAbortError. Passing your own signal in the options aborts the stream the same way; NodeStream exposes no signal property. |

The method is called events, not subscribe: a subscription is already a push-delivery edge between two nodes on the canvas, and one word must not mean two things.

A stream is an observer, never a lifecycle hold. Disconnecting drops your subscriber and leaves the node running. Opening a stream never revives a dormant node — call client.nodes.revive(id) first if that is what you want.

Activity helper

followActivity(stream, { describe? }) turns streamed tool-call events into immutable ActivityStep[] snapshots for a plain activity feed. describeToolDefault(tool, summary) names the common tools; pass describe to use labels for your application or return null to hide a tool call. Tool summaries describe the argument shape, never argument values or raw arguments.

import { followActivity } from '@north-light/crouter-sdk';

for await (const steps of followActivity(stream)) {
  render(steps);
}

A step starts as running; a node.tool_call.completed event changes it to done when status is ok or failed when status is error.

Events

Every event except error carries node_id and sequence_number in addition to the fields listed. error carries only its error object.

| Event | data | |---|---| | node.output_text.delta | { node_id, sequence_number, delta } | | node.output_text.done | { node_id, sequence_number, text } | | node.tool_call.started | { node_id, sequence_number, tool_call_id, tool, summary } | | node.tool_call.completed | { node_id, sequence_number, tool_call_id, tool, status: 'ok' \| 'error', summary } | | node.turn.started | { node_id, sequence_number } | | node.turn.completed | { node_id, sequence_number } | | node.report.pushed | { node_id, sequence_number, report } | | node.status.changed | { node_id, sequence_number, status }, where status can also be stream-only 'dormant' | | node.settled | { node_id, sequence_number, outcome } — terminal; the daemon ends the response after it | | error | { error: { code: 'stream_gap' \| 'stream_dropped' \| 'stream_error', message, details?: { earliest_sequence?: number } } } — stream_gap is recoverable; the other codes terminate the response |

summary on a tool-call event states whether arguments are null, an array and its item count, an object and its field count, or a primitive type. It never includes argument values or tool output.

Deliberately not carried: thinking deltas, the model's tool-call construction deltas, raw tool output, and the system prompt. Those belong to the owner's viewer, not to an application's run stream.

The route

GET /v1/nodes/{id}/events        Accept: text/event-stream
    ?after=<sequence_number>     resume from a cursor (optional)

Server-sent events: one record per event as event: <type>, data: <JSON>, blank line, plus : keepalive comment lines every 15 s. Authentication is the daemon's existing rule — filesystem permission on the unix socket, bearer token over TCP.

What you get depending on the node's state

| Node state when you call | What the stream does | |---|---| | Already settled | Writes retained events when available, ending in node.settled; otherwise writes node.settled with the outcome and ends. Nothing is revived. | | Running | Streams live, seeded as described below. | | Dormant and not settled | Writes node.status.changed { status: 'dormant' } and holds the response open with keepalives. It starts streaming if and when the daemon brings a broker up for that node. |

A node can settle between your create and your events call, so both branches are ordinary. Either way the terminal event is node.settled, read from the same outcome row nodes.outcome reads — a client that joins after settlement and a client that was live at settlement see the same outcome.

Sequence, resume, and failure

sequence_number is monotonic per node while that node's in-memory stream hub exists. The daemon keeps one in-memory ring per streamed node holding the last 512 events or 256 KiB, whichever binds first.

| Situation | Behaviour | |---|---| | You join a run already in progress | First replays already-pushed reports oldest first, then seeds from the broker's snapshot: one node.output_text.done per completed assistant message, then one node.output_text.delta carrying the accumulated partial, then live. No content is lost — only delta granularity. | | A second subscriber joins | Seeded from the ring, so both subscribers see the same sequence numbers. | | after=N within the retained range | Replays events after N. | | after=N outside the retained range, including a future cursor | An error event with code stream_gap carries the earliest available sequence, then the stream continues from that point. It never silently skips. | | One event exceeds the 256 KiB replay budget | Live subscribers receive it, but the daemon clears the resume ring. A cursor before that event gets stream_gap; a cursor on it resumes at the next retained event. | | Your HTTP reader is too slow | The daemon ends the response with an error event whose code is stream_dropped. One slow reader cannot stall the engine. | | The node's broker is replaced or its observer socket closes | The daemon reconnects and emits node.status.changed. Sequence continues while the stream hub remains in memory. | | The daemon restarts | The ring is gone. A resuming client gets stream_gap with earliest_sequence: 1. There is no durable event log. | | You disconnect | Your subscriber is dropped. The node keeps running. |

Resuming

import { APIError } from '@north-light/crouter-sdk';

let cursor: number | undefined;

for (;;) {
  const stream = client.nodes.events(id, { after: cursor });
  try {
    for await (const event of stream) {
      if (event.type === 'error') {
        if (event.error.code === 'stream_gap') {
          console.warn('missed events before', event.error.details?.earliest_sequence);
          continue; // the stream continues from the floor
        }
        throw new APIError(0, event.error.code, event.error.message, event.error.details);
      }
      cursor = event.sequence_number;
      handle(event);
    }
    return; // ended on node.settled
  } catch (error) {
    if (error instanceof APIError && error.code === 'stream_dropped') continue;
    throw error;
  }
}

stream_gap is recoverable and the stream continues after it. stream_dropped is terminal for that HTTP response — reconnect with the cursor you last saw.

Identifier validation

nodes.events(id, options?) validates id before it requests the node or opens SSE. Because it returns a NodeStream synchronously, an invalid node identifier rejects stream.node, stream.finalOutcome(), and an iteration with TypeError; no request is sent. See Errors.

Event shapes depend on the pinned pi version

The delta and tool-call shapes come from the installed @earendil-works/pi-agent-core and pi-ai packages, not from crouter. The daemon's translator is the one place that depends on them. A pi version bump must be checked against it.