Serving it: one turn per request

Step 1

One request, one turn

app.handler carries turn, rest, agui and voice. The first three are one thing wearing three coats: turn is the atom — session in, one run, commit, result out — and it returns a stream. rest drains that stream and returns a JSON body; agui translates it into the AG-UI protocol. Both call turn.

Each takes a config once and hands back a function you call per request. The config is the deployment's business — which agent, which store, which hooks, how much to report. The per-request part is two fields: a sessionId and an input.

interface HandlerInput {
  sessionId?: string
  input: Input // { message?, tools?, state?, initialState? }
}

// app.handler.* fills in the app's name, schema and store, and prepends the app's own
// hooks and error handlers to these.
interface HandlerConfig {
  agent: Runnable
  schema?: StateSchema
  sessionService?: SessionService
  hooks?: Hook[]
  errorHandlers?: ErrorHandler[]
  timeout?: number
  response?: { events?: boolean; usage?: boolean; state?: boolean }
}

Omit the sessionId and the handler mints one. Supply it and the handler loads that row, or starts it if there is nothing there yet. Ids are normalised on the way in, so thread-42 and session_thread-42 are the same thread.

Here is the agent the rest of the page serves. The model is stood down by a hook — beforeAgent returning a string replaces the run's output before any model is reached (see Guardrails and recovery). Everything under it is the shipped request path.

import { adk } from '@animahealth/adk'
import { openai } from '@animahealth/adk/openai'
import { z } from 'zod'

const app = adk({ name: 'bookings' })

const lookupStay = app.tool({
  name: 'lookup_stay',
  description: 'Look up a booking by its reference',
  schema: z.object({ ref: z.string() }),
  execute: (ctx) => ({ ref: ctx.args.ref, nights: 3 }),
})

const concierge = app.agent({
  name: 'concierge',
  model: openai('gpt-5.6-luna'),
  context: [app.context.system('Answer booking questions.'), app.context.history()],
  tools: [lookupStay],
})

// The model, stood down: this hook answers before the model call, so every handler on this page
// runs with no key. Delete it and the same code talks to OpenAI.
const scripted = (text: string) => [{ beforeAgent: () => text }]

concierge.name

Step 2

The REST handler

app.handler.rest(config) returns (input) => Promise<RestResponse>. It runs the turn to completion and hands back a plain object you can serialise. That is the whole surface: no framework, no router, no middleware contract.

const askConcierge = app.handler.rest({
  agent: concierge,
  hooks: scripted('Three nights, checking out Friday.'),
  response: { state: true },
})

const reply = await askConcierge({
  sessionId: 'thread-42',
  input: { message: 'How long is my stay?', state: { channel: 'sms' } },
})

reply

Three fields are always there — sessionId, status, output — and the rest are earned. yieldedTools appears when the turn yielded, error when it failed, warning when the commit was messy. The other three are opt-in, because a chat client and an audit endpoint want different amounts of the turn:

One sharp edge on the last of those: events is typed Event[], but a stream also carries assistant_delta and thought_delta, which are not ledger events. Narrow on type before you trust an entry, or read the ledger from the session instead.

Now the reason this is worth anything. Call the same handler again on the same sessionId: the second turn resumes a conversation nobody kept in memory, because the first one committed before it returned.

const secondReply = await askConcierge({
  sessionId: 'thread-42',
  input: { message: 'And the checkout time?' },
})

const thread = await app.sessions.get('thread-42')

const durable = {
  turns: [reply.status, secondReply.status],
  eventsOnDisk: thread?.events.length,
  version: thread?.version,
  ledger: thread?.events.map((event) => event.type),
}

durable

handler.rest and handler.turn commit after every turn, so serving an agent needs no persistence code of your own — which store the rows land in, and what a bare app.run does instead, are Where a sleeping agent lives.

Behind an HTTP server it stays this small. The handler is built once, at boot; the request body supplies the two fields.

import { createServer } from 'node:http'

import { adk } from '@animahealth/adk'
import { postgresStore } from '@animahealth/adk/stores/postgres'

const bookings = adk({
  name: 'bookings',
  store: postgresStore({ connectionString: process.env.DATABASE_URL }),
})

const ask = bookings.handler.rest({ agent: concierge, response: { usage: true } })

createServer(async (req, res) => {
  const chunks: Buffer[] = []
  for await (const chunk of req) chunks.push(chunk as Buffer)
  const body = JSON.parse(Buffer.concat(chunks).toString())

  const turn = await ask({ sessionId: body.sessionId, input: { message: body.message } })

  res.writeHead(turn.status === 'error' ? 500 : 200, { 'content-type': 'application/json' })
  res.end(JSON.stringify(turn))
}).listen(3000)

