Documentation
¶
Overview ¶
Package controller is the single-writer, event-sourced core of agentsessions. It drives one session's executions against a durable event log and host-mediates every nondeterministic effect through api.EventSink, so it can either invoke-and-record it (live) or serve it from the journal (replay). On replay it re-executes the harness deterministically and never invokes the model — the load-bearing rule that makes resume-on-a-fresh-pod byte-identical. The log is any eventlog.Store (in-memory or sqlite); CAS + fencing are enforced on every append.
This drives the api.Harness Go interface in-process. A gRPC bridge that makes a harness running in a pod look like an api.Harness (the Harness.Connect stream) is a separate transport layer.
Index ¶
- Variables
- func Approve(log eventlog.Store, decision api.ApprovalDecision) (eventlog.Record, error)
- func CanPlace(harness api.Capabilities, runtime api.RuntimeCapabilities) bool
- func Fork(parent, child eventlog.Store, atSeq int64) error
- func ResumeInvocation(log eventlog.Store) (*api.ExecutionStart, error)
- type ApprovalState
- type Controller
- func (c *Controller) Exec(ctx context.Context, har api.Harness, inputs []api.Message, ...) (err error)
- func (c *Controller) Head() (int64, error)
- func (c *Controller) ModelInvocations() int
- func (c *Controller) Outputs() ([]string, error)
- func (c *Controller) Replay(ctx context.Context, har api.Harness) (outputs []string, err error)
- func (c *Controller) Resume(ctx context.Context, har api.Harness) (resumed bool, err error)
- func (c *Controller) ToolInvocations() int
- type ModelFunc
- type Observer
- type Option
- func WithFence(token int64) Option
- func WithHarness(name string) Option
- func WithLogger(logger *slog.Logger) Option
- func WithObserver(o Observer) Option
- func WithSessionUID(sessionUID string) Option
- func WithStart(config []byte, resumeFromSeq int64) Option
- func WithStreamingModel(fn StreamFunc) Option
- func WithToolExecutor(tool ToolFunc) Option
- type StreamFunc
- type ToolCallContext
- type ToolFunc
Constants ¶
This section is empty.
Variables ¶
var ErrApprovalBlocked = errors.New("controller: unresolved approval blocks a new execution")
ErrApprovalBlocked rejects a new turn while this session owns an approval intent without receipt.
var ErrApprovalReceiptReport = errors.New("controller: approval receipts are host owned")
ErrApprovalReceiptReport rejects harness reports and ordinary executor replies carrying host-owned approval receipt fields or unsupported result codes.
ErrApprovalUnavailable is returned by Exec, Resume and Replay for approval on a non-STATELESS_REPLAY harness, or when replay/recovery reproduces a historical pre-intent approval rejection without consuming a different call's evidence.
var ErrHarnessDescribeFailed = errors.New("controller: describe harness failed")
ErrHarnessDescribeFailed identifies failure to Describe the supplied harness before execution. The original cause is retained for callers to distinguish cancellation from endpoint failure.
var ErrHarnessMismatch = errors.New("controller: harness does not match recorded invocation")
ErrHarnessMismatch rejects replay/recovery with a different resolved harness name.
var ErrHarnessVersionMismatch = errors.New("controller: harness version does not match recorded invocation")
ErrHarnessVersionMismatch rejects replay/recovery when a nonempty recorded harness version differs from the supplied harness's advertised version, including an empty advertised version.
var ErrIncompleteInvocation = fmt.Errorf("%w: incomplete invocation", ErrInvalidExecutionLog)
ErrIncompleteInvocation identifies an unfinished trailing turn whose inputs were only partly committed. It wraps ErrInvalidExecutionLog; the caller can retry with Exec and all inputs.
var ErrInheritedToolIntent = fmt.Errorf("%w: inherited tool intent cannot be resumed", ErrInvalidExecutionLog)
ErrInheritedToolIntent rejects recovery of a forked execution whose copied prefix contains an unresolved tool intent. Re-driving it under the child's UID could repeat the parent's effect.
var ErrInvalidApprovalCall = errors.New("controller: invalid approval call identity")
ErrInvalidApprovalCall rejects missing or reused gated call identities before intent is written.
var ErrInvalidApprovalDecision = errors.New("controller: invalid approval decision")
ErrInvalidApprovalDecision identifies a missing, wrong, superseded, inherited, or changed decision tuple. It is distinct from corrupt recorded evidence and operational store failures.
var ErrInvalidExecutionLog = errors.New("controller: invalid execution log")
ErrInvalidExecutionLog is returned when modern execution-scoped events lack execution identity, a legacy-to-modern boundary is ambiguous, or a start marker cannot establish a complete invocation. Replay and resume reject such turns before running the harness.
var ErrMissingIdempotencyKey = errors.New("controller: CONTROLLER_MEDIATED tool call requires an idempotency key")
ErrMissingIdempotencyKey rejects a CONTROLLER_MEDIATED tool call that omits the idempotency key the crash-recovery re-drive needs to dedup its side effect (I3). Without a key, at-most-once silently would not hold, so the host fails loud rather than record an unrecoverable intent. The remaining key contract (generation, TTL) is the §10 spike; this is the minimal guard.
var ErrMissingSessionUID = errors.New("controller: a session UID is required")
ErrMissingSessionUID rejects an unscoped tool executor or a selected legacy harness invocation whose compatibility identity cannot be scoped to a session.
var ErrMissingToolExecutor = errors.New("controller: no tool executor configured")
ErrMissingToolExecutor is a recoverable host precondition, not an executor-failure receipt.
var ErrReplayDiverged = errors.New("controller: replay diverged from the journal")
ErrReplayDiverged is returned when replay or recorded-prefix recovery diverges from the journal: tool request identity or result correlation differs, required tool evidence is missing, or the harness leaves recorded effects unconsumed.
var ErrReplayInvokedModel = errors.New("controller: replay invoked the model (I1 violated)")
ErrReplayInvokedModel is returned when a replay caused a live model call — an I1 violation.
var ErrSinkClosed = errors.New("controller: invocation sink is closed")
ErrSinkClosed rejects effects from a retained sink after its Run has returned.
var ErrUnmediatedToolCall = errors.New("controller: ToolCall requires a host-mediated tier")
ErrUnmediatedToolCall rejects a ToolCall whose mediation tier is not host-executed. ToolCall is only for CONTROLLER_MEDIATED and REQUIRES_APPROVAL; an UNSPECIFIED or IN_HARNESS_REPORTED call must not execute here (the latter uses Report), so it is rejected before anything is recorded — closing the keyless-execute bypass.
Functions ¶
func Approve ¶
Approve commits only a decision, never an effect or a compute transition. Validation and exact historical retries happen before NewFence, so neither invalid input nor retries evict an existing writer. A new decision uses the head from that same validated scan for append CAS. Callers must serialize this operation with other session work; CAS also protects against external writers advancing the journal between validation and append.
func CanPlace ¶
func CanPlace(harness api.Capabilities, runtime api.RuntimeCapabilities) bool
CanPlace reports whether a harness can be scheduled on a runtime. The load-bearing rule: a harness that needs a memory snapshot cannot run on a snapshot-incapable runtime (e.g. a plain pod); a STATELESS_REPLAY harness runs anywhere.
func Fork ¶
Fork branches parent's log at atSeq into child: it copies the prefix [1..atSeq] verbatim (the canonical hashes reproduce identically, so the child shares the prefix chain) and records a LIFECYCLE_FORK marker as the child's first new event, whose prev_hash links to parent@atSeq — turning the hash-chain into a hash-tree. child must be an empty log.
func ResumeInvocation ¶
func ResumeInvocation(log eventlog.Store) (*api.ExecutionStart, error)
ResumeInvocation returns a copy of the start marker for the pending execution Resume would select, so callers can resolve its recorded harness before routing. Completed/empty journals and legacy markerless turns return nil. Invalid invocation evidence fails closed as in Resume. This query is read-only; the caller must serialize routing and recovery with other session work.
Types ¶
type ApprovalState ¶
type ApprovalState struct {
ExecutionID string
Invocation *api.ExecutionStart
Call *api.ToolCall
Request, Decision, Receipt *eventlog.Record
Head int64
Inherited, Completed bool
}
ApprovalState is derived exclusively from records in one session's journal. Request is nil for a cut after the call but before the host request; that cut is not addressable by Approve. Completed means the execution has an END; Receipt independently records call completion. Inherited identifies copied evidence even when a later matching receipt completes the call.
func InspectApproval ¶
func InspectApproval(log eventlog.Store, throughSeq int64) (*ApprovalState, error)
InspectApproval selects the latest execution by first-seen non-lifecycle identity, then its last gated call. A nongated latest execution returns nil. Zero reads the current journal; positive throughSeq bounds both selection and validation to that captured query prefix. Inspection does not fence, invoke a harness, or validate unrelated ordinary invocations. Callers must serialize inspection/decision with other session operations.
type Controller ¶
type Controller struct {
// contains filtered or unexported fields
}
Controller drives one session's log with a single incarnation (fence). It is meant to be driven by a single goroutine: Exec and Replay are NOT safe to call concurrently on the same Controller (liveModelCalls is unsynchronized). Concurrency BETWEEN controllers/processes is safe — the log's CAS + fencing reject a superseded writer.
func New ¶
New starts an incarnation over log. Unless the caller supplies a fence via WithFence, it advances the log's fencing token (superseding any prior incarnation, e.g. a dead pod) and binds to it. When WithFence is supplied (the placement layer minted the fence and stamped it on the incarnation), New uses that token instead — the log remains the single authority either way. A tool executor without a nonempty WithSessionUID returns ErrMissingSessionUID before advancing the fence.
func (*Controller) Exec ¶
func (c *Controller) Exec(ctx context.Context, har api.Harness, inputs []api.Message, expectedLastSeq int64) (err error)
Exec runs one live execution/turn. The first append (EXECUTION_START) is guarded by expectedLastSeq (the single-writer CAS at the session boundary); the harness runs host-mediated, and a completed turn ends with END. A durable approval request seals the sink and returns a typed park without ERROR/END. An owned unresolved gate blocks new turns before Describe.
func (*Controller) Head ¶
func (c *Controller) Head() (int64, error)
Head returns the current log head seq.
func (*Controller) ModelInvocations ¶
func (c *Controller) ModelInvocations() int
ModelInvocations is the number of live model calls made so far (0 across a pure replay).
func (*Controller) Outputs ¶
func (c *Controller) Outputs() ([]string, error)
Outputs returns the recorded assistant outputs in order.
func (*Controller) Replay ¶
Replay reconstructs the session by re-executing each completed modern execution with recorded effects. Pure legacy logs use v0.1.2's whole-log Run, even if unfinished. A modern boundary abandons the legacy tail after its last END for replay only; modern history retains it. Replay asserts the model is never invoked (I1) and each recorded model-input hash matches (I0). It is read-only.
func (*Controller) Resume ¶
Resume re-drives an interrupted last execution (crash-recovery, I4). If the journal's last turn did not complete (no END), Resume re-runs the harness with a hybrid sink that SERVES the already-recorded effects — never re-invoking a model/tool call whose result is recorded (at-most-once, I3) — and switches to LIVE (invoke + record) for anything past the crash point, then appends END. A MODEL_CALL that is the last recorded effect, with no OUTPUT, is the crash point: when the harness re-issues it with the same input hash, the model is invoked once and its completion recorded against that call (see the model re-drive in resumeSink.Model). A legacy ID-less (v0.1.2) journal cut at that point is not re-driven: Resume fails with "recorded completion missing" and does not call the model. If the last turn is complete or the log is empty, it is a no-op (returns false); legacy ERROR also finishes a turn. A start marker with missing or inconsistent input completeness information is rejected before the harness runs. An unresolved tool intent inherited across a fork returns ErrInheritedToolIntent before running the harness or appending events: the child's dedup namespace cannot recover the parent's effect.
An owned unanswered approval returns (true, *api.ApprovalParkedError) without Describe or Run. A call-only cut repairs its request exactly once using the captured head and current fence; repair failure returns (true, operational error). A decided call resumes under the original key.
Outside a legacy ID-less prefix, ERROR events, including those left by earlier failed Resume attempts, are not effects: they stay in the journal as an audit trail, are skipped when the effect stream is rebuilt, and do not end the execution, so a later Resume still recovers it.
func (*Controller) ToolInvocations ¶
func (c *Controller) ToolInvocations() int
ToolInvocations is the number of live tool calls made so far (0 across a pure replay).
type ModelFunc ¶
type ModelFunc func(ctx context.Context, req api.ModelRequest) (api.ModelResponse, error)
ModelFunc performs a live model invocation. It is the nondeterministic op the controller records on the live path and serves from the journal on replay.
ctx is the execution's context: it carries cancellation, the turn deadline, and request correlation, so a provider implementation can abort an in-flight completion when the turn is cancelled rather than running to completion against a caller that has gone away.
type Observer ¶
type Observer struct {
// OnRecord fires once per committed record, after it is durable.
OnRecord func(eventlog.Record)
// OnDelta fires for each ephemeral chunk. Nothing is logged, and replay never calls it.
OnDelta func(api.Delta)
}
Observer receives a turn's events as they happen. Sink callbacks are serialized in effect order for one invocation, so an implementation that writes to a stream needs no synchronization of its own but must not block for long.
type Option ¶
type Option func(*Controller)
Option configures a Controller at construction.
func WithFence ¶
WithFence binds the controller to a fence the caller already minted from the log (via NewFence), instead of minting its own. The placement layer uses this so the fence it stamps on the incarnation and the fence the controller appends under are the same log-minted token — the log stays the single fence authority. Fences are >= 1, so WithFence(0) is a no-op (mint-my-own).
func WithHarness ¶
WithHarness records and checks the resolved registry name independently of Descriptor.ID. Standalone callers that omit it use the supplied harness's descriptor ID instead.
func WithLogger ¶
WithLogger enables structured operational logs. Event payloads and fence values are never logged.
func WithObserver ¶
WithObserver reports a turn's records and streaming chunks as they happen, so a caller can relay them instead of waiting for the turn to finish and re-reading the log.
func WithSessionUID ¶
WithSessionUID binds the session identity passed to tool executors and controller logs. Direct controller users must set it to scope host authorization and durable tool deduplication.
func WithStart ¶
WithStart carries per-execution values the caller supplied straight through to the harness: the opaque config and the resume cursor. Exec journals them before running the harness; Replay and Resume use the recorded values, not this option. Only the harness knows what they mean.
func WithStreamingModel ¶
func WithStreamingModel(fn StreamFunc) Option
WithStreamingModel supplies a model that reports partial output. When set it is used for live invocations instead of the plain ModelFunc; replay ignores it entirely, since replay serves recorded completions and must not reach a provider.
func WithToolExecutor ¶
WithToolExecutor sets the executor for CONTROLLER_MEDIATED and approved tool calls. Without it, a harness that emits a host-mediated ToolCall gets an error (in-harness-reported tools use Report instead). A non-nil executor requires a nonempty WithSessionUID; New rejects an unscoped executor.
type StreamFunc ¶
type StreamFunc func(ctx context.Context, req api.ModelRequest, onChunk func(string)) (api.ModelResponse, error)
StreamFunc is a model invocation that reports partial output as it arrives. onChunk is called zero or more times before the final response, which must still be the complete message: the log records that, not the chunks, so a caller watching the stream and a caller replaying it later see the same finalized output.
It exists because the host mediates the model call. The harness blocks on one sink.Model and is unaware anything streamed, so live output needs no harness cooperation and no change to the harness wire protocol.
type ToolCallContext ¶
type ToolCallContext struct {
SessionUID string // Nonempty identity configured by WithSessionUID.
}
ToolCallContext carries host-owned identity for tool authorization and durable deduplication. It is separate from context.Context, which carries execution cancellation and deadlines.
type ToolFunc ¶
type ToolFunc func(ctx context.Context, scope ToolCallContext, call api.ToolCall) (api.ToolResult, error)
ToolFunc executes a CONTROLLER_MEDIATED or approved REQUIRES_APPROVAL tool. The host calls it between appending the TOOL_CALL intent and appending the TOOL_RESULT (the two-phase write-ahead of §3/I3). The host owns tool/resource authorization and durable deduplication scoped by scope.SessionUID plus the harness-chosen IdempotencyKey, which is only unique within a session. On same-session crash-recovery it receives the same session and recorded call, and MUST NOT repeat the external effect. Resume rejects unresolved inherited intents across a fork. ctx carries the execution's cancellation and deadline.