Help us shape the future of Go! Take the 2026 Go Developer Survey

session

package
v0.9.0 Latest Latest
Warning

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

 Go to latest Published: Oct 20, 2026 License: MIT Imports: 20 Imported by: 0

Documentation

Index

Constants

View Source 
const HeaderDeliveryCursor = "session_delivery_cursor"

HeaderDeliveryCursor carries a confirmed delivery cursor in an envelope.

Variables

View Source 
var (
	ErrSessionClosed   = errdefs.NotAvailablef("runtime session: session is closed")
	ErrPromptUnknown   = errdefs.NotFoundf("runtime session: prompt is unknown")
	ErrPromptDuplicate = errdefs.Conflictf("runtime session: prompt was already replied")
	ErrPromptClosed    = errdefs.NotAvailablef("runtime session: prompt is closed")
	ErrSinkQueueFull   = errdefs.BudgetExceededf("runtime session: sink queue is full")
	// ErrSteerClosed is returned by Turn.Steer once the turn reached a
	// terminal state: nothing will drain the queue after that point.
	ErrSteerClosed = errdefs.NotAvailablef("runtime session: turn is not accepting steer")
	// ErrSteerQueueFull is returned by Turn.Steer when the turn-owned
	// steer queue already holds maxSteerQueue messages. The caller keeps
	// its message and decides whether to retry, queue for the next turn,
	// or surface the rejection.
	ErrSteerQueueFull = errdefs.BudgetExceededf("runtime session: steer queue is full")
	// ErrSteerTooLarge is returned by Turn.Steer for a single message
	// whose JSON form exceeds maxSteerMessageBytes.
	ErrSteerTooLarge = errdefs.BudgetExceededf("runtime session: steer message exceeds the size limit")
)
View Source 
var (
	ErrManagerClosed   = errdefs.NotAvailablef("runtime session: manager is closed")
	ErrManagerDraining = errdefs.NotAvailablef("runtime session: manager is draining")
)

Functions

func PatternPromptRequested added in v0.1.5 

func PatternPromptRequested() event.Pattern

PatternPromptRequested matches every run's prompt-request event.

func PatternPromptResolved added in v0.1.5 

func PatternPromptResolved() event.Pattern

PatternPromptResolved matches every run's prompt-resolution event.

func SubjectPromptRequested 

func SubjectPromptRequested(runID string) event.Subject

SubjectPromptRequested returns the run-scoped prompt request subject.

func SubjectPromptResolved added in v0.1.5 

func SubjectPromptResolved(runID string) event.Subject

SubjectPromptResolved returns the run-scoped prompt resolution subject, sharing the run namespace with SubjectPromptRequested so a "agent.run.<id>.>" subscription observes both.

func WithStreamPolicy added in v0.1.29 

func WithStreamPolicy(ctx context.Context, policy StreamPolicy) context.Context

WithStreamPolicy returns a context that carries policy to nested sessions started within the turn's execution.

Types

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 

func (BaseSessionObserver) OnSessionClosed(*Session, error)

OnSessionClosed is a no-op.

func (BaseSessionObserver) OnSessionClosing 

func (BaseSessionObserver) OnSessionClosing(*Session)

OnSessionClosing is a no-op.

func (BaseSessionObserver) OnSessionStarted 

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

OnSessionStarted is a no-op.

func (BaseSessionObserver) OnTurnFinished 

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

OnTurnFinished is a no-op.

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 (CatalogProviderFunc) NewCatalog 

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

NewCatalog implements CatalogProvider.

type DeliveryCursor 

type DeliveryCursor uint64

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

func DeliveryCursorFromEnvelope 

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

DeliveryCursorFromEnvelope reads the confirmed delivery cursor.

type Deps added in v0.1.14 

type Deps struct {
	Resolver        InstanceResolver
	HostFactory     HostFactory
	CatalogProvider CatalogProvider
	// Checkpoints and Resume are the epoch's session durability
	// settings: turns resolve them from their epoch, so a generation
	// reload can change the store or the resume flag without touching
	// in-flight turns.
	Checkpoints agent.CheckpointStore
	Resume      bool
	Epoch       uint64
}

Deps is one Start's complete consistency snapshot: the resolver, host factory, and catalog provider a turn may use, tagged with the epoch they belong to. A turn bound to an epoch never mixes dependencies from another epoch, so a generation swap can never tear a single execution.

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 (HostFactoryFunc) NewHost 

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

