session

package
v0.12.0 Latest Latest
Warning

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

 Go to latest Published: Jun 18, 2026 License: AGPL-3.0 Imports: 21 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

View Source 
const (
	MetadataKeyModelName            = "ore.model.name"
	MetadataKeyModelThinkingLevel   = "ore.model.thinking_level"
	MetadataKeyModelTemperature     = "ore.model.temperature"
	MetadataKeyModelMaxOutputTokens = "ore.model.max_output_tokens"
)

Model metadata keys. These are the framework contract keys that the slash command in x/tool/set_model (and any other session-level model configurator) writes into Thread.Metadata. The session package owns the canonical form; downstream tools import the constants when they need to set a value, or write the literal string. The "ore.model." prefix reserves the namespace for framework-level model configuration.

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, stream *Stream, emitter loop.Emitter) (InterceptResult, error)
}

Interceptor processes events before they enter the LLM pipeline. It receives a session.Event, the active *session.Stream, and a loop.Emitter for signaling activity, 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.

The *session.Stream parameter is the stream that owns the in-flight event, so interceptors can call stream-scoped methods like SetMetadata. It is never nil for a configured interceptor — stream.processOne passes the active stream unconditionally.

type InterceptorFunc added in v0.6.0 

type InterceptorFunc func(ctx context.Context, event Event, stream *Stream, emitter loop.Emitter) (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, stream *Stream, emitter loop.Emitter) (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) AllMetadata added in v0.11.0 

func (s *Stream) AllMetadata() map[string]string

AllMetadata returns a defensive copy of the thread's metadata map. Mutating the returned map does not affect the underlying thread. The stream mutex is held for the duration of the copy so a concurrent SetMetadata cannot race with the read. Conduits that need to seed their view at Start time (e.g. the TUI status bar) should call AllMetadata() before stream.Subscribe, since Subscribe is live-only and does not replay historical PropertiesEvents.

func (*Stream) AppendTurn added in v0.12.0 

func (s *Stream) AppendTurn(ctx context.Context, role state.Role, artifacts ...artifact.Artifact) error

AppendTurn records a turn into the thread state and broadcasts a TurnCompleteEvent to all subscribers of the stream. It is the canonical entry point for external producers (slash handlers, compaction, future tool-side producers) that need to inject turns without going through step.Turn / step.Submit.

Behavior:

  • The TurnCompleteEvent is emitted via s.step.Emit. The stream's default OnEmit (registered by the Manager) appends the turn to thread.State synchronously, before the event reaches the async FanOut. Subscribers see the event with the canonical state already updated, so live consumers (TUI, telemetry) do not need to call ReloadHistory to observe the new turn.
  • The thread is NOT auto-saved. Callers that want the change to persist to the store should call Save() explicitly, or rely on the stream's normal save flow after the next Process() call. This matches the rest of the Stream's contract: state mutations are persisted by the inference pipeline, not by ad-hoc producers.

Errors:

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

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

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) Spec added in v0.12.0 

func (s *Stream) Spec() (models.Spec, bool)

Spec derives a models.Spec from the thread's metadata. The bool result is false when no recognized metadata keys are set; the caller should use the loop's default in that case. The framework does not merge metadata into a base Spec; the session is responsible for producing the effective Spec for each turn.

Recognized keys (all private to this package; documented here as the framework contract): 

"ore.model.name"              → Spec.Name
"ore.model.thinking_level"    → Spec.ThinkingLevel
"ore.model.temperature"       → Spec.Temperature (parsed float)
"ore.model.max_output_tokens" → Spec.MaxOutputTokens (parsed int)

Unknown keys are ignored. The slash command in x/tool/set_model is the canonical writer; see MetadataKeyModelName for the "set the model" entry point.

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, step *loop.Step, st state.State, prov provider.Provider, spec models.Spec) (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, provider, and a models.Spec derived from the session's metadata (or the empty zero value when no metadata is set, in which case the step's configured default spec applies).

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.