Step 3

A turn that stops

A yielding tool crossing a request boundary needs no extra protocol. The turn comes back yielded_tool with the calls it is waiting on; a later request carries the answers in input.tools. That is the pause-and-resume of Stopping to ask, with HTTP in the gap instead of a runTest step.

// POST /turns { "sessionId": "y1", "message": "book me in" }
const paused = await ask({ sessionId: 'y1', input: { message: 'book me in' } })

// paused.status       → 'yielded_tool'
// paused.output.items → []                       (nothing was said; the turn stopped)
// paused.yieldedTools → [{ callId: 'call_125f…', name: 'ask_guest', args: { q: 'Which night?' } }]

// POST /turns { "sessionId": "y1", "tools": [{ "callId": "call_125f…", "input": { … } }] }
if (paused.status === 'yielded_tool' && paused.yieldedTools) {
  const [pending] = paused.yieldedTools

  const resumed = await ask({
    sessionId: paused.sessionId,
    input: { tools: [{ callId: pending.callId, input: { answer: 'Friday' } }] },
  })

  // resumed.status      → 'completed'
  // resumed.output.text → 'Booked for Friday.'
}

Note the guard. RunResult is a discriminated union, so status === 'yielded_tool' is enough there. RestResponse is a flat object with optional fields, so check yieldedTools yourself.

Between those two requests the agent is rows. Nothing is held open, so the process that asked the question and the process that receives the answer need not be the same one.

Step 4

The stream underneath

app.handler.turn(config) returns (input) => StreamResult<TurnResult> — an async iterable of stream events that returns the turn result. Take it when you want the tokens as they arrive: a server-sent-event endpoint, a websocket, a CLI.

Two things to know before you iterate. The stream is lazy — it starts when something awaits or iterates it, not when you call the handler. And the TurnResult is the generator's return value, which a for await loop throws away, so drive the iterator by hand when you need it.

const streamTurn = app.handler.turn({
  agent: concierge,
  hooks: scripted('Checkout is at eleven.'),
})

const live = streamTurn({ sessionId: 'thread-42', input: { message: 'Remind me again?' } })

const streamedTypes: string[] = []
const iterator = live[Symbol.asyncIterator]()
let step = await iterator.next()
while (!step.done) {
  streamedTypes.push(step.value.type)
  step = await iterator.next()
}

const outcome = {
  invocationId: live.invocationId,
  streamed: streamedTypes,
  status: step.value.status,
  commitStatus: step.value.commitStatus,
  sessionId: step.value.sessionId,
}

outcome

Sparse, because the model on this page never runs: a real turn also streams model_start, assistant_delta, tool_call, model_end. What matters is the pair around them — invocation_start opens the turn, and after invocation_end the result carries a commitStatus the stream never mentioned. That word is the next section.

live.abort() stops the run. An aborted turn does not commit at all: its commitStatus is undefined and its events stay where they were. So does a turn that threw — the result comes back status: 'error' with the message, and nothing is written.

Step 5

The AG-UI handler

AG-UI is a protocol for agent front-ends: a fixed vocabulary of events a generic client can render without knowing what framework produced them. app.handler.agui(config) returns (input) => AsyncIterable<AGUIEvent>, ready to write down an SSE connection.

import { createServer } from 'node:http'

const ui = bookings.handler.agui({ agent: concierge })

createServer(async (req, res) => {
  const url = new URL(req.url ?? '/', 'http://localhost')
  res.writeHead(200, { 'content-type': 'text/event-stream', 'cache-control': 'no-cache' })

  for await (const event of ui({
    sessionId: url.searchParams.get('session') ?? undefined,
    input: { message: url.searchParams.get('message') ?? '' },
  })) {
    res.write(`data: ${JSON.stringify(event)}\n\n`)
  }

  res.end()
}).listen(3001)

Every stream is bracketed the same way. RUN_STARTED carries the session as threadId and the invocation as runId. Then a STATE_SNAPSHOT, which is empty — state reaches the client as STATE_DELTA patches as it changes, not as an opening picture. Then the turn's events, translated one by one:

A text message therefore exists only if the model streamed deltas. The end of the run says which way it went: RUN_FINISHED carrying result: { commitStatus } when the turn finished, a CUSTOM named RUN_INTERRUPTED when it yielded (the pending callId and tool name ride in its value), and RUN_ERROR — with no RUN_FINISHED — when it failed or was aborted.

handler.agui is a thin loop over handler.turn, and the translator it uses is exported. AgUIAdapter takes options the handler does not forward — step events, reasoning off, raw events attached, a transformer per yielding tool — so when you want those, or a protocol that is not AG-UI at all, write the loop yourself.

