session

type AckMode ¶

type AckMode string

const (
	// AckOnDelivery acknowledges a confirmed delivery only after Sink.OnDelta
	// returns nil. If the turn is interrupted while the callback is still
	// running, that delivery is outside the frozen acknowledged prefix.
	AckOnDelivery AckMode = ""
	// AckExplicit requires the authoritative sink to call Turn.Ack. The sink
	// may acknowledge the cursor currently offered to its OnDelta callback
	// before that callback returns.
	AckExplicit AckMode = "explicit"
)

type AskUserFunc ¶

type AskUserFunc func(context.Context, agent.UserPrompt) (agent.UserReply, error)

AskUserFunc is the turn-scoped user-prompt callback supplied to a Host.

type Authority ¶

type Authority string

const (
	AuthorityObserver      Authority = ""
	AuthorityAuthoritative Authority = "authoritative"
)

type BaseSessionObserver ¶

type BaseSessionObserver struct{}

BaseSessionObserver provides no-op default implementations of every SessionObserver method.

func (BaseSessionObserver) OnSessionClosed(*Session, error)

func (BaseSessionObserver) OnSessionClosing(*Session)

func (BaseSessionObserver) OnSessionStarted(*Session, *Turn)

func (BaseSessionObserver) OnTurnFinished(*Session, *Turn, *agent.Result, error)

type CatalogProvider ¶

type CatalogProvider interface {
	NewCatalog(ctx context.Context, instance *agent.Agent) (tool.Session, error)
}

CatalogProvider constructs the per-session tool catalog. The provider runs once per Session (on its first Start) and its result is attached to every turn's execution context, where session-scoped tools (tool_search) and the inference node's all_tools mode pick it up. The Session borrows the catalog.

The provider returns the session-scoped tool.Session (hosts typically use tool.Assembly.NewSession), so the dynamic injection view and the plain catalog share one contract.

type CatalogProviderFunc ¶

type CatalogProviderFunc func(ctx context.Context, instance *agent.Agent) (tool.Session, error)

CatalogProviderFunc adapts a plain function to CatalogProvider.

func (f CatalogProviderFunc) NewCatalog(ctx context.Context, instance *agent.Agent) (tool.Session, error)

type DeliveryCursor ¶

type DeliveryCursor uint64

DeliveryCursor is a turn-global, contiguous confirmed-delivery position.

func DeliveryCursorFromEnvelope(env event.Envelope) (DeliveryCursor, error)

type HostFactory ¶

type HostFactory interface {
	NewHost(context.Context, HostRequest) (agent.Host, error)
}

HostFactory constructs a fresh Host for one turn.

type HostFactoryFunc ¶

type HostFactoryFunc func(context.Context, HostRequest) (agent.Host, error)

HostFactoryFunc adapts a function to HostFactory.

func (f HostFactoryFunc) NewHost(ctx context.Context, request HostRequest) (agent.Host, error)

type HostRequest ¶

type HostRequest struct {
	Key        Key
	RunID      string
	Interrupts <-chan agent.Interrupt
	AskUser    AskUserFunc
}

HostRequest contains the per-turn capabilities needed to construct a Host.

func (r HostRequest) Validate() error

type InstanceResolver ¶

type InstanceResolver interface {
	Instance(id string) (*agent.Agent, bool)
}

InstanceResolver is the minimum borrowed deployment view needed by Manager.

type Key ¶

type Key struct {
	AgentID   string
	ContextID string
}

Key identifies one conversational session. AgentID is part of the identity: two agents using the same ContextID do not share session state.

func (k Key) Validate() error

type Lease ¶

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

Lease keeps a shared Session retained by its Manager.

func (l *Lease) Close() error

func (l *Lease) Session() *Session

type Manager ¶

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

Manager shares Sessions by Key and reclaims unleased idle Sessions. It borrows its resolver, HostFactory, and event router and never closes them.

func NewManager(
	resolver InstanceResolver,
	hostFactory HostFactory,
	router *event.Router,
	options ...ManagerOption,
) (*Manager, error)

func (m *Manager) Close() error

func (m *Manager) GetOrCreate(ctx context.Context, key Key) (*Lease, error)

func (m *Manager) Open(ctx context.Context, key Key) (*Lease, error)

type ManagerOption ¶

type ManagerOption func(*managerOptions) error

ManagerOption configures a Manager.

func WithCatalogProvider(provider CatalogProvider) ManagerOption

func WithCheckpointStore(store agent.CheckpointStore) ManagerOption

func WithDeliveryConcurrency(limit int) ManagerOption

func WithIdleTimeout(timeout time.Duration) ManagerOption

func WithMaxSessions(limit int) ManagerOption

func WithResume(enable bool) ManagerOption

func WithSessionObserver(observer SessionObserver) ManagerOption

func WithSinkBufferSize(size int) ManagerOption

