session

Documentation

Overview

Package session provides the Stream and Manager primitives that orchestrate per-session inference and session lifecycle in the ore framework.

Stream is a per-session primitive that owns the loop.Step, thread.Thread, TurnProcessor, and provider for a single active conversation. It provides ingress (Submit, Process) and egress (Subscribe) plus lifecycle controls (Cancel, Close).

Submit enqueues an event into the stream's internal FIFO queue and returns immediately. A single worker goroutine processes events one at a time in order, so concurrent submissions are naturally serialized without dropping messages.

The worker goroutine starts lazily on the first Submit or Process call via sync.Once, avoiding goroutine overhead for idle streams. Submit returns an error only if the stream has been closed; it never returns ErrSessionBusy (which has been removed).

Process is a convenience wrapper around Submit that blocks until the worker has finished processing the enqueued event. It is useful when the caller needs to know when a turn is complete (e.g. HTTP handlers that must close an NDJSON response stream).

Manager is a factory/registry for Stream handles. It creates and manages active streams, each pairing a persistent thread.Thread with an ephemeral loop.Step. Applications configure a Manager with a provider, step factory, and cognitive pattern (TurnProcessor). Conduits obtain a *Stream from the Manager (via Create, Attach, or Get) and invoke Submit, Process, Subscribe, Cancel, and Close on that handle, never touching loop.Step directly.

Migration note: the Session interface has been removed. Use *session.Stream directly. Event types (Event, UserMessageEvent, InterruptEvent) have moved from the conduit package to session. All event types carry an optional loop.EventContext for routing metadata (e.g., provenance). Set it when constructing the event:

UserMessageEvent{Content: "hello", Ctx: loop.EventContext{Provenance: "slack-123"}}

Lifecycle events:

LifecycleEvent signals phase transitions for the inference pipeline:
"submitted" (message accepted), "streaming" (first artifact arrived),
and "done" (turn or pipeline complete). Conduits should subscribe to
it to drive UI state without inferring lifecycle from data events.

To persist state across turns, wire an OnEmit callback that appends TurnCompleteEvent to the thread's state buffer. Typical composition:

store := thread.NewMemoryStore()
prov, _ := openai.New(openai.WithAPIKey(apiKey), openai.WithModel(model))
stepFactory := func(thr *thread.Thread) (*loop.Step, error) {
	return loop.New(loop.WithOnEmit(func(ctx context.Context, event loop.OutputEvent) {
		if tc, ok := event.(loop.TurnCompleteEvent); ok {
			thr.State.Append(tc.Turn.Role, tc.Turn.Artifacts...)
		}
	})), nil
}
mgr := session.NewManager(store, prov, stepFactory, cognitive.NewTurnProcessor())

The factory receives *thread.Thread so it can bind per-session state. For example, a factory can close over the thread to inject a dynamic system prompt that reads from thread.GetMetadata("persona"):

stepFactory := func(thr *thread.Thread) (*loop.Step, error) {
	// Build transforms that use thr.Metadata, thr.ID, etc.
	return loop.New(loop.WithOnEmit(func(ctx context.Context, event loop.OutputEvent) {
		if tc, ok := event.(loop.TurnCompleteEvent); ok {
			thr.State.Append(tc.Turn.Role, tc.Turn.Artifacts...)
		}
	})), nil
}

// Obtain a *Stream from the manager.
stream, _ := mgr.Create()

// Subscribe to output events via the Stream handle.
ch := stream.Subscribe("text_delta", "turn_complete")

// Submit an event via the Stream handle (non-blocking).
_ = stream.Submit(UserMessageEvent{Content: "hello"})

// HTTP conduit composes with the Manager. UI is enabled by default.
c, _ := httpc.New(mgr, httpc.WithAddr(":8080"))
_ = c.Start(ctx)

// TUI conduit composes with the Manager.
tuiConduit, _ := tui.New(mgr)
_ = tuiConduit.Start(ctx)

