session

package
v0.7.1 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jun 8, 2026 License: AGPL-3.0 Imports: 19 Imported by: 0

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, 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 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 context.Context for routing metadata (e.g., provenance). Set it when constructing the event: 

UserMessageEvent{Content: "hello", Ctx: loop.WithProvenance(context.Background(), "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.

State persistence is handled automatically by the Manager via loop.WithState, which binds the thread's state to the Step so that TurnCompleteEvent is appended after every turn. Applications only need a stepFactory when adding custom transforms, handlers, or other loop.Step options: 

store := NewMemoryStore()
prov, _ := openai.New(openai.WithAPIKey(apiKey), openai.WithModel(model))
stepFactory := func(stream *Stream) ([]loop.Option, error) {
	// Build transforms that use stream.Metadata, stream.ID, etc.
	return nil, nil  // use default step with auto-persistence
}
mgr := session.NewManager(store, prov, stepFactory, cognitive.NewTurnProcessor(cognitive.ReActFactory, tracer))

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

stepFactory := func(stream *Stream) ([]loop.Option, error) {
	sp, _ := systemprompt.New(systemprompt.WithContentFunc(func() string {
		if p, ok := stream.GetMetadata("persona"); ok {
			return "You are a " + p + "."
		}
		return "You are a helpful assistant."
	}))
	return []loop.Option{loop.WithTransforms(sp)}, 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)

// Retrieve the conversation history without exposing the internal Thread.
turns := stream.Turns()

// 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

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type Event 

type Event interface {
	Kind() string
	Context() context.Context
}

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 InterceptResult added in v0.7.0 

type InterceptResult struct {
	// Event is the replacement event to continue processing. If nil, the
	// original event is consumed and no further processing occurs.
	Event Event
	// Feedback holds UI-only messages (artifact.Text) that are emitted as
	// FeedbackEvent and never persisted to state.
	Feedback []artifact.Text
}

InterceptResult is the result of an interceptor processing an event.

type Interceptor added in v0.6.0 

type Interceptor interface {
	Intercept(ctx context.Context, event Event) (InterceptResult, error)
}

Interceptor processes events before they enter the LLM pipeline. It receives a session.Event and can either:

  • Return a non-nil Event in the result to rewrite the event
  • Return a nil Event in the result to consume the event (no further processing)
  • Return an error to abort processing

Feedback messages are ephemeral UI messages that are not persisted to state.

type InterceptorFunc added in v0.6.0 

type InterceptorFunc func(ctx context.Context, event Event) (InterceptResult, error)

InterceptorFunc is a function type that implements Interceptor.

func (InterceptorFunc) Intercept added in v0.6.0 

func (f InterceptorFunc) Intercept(ctx context.Context, event Event) (InterceptResult, error)

Intercept delegates to the function.

type InterruptEvent 

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

InterruptEvent represents the user interrupting the current operation.

func (InterruptEvent) Context 

func (e InterruptEvent) Context() context.Context

Context returns the event's context.Context metadata.

func (InterruptEvent) Kind 

func (e InterruptEvent) Kind() string

Kind returns the event kind identifier.

type JSONStore added in v0.2.1 

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

JSONStore persists threads as individual JSON files in a directory.

func NewJSONStore added in v0.2.1 

func NewJSONStore(dir string) (*JSONStore, error)

NewJSONStore creates a new JSONStore backed by the given directory. The directory is created if it does not exist. Thread data is loaded lazily on first access via Get, List, or GetBy.

Malformed or unreadable .json files are silently skipped during access.

func (*JSONStore) Create added in v0.2.1 

func (s *JSONStore) Create() (*Thread, error)

Create generates a new Thread with a random ID and persists it.

func (*JSONStore) Delete added in v0.2.1 

func (s *JSONStore) Delete(id string) bool

Delete removes a thread from the cache and deletes its file.

func (*JSONStore) Get added in v0.2.1 

func (s *JSONStore) Get(id string) (*Thread, bool)

Get retrieves a thread by ID. If not in cache, it attempts to load from disk.

func (*JSONStore) GetBy added in v0.2.1 

func (s *JSONStore) GetBy(key, value string) (*Thread, bool)

GetBy retrieves a thread by a metadata key-value pair. It scans all thread files on disk and returns the first match.

func (*JSONStore) List added in v0.2.1 

func (s *JSONStore) List() ([]*Thread, error)

List returns all threads in the store.

func (*JSONStore) Save added in v0.2.1 

func (s *JSONStore) Save(thread *Thread) error

Save writes the thread to disk atomically (via a temporary file and os.Rename) and updates the in-memory cache. The thread's UpdatedAt timestamp is also advanced.

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 Store, prov provider.Provider, newStep func(*Stream) ([]loop.Option, 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) CreateWithID added in v0.2.1 

func (m *Manager) CreateWithID(id string) (*Stream, error)

CreateWithID creates a new thread with the given ID and an active stream backed by it.

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) GetBy added in v0.2.1 

func (m *Manager) GetBy(key, value string) (*Thread, bool)

GetBy retrieves a thread by a metadata key-value pair.

func (*Manager) GetThread added in v0.2.1 

func (m *Manager) GetThread(id string) (*Thread, bool)

GetThread retrieves a thread by ID.

func (*Manager) List 

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

List returns handles for all active streams.

func (*Manager) ListThreads added in v0.2.1 

func (m *Manager) ListThreads() ([]*Thread, error)

ListThreads returns all stored threads.

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.

type ManagerOption 

type ManagerOption func(*Manager)

ManagerOption configures a Manager.

func WithDefaultMetadata added in v0.2.5 

func WithDefaultMetadata(fn func(*Stream) map[string]string) ManagerOption

WithDefaultMetadata sets a function that provides default metadata for every stream at creation/attachment time. The function receives the stream so it can include the thread ID or other stream-specific values.

func WithInterceptor added in v0.6.0 

func WithInterceptor(interceptor Interceptor) ManagerOption

WithInterceptor sets an interceptor that processes UserMessageEvent events before they enter the LLM inference pipeline for every stream managed by this Manager.

type MemoryStore added in v0.2.1 

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

MemoryStore is an in-memory Store implementation.

func NewMemoryStore added in v0.2.1 

func NewMemoryStore() *MemoryStore

NewMemoryStore creates a new empty in-memory store.

func (*MemoryStore) Create added in v0.2.1 

func (s *MemoryStore) Create() (*Thread, error)

Create generates a new Thread with a random ID and stores it.

func (*MemoryStore) Delete added in v0.2.1 

func (s *MemoryStore) Delete(id string) bool

Delete removes a thread from the store.

func (*MemoryStore) Get added in v0.2.1 

func (s *MemoryStore) Get(id string) (*Thread, bool)

Get retrieves a thread by ID.

func (*MemoryStore) GetBy added in v0.2.1 

func (s *MemoryStore) GetBy(key, value string) (*Thread, bool)

GetBy retrieves a thread by a metadata key-value pair. It performs a linear scan over all threads and returns the first match.

func (*MemoryStore) List added in v0.2.1 

func (s *MemoryStore) List() ([]*Thread, error)

List returns all threads in the store.

func (*MemoryStore) Save added in v0.2.1 

func (s *MemoryStore) Save(thread *Thread) error

Save updates the thread's UpdatedAt and stores it.

type SessionSwitchEvent added in v0.6.0 

type SessionSwitchEvent struct {
	SessionID string
	Ctx       context.Context
}

SessionSwitchEvent signals a cross-session navigation.

func (SessionSwitchEvent) Context added in v0.6.0 

func (e SessionSwitchEvent) Context() context.Context

Context returns the event's context.Context metadata.

func (SessionSwitchEvent) Kind added in v0.6.0 

func (e SessionSwitchEvent) Kind() string

Kind returns the event kind identifier.

func (SessionSwitchEvent) MarshalJSON added in v0.6.0 

func (e SessionSwitchEvent) MarshalJSON() ([]byte, error)

MarshalJSON serializes the event to JSON. It includes the session_id and an optional context envelope with provenance and traceparent.

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 Store added in v0.2.1 

type Store interface {
	// Create generates a new Thread with a random UUID and stores it.
	Create() (*Thread, error)
	// Get retrieves a Thread by ID. The second return value is false
	// if the thread does not exist.
	Get(id string) (*Thread, bool)
	// GetBy retrieves a Thread by a metadata key-value pair.
	// The second return value is false if no thread matches.
	GetBy(key, value string) (*Thread, bool)
	// Save persists the given Thread, updating its UpdatedAt timestamp.
	Save(thread *Thread) error
	// Delete removes a Thread by ID and returns true if it existed.
	Delete(id string) bool
	// List returns all stored Threads.
	List() ([]*Thread, error)
}

Store abstracts persistence for Thread instances. Implementations must be safe for concurrent use.

type Stream 

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

Stream is a per-session primitive that owns the loop.Step, 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) GetMetadata added in v0.2.1 

