Conversations

Compact long conversations

import { agent } from 'experimental-a2/ai/server'

export const compactedAssistant = agent({
  name: 'compacted-agent',
  model: 'openai/gpt-5.6-terra',
  harness: {
    type: 'ai-sdk',
    // optional: compaction is automatic for supported Gateway models
    compaction: {
      thresholdTokens: 80_000,
      instructions: 'Preserve exact IDs.',
    },
  },
})

Compaction belongs to the AI SDK harness, and is automatic for supported Gateway models. When the estimated prompt crosses the threshold, A2 sends one summary request to the active model with the same instructions, tools, and provider settings, keeping the prefix cacheable. Later requests get the summary plus everything after it; the raw log and state.messages stay complete, and the summary request never appears there as a turn. Successful compaction is permanent across failure, interruption, retry, and recovery, and a failed summary leaves the previous context intact.

The default threshold is 75% of the context window from the Gateway model catalog, lowered for a larger generation.maxOutputTokens; it limits input context, not output. thresholdTokens overrides it, instructions adds guidance to the summary prompt, and false turns off compaction and the catalog lookup, keeping recorded summaries. A resolver can choose per generation, for example from session.state(settingsReducer). Custom shouldCompact(context) and compact(context) callbacks receive the typed messages and the model-ready modelMessages, summary included.

An AI SDK global provider that resolves model IDs to models outside AI Gateway, and Gateway fallback model lists, need an explicit thresholdTokens (for a fallback list, one that fits every model it may use). A custom generate keeps compaction off unless you configure it. Set a threshold too when importing a large conversation: a session does not compact before the catalog lookup lands, so an oversized first prompt is sent as-is, never truncated, and fails with the provider's error. Provider-executed tools, structured output, and forced tool choices skip automatic compaction, and configuring it for them throws; use custom callbacks.

The first generation with each Gateway model in a session records ai.model.metadata.requested; a separate handler fetches the catalog and records ai.model.metadata.resolved, without delaying or failing the generation. state.modelMetadata holds each model's limits and status: pending, resolved, or unavailable (absent from the catalog). A lookup that exhausted its retries stays pending without a new lookup each turn, so inspect that handler's failure or set a threshold. The estimate is serialized UTF-8 bytes divided by four, calibrated by the previous same-model call's usage. It is not a tokenizer, makes no remote calls, and does not precisely measure new image or audio tokens.

Compaction waits until earlier tool calls have final results and all A2-owned tool work has settled, including work from stopped or superseded turns, with one extra state read before each candidate compaction. Dead-lettered tool work defers it until repair or session close. Queued messages stay queued. The summary request runs none of your generation callbacks, stream transforms, or tool hooks and never executes local tools; an empty summary or a summary tool call fails the generation, as do provider context-limit errors. See agent() for every option.

Show compaction in the conversation

'use client'
import type { UIMessage } from 'ai'
import { isCompactionPart, type TranscriptMessage } from 'experimental-a2/ai'

const compactionLabel = {
  running: 'Compacting…',
  completed: 'Context compacted',
  failed: 'Compaction did not complete',
  interrupted: 'Compaction stopped',
}

export function MessageView({
  message,
}: {
  message: TranscriptMessage<UIMessage>
}) {
  return (
    <article>
      {message.parts.map((part, index) => {
        if (isCompactionPart(part))
          return (
            <details key={index}>
              <summary>{compactionLabel[part.data.status]}</summary>
              {part.data.summary ? <p>{part.data.summary}</p> : null}
            </details>
          )
        return part.type === 'text' ? <p key={index}>{part.text}</p> : null
      })}
    </article>
  )
}

The reducer marks each compaction with a data-a2-compaction part in the assistant response. Its data holds a status (running, then completed, failed, or interrupted) and, when known, inputTokens, thresholdTokens, summary, and usage. The part lives in state, so it survives server rendering, reloads, and other tabs, and never enters model context. Because state.messages is TranscriptMessage<M>[], isCompactionPart() narrows even when your message type declares its own data parts. state.compaction holds the current or latest compaction, and indicator is compacting while one runs. See isCompactionPart() for how the part settles.

Store assistant metadata

import type { LanguageModelUsage, UIMessage } from 'ai'
import { agentContract } from 'experimental-a2/ai'
import { agent } from 'experimental-a2/ai/server'

export type MetadataAssistantMessage = UIMessage<{
  status: 'writing' | 'complete'
  usage?: LanguageModelUsage
}>

