session

package
v0.1.1 Latest Latest
Warning

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

 Go to latest Published: Jul 7, 2026 License: MIT Imports: 14 Imported by: 0

Documentation

Overview

Package session is the agent's conversational front door: a streaming Session over the goal runtime that a chat UI drives turn by turn and renders live. It is a new host boundary alongside state and observe, the "streams/sessions/chat" surface.

A Session exposes the mission event spine as a live, replayable event stream. Submit a goal and the session opens, then every model turn, the model's text, each tool call and its result, and the terminal outcome are emitted as ordered session.Events. The events are durable on a spine stream and fanned out over the bus, so a subscriber that joins late replays the conversation from the start and then tails it live. This is the reusable, embeddable generalization of the terminal-first goal runner: the same run, surfaced as a stream a panel renders.

The session does not own the model loop. A caller wires the session's Reporter into a mission executor (mission.WithObserver) and hands the assembled runtime to Submit, so the agent's behaviour and the streaming surface stay decoupled.

Index

Constants

View Source 
const DefaultPoll = 50 * time.Millisecond

DefaultPoll is how often the session re-reads the goal's status to detect the terminal converged or stalled phase, the always-correct floor beneath the live turn events the mission reports directly.

View Source 
const DefaultStreamPoll = 250 * time.Millisecond

DefaultStreamPoll is a subscriber's fallback re-read interval: how often it drains the spine even if no bus wake arrived. The bus delivers events promptly; this is the always-correct floor beneath it, so a coalesced, dropped, or never-delivered wake (a degraded or absent bus) costs at most this much latency rather than stalling the tail.

Variables

This section is empty.

Functions

This section is empty.

Types

type ActionEntry 

type ActionEntry struct {
	// Call is the dispatch correlation id that pairs an admission with its outcome,
	// so a completion or rejection updates the entry its admission created rather
	// than adding a second row for the same action.
	Call int64
	// Action is the governed action's name (empty when the waist recorded none).
	Action string
	// Trust is the trust level the action ran under.
	Trust string
	// State is where the action stands: running, done, or blocked.
	State ActionState
	// Fault is the fault class of a blocked or failed action, empty otherwise.
	Fault string
}

ActionEntry is one governed action in the projection's ledger: what it was, the trust level it ran under, where it stands, and its fault class when it failed or was refused. The governance panel reads the ledger to show the run's recent boundary decisions, which the scalar counts summarize but cannot name.

type ActionState 

type ActionState string

ActionState is where one governed action stands: running once admitted, done once it completes, blocked once it is refused. The states are terminal except running, which resolves to done or blocked when the action's outcome arrives. 

const (
	// ActionRunning is an admitted action that has not yet reported an outcome.
	ActionRunning ActionState = "running"
	// ActionDone is an admitted action that finished (cleanly, or with a fault it
	// carries).
	ActionDone ActionState = "done"
	// ActionBlocked is an action the dispatch waist refused.
	ActionBlocked ActionState = "blocked"
)

type EgressEntry 

type EgressEntry struct {
	// Host is the destination the decision concerned (a resolved literal address).
	Host string
	// Allowed reports whether netguard permitted the dial.
	Allowed bool
	// Reason is netguard's short why for the verdict.
	Reason string
}

EgressEntry is one outbound-network decision in the projection's egress ledger: the destination host netguard was asked to reach, whether it allowed the dial, and its short reason. The governance panel reads the ledger to name the recent destinations the scalar egress counts only tally.

type Event 

