Streaming and cost

Step 1

Two surfaces, one run

Event is a ledger entry. StreamEvent is everything an Event can be, plus two kinds that only ever exist in flight: assistant_delta and thought_delta. Each carries the delta just produced and the text accumulated so far, so a renderer can append or replace, whichever it prefers.

Deltas are never appended to the session. The runner forwards them from the adapter straight to the observer and drops them; the committed assistant event is written once, after the model call finishes. That is why replaying a session gives you the answer and not the typing.

Observation is a hook. onEvent receives every StreamEvent of a run, in order, and returns nothing — guardrails covers the rest of the interface, including the methods that can interrupt. Below, a registered MockAdapter (testing) stands in for the provider the descriptor names (choosing a model), with streamChunks making it emit the reply in pieces.

import { adk } from '@animahealth/adk'
import { openai } from '@animahealth/adk/openai'
import { MockAdapter } from '@animahealth/adk/testing'

// Every cell below re-scripts this adapter, so cells stay deterministic however often you press Run.
const scripted = new MockAdapter({ responses: [] })

const app = adk({ name: 'streaming', adapters: { openai: scripted } })

const narrator = app.agent({
  name: 'narrator',
  model: openai('gpt-5.6-luna'),
  context: [app.context.system('Answer in one sentence.'), app.context.history()],
})

narrator.name

Now watch one run twice: once through onEvent as it happens, once through run.session.events after it is over.

const streamed: string[] = []
const pieces: string[] = []

// `delayMs` buys the run enough wall-clock to be worth measuring in step 2.
scripted.setResponses([
  { text: 'Chunked on the way out.', streamChunks: true, chunkSize: 6, delayMs: 40 },
])

const run = await app.run(narrator, {
  input: 'Say something.',
  hooks: [
    {
      name: 'watch',
      onEvent: (event) => {
        streamed.push(event.type)
        if (event.type === 'assistant_delta') pieces.push(event.delta)
      },
    },
  ],
})

const surfaces = {
  streamed,
  committed: run.session.events.map((event) => event.type),
  pieces,
  reassembled: pieces.join(''),
  finalText: run.output.text,
}

surfaces

The deltas are in streamed and absent from committed. They arrive before model_end, while the committed assistant event lands after it — the ledger writes the answer once the call is accounted for. And committed opens with a user event that streamed never saw: the message was appended to the session before the run began, so it is history, not this run's stream.

Reassembly is pieces.join('') here, but you rarely need it. Each delta already carries text, the accumulation so far, so a renderer can set its buffer to the latest text and never track state of its own.

Step 2

What a model call cost

One model_start/model_end pair is one round-trip to the provider. model_end is where the money is: stepIndex, durationMs, finishReason, and an optional usage with inputTokens, outputTokens, cachedTokens, cacheWriteTokens and reasoningTokens.

The optionality is the important part, and this cell shows it. durationMs is the runner's own stopwatch around the call, so it is always a number. usage and finishReason come off the provider's response, so a scripted model reports neither — and run.usage, which is rolled up from the model_end events that have usage, is undefined for the whole run. Deterministic tests can assert on latency and call counts; they cannot assert on cost. For cost you need step 5.

import { isModelEndEvent } from '@animahealth/adk'

const perCall = run.session.events.filter(isModelEndEvent).map((event) => ({
  stepIndex: event.stepIndex,
  durationMs: event.durationMs,
  finishReason: event.finishReason,
  usage: event.usage,
}))

const accounting = { runUsage: run.usage, perCall }

accounting

When usage is present, run.usage is the roll-up: modelCalls, the token totals, a models array broken down by model name, and an optional cost of { inputCost, outputCost, totalCost, currency: 'USD' }. The model name is the one you wrote in the descriptor, not whatever the endpoint resolved it to.

No provider sends you a price. The ADK derives one from the token counts and a pricing table internal to the package, so a model the table does not know has no cost field and its token counts stand alone — and a price you negotiated yourself is your own multiplication over those totals.

For the streaming case there is a built-in that does the bookkeeping: app.hook.metrics({ onModelResult }) fires per model_end with the agent name, model name, duration and token counts — one callback into whatever your metrics backend is, with no event filtering of your own.

Step 3

Usage is per session, not per run

Read this before you build a budget on it. The ledger belongs to the session, and run.usage is computed from that whole event list — not from the slice this run appended. That is deliberate: a session accrues its cost and its history as it goes. Give two runs the same session and the second result accounts for both.

const shared = await app.sessions.create()

scripted.setResponses([{ text: 'First answer.' }, { text: 'Second answer.' }])

const firstRun = await app.run(narrator, { session: shared, input: 'One?' })

// The ledger is live, not a snapshot, so measure it now — the second run appends to this array.
const afterFirstRun = firstRun.session.events.length

const secondRun = await app.run(narrator, { session: shared, input: 'Two?' })

const scope = {
  afterFirstRun,
  afterSecondRun: secondRun.session.events.length,
  sessionEvents: shared.events.length,
  modelEndsInSession: secondRun.session.events.filter(isModelEndEvent).length,
}

