useAssistantTransport

Overview

- Similar API as `useDataStreamRuntime`.
- Built on an external-store runtime; the external store issues "commands".
- Exactly one run is active at a time (single-flight).
- Runs take queued commands as input and consume an assistant stream that yields state snapshots.
- Every run flushes the entire command queue; a single run processes all pending commands.

Command Scheduling

- When commands are enqueued:
  - If a run is in progress: do not start another; mark that a follow-up run is pending.
  - When the current run ends: if commands were scheduled during the run, start a new run and publish them.
  - If no run is in progress: start a run immediately and flush commands to the server.
- Scheduling uses `queueMicrotask` to coalesce multiple synchronous enqueues into a single run start.

Command Queue

`useCommandQueue({ onQueue() { runManager.schedule(); } })`

- `enqueue(cmd)`: Adds a command to the queue. Calls `onQueue` when transitioning from empty → non-empty (coalesced via `queueMicrotask`).
- `flush(): Command[]`: Returns all queued commands, moves them into `inTransit`, and clears the queue.
- Internal state tracks `inTransit: Command[]` and `queued: Command[]`.

Run Manager

`useRunManager({
async onRun(signal) {
const commands = commandQueue.flush();
setInTransitCommands(commands);

    try {
      const response = await fetch(backendUrl, { signal });
      const stream = response.body
        .pipeThrough(new DataStreamDecoder())
        .pipeThrough(
          new AssistantMessageAccumulator({
            initialMessage: createInitialMessage({
              unstable_state: (state.state as ReadonlyJSONValue) ?? null,
            }),
          }),
        );

      for await (const snapshot of stream) {
        // Clear in-transit commands after the first response chunk.
        // Use a stable empty array to avoid unnecessary re-renders.
        setInTransitCommands(EMPTY_ARRAY);
        setSnapshot(snapshot);
      }
    } catch (error) {
      // Do not restore commands. Surface error to onError for state update.
      callbacks.onError?.({
        error,
        commands: getCurrentInTransitCommands(),
        updateState(updater) {
          setSnapshot((prev) => updater(prev));
        },
      });
    }

},
})`

- `schedule()`: Starts immediately if idle, or schedules at most one follow-up run to start right after the current run.
- `cancel()`: Aborts the active run via `signal` and clears any scheduled follow-up run. Does not restore commands.
- `isRunning: boolean`: Indicates whether a run is currently active (internal to scheduling).
  UI-facing `isRunning` is controlled by the converter output (see Converter).
- On cancellation, invoke `callbacks.onCancel?.({ commands, updateState })` where `commands` contains all pending work at the time of cancel: `[...inTransitCommands, ...queuedCommands]`. Note: after the first snapshot arrives, `inTransitCommands` are cleared to `[]`, so cancels after first byte will not include them.
- RunConfig is not supported for now; any provided run configuration is ignored.

Converter

`useConverter({
  converter,
  agentState,
  queuedCommands,
  inTransitCommands,
})`

- Reactive pattern: do not imperatively set converted state. Maintain an `agentState` snapshot variable (updated via stream), and compute the converted UI state with a memoized converter.
  - Example: `const pending = [...inTransitCommands, ...queuedCommands]; const converted = useMemo(() => converter(agentState, { pendingCommands: pending, isSending }), [agentState, pending, isSending])`
  - `isSending` should be sourced from the run manager’s `isRunning` flag.
- Returns `AssistantTransportState` with `{ messages, isRunning }` derived from inputs via `converter`.
- The converter controls UI `isRunning`. Typical mapping: `isRunning = isSending`. Advanced policies are allowed (e.g., extend running while reconciling, or suppress during background tool results).
- Assistant stream deltas are applied by `AssistantMessageAccumulator`, which emits immutable full-state snapshots; no additional delta handling is required in the converter.

Tool Invocations

`useToolInvocations({
  messages,
  onResult(result) { commandQueue.enqueue(result); },
})`

- Uses a ToolCall differ to diff tool calls across successive snapshots (e.g., ToolCallDiffer).
- When a tool call’s argsText transitions from incomplete → complete and `result` is undefined, synthesize a tool-execution event and enqueue an `add-tool-result` command via `onResult`.
- `onResult` is for frontend function calling (client-side tool calls producing results to enqueue).
- No return value.

External Store Runtime Bridge

`useExternalStoreRuntime({
  isRunning,
  messages,
  onNew(command) { commandQueue.enqueue(command); },
  onCancel() { runManager.cancel(); },
  onAddToolResult(result) { commandQueue.enqueue(result); },
})`

- `onAddToolResult` typically reflects userland-triggered results (e.g., human/tool calling) coming from the external store runtime.

Notes

- Use a stable `EMPTY_ARRAY` when clearing in-transit commands to minimize re-renders via referential equality.
- "Assistant stream" refers to the incremental response stream that yields state snapshots.

Callbacks

- `onError({ error, commands, updateState })`: invoked on network/stream errors. Commands are not restored; use `updateState(state => newState)` to reflect the error in state and/or messages. `commands` reflects the current in-transit commands at the moment of error (often `[]` after the first snapshot).
- `onCancel({ commands, updateState })`: invoked after a cancellation. Commands are not restored. `commands` contains all pending work at cancel time (`inTransitCommands` plus queued). Use `updateState` to reflect cancellation in state and/or messages. The last received snapshot remains committed.

Return Value

- `useAssistantTransport` returns the runtime object from `useExternalStoreRuntime` (e.g., `{ isRunning, messages, ... }`), rather than a custom wrapper shape.