export const metadataAssistant = agent({
  contract: agentContract<MetadataAssistantMessage>({
    name: 'metadata-assistant',
  }),
  model: 'openai/gpt-5.6-terra',
  harness: {
    type: 'ai-sdk',
    messageMetadata: ({ part }) => {
      if (part.type === 'start') return { status: 'writing' as const }
      if (part.type === 'finish') {
        return { status: 'complete' as const, usage: part.totalUsage }
      }
      return undefined
    },
  },
})

A narrower UIMessage type, carried by agentContract(), keeps generation status and usage on the assistant message. The synchronous harness.messageMetadata callback gets the generate context plus the current TextStreamPart as part. What it returns for start and finish is logged with the stream and shallow-merged onto the assistant message, so the example shows writing at once and keeps final usage. useAgent<typeof metadataAssistant>() types messages with it.

Add your own events and state

Apps carry more than the conversation: a plan, a selected ticket, a chosen model. Put those events and their state in a shared agentContract(). It takes a core contract()'s events, and its reducer takes the same name, initialState, and fold:

import { z } from 'zod'
import { agentContract } from 'experimental-a2/ai'

const noFlags: string[] = []

export const supportDesk = agentContract({
  name: 'support-desk',
  events: {
    'plan.changed': z.object({ plan: z.enum(['free', 'pro']) }),
    'ticket.flagged': z.object({ ticketId: z.string() }),
  },
  reducer: {
    name: 'support-desk.v1',
    initialState: { plan: 'free' as 'free' | 'pro', flagged: noFlags },
    fold: (state, event) =>
      event.type === 'plan.changed'
        ? { ...state, plan: event.payload.plan }
        : event.type === 'ticket.flagged'
          ? { ...state, flagged: [...state.flagged, event.payload.ticketId] }
          : state,
  },
})

Keep this module isomorphic. fold sees your state and every event, never the transcript. The session's state is AIState merged with yours, so a key that shadows one of AIState's is a type error. The reducer's name is its snapshot identity: rename it when fold changes.

Serve it with agent({ contract }). Every resolver receives your state as state:

import { agent } from 'experimental-a2/ai/server'
import { supportDesk } from '../support-desk'

export const supportDeskAgent = agent({
  contract: supportDesk,
  model: ({ state }) =>
    state.plan === 'pro'
      ? { model: 'openai/gpt-5.6-terra', reasoning: 'high' }
      : { model: 'openai/gpt-5.6-luna', speed: 'fast' },
  handlers: {
    'ticket.flagged': async ({ event }) => {
      // your side effect; event.id makes a stable idempotency key:
      // await notifyOnCall(event.payload.ticketId, { idempotencyKey: event.id })
    },
  },
})

state comes from the snapshot the generation already captured, so it costs no extra read. Each model step resolves again, so a change during a turn applies from the next step. A resolver that throws fails the turn with its message, like a provider error; retry() runs it again.

In the browser, pass the contract to useAgent(). This import is real, not type-only, because the browser folds your state too:

'use client'
import { useAgent } from 'experimental-a2/ai/react'
import { supportDesk } from '@/support-desk'

export function Desk({ id }: { id: string }) {
  const { state, push, sendMessage } = useAgent({
    api: '/api/support',
    id,
    contract: supportDesk,
  })

  return (
    <section>
      <p>
        {state.plan} plan, {state.flagged.length} flagged tickets
      </p>
      <button
        onClick={() =>
          void push({ type: 'ticket.flagged', payload: { ticketId: 'T-1' } })
        }
      >
        Flag T-1
      </button>
      <button onClick={() => void sendMessage('Summarize the flagged tickets.')}>
        Ask
      </button>
    </section>
  )
}

Browsers push a shared contract's application events like any A2 client. The agent refuses only its own server-authored ai.* events; the route's authorize decides the rest. Here the plan comes from your billing code, not the browser:

import { supportDeskAgent } from '@/server/support-desk'

function handle(request: Request): Promise<Response> {
  // here's where you'd do auth, or any other checks:
  return supportDeskAgent.fetch(request, {
    authorize: (operation) =>
      operation.type !== 'push' ||
      operation.events.every((event) => event.type !== 'plan.changed'),
  })
}

export const GET = handle
export const POST = handle

Server appends, such as supportDeskAgent.session(id).append({ type: 'plan.changed', payload }), are trusted and skip authorize. push() also accepts any inputs builder, and handlerContext(supportDesk) types application events inside tools.

The contract-free client, createAgentClient() or useAgent({ api | client }) with a type-only import, folds only the built-in reducer. It accepts agent({ name }), or agent({ contract }) with an agentContract() that has no reducer; the types enforce this through AI_STATE_REDUCER.

When the agent belongs inside something larger, such as a shared chat room, keep two sessions. The room's handler appends each message the agent should see to the agent's session with inputs.message(). Event IDs are unique across a store's sessions, so derive the message ID from the room event's ID with idempotentId(); a retried handler then converges on the same message. A tool the agent calls to speak appends to the room with an idempotentId() of its toolCallId. Neither fold reads the other's state.