// Emit custom output events (e.g. status updates) into the stream's FanOut.
_ = stream.Emit(ctx, loop.PropertiesEvent{Properties: map[string]string{"thread_id": stream.ID()}})

Index

• type Event
• type InterruptEvent
  • func (e InterruptEvent) Context() loop.EventContext
  • func (e InterruptEvent) Kind() string
• type Manager
  • func NewManager(store thread.Store, prov provider.Provider, ...) *Manager
  • func (m *Manager) Attach(threadID string) (*Stream, error)
  • func (m *Manager) Close(sessionID string) error
  • func (m *Manager) Create() (*Stream, error)
  • func (m *Manager) Get(sessionID string) (*Stream, error)
  • func (m *Manager) List() []*Stream
  • func (m *Manager) RegisterSink(kinds []string, fn SinkFunc) func()
  • func (m *Manager) Store() thread.Store
• type ManagerOption
• type SinkFunc
• type Stream
  • func (s *Stream) Cancel() error
  • func (s *Stream) Close() error
  • func (s *Stream) Emit(ctx context.Context, event loop.OutputEvent) error
  • func (s *Stream) ID() string
  • func (s *Stream) Process(ctx context.Context, event Event) error
  • func (s *Stream) Submit(event Event) error
  • func (s *Stream) Subscribe(kinds ...string) <-chan loop.OutputEvent
• type TurnProcessor
• type UserMessageEvent
  • func (e UserMessageEvent) Context() loop.EventContext
  • func (e UserMessageEvent) Kind() string

Types

type Event

type Event interface {
	Kind() string
	Context() loop.EventContext
}

Event is the base interface for all ingress events to a session Stream. Custom event types can be defined in other packages by implementing the public Kind() and Context() methods.

type InterruptEvent

type InterruptEvent struct {
	// Ctx carries the provenance/context metadata for the interrupt event.
	Ctx loop.EventContext
}

InterruptEvent represents the user interrupting the current operation.

func (InterruptEvent) Context

func (e InterruptEvent) Context() loop.EventContext

Context returns the event's loop.EventContext metadata.

func (InterruptEvent) Kind

func (e InterruptEvent) Kind() string

Kind returns the event kind identifier.

type Manager

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

Manager owns the Thread↔Step binding and acts as a factory/registry for Stream handles.

func NewManager

func NewManager(store thread.Store, prov provider.Provider, newStep func(*thread.Thread) (*loop.Step, error), processor TurnProcessor, opts ...ManagerOption) *Manager

NewManager creates a new Manager with the given dependencies.

func (*Manager) Attach

func (m *Manager) Attach(threadID string) (*Stream, error)

Attach gets or creates an active stream for an existing thread. If the thread does not exist in the store, an error is returned. May also return an error if the step factory fails.

func (*Manager) Close

func (m *Manager) Close(sessionID string) error

Close closes a stream and removes it from the active map. The underlying thread is NOT deleted from the store.

func (*Manager) Create

func (m *Manager) Create() (*Stream, error)

Create creates a new thread and an active stream backed by it. Returns an error if the underlying thread store cannot create a thread or if the step factory fails.

func (*Manager) Get

func (m *Manager) Get(sessionID string) (*Stream, error)

Get returns the active Stream for a given session ID. An error is returned if the stream does not exist.

func (*Manager) List

func (m *Manager) List() []*Stream

List returns handles for all active streams.

func (*Manager) RegisterSink

func (m *Manager) RegisterSink(kinds []string, fn SinkFunc) func()

RegisterSink registers a callback that receives OutputEvents from all active and future streams matching the given kinds. An empty kinds slice means all event kinds. Registration also triggers forwarding for any streams that already exist at the time of registration.

It returns a function that removes the sink from the manager. Calling the returned function more than once is safe and idempotent.

func (*Manager) Store

func (m *Manager) Store() thread.Store

Store returns the underlying thread.Store.

