FlowAgent AG-UI Bridge — Live Text/Agentic Surface (staff + guest)
Audiences: developer, internal, clinical-buyer
FlowAgent AG-UI Bridge — Live Text/Agentic Surface (staff + guest)
The SECOND live surface for a FlowAgent session. Where the voice bridge (livekit-flowagent-ts.md) drives the patient's mic over a LiveKit DataChannel, the AG-UI bridge streams the SAME 17 FlowAgent runtime events — mapped onto Condor's hand-rolled AG-UI event taxonomy — over a dedicated raw SSE wire, so a browser (staff OR guest) can render the live intake as a text/agentic UI (field document + transcript + UI-action commands) with zero-loss resume on refresh.
Business Purpose
FlowAgent authors a scenario once and runs it as voice. But two consumers need the run WITHOUT a mic:
Clinic staff watching a patient's intake in real time (the doctor/operator surface) — they want the live field document + transcript, not audio.
A guest patient self-completing an intake from an invite link — no clinic login, no LiveKit room, a plain browser SSE surface gated by a time-boxed invite token.
The AG-UI bridge serves both from ONE contract: the publisher already emits 17 SSE events per run; the bridge tees each into an in-process bus, maps it to an AG-UI event, persists it to a durable store, and streams it over /agui/live. The web consumer folds the stream into a render-ready FlowAgentRunState. This is what makes FlowAgent a text-or-voice product from a single authored scenario.
Audiences
Developer/partner: The bridge is contract-first — flowagent-agui.contract.ts fixes the 17-event → AG-UI-variant map (imported, never re-derived). It REUSES Condor's AG-UI primitives (AguiEventStore, agui-sse.helper.ts, the AgUiEvent union) as read-only types — it never edits ai/condor/**.
Internal (ops/support): The wire is a dedicated raw RFC 8895 SSE transport, deliberately NOT the @ddx/sse-contract v4 SseEnvelope — the global EDA bus path (publishV3) is preserved untouched (dual-emit). Two disjoint sinks, one publisher.
Clinical-buyer: The guest surface lets a patient complete intake from a link with no account, gated by an expiring invite token that carries NO clinic RBAC claims.
Architecture
diagram
FlowAgent runtime (voice bridge OR HTTP guest message)
│ emits one of 17 FlowAgent SSE events
▼
ddx-api FlowAgentEventsPublisher.publishV3(name, meta, data) [DUAL-EMIT]
├─► @ddx/sse-contract global bus (Redis Streams) ── UNCHANGED, v4 SseEnvelope
└─► FlowAgentAguiMapperService.mapFlowAgentEvent(name, data) → AgUiEvent[]
└─► FlowAgentAguiBus.publish(sessionId, event)
1. AguiEventStore.persistOne(sessionId, …) ── persist FIRST (AC-5)
2. Subject.next({ sessionId, seq, event }) ── live tap
▼
GET /api/ai/flowagent/sessions/:id/agui/live (FlowAgentAguiController — staff)
GET /api/ai/flowagent/guest/sessions/:id/agui/live (FlowAgentGuestController — guest)
│ streamSessionAguiEvents: replay store from Last-Event-ID → tail bus
│ raw RFC 8895 frames (id: <seq>, data: <AgUiEvent JSON>)
▼
ddx-web BFF /api/flowagent/agui/[id]/route.ts (proxy, session-scoped)
▼
useFlowAgentAguiStream(sessionId) ── fetch + ReadableStream, Last-Event-ID resume
└─► createFlowAgentRunStateFold ── RFC6902 patch over FlowAgentState + transcript
└─► FlowAgentAguiProvider → renderers (field document, transcript, ui-action)
Tech Stack & Choices
Contract-first (INV-2): flowagent-agui.contract.ts owns FLOWAGENT_SSE_EVENTS (the 17 authoritative names) and FLOWAGENT_AGUI_EVENT_MAP (name → Condor AgUiEvent['type']). Compile-time conformance proofs (FlowAgentContractConformance, AssertAssignable<...>) force tsc to prove every mapped variant is assignable to the read-only Condor union — a Condor rename breaks THIS file's build.
The 17 → AG-UI map (contract §4): lifecycle → RunStarted/RunFinished/RunFailed; tool:start/tool:end → ToolCallStart/ToolCallEnd; agent:message → TextMessageContent; ui:command → Custom{name:'flowagent.ui-action'}; the 11 paused/section/field/checkpoint/language/score events → StateDelta carrying an RFC6902 JSON-Patch over FlowAgentState. session:error maps to RunFailed (a failed run ends the run), NOT the special Error variant.
Never-drop (AC-1): any event the mapper cannot classify becomes Custom{name:'unmapped-chunk'} — never silently dropped.
Dual-emit, disjoint transports: publishV3 keeps the @ddx/sse-contract global-bus path AND tees the mapped event to FlowAgentAguiBus. The /agui/live wire is raw RFC 8895 (agui-sse.helper.ts), DELIBERATELY not SseEnvelope — cross-referencing the SSE contract, the bridge is a separate agentic wire, so it does not go through the v3/v4 envelope bridge (which would 422 a non-envelope frame).
Persist-then-publish (AC-5): FlowAgentAguiBus.publish chains each session's events on a per-session promise tail — persist to AguiEventStore FIRST, then Subject.next(). A reconnecting consumer replays every event from the store with zero loss; a persist miss still emits live (seq=null) so a store failure never silences the tap.
In-process RxJS Subject (not Redis): NestJS is single-process here; the live tap only needs same-process fan-out from publisher to the open SSE response. Cross-instance durability is the durable-store concern (Condor's AguiEventStore, keyed here by sessionId as the partition — FlowAgent has no per-activation runId).
Reader-without-writer guard: the bus doc explicitly wires BOTH ends in the same change — WRITER = publishV3 (all 17 methods), READER = the two controllers via streamSessionAguiEvents. Neither ships without the other.
Data Flow
Emit: the FlowAgent runtime (voice bridge, or a guest HTTP POST /guest/sessions/:id/message → SessionService.processMessage) triggers FlowAgentEventsPublisher, which calls publishV3 for each of the 17 event methods.
Map: publishV3 calls FlowAgentAguiMapperService.mapFlowAgentEvent(name, data, base) → an AgUiEvent[] (fixed by the contract table; StateDelta events build an RFC6902 patch via buildStatePatch).
Tee + persist: each mapped event → FlowAgentAguiBus.publish(sessionId, event) → AguiEventStore.persistOne (seq assigned) → Subject.next.
Live-bridge: a browser opens GET …/agui/live (staff or guest route). Auth + ownership/tenant (staff) or invite-token (guest) resolve BEFORE the first SSE byte. streamSessionAguiEvents REPLAYS the durable log from Last-Event-ID, then TAILS the live bus (replay-before-tail, mirrors the canonical hub).
BFF: ddx-web /api/flowagent/agui/[id]/route.ts proxies session-scoped (no runId param — FlowAgent is session-scoped, unlike Condor's runId scope).
Consume: useFlowAgentAguiStream(sessionId) reads the stream via fetch + ReadableStream (not EventSource — needs Last-Event-ID header + AbortController). It persists the frame cursor (id: line = String(seq), else event.timestamp) to localStorage (ddx.flowagent.lastSeen.<sessionId>) so a full refresh resumes; it clears the cursor on terminal lifecycle so a finished run never replays.
Fold: createFlowAgentRunStateFold folds each AG-UI frame into ONE FlowAgentRunState — StateDelta applies the RFC6902 patch over FlowAgentState; TextMessageContent coalesces per-messageId deltas into a rolling transcript line (cap 500); Custom{flowagent.ui-action} is surfaced verbatim to the provider's onUiAction sink (NOT a state mutation); tool-call frames have no state projection.
Auth Model (two controllers, one wire)
Staff (flowagent-agui.controller.ts): class-level RolesGuard + @Roles(DOCTOR, NURSE, CLINIC_ADMIN, ORG_ADMIN, SUPER_ADMIN), NO @RbacExempt. loadOwnedSession enforces operator-ownership + tenant (defense-in-depth; tolerates the ddx_api_ai slug-vs-UUID org skew — mismatch enforced only when both sides present and unequal).
Guest (flowagent-guest.controller.ts): a SIBLING controller (NOT a method on the staff controller — a guest has no role and the class-level RolesGuard would reject it before its own @RbacExempt applied). Every handler individually pairs @Public() + @RbacExempt() + @UseGuards(FlowAgentInviteTokenGuard) — a bare @Public/@RbacExempt without the token guard is forbidden (the task validation greps for exactly that).
Invite token (flowagent-invite-token.guard.ts): a stateless JWT (no token table, no Prisma change) carrying ONLY { sessionId, scenarioId, kind } — NO clinic/RBAC claims (AC-4). Expiry lives in the JWT (expiresIn set at mint time in session.service); after the window verify throws → 401 BEFORE any SSE byte (AC-5). The token's sessionId MUST match the route :id (a token for session A can never open session B). The guest's real org + operator are resolved from the session ROW, never trusted from the token; the guest runs as a zero-permission PATIENT GatewayUser.
Implicated Code
MANDATORY: ≥3 file:line citations.
ddx-api/src/ai/flowagent-agui/contracts/flowagent-agui.contract.ts:38-56 — FLOWAGENT_SSE_EVENTS (the 17 authoritative event names).
ddx-api/src/ai/flowagent-agui/flowagent-agui.controller.ts:89-122 — staff GET :id/agui/live (replay-before-tail; ownership resolved before first byte).
ddx-api/src/ai/flowagent-agui/flowagent-guest.controller.ts:91-128 — guest POST :id/message (@Public+@RbacExempt+FlowAgentInviteTokenGuard).
ddx-api/src/ai/flowagent-agui/flowagent-guest.controller.ts:205-217 — buildGuestUser (zero-permission PATIENT, real org from session row).
ddx-api/src/ai/flowagent/sse/flowagent-events.publisher.ts:96-160 — publishV3 dual-emit (global bus + aguiMapper → aguiBus).
Two live surfaces, one session: the DataChannel voice path (flowagent-client/events/data-channel-handler.ts: state_changed/transcript/agent_speech/…) and this AG-UI SSE path both observe the same run. The AG-UI wire is the text/agentic surface; the DataChannel is the in-room voice surface. Don't conflate — a guest text session uses ONLY the AG-UI wire (no LiveKit room).
Resume: the web cursor key is ddx.flowagent.lastSeen.<sessionId> (distinct from Tucan's per-channel ddx.sse.lastEventId.<channel>). The AG-UI wire uses AguiEventStore (30-day TTL, no MAXLEN trim) so a refresh within the window always finds its events — there is intentionally NO replay:gap handling (that is a platform-hub-only signal).
No secrets in this doc: invite-token secret name lives in session.service/guard; LiveKit keys in the voice bridge — names only, never values.
Module wiring: flowagent-agui.module.ts imports CondorModule (for AguiEventStore + AgenticFlowRouter) and wires the mapper/bus/controllers as REAL providers (no @Optional); FlowAgentModule imports it via forwardRef to break the module cycle.
Related Topics
FlowAgent Scenarios — the authored scenario + the 17 runtime events this bridge maps