agui

Documentation

Overview

Package agui is the AG-UI protocol transport adapter (Slice 8a). It consumes Aura's in-process iter.Seq2[*agent.Event, error] stream and maps it onto the official AG-UI community Go SDK event surface. The boundary is one-way: agui imports agent, the agent runtime NEVER imports agui (D-17, CI-enforced via scripts/agui_boundary_check.sh). The translator (translator.go) is a pure function — no I/O, no goroutines — so it is property- and golden-testable in isolation.

Index

• Constants
• Variables
• func SanitizeString(msg string) string
• func Subscribe(ctx context.Context, threadID, runID string, idgen IDGenerator, ...) <-chan Event
• func Translate(threadID, runID string, idgen IDGenerator, seq iter.Seq2[*agent.Event, error], ...) iter.Seq2[events.Event, error]
• func ValidateRunInput(in types.RunAgentInput) error
• type ConversationStore
• type Event
• type EventType
• type Fanout
  • func NewFanout(source iter.Seq2[events.Event, error]) *Fanout
  • func (f *Fanout) Run(ctx context.Context)
  • func (f *Fanout) Subscribe() <-chan events.Event
• type IDGenerator
  • func NewIDGenerator() IDGenerator
• type ReadinessProbe
• type Runner
• type Server
  • func NewServer(run Runner, conv ConversationStore, cfg ServerConfig) *Server
  • func (s *Server) Mux() http.Handler
• type ServerConfig

Constants

const ArtifactEventName = "aura.artifact"

ArtifactEventName is the stable AG-UI CUSTOM-event name the artifact branch emits (D-06). A channel consumer keys on this name to render a delivered file (Telegram → sendDocument, CLI → print path, AG-UI HTTP → pass-through). It is namespaced so it never collides with another application's custom events. Exported so the channel consumers (e.g. telegram artifact.go, plan 13-06) key on the SAME canonical name rather than a duplicated magic string.

Variables

var (
	ErrEmptyThreadID = errors.New("agui: threadId must not be empty")
	ErrNoMessages    = errors.New("agui: messages must not be empty")
)

ErrEmptyThreadID and ErrNoMessages are the Aura-semantic validation sentinels ValidateRunInput returns. They are distinct from the SDK's parse errors: the SDK's RunAgentInput.UnmarshalJSON owns camel/snake JSON shape, while these guard Aura's run preconditions (a thread to resolve, a message to drive a turn).

Functions

func SanitizeString

func SanitizeString(msg string) string

SanitizeString strips credential-bearing substrings from an arbitrary string: whole DB DSNs collapse to a scheme marker, generic URL userinfo collapses to `scheme://[redacted]@`, and bearer/api-key/token shapes collapse to a `prefix[redacted]` marker (WR-03). The DSN pass runs first so the generic userinfo pass never reaches the five DB schemes.

func Subscribe

func Subscribe(ctx context.Context, threadID, runID string, idgen IDGenerator, turn iter.Seq2[*agent.Event, error]) <-chan Event

