Documentation
¶
Index ¶
- func JobsTopic(agentID uuid.UUID) uuid.UUID
- type BuildEventPublisher
- func (p *BuildEventPublisher) PublishBuildEvent(ctx context.Context, agentID, buildID uuid.UUID, status, errMsg, phase string, ...)
- func (p *BuildEventPublisher) PublishBuildLogLine(ctx context.Context, agentID, buildID uuid.UUID, seq int64, ...)
- func (p *BuildEventPublisher) PublishBuildTodos(ctx context.Context, buildID uuid.UUID, seq int64, todosJSON []byte)
- type Conn
- func (c *Conn) JobsSubscriptions() []uuid.UUID
- func (c *Conn) ReadPump(ctx context.Context, handler MessageHandler)
- func (c *Conn) Send(data []byte)
- func (c *Conn) SendEnvelope(env Envelope)
- func (c *Conn) TrackJobsSubscription(agentID uuid.UUID)
- func (c *Conn) UntrackJobsSubscription(agentID uuid.UUID)
- func (c *Conn) WritePump(ctx context.Context)
- type Envelope
- type Handler
- type Hub
- func (h *Hub) BroadcastToTopic(topicID uuid.UUID, env Envelope)
- func (h *Hub) ClearTopicBuffer(topicID uuid.UUID)
- func (h *Hub) ConnCount() int
- func (h *Hub) OnTopicSubscriptionChange(onFirst func(uuid.UUID), onLast func(uuid.UUID))
- func (h *Hub) Register(conn *Conn)
- func (h *Hub) ResyncAll()
- func (h *Hub) SendToConnection(connID string, env Envelope)
- func (h *Hub) Subscribe(conn *Conn, topicID uuid.UUID)
- func (h *Hub) TopicConnCount(topicID uuid.UUID) int
- func (h *Hub) Unregister(conn *Conn)
- func (h *Hub) Unsubscribe(conn *Conn, topicID uuid.UUID)
- type JobEventRelay
- type MessageHandler
- type PubSub
- type SubagentInfo
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
Types ¶
type BuildEventPublisher ¶
type BuildEventPublisher struct {
// contains filtered or unexported fields
}
BuildEventPublisher adapts PubSub to the builder.EventPublisher interface.
Build events split across two topics: lightweight lifecycle status (agent.build, carrying tasks_done/total for the badge) broadcasts on the AGENT topic so every member's UI updates without opening the Build page; the verbose stream (log lines, live actions, todos) publishes on the per-BUILD topic, which the Build page subscribes to only while open.
func NewBuildEventPublisher ¶
func NewBuildEventPublisher(ps *PubSub, hub *Hub) *BuildEventPublisher
NewBuildEventPublisher creates a BuildEventPublisher.
func (*BuildEventPublisher) PublishBuildEvent ¶
func (p *BuildEventPublisher) PublishBuildEvent(ctx context.Context, agentID, buildID uuid.UUID, status, errMsg, phase string, tasksDone, tasksTotal int32)
PublishBuildEvent publishes an agent build lifecycle event on the agent topic. tasksDone/tasksTotal drive the "Building N/M tasks" badge.
func (*BuildEventPublisher) PublishBuildLogLine ¶
func (p *BuildEventPublisher) PublishBuildLogLine(ctx context.Context, agentID, buildID uuid.UUID, seq int64, stream, line string)
PublishBuildLogLine publishes a single build log line on the per-build topic. seq is a monotonic counter per build (across both sol and docker streams); the frontend uses it to dedupe against a REST snapshot of the persisted log.
func (*BuildEventPublisher) PublishBuildTodos ¶ added in v0.4.0
func (p *BuildEventPublisher) PublishBuildTodos(ctx context.Context, buildID uuid.UUID, seq int64, todosJSON []byte)
PublishBuildTodos publishes the agent's full todo list on the per-build topic. todosJSON is the same jsonb blob persisted to agent_builds.todos.
type Conn ¶
type Conn struct {
ID string
UserID uuid.UUID
Email string
// TenantRole is fixed for the lifetime of the connection. The live
// authorization monitor closes the connection if the user's role changes.
TenantRole auth.Role
// SinceSeq is the client's replay cursor from the ?since= connect
// param: the max Envelope.Seq it has already processed. Set once by
// the WS accept handler before the Subscribe loop; the hub replays
// only seq>SinceSeq per topic (0 = fresh connect, no replay).
SinceSeq uint64
// contains filtered or unexported fields
}
Conn wraps a WebSocket connection with user identity and a send buffer.
func NewConn ¶
func NewConn(ws *websocket.Conn, userID uuid.UUID, email string, tenantRole auth.Role, disconnect context.CancelFunc, logger *zap.Logger) *Conn
NewConn creates a new Conn for the given WebSocket and user.
func (*Conn) JobsSubscriptions ¶ added in v0.5.0
JobsSubscriptions returns a snapshot of dynamically subscribed agent IDs.
func (*Conn) ReadPump ¶
func (c *Conn) ReadPump(ctx context.Context, handler MessageHandler)
ReadPump reads messages from the WebSocket and dispatches to handler. Blocks until the connection closes or ctx is cancelled.
func (*Conn) SendEnvelope ¶
SendEnvelope marshals and enqueues an envelope. Correctness control messages disconnect a slow client instead of being dropped so it reloads state.
func (*Conn) TrackJobsSubscription ¶ added in v0.5.0
TrackJobsSubscription records an authorized dynamic jobs subscription.
func (*Conn) UntrackJobsSubscription ¶ added in v0.5.0
UntrackJobsSubscription removes a dynamic jobs subscription.
type Envelope ¶
type Envelope struct {
Type string `json:"type"`
RequestID string `json:"requestId,omitempty"`
TopicID string `json:"topicId,omitempty"`
UserID string `json:"userId,omitempty"`
// Seq is a hub-global monotonic sequence stamped at publish. The
// client keeps the max seq it has processed and presents it as
// ?since= on (re)connect; the hub replays only seq>since per topic,
// or sends a `resync` when the gap is wider than the buffer. Opaque
// to the client ("bigger = newer"); when realtime moves to a shared
// bus the source changes without touching the wire contract.
Seq uint64 `json:"seq,omitempty"`
ConversationID string `json:"conversationId,omitempty"`
Subagent *SubagentInfo `json:"subagent,omitempty"`
Payload json.RawMessage `json:"payload,omitempty"`
}
Envelope is the JSON wire format for all WebSocket messages.
UserID is the principal who "owns" the event — typically the human at the top of a run chain. Subscribers gate delivery on env.UserID == conn.UserID, so a tenant admin who happens to be a member of agent X does not see live events from runs another user started on X. Pre-existing system-level broadcasts leave it empty and fall through to the "deliver to every subscriber on the topic" behaviour. ConversationID lets the frontend route an event to the correct chat card without payload introspection. Subagent tags an envelope as a sub-run event so the chat store can render it underneath the parent run's tool-call instead of as a top-level message.
func NewEnvelope ¶
NewEnvelope creates an Envelope, marshaling the payload via protojson. The proto.Message constraint provides compile-time enforcement — passing map[string]any or dbq types is a compile error.
This constructor leaves UserID empty, which means "deliver to every subscriber on the topic" — appropriate for system-level broadcasts (agent.synced, build events). Per-user events should use NewEnvelopeForUser so the WS hub can apply the user-id gate.
func NewEnvelopeForUser ¶ added in v0.4.0
func NewEnvelopeForUser(eventType, topicID, userID, conversationID string, payload proto.Message) Envelope
NewEnvelopeForUser is NewEnvelope plus per-user gating + conversation routing. Callers that publish run events (text-delta, tool-call, etc.) use this so subscribers other than the run's owner don't receive the live stream.
func (Envelope) WithSubagent ¶ added in v0.4.0
func (e Envelope) WithSubagent(info SubagentInfo) Envelope
WithSubagent tags the envelope as a sub-run event and returns it. Chainable on the constructors above.
type Handler ¶
type Handler struct {
// contains filtered or unexported fields
}
Handler routes dynamic WebSocket subscriptions that require narrower access than the agent topics installed by the WS upgrade handler.
func NewHandler ¶
NewHandler creates a new inbound message handler.
func (*Handler) HandleMessage ¶
HandleMessage routes an inbound client message.
type Hub ¶
type Hub struct {
// contains filtered or unexported fields
}
Hub manages WebSocket connections and topic subscriptions. A single connection can be subscribed to many topics at once — typically every agent the user is a member of, wired up at WS-accept time.
func (*Hub) BroadcastToTopic ¶
BroadcastToTopic sends an envelope to all connections subscribed to a topic.
User-id gating: if env.UserID is set, only connections whose authenticated user matches receive the event. Empty env.UserID falls through to the historical "deliver to every subscriber" behaviour (used for tenant-wide / system broadcasts: agent.synced, build events, etc).
The replay buffer stores (bytes, userID) pairs so a late subscriber gets the same gate applied on replay — without it, joining late would leak events from other users' runs that happened before the join.
func (*Hub) ClearTopicBuffer ¶
ClearTopicBuffer removes the replay buffer for a topic. Called after a terminal event (build complete/failed) since replay is no longer needed.
func (*Hub) OnTopicSubscriptionChange ¶
OnTopicSubscriptionChange sets callbacks for when the first local connection subscribes to a topic and when the last unsubscribes. Used by PubSub to manage Redis subscriptions.
func (*Hub) ResyncAll ¶ added in v0.5.0
func (h *Hub) ResyncAll()
ResyncAll asks every connected client to reload authoritative REST state.
func (*Hub) SendToConnection ¶
SendToConnection sends an envelope to a specific connection by ID.
func (*Hub) Subscribe ¶
Subscribe adds the connection to the given topic if not already subscribed. Pre-existing topic subscriptions on this connection are not affected — Subscribe is additive, so the WS accept handler can call it in a loop over the user's member agents without disturbing anything else. Replays any buffered events for the topic to the newly-subscribed connection.
func (*Hub) TopicConnCount ¶
TopicConnCount returns the number of connections subscribed to a topic.
func (*Hub) Unregister ¶
Unregister removes a connection and all its topic subscriptions, firing onLastUnsubscribe for any topic that's now empty.
func (*Hub) Unsubscribe ¶ added in v0.4.0
Unsubscribe removes one topic subscription from a connection (the inverse of a single Subscribe). Used when the client leaves a dynamically-subscribed topic — e.g. the Build page unmounting its per-build subscription — while the connection's other subscriptions stay intact. Idempotent.
type JobEventRelay ¶ added in v0.5.0
type JobEventRelay struct {
// contains filtered or unexported fields
}
JobEventRelay fans committed PostgreSQL job notifications out to this replica's local WebSocket subscribers.
func NewJobEventRelay ¶ added in v0.5.0
type MessageHandler ¶
MessageHandler processes an inbound WebSocket message from a client.
type PubSub ¶
type PubSub struct {
// contains filtered or unexported fields
}
PubSub delivers real-time envelopes to local WebSocket subscribers via the Hub. This is the single-instance implementation. For multi-instance fan-out (e.g. via Redis pub/sub), replace or extend this with a cross-process transport.
func (*PubSub) ClearTopicBuffer ¶
ClearTopicBuffer removes the replay buffer for a topic. Call after terminal events (run complete, build complete) so reconnecting clients don't replay stale streaming events.
type SubagentInfo ¶ added in v0.4.0
type SubagentInfo struct {
AgentID string `json:"agentId"`
RunID string `json:"runId"`
Slug string `json:"slug,omitempty"`
}
SubagentInfo identifies a sub-run when an A2A child agent's events are mirrored to the parent's topic. Frontend chat-store reads it to attach the event to the parent run's active tool-call card.
Source Files