stream

package
v0.10.2 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 1, 2026 License: MIT Imports: 9 Imported by: 0

Documentation

Overview

Package stream provides event streaming for real-time observability.

Index

Constants

View Source 
const (
	EventNodeStart      = "node_start"
	EventNodeEnd        = "node_end"
	EventEdgeTransition = "edge_transition"
	EventToolCall       = "tool_call"
	EventToolResult     = "tool_result"
	EventModelCall      = "model_call"
	EventModelDelta     = "model_delta"
	EventModelResponse  = "model_response"
	EventCheckpoint     = "checkpoint"
	EventInterrupt      = "interrupt"
	EventCompleted      = "completed"
	EventError          = "error"
	EventPlanUpdate     = "plan_update"
	EventCustom         = "custom"
)

Standard event types published during graph and agent execution.

Variables

View Source 
var TopicQueryParams = []string{"topic", "session"}

TopicQueryParams are the request query parameters, in priority order, that SSEHandler reads to determine the topic a client subscribes to.

Functions

func Emit added in v0.3.0 

func Emit(ctx context.Context, eventType string, data any)

Emit sends a custom event from within a node function or tool handler. It is a no-op if no emitter is attached to the context.

func WithEmitter added in v0.3.0 

func WithEmitter(ctx context.Context, ch chan<- Event) context.Context

WithEmitter adds an event emitter channel to the context, allowing node functions and tool handlers to emit custom events during execution.

Types

type Broker 

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

Broker manages SSE subscribers. It routes events per topic (typically a session or tenant id) so a subscriber only receives events for its own topic, while a broadcast Publish still reaches every subscriber for backward compatibility. Delivery is delegated to a pluggable FanOut.

func NewBroker 

func NewBroker(opts ...Option) *Broker

NewBroker creates a new event broker. Without options it uses an in-process FanOut with sane defaults (max 1024 subscribers, 64-event buffers, 15s heartbeat).

func (*Broker) Close added in v0.4.0 

func (b *Broker) Close() error

Close releases all subscriptions and underlying resources.

func (*Broker) Heartbeat added in v0.10.0 

func (b *Broker) Heartbeat() time.Duration

Heartbeat returns the broker's configured SSE keepalive interval, so alternative SSE handlers (e.g. the AG-UI stream) can honor the same configuration instead of hardcoding their own. A value <= 0 means heartbeats are disabled.

func (*Broker) Publish 

func (b *Broker) Publish(evt Event)

Publish broadcasts an event to every subscriber, regardless of topic. Prefer PublishTopic to route to a single session/tenant and avoid cross-session leakage.

func (*Broker) PublishTopic added in v0.4.0 

func (b *Broker) PublishTopic(topic string, evt Event)

PublishTopic delivers an event only to subscribers of the given topic.

func (*Broker) SSEHandler 

func (b *Broker) SSEHandler(defaultTopic string) http.HandlerFunc

SSEHandler returns an HTTP handler that streams events to the client over Server-Sent Events. The topic each client subscribes to is taken from the request query (see TopicQueryParams), falling back to defaultTopic. Each connection gets a unique subscriber id, so concurrent clients never clobber one another. Periodic keepalive comments (see WithHeartbeat) keep idle connections and intermediary proxies from timing out, and the subscription is released when the client disconnects (request context is canceled) or the subscriber cap is exceeded (HTTP 503).

func (*Broker) Subscribe 

func (b *Broker) Subscribe(id string) <-chan Event

Subscribe creates a firehose subscription under the caller-supplied id and returns its channel. It exists for backward compatibility: with the empty topic the subscription receives every event — both broadcast Publish and any PublishTopic to a named session/tenant. (This is a deliberate change from the pre-routing behavior, where it saw only broadcasts; it is what the dashboard and monitor rely on.) Use SubscribeTopic for a session-scoped stream that is isolated from other sessions. Re-subscribing with an existing id replaces the previous subscription. If the subscriber cap is reached the returned channel is already closed.

func (*Broker) SubscribeTopic added in v0.4.0 

func (b *Broker) SubscribeTopic(topic string) (*Subscription, error)