type Event struct {
	Seq   int64           `json:"seq"`
	Time  time.Time       `json:"time"`
	Kind  Kind            `json:"kind"`
	Actor spine.ActorType `json:"actor"`

	// Goal is the id of the goal the event belongs to: the run's root goal (its id is
	// the run id) for a conversation event, or a delegated child's own id under
	// fan-out. It attributes each event to its originating goal so a fan-out's
	// concurrent children stay distinguishable on the one shared stream. Empty on
	// session-level events and on a single conversation's root, both of which are the
	// run itself.
	Goal string `json:"goal,omitempty"`
	// Child is the id of a spawned child goal (child.spawned only): the other end of
	// the parent-to-child edge whose parent is Goal.
	Child string `json:"child,omitempty"`
	// Turn is the 1-based model turn the event belongs to (0 for session-level
	// events: started, converged, stalled).
	Turn int `json:"turn,omitempty"`
	// Text carries the objective (started), the model's message (assistant), or the
	// final answer (converged).
	Text string `json:"text,omitempty"`
	// Tool is the tool's name (tool.call, tool.result).
	Tool string `json:"tool,omitempty"`
	// ToolUseID correlates a tool.call with the tool.result that answers it.
	ToolUseID string `json:"toolUseID,omitempty"`
	// Input is the tool's JSON arguments (tool.call).
	Input json.RawMessage `json:"input,omitempty"`
	// Result is the tool's output text (tool.result).
	Result string `json:"result,omitempty"`
	// IsError reports a failed tool call (tool.result).
	IsError bool `json:"isError,omitempty"`
	// StopReason is why the turn ended (turn.completed).
	StopReason string `json:"stopReason,omitempty"`
	// Usage is the token cost of a turn (turn.completed), nil for other events. It
	// is carried on the stream, not just metered internally, so a live UI and a
	// replay both show per-turn spend and cache effectiveness.
	Usage *Usage `json:"usage,omitempty"`
	// Err carries the stall reason (stalled).
	Err string `json:"error,omitempty"`

	// Action is the governed action's name (action.admitted/completed/rejected): a
	// tool, the model call, or the spawn that delegates a sub-goal. It is broader than
	// Tool, which names only a model-requested tool.
	Action string `json:"action,omitempty"`
	// Call correlates an action.admitted with the action.completed or action.rejected
	// that answers it, so a renderer pairs an admission with its outcome and a panel
	// can show only the actions still in flight.
	Call int64 `json:"call,omitempty"`
	// Trust is the action's trust level (action.* events): how far the work is
	// trusted, which is the containment a gate requires to admit it.
	Trust string `json:"trust,omitempty"`
	// Fault is the fault class on a refused or failed action: the denial reason on
	// action.rejected (capability_denied, budget_exceeded, needs_approval, ...) or the
	// failure class on an action.completed that ran but errored (transient, ...). Empty
	// on a clean completion.
	Fault string `json:"fault,omitempty"`

	// Host is the destination an egress decision concerned (net.egress): the literal
	// address netguard resolved before deciding whether to connect.
	Host string `json:"host,omitempty"`
	// Allowed reports whether an egress decision permitted the dial (net.egress). A
	// false Allowed with a set Host is a blocked destination.
	Allowed bool `json:"allowed,omitempty"`
	// Reason is the short why for an egress decision (net.egress): why netguard allowed
	// the dial (public, allowlisted) or refused it (private or reserved address, ...).
	Reason string `json:"reason,omitempty"`
}

Event is one record on a session's event stream: an ordered, replayable view of the conversation a UI renders live. Seq is monotonic within a session and Time is the moment it was recorded; the remaining fields are populated by Kind and are otherwise zero. It is the public "streams/sessions/chat" surface, a typed projection of the underlying event spine.

func History 

func History(ctx context.Context, log spine.Log, id string) ([]Event, error)

History returns every recorded event for the run identified by id, read in order straight from the durable log. Unlike Subscribe it is finite: it does not wait for new events, so it is the primitive for inspecting or replaying a run that has already finished. An unknown id yields an empty slice, not an error.

type FanoutChild 

type FanoutChild struct {
	// ID is the child goal's id, the value its own events carry in Goal.
	ID string
	// Parent is the id of the goal that spawned this child.
	Parent string
	// Objective is the sub-objective the parent delegated.
	Objective string
	// State is where the child stands: running, done, or failed.
	State FanoutState
	// Turns counts the model turns the child has completed so far.
	Turns int
	// Result is the child's folded answer once it is done or failed, empty while it
	// runs.
	Result string
	// Trust is the trust level of this child's most recent governed action, its own
	// containment posture (the per-child analogue of the run's Containment). Empty until
	// the child has an action admitted or rejected.
	Trust string
	// Blocked counts this child's governed actions the dispatch waist refused. A non-zero
	// Blocked is the signal that this child, not a sibling, hit a governance boundary.
	Blocked int
	// Seal is the child's record lifecycle within the run's sealed record: recording
	// while its events are still being appended, sealed once it has folded back and the
	// run's stream is signed (its events are then under the signed Merkle root, provable
	// with a per-event proof), verified once the run's sealed record passes verification.
	// A child still running when a seal lands stays recording, since its events are not
	// yet final. It is the per-child analogue of the run-level Record above.
	Seal RecordState
}

