Streaming

The streaming overloads on recall.remember() and recall.search() return RecallStream<PipelineEvent, Final> where Final is RememberResult or SearchResult respectively. The class is AsyncIterable<PipelineEvent> for progress and exposes finalResult() for the accumulated outcome.

RecallStream

class RecallStream<Event, Final> implements AsyncIterable<Event> {
  [Symbol.asyncIterator](): AsyncIterator<Event>;
  finalResult(): Promise<Final>;
}

The class has two responsibilities: deliver per-event progress to a for await loop, and resolve to the accumulated final outcome via finalResult(). It deliberately does not implement PromiseLike — if it did, await Promise<RecallStream> (the natural shape returned by the streaming overload) would recursively unwrap to Final, hiding the stream from your code.

The two consumption patterns

Pattern A — iterate, then read final

const stream = await recall.remember({ messages, stream: true });

for await (const event of stream) {
  if (event.type === 'stage')    console.log(`stage ${event.stage} ${event.latencyMs}ms`);
  if (event.type === 'llm_call') console.log(`llm ${event.model} ${event.tokens} tok`);
}

const result = await stream.finalResult(); // RememberResult
console.log(`stored=${result.stored.length} traceId=${result.traceId}`);

The async iterator drains the underlying source. When it completes, the stream's extractFinal reducer is run and finalResult() resolves. A try/catch around the for await catches mid-stream pipeline failures.

Pattern B — final only, no iteration

const stream = await recall.remember({ messages, stream: true });
const result = await stream.finalResult(); // drains internally, no event handling

When you call finalResult() without iterating first, the stream drains the source internally and resolves the promise. Use this when you only want the final outcome but happened to be on a streaming endpoint (e.g. you are forced into streaming mode by an upstream caller).

STREAM_CONSUMED guard

The underlying SSE source can be consumed exactly once. Iterating twice is a programmer error:

const stream = await recall.remember({ messages, stream: true });
for await (const _ of stream) { /* … */ }
for await (const _ of stream) { /* … */ } // throws RecallError('STREAM_CONSUMED')

If you need to fan out events to multiple consumers, accumulate them once and dispatch yourself.

The PipelineEvent union

Both write and read streams yield PipelineEvent, a discriminated union with three variants. Switch on event.type:

type PipelineEvent = StageEvent | LlmCallEvent | CompleteEvent;

StageEvent

A pipeline stage completed. Emitted once per stage.

LlmCallEvent

An LLM invocation completed. Zero or more per stream.

CompleteEvent

The terminal event yielded just before the stream closes. Carries the same headline metrics the non-streaming response would have returned, so callers can render a summary without awaiting finalResult().

Wire format — the seven SSE event types

Internally, the server emits seven SSE event names. The SDK maps these to the public PipelineEvent union as follows:

The SDK normalises snake_case server fields to the SDK's camelCase. For example, the server's latency_us becomes latencyMs (rounded), cost_usd becomes costUsd, etc.

Mid-stream errors

If the server emits event: error after the stream has begun, the SDK throws RecallServerError('PIPELINE_FAILED', 500) from inside the for await loop:

import { RecallServerError } from '@arc-labs/recall';

try {
  const stream = await recall.search({ query, stream: true });
  for await (const event of stream) { /* … */ }
} catch (err) {
  if (err instanceof RecallServerError && err.code === 'PIPELINE_FAILED') {
    console.error('pipeline failed mid-stream:', err.message);
  }
}

If the stream throws and you have not yet iterated, await stream.finalResult() rejects with the same error. The shared completion promise is wired to surface failures from either pattern.

Cancellation

Pass an AbortSignal via options.signal to cancel mid-stream:

const ctrl = new AbortController();
setTimeout(() => ctrl.abort(), 5000);

const stream = await recall.search({ query, stream: true }, { signal: ctrl.signal });
for await (const event of stream) { /* … */ }
// throws RecallTimeoutError after 5 s

Cancellation is composed with the per-request timeoutMs — whichever fires first wins. See AbortSignal for the full pattern.

parseSse — the low-level helper

For callers who want to consume SSE directly (custom transports, replay tools, debugging proxies), the SDK exports parseSse:

import { parseSse, type SseEvent } from '@arc-labs/recall';

const response = await fetch(url, { headers: { Accept: 'text/event-stream' } });
if (response.body) {
  for await (const event of parseSse(response.body)) {
    // event: { type: string; data: string }
    console.log(event.type, event.data);
  }
}

parseSse(body: ReadableStream<Uint8Array>) is an async generator that yields SseEvent records as they arrive. It implements the SSE spec correctly: records separated by blank lines, multi-line data: joined with \n, comment lines (:foo) ignored, optional leading-space stripping on field values. id: and retry: lines are accepted but ignored — Recall does not use them.

When event: is absent, event.type is 'data' (per spec). When data: is absent, event.data is ''. The generator yields a final partial event on stream end without a trailing blank line, so trailing data is never lost.

This is a correctness-focused parser, not a high-level abstraction. For typed event consumption, stay with recall.remember({ stream: true }) and recall.search({ stream: true }).

When to stream

Stream when you want UI progress (a "thinking" indicator backed by stage events), real-time cost tracking (sum LlmCallEvent.costUsd as you go), or to start displaying results before the pipeline completes.

Don't stream when you only want the final outcome — the non-streaming overload is simpler, returns the same Result, and avoids the SSE wire overhead.

Common pitfalls

The streaming overload returns Promise<RecallStream<…>>, not RecallStream<…> directly. You always await it:

// ✗ wrong — stream is a Promise here
const stream = recall.remember({ messages, stream: true });
for await (const e of stream) { … } // TypeError: not iterable

// ✓ right
const stream = await recall.remember({ messages, stream: true });
for await (const e of stream) { … }

finalResult() and the iterator share the same source. If iteration throws, finalResult() rejects. If iteration succeeds, finalResult() resolves. There is no way to "drain past an error" — handle it once.

stream.finalResult() is not equivalent to the non-streaming form. The non-streaming form posts to /v1/remember (or /v1/search); the streaming form posts to /v1/remember/stream. They share semantics but are different endpoints — server logs and rate-limit buckets are separate.