func WithSpeculativeBufferLimits(events, bytes int) ManagerOption

type PromptRequested ¶

type PromptRequested struct {
	RunID    string           `json:"run_id"`
	TurnID   string           `json:"turn_id"`
	PromptID string           `json:"prompt_id"`
	Prompt   agent.UserPrompt `json:"prompt"`
}

PromptRequested is published when a turn begins waiting for user input.

type Session ¶

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

Session owns conversational activity for one Key while borrowing all deployment and event-routing dependencies from the runtime.

func (s *Session) Key() Key

func (s *Session) Start(ctx context.Context, request agent.Request, sinks ...SinkSpec) (*Turn, error)

type SessionObserver ¶

type SessionObserver interface {
	// OnSessionStarted fires after Start successfully scheduled a new turn
	// and before Start returns to its caller. turn is the running turn.
	OnSessionStarted(session *Session, turn *Turn)

	// OnSessionClosing fires exactly once when the Session begins closing,
	// covering explicit close, idle timeout reclamation, and Manager
	// shutdown. Active turns may still be finalizing at this point.
	OnSessionClosing(session *Session)

	// OnSessionClosed fires exactly once after closing completed and any
	// active turn has been shut down and awaited. err is the close error,
	// or nil on a clean close.
	OnSessionClosed(session *Session, err error)

	// OnTurnFinished fires when a turn reaches its terminal state.
	// result and err are the same values Turn.Wait returns.
	// It is called before Turn.Wait unblocks, so observers MUST NOT call
	// Turn.Wait from this callback.
	OnTurnFinished(session *Session, turn *Turn, result *agent.Result, err error)
}

SessionObserver observes session-level lifecycle transitions without affecting their outcome. It is the public hook for metrics, tracing, and accounting that wants the session view (Key, RunID, terminal outcome) rather than per-run agent events.

Concurrency contract:

• Every method is called synchronously from the goroutine performing the transition, outside Session and Manager locks.
• Implementations must return promptly and MUST NOT call back into Session or Manager methods; long-running side effects must be dispatched asynchronously by the observer itself.
• Implementations must be safe for concurrent use: SessionStarted and TurnFinished can be called from a turn goroutine while Closing and Closed run on a close/reclaim goroutine.

Embed BaseSessionObserver to satisfy the interface with no-op defaults when only a subset of the lifecycle is interesting.

type SinkSpec ¶

type SinkSpec struct {
	ID        string
	Sink      agent.StreamSink
	QueueSize int
	// DeliveryTimeout bounds each Sink.OnDelta call. Zero uses the 30-second
	// runtime default; a sink that exceeds the deadline is detached.
	DeliveryTimeout time.Duration
	OnDetach        func(error)
	Visibility      Visibility
	Authority       Authority
	// AckMode controls when confirmed authoritative deliveries become part of
	// the committable prefix. See AckOnDelivery and AckExplicit.
	AckMode    AckMode
	MaxUnacked int
}

SinkSpec describes one independently buffered stream attachment.

func (s SinkSpec) Validate() error

type Turn ¶

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

Turn is one asynchronous execution owned by a Session.

func (t *Turn) Ack(sinkID string, cursor DeliveryCursor) error

func (t *Turn) After(_ context.Context, _ agent.Identity, _ *agent.Request, res *agent.Result) (agent.Decision, error)

func (t *Turn) CommitView(_ context.Context, _ agent.Identity, _ *agent.Request, res *agent.Result) (agent.CommitView, error)

func (t *Turn) Interrupt(interrupt agent.Interrupt) error

func (t *Turn) OnInterrupt(context.Context, agent.Identity, agent.Interrupt)

func (t *Turn) OnRunEnd(context.Context, agent.Identity, *agent.Result)

func (t *Turn) OnRunRevise(
	_ context.Context,
	_ agent.Identity,
	_ *agent.Result,
	nextAttempt int,
)

func (t *Turn) OnRunStart(context.Context, agent.Identity, *agent.Request)

func (t *Turn) Reply(ctx context.Context, promptID string, reply agent.UserReply) error

func (t *Turn) RunID() string

func (t *Turn) State() TurnState

func (t *Turn) Wait(ctx context.Context) (*agent.Result, error)

type TurnState ¶

type TurnState string

TurnState is the externally observable lifecycle state of a Turn.

const (
	TurnStarting     TurnState = "starting"
	TurnRunning      TurnState = "running"
	TurnInterrupting TurnState = "interrupting"
	TurnCompleted    TurnState = "completed"
	TurnInterrupted  TurnState = "interrupted"
	TurnCanceled     TurnState = "canceled"
	TurnFailed       TurnState = "failed"
	TurnAborted      TurnState = "aborted"
)

type Visibility ¶

type Visibility string

const (
	VisibilityRaw       Visibility = ""
	VisibilityConfirmed Visibility = "confirmed"
)