FanoutChild is one delegated child in a run's fan-out tree: which goal spawned it, the sub-objective it was given, where it stands, how many turns it has taken, its own governance posture (trust level and blocked count, folded from the actions it ran), its seal state within the run's sealed record, and its folded answer once it finishes. The tree is flat with a Parent id on each node rather than nested, so a renderer builds the hierarchy and a deeper delegation (a child that itself spawns) needs no change here. Children are kept in spawn order.

type FanoutState 

type FanoutState string

FanoutState is where a delegated child stands in the fan-out tree: running from the moment it is spawned until it folds back into its parent, then done or failed by the outcome it folded. The states are terminal except running. 

const (
	// FanoutRunning is a spawned child that has not yet folded back into its parent.
	FanoutRunning FanoutState = "running"
	// FanoutDone is a child that folded back with a clean result.
	FanoutDone FanoutState = "done"
	// FanoutFailed is a child that folded back as an error.
	FanoutFailed FanoutState = "failed"
)

type Kind 

type Kind string

Kind classifies a session event. The set spans a conversation's whole arc: the session opening, each model turn and the text and tool calls within it, and the terminal outcome. A renderer switches on Kind to draw the live transcript. 

const (
	// KindSessionStarted is the first event: the session was opened with an
	// objective.
	KindSessionStarted Kind = "session.started"
	// KindTurnStarted marks the start of one model turn.
	KindTurnStarted Kind = "turn.started"
	// KindAssistant carries the model's natural-language text for a turn.
	KindAssistant Kind = "assistant.message"
	// KindToolCall is the model requesting a tool, with its arguments.
	KindToolCall Kind = "tool.call"
	// KindToolResult is the outcome of a tool call.
	KindToolResult Kind = "tool.result"
	// KindTurnCompleted marks the end of a turn, with why the model stopped.
	KindTurnCompleted Kind = "turn.completed"
	// KindChildSpawned is a fan-out delegation: the goal named by an event's Goal
	// spawned the child named by Child, whose sub-objective is in Text. It records the
	// parent-to-child edge on the stream, so a live fan-out tree is built from the
	// record rather than a separate store.
	KindChildSpawned Kind = "child.spawned"
	// KindChildCompleted marks a spawned child folding back into its parent: the child
	// named by Child finished, its answer in Result (IsError set when it failed). It
	// closes the edge KindChildSpawned opened so the tree flips the child to done.
	KindChildCompleted Kind = "child.completed"
	// KindConverged is the terminal success event: the goal's stop condition was
	// met, with the model's final answer.
	KindConverged Kind = "session.converged"
	// KindStalled is the terminal failure event: the goal ran out of budget or a
	// step failed terminally, with the reason.
	KindStalled Kind = "session.stalled"

	// KindActionAdmitted is a governed action admitted by the dispatch waist, the
	// waist's start event projected onto the conversation stream. The waist records
	// every governed action (a tool, the model call, a delegation) on the run's own
	// stream, so its admission decisions interleave with the turns they belong to.
	// This projection never changes the stored event type (dispatch.start), which
	// stays the verifiable governance record a sealer signs and a verifier checks.
	KindActionAdmitted Kind = "action.admitted"
	// KindActionCompleted is a governed action that finished, carrying a fault class
	// when it ran but errored. Projected from the waist's end event (dispatch.end).
	KindActionCompleted Kind = "action.completed"
	// KindActionRejected is a governed action the waist refused, carrying the denial's
	// fault class. Projected from the waist's rejection event (dispatch.rejected).
	KindActionRejected Kind = "action.rejected"

	// KindRecordSealed marks the run's stream sealed into a signed record, projected
	// from the spine.record event the sealer stores on the stream.
	KindRecordSealed Kind = "record.sealed"
	// KindRecordVerified marks a sealed record checked against its tiers. It is emitted
	// when verification runs, so a replay reproduces the record badge's transitions.
	KindRecordVerified Kind = "record.verified"

	// KindEgressDecision is one outbound-network decision netguard reached at dial time,
	// projected onto the run's stream: the destination host, whether the dial was allowed
	// or blocked, and a short reason. It records the run's egress posture (which
	// destinations it reached and which the policy refused) alongside its governed
	// actions, so the same panel that shows admission decisions shows egress ones.
	KindEgressDecision Kind = "net.egress"
)

type Option 

type Option func(*Session)

Option configures a Session.

func WithClock 

func WithClock(clk clock.Timing) Option

WithClock sets the clock the lifecycle watch and the subscriber poll floor time their re-reads on, so a test drives them with a Manual clock instead of sleeping. Production leaves the default System clock.

func WithID 

func WithID(id string) Option

