session

Documentation

Overview

Package session provides the per-conversation primitives that replace the legacy junk.Thread / junk.Stream / junk.Manager combination. A Session is a per-conversation handle that owns its identity, ledger thread, conduit-mapping metadata, and the long-lived loop.Step used for subscriber fanout. A Runner is the process-wide wire that drives inference against sessions.

The primitives intentionally split "data" from "control": Session is pure data; Runner is the active component that handles events, runs interceptors (e.g. slash commands), builds ephemeral agents, and routes output events through sink routers. This separation lets callers compose the parts they need without inheriting a fat Manager.

Index

• Constants
• Variables
• type AgentFactory
• type DefaultFactory
  • func NewDefaultFactory(p provider.Provider, pat cognitive.Pattern, tr trace.Tracer) *DefaultFactory
  • func (f *DefaultFactory) Build(sess *Session) (*agent.Agent, error)
• type Event
• type InterceptResult
• type Interceptor
• type InterruptEvent
  • func (e InterruptEvent) Context() context.Context
  • func (e InterruptEvent) Kind() string
• type Option
• type Runner
  • func NewRunner(opts ...RunnerOption) *Runner
  • func (r *Runner) Close() error
  • func (r *Runner) Create(ctx context.Context, id string, sess *Session) error
  • func (r *Runner) Get(id string) (*Session, error)
  • func (r *Runner) Register(sess *Session)
  • func (r *Runner) Run(ctx context.Context, sess *Session, evt Event) error
  • func (r *Runner) WithSink(kinds []string, fn SinkFunc) func()
• type RunnerOption
  • func WithFactory(f AgentFactory) RunnerOption
  • func WithInterceptor(i Interceptor) RunnerOption
• type Session
  • func New(id string, thread *ledger.Thread, opts ...Option) *Session
  • func (s *Session) AllMetadata() map[string]string
  • func (s *Session) Close() error
  • func (s *Session) Emitter() loop.Emitter
  • func (s *Session) GetMetadata(key string) (string, bool)
  • func (s *Session) ID() string
  • func (s *Session) Run(ctx context.Context, evt Event) error
  • func (s *Session) SetMetadata(key, value string)
  • func (s *Session) Subscribe(kinds ...string) <-chan loop.OutputEvent
  • func (s *Session) Thread() *ledger.Thread
  • func (s *Session) Turns() []ledger.Turn
• type SessionSwitchEvent
  • func (e SessionSwitchEvent) Context() context.Context
  • func (e SessionSwitchEvent) Kind() string
• type SinkFunc
• type SinkRouter
  • func (r *SinkRouter) Add(kinds []string, fn SinkFunc) func()
  • func (r *SinkRouter) Deliver(sessionID string, event loop.OutputEvent)
• type UserMessageEvent
  • func (e UserMessageEvent) Context() context.Context
  • func (e UserMessageEvent) Kind() string

Constants

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

Model metadata keys written by slash commands (e.g. x/tool/set_model) and read by the factory to derive a per-turn model spec.

These constants are the framework contract. Slash handlers write under these keys; the factory reads them and translates the metadata into a models.Spec. They are duplicated in junk/stream.go for backwards compatibility; that duplication is removed in Task 10.

Variables

var ErrSessionAlreadyExists = errors.New("session already exists")

Create creates a new session with the given ID. The session is registered and returned. If a session with the same ID already exists, ErrSessionAlreadyExists is returned (the existing session is left untouched).

var ErrSessionNotFound = errors.New("session not found")

ErrSessionNotFound is the sentinel returned by Runner.Get when no session matches the requested identifier. Callers can detect this case with errors.Is.

Types

type AgentFactory

type AgentFactory interface {
	Build(sess *Session) (*agent.Agent, error)
}

AgentFactory builds an *agent.Agent for a given Session. The factory reads session.Metadata and other session state to construct the agent's spec, transforms, and other per-turn configuration.

The factory is invoked by Runner on every cognitive-pattern invocation; it is the canonical place where "session configuration influences agent design" is expressed.

type DefaultFactory

type DefaultFactory struct {
	Provider provider.Provider
	Pattern  cognitive.Pattern
	Tracer   trace.Tracer
}

DefaultFactory builds an agent from a Session's metadata, using the factory's configured Provider, Pattern, and Tracer. It is the reference implementation of AgentFactory.