func (s *Stream) GetMetadata(key string) (string, bool)

GetMetadata retrieves a metadata value from the underlying thread.

func (*Stream) ID 

func (s *Stream) ID() string

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

func (*Stream) LoadTurns added in v0.6.2 

func (s *Stream) LoadTurns(turns []state.Turn)

LoadTurns replaces the thread's turn state with the provided slice. It acquires the stream's mutex to ensure thread-safe state mutation.

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) Save added in v0.2.1 

func (s *Stream) Save() error

Save persists the underlying thread to the store.

func (*Stream) SetMetadata added in v0.2.1 

func (s *Stream) SetMetadata(key, value string)

SetMetadata sets a metadata value on the underlying thread.

func (*Stream) Submit added in v0.1.3 

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.

Subscribe is live-only: it delivers events from the point of subscription onward and does not replay historical events. Conduits that need historical state should fetch it via Turns() before subscribing, or load it via LoadTurns() after external mutations (e.g. compaction).

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
}

func (*Stream) Turns added in v0.5.2 

func (s *Stream) Turns() []state.Turn

Turns returns a defensive (shallow) copy of the thread's turn history. The slice of Turns is copied, but each Turn's Artifacts slice is shared. Callers should treat the returned artifacts as immutable.

type Thread added in v0.2.1 