WithID sets the session's identity, which names its spine stream and bus subject. A host embeds its own conversation id here so the stream lines up with its records. The id must be a valid bus subject token (no whitespace, dots, or wildcards). The default is a freshly generated UUIDv7, globally unique and stable across restarts so a run's event stream stays addressable.

func WithPollInterval 

func WithPollInterval(d time.Duration) Option

WithPollInterval overrides how often the lifecycle watch re-reads goal status.

func WithStreamPoll 

func WithStreamPoll(d time.Duration) Option

WithStreamPoll overrides a subscriber's fallback re-read interval (see DefaultStreamPoll). A non-positive value disables the floor, leaving the bus the sole liveness signal, which a host should do only when it trusts the bus to deliver.

type Projection 

type Projection struct {
	// Objective is the run's opening goal; Result is its final answer once converged.
	Objective string
	Result    string

	// Turns counts the model turns completed. Usage is the run's cumulative token cost
	// across those turns, summed from each turn.completed.
	Turns int
	Usage Usage

	// Record is the run's record lifecycle for the badge.
	Record RecordState

	// Containment is the trust level of the most recent governed action, the run's
	// current containment posture. Empty until the first action is admitted.
	Containment string

	// Admitted, Completed, and Rejected count governed actions by outcome. Admitted
	// minus Completed minus the admitted-then-rejected races is the in-flight count; a
	// non-zero Rejected is the signal a run hit a governance boundary.
	Admitted  int
	Completed int
	Rejected  int

	// Actions is a bounded ledger of the run's most recent governed actions, newest
	// last, each carrying its name, trust level, state, and fault class. It names the
	// decisions the scalar counts above only tally, for a governance panel to show.
	// It is capped at maxLedger entries; older actions fall off the front.
	Actions []ActionEntry

	// EgressAllowed and EgressBlocked count outbound dials netguard permitted and
	// refused over the run. A non-zero EgressBlocked is the signal the run tried to reach
	// a destination the policy denied. Both stay zero for a run that made no
	// netguard-gated egress (a local, loopback model), so the egress row stays silent.
	EgressAllowed int
	EgressBlocked int

	// Egress is a bounded ledger of the run's most recent egress decisions, newest last,
	// each naming the destination host and whether it was allowed. It names the
	// destinations the counts above only tally, for a governance panel to show. It is
	// capped at maxEgress entries; older decisions fall off the front.
	Egress []EgressEntry

	// Fanout is the run's delegated children in spawn order, each with its parent, its
	// sub-objective, where it stands, and its turn count. It is empty for a single
	// conversation and grows as a fan-out spawns children, so a live tree view reads
	// the run's delegation shape straight from the projection.
	Fanout []FanoutChild

	// Terminal reports whether the run reached a terminal event. Err is the stall
	// reason when it stalled, empty on a clean convergence.
	Terminal bool
	Err      string
}

Projection is the current status of a run, folded from its event stream: the facts an always-on status line and a governance panel read. It is a pure function of the events seen so far, so a live client, a replay, and a reattached client that folds from Seq 0 all compute the same status. It carries no presentation, only the run's state; a renderer decides how to show it (and combines it with UI-local facts a stream does not carry, such as the model's context-window size, to derive a context-fill percentage from Usage).

func NewProjection 

func NewProjection() Projection

NewProjection returns the status of a run with no events yet: recording, and otherwise zero. Fold events into it with Reduce, or use Project over a slice.

func Project 

func Project(evs []Event) Projection

Project folds a whole event slice into the run's status, the replay counterpart of Reduce: History or a resume picker reads a run's status by projecting its recorded events. It seeds from NewProjection, so an empty slice is a recording run with no activity yet.

func Reduce 

func Reduce(p Projection, ev Event) Projection

Reduce folds one event into a running Projection and returns the updated status. It is the single place a stream becomes status, shared by the live status line (folding each event as it arrives) and a replay (folding History). Unrecognized events leave the status unchanged, so the projection is stable as the event vocabulary grows.

type RecordState 

type RecordState string

RecordState is a run's record lifecycle, the value the status badge shows. A run records until it is sealed into a signed record, and a sealed record can be verified against its tiers. The states are ordered: a run only ever moves recording -> sealed -> verified. 

const (
	// RecordRecording is the live state: events are still being appended to the run.
	RecordRecording RecordState = "recording"
	// RecordSealed is set once the run's stream is sealed into a signed record.
	RecordSealed RecordState = "sealed"
	// RecordVerified is set once a sealed record has passed verification.
	RecordVerified RecordState = "verified"
)