NewHost implements HostFactory.

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 (HostRequest) Validate 

func (r HostRequest) Validate() error

Validate checks the required host-construction contract.

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 (Key) Validate 

func (k Key) Validate() error

Validate rejects empty, whitespace-only, and padded identifiers.

type Lease 

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

Lease keeps a shared Session retained by its Manager.

func (*Lease) Close 

func (l *Lease) Close() error

Close releases this Lease. It is safe to call repeatedly or concurrently.

func (*Lease) Session 

func (l *Lease) Session() *Session

Session returns the Session retained by this Lease.

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 

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

NewManager constructs a Session manager over borrowed runtime dependencies.

func (*Manager) Close 

func (m *Manager) Close() error

Close stops reclamation, refuses new leases, and closes every Session. Borrowed deployment, router, and host dependencies remain owned by Runtime.

func (*Manager) DeleteSession added in v0.1.27 

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

DeleteSession removes one session's durable state and closes its live Session, identified by key (agent id + context id). It is the by-key counterpart of RemoveAgent for the desktop lifecycle (delete/archive a conversation): after it returns, the checkpoint store no longer carries the session's committed board (history), parked run checkpoint, or resumable request for that key, and a later Open starts a fresh session with empty history.

Semantics mirror RemoveAgent:

  • New opens for the key are refused until deletion finishes.
  • Any live Session is drained (bounded by ctx), then closed. On ctx expiry the delete marker is rolled back, the session is left intact, and no partial removal state is produced — callers may retry.
  • Repeated calls are idempotent: a key with no live Session and no persisted state deletes successfully.

Deletion of the checkpoint store entries runs only after the live Session confirms its active turn has stopped (afterTurn has run), so a turn that started in the drain window cannot resurrect state underneath the delete. Session.close's shutdown wait is budgeted: if it times out — or the session is already closing with an active turn — DeleteSession returns a retryable error with the durable state left in place, and the caller retries once the turn has stopped. Store failures are returned (joined with close errors) so callers can retry.

func (*Manager) Drain added in v0.2.5 

func (m *Manager) Drain(ctx context.Context) error

Drain quiesces the manager for offline replacement: it refuses new Opens, new GetOrCreate calls, and new Starts on already-open leases, then waits (bounded by ctx) for every live Session to finish its active turns, prompts, and sinks naturally. Drain never interrupts running turns. After Drain returns, the manager stays draining and the caller may retry a timed-out Drain or proceed to Close.

func (*Manager) GetOrCreate 

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

GetOrCreate lazily creates a Session and returns an independent Lease.

func (*Manager) Idle added in v0.2.5 

func (m *Manager) Idle() bool

Idle reports whether no live Session currently has an active turn, prompt, or sink. A manager with no live Sessions is idle.

func (*Manager) Open 

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

Open returns an independent Lease over the Session identified by key.

func (*Manager) RemoveAgent added in v0.1.11 

func (m *Manager) RemoveAgent(ctx context.Context, name string) error

RemoveAgent blocks new session activity for the named agent and drains its live sessions: it waits (bounded by ctx) for every active turn to finish naturally, then closes the now-idle sessions. On ctx expiry the tombstone stays in place (new opens keep failing), the sessions are left intact, and no partial removal state is produced — callers may retry. Repeated calls are idempotent.

func (*Manager) ReopenAgent added in v0.1.11 

func (m *Manager) ReopenAgent(name string)

ReopenAgent clears the tombstone left by RemoveAgent, re-admitting session activity for the name. It is used when re-registering an agent under the same ID.

func (*Manager) SwapDeps added in v0.1.14 

func (m *Manager) SwapDeps(
	deps Deps,
	onRetired func(epoch uint64, d Deps),
) error

SwapDeps atomically retires the current epoch and installs deps as the new current epoch. onRetired, when non-nil, is invoked exactly once when the retired epoch's references reach zero (immediately, if none are outstanding). SwapDeps fails after the manager is closed.

func (*Manager) UpdateTunables added in v0.1.14 

func (m *Manager) UpdateTunables(t Tunables) error

UpdateTunables validates and atomically applies new manager tunables. Validation mirrors the With* ManagerOptions (positive bounds).