Replace generation, not durability

import { convertToModelMessages, streamText } from 'ai'
import { agent } from 'experimental-a2/ai/server'

export const customGeneration = agent({
  name: 'custom-generation',
  model: 'openai/gpt-5.6-terra',
  harness: {
    type: 'ai-sdk',
    generate: async ({ event, messages, model, instructions, signal }) => {
      const result = streamText({
        model,
        messages: await convertToModelMessages(messages),
        abortSignal: signal,
        ...(instructions === undefined ? {} : { instructions }),
      })

      return result.toUIMessageStream({
        originalMessages: messages,
        generateMessageId: () => event.payload.responseMessageId,
      })
    },
  },
})

harness.generate replaces only the model-facing step, for fixtures, provider routing, or a different AI SDK assembly. It receives the resolver context (with messages already compacted), the resolved settings, modelMessages, and the step's generationId, and returns exactly one model step as a ReadableStream<UIMessageChunk> that writes the assistant message event.payload.responseMessageId. A tool-aware replacement sends tool definitions without running local execute functions. A2 still owns durable progress, tool execution, approval, continuation, interruption, and failure. messageMetadata applies only to the default generator, so a custom generate emits its own metadata.

Checkpoints and recovery reads

Samples from here on import assistant, the agent from Agents:

import type { UIMessage } from 'ai'
import type { TranscriptMessage } from 'experimental-a2/ai'
import { assistant } from './assistant'

export async function readMessages(
  sessionId: string,
): Promise<TranscriptMessage<UIMessage>[]> {
  const snapshot = await assistant.session(sessionId).state()
  return snapshot.state.messages
}

session(id).state() returns AIState and its log index, ready for server code or client hydration. Tool inputs, outputs, and provider metadata live in message parts; state.tools keeps only identities and execution status. reduceAIState() is the reducer's pure transition over confirmed AIState, for projecting raw event histories.

Model, instruction, compaction, and custom-generation callbacks receive prompt messages, the triggering request, and your contract's state, never a raw history. Read any other reducer with session.state(yourReducer), at the captured prompt boundary unless you pass through. Recovery reads at fixed boundaries too: a tool recovered in a fresh process sees its original prompt, and A2 rebuilds a missing or invalid reducer snapshot from the log.

While tool arguments stream, input-streaming parts carry a best-effort partial JSON input, so a command string can grow before its closing quote. It is a display draft, not validated input; tool-input-available supplies the completed input.

Delivery semantics

Model steps and tools run inside at-least-once A2 handlers. A successful completion commits atomically with its terminal events, and deterministic IDs make scheduling and lifecycle appends effectively once. An incomplete model attempt is marked superseded before its replacement starts. An expired claim or superseded attempt keeps its abort reason through model, resolver, compaction-policy, and tool calls, and retries at the cost of one failed attempt; an explicit interruption stays terminal. A retried tool can stream different preliminary output under the same tool-call ID.

Provider calls and external side effects sit outside that transaction, so recovery may repeat a call whose effect landed before its handler committed. Give side-effecting tools their own idempotency, normally keyed by the AI SDK toolCallId. A fast tool can also start before its model step fails; the failure blocks the continuation but cannot undo the effect.

For a tool whose effect must not repeat and has no idempotency key, such as a shell command, set toolRecovery: { yourTool: 'report' }. When a crash, a timeout, Stop, or Pause now may have cut a call short, A2 does not run it again, even on resume; the model gets an error result saying the call was interrupted and may have partly run. A call that failed before reaching the tool still retries, and so does one where A2 itself fails while or just after the tool runs. Every other tool uses the default, 'rerun'. See agent() for the exact cases.

Configure a production scheduler as for any A2 server. It recovers an incomplete generation after the original serverless invocation disappears.

Inbox and turn controls

'use client'
import { useAgent } from 'experimental-a2/ai/react'
import type { assistant } from '@/server/assistant'

export function Queue({ sessionId }: { sessionId: string }) {
  const { state, queue, steer, pause, resume } = useAgent<typeof assistant>({
    api: '/api/agent',
    id: sessionId,
  })

  return (
    <section>
      <ol aria-label="Queued messages">
        {state.inbox.items.map((item) => (
          <li key={item.id}>
            {item.message.parts.map((part, index) =>
              part.type === 'text' ? <span key={index}>{part.text}</span> : null,
            )}
            <button onClick={() => void queue.sendNow(item.id)}>Send now</button>
            <button
              onClick={() =>
                void queue.move({ inputId: item.id, beforeId: null })
              }
            >
              Move to end
            </button>
            <button onClick={() => void queue.remove(item.id)}>Remove</button>
          </li>
        ))}
      </ol>
      <button onClick={() => void steer('Check the billing service first.')}>
        Steer
      </button>
      {state.inbox.paused ? (
        <button onClick={() => void resume()}>Resume</button>
      ) : (
        <button onClick={() => void pause({ when: 'after-turn' })}>
          Pause after this turn
        </button>
      )}
    </section>
  )
}