Path B (locked in): the agent continues to construct its own internal *loop.Step at agent.New time. Long-term subscribers attach to the session's step (via Session.Subscribe), which is distinct from the agent's step. Re-routing the agent's events through the session's step requires an agent.WithStep option (or equivalent), which is tracked separately as #523. Until that lands, the agent's internal step is unused for fanout — subscribers read events from the session's step directly, which receives lifecycle events emitted by the Runner and any PropertiesEvents emitted by Session.SetMetadata.

func NewDefaultFactory

func NewDefaultFactory(p provider.Provider, pat cognitive.Pattern, tr trace.Tracer) *DefaultFactory

NewDefaultFactory constructs a DefaultFactory.

func (*DefaultFactory) Build

func (f *DefaultFactory) Build(sess *Session) (*agent.Agent, error)

Build constructs an *agent.Agent for the given session. It reads the session's metadata to derive a model spec, then constructs the agent with the factory's configured provider, pattern, and tracer.

The session is bound as the agent's default state (for the auto-append behavior of TurnCompleteEvent). Per-call overrides by the pattern still win over the bound spec.

type Event

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

Event is the base interface for ingress events to a Session.

type InterceptResult

type InterceptResult struct {
	Event  Event
	Notice []loop.Notice
}

InterceptResult is the result of an Interceptor processing an event.

Event, when non-nil, replaces the original event for downstream processing. A nil Event means the original event was consumed and no further processing occurs.

Notice carries ephemeral, user-visible messages emitted as loop.NoticeEvent after Intercept returns. Each Notice carries a Severity that conduits use to pick a rendering style (Success, Info, Warn, Error).

type Interceptor

type Interceptor interface {
	Intercept(ctx context.Context, event Event, sess *Session, emitter loop.Emitter) (InterceptResult, error)
}

Interceptor processes events before they enter the LLM pipeline. It receives the session so it can call session-scoped methods like SetMetadata, and an emitter for signaling activity.

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 Option

type Option func(*Session)

Option configures a Session at construction.

type Runner

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

Runner is the process-wide wire that drives inference against Sessions. It owns:

• the AgentFactory that builds per-pattern agents from session metadata
• a chain of Interceptors (slash commands, etc.) that run before the agent
• a SinkRouter that forwards session events to external subscribers (conduits)
• a registry of active sessions indexed by ID

Conduits interact with the Runner, not the Session, for inference-driving operations.

func NewRunner

func NewRunner(opts ...RunnerOption) *Runner

NewRunner constructs a Runner with the given options. The Runner's internal SinkRouter is always present; all other fields default to nil until configured.

func (*Runner) Close

func (r *Runner) Close() error

Close closes all sessions and releases resources. The Runner is unusable after Close.

func (*Runner) Create

func (r *Runner) Create(ctx context.Context, id string, sess *Session) error

func (*Runner) Get

func (r *Runner) Get(id string) (*Session, error)

Get retrieves an active session by ID.

func (*Runner) Register

func (r *Runner) Register(sess *Session)

Register inserts a session into the registry without checking for duplicates. It is the caller's responsibility to ensure uniqueness.

func (*Runner) Run

func (r *Runner) Run(ctx context.Context, sess *Session, evt Event) error

Run is the canonical entry point for inference-driving events. It invokes the registered interceptors in order, builds an agent from the session's metadata via the configured AgentFactory, runs the agent, and broadcasts emitted events through the SinkRouter.

Run is synchronous from the caller's perspective: it returns once the agent's Run has returned (or the event was consumed by an interceptor). It does NOT use the session's work queue; events are processed in the caller's goroutine.

func (*Runner) WithSink

func (r *Runner) WithSink(kinds []string, fn SinkFunc) func()

WithSink registers a sink on the Runner's SinkRouter. The returned function unregisters the sink when called.

type RunnerOption

type RunnerOption func(*Runner)

RunnerOption configures a Runner.

func WithFactory

func WithFactory(f AgentFactory) RunnerOption

WithFactory sets the AgentFactory used to build per-pattern agents. Required; NewRunner returns a Runner that panics if no factory is configured before the first Run.

func WithInterceptor

func WithInterceptor(i Interceptor) RunnerOption

