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 (Process) and egress (Subscribe) plus lifecycle controls (Cancel, Close).
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 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"}}
Typical composition:
store := thread.NewMemoryStore()
prov := openai.New(apiKey, model)
stepFactory := func(*thread.Thread) (*loop.Step, error) { return loop.New(), 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(), 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")
// Process an event via the Stream handle.
_ = stream.Process(ctx, 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)
Index ¶
- Variables
- type Event
- type InterruptEvent
- type Manager
- func (m *Manager) Attach(threadID string) (*Stream, error)
- func (m *Manager) Check(sessionID string) 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
- type TurnProcessor
- type UserMessageEvent
Constants ¶
This section is empty.
Variables ¶
var ErrSessionBusy = errors.New("session is busy processing another turn")
ErrSessionBusy is returned when a stream is already processing a turn.
Functions ¶
This section is empty.
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 ¶
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) Check ¶
Check verifies that the stream exists and is not busy. It returns nil if the stream is ready to process.
Errors:
- "session not found" if the stream does not exist
- ErrSessionBusy if the stream is already processing a turn
func (*Manager) Close ¶
Close closes a stream and removes it from the active map. The underlying thread is NOT deleted from the store.
func (*Manager) Create ¶
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 ¶
Get returns the active Stream for a given session ID. An error is returned if the stream does not exist.
func (*Manager) RegisterSink ¶
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.
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) and egress (Subscribe) for the session, plus lifecycle controls (Cancel, Close).
func (*Stream) Close ¶
Close closes the stream's Step and marks it as closed. The underlying thread is NOT deleted from the store.
func (*Stream) Emit ¶
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. Unlike Process(), Emit() does not check whether the stream is busy: handlers running during an active turn may need to emit events (e.g., session-switch signals from slash commands).
Errors:
- "session %s is closed" if the stream has been closed
func (*Stream) Process ¶
Process submits the event to the stream's state and runs the inference pipeline. The stream must not be busy. Context cancellation aborts the running TurnProcessor.
Errors:
- ErrSessionBusy if the stream is already processing a turn
- "unsupported event kind" for unknown event types
- "process event: ..." wrapping any TurnProcessor or save error
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.
Source Files