@nifrajs/ag-ui

Mount a nifra agent as an AG-UI (Agent-User Interaction protocol) endpoint: one POST accepting RunAgentInput, a stream of AG-UI events out over SSE - run lifecycle, step and tool-call events projected from the runner's step evidence, text message events for the output, and a typed continuation for human-in-the-loop resume.

A protocol bridge over @nifrajs/agent - the request body reuses core's single bounded, prototype-guarded trust boundary, and the model, durable state store, and approval transport are injected per request through the same ports factory mountAgent uses. Dependency-free beyond the two nifra peers.

Install

bun add @nifrajs/ag-ui @nifrajs/agent

Mount

import { server } from "@nifrajs/core"
import { mountAgUI } from "@nifrajs/ag-ui"
import { agent } from "./agent" // an AgentDefinition

const app = server()
mountAgUI(app, {
  agent,
  ports: (c) => ({
    model: myModelPort,        // scope to the request subject
    capabilities: ["search.read"],
    state: myStateStore,       // required for resume
  }),
})
// POST /agui - RunAgentInput in, AG-UI events out (SSE)

Event mapping

| Runner | AG-UI events | | --- | --- | | run starts | RUN_STARTED, then CUSTOM { name: "nifra.turn" } announcing the turn id | | tool evidence | TOOL_CALL_START / TOOL_CALL_END, then TOOL_CALL_RESULT whose content is the token-only outcome { "outcome": "committed" \| "failed" \| "denied", "code"? } | | model, approval, budget, state evidence | STEP_STARTED / STEP_FINISHED | | model text deltas | live TEXT_MESSAGE_START / _CONTENT (see streaming below) | | model reasoning deltas | REASONING_START / REASONING_MESSAGE_START / _CONTENT / _END / REASONING_END | | model tool-args deltas | provisional TOOL_CALL_START + TOOL_CALL_ARGS, closed by the call's tool evidence | | shared state | STATE_SNAPSHOT on the seed and STATE_DELTA (RFC 6902 ops) per patch (see below) | | model usage deltas | summed per (provider, model) into the usage: TokenUsage[] array on RUN_FINISHED | | completed output | TEXT_MESSAGE_START / _CONTENT / _END (unless text was streamed), an optional MESSAGES_SNAPSHOT (see below), then RUN_FINISHED with result and outcome: { "type": "success" } | | completed with a typed error | RUN_ERROR | | suspended | CUSTOM { name: "nifra.pending" } with the continuation, then RUN_FINISHED with outcome: { "type": "interrupt", "interrupts": [...] } |

Evidence-derived events carry the evidence timestamp. The runtime is token-only by design - TOOL_CALL_RESULT reports the outcome and error code, never the tool's payload.

Agent input is forwardedProps.input when present, otherwise the content of the last role: "user" message.

Token streaming

A model port that calls the request's optional onDelta callback (see @nifrajs/agent's AgentModelDelta) streams live:

• { kind: "text", text } opens a TEXT_MESSAGE and streams each chunk as TEXT_MESSAGE_CONTENT. When any text was streamed, the terminal text block is suppressed - the streamed text IS the assistant message, so a streaming port must stream all user-visible text. A non-streaming port keeps the single terminal block; nothing changes for it.
• { kind: "reasoning", text } streams a REASONING_* message.
• { kind: "tool-args", name?, argsText } opens a provisional TOOL_CALL_START (id <turnId>:call:<n>) and streams TOOL_CALL_ARGS. The tool evidence that follows closes the same call - TOOL_CALL_END plus the token-only TOOL_CALL_RESULT - instead of opening a second one.
• { kind: "usage", provider?, model?, inputTokens?, ... } never becomes a frame. The counts are summed per (provider, model) across the run's model decisions and stamped as the spec usage: TokenUsage[] array on the terminal RUN_FINISHED (success and interrupt alike); non-finite figures are ignored.

Each stream closes when a different delta kind starts, when tool evidence lands, or at the end of the run. Deltas are transient observer data: a Last-Event-ID replay carries evidence frames and the stored, unsuppressed terminal events only - the stored RUN_FINISHED keeps its usage.

Shared state