func (*Manager) WaitIdle added in v0.2.5 

func (m *Manager) WaitIdle(ctx context.Context) error

WaitIdle blocks until no live Session has an active turn, prompt, or sink, bounded by ctx. It is a pure wait and never changes manager state; a manager with no live Sessions returns immediately. The error is nil on success or the ctx error when the deadline is exceeded or the context is cancelled.

type ManagerBinder added in v0.1.23 

type ManagerBinder interface {
	BindSessionManager(manager *Manager) error
}

ManagerBinder receives the Runtime's session manager when a deployment generation is built or rebuilt. Consumers such as the delegation service implement it to upgrade subagent execution from the legacy bare-run path to session lifecycle. Binding is set-once: a second bind must return an error so the runtime can roll back a reload.

type ManagerOption 

type ManagerOption func(*managerOptions) error

ManagerOption configures a Manager.

func WithCatalogProvider 

func WithCatalogProvider(provider CatalogProvider) ManagerOption

WithCatalogProvider wires a per-session tool catalog provider. When set, every Session created by this Manager builds one catalog on its first Start, attaches it to each turn's execution context, and closes it when the Session closes.

func WithCheckpointStore 

func WithCheckpointStore(store agent.CheckpointStore) ManagerOption

WithCheckpointStore wires the store used for end-to-end resume. It is required when WithResume is enabled.

func WithDeliveryConcurrency 

func WithDeliveryConcurrency(limit int) ManagerOption

WithDeliveryConcurrency bounds how many sink callbacks may be in flight concurrently for one sink.

func WithIdleTimeout 

func WithIdleTimeout(timeout time.Duration) ManagerOption

WithIdleTimeout sets how long an unleased, idle Session is retained.

func WithMaxSessions 

func WithMaxSessions(limit int) ManagerOption

WithMaxSessions bounds the number of distinct live sessions the manager retains. Values <= 0 leave the manager default (1024).

func WithResume 

func WithResume(enable bool) ManagerOption

WithResume enables session-level durability across turns:

  • Every turn gets a fresh run id; conversation history carries over from the session's last committed board, persisted under a session-scoped key in the checkpoint store.
  • Turns that end without committing park their run id; Resume replays that specific interrupted execution from its checkpoint.
  • Committed turns delete their run checkpoint and advance the session board.

It requires a checkpoint store (WithCheckpointStore) whose implementation also satisfies agent.CheckpointDeleter.

func WithSessionObserver 

func WithSessionObserver(observer SessionObserver) ManagerOption

WithSessionObserver attaches a session-level lifecycle observer to every Session created by this Manager. See SessionObserver for the callback contract.

func WithSinkBufferSize 

func WithSinkBufferSize(size int) ManagerOption

WithSinkBufferSize sets the queue size used when SinkSpec.QueueSize is zero.

func WithSpeculativeBufferLimits 

func WithSpeculativeBufferLimits(events, bytes int) ManagerOption

WithSpeculativeBufferLimits bounds aggregate pending confirmed branch data per turn.

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 PromptResolved added in v0.1.5 

type PromptResolved struct {
	RunID    string       `json:"run_id"`
	TurnID   string       `json:"turn_id"`
	PromptID string       `json:"prompt_id"`
	Status   PromptStatus `json:"status"`
}

PromptResolved is published when a prompt stops waiting for user input, so consumers (UI views) can close or invalidate the pending interaction immediately instead of waiting for the whole turn to end. It is a best-effort notification: publish failures do not change the Reply / AskUser result.

type PromptStatus added in v0.1.5 

type PromptStatus string

PromptStatus is the terminal lifecycle state of one prompt. 

const (
	// PromptReplied is published when a Reply satisfied the prompt.
	PromptReplied PromptStatus = "replied"
	// PromptExpired is published when the AskUser context ended before
	// a reply arrived (timeout or host cancellation).
	PromptExpired PromptStatus = "expired"
	// PromptInterrupted is published when a turn interrupt closed the
	// prompt.
	PromptInterrupted PromptStatus = "interrupted"
	// PromptClosed is published when the turn ended while the prompt
	// was still pending, or when the prompt could not be published.
	PromptClosed PromptStatus = "closed"
)

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 (*Session) Key 

func (s *Session) Key() Key

Key returns this Session's immutable identity.