import { AgUIAdapter } from '@animahealth/adk/agui'

const runTurn = bookings.handler.turn({ agent: concierge })

async function* protocol(sessionId: string, message: string) {
  const live = runTurn({ sessionId, input: { message } })
  const adapter = new AgUIAdapter(sessionId, live.invocationId, { includeSteps: true })

  yield adapter.runStarted()

  const iterator = live[Symbol.asyncIterator]()
  let step = await iterator.next()
  while (!step.done) {
    yield* adapter.transform(step.value)
    step = await iterator.next()
  }

  yield adapter.runFinished({ commitStatus: step.value.commitStatus })
}

Step 6

Four ways a turn ends up on disk

Sessions are guarded by a version, not a lock — that is the store's bargain, argued in Where a sleeping agent lives — so a handler must decide what to do when a commit is refused, and commitStatus on the turn result reports which way it decided:

Underneath are the two calls the handler makes for you, and you can make them yourself. Load one session twice — two requests, two processes, same row — and watch the second commit be refused.

await app.sessions.create({ sessionId: 'thread-shared' })

const replicaOne = await app.sessions.get('thread-shared')
const replicaTwo = await app.sessions.get('thread-shared')
if (!replicaOne || !replicaTwo) throw new Error('no session thread-shared')

const loadedAt = [replicaOne.version, replicaTwo.version]

replicaOne.input.message('from the first request')
replicaTwo.input.message('from the second request')

const won = await app.sessions.commit(replicaOne)
const lost = await app.sessions.commit(replicaTwo)
const rescued = await app.sessions.merge(replicaTwo)
const settled = await app.sessions.get('thread-shared')

const conflict = {
  loadedAt,
  won,
  lost,
  rescued,
  version: settled?.version,
  ledger: settled?.events.map((event) => (event.type === 'user' ? event.text : event.type)),
}

conflict

The refusal is data, not an exception: { ok: false, conflict: true, currentVersion }, and it tells you the version you actually needed. Nothing was written — not the events, not the scoped state riding with them. merge then re-reads the head and appends this writer's new events onto it, so both messages survive. Losing a race costs an extra round trip, not a turn.

Step 7

Two turns, one session

Now the same collision through the handlers, with no conflict simulated. Two turns are started against one sessionId. In the first pair a slow turn is overtaken by a message that arrives while it is still thinking. In the second, both arrive at once.

const slowConcierge = app.handler.turn({
  agent: concierge,
  hooks: [
    {
      beforeAgent: async () => {
        await new Promise((resolve) => setTimeout(resolve, 150))
        return 'the slow answer'
      },
    },
  ],
})
const fastConcierge = app.handler.turn({ agent: concierge, hooks: scripted('the fast answer') })

// Promise.resolve starts a lazy turn stream, so the second request lands mid-flight.
const slowTurn = Promise.resolve(
  slowConcierge({ sessionId: 'race', input: { message: 'first question' } }),
)
await new Promise((resolve) => setTimeout(resolve, 30))
const fastTurn = Promise.resolve(
  fastConcierge({ sessionId: 'race', input: { message: 'second question' } }),
)

const [slowOutcome, fastOutcome] = await Promise.all([slowTurn, fastTurn])
const raced = await app.sessions.get('race')

const together = app.handler.turn({ agent: concierge, hooks: scripted('either answer') })
const [firstOutcome, secondOutcome] = await Promise.all([
  Promise.resolve(together({ sessionId: 'tie', input: { message: 'A' } })),
  Promise.resolve(together({ sessionId: 'tie', input: { message: 'B' } })),
])
const tied = await app.sessions.get('tie')

const said = (session: typeof raced) =>
  session?.events.map((event) => (event.type === 'user' ? event.text : event.type))

const race = {
  overtaken: {
    slow: [slowOutcome.status, slowOutcome.commitStatus],
    fast: [fastOutcome.status, fastOutcome.commitStatus],
    ledger: said(raced),
  },
  together: {
    first: [firstOutcome.status, firstOutcome.commitStatus],
    second: [secondOutcome.status, secondOutcome.commitStatus],
    ledger: said(tied),
  },
}

race

Read the first ledger. The slow turn ran to completed and produced an answer, and none of it is there — not the reply, not even the question that started it. The rule is a timestamp comparison: on a rejected commit the handler re-reads the row, and if it holds a user message newer than this turn's own, the whole uncommitted slice is dropped. The guest moved on, so the answer to the old question would arrive out of order.

The second ledger is the other outcome: nothing overtook anything, so the loser merged and both exchanges are on the row, in commit order. One question separates the two — did a newer user message land while this turn was running.