type Session 

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

Session is one conversational run: its event stream, and the lifecycle watch that turns the goal reaching a terminal phase into a terminal event and a result a caller can await. Construct it with New, feed its Reporter into a mission executor, then drive it with Submit.

func New 

func New(log spine.Log, b bus.Bus, opts ...Option) *Session

New builds a Session whose events are recorded on log and fanned out over b. Its identity defaults to a fresh UUIDv7 (override with WithID); that id names the run's event stream and, via Submit, its goal resource, so one value addresses the whole run for replay, audit, and sync.

func (*Session) ID 

func (s *Session) ID() string

ID returns the session's identity (its spine stream and bus subject suffix).

func (*Session) Reporter 

func (s *Session) Reporter() mission.Reporter

Reporter returns the mission.Reporter that records the conversation onto this session's stream. Wire it into the executor with mission.WithObserver before handing the runtime to Submit.

func (*Session) Resume 

func (s *Session) Resume(ctx context.Context, rt *runtime.Runtime, key resource.Key)

Resume attaches the session to an already-submitted goal identified by key and watches it to its terminal phase, without re-opening the session or re-submitting the goal. It is how a run is continued: the caller arranges for the runtime to re-drive the existing goal (see runtime.Resume), and the session streams the rest of the conversation onto the same stream as the original run. It emits no session.started, since the run was already opened.

func (*Session) Submit 

func (s *Session) Submit(ctx context.Context, rt *runtime.Runtime, spec goal.Spec) (resource.Key, error)

Submit opens the session, submits spec as a goal to rt, and starts watching it. The goal is named after the session id, so the run's event stream and its goal resource share one identity: the run is addressable by a single id for replay, audit, and (later) ownership and sync. It emits the session.started event, returns the goal's key, and from then on the session streams the conversation (via the Reporter the caller wired in) and, when the goal converges or stalls, emits the terminal event and releases Wait. The caller owns rt's lifecycle: rt.Start must be running for the goal to make progress.

func (*Session) Subscribe 

func (s *Session) Subscribe(ctx context.Context, afterSeq int64) (<-chan Event, error)

Subscribe returns a live, ordered event channel beginning after afterSeq (0 replays from the start). See Stream.Subscribe for the delivery guarantees.

func (*Session) Wait 

func (s *Session) Wait(ctx context.Context) (string, error)

Wait blocks until the session reaches a terminal state and returns the model's final answer on convergence, or a non-nil error on stall or cancellation.

type Stream 

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

Stream is an ordered, replayable, fan-out event channel for one session. Each appended event is recorded on a durable spine stream (which assigns its monotonic Seq) and a wake is published on the bus. A subscriber receives the full backlog followed by every later event, in Seq order, with no loss and no duplication.

The split of roles is deliberate and mirrors the foundation: the spine is the durable, ordered source of truth, and the bus carries only a liveness signal. Subscribers never read content off the bus, so a coalesced or dropped wake costs nothing: the next wake (or the poll floor) re-reads everything past the subscriber's cursor from the spine. That makes the live tail correct even though the bus is at-most-once, and keeps it making progress even if the bus delivers nothing at all.

func (*Stream) Subscribe 

func (s *Stream) Subscribe(ctx context.Context, afterSeq int64) (<-chan Event, error)

Subscribe returns a channel of every event after afterSeq, beginning with the backlog already on the spine and continuing with events as they arrive. Pass 0 to replay the session from the start. The channel is closed when ctx is cancelled or the underlying bus subscription ends; the caller owns draining it.

Ordering and exactly-once delivery come from a single advancing cursor: each wake drains the spine from the cursor forward and advances it past every event emitted, so no event is sent twice and a burst of wakes collapses into one read. Backpressure is per-subscriber: a slow reader blocks only its own drain, never the bus or other subscribers.

type Usage 

type Usage struct {
	InputTokens      int `json:"inputTokens,omitempty"`
	OutputTokens     int `json:"outputTokens,omitempty"`
	CacheReadTokens  int `json:"cacheReadTokens,omitempty"`
	CacheWriteTokens int `json:"cacheWriteTokens,omitempty"`
}

Usage is the token cost of one turn, projected onto the conversation stream. It mirrors the model port's accounting in the chat surface's own terms so a session consumer does not depend on the model package. InputTokens is the total input processed including any served from cache; CacheReadTokens is the cached subset (the discounted win), so cache-hit-rate is CacheReadTokens over InputTokens.

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.