SubscribeTopic creates a subscription scoped to topic. Only events published to that topic (or broadcast events) are delivered. Unsubscribe with the returned Subscription.ID.

func (*Broker) Unsubscribe 

func (b *Broker) Unsubscribe(id string)

Unsubscribe removes a subscription. It accepts either a caller-supplied Subscribe id or a Subscription.ID returned by SubscribeTopic.

type Event 

type Event struct {
	Type string `json:"type"`
	Data any    `json:"data"`
}

Event is a server-sent event.

type FanOut added in v0.4.0 

type FanOut interface {
	// Subscribe registers a new subscriber for topic and returns its handle.
	// It returns an error when the subscriber cap is reached.
	Subscribe(topic string) (*Subscription, error)
	// Unsubscribe removes the subscriber with the given id and releases its
	// resources. It is a no-op for an unknown id.
	Unsubscribe(id string)
	// Publish delivers evt to subscribers of topic plus any firehose
	// (empty-topic) subscribers. An empty topic is a broadcast that reaches
	// every subscriber regardless of topic. Delivery is best-effort: a
	// subscriber whose buffer is full drops the event rather than blocking
	// publishers.
	Publish(topic string, evt Event)
	// Close releases all subscriptions and underlying resources.
	Close() error
}

FanOut delivers published events to interested subscribers. The default implementation (newInProcessFanOut) keeps all state in memory. Multi-replica deployments can supply an external implementation (backed by Redis, NATS, etc.) via WithFanOut so events published on one replica reach subscribers on another. Implementations must be safe for concurrent use.

type Option added in v0.4.0 

type Option func(*Broker)

Option configures a Broker.

func WithBufferSize added in v0.4.0 

func WithBufferSize(n int) Option

WithBufferSize sets the per-subscriber channel buffer size. Events are dropped for a subscriber once its buffer fills, so slow consumers never block publishers.

func WithFanOut added in v0.4.0 

func WithFanOut(f FanOut) Option

WithFanOut installs a custom FanOut, allowing event delivery to be externalized (e.g. Redis/NATS) for multi-replica deployments. When set, the max-subscriber and buffer-size options apply to the supplied FanOut only if it honors them; the Broker delegates delivery to it verbatim.

func WithHeartbeat added in v0.4.0 

func WithHeartbeat(d time.Duration) Option

WithHeartbeat sets the interval at which SSEHandler emits keepalive comments. A value <= 0 disables heartbeats.

func WithMaxSubscribers added in v0.4.0 

func WithMaxSubscribers(n int) Option

WithMaxSubscribers bounds the number of concurrent subscribers. Subscriptions beyond the cap are rejected. A value <= 0 disables the cap.

type StreamConfig added in v0.3.0 

type StreamConfig struct {
	Mode StreamMode
}

StreamConfig holds configuration for how events are filtered and delivered.

func DefaultStreamConfig added in v0.3.0 

func DefaultStreamConfig() StreamConfig

DefaultStreamConfig returns a config that streams all events (debug mode).

func (StreamConfig) ShouldInclude added in v0.3.0 

func (c StreamConfig) ShouldInclude(eventType string) bool

ShouldInclude returns true if the given event type passes the mode's filter. ModeDebug includes all events.

type StreamMode added in v0.3.0 

type StreamMode string

StreamMode defines which events are included in a stream subscription. 

const (
	// ModeValues streams full state after each node execution.
	ModeValues StreamMode = "values"
	// ModeUpdates streams only changed state keys after each node.
	ModeUpdates StreamMode = "updates"
	// ModeCustom streams only user-emitted custom events.
	ModeCustom StreamMode = "custom"
	// ModeMessages streams LLM token-level events (model_call, model_response).
	ModeMessages StreamMode = "messages"
	// ModeDebug streams all internal execution details.
	ModeDebug StreamMode = "debug"
)

type Subscription added in v0.4.0 

type Subscription struct {
	ID    string
	Topic string
	C     <-chan Event
}

Subscription is a handle to an active fan-out subscription. ID is unique for the lifetime of the FanOut and is used to Unsubscribe; C delivers events for the subscription's Topic.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.