type Thread struct {
	// ID is the unique identifier for this thread (random UUID).
	ID string
	// State holds the mutable thread turn history.
	State *state.Buffer
	// CreatedAt is set when the thread is first created.
	CreatedAt time.Time
	// UpdatedAt is advanced on every successful Save.
	UpdatedAt time.Time
	// Metadata holds arbitrary key-value pairs for conduit-specific
	// thread mapping (e.g., external system identifiers).
	Metadata map[string]string
}

Thread represents a persistent thread with identity, state, and metadata.

func (*Thread) GetMetadata added in v0.2.1 

func (c *Thread) GetMetadata(key string) (string, bool)

GetMetadata retrieves a metadata value from the thread.

func (*Thread) MarshalJSON added in v0.2.1 

func (c *Thread) MarshalJSON() ([]byte, error)

MarshalJSON serializes the thread to JSON.

func (*Thread) SetMetadata added in v0.2.1 

func (c *Thread) SetMetadata(key, value string)

SetMetadata sets a metadata value on the thread.

func (*Thread) UnmarshalJSON added in v0.2.1 

func (c *Thread) UnmarshalJSON(data []byte) error

UnmarshalJSON deserializes a thread from JSON.

type TurnProcessor 

type TurnProcessor func(ctx context.Context, executor loop.TurnExecutor, 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 holds the raw user message text.
	Content string

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

UserMessageEvent represents the user submitting a text message.

func (UserMessageEvent) Context 

func (e UserMessageEvent) Context() context.Context

Context returns the event's context.Context metadata.

func (UserMessageEvent) Kind 

func (e UserMessageEvent) Kind() string

Kind returns the event kind identifier.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.