sendMessage() puts a message in the durable inbox, state.inbox.items. The controller admits one input at a time and adds it to state.messages; pending inputs never enter the model prompt, and an edit after admission is rejected. The log stores a message once, in the command that sent it or the edit that last changed it. state.active is the logical turn: its ID, its input, and phase. The turn ID holds across model steps, tools, and a pause. Stop ends the turn and admits the next input when admission is running.

queue.edit(message) replaces a pending input's content; edits are last-write-wins in log order. An edit keeps the message's identity and role; changing the role returns role-mismatch. queue.move() takes beforeId: null for the end. queue.remove() keeps the original submission in the log, so replay cannot recreate the prompt. Removing an input that is no longer pending (admitted, removed, or never queued) returns not-pending, as does an edit that loses to admission; a Stop aimed at a finished turn returns stale-turn.

steer() puts new input right after the turn's earlier steering, or first when there is none, so steering hands off in the order it was sent. If the observed turn is still active, it finishes its current model response and that response's tools, including authorized deferred provider results, then hands off before another model step. Everything steering that turn enters together as one new turn: the earlier messages as context and the last one generating, so the model answers them once. Output produced after the click stays, and unanswered approval and input requests close, so the unapproved tool never runs; a denied approval never blocks the handoff. With no active turn, steer() sends like sendMessage().

queue.sendNow() makes the same handoff with a queued message, after the steering sent before it, keeping its ID and latest accepted content and promoting passive user input to generation. Non-user context returns not-user-input, and a stale observed turn returns stale-turn without changing the queue. Send now uses the accepted content, so save editor text first. steer() and queue.sendNow() both preserve pause: a paused queue keeps the selected input first until Resume. Stop remains the immediate interruption.

A queued item's optional afterStepOf records the observed turn for steering. Editing keeps it, removing the item cancels it, and queue order decides which inputs are admitted at the handoff: the run at the front that steers the same turn. See inputs for the full handoff rules.

Input identity

New-input commands from inputs.message(), inputs.seed(), and inputs.steer() use the message ID as their event ID, and hand-written commands must too. Caller IDs cannot start with the reserved a2.ai: prefix and must be unique across the store; include the contract and session when deriving them with idempotentId(). An identical submission replays without new work, even after removal or compaction. Reusing an ID with different content, action, or session rejects with PARTIAL_DUPLICATE_BATCH: use queue.edit() to change a pending input and a fresh ID for a new one.

Optimistic conversation display

Every client folding the AI reducer, an extended agentContract() included, applies pending commands to state with no extra I/O, so render the conversation and queue from state. An idle, unpaused send appears in state.messages at once with state.starting: true: a turn awaiting acceptance, not a running model. Later sends, and every send while paused, wait in state.inbox.items; steering, and Send now aimed at an active turn, appear in state.messages after the current response instead. Another client's accepted turn can move a speculative send back into the queue, while steering stays visible as a message. A pending Stop clears state.active at once. Close an editor as soon as its save is pushed, and keep the draft if the push fails. See Optimistic AI client state.

Persistence and application are separate

A push confirms that the command is persisted; ai.control.decided confirms its application, with commandId (the submitted event's ID) and an outcome of applied or rejected. A rejection carries a reason such as not-pending or stale-turn. state.receipt is the most recent user command receipt. To correlate several in flight, each action resolves to its persisted command events, whose id is the receipt's commandId, or to an empty array when there is nothing to address, such as Stop with no active turn. A rejection is a normal result and spends no handler retry.

Pause and resume

pause({ when: 'after-turn' }) closes inbox admission and lets the active turn finish. pause({ when: 'now' }) also cancels current model and tool work; state.active.phase is pausing while old work settles, then paused. resume() resumes retained work before admitting another input.

A turn paused with Pause now keeps its steering until Resume. Every steer() or queue.sendNow() sent while it's paused targets it, and on Resume they enter together as one turn; the paused turn's own work does not resume.

Resume starts a fresh model request from durable state; it does not resume the provider's old stream or undo completed tool effects. While paused, accepted tool calls are retained with their outcomes recorded, and the turn holds no model or tool claim. A cancelled tool without a terminal result can run again after resume, so use the stable toolCallId for external idempotency.