type ManagerOption

type ManagerOption func(*Manager)

ManagerOption configures a Manager.

type SinkFunc

type SinkFunc func(streamID string, event loop.OutputEvent)

SinkFunc receives OutputEvents from a specific stream, together with the originating stream ID. It may be invoked concurrently for multiple streams.

type Stream

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

Stream is a per-session primitive that owns the loop.Step, thread.Thread, TurnProcessor, and provider for a single active conversation. It provides ingress (Process, Submit) and egress (Subscribe) for the session, plus lifecycle controls (Cancel, Close).

Events submitted via Submit() are enqueued in an unbounded FIFO queue and processed serially by a single internal worker goroutine. Process() also enqueues but blocks until the event has been fully processed.

func (*Stream) Cancel

func (s *Stream) Cancel() error

Cancel aborts an ongoing turn by cancelling its context.

func (*Stream) Close

func (s *Stream) Close() error

Close closes the stream's Step and marks it as closed. The underlying thread is NOT deleted from the store.

Close cancels any in-flight turn, waits for the worker goroutine to exit, and then closes the step. This ensures all pending Process calls receive errors and no goroutine leak occurs.

func (*Stream) Emit

func (s *Stream) Emit(ctx context.Context, event loop.OutputEvent) error

Emit injects a custom output event into the stream's FanOut, allowing handlers, interceptors, and application logic to emit meta-events that are delivered to all subscribers alongside standard artifact and turn-complete events.

The stream must not be closed.

Errors:

• "session %s is closed" if the stream has been closed

func (*Stream) ID

func (s *Stream) ID() string

ID returns the stream's unique identifier (same as the thread ID).

func (*Stream) Process

func (s *Stream) Process(ctx context.Context, event Event) error

Process enqueues the event and blocks until the worker has finished processing it. Context cancellation aborts the waiting, not the in-flight turn; use Cancel() to abort a running turn.

After the TurnProcessor returns (including all tool-call loops), Process emits a LifecycleEvent{Phase: "done"} to signal pipeline completion before performing save cleanup. Subscribers can use this event for lifecycle signalling (audio notifications, UI state finalization).

Errors:

• "session %s is closed" if the stream has been closed
• "unsupported event kind" for unknown event types
• "process event: ..." wrapping any TurnProcessor or save error

func (*Stream) Submit

func (s *Stream) Submit(event Event) error

Submit enqueues the event in the stream's unbounded FIFO queue and returns immediately. A single internal worker goroutine drains the queue and processes events serially, one at a time.

InterruptEvent clears all pending events from the queue before being enqueued itself, and cancels any in-flight turn via Cancel().

Errors:

• "session %s is closed" if the stream has been closed

func (*Stream) Subscribe

func (s *Stream) Subscribe(kinds ...string) <-chan loop.OutputEvent

Subscribe returns a filtered output event channel for the stream's loop.Step FanOut. If no kinds are provided, the channel receives all events regardless of kind. If the stream is closed, the returned channel is immediately closed.

The returned channel is closed when the session is closed. Callers should range over the channel and handle closure:

ch := stream.Subscribe("text_delta", "turn_complete")
for event := range ch {
    // process event
}

type TurnProcessor

type TurnProcessor func(ctx context.Context, step *loop.Step, st state.State, prov provider.Provider) (state.State, error)

TurnProcessor runs the full inference pipeline for a single turn after the user event has been submitted to state. It is called with the stream's loop.Step, state, and provider.

type UserMessageEvent

type UserMessageEvent struct {
	Content string

	// Ctx carries the provenance/context metadata for the user message.
	Ctx loop.EventContext
}

UserMessageEvent represents the user submitting a text message.

func (UserMessageEvent) Context

func (e UserMessageEvent) Context() loop.EventContext

Context returns the event's loop.EventContext metadata.

func (UserMessageEvent) Kind

func (e UserMessageEvent) Kind() string

Kind returns the event kind identifier.