scope

afterSecondRun equals sessionEvents, and the ledger read from the second run's result holds two model calls after that run made one of them. That total is the conversation's, which is usually what you bill it for. output.text follows the same rule: it is the session's last assistant event. A fresh session per run — which is what app.run creates when you pass a bare string instead of a session — makes every reading a single turn's, so the accumulation only shows once you keep a conversation going. To attribute cost to one turn inside a longer one, diff the session's event list around the run, or accumulate from onEvent: a model_end observed during the run is unambiguously the run's own.

Step 4

Stopping a run

app.run does not take an AbortSignal. It returns a handle — a StreamResult, which is at once awaitable and iterable — and that handle has abort(). Calling it aborts the controller the runner made for this run: the signal is already threaded into the model adapter, so an in-flight provider request is cancelled rather than ignored, and the event channel is torn down. Awaiting an aborted run does not return a result. It rejects, with Aborted.

An AbortSignal you already own connects with one line — signal.addEventListener('abort', () => stream.abort()) — which is what app.ask does with its own signal option, the one place in the API where a signal goes in directly. The cell proves both paths, and both time out well inside the three seconds the scripted model was told to take.

const observed: string[] = []

scripted.setResponses([{ text: 'A reply that never arrives.', delayMs: 3000 }])

const inflight = app.run(narrator, {
  input: 'Take your time.',
  hooks: [{ name: 'watch-abort', onEvent: (event) => observed.push(event.type) }],
})

setTimeout(() => inflight.abort(), 50)

let stopped = 'the run finished first'
try {
  await inflight
} catch (error) {
  stopped = (error as Error).message
}

// The same stop, reached through a signal you own.
scripted.setResponses([{ text: 'Also never arrives.', delayMs: 3000 }])

const controller = new AbortController()
setTimeout(() => controller.abort(), 50)

let asked = 'the ask finished first'
try {
  await app.ask('Anything.', { model: openai('gpt-5.6-luna'), signal: controller.signal })
} catch (error) {
  asked = (error as Error).message
}

const stopping = { stopped, asked, observed }

stopping

observed is where the run got to before it was cut off — the events that had already been produced, and nothing after. Abort is not the only stop, and the others are not interchangeable. app.run(…, { timeout }) races the whole stream and rejects with Timeout after Nms; an agent's maxSteps ends the run with status max_steps instead of throwing; a tool's own timeout is a recoverable phase error. Guardrails is where those live.

Step 5

Real tokens, real money

Everything above ran against a script, so no number in it was a price. This cell calls OpenAI with the key from the box at the top of the page, on a fresh app so the scripted adapter is out of the way. It times the first delta — the number your users feel — and then reads the usage the provider reported.

const liveApp = adk({ name: 'streaming-live' })

const scribe = liveApp.agent({
  name: 'scribe',
  model: openai('gpt-5.6-luna'),
  context: [liveApp.context.system('Answer in two short sentences.'), liveApp.context.history()],
})

const firstDelta: number[] = []
const startedAt = Date.now()

const answered = await liveApp.run(scribe, {
  input: 'What does a hospital ward do at night?',
  hooks: [
    {
      name: 'time-to-first-token',
      onEvent: (event) => {
        if (event.type === 'assistant_delta' && firstDelta.length === 0) {
          firstDelta.push(Date.now() - startedAt)
        }
      },
    },
  ],
})

const measured = {
  msToFirstDelta: firstDelta[0],
  text: answered.output.text,
  usage: answered.usage,
}

measured

msToFirstDelta against durationMs on the model_end event is the streaming argument in two numbers: the reader starts reading at the first, the run ends at the second. The OpenAI adapter always streams, so those deltas arrive whether or not you asked for them — streaming is not a mode you switch on, it is what the transport does, and onEvent is where you decide to care.

Edit the question and run it again to watch the tokens and the cost move. A longer question moves inputTokens; a longer answer moves outputTokens, which on this model is priced six times higher.

Step 6

Getting the stream off the box

A hook is in-process. To reach a browser, the run has to be iterated, and app.handler.turn is the surface for it: it returns a handle that is both an async iterable of stream events and a promise of the committed TurnResult. Driving that handle by hand — and the trap in not doing so — belongs to serving; what is left here is one line down the wire.

// Inside the loop over the turn's stream events.
if (event.type === 'assistant_delta') {
  response.write(`data: ${JSON.stringify({ delta: event.delta })}\n\n`)
}

Forwarding only the deltas is deliberate. The same stream carries a state_change event for every state write, patient and practice scopes included. To send more than text, drop each event for which mayLeaveProcess(event) returns false before it goes down the wire.

For a front end that speaks the AG-UI protocol, app.handler.agui is the same run translated: it hands back an AsyncIterable of AG-UI events, mapping deltas to text-message content frames and tool calls to their protocol equivalents, so a compatible client renders the stream without knowing the ADK exists. Both handlers, and the sessions they commit into, live in serving.

Next: Install and setup — get this on your machine.