WithInterceptor appends an Interceptor to the Runner's chain. The Runner invokes interceptors in registration order; the first to return a non-nil InterceptResult.Event replaces the original event for downstream processing.

type Session

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

Session is the per-conversation primitive. It owns the identity, the ledger thread, the conduit-mapping metadata, the long-lived loop.Step used for subscriber fanout, and the work queue that serializes ingress events.

Session is not safe for concurrent closure. Submit, Subscribe, metadata access, and close are all serialized via Session.mu. The internal worker goroutine is the only reader of the work queue.

func New

func New(id string, thread *ledger.Thread, opts ...Option) *Session

New constructs a Session with the given identity and thread. The thread must not be nil. The session is fully initialized: an internal loop.Step is created for fanout, and a worker goroutine is started to drain the work queue.

Future tasks will wire agent invocation into the worker. For now, the worker is a stub that emits a LifecycleEvent{Phase: "done"} after processing each event.

func (*Session) AllMetadata

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

AllMetadata returns a defensive copy of the metadata map. Mutating the returned map does not affect the session's state.

func (*Session) Close

func (s *Session) Close() error

Close closes the session's step, drains the worker, and marks the session as closed. Subsequent calls to Run return errSessionClosed; Subscribe returns an immediately-closed channel. Close is safe to call multiple times.

func (*Session) Emitter

func (s *Session) Emitter() loop.Emitter

Emitter returns a loop.Emitter that publishes events through the session's step. Interceptors (slash commands, etc.) receive this in their Intercept call to publish events (e.g. PropertiesEvent) during pre-LLM processing.

func (*Session) GetMetadata

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

GetMetadata retrieves a metadata value.

func (*Session) ID

func (s *Session) ID() string

ID returns the session's unique identifier.

func (*Session) Run

func (s *Session) Run(ctx context.Context, evt Event) error

Run enqueues an event for processing by the session's worker. It returns immediately; processing happens asynchronously. If the session is closed, Run returns errSessionClosed.

func (*Session) SetMetadata

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

SetMetadata sets a metadata value and emits a PropertiesEvent so subscribers (TUI status bar, HTTP web UI, etc.) react to the change. The event is emitted via the session's loop.Step FanOut.

func (*Session) Subscribe

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

Subscribe returns a filtered output event channel from the session's loop.Step FanOut. If no kinds are provided, the channel receives all events regardless of kind. If the session 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.

func (*Session) Thread

func (s *Session) Thread() *ledger.Thread

Thread returns the session's ledger.Thread. The returned pointer is the same one passed to New; the session does not copy.

func (*Session) Turns

func (s *Session) Turns() []ledger.Turn

Turns returns the active-path projection of the session's tree, as a defensive copy.

type SessionSwitchEvent

type SessionSwitchEvent struct {
	SessionID string
	Ctx       context.Context
}

SessionSwitchEvent signals a cross-session navigation. Slash handlers emit it to redirect the conduit to another session.

func (SessionSwitchEvent) Context

func (e SessionSwitchEvent) Context() context.Context

Context returns the event's context.Context metadata.

func (SessionSwitchEvent) Kind

func (e SessionSwitchEvent) Kind() string

Kind returns the event kind identifier.

type SinkFunc

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

SinkFunc receives output events from sessions, together with the originating session ID. It may be invoked concurrently for multiple sessions. Implementations should be non-blocking or short-lived.

type SinkRouter

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

SinkRouter is a process-wide registry of external subscribers (typically conduits) that want to receive session events. It is concurrency-safe; subscribers may be added or removed at any time.

Each registered sink receives every event from a session whose Session.Subscribe channel yields the event, filtered by the kinds the sink registered for. An empty kinds list means all events.

Sinks are invoked on the session's emit goroutine; implementations must be non-blocking or short-lived. A panicking sink is recovered and logged.

func (*SinkRouter) Add

func (r *SinkRouter) Add(kinds []string, fn SinkFunc) func()

Add registers a sink. It returns a function that removes the sink when called. Calling the returned function more than once is safe and idempotent.

func (*SinkRouter) Deliver

func (r *SinkRouter) Deliver(sessionID string, event loop.OutputEvent)

Deliver invokes all registered sinks for the given sessionID and event, applying per-sink kind filters. Sinks that panic are recovered and logged.

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.