func (*Session) ParkedRequest added in v0.1.25 

func (s *Session) ParkedRequest(ctx context.Context) (agent.Request, bool, error)

ParkedRequest returns the original request of the session's parked interrupted run, when one exists and its request was stored. ok is false whenever there is nothing to resume: a fresh or committed session, an ephemeral session, resume not configured, or a parked marker without a stored request. The returned request is a copy.

func (*Session) Resumable added in v0.1.6 

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

Resumable reports whether a previous turn ended without committing and can be replayed via Resume. It returns the parked run id.

func (*Session) Resume added in v0.1.6 

func (s *Session) Resume(ctx context.Context, sinks ...SinkSpec) (*Turn, error)

Resume re-executes the session's parked interrupted run. The parked run is the most recent turn that ended without committing; its checkpoint is loaded and replayed under the original run id and request, so the engine picks up where it stopped instead of starting fresh. A new user message MUST go through Start — Resume does not carry a request.

func (*Session) ResumeWithOptions added in v0.1.23 

func (s *Session) ResumeWithOptions(ctx context.Context, opts ...StartOption) (*Turn, error)

ResumeWithOptions is Resume with option-based configuration. WithEphemeral is rejected: the session ephemeral property is fixed by the first Start.

func (*Session) Start 

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

Start installs and asynchronously executes a new turn. A previous turn is interrupted and fully finalized before the replacement is constructed.

func (*Session) StartWithOptions added in v0.1.23 

func (s *Session) StartWithOptions(ctx context.Context, request agent.Request, opts ...StartOption) (*Turn, error)

StartWithOptions is Start with option-based configuration: stream sinks, an AskUser override, and the ephemeral session property. The plain Start signature is unchanged for existing callers.

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 receives stream deltas for this attachment. A Sink may be
	// shared across sessions: session inheritance (see StreamPolicy)
	// attaches the caller's sink to delegated subagent turns as well,
	// so implementations must be safe for concurrent OnDelta calls.
	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 (SinkSpec) Validate 

func (s SinkSpec) Validate() error

Validate checks a sink before it is attached to a turn.

type StartOption added in v0.1.23 

type StartOption func(*startConfig) error

StartOption configures one StartWithOptions / ResumeWithOptions call. Existing callers of Start / Resume never see these options.

func WithAskUserOverride added in v0.1.23 

func WithAskUserOverride(askUser AskUserFunc) StartOption

WithAskUserOverride replaces the turn's default PromptRequested-based asker. The override is invoked directly by the turn host without publishing prompt lifecycle events or blocking for a reply.

func WithEphemeral added in v0.1.23 

func WithEphemeral() StartOption

WithEphemeral marks the session ephemeral (fixed by the first Start): no session-state or run-checkpoint writes, no history seeding, canceled turns never park, and Resume reports not found. Mixing ephemeral and persistent starts on the same session is rejected.

func WithSinks added in v0.1.23 

func WithSinks(sinks ...SinkSpec) StartOption

WithSinks attaches stream sinks, mirroring the variadic sinks of Session.Start for the option-based entry point.

type StreamPolicy added in v0.1.29 

type StreamPolicy struct {
	// Sinks are attached to every nested session started inside the
	// turn when Inheritable is true. Delegation attaches them as
	// observers.
	Sinks []SinkSpec
	// Inheritable reports whether nested sessions should inherit Sinks.
	// Session starts stamp the policy with Inheritable=true; callers may
	// stamp false to stop inheritance at a boundary.
	Inheritable bool
}

StreamPolicy describes how a turn's stream sinks are inherited by sessions started within the turn's execution (for example delegated subagent runs). Session starts stamp their attached sinks so nested runs inherit the caller's stream without any explicit wiring.

Inheritance shares the same Sink instance across sessions: a delegated turn attaches the caller's sinks too, so a sink must be safe for concurrent OnDelta calls. The delegation boundary downgrades inherited specs to observers (no authority, no explicit acks), because the subagent turn is never handed to the sink.

func StreamPolicyFromContext added in v0.1.29 

func StreamPolicyFromContext(ctx context.Context) (StreamPolicy, bool)

StreamPolicyFromContext returns the nearest stream policy carried by ctx, if any.

type Tunables added in v0.1.14 

