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 ForkWithComputeInstance(parent, child eventlog.Store, atSeq int64, instance string) error
- func LatestComputeInstance(recs []eventlog.Record, sameHarness func(recorded string) bool) (instance string, seq int64)
- func ResumeInvocation(log eventlog.Store) (*api.ExecutionStart, error)
- func VerifyForkPrefix(parent eventlog.Store, atSeq int64) error
- func VerifyJournal(log eventlog.Store) 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 WithComputeInstance(instance string, sameHarness func(recorded string) bool) Option
- func WithDurableDedup(declared func(tool string) bool) Option
- func WithFence(token int64) Option
- func WithHarness(name string) Option
- func WithLogger(logger *slog.Logger) Option
- func WithModelDefault(model string) Option
- func WithModelPolicy(policy api.ModelPolicy) 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 ErrJournalTampered = errors.New("controller: refusing a journal that failed hash-chain verification")
ErrJournalTampered refuses a journal whose hash chain fails verification (CHAIN-12). Exec, Replay, Resume and Outputs return it before they run a harness, call a model or tool, serve a recorded effect or append; Fork returns it before it fences or writes the child, and VerifyForkPrefix before a placer checkpoints the parent or provisions a child. The returned error also wraps eventlog.ErrTampered (which is sqlitelog.ErrTampered) and names the first bad seq.
The chain is unkeyed, so this catches storage corruption and edits that do not recompute the hashes. A writer that rewrites events and recomputes every later hash is not detected, and neither is removal of records from the end of the journal by a controller that has not already verified those records.
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 ErrUnansweredToolCall = errors.New("controller: tool call has no recorded result; repeating it could repeat the action")
ErrUnansweredToolCall refuses Resume of an execution whose last recorded effect is a TOOL_CALL that is not approval-gated and has no TOOL_RESULT when the configured executor has not declared durable dedup for that tool (WithDurableDedup). Whether the executor already applied the effect is unknown, so re-running the call could repeat it. Resume returns it before Describe, Run or any append, so the journal is unchanged and a later Resume under a declaring executor can still re-drive the call.
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. The marker records no compute instance; ForkWithComputeInstance records the one the child was cloned onto. A copied prefix that fails hash-chain verification returns ErrJournalTampered before the child is fenced or written.
func ForkWithComputeInstance ¶ added in v0.3.0
ForkWithComputeInstance is Fork, and also records instance (the child's api.Incarnation.ComputeInstance, the clone a stateful fork made) on the FORK marker. Until the child records an instance of its own, LatestComputeInstance returns it, so a turn, Resume or snapshot placed on another instance (a replacement created under the child's name before its first turn) is refused with api.ErrMemoryStateLost. An empty instance records nothing, as Fork does.
func LatestComputeInstance ¶ added in v0.3.0
func LatestComputeInstance(recs []eventlog.Record, sameHarness func(recorded string) bool) (instance string, seq int64)
LatestComputeInstance returns the compute instance the session's latest record of one names, and that record's seq. Only records after the last FORK marker count, plus the marker itself:
- a turn's EXECUTION_START.compute_instance, counting only turns sameHarness attributes to the caller's harness (nil counts every turn);
- the FORK marker's Lifecycle.compute_instance, the clone a stateful fork made for the child. It is attributed like a turn with no recorded harness name (sameHarness("")), since the clone runs the session's own harness.
Records that name no instance are skipped. It returns "" when no counted record names one: a session with no memory-tier turn, or a fork child whose marker predates the field (or a stateless fork) and that has not run a turn yet.
A forked child's own compute is a clone, not the parent's instance, so instances its copied prefix recorded do not count. Exec and Resume compare the instance they were placed on with it, and Suspend and a stateful Fork pass it to Runtime.Snapshot as the instance the session must still be.
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.
func VerifyForkPrefix ¶ added in v0.3.0
VerifyForkPrefix checks the hash chain of the parent records Fork copies, seq 1 through atSeq, and returns ErrJournalTampered on a mismatch (CHAIN-12). A caller that checkpoints the parent or provisions child compute before ForkWithComputeInstance runs calls it first, so a tampered parent is refused before any of that work.
func VerifyJournal ¶ added in v0.3.0
VerifyJournal checks the hash chain of every record in log and returns ErrJournalTampered on a mismatch (CHAIN-12). A caller that reports or extends a journal without a Controller (an approval park or request repair) calls it first.
Types ¶
type ApprovalState ¶
type ApprovalState struct {
ExecutionID string
Invocation *api.ExecutionStart
Call *api.ToolCall
Request, Decision, Receipt *eventlog.Record
Head int64
Inherited, Completed bool
// contains filtered or unexported fields
}
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. A journal whose hash chain fails verification returns ErrJournalTampered before Describe or any append.
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, after the same chain verification as Replay.
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. A journal whose hash chain fails verification returns ErrJournalTampered before the harness runs.
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. Any trailing TOOL_CALL that is not approval-gated and has no TOOL_RESULT is re-driven under its recorded key only when the executor declares durable dedup for its tool (WithDurableDedup); otherwise Resume returns ErrUnansweredToolCall, also before Describe, Run or any append. Without a tool executor the call is not refused here; re-issuing it fails as on the live path.
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.
A journal whose hash chain fails verification returns ErrJournalTampered before Describe, Run, an approval repair or any append, even when its last turn is complete.
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 WithComputeInstance ¶ added in v0.3.0
WithComputeInstance binds the controller to the compute instance the caller placed the session on (api.Incarnation.ComputeInstance). Exec records it in the turn's EXECUTION_START. Exec and Resume first compare it with the latest instance the journal recorded for the same harness since the session's FORK marker, or the clone that marker recorded (LatestComputeInstance), and refuse with api.ErrMemoryStateLost when the two differ, before they append anything or run the harness.
sameHarness reports whether a turn recorded under a harness name ran on the harness this controller drives (the Placer resolves names through its registry); nil treats every recorded name as this harness. An empty instance records nothing and skips the comparison.
Exec compares against the journal it then appends to under its compare-and-swap, so a turn another host journaled before this one cannot be missed: either it is in the history the comparison read, or the append fails.
func WithDurableDedup ¶ added in v0.3.0
WithDurableDedup declares, per tool name, that the WithToolExecutor executor deduplicates that tool durably by (scope.SessionUID, call.IdempotencyKey), or that repeating the call is harmless (a read-only or naturally idempotent tool). Resume consults declared with the recorded call's tool name only when an interrupted execution's last recorded effect is a TOOL_CALL that is not approval-gated and has no TOOL_RESULT: true re-runs that call under its recorded key; false, or no WithDurableDedup at all, makes Resume return ErrUnansweredToolCall. First attempts, recorded results and approved REQUIRES_APPROVAL calls never consult it. declared must not block.
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 WithModelDefault ¶ added in v0.3.0
WithModelDefault supplies a session's requested default for a new policy-enabled execution. Empty uses the policy default. This option alone enables nothing; Replay/Resume always restore the invocation's recorded default, including empty for pre-policy turns.
func WithModelPolicy ¶ added in v0.3.0
func WithModelPolicy(policy api.ModelPolicy) Option
WithModelPolicy opts new executions into host-owned model admission. Default and Allowed are explicit; Descriptor.Models never supplies a default. Replay serves recorded calls independently of today's policy. Resume applies it only to live effects of invocations with a recorded Model. Without this option, new executions keep the harness's existing model behavior.
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. Resume re-drives an interrupted call only for tools declared with WithDurableDedup.
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. ctx carries the execution's cancellation and deadline.
Resume re-runs a call that is not approval-gated and whose TOOL_CALL is the interrupted execution's last recorded effect (including a call whose executor returned an error the harness did not handle) only for a tool the executor declares through WithDurableDedup; it then receives the same session and recorded call and MUST NOT repeat the external effect. For an undeclared tool Resume returns ErrUnansweredToolCall instead. Resume rejects unresolved inherited intents across a fork whether or not the tool is declared. An approved REQUIRES_APPROVAL call runs on Resume without consulting the declaration.
Source Files