The ports factory receives (c, run); run.sharedState is the run's AgentSharedState channel and run.turnId the resolved turn id. body.state seeds the document and is announced with an upfront STATE_SNAPSHOT; every patch from a port or tool executor streams as STATE_DELTA carrying the RFC 6902 ops. A first patch on an unseeded document announces itself as a STATE_SNAPSHOT instead of a delta against a base the client never saw. The channel is per-run and transient - nothing in it is persisted.

mountAgUI(app, {
  agent,
  ports: (c, run) => ({
    model: myStreamingModelPort(run.sharedState), // patch progress as the run advances
    capabilities: ["search.read"],
  }),
})

With emitMessagesSnapshot: true, a successful run also emits MESSAGES_SNAPSHOT - the request's messages echoed back with the assistant output message appended - before RUN_FINISHED. It is off by default: the snapshot echoes client message payloads, and terminal events persist to the evidence log when one is configured.

Human-in-the-loop resume

A suspended run finishes with outcome: { "type": "interrupt", "interrupts": [interrupt] } where the interrupt is:

{
  "id": "<turnId>",
  "reason": "approval",                    // approval | budget | model | cancelled | max_turns
  "toolCallId": "<effectId>",              // present for tool suspensions
  "responseSchema": { /* JSON Schema for the resume payload */ },
  "metadata": { "turnId": "…", "continuation": { "kind": "approval", "tool": "…", "effectId": "…" } }
}

Resume with the AG-UI resume array. The runtime keeps state token-only - it holds no interrupt registry - so the payload must echo metadata.continuation, with the suspended tool's input replayed in continuation.input. The turn keeps a digest of that input: a continuation naming another tool, effect, kind or input is refused, and an approval answers only an approval interrupt:

{
  "threadId": "thread-1", "runId": "run-2", "messages": [],
  "forwardedProps": { "input": { "prompt": "go" } },
  "resume": [{
    "interruptId": "<interrupt id>",
    "status": "resolved",                  // "cancelled" without an approval resumes as a denial
    "payload": {
      "continuation": { "kind": "approval", "tool": "search.read", "effectId": "…", "input": { "q": "…" } },
      "approval": { "granted": true }
    }
  }]
}

The pre-interrupt form - the same { continuation, approval? } object in forwardedProps.resume plus forwardedProps.turnId - keeps working. A resume that fails validation is ignored and the POST starts a fresh run.

Resumable streams

Pass an evidenceLog to make a dropped SSE connection resumable. Evidence-derived frames then carry SSE id: <seq>; a client reconnects by re-POSTing the same body with a Last-Event-ID header and receives the missed events, rejoining a still-running turn live or replaying the stored terminal events - the run is never re-executed. A disconnected run keeps going and records its real terminal events for that reconnect. Without an evidenceLog nothing can rejoin a dropped stream, so the disconnect aborts ports.signal and the turn suspends as cancelled.

Exceeding maxOutputBytes detaches the viewer in the same way, including an oversized initial state snapshot. The response closes immediately, with an output_limit error frame when the remaining budget permits it. Without a log the turn is aborted; with a log it continues recording its real terminal result for reconnect rather than storing the viewer's output-limit error as the run result.

import { createMemoryAgentEvidenceLog } from "@nifrajs/agent/events"

mountAgUI(app, { agent, ports, evidenceLog: createMemoryAgentEvidenceLog() })

The in-memory log is the single-process dev/test reference; a durable, multi-process log is an adapter implementing the same AgentEvidenceLog interface.

When more than one caller shares the route, also pass evidenceOwner: (c) => <the caller's user or tenant id>: turns are recorded and replayed under that owner, so another caller's reconnect or reused runId finds none of them. Without it a turn id (by default the client's runId) works as a bearer capability. The owner function runs before any replay; throwing from it refuses the request with 403, and a thrown Response answers the request as built, as one thrown from ports does.

The seam performs no authentication or authorization - wrap it with your app's route guards and scope every port in ports(c) to the caller.

Docs

Part of the nifra full-stack TypeScript framework - one core, five UI libraries, every runtime. Scaffold a new app with bun create nifra.

For AI agents, see LLM.md and the full corpus ../../llms-full.txt.