session

Documentation

Overview

Package session provides the session layer handler that routes WebSocket protocol messages to domain-specific handlers (conversation, intent, etc.).

Package session provides the session layer handler that routes WebSocket protocol messages to domain-specific handlers.

Index

• Constants
• Variables
• type EditHandler
• type Handler
  • func New(opts ...Option) *Handler
  • func (h *Handler) HandleMessage(ctx context.Context, connID string, data []byte)
  • func (h *Handler) OnConnectionClose(_ context.Context, connID string)
  • func (h *Handler) SendNonPersistentUpdate(ctx context.Context, scope, scopeID string, kind string, payload []byte) error
  • func (h *Handler) SendPersistentUpdate(ctx context.Context, scope, scopeID string, kind string, payload []byte, ...) (uint64, error)
• type MethodHandler
• type Option
  • func WithCache(c cache.Cache) Option
  • func WithCanceller(c cancelCanceller) Option
  • func WithConnectionManager(cm *websocket.ConnectionManager) Option
  • func WithEditHandler(eh EditHandler) Option
  • func WithFetchDebounceDuration(d time.Duration) Option
  • func WithLogger(l *slog.Logger) Option
  • func WithMaxContentLength(n int) Option
  • func WithPendingResults(pr PendingResults) Option
  • func WithQueue(q task.TaskQueue) Option
  • func WithStore(s store.Store) Option
  • func WithWebhookMaxRetry(n int) Option
  • func WithWebhookURL(url string) Option
• type PendingResults
• type ResultPayload

Constants

const MaxFetchRange = 1000

MaxFetchRange is the maximum number of seq values that can be fetched in a single request.

Variables

var (
	// ErrSeqAllocation is returned when sequence or message ID allocation fails.
	ErrSeqAllocation = errors.New("allocate seq")
)

Sentinel errors for session operations.

Types

type EditHandler

type EditHandler interface {
	// Handle processes an edit_message request.
	Handle(ctx context.Context, connID, conversationID string, messageID uint64, newContent string) error
}

EditHandler processes edit_message requests. This interface is defined locally to avoid importing the agent package.

type Handler

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

Handler routes incoming WebSocket messages to the appropriate method handler.

func New

func New(opts ...Option) *Handler

New creates a Handler with the given options and registers all method routes. Defaults logger to slog.Default() if not provided.

func (*Handler) HandleMessage

func (h *Handler) HandleMessage(ctx context.Context, connID string, data []byte)

HandleMessage is the entry point for incoming WebSocket messages. It parses, validates, and routes the message to the appropriate handler.

func (*Handler) OnConnectionClose

func (h *Handler) OnConnectionClose(_ context.Context, connID string)

OnConnectionClose is called when a connection is closed. It cleans up any pending fetch debounce state. This is the public entry point for connection lifecycle cleanup. Should be called from the server layer when a connection closes.

func (*Handler) SendNonPersistentUpdate

func (h *Handler) SendNonPersistentUpdate(
	ctx context.Context,
	scope, scopeID string,
	kind string,
	payload []byte,
) error

SendNonPersistentUpdate publishes an update directly without seq allocation or database persistence. Used for typing indicators and retry signals.

Conversation-scope fanout: when scope is "conversation", the update is published to each member's user channel (no persistence for non-persistent kinds).

Returns an error if the publish fails.

func (*Handler) SendPersistentUpdate

func (h *Handler) SendPersistentUpdate(
	ctx context.Context,
	scope, scopeID string,
	kind string,
	payload []byte,
	convID string,
) (uint64, error)

SendPersistentUpdate allocates a sequence number, persists the update in a database transaction, and publishes it via Pub/Sub.

For update kinds that require a message ID (tool_call, result, reply, error), pass a non-empty convID to allocate both seq and msgID via IncrSeqAndMsgID. For other persistent kinds (conversation_created, etc.), pass empty convID.

Conversation-scope fanout: when scope is "conversation", the update is fanned out to each conversation member's user channel. Each member gets their own per-user copy (scope=user, scopeID=member, seq=member's user seq) so that fetch_updates (which queries user scope) can recover missed updates after reconnection. This is the "write fanout" design: no conversation channels.

The publish step happens outside the transaction — a publish failure is logged but does not cause the method to return an error (the update is already persisted).

type MethodHandler

type MethodHandler func(ctx context.Context, conn *websocket.Connection, req *protocol.Request) *protocol.ResponseBuilder

MethodHandler is a function that handles a specific protocol method. Returns a ResponseBuilder to be marshaled and sent, or nil if the handler already sent the response (e.g., GetMaxSeq).

type Option

type Option func(*options)

Option configures a Handler.

func WithCache

func WithCache(c cache.Cache) Option

WithCache sets the Cache dependency.

func WithCanceller

func WithCanceller(c cancelCanceller) Option

WithCanceller sets the cancel registry for intent cancellation.

func WithConnectionManager

func WithConnectionManager(cm *websocket.ConnectionManager) Option

WithConnectionManager sets the ConnectionManager dependency.

func WithEditHandler

func WithEditHandler(eh EditHandler) Option

WithEditHandler sets the EditHandler dependency for edit_message routing.

func WithFetchDebounceDuration

func WithFetchDebounceDuration(d time.Duration) Option

WithFetchDebounceDuration sets the debounce duration for fetch_updates coalescing. Default: 300ms.

func WithLogger

func WithLogger(l *slog.Logger) Option

WithLogger sets the logger.

func WithMaxContentLength

func WithMaxContentLength(n int) Option

WithMaxContentLength sets the maximum allowed length for user message content.

func WithPendingResults

func WithPendingResults(pr PendingResults) Option

WithPendingResults sets the pending tool results registry for result routing.

func WithQueue

func WithQueue(q task.TaskQueue) Option

WithQueue sets the TaskQueue dependency.

func WithStore

func WithStore(s store.Store) Option

WithStore sets the Store dependency.

func WithWebhookMaxRetry

func WithWebhookMaxRetry(n int) Option

WithWebhookMaxRetry sets the maximum retry count for webhook delivery.

func WithWebhookURL

func WithWebhookURL(url string) Option

WithWebhookURL sets the webhook delivery URL for intent completion notifications.

type PendingResults

type PendingResults interface {
	// Register adds a pending tool execution and returns a channel to receive the result.
	// The connID and userID are stored for ownership validation.
	Register(requestID, toolName, connID, userID string) <-chan task.ToolResult
	// Complete delivers a result to a pending tool execution.
	// Returns ErrPendingResultOwnership if the submitting connID/userID don't match.
	Complete(requestID string, result task.ToolResult, connID, userID string) (bool, error)
}

PendingResults manages pending tool execution results. This interface is defined locally to avoid importing the agent package.

type ResultPayload

type ResultPayload struct {
	RequestID string          `json:"request_id" validate:"required"`
	Status    string          `json:"status" validate:"required"` // success / error
	Output    json.RawMessage `json:"output,omitempty"`
	Error     string          `json:"error,omitempty"`
}

ResultPayload is the payload for the result method from the client.