Subscribe composes Plan 01's Translate with a single-subscriber Fanout and returns a receive-only channel of AG-UI events for ONE in-process consumer of an agent turn. Given the threadID/runID, an IDGenerator, and the agent's iter.Seq2[*agent.Event, error] turn stream, it translates the turn and pumps it through the cap-64, drop-on-full, ctx-cancel-clean Fanout — the same backpressure + leak discipline a future Telegram channel consumes (amendment #35 agui_subscriber.go seam). The channel closes when the turn ends or ctx is cancelled.

This seam is transport-free: no HTTP, no SSE (that is server.go in a later plan). For N concurrent consumers, build a *Fanout directly and Subscribe() per consumer; this helper is the single-consumer convenience over that same pump.

Reasoning forward-compat: the translated stream may later carry REASONING_* events (the dual-field reasoning/reasoning_content data-plane, amendment #57). This phase only surfaces the alias — no reasoning parsing is added here.

func Translate

func Translate(threadID, runID string, idgen IDGenerator, seq iter.Seq2[*agent.Event, error], showReasoning bool) iter.Seq2[events.Event, error]

Translate maps Aura's per-token *agent.Event stream onto a valid AG-UI events.Event sequence. It is a PURE function (no I/O, no goroutines): RUN_STARTED first, RUN_FINISHED(success) last, with a chunk-coalescing text-message state machine in between. idgen owns id minting so every emitted messageId/toolCallId is non-empty (both are Validate()-required by the SDK).

Run-boundary policy (locked OQ1 a): the real (*LlmAgent).Run streams per-token chunk Events AND a final Event carrying the full answer + FinishReason. We stream the deltas (one TEXT_MESSAGE_CONTENT per non-empty delta) and treat the final Event as END-only — its Content is NOT re-emitted as a CONTENT (no double-stream); its usage StateDelta becomes a STATE_DELTA event.

showReasoning controls the REASONING_MESSAGE_CONTENT payload: false (the default privacy posture) substitutes redactedReasoningDelta; true passes the real CoT delta through so an opted-in consumer (Telegram's live 💭 window) can display it. The REASONING_* lifecycle envelope is emitted either way — only the delta text differs.

func ValidateRunInput

func ValidateRunInput(in types.RunAgentInput) error

ValidateRunInput checks the Aura-semantic preconditions for a run: a non-empty threadId (resolved to a conversation downstream by conversations.Get, which owns the UUID-shape check) and at least one message to drive the turn. It does NOT re-implement JSON parsing (the SDK's UnmarshalJSON does) nor re-validate the UUID shape (conversations.Get's parseUUID does).

Types

type ConversationStore

type ConversationStore interface {
	Get(ctx context.Context, conversationID string) (conversations.Conversation, error)
	LoadHistory(ctx context.Context, conversationID string) ([]llm.Message, error)
}

ConversationStore is the narrow conversation surface the AG-UI server consumes (D-A2-02, "accept interfaces, return structs"). *conversations.Store satisfies it implicitly. Get resolves a threadId → conversation (404 chokepoint); LoadHistory rehydrates the byte-identical turn history for the GET messages projection. Declared consumer-side so agui depends only on the two methods it calls, not the whole Store.

type Event

type Event = events.Event

Event is the AG-UI event surface re-exported under the agui package so in-process consumers (e.g. the Phase-13 Telegram channel) reference agui.Event, not the external events.Event — keeping github.com/ag-ui-protocol/... out of every call site (PRD Slice 8: "Type aliases per evitare leak del package esterno nei call sites"). Only the subscriber surface is aliased; the SDK is not blanket re-exported.

type EventType

type EventType = events.EventType

EventType is the aliased AG-UI event discriminator (agui.Event.Type() returns it), re-exported alongside Event so a consumer can switch on event kind without importing the SDK.

type Fanout

type Fanout struct {
	// contains filtered or unexported fields
}

Fanout distributes a single translated AG-UI event stream to N concurrent in-process subscribers. It wraps the Translate output (iter.Seq2[events.Event, error]) and, for each event, sends to every subscriber via a cap-64 buffered, drop-on-full, never-block pump (the openai_compat Stream analog plus a default-drop arm). The single producer goroutine started by Run is the SOLE sender on every subscriber channel and closes them all on exit (ctx-cancel OR source end), so the stream is goleak-clean (Pitfall 4).

Fanout is NOT on the HTTP path (RESEARCH diagram line 134): server.go does not route through it. It is the in-process seam the Phase-13 Telegram channel consumes.

Errors from the wrapped source are already mapped to a RUN_ERROR event by Translate (the translated stream is pure events.Event), so Fanout never inspects the error slot — it just forwards whatever Translate yields.

func NewFanout

func NewFanout(source iter.Seq2[events.Event, error]) *Fanout

NewFanout wraps a translated event stream. Subscribe before Run to register a consumer; Run then drives the source once and fans out to all registered subscribers.

func (*Fanout) Run

func (f *Fanout) Run(ctx context.Context)

Run starts the single producer goroutine that ranges the wrapped source once and fans each event to every subscriber. It returns immediately; the goroutine exits on ctx-cancel or source end, closing every subscriber channel on the way out (sole sender closes — golang-concurrency principle #4). Run is non-blocking and must be called AT MOST ONCE: a second Run would launch a second producer that double-closes the subscriber channels (panic: close of closed channel). The started guard makes the misuse loud at the call site instead (WR-06).

func (*Fanout) Subscribe

func (f *Fanout) Subscribe() <-chan events.Event

Subscribe registers a new consumer and returns its receive-only channel. The channel is cap-64 buffered and is closed by the producer goroutine when the source ends or ctx is cancelled. Subscribe MUST be called before Run: a subscriber registered after the producer has snapshotted f.subs is never fed (and racing the snapshot is a data race), so Subscribe-after-Run panics loudly rather than failing silently (WR-06).

type IDGenerator

type IDGenerator interface {
	NewMessageID() string
	NewReasoningID() string
	NewToolResultID(toolCallID string) string
}

IDGenerator owns AG-UI id minting so the translator can guarantee non-empty messageId/toolCallId (both are Validate()-required by the SDK) and so tests can inject deterministic ids for stable golden compares. NewMessageID mints an assistant message run id; NewReasoningID mints the SEPARATE rsn- message id a REASONING lifecycle carries (distinct from the assistant TEXT_MESSAGE id so a consumer never conflates CoT with the answer); NewToolResultID mints the message id carrying a tool result (correlated to its toolCallID).

func NewIDGenerator

func NewIDGenerator() IDGenerator

NewIDGenerator returns the default uuid-v4-backed IDGenerator.

type ReadinessProbe

type ReadinessProbe struct {
	Name  string
	Check func(context.Context) error
}

ReadinessProbe is one named dependency-readiness check the server runs on /readyz. Check returns nil when the dependency is reachable, a non-nil error (surfaced sanitized in the 503 body) when it is not. Probes are injected by the composition root (cmd/aura wires the real PG + Neo4j probes) so they stay testable and the set is easy to extend without touching the handler.

type Runner

type Runner interface {
	Turn(ctx context.Context, convID string, userMsg *string) iter.Seq2[*agent.Event, error]
	SubmitAnswers(ctx context.Context, answers map[string]runner.ResponseInput) (int, error)
}

Runner is the narrow agent-driver surface the server consumes (D-A2-02; *runner.Runner satisfies it implicitly). Turn drives one round over a thread and yields the agent Event stream; SubmitAnswers resolves protocol-native HITL resume entries (resolved→accept, cancelled→cancel) before the Turn. Declared consumer-side so the server depends only on the two methods it calls, not the whole Runner — and so unit tests pass scripted fakes.

type Server

type Server struct {
	// contains filtered or unexported fields
}

Server is the minimal AG-UI HTTP gateway (Slice 8b): POST /agent/run streams a translated agent turn as SSE, GET /threads/{id}/messages returns the persisted history as a MESSAGES_SNAPSHOT JSON body. It is the thinnest glue over EXISTING seams — the narrow Runner + ConversationStore, the translator, and the SDK SSE writer. The bind is hardcoded loopback by the daemon (auth deferred this phase, amendment #35); the loopback bind IS the compensating control (T-12-08).

func NewServer

func NewServer(run Runner, conv ConversationStore, cfg ServerConfig) *Server

NewServer builds the gateway over the supplied driver + store + config. The IDGenerator is the default uuid-v4 minter; tests inject a deterministic one via the exported field for stable frame ids.

func (*Server) Mux

func (s *Server) Mux() http.Handler

Mux registers the two routes using Go 1.22+ method-pattern routing (no chi/gorilla — matches the no-router codebase posture). When CORSPermissive is on (the dev knob) the handler is wrapped in withCORS so a browser cross-origin POST works end to end: the preflight OPTIONS is answered and ACAO is set on every response, including errors.

type ServerConfig

type ServerConfig struct {
	CORSPermissive  bool
	BufferCap       int
	HealthCheck     func(context.Context) error
	HealthDetails   func() map[string]any
	ReadinessProbes []ReadinessProbe
}

ServerConfig carries the AG-UI server knobs the daemon resolves from config (AURA_AGUI_*). CORSPermissive gates the `Access-Control-Allow-Origin: *` header (default restrictive, T-12-13); BufferCap is the cap of the per-connection SSE pump channel (drop-on-full, never block the Loop, T-12-09). A non-positive BufferCap falls back to the fanout default. ReadinessProbes are the required-dependency checks /readyz runs (O-05/AP-14): /healthz stays a cheap LIVENESS check, /readyz reflects whether the required backends (PG + Neo4j) are reachable. An empty list reports ready (the daemon was started without gated deps).