Documentation
¶
Overview ¶
Package streamhub provides task-scoped stream buffers for worker-push and client subscribe (SSE).
Package wsconn defines the typed event protocol for WebSocket communication between the portal frontend and the server. All messages are JSON envelopes with a "type" field and a typed "payload".
Index ¶
- Constants
- func DecodePayload[T any](env Envelope) (T, error)
- func Encode(eventType string, payload any) ([]byte, error)
- func Serve(w http.ResponseWriter, r *http.Request, userID, spaceID string, deps ConnDeps)
- type Conn
- type ConnDeps
- type ConnRegistry
- func (r *ConnRegistry) Broadcast(spaceID, userID, eventType string, payload any)
- func (r *ConnRegistry) BroadcastSink(spaceID, userID, conversationID string) llm.StreamSink
- func (r *ConnRegistry) Consume(incoming <-chan []byte)
- func (r *ConnRegistry) ForUser(userID string) []*Conn
- func (r *ConnRegistry) Register(userID string, c *Conn)
- func (r *ConnRegistry) SetPublisher(fn func([]byte))
- func (r *ConnRegistry) Unregister(userID string, c *Conn)
- type ConversationCreate
- type ConversationCreated
- type ConversationError
- type ConversationMessage
- type Envelope
- type MessageCompleted
- type MessageDelta
- type MessageDequeued
- type MessageQueued
- type StreamHub
- type SubscribeTask
- type SystemError
- type TaskStatusChanged
- type TaskStreamDelta
- type TaskStreamDone
- type Turner
- type UnsubscribeTask
Constants ¶
const ( TypeConversationCreate = "conversation.create" TypeConversationMessage = "conversation.message" TypeSubscribeTask = "subscribe.task" TypeUnsubscribeTask = "unsubscribe.task" )
const ( TypeConversationCreated = "conversation.created" TypeMessageDelta = "conversation.message.delta" TypeMessageQueued = "conversation.message.queued" TypeMessageDequeued = "conversation.message.dequeued" TypeMessageCompleted = "conversation.message.completed" TypeConversationError = "conversation.error" TypeTaskStatusChanged = "task.status.changed" TypeTaskStreamDelta = "task.stream.delta" TypeTaskStreamDone = "task.stream.done" TypeSystemError = "system.error" )
const ErrorCodeQueueFull = "queue_full"
ErrorCodeQueueFull marks a conversation error that refused a new message because the conversation's queue is full. It is not a failure of the turn in progress, which is still running — a client must not read it as "the conversation went idle".
const StreamEventDone = "[[DONE]]"
StreamEventDone is sent on the subscription channel when the run finishes (SUCCEEDED/FAILED).
Variables ¶
This section is empty.
Functions ¶
func DecodePayload ¶
DecodePayload unmarshals the envelope's raw payload into the target type T.
Types ¶
type Conn ¶
type Conn struct {
// contains filtered or unexported fields
}
Conn manages a single WebSocket connection for one authenticated user.
type ConnDeps ¶
type ConnDeps struct {
Conversations coreconv.Store
Turns *turnqueue.Registry
// Turner runs one conversation turn. An interface so this package does not
// depend on the service that assembles agents, models, and tools.
Turner Turner
Registry *ConnRegistry
// CORSOrigin is checked on the upgrade. Empty or "*" accepts any origin,
// which is what a deployment serving Portal from the same host has.
CORSOrigin string
}
ConnDeps is everything a live connection needs. It is a struct rather than a Handler because this package should not be able to reach a store it has no use for: a socket creates and reads conversations and runs turns, and that is the whole list.
type ConnRegistry ¶
type ConnRegistry struct {
// contains filtered or unexported fields
}
ConnRegistry tracks active WebSocket connections per user.
A single replica holds only the connections that landed on it. When a publisher is set, an event is published to the shared coordination bus instead of being fanned out inline, and every replica's Consume loop — including this one's — delivers it to the connections that replica holds. That is what makes an event raised on one replica reach a socket on another. See docs/design/server-coordination.md §6.
func NewConnRegistry ¶
func NewConnRegistry() *ConnRegistry
func (*ConnRegistry) Broadcast ¶
func (r *ConnRegistry) Broadcast(spaceID, userID, eventType string, payload any)
Broadcast sends one event to every connection watching the space.
Nothing here picks a single connection. A person reads the Portal from as many tabs as they like and from none at all, and a spacemate reads the same conversation from their own; an event that reaches one of those and not the rest leaves the others showing state that has already changed.
The event is encoded once. With a publisher set it goes to the coordination bus and this replica delivers it through its own Consume loop, so the publish path never fans out locally and no connection is written twice.
func (*ConnRegistry) BroadcastSink ¶
func (r *ConnRegistry) BroadcastSink(spaceID, userID, conversationID string) llm.StreamSink
BroadcastSink streams one turn's deltas to everyone watching the space.
A turn the server starts on its own — reporting a finished run, say — has no socket of its own to write to. This is what it streams through instead, and it costs nothing when nobody is connected.
func (*ConnRegistry) Consume ¶
func (r *ConnRegistry) Consume(incoming <-chan []byte)
Consume delivers envelopes received from the coordination bus to this replica's own connections. It runs until incoming is closed, which the bus does on shutdown.
func (*ConnRegistry) ForUser ¶
func (r *ConnRegistry) ForUser(userID string) []*Conn
func (*ConnRegistry) Register ¶
func (r *ConnRegistry) Register(userID string, c *Conn)
func (*ConnRegistry) SetPublisher ¶
func (r *ConnRegistry) SetPublisher(fn func([]byte))
SetPublisher routes broadcasts through the coordination bus. It is called once at assembly when a Redis coordination backend is configured.
func (*ConnRegistry) Unregister ¶
func (r *ConnRegistry) Unregister(userID string, c *Conn)
type ConversationCreate ¶
type ConversationCreate struct {
Channel string `json:"channel,omitempty"`
Message string `json:"message"`
}
ConversationCreate is the payload for TypeConversationCreate.
type ConversationCreated ¶
type ConversationCreated struct {
ConversationID string `json:"conversation_id"`
}
ConversationCreated is the payload for TypeConversationCreated.
type ConversationError ¶
type ConversationError struct {
ConversationID string `json:"conversation_id,omitempty"`
Error string `json:"error"`
// Code is an optional machine-readable reason. Empty means the turn itself failed.
Code string `json:"code,omitempty"`
}
ConversationError is the payload for TypeConversationError.
type ConversationMessage ¶
type ConversationMessage struct {
ConversationID string `json:"conversation_id"`
Content string `json:"content"`
}
ConversationMessage is the payload for TypeConversationMessage.
type Envelope ¶
type Envelope struct {
Type string `json:"type"`
Payload json.RawMessage `json:"payload"`
}
Envelope is the wire format for every WebSocket message (both directions).
type MessageCompleted ¶
type MessageCompleted struct {
ConversationID string `json:"conversation_id"`
// QueuedRemaining is how many messages are still waiting for their turn. A
// client uses it to decide whether the conversation is idle or merely between
// turns.
QueuedRemaining int `json:"queued_remaining,omitempty"`
}
MessageCompleted is the payload for TypeMessageCompleted.
type MessageDelta ¶
type MessageDelta struct {
ConversationID string `json:"conversation_id"`
Delta string `json:"delta"`
}
MessageDelta is the payload for TypeMessageDelta.
type MessageDequeued ¶
type MessageDequeued struct {
ConversationID string `json:"conversation_id"`
Content string `json:"content"`
}
MessageDequeued is the payload for TypeMessageDequeued: a queued message is starting its own turn now.
type MessageQueued ¶
type MessageQueued struct {
ConversationID string `json:"conversation_id"`
Content string `json:"content"`
// Position is 1-based: 1 is the next turn to run after the current one.
Position int `json:"position"`
}
MessageQueued is the payload for TypeMessageQueued: the message arrived while a turn was running and will run as its own turn once that one finishes.
type StreamHub ¶
type StreamHub interface {
// Append adds a delta to the buffer and broadcasts to all subscribers for that task.
Append(taskID, delta string)
// Buffer returns the current buffered content for the task. Empty string if task has no buffer or was Done.
Buffer(taskID string) string
// Done marks the task's current run as finished; sends StreamEventDone to subscribers and clears state.
Done(taskID string)
// Subscribe returns a channel of deltas (and finally StreamEventDone) for the task. unsub must be called when done.
Subscribe(taskID string) (events <-chan string, unsub func())
}
StreamHub is the interface for task-scoped stream buffers. Keys are task_id (one active run per task). Implementations may be in-memory (single instance) or backed by Redis (multi-instance).
func NewStreamHub ¶
func NewStreamHub() StreamHub
NewStreamHub returns an in-memory StreamHub. Multi-instance scaling requires a Redis-backed impl.
type SubscribeTask ¶
type SubscribeTask struct {
TaskID string `json:"task_id"`
}
SubscribeTask is the payload for TypeSubscribeTask.
type SystemError ¶
type SystemError struct {
Error string `json:"error"`
}
SystemError is the payload for TypeSystemError.
type TaskStatusChanged ¶
type TaskStatusChanged struct {
TaskID string `json:"task_id"`
Status string `json:"status"`
Title string `json:"title,omitempty"`
}
TaskStatusChanged is the payload for TypeTaskStatusChanged.
type TaskStreamDelta ¶
TaskStreamDelta is the payload for TypeTaskStreamDelta.
type TaskStreamDone ¶
type TaskStreamDone struct {
TaskID string `json:"task_id"`
}
TaskStreamDone is the payload for TypeTaskStreamDone.
type Turner ¶
type Turner interface {
HandleTurn(ctx context.Context, cmd conversation.HandleTurnCmd) (conversation.ConversationResult, error)
}
Turner runs one conversation turn to completion, streaming as it goes.
type UnsubscribeTask ¶
type UnsubscribeTask struct {
TaskID string `json:"task_id"`
}
UnsubscribeTask is the payload for TypeUnsubscribeTask.