type Tunables struct {
	IdleTimeout         time.Duration
	SinkBuffer          int
	SpeculativeEvents   int
	SpeculativeBytes    int
	DeliveryConcurrency int
	MaxSessions         int
}

Tunables are the manager-level session settings a generation reload may change. Semantics:

  • IdleTimeout and MaxSessions are read live by the manager, so a change applies immediately (future timers and open checks).
  • SinkBuffer, SpeculativeEvents, SpeculativeBytes, and DeliveryConcurrency are captured by each Session at creation, so a change applies to sessions created afterwards.

type Turn 

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

Turn is one asynchronous execution owned by a Session.

func (*Turn) Ack 

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

Ack cumulatively acknowledges confirmed deliveries from the authoritative explicit sink. It may be called from Sink.OnDelta for the cursor currently being offered to that callback.

func (*Turn) After 

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

After implements the turn-scoped interruption referee.

func (*Turn) Cancel added in v0.1.6 

func (t *Turn) Cancel()

Cancel immediately stops the turn by cancelling its execution context. Unlike Turn.Interrupt, which the engine only observes at its next safe checkpoint, Cancel takes effect as soon as the engine notices the context cancellation — typically while it is blocked in I/O or an inference call.

The run settles as agent.StatusCanceled, a non-committed outcome: with resume enabled the session parks the run and its checkpoint stays resumable via Session.Resume. Cancel is idempotent and safe to call after the turn has finished.

func (*Turn) CommitView 

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

CommitView implements the turn-scoped Committer projection.

func (*Turn) DrainSteer added in v0.4.6 

func (t *Turn) DrainSteer() []message.Message

DrainSteer returns the messages queued for this turn and empties the queue. It backs the turn's agent.SteerSource: non-blocking, safe for concurrent callers, and destructive — a message is never returned twice. A nil result means nothing was queued. It stays callable after the turn is terminal so the owner can retrieve messages that were never delivered (Turn.PendingSteer reports their count).

func (*Turn) Interrupt 

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

Interrupt cooperatively asks the engine to stop. The first cause wins.

func (*Turn) OnInterrupt 

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

OnInterrupt implements agent.Observer. Turn.finish owns logical completion.

func (*Turn) OnRunEnd 

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

OnRunEnd implements agent.Observer. Turn.finish owns logical completion.

func (*Turn) OnRunRevise 

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

OnRunRevise moves commit authority to the next attempt while preserving turn-global delivery and acknowledgement cursors.

func (*Turn) OnRunStart 

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

OnRunStart implements agent.Observer.

func (*Turn) PendingSteer added in v0.4.6 

func (t *Turn) PendingSteer() int

PendingSteer reports how many steered messages are queued right now. After the turn is terminal a non-zero value means those messages never reached the conversation; they are also recorded on the result state under [steerPendingStateKey].

func (*Turn) Reply 

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

Reply supplies a response to one prompt without affecting other prompts.

func (*Turn) RunID 

func (t *Turn) RunID() string

RunID returns the immutable root execution identifier.

func (*Turn) State 

func (t *Turn) State() TurnState

State returns the current lifecycle state.

func (*Turn) Steer added in v0.4.6 

func (t *Turn) Steer(msg message.Message) error

Steer queues msg for the running turn. The message is delivered when something drains the turn's steer source — for a graph deployment, a node the author placed at a round boundary (host.drainSteer() in a script). Steer never injects mid-stream: the queue is read at boundaries the consuming document chose.

Unlike Turn.Interrupt, which is idempotent and silently ignores a dead turn, Steer always reports non-delivery:

  • errdefs.Validation for a malformed message;
  • ErrSteerTooLarge for a message over maxSteerMessageBytes;
  • ErrSteerQueueFull when maxSteerQueue messages are already queued;
  • agent.Interrupted once an interrupt was requested — the remaining waves are not guaranteed to run, so accepting would promise a delivery that cannot happen (mirrors Turn.askUser). The error classifies as interrupted (errdefs.IsInterrupted) and carries the pending interrupt, so recover it with errors.As on agent.InterruptedError;
  • ErrSteerClosed once the turn is terminal.

The message is stored as a deep copy: later mutation of msg does not affect the queued content. The queue is turn-scoped and never persisted, so a resumed run starts with an empty one.

func (*Turn) Wait 

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

Wait waits for terminal completion without affecting execution.

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"
)

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.