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

Three things to read off that. 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 honestly. 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

This one costs people money, so read it before you build a budget on it. run.usage and run.stepEvents are both computed from the session's entire event list, not from the slice this run appended. Give two runs the same session and the second result reports the first one's model calls as well as its own.

const shared = await app.sessions.create()

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

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

const scope = {
  firstRunStepEvents: firstRun.stepEvents.length,
  secondRunStepEvents: secondRun.stepEvents.length,
  sessionEvents: shared.events.length,
  modelEndsSeenBySecondRun: secondRun.stepEvents.filter(isModelEndEvent).length,
}

scope

secondRunStepEvents equals sessionEvents, and the second run counts two model calls after making one. A fresh session per run — which is what app.run creates when you pass a bare string instead of a session — makes the two readings identical, which is why the trap stays hidden until you keep a conversation going. To attribute cost to one turn of a long conversation, 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 merely 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 exactly 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, which is why 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 actually 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`)
}

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, belong to serving.

Next: The next hour — get this on your machine.