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

agent

package
v0.9.0 Latest Latest
Warning

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

 Go to latest Published: Oct 1, 2026 License: Apache-2.0 Imports: 36 Imported by: 0

Documentation

Overview

Package agent is a durability-first, Go-idiomatic toolkit for building AI agents.

The agent loop journals every step to a Durable store via named-step memoization (Durable.Do) and resumes after a crash — reusing recorded steps, re-running retry-safe tools, halting (rather than double-executing) a non-idempotent tool whose outcome is unknown, and pausing durably for human approval when a tool requires it. Orchestration is plain Go (Option B); the model call is wrapped by a func(Handler) Handler middleware chain. Reliability wrappers ship in the middleware package: per-attempt timeouts, classified retry with backoff, hedged model calls (middleware.Hedge), and rate limiting.

Scope: this resumes AROUND tool boundaries, not the internals of a single in-flight tool call. The precise promise is "survives crashes around tool calls, never double-fires an unsafe side effect" — not "mid-tool-call resume".

Index

Examples

Constants

View Source 
const (
	ReasonNotEligible = "not an eligible approver"
	ReasonSuperseded  = "superseded by the approver's earlier valid decision"
	ReasonNoKey       = "no key for this approver"
	ReasonBadSig      = "signature does not verify for this call"
	ReasonSharedKey   = "the approver's signing key also serves another approver"
	ReasonNoKeyID     = "the approver's verifier reports no key identity"
	ReasonAlg         = "signed under another scheme than the approver's key"
)

Reasons a decision record does not count.

View Source 
const JournalFormat = "bide.journal.v1-dev"

JournalFormat is the journal format this version writes: the format every run's header names (see StepHeader). Before 1.0 it is one dev tag, "bide.journal.v1-dev", the same in every pre-release and never bumped per change: a pre-release journal is not promised to be readable by a later pre-release. It becomes "bide.journal.v1" at the 1.0 tag, so a 1.0 binary refuses every pre-release journal and none is read under rules it was not written for.

View Source 
const SaltSize = 32

SaltSize is the length of Record.Salt: 32 bytes (256 bits) from crypto/rand.

Variables

View Source 
var (
	// ErrConfig is a programmer-facing misuse or misconfiguration, not a runtime fault.
	ErrConfig = errors.New("config")
	// ErrModel is a model-provider call that failed: HTTP status, decode, stream error.
	ErrModel = errors.New("model")
	// ErrTool is a tool that failed or was misused (unknown tool, bad arguments).
	ErrTool = errors.New("tool")
	// ErrStorage is a durable-store I/O failure: marshal, insert, query, history.
	ErrStorage = errors.New("storage")
	// ErrProtocol is malformed wire or journal data (bad log entry, unknown content part).
	ErrProtocol = errors.New("protocol")
	// ErrBudget is a middleware limit tripping (e.g. a token budget).
	ErrBudget = errors.New("budget")
)

The toolkit classifies failures with sentinel errors matched by errors.Is — the standard-library idiom (io.EOF, sql.ErrNoRows). There are two tiers:

  • Category sentinels (ErrModel, ErrTool, ErrStorage, ErrConfig, ErrProtocol, ErrBudget) — the coarse class of failure, for a broad branch or a metric label.
  • Condition sentinels (ErrUnknownTool, ErrToolArgs, …) — a specific cause. Each wraps its category, so errors.Is(err, ErrUnknownTool) and errors.Is(err, ErrTool) both hold.

Every error the toolkit returns wraps exactly one category (and usually the underlying cause too, via a second %w). Match the class you care about: 

if errors.Is(err, agent.ErrModel)   { retryLater() }   // any model-provider fault
if errors.Is(err, agent.ErrUnknownTool) { … }          // one specific condition

The control-flow signals are richer than a category, so they stay concrete types matched with errors.As, not sentinels: *PendingApproval (human approval needed), *ResumeHalt (unsafe to resume), *SagaAborted (transaction rolled back). context cancellation surfaces as the usual context.Canceled / context.DeadlineExceeded.

View Source 
var (
	ErrUnknownTool = fmt.Errorf("unknown tool: %w", ErrTool)
	ErrToolArgs    = fmt.Errorf("invalid tool arguments: %w", ErrTool)
	// ErrToolOutcomeUnknown is a tool call that may or may not have taken effect: the tool
	// sent its request but lost the connection (or its deadline passed) before the answer
	// came back. A tool returns an error wrapping it instead of an ordinary failure when it
	// cannot tell. For a tool that is not retry-safe the agent records no result, as it does
	// for a cancelled call: the run stops with this error, and a resume halts (*ResumeHalt)
	// rather than run the side effect a second time. For a retry-safe tool the call is an
	// ordinary failure the model sees, since running it again does no harm.
	ErrToolOutcomeUnknown = fmt.Errorf("tool outcome unknown: %w", ErrTool)
	// ErrToolReinvoked is a tool middleware calling a tool that is not retry-safe a second
	// time for the same tool call. The agent refuses and does not run the tool again.
	ErrToolReinvoked    = fmt.Errorf("tool invoked again for one call: %w", ErrConfig)
	ErrNoRecordedOutput = fmt.Errorf("no recorded model output: %w", ErrModel)
	// ErrIncompleteResponse is a model stream that ended without a Finish event: the
	// response stopped partway through a turn, so what arrived is not the model's answer.
	ErrIncompleteResponse = fmt.Errorf("model response ended before the turn finished: %w", ErrModel)
	ErrTruncatedToolArgs  = fmt.Errorf("truncated tool-call arguments: %w", ErrProtocol)
	// ErrOutputTruncated is a model turn that ended at its output token limit (or the context
	// window): Finish reason FinishLength. What arrived is only the start of the model's answer,
	// so it is not journaled. It wraps ErrModel, so middleware.Retryable retries it; the same
	// request usually ends the same way, so raise the limit (MaxTokens) rather than rely on a retry.
	ErrOutputTruncated = fmt.Errorf("model output cut off at its token limit: %w", ErrModel)
	// ErrOutputFiltered is a model turn that a provider's safety or content filter stopped, or
	// that the model refused: Finish reason FinishFiltered. What arrived is not the model's whole
	// answer, so it is not journaled. It wraps ErrModel, so middleware.Retryable retries it.
	ErrOutputFiltered = fmt.Errorf("model output stopped by a content filter or refusal: %w", ErrModel)
	// ErrStreamProtocol is a model stream that broke its provider's event protocol, such as
	// content after the turn's end or an event after the Finish. What arrived is not a turn the
	// model finished, so it is not the answer. It wraps ErrModel as well as ErrProtocol: the
	// turn is not journaled and a fresh attempt can succeed, so a retry middleware retries it.
	ErrStreamProtocol = fmt.Errorf("model stream broke its protocol: %w (%w)", ErrProtocol, ErrModel)
	// ErrToolUseIDReused is a model turn whose tool call has no ID, repeats an ID from an
	// earlier turn of the conversation, repeats one within the turn, or has an ID that is not
	// valid UTF-8 (the journal cannot hold it). The loop keys each call's result and journal
	// step by its ID, so a reused ID would pass the call off as already done. Any other ID is
	// safe: the keys derived from it encode it (see ToolResultStep and SubRunID). It wraps
	// ErrModel as well as ErrProtocol: the fault is in the model's output, the turn is not
	// journaled, and a fresh attempt can issue a valid turn, so a retry middleware treats it as
	// retryable.
	ErrToolUseIDReused = fmt.Errorf("tool-use id missing, reused, or malformed: %w (%w)", ErrProtocol, ErrModel)
	// ErrNegativeUsage is token usage with a negative count, reported by a model or returned by
	// a middleware. The usage is rejected rather than lowering a run's totals and budget. It
	// wraps ErrModel as well as ErrProtocol.
	ErrNegativeUsage = fmt.Errorf("negative token usage: %w (%w)", ErrProtocol, ErrModel)
	// ErrQuotaExhausted is a provider refusing a call because the account's quota or credit
	// is used up (OpenAI insufficient_quota, Anthropic billing_error, HTTP 402, a Gemini daily
	// quota), as opposed to a momentary rate limit. Waiting a few seconds does not lift it, so
	// middleware.Retryable does not retry it. It arrives as an *APIError.
	ErrQuotaExhausted = fmt.Errorf("provider quota or credit exhausted: %w", ErrModel)
	// ErrResponseTooLarge is a streamed response with a line longer than the adapter reads, or
	// longer in all than its adapter's cap (see MaxSSELine and LimitResponse in model/provider). The same
	// request would produce it again, so middleware.Retryable does not retry it.
	ErrResponseTooLarge = fmt.Errorf("model response line too large: %w", ErrModel)
	ErrBudgetExceeded   = fmt.Errorf("budget exceeded: %w", ErrBudget)
	ErrMaxTurns         = fmt.Errorf("max turns exceeded: %w", ErrBudget)
	// ErrInvalidApproval is an approver decision rejected at submission by SubmitDecision's
	// WithDecisionCheck: no such tool call, an unknown approver, or a signature that does not
	// verify for this exact call.
	ErrInvalidApproval = fmt.Errorf("invalid approval: %w", ErrConfig)
	// ErrAlreadyDecided is a decision submitted (with WithDecisionCheck) by an approver whose
	// earlier valid decision on the same call already counts; the new one would be ignored.
	ErrAlreadyDecided = fmt.Errorf("approver already decided: %w", ErrConfig)
)

Condition sentinels — specific causes, each wrapping its category.

View Source 
var ErrAlreadyResolved = fmt.Errorf("halt already has a different recorded outcome: %w", ErrConfig)

ErrAlreadyResolved is the condition of a resolution refused because the operation already has a different recorded outcome (see HaltAlreadyResolved). It wraps ErrConfig.

View Source 
var ErrJournalVersion = fmt.Errorf("unsupported journal format: %w", ErrProtocol)

ErrJournalVersion is a run whose journal this version cannot read or write: its header names a journal format this version does not support, or the journal has no header first (written before the header existed, or by something other than a Journal). It wraps ErrProtocol. The error the journal returns is a *JournalVersionError, which wraps it.

View Source 
var ErrLeaseLost = errors.New("run lease lost")

ErrLeaseLost ends a drive whose run lease was lost while it ran: the store reported the lease no longer held (it lapsed and another process took it), or no renewal succeeded before the renewal cutoff (see Lease). The drive's context is cancelled with it as the cause (context.Cause), and Lease returns the drive's error wrapped with it. It is a coordination signal, not a fault, so like context.Canceled it wraps no category: the run is not failed, it is another driver's to continue. Recover does not count it as a failure.

View Source 
var ErrNoLiveAttempt = fmt.Errorf("operation has no live attempt to resolve: %w", ErrConfig)

ErrNoLiveAttempt is the condition of a resolution refused because the operation named has no live attempt marker: it never halted (it was never attempted, or its only attempts are recorded as not started and are re-attempted by the next drive). It wraps ErrConfig.

Functions

func Ack 

func Ack(ctx context.Context, d Durable, runID, channel, key string) error

Ack marks a message consumed so Receive advances past it. Idempotent (the first ack for a (runID, channel, key) wins) and journaled, so the advance survives a crash and a resume.

func AnswerInterrupt added in v0.9.0 

func AnswerInterrupt[T any](ctx context.Context, d Durable, runID, name string, value T) error

AnswerInterrupt records the typed value a paused run is waiting for at the interrupt point name (see Interrupt); then re-invoke Run with the pause's RootRunID to continue. It is idempotent: the first value for a (runID, name) wins. The value survives a crash: it is a journaled step.

func ApprovalDecisionBytes added in v0.4.0 

func ApprovalDecisionBytes(s ApprovalSubject, approverID string, approved bool) []byte

ApprovalDecisionBytes returns the canonical bytes an approver signs and every verifier checks. The layout, stable and reproducible outside Go: 

"bide.approval.v2\n"
len32(RunID) RunID
len32(ToolUseID) ToolUseID
len32(ToolName) ToolName
SHA-256(canonical args), 32 bytes
len32(approverID) approverID
1 byte: 1 approved, 0 denied

len32 is a 4-byte big-endian length, so no two distinct decisions encode to the same bytes. The canonical args are the arguments parsed as JSON and re-serialized with object keys sorted, no insignificant whitespace, no HTML escaping, and number literals kept verbatim; empty arguments are "{}", and bytes that are not a single JSON value are used as-is. The canonical form makes the signature independent of how the arguments were formatted in transit (a live model stream, the journal, an evidence file).

func ApprovalTallyStep added in v0.4.0 

func ApprovalTallyStep(toolUseID string) string

ApprovalTallyStep is the journal name of the gate's terminal tally for toolUseID.

func Approve 

func Approve(ctx context.Context, d Durable, runID, toolUseID string, approved bool) error

Approve durably records a human approve/deny decision for a tool call (HITL). It is idempotent: the first decision for a (runID, toolUseID) wins. After approving, re-run the agent with the pause's RootRunID and it resumes past the *ApprovalPending pause. The decision survives a crash because it's a journaled step like any other.

func Await 

func Await[T any](ctx context.Context, name string) (T, error)

Await blocks the current run until a single-shot signal named `name` is delivered, then returns its payload. Call it from inside a tool (the agent loop supplies the run context). On first encounter, with no signal recorded, it returns the zero T and a *SignalPending error that propagates out of Run, pausing the run durably. After Signal records a payload for the same name and Run is re-invoked, Await returns that payload and execution continues past this point. The payload survives a crash: it is a journaled step.

Await is the externally-pushed counterpart of Interrupt. Like Interrupt and Sleep it must be called from a retry-safe tool (Safety.ReadOnly or Idempotent): on resume the tool re-runs from the top until the await resolves, so everything before the Await call must be safe to repeat. Use distinct names for distinct awaits; each pauses and resolves independently.

func AwaitFor 

func AwaitFor[T any](ctx context.Context, name string, d time.Duration) (T, bool, error)

AwaitFor blocks the current run until either a single-shot signal named `name` is delivered or the duration d elapses, whichever happens first. It returns (payload, true, nil) when the signal wins, or (zero, false, nil) when the timeout wins. Call it from inside a retry-safe tool (Safety.ReadOnly or Idempotent), like Await and Sleep: on resume the tool re-runs from the top until the await resolves, so everything before the AwaitFor call must be safe to repeat. A Waker that fails to schedule the deadline's wake fails the run as it does for Sleep.

It is Await composed with a durable timer. The deadline is journaled once, on the first encounter (at-most-once by name), as now()+d, so a resumed or crash-recovered run races against the same absolute instant rather than restarting the clock. While neither side has resolved, AwaitFor schedules a wake for the top-level run with the bound Waker (if any) and returns *SignalPending, pausing the run durably exactly like Await. The bool return is only meaningful when err is nil.

The outcome is journaled too, at-most-once by name, the first time either side wins. Every later evaluation of the same await (the tool re-runs because it pauses afterwards, or the process dies before the tool's result is recorded) returns that recorded outcome, so a signal delivered after the timeout won cannot flip the call to the signal branch.

func Capability added in v0.8.0 

func Capability[T any](store Store) (T, bool)

Capability returns store's implementation of the optional capability T (Lister, Leaser, or any other interface a store may implement beyond Store), looking through wrappers.

A store that wraps another store exposes the store it wraps by implementing 

Unwrap() Store

Capability checks store itself first, then follows Unwrap until a store implements T, and reports false once a store has no Unwrap method or Unwrap returns nil. A wrapper therefore exposes exactly the capabilities of the store it wraps, with no forwarding methods to keep in step with new capabilities. A wrapper that must change what a capability means implements that capability itself, which takes precedence over the wrapped store's, as with errors.As. Only a wrapper that passes run IDs and names through unchanged may implement Unwrap (see Store). The same holds for a Durable wrapper's Unwrap() Durable (see Durable), which the engine follows to the store beneath for its capabilities and for what the process keeps per run.

func EncodeRecord added in v0.7.0 

func EncodeRecord(r Record) ([]byte, error)

EncodeRecord returns the journal encoding of r: the bytes a store persists for it and the bytes an audit leaf commits to. Every store writes exactly these bytes, so a record's journal form does not depend on the backend, and every store hands back DecodeRecord of them, on the live path as on replay, so a resumed run rebuilds exactly the conversation the live run had.

The encoding is JSON with two properties that make it canonical:

  • It is a fixed point: decoding it with DecodeRecord and encoding the result again yields the same bytes. So a store holds exactly EncodeRecord(rec) for the record rec it hands back, and an audit leaf computed from a record read back from any store is the bytes that store persisted.
  • It keeps the content of every json.RawMessage (a tool's arguments and result, a step's value, a reconciler's evidence) byte for byte, except whitespace between JSON tokens, which is removed. Nothing is HTML-escaped: a tool result that says "<b>" is journaled as "<b>", not with the escapes json.Marshal writes for < and >, so an adapter that forwards the JSON text to a model sends what the tool returned. Key order, number spelling ("1.50", "1e20"), escapes already in the text, and even invalid UTF-8 inside a JSON string are kept as they are.

Insignificant whitespace is dropped because encoding/json compacts every raw value it embeds, and one spelling per JSON value is what an audit leaf should commit to. For the same reason the line and paragraph separators U+2028 and U+2029 are always written as their JSON escapes, which encoding/json emits in some builds but not others. In a Go string field (message text, a reasoning signature), invalid UTF-8 becomes U+FFFD, as encoding/json writes it; since the stores hand out only the decoded form, the live and replayed conversations agree.

func Enqueue added in v0.9.0 

func Enqueue[T any](ctx context.Context, d Durable, runID, channel, key string, payload T) error

Enqueue appends a message to an ordered per-run channel, deduped by key: a redelivery with the same (runID, channel, key) is a no-op and the first payload wins (at-most-once intake over at-least-once transport). Safe to call from any process; the store's PK / ON CONFLICT is the cross-process dedup, exactly as for Signal. After enqueueing, re-invoke a run paused on the channel (*SignalPending) with the pause's RootRunID.

func Generate 

func Generate(ctx context.Context, m Model, req Request) (Message, Usage, error)

Generate is the convenience drain: stream and assemble in one call.

func InSaga 

func InSaga(ctx context.Context) bool

InSaga reports whether the current tool execution is inside a saga run. Sub-agents use it to run transactionally (RunSaga) so a failure deep in the tree aborts the whole tree.

func Interrupt 

func Interrupt[T any](ctx context.Context, name string, prompt any) (T, error)

Interrupt pauses the current run to request typed human input, identified by name. Call it from inside a tool (the agent loop supplies the run context). On first encounter it returns the zero T and an *InterruptPending error that propagates out of Run, pausing the run durably. After AnswerInterrupt records a value for the same name and Run is re-invoked, Interrupt returns that value and execution continues past this point. This generalizes approve/deny (a bool) to an arbitrary typed answer.

Interrupt must be called from a retry-safe tool (Safety.ReadOnly or Idempotent): on resume the tool re-runs from the top until the interrupt resolves, so everything before the Interrupt call must be safe to repeat. Use distinct names for multiple interrupt points; each pauses and resumes independently.

func IsApprovalDecision added in v0.4.0 

func IsApprovalDecision(r Record, toolUseID string) bool

IsApprovalDecision reports whether r is an m-of-n approver decision (written by SubmitDecision) on toolUseID. The single-approver Approve record has no Approver and is not one.

func IsComplete 

func IsComplete(ctx context.Context, store Durable, runID string) (bool, error)

IsComplete reports whether runID has reached its terminal answer, by checking the journal for the durable completion marker the agent loop records at the end of a run (see runCompleteStep). A crash-recovery supervisor uses it to skip finished runs.

func IsPause added in v0.9.0 

func IsPause(err error) bool

IsPause reports whether err, or any error it wraps, is a Pause: the run stopped durably and is waiting, rather than failed.

func IsReservedStepName added in v0.8.0 

func IsReservedStepName(name string) bool

IsReservedStepName reports whether name starts with a prefix the engine reserves for its own journal keys. Step and Parallel refuse such a name, and so do ResolveHaltRef and ResolveStepHalt for a step, except a plan flow node's key ("node:<name>"), which halts as a Step does.

func IsSessionRun added in v0.8.0 

func IsSessionRun(runID string) bool

IsSessionRun reports whether runID is a session's journal or one of its turn runs, which the session drives (see Session). A sub-agent called in a turn is IsSubRun instead.

func IsSubRun added in v0.8.0 

func IsSubRun(runID string) bool

IsSubRun reports whether runID is a sub-agent's run (SubRunID), which its root run drives: its last '>' is followed by an encoded call ID, not by a session's mark.

func JournalEntry added in v0.7.0 

func JournalEntry(name string, rec Record) ([]byte, error)

JournalEntry returns the bytes a store persists when it records rec as the step named name: rec with Name set to name and a fresh random Salt (see Record.Salt; a salt the engine drew for the record is kept), in its journal encoding (EncodeRecord). Every Durable implementation must record a new step through it, so every record carries a salt; the audit package refuses to commit a record without one. It errors only if the system's random source fails or rec cannot be encoded.

func Lease 

func Lease(ctx context.Context, store Durable, runID string, drive func(context.Context) error, opts ...RecoverOption) (bool, error)

Lease runs drive under an exclusive, auto-renewed lease on runID, so a primary driver and a recoverer (or two workers) do not drive the same run at once. If the store implements Leaser and another holder currently leases the run, drive is NOT called and Lease returns (false, nil). If neither the store nor a store it wraps implements Leaser (see Capability), drive runs unconditionally. The bool reports whether drive ran; the error is the acquisition error (when false) or drive's own error (when true).

If the lease is lost while drive runs (see driveWithRenew for the renewal schedule), drive's context is cancelled with ErrLeaseLost as its cause (context.Cause), and a drive that then returns an error has it wrapped with ErrLeaseLost, so errors.Is(err, ErrLeaseLost) tells a lost lease from a shutdown or a genuine failure. A drive that returns nil succeeded regardless.

A primary driver wraps Agent.Run so a live run and a recoverer never both drive it, using the same lease the recoverer respects: 

driven, err := agent.Lease(ctx, store, runID, func(ctx context.Context) error {
    _, err := ag.Run(ctx, runID, input)
    return err
}, agent.WithLeaseHolder("worker-1"))

func NextOnceKey added in v0.7.0 

func NextOnceKey(ctx context.Context) string

NextOnceKey returns a key for the next exactly-once operation (an idempotent append, say) that the tool call running in ctx makes, or "" outside a tool call. The call's operations are numbered in the order they ask for keys, and the numbering starts again each time the call runs, so an operation keeps its key when the call runs again after a crash, and two operations of one call never share one.

The numbering is stable only if the call asks for its keys in the same order each time it runs. Work a call does concurrently must therefore run in steps (Step, or Parallel's tasks): each step numbers its own operations, under a key that names the step, so concurrent steps keep their keys whatever order they run in.

func Parallel 

func Parallel[T any](ctx context.Context, d Durable, runID string, maxConcurrency int, tasks ...Task[T]) ([]T, error)

Parallel runs tasks concurrently, each as a durable named step (see Step), and returns their results in task order. It is the durable, auditable fan-in that compliance-style pipelines want: run several independent checks at once (sanctions, credit, fraud), each one crash-safe, each result committed to the journal and provable on its own, then aggregate.

All tasks run even if some fail, so the full set of results is preserved (a failed check does not hide the others). The returned error joins every task's error and is nil only if all succeeded. Succeeded tasks are memoized on a later resume. A failed task was not journaled: one marked retry-safe (Task.Safety) re-runs, and a side effect halts with *OutcomeUnknown, as Step does, since a failed effect may still have landed. Task names must be unique and not empty; a repeated or empty name is ErrConfig. maxConcurrency caps in-flight tasks; <= 0 means one goroutine per task.

This is deliberately a thin primitive over the durable journal, not a graph engine: dynamic, model-driven routing stays in plain Go and sub-agents; Parallel covers the static fan-out/fan-in that a governed workflow's parallel-checks-then-decide stage is made of.

As with Step, each task result is journaled as JSON, so T must be JSON-serializable; a struct's unexported fields silently round-trip to their zero values.

func PromptDigest added in v0.9.0 

func PromptDigest(msgs []Message) string

PromptDigest is the hex SHA-256 digest a model record journals of the system prompt a turn was sent: the text of every system message in msgs, in order, each length-prefixed under a domain tag, so no two different prompts share a digest. It is "" when msgs holds no system message.

func Recover 

func Recover(ctx context.Context, store Durable, resume func(ctx context.Context, runID string) error, opts ...RecoverOption) (int, error)

Recover re-drives the runs that were in flight when the process died, in one pass. It enumerates the runs the store holds that are not over (via Lister, which filters out every run holding a terminal marker: run:complete, run:aborted or run:cancelled), and calls resume for each to push it forward. A run another driver finished after the listing is not resumed: holding the run's lease, Recover checks the terminal markers again before it calls resume (over a Journal, three point reads per run it drives). It returns how many runs it re-drove and the joined genuine failures (nil if none).

The store must implement Lister; a store that cannot enumerate its runs (the base Durable contract does not require it) yields an ErrConfig-wrapped error.

If the store also implements Leaser, Recover coordinates across processes: it claims an exclusive, renewed lease per run before driving it and skips a run another holder currently leases, so competing recoverers do not both re-drive the same run. A crash lets the lease expire (see WithLeaseTTL), and the next pass of another process takes the run over: Recover is one pass, so run RecoverLoop to keep passing. Without Leaser, Recover drives every enumerated run, which is safe under at-most-once memoization but redundant across processes.

A pause is a SUCCESS, not a failure. A re-driven run that is still waiting returns a Pause (*ApprovalPending, *InterruptPending, *TimerPending, *SignalPending, *OutcomeUnknown); Recover detects it with IsPause and does NOT record it as an error: it means "recovered, still waiting", and the run resumes later when its condition is met (a human approves, an interrupt is answered, a timer fires). Only a genuine error (a model or storage fault, a bad tool) is joined into the returned error. A run whose lease was lost mid-drive (ErrLeaseLost) is not joined either: another process holds it now and carries it on.

resume is deployment POLICY, not a mechanism the SDK can supply: it knows which agent drives a run and any Waker or clock to bind onto the context (a Waker-bound resume rebuilds the timer set for sleeping runs, since Sleep re-registers its wake on replay). The run's input and entry point (Run or RunSaga) are in its journal (see RecordedStart), and a resume with another input or entry point is ErrConfig. A typical resume is: 

func(ctx context.Context, runID string) error {
    start, ok, err := agent.RecordedStart(ctx, store, runID)
    if err != nil {
        return err
    }
    if !ok {
        start = startFor(runID) // your own record, for a run not driven under this version
    }
    ctx = agent.WithWaker(ctx, w)
    if start.Saga {
        _, err = a.RunSaga(ctx, runID, start.Input)
    } else {
        _, err = a.Run(ctx, runID, start.Input)
    }
    return err
}

Recover skips a sub-agent's run (IsSubRun): its root run drives it, and re-running the root resumes it. It skips a session's journal and turn runs (IsSessionRun) too: a turn is seeded with the transcript before its message, which only the session holds, and only the session records its answer, so an unfinished turn resumes when its message is sent again (Send with the same input, or the redelivered SendOnce). resume should no-op any other runID it does not own; Recover re-drives every other incomplete run it enumerates.

func RecoverLoop added in v0.8.0 

func RecoverLoop(ctx context.Context, store Durable, resume func(ctx context.Context, runID string) error, opts ...RecoverOption) error

RecoverLoop re-drives in-flight runs until ctx is done, so a run whose holder dies is taken over automatically: Recover is a single pass, and a run another holder leases at that moment is left for a later one. Each pass, started every WithRecoverInterval (half the lease TTL by default), enumerates the store's runs as Recover does and drives each incomplete run it can lease, so a dead holder's run is picked up within about one interval of its lease expiring (the TTL after the holder's last renewal) while a pass is short. A pass costs about five store round trips for each unfinished run it lists, halted runs included, and the next pass does not start before this one has started all of its drives, so with many unfinished runs or a slow store a pass can outlast the interval, and pickup can take up to a pass's length longer.

Runs are driven concurrently, up to WithRecoverConcurrency at once (16 by default), so one long drive does not hold up the others; a run this loop is already driving is not started again. A pass that finds every slot busy waits for one, so each pass reaches every run it listed; the next pass starts when this one has started all of its drives and the interval has elapsed. Genuine failures go to the WithRecoverErrors handler, and the run is retried on the next pass; pauses and lost leases are not failures (see Recover). A run that failed because its Waker could not schedule a wake (an error wrapping ErrStorage) recorded nothing for the sleeping call, so the next pass reaches the Sleep again and schedules again.

Run it once per process, for the life of the process, with the same resume Recover takes: 

go func() {
    err := agent.RecoverLoop(ctx, store, resume,
        agent.WithLeaseHolder("worker-1"),
        agent.WithRecoverErrors(func(err error) { log.Print(err) }))
    // err is ctx's error once ctx is done
}()

A configuration error (a store that does not implement Lister, a non-positive TTL, interval or concurrency) is returned at once. Otherwise RecoverLoop returns ctx's error when ctx is done, after the drives it started (whose contexts derive from ctx) have returned.

func RedactURLs added in v0.8.0 

func RedactURLs(s string) string

RedactURLs returns s with every URL in it redacted: its userinfo, each query parameter's value, and its fragment become REDACTED, and its scheme, host, path, and query parameter names are kept. It is the redaction the agent applies to a failed tool call's text; use ToolErrorText for a tool error, and this for other error text that can quote a URL.

func RenderMermaid 

func RenderMermaid(ctx context.Context, d Durable, runID string) (string, error)

RenderMermaid returns a Mermaid flowchart of a run's journaled steps — the "graph as derived OUTPUT" (Option B): you write plain Go, and the graph is rendered from what actually ran, not hand-authored. Feed it to the dev UI, a trace viewer, or a PR.

func ResolveHalt deprecated 

func ResolveHalt(ctx context.Context, store Durable, runID, toolUseID string, result any, isError bool, opts ...ResolveOption) error

ResolveHalt clears an *OutcomeUnknown halt on the tool call toolUseID of run runID, recording result (and isError) as the call's outcome. It is ResolveHaltRef for a tool call, with the cause taken as HaltCrashed, and the same live-driver check.

Deprecated: transitional; replaced by ResolveHaltRef, which the 1.0 rewrite renames to ResolveHalt.

func ResolveHaltRef added in v0.9.0 

func ResolveHaltRef(ctx context.Context, store Durable, ref HaltRef, out Outcome, opts ...ResolveOption) error

ResolveHaltRef is the sanctioned escape from an *OutcomeUnknown halt. After a non-retriable tool call or Step halted with an unknown outcome, an operator (or a reconciler) who has verified the real side effect out of band records it as the operation's result, so a re-run proceeds past the halt instead of halting again: 

if halt, ok := errors.AsType[*agent.OutcomeUnknown](err); ok {
    // confirm out of band that the charge did go through
    err = agent.ResolveHaltRef(ctx, store, halt.Ref(), agent.Outcome{Result: "charged (operator-confirmed)"},
        agent.WithMinHaltAge(time.Minute))
    msg, err = a.Run(ctx, halt.RootRunID, input) // resumes past the halt
}

It is idempotent: the first result for an operation wins, so calling it twice or racing a concurrent driver records at most one. Resolving again with the same outcome returns nil; an operation that already has a different outcome (an earlier resolution, or the live driver's own result) is refused with *HaltAlreadyResolved, which reports the outcome that stands.

It refuses to resolve an effect a driver may still be running, whatever the halt's Cause:

  • with a store that leases runs (Leaser: MemStore, store/sqlite, store/postgres, found through a Journal and through wrappers that implement Unwrap() Store, see Capability), it takes the root run's lease for the resolution and returns *HaltInFlight while any driver holds it. Only drivers that lease the run (Lease, Recover, RecoverLoop) are seen; a plain Run holds no lease, which the claim below covers.
  • with a store that cannot (a custom store with no Leaser, or a Durable that exposes none), it requires WithMinHaltAge, so the halt is resolved only once no driver can still be running it.

On either path, it then claims the attempt after the live one, under a claim of its own, before it records the outcome, and returns *HaltInFlight if a driver holds that claim already: a process that could not record that its claim never started may void the live attempt after the check and claim the next (a plain Run, which holds no lease, included), and its effect must not be overridden. The resolution's claim is journaled as an attempt marker of the operation. If recording the outcome then fails, the claim stays live (the outcome may have been recorded all the same), so the operation halts until it is resolved again; that resolution finds the resolution's attempt live and claims the one after it (with WithMinHaltAge, once that attempt is old enough).

WithoutLiveDriverCheck skips all of these, for an operator who knows no driver is running.

A plan flow's node runs as a Step under its node key, so its halt is OpRef{Kind: OpStep, ID: "node:<name>"} (or "node:iter:<n>:<name>" inside a loop), resolved like any Step's; the recorded Result is the node's output, as JSON of the node's output type.

It refuses (ErrNoLiveAttempt, which wraps ErrConfig) an operation with no live attempt marker: one that never halted, because it was never attempted or its attempts are recorded as not started. It refuses (ErrConfig) a ref with no valid Cause or Op.Kind, and an operation that only the other kind of operation attempted. WithMinHaltAge(d) refuses (*HaltTooYoung) a halt younger than d, measured from the live attempt's marker, so a reconciler cannot query and resolve before the provider's record has settled and thereby re-fire the effect. A HaltContended halt requires WithMinHaltAge: another driver may be running the effect, and resolving it while it is young could record an outcome that driver is about to contradict.

Deciding the true outcome is a human (or reconciler) judgment the runtime cannot make for you.

func ResolveStepHalt deprecated added in v0.8.0 

func ResolveStepHalt(ctx context.Context, store Durable, runID, name string, result any, isError bool, opts ...ResolveOption) error

ResolveStepHalt clears an *OutcomeUnknown halt on the Step named name, as ResolveHalt does for a tool call, with the same live-driver check.

Deprecated: transitional; use ResolveHaltRef with OpRef{Kind: OpStep}.

func Resume deprecated 

func Resume[T any](ctx context.Context, d Durable, runID, key string, value T) error

Resume is the former name of AnswerInterrupt.

Deprecated: transitional; renamed by the 1.0 rewrite. Use AnswerInterrupt.

func RunScope 

func RunScope(ctx context.Context) string

RunScope returns the hierarchical run scope for the current tool execution (SubRunID: the parent run ID, '>', the encoded tool-use ID). Sub-agents use it to derive a stable, resumable sub-run ID.

func RunTyped 

func RunTyped[T any](ctx context.Context, a *Agent, runID, input string) (T, error)

RunTyped runs the agent to completion like Run, but returns a typed result T instead of a free-form Message. It injects a synthetic final_answer tool whose JSON schema is derived from T (via the schema package) and instructs the model to call it once, with the structured answer, when its work is done. Real tools still run first, so a tool-using agent can do work and then answer typed. The first final_answer call the tool accepts ends the run: the model is not asked for another turn.

The answer is the arguments the final_answer tool accepted: what it received, after any tool middleware (a middleware that rewrites arguments rewrites the answer, as it would for any tool). The tool decodes them strictly as T (see Func: a missing required field, an unknown or case-variant name, a duplicate name, and so on are a tool error the model corrects) and journals them as its result, {"accepted": <arguments>}. RunTyped reads that JOURNALED result, not a value captured live, so it is resume-safe: on resume, or when the run already finished, the answer is recovered from the log without re-running anything. (A run journaled before final_answer recorded its arguments has the result {}; its answer is the model's arguments, decoded with encoding/json as the tool then accepted them.)

Only if the model never makes an accepted final_answer call (it answers in plain text instead) does RunTyped parse the text of the run's final turn (the message Run returns) as T, strictly too; a text that does not decode is ErrProtocol.

T must be a type whose schema is a JSON object (a struct, a pointer to one, or a map): the answer is a tool call's arguments, which providers take only as an object. Any other T is an ErrConfig before the run starts. Because Go methods cannot add type parameters, this is a package function: agent.RunTyped[MyResult](ctx, a, id, in).

Example

ExampleRunTyped shows structured output: RunTyped returns a typed value, decoded from a schema-guided final_answer tool call (a real adapter streams the call the same way). 

type Weather struct {
	City  string `json:"city"`
	TempF int    `json:"temp_f"`
}
m := &scriptModel{turns: [][]Emit{
	toolTurn("c1", "final_answer", `{"city":"SF","temp_f":68}`),
	textTurn("done"),
}}
a := New(m, NewMemStore())

w, err := RunTyped[Weather](context.Background(), a, "run-1", "weather in SF?")
if err != nil {
	fmt.Println("error:", err)
	return
}
fmt.Printf("%s is %d°F\n", w.City, w.TempF)

Output:
SF is 68°F

func RunTypedNative 

func RunTypedNative[T any](ctx context.Context, a *Agent, runID, input string) (T, error)

RunTypedNative is like RunTyped but uses the provider's NATIVE structured-output constraint (a JSON-schema response format) instead of the injected final_answer tool: it sets Request.ResponseFormat from T's schema and decodes the model's direct JSON output. Prefer it on OpenAI-compatible providers with strict structured outputs (schema adherence is enforced provider-side, no tool round-trip). An adapter that does not support response formats (Anthropic) fails the run with ErrConfig before calling the model; use the provider-agnostic RunTyped there. Package function (Go methods can't add type parameters).

func Send deprecated 

func Send[T any](ctx context.Context, d Durable, runID, channel, key string, payload T) error

Send is the former name of Enqueue.

Deprecated: transitional; renamed by the 1.0 rewrite. Use Enqueue.

func Signal 

func Signal[T any](ctx context.Context, d Durable, runID, name string, payload T) error

Signal delivers a single-shot signal to a run, journaled at-most-once by name: a redelivery (a retried webhook, an at-least-once queue) is a no-op and the first payload wins. This turns at-least-once transport into exactly-once application to the run. Safe to call from any process; the store's primary-key / ON CONFLICT is the cross-process dedup.

Signal only records the payload. After delivering, re-invoke Run with the pause's RootRunID to resume the awaiting run: directly, or via a Waker scheduled at the current time.

func Sleep 

func Sleep(ctx context.Context, name string, d time.Duration) error

Sleep pauses the current run until d has elapsed from the FIRST time this timer was reached, durably. Call it from inside a retry-safe tool (the agent loop supplies the run context). On the first encounter it journals the wake time (now + d) and returns *TimerPending, pausing the run; on a later resume it returns nil once the wake time has passed, and continues past this point. The wake time is fixed on the first call and memoized, so a resumed or crash-recovered run waits to the same absolute instant rather than restarting the clock. Use distinct names for distinct timers. Sleep requires a retry-safe tool (Safety.ReadOnly or Idempotent), like Interrupt.

With a Waker bound (WithWaker), Sleep schedules the wake before it pauses. If the Waker fails, Sleep returns an error wrapping ErrStorage instead of pausing: a pause with no wake scheduled could sleep forever. The run then fails and its tool call records nothing (the memoized wake time aside), so re-driving the run (RecoverLoop does, on its next pass) reaches this Sleep again and schedules again.

func Step 

func Step[T any](ctx context.Context, d Durable, runID, name string, fn func(context.Context) (T, error), opts ...StepOption) (T, error)

Step runs fn as a named durable step and returns its typed result. On resume, a completed step returns its recorded result without re-running fn. This is the Option-B authoring primitive: write plain Go control flow, and name the operations that must survive a crash. 

inv, err := agent.Step(ctx, dur, runID, "fetch-invoice", fetchInvoice, agent.StepSafety(agent.Safety{ReadOnly: true}))
res, err := agent.Step(ctx, dur, runID, "reserve", reserve) // a side effect: at most once

A step runs at most once, like a tool call. By default it is treated as a side effect: an attempt marker is journaled before fn runs, so if the process dies after fn's effect and before its result is recorded, the resumed step returns *OutcomeUnknown instead of running fn again. Clear it with ResolveHaltRef (the halt's Op is OpRef{Kind: OpStep, ID: name}) once the true outcome is known. A step that is safe to re-run declares it with StepSafety (ReadOnly, Idempotent, or an IdempotencyKey); it then skips the marker and simply re-runs after a crash.

If fn returns an error, nothing is recorded but the marker: a side-effecting step whose fn failed halts on the next attempt too, since a failed call may still have taken effect.

A step that is not retry-safe must not pause: if fn returns a pause (Interrupt, Sleep, Await, a pending approval, a halt from a sub-agent it drives), Step returns an ErrConfig error instead, and the marker stays, so the step halts on the next attempt, since fn may have done something before it paused. Put the pause in a retry-safe step of its own, before or after the side effect.

T must be JSON-serializable: the result is marshaled into the journal, so a struct with unexported fields round-trips those fields to their zero values (encoding/json skips them) with no error reported. Return exported fields, a map, or a pointer whose fields are exported.

Inside the body of a plan flow node of the same run, the step is recorded under the node's key ("node:<node>:step:<name>", or "node:iter:<n>:<node>:step:<name>" in a loop body), so a loop body's step runs once per iteration, and name need be unique only within the node. Its halt names that key (OutcomeUnknown.Op.ID); resolve it with the halt's Ref.

name must not be empty, and must not start with a prefix the engine reserves for its own journal keys ("@", "run:", "tool:", "attempt:", "approval:", "signal:", and the rest; see IsReservedStepName): such a name is ErrConfig.

func SubRunID added in v0.8.0 

func SubRunID(parent, toolUseID string) string

SubRunID returns the run ID of the sub-agent run that the call toolUseID of run parent starts: parent, '>', then toolUseID encoded (see encodeID). It is RunScope inside that call.

func SubmitDecision added in v0.9.0 

func SubmitDecision(ctx context.Context, store Durable, d Decision, opts ...ApproveOption) error

SubmitDecision records d, one approver's signed decision on an m-of-n gated call. After enough decisions land, re-run the pause's RootRunID. (A single-approver gate uses Approve.)

Each distinct decision is its own journal record; resubmitting the identical decision is a no-op. The gate counts each approver's first valid decision and ignores the rest, so an invalid record never blocks a valid one. Without WithDecisionCheck, SubmitDecision does not verify: a decision that will not count is recorded and reported by the gate as ignored.

func TallyApprovals added in v0.4.0 

func TallyApprovals(recs []Record, s ApprovalSubject, p ApprovalPolicy, verifierFor ApproverVerifierFor) (ApprovalTally, []DecisionCheck)

TallyApprovals is the m-of-n counting rule, shared by the gate and offline verification so the two cannot drift. Over recs, in journal order, it considers every decision record on s.ToolUseID (IsApprovalDecision). An approver's decision is their FIRST record that is valid: the approver is in p.Approvers, verifierFor resolves their key, and the signature verifies over ApprovalDecisionBytes(s, approver, approved) under the scheme the record names (ApproverAlg), which must be the scheme of the approver's key. Records that are not valid never occupy an approver's place, so a forged or mistaken decision cannot block the approver's real one; later valid records from an approver who already decided are superseded.

A seat is a key, not an id. verifierFor is called once per approver, in policy order, and every record is checked with that one resolution. An approver whose verifier reports a key identity (ApproverVerifier.KeyIDs) that another approver's verifier also reports, or that reports none, never counts (ReasonSharedKey, ReasonNoKeyID), whichever of them signed first.

It returns the tally and a classification of every record it considered. It does not validate p (see ApprovalPolicy.ValidateKeys, which the gate runs first and which refuses a policy this rule would count short); a nil verifierFor counts nothing.

func ToolErrorText added in v0.8.0 

func ToolErrorText(ctx context.Context, tool string, err error) string

ToolErrorText returns the text an agent journals, and sends to the model, for a call to the tool named tool that failed with err: the text chosen by the agent's WithToolErrorRedactor, if ctx is a tool call's context (as a ToolMiddleware receives it) from an agent with one set, and otherwise err's own text, in either case with every URL in it redacted. A tool middleware that records a failed call's error text (a trace span, a log line) uses it to record no more than the journal holds.

func ToolResultStep added in v0.8.0 

func ToolResultStep(toolUseID string) string

ToolResultStep is the key of the result of the call toolUseID: the record the loop, a saga rollback, or ResolveHalt writes when the call's outcome is known.

func ToolsDigest added in v0.9.0 

func ToolsDigest(tools []Tool) string

ToolsDigest is the hex SHA-256 digest a model record journals of the tool set a turn was sent: each tool's name, description and argument schema, length-prefixed under a domain tag, in the order of their names, so the digest does not depend on the order the tools were listed in. It is "" for no tools.

func WaitUntil 

func WaitUntil(ctx context.Context, name string, until time.Time) error

WaitUntil pauses the current run until the absolute time `until`, durably. It is the deadline form of Sleep: same semantics, but the wake time is the given instant rather than a relative delay.

func WithClock 

func WithClock(ctx context.Context, now func() time.Time) context.Context

WithClock binds a clock to ctx for the durable timer to read "now". Deployments leave it unset (defaulting to time.Now); tests inject a controllable clock to advance time deterministically.

func WithIdentity 

func WithIdentity(ctx context.Context, id Identity) context.Context

WithIdentity binds an acting identity to ctx. The deployment calls this once at run start (for example, agent.Run(agent.WithIdentity(ctx, id), runID, input)); it then propagates to tool calls and, through the run scope, to sub-agents, so every governed action inherits the same attribution without threading it by hand.

func WithToolSafety added in v0.6.0 

func WithToolSafety(ctx context.Context, s Safety) context.Context

WithToolSafety returns ctx carrying s as the Safety that ToolSafety reports. The agent sets it for every tool call; call it yourself to run tool middleware outside an agent, as in a test. It informs middleware only: the agent's at-most-once check reads the registered tool's Safety, never this value.

func WithWaker 

func WithWaker(ctx context.Context, w Waker) context.Context

WithWaker binds a Waker to ctx so a durable Sleep registers its wake automatically. With no Waker bound, Sleep still pauses durably; the deployment is then responsible for re-invoking the run at or after the wake time on its own schedule.

Types

type APIError 

type APIError struct {
	StatusCode int
	Body       string
	Message    string // the provider's error message
	Type       string // the provider's error type (OpenAI, Anthropic) or status (Gemini)
	Code       string // the provider's error code (OpenAI)
	Err        error
}

APIError is returned by a provider adapter for a non-2xx HTTP response that is not a rate limit (which uses RateLimited), and for an exhausted quota whatever its status (wrapping ErrQuotaExhausted). It carries the StatusCode so a retry classifier can tell a transient failure (5xx, 408) from a terminal one (most 4xx: auth, validation), the provider's error fields when the body held its error object, and the Body (cut to 8KB) for diagnostics. It wraps Err (typically ErrModel) so errors.Is(err, ErrModel) holds. StatusCode is 0 for an error the provider sent partway through a stream without one.

func (*APIError) Error 

func (e *APIError) Error() string

func (*APIError) Unwrap 

func (e *APIError) Unwrap() error

type Agent 

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

Agent binds a model, a tool set, a durable store, and a middleware chain.

The WithX/Use/SetX builder methods MUTATE the receiver in place and return it for chaining; they do not copy. This is deliberate: it lets configuration be applied after construction and after the value has been shared, e.g. a.Use(trace.Model(...)) or a.Use(agent.WithRetrieval(...)) wiring a cross-cutting concern onto an agent other code already holds. The consequence is that two variables assigned from the same New(...) are aliases: reconfiguring one reconfigures both. When you need an independently configured variant, construct a fresh Agent rather than expecting a builder to fork. (The internal clone/cloneWith, used by RunTyped, do copy; they are not exported.)

An Agent is safe for concurrent Run/Stream/Session calls once configured; the builder methods are not safe to call concurrently with a run or with each other. Configure first, then run.

func New 

func New(model Model, store Durable, tools ...Tool) *Agent

New constructs an Agent. It panics if model or store is nil: both are load-bearing on every run (the model drives turns, the store journals them for at-most-once resume), so a nil is a construction-time programmer error, not a runtime condition to thread through every call. Tool names must be unique: if two tools share a name, every run fails with ErrConfig.

func (*Agent) Run 

func (a *Agent) Run(ctx context.Context, runID, input string) (Message, error)

Run drives the agent to completion for runID, resuming from the journal if steps already exist. Completed steps are reused; retry-safe tools with no recorded result are re-run; a non-retry-safe tool with no result triggers OutcomeUnknown; a tool that requires approval with no recorded decision triggers ApprovalPending. A run that already finished is final: Run returns its recorded answer without calling the model, whatever input is passed, so retrying a completed run never repeats its side effects.

The first drive of a run records its input, and a run that has not finished resumes only with that input: another input is ErrConfig, as is resuming through RunSaga a run started through Run, or the reverse (see RunStart; RecordedStart reads the recorded input back).

func (*Agent) RunResult 

func (a *Agent) RunResult(ctx context.Context, runID, input string) (*Result, error)

RunResult is the envelope-returning counterpart of Run. It drives the agent to completion and returns a *Result carrying the final message plus accumulated telemetry (usage totals, turn count, duration). Run remains unchanged; existing callers need not change.

func (*Agent) RunSaga 

func (a *Agent) RunSaga(ctx context.Context, runID, input string) (Message, error)

RunSaga runs the agent as a transaction: on success it behaves like Run; if a step fails after earlier writes succeeded, it compensates the completed writes in reverse order (recursing into sub-agent trees) and returns *SagaAborted.

The abort is derived from the journal (a durable StepSagaFail record), so a crash at any point resumes correctly: on re-entry a recorded failure sends us straight to rollback, and each compensation is a durable memoized step: once recorded it never runs again, and a crash mid-compensation re-runs it (so Compensate must be idempotent).

Note: if a non-retriable step's outcome is genuinely unknown (crashed after its attempt marker but before any result), resume returns *OutcomeUnknown instead: you can't safely auto-roll-back a step that may have committed; a human decides. A failure a human then records with ResolveHaltRef (Outcome.IsError) is a failed step: the next RunSaga rolls back.

func (*Agent) RunSagaResult 

func (a *Agent) RunSagaResult(ctx context.Context, runID, input string) (*Result, error)

RunSagaResult is the envelope-returning counterpart of RunSaga. It behaves identically to RunSaga (transactional run with reverse-order compensation on failure) but returns the richer *Result envelope on success.

func (*Agent) Session 

func (a *Agent) Session(ctx context.Context, id string) (*Session, error)

Session opens (or reopens) a multi-turn conversation with the given id, rebuilding the transcript from the store so a restarted process continues where it left off.

The id must not contain '>'. The session journals under "<id>>@session" and runs its turns under "<id>>@turn/<n>" and "<id>>@event/<encoded key>" (and their sub-agents under SubRunID(<turn run>, <call>)): everything up to the first '>' is the session id, so two sessions never share a run, and a root run ID may not contain '>', so no root run shares one with a session either.

func (*Agent) SetMaxConcurrency 

func (a *Agent) SetMaxConcurrency(n int) *Agent

SetMaxConcurrency bounds how many tool calls run in parallel within a single turn. n <= 0 means unbounded (the default). Returns the agent for chaining; use SetMaxConcurrency(1) to force fully sequential tool execution.

func (*Agent) Stream 

func (a *Agent) Stream(ctx context.Context, runID, input string) *AgentStream

Stream drives the agent like Run but returns a live AgentStream: token deltas, turn boundaries, and tool start/finish arrive as events while the durable loop runs. Resume, approval, and side-effect safety are identical to Run — Stream and Run share one loop; Run is Stream(...).Final().

Example

ExampleAgent_Stream shows the streaming API: range Events for live progress, then call Final for the terminal answer. Run is the blocking equivalent of Final. 

// A model that answers directly (a real adapter streams token deltas the same way).
m := &scriptModel{turns: [][]Emit{textTurn("hello world")}}
a := New(m, NewMemStore())

stream := a.Stream(context.Background(), "run-1", "hi")
for ev := range stream.Events() {
	switch e := ev.(type) {
	case TurnStarted:
		fmt.Println("turn started")
	case ModelEvent:
		if d, ok := e.Event.(TextDelta); ok {
			fmt.Printf("delta: %q\n", d.Text)
		}
	case Finished:
		fmt.Println("finished")
	}
}
answer, err := stream.Final()
if err != nil {
	fmt.Println("error:", err)
	return
}
fmt.Println("answer:", textOf(answer))

Output:
turn started
delta: "hello world"
finished
answer: hello world

func (*Agent) StreamSaga 

func (a *Agent) StreamSaga(ctx context.Context, runID, input string) *AgentStream

StreamSaga is the streaming counterpart of RunSaga (transactional run with reverse- order compensation on failure).

func (*Agent) Use 

func (a *Agent) Use(mw ...Middleware) *Agent

Use appends middleware wrapping the model call (first added = outermost). Returns the agent for chaining.

func (*Agent) UseTool 

func (a *Agent) UseTool(mw ...ToolMiddleware) *Agent

UseTool appends tool middleware wrapping every tool call (first added = outermost). Returns the agent for chaining. Composes with model middleware (Use) independently: Use wraps the model call, UseTool wraps tool calls.

func (*Agent) WithApproverVerifiers added in v0.4.0 

func (a *Agent) WithApproverVerifiers(fn ApproverVerifierFor) *Agent

WithApproverVerifiers configures how the m-of-n approval gate resolves an approver id to the verifier for its signature. Required whenever any tool carries a non-nil Safety.Approval: a gated call with no resolver configured fails with ErrConfig rather than silently counting zero decisions. Returns the agent for chaining.

func (*Agent) WithMaxTurns 

func (a *Agent) WithMaxTurns(n int) *Agent

WithMaxTurns caps the number of model turns a single run may take, so a model that keeps calling tools can't loop forever. n <= 0 means unbounded (the default). When the cap is reached the run returns ErrMaxTurns (category ErrBudget). Returns the agent for chaining. The cap is per run (per Session.Send turn), not per session.

func (*Agent) WithSampling 

func (a *Agent) WithSampling(opts ...SamplingOption) *Agent

WithSampling sets generation controls applied to every model call (last write wins per field). Returns the agent for chaining: New(...).WithSampling(agent.Temperature(0), agent.MaxTokens(500)).

func (*Agent) WithSystemPrompt 

func (a *Agent) WithSystemPrompt(s string) *Agent

WithSystemPrompt sets a system message that is prepended to the conversation on every model turn. The message is re-seeded on each Run (including resumes), so it is always present regardless of journal replay. Returns the agent for chaining: New(...).WithSystemPrompt("you are a concise assistant").

The system message is configuration, not journal: it is not recorded, and the model turns a drive makes are sent the message the agent holds at that drive. A run resumed after the prompt was changed sends its remaining turns the new prompt beside turns that answered the old one. Its recorded turns are not affected, and neither is any decision the loop makes on resume.

func (*Agent) WithSystemPromptFunc 

func (a *Agent) WithSystemPromptFunc(fn func(context.Context) string) *Agent

WithSystemPromptFunc sets a system message computed per run, so it can inject dynamic context (current date, tenant, retrieved state) each turn. It takes precedence over WithSystemPrompt. Returns the agent for chaining.

fn is called once per drive of a run (each Run, Stream, or resume), not once per run, and its result is not journaled (see WithSystemPrompt): a run resumed later is sent what fn returns then. Context that the run's later turns must see unchanged belongs in the input, which is journaled (see RunStart), or in a tool result.

func (*Agent) WithTokenBudget added in v0.6.0 

func (a *Agent) WithTokenBudget(max int) *Agent

WithTokenBudget caps the tokens a single run may use: once the run's model calls have used max tokens or more (Usage.TotalTokens, cached input included), the run makes no further model call and returns ErrBudgetExceeded (category ErrBudget). A call's usage is known only after it returns, so the call that crosses max completes; the next one is refused. max <= 0 means unbounded (the default). Returns the agent for chaining.

The budget counts every model request the run sent, not only the responses it recorded: a failed attempt a middleware retried, a losing hedge target, and a model call that failed for good all count (see Result.Spend). It is durable: each turn's usage and discarded spend are journaled with it, a failed call's spend is journaled as its own record, and the count is rebuilt from the journal, so a resumed run, on any process, is held to what it has already used. A request still running when the run returns (a hedge loser whose model ignores cancellation) is not counted. Like WithMaxTurns it applies per run (per Session.Send turn).

The budget covers the run's agent tree: every run started from its tool calls (sub-agents, and theirs) counts against it, and each of those runs checks it, with its own budget if it has one, before each model call. A resumed tree first counts the journals of its sub-agents cut off mid-run, so it is held to everything it used before any of it calls the model again. Overshoot: a check sees every call that has returned, so the tree passes max only by calls in flight when it reached max. Each agent run has at most one model call in flight, so with k runs of the tree calling the model at that moment (k parallel sub-agents; 1 with none) the tree uses less than max plus k calls' usage.

func (*Agent) WithToolChoice 

func (a *Agent) WithToolChoice(tc ToolChoice) *Agent

WithToolChoice sets the tool-choice control applied to every model call (see ToolChoice for the modes). Returns the agent for chaining.

CAVEAT: forcing Mode "required" or "tool" on every turn of the multi-turn loop keeps the model from ever producing a final answer (it must always call a tool), so the run cannot terminate normally. Reserve those modes for single-turn or typed/structured calls; "auto" (the default when unset) is the norm for the agent loop.

func (*Agent) WithToolErrorRedactor added in v0.7.0 

func (a *Agent) WithToolErrorRedactor(fn func(tool string, err error) string) *Agent

WithToolErrorRedactor sets the text recorded for a tool call that fails. A failed call's error text is journaled as the call's result, sent to the model, and hashed into the audit trail, where a proof can disclose it, so it must not carry secrets. By default the text is the error's own, with every URL in it redacted: its userinfo, each query parameter's value, and its fragment become REDACTED (a net/http *url.Error quotes the whole request URL, API key parameter included). fn receives the tool's name and its error and returns the text to record instead; the same URL redaction then applies to what fn returns. Use it to scrub what URL redaction cannot know about (an account number, a token in a header echoed back), and keep the text useful: the model reads it to decide what to do next.

It applies to every failed call, a sub-agent's included (a sub-agent's failure is its tool call's error), and to the failure a saga journals. The error the caller gets back (SagaAborted.Cause, for one) is the tool's own. Returns the agent for chaining.

type AgentEvent 

type AgentEvent interface {
	// contains filtered or unexported methods
}

AgentEvent is a lifecycle event emitted by Agent.Stream as the run loop advances: turn boundaries, live model token deltas, and tool start/finish. It is the SEMANTIC-layer companion to the model's byte-level Event — a UI ranges over these to render progress while the durable loop runs underneath.

The terminal answer and any error are NOT events: they come from AgentStream.Final (mirroring how the model Stream yields Events but returns the assembled Message separately). A run that pauses for approval or halts on an unsafe resume emits the relevant lifecycle event and then surfaces the *ApprovalPending / *OutcomeUnknown via Final, exactly as Run returns it.

func ProjectEvents added in v0.7.0 

func ProjectEvents(recs []Record) (events []AgentEvent, sources []int)

ProjectEvents returns the events ReplayEvents returns for a journal holding recs, and for each event the index in recs of the record it projects: each event comes from exactly one record. The audit package uses the index to salt a projected event from its record's salt.

func ReplayEvents 

func ReplayEvents(ctx context.Context, store Durable, runID string) ([]AgentEvent, error)

ReplayEvents returns the semantic lifecycle events implied by runID's DURABLE journal: the same AssistantTurn and ToolCompleted events Agent.Stream re-emits when it resumes from that journal, in persisted order. Because the sequence is a pure function of the recorded steps, it — and any commitment built over it (see audit.EventLogFromJournal) — is identical before and after a crash, which is what makes it a resume-stable audit artifact.

It reconstructs from the journal alone, without re-running the model or tools. Only journaled facts are reproduced: assembled assistant turns (StepModel) and completed tool calls with their results (StepToolResult). Live-loop-only signals — token-level ModelEvent deltas, TurnStarted, TurnRestarted, ToolStarted, and the terminal Finished — are not journaled and so are not part of the durable projection; the durable content is the turns and tool results.

type AgentStream 

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

AgentStream is a live view of a running agent: range Events for progress, then call Final for the terminal answer (or error). It is the streaming counterpart of Run, the same way the model Stream is the counterpart of Generate — Run is literally Stream(...).Final().

A Stream is consumed once. Either range Events fully (Final then returns immediately) or call Final directly (it drains and discards events); calling Final after breaking out of Events early drains whatever remains. Cancel ctx to abandon a run without draining — pending emits then unblock on ctx.Done rather than leaking.

func (*AgentStream) Events 

func (as *AgentStream) Events() iter.Seq[AgentEvent]

Events returns a range-over-func iterator over lifecycle events until the run ends. Per-event errors do not occur here (a mid-stream model error aborts the run and is reported by Final); the iterator simply ends when the run stops producing events.

func (*AgentStream) Final 

func (as *AgentStream) Final() (Message, error)

Final drains any un-consumed events and returns the run's terminal message and error (including *ApprovalPending / *OutcomeUnknown, matching Run). Safe to call after fully or partially ranging Events, or on its own.

type Alg added in v0.9.0 

type Alg string

Alg names a signature scheme: "ed25519", "ml-dsa-65", or "ed25519+ml-dsa-65". The audit package defines the schemes (audit.AlgEd25519, audit.AlgMLDSA65, audit.AlgHybrid) and their signers and verifiers; the agent package only journals the name, with each approver decision, so a decision is checked under the scheme it was signed with.

type ApprovalPending added in v0.9.0 

type ApprovalPending struct {
	RunRef
	ToolUseID string
	ToolName  string
	Args      json.RawMessage
	// Quorum is non-nil for an m-of-n gate: the running tally at the pause. Record
	// decisions with SubmitDecision, then re-run.
	Quorum *ApprovalTally
}

ApprovalPending is returned by Agent.Run when a tool requiring human approval has no recorded decision yet. The run has paused durably; record a decision (Approve, or SubmitDecision for an m-of-n gate) against RunID, then re-run RootRunID to resume.

func (*ApprovalPending) Error added in v0.9.0 

func (e *ApprovalPending) Error() string

Error names the run and the call awaiting approval.

func (*ApprovalPending) Subject added in v0.9.0 

func (e *ApprovalPending) Subject() ApprovalSubject

Subject returns the approval subject of the paused call, which is what approvers sign: 

sig := signer.Sign(agent.ApprovalDecisionBytes(pend.Subject(), "finance", true))

For a gate inside a sub-agent, RunID is the sub-run's id.

type ApprovalPolicy added in v0.4.0 

type ApprovalPolicy struct {
	Need      int      // decisions required to proceed (k), 1 <= Need <= len(Approvers)
	Approvers []string // eligible approver ids; the bounded set (n)
}

ApprovalPolicy declares a k-of-n human gate. Need is k; the eligible approvers are the bounded set (n = len(Approvers)). A tool with a non-nil Approval requires approval whether or not RequiresApproval is also set.

func (ApprovalPolicy) Validate added in v0.4.0 

func (p ApprovalPolicy) Validate() error

Validate reports whether the policy is well formed: at least one approver, every approver id non-empty valid UTF-8, no two ids the same approver, and 1 <= Need <= len(Approvers). A policy that fails is a configuration error; the gate refuses it with ErrConfig rather than guessing.

Approver ids are compared as exact bytes everywhere else (eligibility, the signed decision bytes, the verifier lookup). A policy may still not list two ids that differ only by case or by Unicode normalization ("alice" and "Alice", an NFC and an NFD "café", a fullwidth and an ASCII spelling): a person reading the policy, or a key lookup that folds case, takes them for one approver, who could then fill two seats. Two ids are the same approver when their approverFoldKey is equal, and such a policy is refused as ambiguous.

func (ApprovalPolicy) ValidateKeys added in v0.9.0 

func (p ApprovalPolicy) ValidateKeys(verifierFor ApproverVerifierFor) error

ValidateKeys reports whether the policy is well formed (Validate) and holds one seat per signing key under verifierFor: it resolves every approver once and refuses, with ErrConfig, a nil verifierFor, an approver whose verifier reports no key identity (an empty KeyIDs, or an empty entry), and two approvers whose verifiers report a common key identity. An approver the resolver does not know is not refused: its decisions cannot verify, so it fills no seat.

The gate runs this check on every evaluation, before it reads a recorded tally, so a resolver that changes between runs is checked again; TallyApprovals also never counts an approver whose key is shared, whatever resolver it is given. The check guards the tallies this version counts and records. A terminal tally already in the journal stays authoritative and is not recounted, so one recorded before this check existed is reused as recorded, even if it counted two approvers on one key. A caller that builds policies and resolvers ahead of a run can call ValidateKeys to refuse a bad pairing before any call pauses.

type ApprovalRequired 

type ApprovalRequired struct {
	ToolUseID string          `json:"tool_use_id"`
	Name      string          `json:"name"`
	Args      json.RawMessage `json:"args"`
	// Quorum is the running tally when the tool has an m-of-n Safety.Approval policy, so a
	// streaming UI can show progress ("1 of 2 approved") without waiting for Final. Nil for
	// a 1-of-1 gate.
	Quorum *ApprovalTally `json:"quorum,omitempty"`
}

ApprovalRequired fires immediately before the run pauses for a human decision on a tool that requires approval. The run then returns *ApprovalPending from Final; record a decision (Approve, or SubmitDecision for an m-of-n gate) and re-invoke to continue.

type ApprovalSubject added in v0.4.0 

type ApprovalSubject struct {
	RunID     string
	ToolUseID string
	ToolName  string
	Args      json.RawMessage
}

ApprovalSubject is exactly what an approver decides on: one tool call, with its name and arguments, in one run. The decision bytes bind all of it, so a signature approves this call and no other: not another run, not another call, and not the same call id with different arguments.

type ApprovalTally added in v0.4.0 

type ApprovalTally struct {
	Need       int      `json:"need"`                  // the policy's k
	Approvers  []string `json:"approvers"`             // the eligible set the gate enforced, in policy order
	Approved   int      `json:"approved"`              // approvers whose counted decision is an approval
	Denied     int      `json:"denied"`                // approvers whose counted decision is a denial
	ApprovedBy []string `json:"approved_by,omitempty"` // those approvers, in journal order
	DeniedBy   []string `json:"denied_by,omitempty"`   // those approvers, in journal order
	Pending    []string `json:"pending,omitempty"`     // eligible approvers with no valid decision yet who can still make one, in policy order
	// Excluded names the eligible approvers who fill no seat, in policy order: their verifier
	// shares a key identity with another approver's, reports none, or is a typed nil
	// (ReasonSharedKey, ReasonNoKeyID). They never count, so Unreachable leaves them out. The gate
	// refuses such a policy before counting, so a tally it journals has none.
	Excluded []string `json:"excluded,omitempty"`
	// Records names every decision record on this call the gate read, valid or not, in
	// journal order. Evidence must disclose all of them, so an omitted decision is detectable.
	Records []string `json:"records,omitempty"`
}

ApprovalTally is the m-of-n gate's count for one call. At a pause it is the running tally (ApprovalPending.Quorum, ApprovalRequired.Quorum). At a terminal outcome the gate journals it under ApprovalTallyStep, before the tool runs, as the record of what it enforced and what it read.

func (ApprovalTally) Passed added in v0.4.0 

func (t ApprovalTally) Passed() bool

Passed reports whether Need approvals are in.

func (ApprovalTally) Unreachable added in v0.4.0 

func (t ApprovalTally) Unreachable() bool

Unreachable reports whether Need can no longer be reached: fewer approvers remain who have not denied, and are not Excluded, than approvals are required. Only valid denials count toward this, so an invalid record cannot force a denial.

type ApproveOption added in v0.4.0 

type ApproveOption func(*approveOptions)

ApproveOption configures SubmitDecision.

func WithDecisionCheck added in v0.4.0 

func WithDecisionCheck(verifierFor ApproverVerifierFor) ApproveOption

WithDecisionCheck makes SubmitDecision check the decision before recording it, so an approver learns at submission time that it will not count: it reads the run, finds the call, and verifies the signature against that exact call with verifierFor. It returns ErrInvalidApproval (no such call, unknown approver, or a signature that does not verify) or ErrAlreadyDecided (the approver's earlier valid decision already counts), and records nothing. The check is for feedback; the gate re-verifies every record itself, so correctness never depends on it. Eligibility is the policy's, which SubmitDecision does not see.

type ApproverVerifier added in v0.4.0 

type ApproverVerifier interface {
	// Alg names the verifier's scheme. A decision counts only if it was journaled under this scheme.
	Alg() Alg
	// Verify reports whether sig is a valid signature over message.
	Verify(message, sig []byte) bool
	// KeyIDs identifies the signing keys behind Verify: one entry per public key whose private
	// key takes part in producing a signature Verify accepts. A single-key verifier reports one
	// entry; a verifier that accepts a signature by any of several keys (a rotation window)
	// reports one per key; a verifier that requires several signatures (audit.HybridVerifier)
	// reports one per component key.
	//
	// An entry identifies a key, not an approver: it is derived from the public key's bytes,
	// never from a configured name, so two verifiers over one key report the same entry however
	// they were built. The audit verifiers report the scheme and the hex SHA-256 of the public
	// key's encoding ("ed25519:3b6a27bc..."). Entries are compared as exact strings.
	//
	// The gate counts one seat per key: it refuses a policy two of whose approvers report a
	// common entry, or whose approver's verifier reports no entry or an empty one, with
	// ErrConfig (see ApprovalPolicy.ValidateKeys). Otherwise the holder of a shared key could
	// sign as each approver that key serves and meet a quorum alone. Enrolling one approver's
	// public key under a second approver makes the gate refuse the whole policy: it fails closed,
	// and no call under it runs until the keys are corrected.
	//
	// Trust boundary: a verifier is trusted code, like the resolver that returns it. The gate
	// takes its KeyIDs on faith; it cannot check that they name the keys Verify really accepts.
	// A verifier that under-reports its keys, or reports an identity that is not derived from a
	// key, defeats the check. Use the audit verifiers, or derive identities the same way.
	KeyIDs() []string
}

ApproverVerifier checks an approver's signature over the decision bytes under one scheme and names the signing keys behind it. audit.Ed25519Verifier, audit.MLDSAVerifier and audit.HybridVerifier implement it, so a deployment reuses the existing verifiers with no new cryptography and no agent->audit import.

type ApproverVerifierFor added in v0.4.0 

type ApproverVerifierFor func(approverID string) (ApproverVerifier, bool)

ApproverVerifierFor resolves the verifier for an approver id (the caller's PKI, mirroring audit's issuer lookup). It reports ok=false for an unknown approver; the counting rule treats that approver's decisions as unverifiable.

type AssistantTurn 

type AssistantTurn struct {
	Message  Message `json:"message"`
	Replayed bool    `json:"replayed"`
}

AssistantTurn is the fully-assembled assistant message for a turn. Replayed is true when it was reconstructed from the journal on resume rather than produced by a live model call (in which case no ModelEvent deltas preceded it).

type Awaiting deprecated 

type Awaiting = SignalPending

Awaiting is the former name of SignalPending.

Deprecated: transitional; renamed by the 1.0 rewrite. Use SignalPending.

type Compensator 

type Compensator interface {
	// Compensate undoes a completed call. args are the arguments the tool accepted (after tool
	// middleware; see CompensatedFunc); result
	// is what Call returned. Must be idempotent — on a crash mid-rollback it may re-run.
	Compensate(ctx context.Context, args, result json.RawMessage) error
}

Compensator is an optional interface a Tool implements to declare how to UNDO its side effect. In a saga run (RunSaga), if a step fails after earlier writes succeeded, the completed compensatable writes are rolled back in reverse order — automatically, and recursively through sub-agent trees.

This is the transactional / saga agent: "charged the card and booked the flight, then failed on the hotel → cleanly refund + cancel." It's the gsm compensation idea applied to the agent runtime — the journal records exactly which writes completed, so well-founded reverse compensation falls out of the durable substrate.

A saga step must be ATOMIC: it must not leave a partial external side effect before returning an error, because the failing step itself is not compensated (there's no recorded result to drive Compensate). Make forward steps all-or-nothing, or idempotent.

type Decision added in v0.9.0 

type Decision struct {
	RunID      string // the run whose journal holds the call: the pause's RunID
	ToolUseID  string
	ApproverID string
	Approved   bool
	Alg        Alg // the signature scheme, e.g. audit.AlgEd25519
	Signature  []byte
}

Decision is one named approver's signed decision on a tool call gated by an m-of-n Approval policy. Signature is the approver's signature over ApprovalDecisionBytes(pend.Subject(), ApproverID, Approved), where pend is the run's *ApprovalPending, under the scheme Alg names (the signer's Alg). Alg is journaled with the decision as Record.ApproverAlg, and the gate counts the decision only under a verifier of that scheme.

type DecisionCheck added in v0.4.0 

type DecisionCheck struct {
	Step     string // the record's journal name
	Approver string
	Approved bool
	Counted  bool   // this is the approver's decision: their first valid one
	Reason   string // why it did not count; empty when Counted
}

DecisionCheck is how the counting rule classified one decision record.

type Describer added in v0.8.0 

type Describer interface {
	Describe() ModelInfo
}

Describer is implemented by a Model that can say what it is. It is optional: the agent never requires it, and ModelInfoOf reports false for a Model that does not implement it.

type Doc 

type Doc struct {
	ID       string         `json:"id,omitempty"`
	Text     string         `json:"text"`
	Score    float64        `json:"score,omitempty"`
	Metadata map[string]any `json:"metadata,omitempty"`
}

Doc is a document returned by a Retriever: its text plus optional id, similarity score, and metadata. This is the neutral shape the retrieval helpers speak; your Retriever maps your store's results onto it. A NaN or infinite Score (cosine similarity against a zero vector is 0/0) has no JSON encoding, so the retrieval helpers drop it (report 0).

type Durable deprecated 

type Durable interface {
	// Do returns the recorded Record for (runID, name) without running fn if present; otherwise
	// runs fn, records the returned Record (with Name set and a fresh salt), and returns the
	// record the journal holds. If fn errors, nothing is recorded and the step runs again on the
	// next attempt. Once fn has returned a record, Do records it even if ctx was cancelled
	// meanwhile: fn may have fired a side effect, and its outcome must not be lost.
	Do(ctx context.Context, runID, name string, fn func(context.Context) (Record, error)) (Record, error)
	// History returns all recorded steps for a run, in order, the journal header first.
	History(ctx context.Context, runID string) ([]Record, error)
}

Durable is the step-memoization interface the engine and its callers use during the transition to Journal: Do runs a step at most once per (runID, name), and History reads a run back. *Journal implements it, and so do MemStore and the SQL stores, whose Do and History go through a Journal over themselves.

Deprecated: transitional; renamed by the 1.0 rewrite. Use *Journal.

A Durable that is not a Journal must keep the Journal's guarantees, which the engine's accounting relies on as much as its at-most-once does: it records a step's result at most once under one name (a second record under the name is never written, and Do returns the one the journal holds); it calls fn at most once per record it writes, never again for a name already recorded; and it keeps each record's bytes, the salt among them, exactly as JournalEntry built them (a salt a step's record carries from the engine is the one the journal must hold).

A Durable that wraps another (such as audit.AuditedStore) may implement 

Unwrap() Durable

so the engine reaches the store beneath it: its capabilities (Lister, Leaser, as Capability finds them), and the identity under which the process keeps what it keeps per run (claims it could not record as not started, spend it could not journal), which every Journal and wrapper over that store then shares. It may do so only if it passes run IDs and step names through unchanged, as for Unwrap() Store. A wrapper that rewrites keys (a tenant prefix, say) must not implement Unwrap: its runs would be kept under the run IDs it was given, and two tenants' runs of one name would share them. storetest.CheckDurableWrapper checks this.

type Emit 

type Emit struct {
	Event Event
	Err   error
}

Emit is what an adapter pushes onto its event channel (event or terminal error).

type Entry added in v0.9.0 

type Entry struct {
	// Seq is the entry's position: opaque, strictly increasing in commit order within the run
	// (see Store). It is not a count.
	Seq  int64
	Name string
	Data []byte
}

Entry is one stored entry of a run: its position, name and bytes.

type Event 

type Event interface {
	// contains filtered or unexported methods
}

Event is a normalized streamed model event. The stream/ package maps every provider wire format (OpenAI SSE / Anthropic typed / Gemini NDJSON / Bedrock binary) onto these.

type Finish 

type Finish struct {
	Reason    FinishReason `json:"reason"`
	Raw       string       `json:"raw"` // the provider's own finish reason, as it sent it ("end_turn", "STOP"); empty if it sent none
	Usage     Usage        `json:"usage"`
	Discarded Usage        `json:"discarded"`
	// contains filtered or unexported fields
}

Finish ends a model turn. Reason says why the turn ended, in the neutral vocabulary below; an adapter maps its provider's reasons onto it where they enter, and keeps the provider's own value in Raw. A first-party adapter always sets Reason: a provider reason it does not know is passed through unchanged (Reason is then Raw), and a turn the provider ended without naming a reason is FinishStop with an empty Raw. A Model that does not know the reason may leave Reason empty, which counts as a natural stop.

The reason decides whether the turn is the model's answer. Stream.Message (and so every agent run) returns ErrOutputTruncated for FinishLength and ErrOutputFiltered for FinishFiltered rather than a message, since what arrived is only part of what the model would have said, and a run that recorded it would end with that part as its final answer. Any other reason is ErrStreamProtocol: the core does not guess what a provider's own word means.

The reason never decides whether tools run: the calls the turn carries do. A turn with calls runs them whatever its reason says (OpenAI reports stop under a forced tool_choice), and a FinishToolUse turn with no call is ErrStreamProtocol, since the calls it was for were lost.

Usage is what the turn's own response cost. Discarded is usage billed for responses the turn threw away (failed attempts a middleware retried, losing hedge targets); a live adapter leaves it zero, and a Model that replays a recorded turn reports the discarded spend it recorded there.

type FinishReason added in v0.8.0 

type FinishReason string

FinishReason is why a model turn ended, in a neutral vocabulary every adapter maps its provider's reasons onto (see Finish). The set is closed: the core accepts only the constants below and the empty reason. 

const (
	FinishStop     FinishReason = "stop"     // the model ended its answer (end of turn, a stop sequence)
	FinishToolUse  FinishReason = "tool_use" // the model ended its turn to call tools
	FinishLength   FinishReason = "length"   // the output hit its token limit (or the context window) and was cut off
	FinishFiltered FinishReason = "filtered" // a safety or content filter, or a refusal, stopped the output
)

The neutral finish reasons (see Finish).

type Finished 

type Finished struct {
	Final Message `json:"final"`
}

Finished carries the terminal assistant answer — the same Message that Run returns and that Final reports. The event stream closes after this.

type FlowRef added in v0.9.0 

type FlowRef struct {
	Name string `json:"name"`
}

FlowRef names the plan flow that drives a run of kind RunKindFlow.

type HaltAlreadyResolved added in v0.9.0 

type HaltAlreadyResolved struct {
	RunID   string
	Op      OpRef
	Result  json.RawMessage // the recorded result
	IsError bool            // whether the recorded outcome is a failure
}

HaltAlreadyResolved is returned by ResolveHaltRef when the operation already has a recorded outcome that differs from the one given: an earlier resolution, or the result the live driver recorded. The recorded outcome stands; Result and IsError report it. Resolving again with the same outcome is not an error. It wraps ErrAlreadyResolved.

func (*HaltAlreadyResolved) Error added in v0.9.0 

func (e *HaltAlreadyResolved) Error() string

Error names the operation and the outcome that stands.

func (*HaltAlreadyResolved) Unwrap added in v0.9.0 

func (e *HaltAlreadyResolved) Unwrap() error

Unwrap returns ErrAlreadyResolved.

type HaltCause added in v0.9.0 

type HaltCause string

HaltCause says why a run halted on an operation whose outcome is unknown. 

const (
	// HaltCrashed: the operation's attempt marker is recorded with no result, and the halting
	// driver knows of no live claimant. That is what the halting driver saw, not proof: a
	// driver of the same run may still be running the effect (a Step that lost its claim with no
	// call of it in flight in this process, or a drive that read the journal after the claim,
	// cannot tell). ResolveHaltRef therefore checks for a live driver itself, whatever the cause.
	HaltCrashed HaltCause = "crashed"
	// HaltContended: another driver of the same run won the claim on the operation while this
	// one was running: a tool call's claim lost after this drive read no marker for it, or a
	// Step's claim lost to a call of the step in flight in this process. That driver owns the
	// effect and may be running it now.
	HaltContended HaltCause = "contended"
)

type HaltInFlight added in v0.9.0 

type HaltInFlight struct {
	RunID     string // the run whose journal holds the operation
	RootRunID string // the run whose lease is held (the halted run's root)
	Op        OpRef
	// Attempt is the attempt marker key a driver holds, when that is why the resolution was
	// refused; empty when the lease is held.
	Attempt string
}

HaltInFlight is returned by ResolveHaltRef when a driver may be running the operation's effect right now, and its own result must win over a resolution: a driver holds the lease on the halted run's root, or (see ResolveHaltRef) a driver holds the claim on the attempt after the live one, having found the live one recorded as never started. Retry once the driver has finished (its lease released or expired, or its result recorded); a retry reads the run again.

func (*HaltInFlight) Error added in v0.9.0 

func (e *HaltInFlight) Error() string

Error names the operation and what a driver holds.

type HaltRef added in v0.9.0 

type HaltRef struct {
	RunID string
	Op    OpRef
	Cause HaltCause
}

HaltRef is what ResolveHaltRef needs to clear a halt: the run whose journal holds the operation, the operation, and why it halted. Take it from the halt with OutcomeUnknown.Ref.

Cause matters: a HaltContended halt may still be running in another driver, and ResolveHaltRef resolves it only once it is older than WithMinHaltAge. An operator building a HaltRef by hand from a journal listing must state the cause.

type HaltTooYoung added in v0.3.0 

type HaltTooYoung struct {
	RunID     string
	ToolUseID string        // the operation's ID: a tool-use ID, or a step name
	Age       time.Duration // elapsed since the effect was attempted
	Min       time.Duration // the required minimum
}

HaltTooYoung is returned by ResolveHaltRef when WithMinHaltAge is set and the halt has not aged past the grace period yet. Wait and retry the resolution later.

func (*HaltTooYoung) Error added in v0.3.0 

func (e *HaltTooYoung) Error() string

type Identity 

type Identity struct {
	// Actor is the acting agent instance: a deployment plus model version, e.g. "exec-agent@1.4.2".
	Actor string `json:"actor,omitempty"`
	// OnBehalfOf is the principal the agent acts for: a desk, an account, a user.
	OnBehalfOf string `json:"on_behalf_of,omitempty"`
	// AuthorityRef identifies the grant that authorizes this action: the digest or id of a signed
	// capability/authority record (see audit anchoring), so the leaf links to what permitted it.
	AuthorityRef string `json:"authority_ref,omitempty"`
}

Identity is the accountability binding for a run: which agent instance acted, on whose behalf, and under what authority. It is assigned by the DEPLOYMENT, not by the model: the operator sets it at run start from its own auth layer (an IdP, a signed capability grant, a service identity), and it rides the context so every governed action and every sub-agent call inherits it. The SDK anchors and proves the identity CLAIM (these fields, committed in the audit log); it does not authenticate the principal, which is the operator's IdP/PKI, and the attribution is only as strong as custody of whatever key signs the run's tree heads and grants.

The model has no inherent identity. "The agent's identity" is the identity the deployment binds to this run, so a compromised or misconfigured deployment attributes to whatever it claims. That boundary is deliberate and matches the model/governance split: prove the record and the granted authority, do not overclaim the intent behind them.

func IdentityFrom 

func IdentityFrom(ctx context.Context) (Identity, bool)

IdentityFrom returns the acting identity bound to ctx, and whether one was set.

func (Identity) Empty 

func (i Identity) Empty() bool

Empty reports whether the identity carries no attribution at all.

type Image 

type Image struct {
	Mime string `json:"mime,omitempty"` // MIME type for Data (required when Data is set), e.g. "image/jpeg"; ignored for URL
	Data []byte `json:"data,omitempty"` // raw image bytes; adapters base64-encode. Mutually exclusive with URL.
	URL  string `json:"url,omitempty"`  // hosted image URL. Mutually exclusive with Data.
}

Image is a provider-neutral image input. Exactly one of Data or URL is set (Data XOR URL). For raw bytes, set Data and Mime (e.g. "image/png"); the adapter base64-encodes it (Anthropic base64 source / OpenAI data URI). For a hosted image, set URL and leave Data nil; Mime is not required for URLs. This is an input-only part: models emit text, reasoning, and tool calls, never images, so nothing produces an Image on the response path.

func ImageData 

func ImageData(mime string, data []byte) Image

ImageData builds an Image part from raw bytes with the given MIME type; the adapter base64-encodes the bytes onto the wire.

func ImageURL 

func ImageURL(url string) Image

ImageURL builds an Image part referencing a hosted image by URL.

type InterruptPending added in v0.9.0 

type InterruptPending struct {
	RunRef
	Name   string
	Prompt any // caller-defined payload for the human: a question, options, current state
}

InterruptPending is returned by Run when a tool called Interrupt and no answer has been recorded for that name yet. The run has paused durably at the interrupt point. Inspect Prompt to decide what to ask the human, record an answer with AnswerInterrupt (same RunID and Name), then re-invoke Run with RootRunID to continue.

func (*InterruptPending) Error added in v0.9.0 

func (e *InterruptPending) Error() string

Error names the run and the interrupt point.

type Interrupted deprecated 

type Interrupted = InterruptPending

Interrupted is the former name of InterruptPending.

Deprecated: transitional; renamed by the 1.0 rewrite. Use InterruptPending.

type Journal added in v0.9.0 

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

Journal is the durable journal the engine writes through: it owns everything a run's records mean on top of a Store. It encodes each record (EncodeRecord) with a fresh salt, memoizes named steps, writes and checks the journal format header, runs the exclusive attempt claims that keep side effects at most once, and records a step's outcome even when its caller's context is cancelled. A Journal is safe for concurrent use, and any number of Journals may share a store, in one process or many: the store's single-winner Insert (A1) decides every race, and Journals in one process over the same store also share in-flight steps.

The header. Before its first write to a run a Journal inserts the header, a StepHeader record named "@journal" that names JournalFormat, and requires it to be the run's first entry. Every read checks that the first entry is a header naming a supported format; a run in another format, or with no header first, is refused with a *JournalVersionError before anything is read from it or written to it. A Journal remembers the runs it has checked (the 65536 most recently used), so a run is checked once per process, not on every step. Run IDs must never be reused after a run is deleted from its store: a Journal that remembers the deleted run writes into the new one with no header, which every reader refuses.

func NewJournal added in v0.9.0 

func NewJournal(s Store, opts ...JournalOption) (*Journal, error)

NewJournal returns a Journal over s. A nil s is ErrConfig.

func (*Journal) Do deprecated added in v0.9.0 

func (j *Journal) Do(ctx context.Context, runID, name string, fn func(context.Context) (Record, error)) (Record, error)

Do runs fn as the named step name of runID, at most once: if the step is recorded, Do returns the recorded record without calling fn; otherwise it calls fn and records the record fn returns (with Name set and a fresh salt), and returns the record the journal holds, which is another caller's if one recorded the step first. If fn errors, nothing is recorded. Once fn has returned a record, Do records it even if ctx was cancelled meanwhile, since fn may have fired a side effect whose outcome must not be lost. Callers of the same step in this process, through any Journal over the same store, share one call of fn.

Deprecated: transitional; the 1.0 rewrite unexports it. Engine code writes through the journal's own paths; audit and plan reach it through internal/journalhook.

func (*Journal) Format added in v0.9.0 

func (j *Journal) Format(ctx context.Context, runID string) (string, error)

Format returns the journal format runID's header names, "" for a run with no entries, and a *JournalVersionError for a run whose first entry is not a header. It reports a format this version does not support rather than refusing it.

func (*Journal) Get added in v0.9.0 

func (j *Journal) Get(ctx context.Context, runID, name string) (Record, bool, error)

Get returns the record named name in runID's journal, if there is one. It reads the record first and then checks the run's header, so a read that races the run's first write never takes a record for one without a header: the header commits before any record. A run this version cannot read is a *JournalVersionError whether or not it holds the name; a run with no entries holds nothing.

func (*Journal) History added in v0.9.0 

func (j *Journal) History(ctx context.Context, runID string) ([]Record, error)

History returns runID's records in journal order, the header first (see Records). A run with no entries has an empty history.

func (*Journal) Records added in v0.9.0 

func (j *Journal) Records(ctx context.Context, runID string) iter.Seq2[Record, error]

Records yields runID's records in journal order, the header first. It checks the header as it streams: a run whose first entry is not a header naming a supported format yields a *JournalVersionError and nothing else.

func (*Journal) Store added in v0.9.0 

func (j *Journal) Store() Store

Store returns the store the journal writes to.

type JournalOption added in v0.9.0 

type JournalOption interface {
	// contains filtered or unexported methods
}

JournalOption configures NewJournal.

type JournalVersionError added in v0.9.0 

type JournalVersionError struct {
	RunID     string
	Found     string
	Supported []string
}

JournalVersionError is returned by a Journal asked to read or write a run whose journal format it does not support (see JournalFormat). Found is the format the run's header names, or "" for a journal with no header first (an unversioned journal). Supported lists the formats this version reads and writes. It wraps ErrJournalVersion. A journal refuses such a run before it writes to it, so an older binary never adds records to a run a newer one started.

func (*JournalVersionError) Error added in v0.9.0 

func (e *JournalVersionError) Error() string

Error describes the run, the format found, and the formats supported.

func (*JournalVersionError) Unwrap added in v0.9.0 

func (e *JournalVersionError) Unwrap() error

Unwrap returns ErrJournalVersion.

type Leaser 

type Leaser interface {
	// AcquireLease claims runID for holder until now+ttl. It returns true if granted (the run is
	// unleased or the live lease is already holder's, which renews it), false if another holder
	// currently holds a live lease. An expired lease is available to any holder.
	AcquireLease(ctx context.Context, runID, holder string, ttl time.Duration) (bool, error)
	// RenewLease extends holder's lease on runID, returning false if holder no longer holds it (it
	// expired or was taken). A driver whose work outlasts ttl renews to keep the lease.
	RenewLease(ctx context.Context, runID, holder string, ttl time.Duration) (bool, error)
	// ReleaseLease relinquishes runID if held by holder (a no-op otherwise) so another process can
	// take it immediately rather than waiting for expiry.
	ReleaseLease(ctx context.Context, runID, holder string) error
}

Leaser is the optional capability that coordinates driving a run across processes. Without it, several processes recovering against a shared store all re-drive the same in-flight runs. That is safe but wasteful: side effects stay at-most-once because each non-idempotent call is guarded by an exclusive attempt claim (ClaimAttempt), but the drivers duplicate model calls and a loser of a claim halts with ResumeHalt. A store that implements Leaser lets a driver take a time-bounded lease on a run so normally only the holder drives it; a dead holder's lease expires and another process takes over, which is the high-availability property. Recover uses it automatically when the store provides it. A lease is an efficiency and liveness mechanism, not the safety one: no lease can guarantee mutual exclusion against a holder that stalls past its TTL, which is why the claim, not the lease, is what keeps side effects at-most-once.

The Store port does not require leasing. MemStore's implementation is in-process (for tests and as the reference), SQLite's coordinates the processes sharing one database file, and Postgres's the nodes sharing one database, each with an atomic upsert over a leases table.

type Lister 

type Lister interface {
	// Runs yields the IDs of the runs f admits, in ascending byte order. It yields an error at
	// most once, as its last element, and holds no connection or lock across a yield.
	Runs(ctx context.Context, f RunFilter) iter.Seq2[string, error]
}

Lister is the optional store capability a crash-recovery supervisor needs: it enumerates the runs a store holds, so Recover can find the ones to re-drive after a restart. A store opts in by implementing Runs; Recover finds it with Capability, so a wrapper that implements Unwrap keeps its inner store's Lister.

type MemStore 

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

MemStore is an in-memory Store for tests and local development. It also implements Lister and Leaser (leases are in-process), so Recover and Lease work on it as on the SQL stores. SQLite (store/sqlite) is the on-disk default and store/postgres the high-availability backend.

Each run's entries get consecutive Seq values from 0, in insertion order, so MemStore meets the Store requirements trivially: an Insert is atomic under its mutex, and a read sees every entry inserted before it.

func NewMemStore 

func NewMemStore() *MemStore

NewMemStore returns an empty MemStore.

func (*MemStore) AcquireLease 

func (m *MemStore) AcquireLease(_ context.Context, runID, holder string, ttl time.Duration) (bool, error)

AcquireLease implements Leaser.

func (*MemStore) Do deprecated 

func (m *MemStore) Do(ctx context.Context, runID, name string, fn func(context.Context) (Record, error)) (Record, error)

Do runs a memoized step through m's Journal (see Journal.Do).

Deprecated: transitional; removed by the 1.0 rewrite. Use a Journal over the store.

func (*MemStore) Get added in v0.9.0 

func (m *MemStore) Get(ctx context.Context, runID, name string) (Entry, bool, error)

Get implements Store.

func (*MemStore) History deprecated 

func (m *MemStore) History(ctx context.Context, runID string) ([]Record, error)

History reads a run back through m's Journal (see Journal.History).

Deprecated: transitional; removed by the 1.0 rewrite. Use a Journal over the store.

func (*MemStore) Insert added in v0.9.0 

func (m *MemStore) Insert(ctx context.Context, runID, name string, data []byte) (Entry, bool, error)

Insert implements Store.

func (*MemStore) Journal deprecated added in v0.9.0 

func (m *MemStore) Journal() *Journal

Journal returns the Journal over m that its Do and History shims delegate to.

Deprecated: transitional; removed by the 1.0 rewrite. Use NewJournal(m).

func (*MemStore) Load added in v0.9.0 

func (m *MemStore) Load(ctx context.Context, runID string, after int64) iter.Seq2[Entry, error]

Load implements Store. It takes the run's entries as of the call and yields them without holding the store's mutex, so the caller may write to the store inside the loop.

func (*MemStore) ReleaseLease 

func (m *MemStore) ReleaseLease(_ context.Context, runID, holder string) error

ReleaseLease implements Leaser.

func (*MemStore) RenewLease 

func (m *MemStore) RenewLease(_ context.Context, runID, holder string, ttl time.Duration) (bool, error)

RenewLease implements Leaser.

func (*MemStore) Runs 

func (m *MemStore) Runs(ctx context.Context, f RunFilter) iter.Seq2[string, error]

Runs implements Lister: the runs f admits, in ascending order.

type MemWaker 

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

MemWaker is an in-process Waker: it holds pending timers and, when Fire is called with the current time (or via a Start ticker), resumes every run whose timer is due by calling the resume function it was built with. resume is how the deployment re-invokes a run, typically agent.Run(ctx, runID, savedInput); a resumed run whose wait is over proceeds, and one that sleeps again re-registers a new wake here. MemWaker is a reference and a local-dev default, not a durable scheduler: a process exit loses its in-memory timer set, so the wake times must also live in the durable journal (they do, Sleep journals them), and a restarted deployment rebuilds pending wakes by scanning runs, or an external scheduler owns the trigger.

func NewMemWaker 

func NewMemWaker(resume func(ctx context.Context, runID string) error) *MemWaker

NewMemWaker builds an in-process waker that resumes due runs via resume.

func (*MemWaker) Fire 

func (w *MemWaker) Fire(ctx context.Context, now time.Time) (int, error)

Fire resumes every run that has a timer due at or before now, each run once even if several of its timers are due, drops the fired timers, and returns how many runs it resumed. Resume callbacks run without the lock held, so a resumed run may register a new wake. Errors from resume are joined.

A resume that fails (the model provider or the store was briefly unavailable) leaves the run where it was, so its due timers are put back and the next Fire retries it; without that, one transient failure would leave the run asleep with nothing to wake it. A resume that ends in a Pause (IsPause: the run slept again, or awaits approval, an interrupt, a signal, or a halt resolution) is not retried: the run is waiting on something else, and waking it every tick would only spin.

func (*MemWaker) Schedule 

func (w *MemWaker) Schedule(_ context.Context, wk Wake) error

Schedule registers a wake, idempotent per (w.RunID, w.Name): re-scheduling the same timer overwrites its fire time rather than adding a duplicate. Timers of the same name in different sub-runs of one root are distinct, and the root is resumed once when either is due. An empty RootRunID resumes RunID. An empty RunID is ErrConfig.

func (*MemWaker) Start 

func (w *MemWaker) Start(ctx context.Context, every time.Duration, onError func(error)) <-chan struct{}

Start runs Fire on a ticker every `every` until ctx is cancelled, so sleeping runs wake on their own. Resume errors go to onError if non-nil. It is the turnkey local loop; a production deployment may prefer an external scheduler that owns the trigger and durable timer set.

The returned channel closes once the loop has stopped, after any Fire in flight when ctx was cancelled has returned. Wait on it before closing the store the resumed runs write to.

type Message 

type Message struct {
	Role  Role
	Parts []Part
}

Message is one turn. Its content is a sequence of typed Parts (NOT a flat string), so reasoning/thinking parts (which Anthropic requires echoed back across turns), tool calls, tool results, and image inputs all survive round-trips intact. This is the provider-neutral type; adapters translate to/from provider wire formats.

func SystemText 

func SystemText(s string) Message

func UserParts 

func UserParts(parts ...Part) Message

UserParts builds a user message from mixed parts, e.g. a prompt and one or more images: UserParts(Text{"what is this?"}, ImageData("image/png", raw)). It is the multimodal counterpart to UserText.

func UserText 

func UserText(s string) Message

UserText and SystemText are convenience constructors.

func (Message) MarshalJSON 

func (m Message) MarshalJSON() ([]byte, error)

func (Message) Text 

func (m Message) Text() string

Text returns the concatenation of the message's Text parts (ignoring reasoning, tool calls, and tool results). Empty if the message carries no text.

func (*Message) UnmarshalJSON 

func (m *Message) UnmarshalJSON(b []byte) error

type Middleware 

type Middleware func(next ModelHandler) ModelHandler

Middleware wraps a ModelHandler: the net/http-style func(Handler) Handler chain, at the semantic layer (it sees messages, tool calls and token usage, not bytes). Batteries live in the middleware package (Retry, Hedge, RateLimit, Cost). A middleware passes on the call it received, or a copy of it with fields changed (c := call; c.Model = backup): the call carries the agent's per-turn state in unexported fields, and the agent's model handler refuses a ModelCall that was built from scratch rather than derived from the one the middleware received (ErrConfig).

func WithRetrieval 

func WithRetrieval(r Retriever, k int) Middleware

WithRetrieval is model middleware that auto-injects retrieved context (classic RAG): it retrieves the top-k documents for the run's user message and adds them, on every model call of the run (so the call that follows a tool result still has them), as a user message placed just before that user message: after the agent's system prompt and any earlier turns. The documents are data the operator does not control, so they are not given system authority, and each is written as one line of JSON, so no document can forge another entry (see formatDocs).

Inside an agent run the retrieval is a journaled step (read-only, so a crash before it is recorded simply retrieves again): the run retrieves once, every later call of the run (including after a crash and resume, or from a second driver) is given the documents it recorded, and the journal holds what the model was shown. Retrieved text is therefore durable content, stored like a tool result. Called outside a run, where there is no journal, it retrieves for the latest user message on every call.

A retrieval error aborts the model call (and nothing is recorded, so the next attempt retrieves again): have your Retriever return (nil, nil) instead of an error if you prefer to degrade to no context. A Retriever that returns more than k documents is cut to its first k. It panics if k is below 1.

type Model 

type Model interface {
	Stream(ctx context.Context, req Request) (*Stream, error)
}

Model is the provider primitive. Streaming is FIRST-CLASS: Stream is the only method an adapter must implement. Generate is a convenience drain built on top — the opposite of frameworks that make blocking generation primary and bolt streaming on later.

func Replay 

func Replay(ctx context.Context, source Durable, runID string) (Model, error)

Replay returns a Model that re-emits the model outputs recorded in `source` for runID, in order, instead of calling a live LLM. Run an agent with it against a FRESH store to deterministically re-execute a past run:

  • time-travel debugging: step through exactly what happened, offline and free;
  • regression tests: capture a production run, replay it in CI (pairs with testing/synctest), assert behavior didn't drift;
  • evals over real traffic: the journal IS a golden dataset.

Because the journal captures every model output across the whole (possibly nested) tree, the replay is exact. Each turn ends with the Finish its record holds: the reason and the provider's raw reason the original turn journaled (a record written before they were journaled gets the reason its message implies, "tool_use" when it calls tools, else "stop"), and the replayed turn's record names the model its original record named (Record.Model), not the replaying Model. Spend is replayed too: a turn reports the usage the original turn discarded (failed attempts, losing hedge targets, and requests journaled in a late spend record after it) in its Finish's Discarded, beside its own Usage, and a model call that failed for good in the original fails again at the same point, reporting the same usage, so the replayed run's Result.Spend, journal and WithTokenBudget stops match the original's. A middleware in the replaying agent that retries such a failure moves on to the next recorded response instead. This is only possible because the durable substrate records a complete, replayable history in the first place.

type ModelAttempt added in v0.9.0 

type ModelAttempt struct {
	// Response is what the request returned. When Err is set, only its Usage is meaningful: the
	// usage the request reported before it failed, which the provider may still bill.
	Response ModelResponse
	// Discarded is usage the request's model reported as billed for responses it threw away (see
	// Finish.Discarded): zero from a live adapter, the recorded discarded spend from Replay.
	Discarded Usage
	// Err is the request's error, or the agent's rejection of its response (a reused tool-use
	// id), or the error of a Before hook that kept it from being sent.
	Err error
}

ModelAttempt is how one request ended, as an After hook sees it.

type ModelCall added in v0.9.0 

type ModelCall struct {
	// Request is what the call asks of the model.
	Request Request
	// Model is where the agent's model handler sends the call's requests. A middleware may
	// retarget a copy of the call (Hedge sends its backups this way).
	Model Model
	// RunID is the run the call belongs to, and Turn the model turn's sequence number within it
	// (0 for the first). Both are zero for a call made through CallModel.
	RunID string
	Turn  int
	// contains filtered or unexported fields
}

ModelCall is one model call as it passes through the middleware chain. It is a value: a middleware changes a copy and passes that on, so what one middleware changes is seen only by the handlers it calls, never by its caller or by a sibling (a Hedge's other targets).

Request.Messages and Request.Tools arrive at every handler slices.Clip'ped, so a handler that appends to them gets a new backing array and never writes into a slice another handler holds. The elements are shared: replace an element through a copy of the slice rather than assign into it.

func (ModelCall) AddHook added in v0.9.0 

func (c ModelCall) AddHook(h ModelCallHook) ModelCall

AddHook returns a copy of c with h appended to its hooks, which the agent's model handler runs around every request it sends for the returned call and every call derived from it. Hooks are append-only: a middleware can add one, never remove or reorder those added outside it, and the agent's own spend accounting is not a hook at all, so no middleware can hide a request from the run's budget or Result.Spend.

func (ModelCall) Attempt added in v0.9.0 

func (c ModelCall) Attempt() int

Attempt is the number of the request a hook is observing. The turn numbers each request when it reaches the agent's model handler, before its Before hooks run, from one counter for the whole turn: 1 for the first, 2 for the next, and so on, so the requests of every Retry attempt and every Hedge target are numbered apart and every hook of a request, Before and After, sees the same number. A request a Before hook refused keeps its number although it was not sent, so the numbers of the requests sent can have gaps. It is 0 on a call that has not reached the model handler (as a middleware receives it).

func (ModelCall) OnAnswer added in v0.9.0 

func (c ModelCall) OnAnswer(key any, fn func(ctx context.Context, resp ModelResponse)) bool

OnAnswer registers fn to run once with the call's answer: the response the whole chain returns to the agent, after the agent's checks, once the journal holds it as the turn's record (for CallModel, which has no journal, once the call returns it). It runs after every request of the call has had its hooks, only when the call succeeds, and not at all when the turn's record is not written (the write failed, or another driver of the run recorded the turn first): fn runs once per recorded turn. Registrations are keyed: a key registered again for the same turn (a middleware inside a Hedge or a Retry registers once per target or attempt) keeps the first registration, so fn runs once per turn wherever the middleware sits; key must be comparable, and a pointer the middleware allocates is the usual choice. Cost counts answers this way.

It reports whether the call belongs to a turn. A call a middleware kept and passes on after its turn is over belongs to that turn: OnAnswer reports true and fn never runs, since the turn's answer is already decided. It reports false only for a call that belongs to no turn (a handler called directly, not through an agent or CallModel).

type ModelCallHook added in v0.8.0 

type ModelCallHook struct {
	// Before runs before the request is sent, with the request's context and the call as it
	// reached the model handler (Attempt set). An error fails the request without sending it:
	// the hooks after it do not run, and the hooks before it get their After with that error.
	Before func(ctx context.Context, call ModelCall) error
	// After runs once the request has ended, successfully or not.
	After func(ctx context.Context, call ModelCall, a ModelAttempt)
}

ModelCallHook runs around each request the agent's model handler sends: every retried attempt and every hedged target, wherever the middleware that added it (ModelCall.AddHook) sits in the chain. Middleware whose effect belongs to requests actually sent, rather than to calls through it, uses one: RateLimit waits for a token in Before, Cost counts spend in After.

The hooks of a call run in the order they were added, so a middleware nearer the outside of the chain runs its Before first. Each Before runs once per request; a hook whose Before returned nil gets exactly one After for that request.

type ModelEvent 

type ModelEvent struct {
	Event Event `json:"event"`
}

ModelEvent forwards one live model stream Event (TextDelta, ReasoningDelta, ToolCallDelta, Finish) from the current turn — the token-by-token feed. Emitted only for a FRESH model call; on durable replay the turn is reused from the journal and produces no deltas (an AssistantTurn with Replayed=true is emitted instead). When the turn's model call starts over (see TurnRestarted), the deltas before it are not part of the recorded turn.

type ModelHandler 

type ModelHandler func(ctx context.Context, call ModelCall) (ModelResponse, error)

ModelHandler answers one model call: it returns the response the turn records. Middleware wraps it; the innermost handler, which the agent (or CallModel) supplies, sends each request to call.Model. On error the response carries at most the usage the failed request reported.

type ModelInfo added in v0.8.0 

type ModelInfo struct {
	Provider       string `json:"provider,omitempty"`
	Model          string `json:"model,omitempty"`
	ResponseFormat bool   `json:"response_format,omitempty"`
}

ModelInfo identifies a Model. Provider names the API it calls ("anthropic", "openai", "gemini"), Model is the provider's model ID the requests name, and ResponseFormat reports whether the Model honours Request.ResponseFormat (a Model that does not fails a request that sets it, rather than drop the constraint).

func ModelInfoOf added in v0.8.0 

func ModelInfoOf(m Model) (ModelInfo, bool)

ModelInfoOf describes m. It returns m's own Describe when m implements Describer, and otherwise follows an Unwrap() Model method, as a wrapping Model (middleware, a test double) offers, to the first Model in the chain that does. It reports false when no Model in the chain describes itself, when the chain ends in nil, or when it is longer than 64 links.

type ModelResponse added in v0.9.0 

type ModelResponse struct {
	Message   Message
	Usage     Usage
	Finish    FinishReason
	RawFinish string // the provider's own finish reason, as it sent it; empty if it sent none
	// contains filtered or unexported fields
}

ModelResponse is the answer to a model call: the assistant message, the usage of the request that produced it, and why the turn ended. Finish is never empty in a response the agent records: an empty reason, which a Model may send when it does not know one, counts as FinishStop (see Finish). The loop decides whether to run tools from the message's tool calls, never from Finish (an OpenAI forced tool_choice reports stop alongside tool calls).

func CallModel added in v0.9.0 

func CallModel(ctx context.Context, m Model, req Request, mw ...Middleware) (ModelResponse, error)

CallModel sends one model call outside an agent, through mw (first = outermost) and the same model handler an agent uses: hooks run around every request, Messages and Tools arrive clipped, and the response is checked as an agent checks it (usage, finish reason, tool-use ids). There is no journal, no stream and no run, so RunID and Turn are zero. It returns ErrConfig when m is nil.

type OpKind added in v0.9.0 

type OpKind string

OpKind is the kind of operation a halt is on. 

const (
	// OpTool is a tool call; OpRef.ID is its tool-use ID.
	OpTool OpKind = "tool"
	// OpStep is a Step; OpRef.ID is the step name.
	OpStep OpKind = "step"
)

type OpRef added in v0.9.0 

type OpRef struct {
	Kind     OpKind
	ID       string
	ToolName string
}

OpRef names an operation in a run's journal: a tool call (ID is the tool-use ID and ToolName the tool's name) or a Step (ID is the step name, ToolName is empty).

type Outcome added in v0.9.0 

type Outcome struct {
	Result   any
	IsError  bool
	Evidence any
}

Outcome is the verified outcome of a halted operation, recorded by ResolveHaltRef as the operation's result. Result is the value an actual call would have returned (JSON-marshalled); IsError marks an outcome the model should read as a failure (in a saga, a failed step, which rolls the saga back). A non-nil Evidence marks the resolution reconciled from evidence and stores Evidence (JSON-marshalled) beside the result, as WithEvidence does.

type OutcomeUnknown added in v0.9.0 

type OutcomeUnknown struct {
	RunRef
	Op OpRef
	// AttemptedAt is when the effect was attempted (the live attempt marker's timestamp), zero
	// if unknown (including a marker stamped zero or negative, see markerTime). A reconciler uses
	// it to honor a grace period before resolving (see WithMinHaltAge) so it does not query the
	// provider before its record has settled.
	AttemptedAt time.Time
	Cause       HaltCause
}

OutcomeUnknown is returned when a run cannot safely proceed past an operation whose outcome is unknown: a non-retriable tool call or Step was attempted (its marker is recorded) but no result was. The run stops for confirmation rather than risk a double side effect (a double charge). Cause says why: HaltCrashed when the marker was already there with no live claimant known (the driver that claimed it most likely died mid-effect), HaltContended when another driver of the same run won the claim while this one ran (for example one that took over after this node's lease lapsed); that driver owns the effect, and once it records the result, re-running proceeds normally.

Clear it with ResolveHaltRef(ctx, store, halt.Ref(), outcome), then re-run RootRunID.

func (*OutcomeUnknown) Error added in v0.9.0 

func (e *OutcomeUnknown) Error() string

Error names the operation whose outcome is unknown, and says when another driver holds its claim.

func (*OutcomeUnknown) Ref added in v0.9.0 

func (e *OutcomeUnknown) Ref() HaltRef

Ref returns the reference ResolveHaltRef takes to clear this halt.

type Part 

type Part interface {
	// contains filtered or unexported methods
}

Part is a single piece of message content. Concrete parts: Text, Reasoning, ToolUse, ToolResult, Image.

type Pause added in v0.9.0 

type Pause interface {
	error
	// Paused names the paused run: RunID is the journal the pause lives in (a sub-agent's
	// sub-run for a pause raised inside a sub-agent) and RootRunID is the run to re-invoke.
	Paused() RunRef
	// contains filtered or unexported methods
}

Pause is the error a run returns when it has stopped durably and is waiting on something outside it: a human's approval (*ApprovalPending), an answer to an interrupt (*InterruptPending), a signal or a channel message (*SignalPending), a durable timer (*TimerPending), or a verdict on a side effect whose outcome is unknown (*OutcomeUnknown). A pause is not a failure: the run's journal is intact, and once the condition is met the run is re-invoked with Paused().RootRunID and carries on.

The set is sealed: only this package's five types implement Pause, so a type switch over them is exhaustive. Test for a pause with IsPause and read it with AsPause, which see through wrapping.

func AsPause added in v0.9.0 

func AsPause(err error) (Pause, bool)

AsPause returns the first Pause in err's chain, if there is one.

type PendingApproval deprecated 

type PendingApproval = ApprovalPending

PendingApproval is the former name of ApprovalPending.

Deprecated: transitional; renamed by the 1.0 rewrite. Use ApprovalPending.

type RateLimited 

type RateLimited struct {
	RetryAfter time.Duration
	Message    string // the provider's error message, if it sent one
	Err        error
}

RateLimited is returned by a provider adapter when it is rate limited (HTTP 429 that is not an exhausted quota; see ClassifyHTTPError). It carries an optional RetryAfter hint from the Retry-After response header or the provider's own retry delay (0 means no hint was provided), and the provider's error message. It wraps the caller-supplied Err (typically wrapping ErrModel) so errors.Is(err, ErrModel) holds.

func (*RateLimited) Error 

func (e *RateLimited) Error() string

func (*RateLimited) Unwrap 

func (e *RateLimited) Unwrap() error

type Reasoning 

type Reasoning struct {
	Text      string `json:"text"`
	Signature string `json:"signature,omitempty"`
	Redacted  string `json:"redacted,omitempty"`
}

Reasoning is model thinking. Signature is the provider's opaque token that must be echoed back on later turns (Anthropic extended thinking); dropping it corrupts the conversation — which is exactly why Content can't be a flat string.

Redacted holds reasoning the provider returned encrypted (Anthropic redacted_thinking data), which must also go back unchanged; Text and Signature are empty when it is set. Each thinking block is its own Reasoning part, in the order the model produced them.

type ReasoningDelta 

type ReasoningDelta struct {
	Text      string `json:"text"`
	Signature string `json:"signature"` // opaque provider token to echo back on later turns (Anthropic thinking)
	Redacted  string `json:"redacted"`  // encrypted reasoning to echo back unchanged (Anthropic redacted_thinking)
}

ReasoningDelta is a fragment of model thinking. A Signature ends the thinking block it belongs to, so the next ReasoningDelta starts a new Reasoning part. Redacted carries a whole encrypted block (Anthropic redacted_thinking), which becomes a Reasoning part of its own.

type Received 

type Received[T any] struct {
	Key     string
	Payload T
}

Received is one message pulled from a channel.

func Receive 

func Receive[T any](ctx context.Context, channel string) (Received[T], error)

Receive returns the oldest not-yet-acked message on channel, in delivery order. If the channel has no unacked message it returns *SignalPending and the run pauses durably. The run must Ack(channel, msg.Key) after it has durably handled the message; until then Receive keeps returning the SAME message, which is what makes processing replay-safe and exactly-once. Call from a retry-safe tool (Safety.ReadOnly or Idempotent), like Await.

type Record 

type Record struct {
	Name    string   `json:"name"`
	Kind    StepKind `json:"kind"`
	Message *Message `json:"message,omitempty"` // StepModel
	// Usage is token usage. On a StepModel record, the call's. On a StepToolResult or StepSagaFail
	// record, the Usage of the runs the tool call started (sub-agents): their recorded responses.
	Usage *Usage `json:"usage,omitempty"`
	// DiscardedUsage is billed usage no recorded response carries. On a StepModel record, the
	// turn's other model requests: failed attempts a middleware retried, losing hedge targets. On
	// a StepValue record named "@spend/<id>", a model call that failed for good; on one named
	// "@spend-late/<id>", requests that ended after their turn was recorded, or whose turn another
	// driver recorded. On a
	// StepToolResult or StepSagaFail record, the rest of the Spend of the runs the tool call
	// started. Nil when there was none. WithTokenBudget counts it.
	DiscardedUsage *Usage `json:"discarded_usage,omitempty"`
	// Finish is why the model turn ended, and RawFinish the provider's own reason as it sent it
	// (see Finish). StepModel only; empty on a record written before they were journaled.
	Finish    FinishReason `json:"finish,omitempty"`
	RawFinish string       `json:"raw_finish,omitempty"`
	// Model identifies the model that answered the turn (see ModelInfoOf), for audit and for a
	// run whose turns a middleware sent to different providers. StepModel only; nil when the
	// model does not describe itself or no request produced the response (a middleware built it).
	Model *ModelInfo `json:"model,omitempty"`
	// PromptDigest and ToolsDigest are digests of the system prompt and the tool set the turn was
	// sent (see the functions of the same names), so an auditor can tell which instructions and
	// which tools each answer was given, although agent-level defaults stay live across a
	// redeploy. StepModel only; empty when the turn was sent no system prompt, or no tools.
	PromptDigest string          `json:"prompt_digest,omitempty"`
	ToolsDigest  string          `json:"tools_digest,omitempty"`
	ToolUseID    string          `json:"tool_use_id,omitempty"`  // StepToolResult
	Result       json.RawMessage `json:"result,omitempty"`       // StepToolResult / StepValue
	IsError      bool            `json:"is_error,omitempty"`     // StepToolResult
	Approved     bool            `json:"approved,omitempty"`     // StepApproval
	Approver     string          `json:"approver,omitempty"`     // StepApproval written by SubmitDecision
	ApproverAlg  Alg             `json:"approver_alg,omitempty"` // StepApproval written by SubmitDecision: the scheme Signature is under
	Signature    []byte          `json:"signature,omitempty"`    // StepApproval written by SubmitDecision
	// AttemptedAt is the Unix-millis wall-clock time an attempt marker (StepAttempt) was
	// written, i.e. just before a non-retriable side effect fired. It is set once and read
	// back verbatim on replay, so it stays deterministic. Zero (and omitted) on every
	// other kind; a reconciler uses it to honor a grace period before resolving a halt.
	AttemptedAt int64 `json:"attempted_at,omitempty"`
	// Reconciled marks a StepToolResult that ResolveHalt injected from verified evidence
	// rather than one the tool produced by running. It lets a later reader (and bide-audit)
	// tell a reconciled outcome from a clean one at a glance.
	Reconciled bool `json:"reconciled,omitempty"`
	// ReadOnly marks a StepToolResult for a call whose tool was declared ReadOnly when the call
	// ran. A saga rollback skips such a call (it changed nothing) and treats any other completed
	// call as a write, whatever the tool is declared as by the time the rollback runs. A result
	// that ResolveHalt injected is never ReadOnly: only a call that was not retry-safe halts.
	ReadOnly bool `json:"read_only,omitempty"`
	// Evidence is what a reconciler read to decide the outcome (a queried provider record,
	// a message id, a log line). It is carried on the reconciled result and signed with it,
	// so the verdict and its basis live in the journal beside the outcome.
	Evidence json.RawMessage `json:"evidence,omitempty"`
	// Format is the journal format a StepHeader record names (see JournalFormat). Empty on every
	// other kind.
	Format string `json:"format,omitempty"`
	// Redacted marks a record whose stored bytes a redaction replaced with a tombstone, the
	// reserved form {"redacted":{"leaf_hash":"<hex>","at_ms":<ms>}}: the hex audit leaf hash of
	// the bytes it replaced and the Unix-millis time of the redaction. Only Name is meaningful on
	// such a record, and Raw returns the tombstone. It is never journaled: the tombstone is what the
	// store holds. A run holding a redacted record is over and cannot be driven again.
	Redacted bool `json:"-"`
	// contains filtered or unexported fields
}

Record is one durably-recorded, named event. The ordered sequence of Records for a run is its full, replayable history; resume rebuilds state by replaying them. Name is the stable identity used for memoization (see Durable.Do).

func ClaimAttempt added in v0.5.0 

func ClaimAttempt(ctx context.Context, d Durable, runID, name string, rec Record) (won bool, got Record, err error)

ClaimAttempt writes the attempt marker named name as an exclusive claim and reports whether the caller won it. It stamps rec with a fresh random claim id and inserts it; because a journal records a name at most once, exactly one driver's marker is stored, and a driver won only if the marker it gets back carries its own claim (see Record.ClaimID). A driver that loses must not run the side effect the marker guards: another driver owns it and may be running it right now.

This makes at-most-once independent of leasing. A lease cannot guarantee mutual exclusion (a holder stalled past its TTL wakes still believing it holds the lease), but two drivers that overlap still cannot both win the claim, since the store's atomic insert decides the winner. With a single driver the claim is always won, so a run with no contention behaves exactly as before.

Every claim is made with a fresh id. If the marker cannot be written, the claim records that its attempt did not start, and if even that record cannot be written, the process remembers it and writes it again at the next claim of the same name, which then loses to the marker its earlier claim may have left: a caller that claims one name only (rather than numbered re-attempts, as Step and tool calls do) halts on it, which is safe.

func DecodeRecord added in v0.7.0 

func DecodeRecord(b []byte) (Record, error)

DecodeRecord decodes a record from its journal encoding (see EncodeRecord) into an independent copy that shares no memory with b. The record's Raw is a copy of b.

func DecodeStoredRecord added in v0.8.0 

func DecodeStoredRecord(runID, name string, b []byte) (Record, error)

DecodeStoredRecord decodes the record a store holds for the step name of run runID (see DecodeRecord) and checks that the record carries that name. Every record is journaled with the name it is stored under (JournalEntry), and the engine finds records by the name inside them (IsComplete, the approval tally, a signal, an attempt marker), so a row whose record names another step (a row edited or copied in the backing store) would be read as that step: a row under "x" holding a record named run:complete would mark an unfinished run complete. Such a row is ErrStorage, naming the run and the key, like any other stored record that does not decode: the store's contents are wrong, whatever wrote them. Fields this version does not know still decode, so a journal a newer version wrote stays readable.

A redaction tombstone (see Record.Redacted) decodes as a record that carries only its name.

Every Durable implementation must read a record back through it, in Do and in History.

func (Record) ClaimID added in v0.9.0 

func (r Record) ClaimID() string

ClaimID returns the random id of the driver that wrote this attempt marker or not-started record (see ClaimAttempt), or "" for any other record. The journal sets it when it records the marker, and it is read back verbatim.

func (Record) MarshalJSON added in v0.9.0 

func (r Record) MarshalJSON() ([]byte, error)

MarshalJSON returns the record's JSON encoding, claim and salt included, without HTML escaping (see EncodeRecord for the canonical form a store persists).

func (Record) Raw added in v0.9.0 

func (r Record) Raw() []byte

Raw returns the bytes the store holds for the record, verbatim, for a record read back from a journal (Journal.Get, Journal.History, Journal.Records, or a Durable's Do and History). It is nil for a record built in memory. The returned slice is a copy.

func (Record) Salt added in v0.7.0 

func (r Record) Salt() []byte

Salt returns the record's salt: SaltSize random bytes the journal sets when it first records the record (see JournalEntry), replacing any salt the step returned (the engine draws the salt of its own model records itself, just as randomly, to tell its record from another writer's). It is persisted with the record and read back verbatim, so it is stable across replay and across stores. It has no meaning to the run; the audit trail needs it. An audit leaf commits to the record's journal encoding, salt included, and an inclusion proof for one record carries its neighbours' leaf hashes, so without the salt anyone holding a proof could confirm a guessed neighbour (an approval, a small tool result) by hashing it. The salt is disclosed only with its own record. The returned slice is a copy, nil for a record built in memory.

func (*Record) UnmarshalJSON added in v0.9.0 

func (r *Record) UnmarshalJSON(b []byte) error

UnmarshalJSON decodes a record from its JSON encoding, claim and salt included.

type RecoverOption 

type RecoverOption func(*recoverConfig)

RecoverOption configures Recover. Lease options apply only when the store implements Leaser.

func WithLeaseHolder 

func WithLeaseHolder(id string) RecoverOption

WithLeaseHolder names the worker that claims leases, as the leases table and logs show it. Defaults to a host/pid/random string. The name identifies, it does not grant: each Lease call (and each run Recover drives) claims under a token of its own derived from it, so two drivers sharing a name still exclude each other, and a restarted worker waits out its predecessor's lease (the TTL) like any other process rather than taking it over.

func WithLeaseTTL 

func WithLeaseTTL(d time.Duration) RecoverOption

WithLeaseTTL sets how long an acquired lease is valid. Lease renews it while a run is driving, so a crash lets the lease expire after roughly this long and another process (RecoverLoop) takes over. Defaults to 30s. Set it well above the store's round-trip time and the longest pause you expect a process to take: renewal starts at half the TTL, and a drive that cannot renew within three quarters of it is cancelled with ErrLeaseLost (see Lease). It must be positive: Lease, Recover and RecoverLoop return an ErrConfig error otherwise.

func WithRecoverConcurrency added in v0.8.0 

func WithRecoverConcurrency(n int) RecoverOption

WithRecoverConcurrency caps how many runs RecoverLoop drives at once. Defaults to 16. It must be at least 1. Only RecoverLoop reads it.

func WithRecoverErrors added in v0.8.0 

func WithRecoverErrors(fn func(error)) RecoverOption

WithRecoverErrors sets the function RecoverLoop hands each genuine failure to: a store error while enumerating runs or acquiring a lease, or a drive that failed. Pauses and lost leases are not failures and are not reported. Without it, RecoverLoop drops failures (the next pass retries the run). It may be called from several goroutines, one call at a time. Only RecoverLoop reads it.

func WithRecoverInterval added in v0.8.0 

func WithRecoverInterval(d time.Duration) RecoverOption

WithRecoverInterval sets how often RecoverLoop starts a recovery pass. Defaults to half the lease TTL, so a dead holder's run is taken over within about 1.5 TTLs of its last renewal. It must be positive. Only RecoverLoop reads it.

type Request 

type Request struct {
	Messages       []Message
	Tools          []Tool
	Sampling       Sampling        // generation controls; zero value = provider/model defaults
	ResponseFormat *ResponseFormat // nil = free-form; set = constrain output to a JSON schema
	ToolChoice     *ToolChoice     // nil = provider default (auto); see ToolChoice
}

Request is a single model call.

type ResolveOption added in v0.3.0 

type ResolveOption func(*resolveConfig)

ResolveOption configures ResolveHaltRef (and its wrappers ResolveHalt and ResolveStepHalt).

func WithEvidence added in v0.3.0 

func WithEvidence(v any) ResolveOption

WithEvidence records the resolution as reconciled from verified evidence rather than an operator's bare assertion: it marks the injected result Reconciled and stores v (JSON-marshalled) as its Evidence, signed alongside the outcome. Use it so a later reader can tell a reconciled step from a clean one and re-check the basis of the verdict. Outcome.Evidence does the same; giving both is ErrConfig.

func WithMinHaltAge added in v0.3.0 

func WithMinHaltAge(d time.Duration) ResolveOption

WithMinHaltAge refuses to resolve a halt younger than d, measured from the live attempt marker's AttemptedAt to now (an attempt recorded as never started does not count). A reconciler passes this so it cannot resolve before the provider's record has had time to settle (a sent-message id can appear seconds after the send): resolving too early reads "absent" and re-fires the very side effect the halt exists to prevent. A HaltContended halt requires it, and so does any halt on a store that cannot lease runs (no Leaser), unless WithoutLiveDriverCheck is given. d <= 0 skips the check. When d > 0 but no attempt timestamp is found, resolution errors rather than resolve blind; a marker whose AttemptedAt is zero or negative carries none (a negative value is not one the engine writes). Returns *HaltTooYoung when the halt has not aged enough, including a marker stamped in the future, so the caller waits and retries later.

func WithNow added in v0.3.0 

func WithNow(now func() time.Time) ResolveOption

WithNow overrides the clock WithMinHaltAge measures against (default time.Now). For tests and callers that carry their own clock.

func WithoutLiveDriverCheck added in v0.9.0 

func WithoutLiveDriverCheck() ResolveOption

WithoutLiveDriverCheck skips ResolveHaltRef's check that no driver may still be running the halted operation: the lease probe on a store that leases runs, and the WithMinHaltAge requirement on one that does not. With it, a resolution can race a live driver's own result. Pass it only when you know no driver of the run is running, for example with every worker stopped.

type ResponseFormat 

type ResponseFormat struct {
	Name   string
	Schema json.RawMessage
}

ResponseFormat asks the provider to constrain the model's output to a named JSON schema (OpenAI structured outputs / "json_schema" response format). An adapter that does not support it (Anthropic) fails the request with ErrConfig rather than drop the constraint. Name is the schema name; Schema is provider-neutral JSON Schema (as from the schema package). See RunTypedNative.

type Result 

type Result struct {
	// Message is the final assistant answer, identical to what Run / RunSaga return.
	Message Message

	// Usage is the sum of the usage of the model responses the run recorded, one per turn
	// (input, output, and cache tokens). It is the whole run's, read from the journal: the same
	// whether the run finished in one invocation or was resumed after a crash, and the same again
	// when a finished run is re-entered.
	Usage Usage

	// Spend is every token the run's model requests used: Usage plus the usage of requests whose
	// responses were discarded (failed attempts a middleware retried, losing hedge targets) and of
	// model calls that failed. It is what the provider bills, and what WithTokenBudget counts.
	// Like Usage it is the whole run's, from the journal (see Record.DiscardedUsage). A request
	// still in flight when an invocation ends (a hedge loser, a request a middleware left
	// running) is waited for, for at most two seconds and not past the run's context, and counted;
	// one that runs longer than that is not in Spend.
	Spend Usage

	// Turns is the number of live model turns this invocation executed. Turns replayed from
	// the journal are not counted: unlike Usage and Spend, it describes this invocation.
	Turns int

	// Duration is the wall-clock elapsed time for the run (from entry to return).
	Duration time.Duration

	// RunID echoes the run identifier passed to RunResult / RunSagaResult.
	RunID string
}

Result is a rich envelope returned by RunResult and RunSagaResult. It carries the terminal assistant message plus telemetry accumulated over the whole run: total token usage (sum of every model-call Usage), the number of model turns, wall-clock duration, and the run ID.

Run and RunSaga are unchanged and remain the idiomatic path for callers that only need the final message; RunResult / RunSagaResult are the additive counterparts for callers that need observability data.

type ResumeHalt deprecated 

type ResumeHalt = OutcomeUnknown

ResumeHalt is the former name of OutcomeUnknown.

Deprecated: transitional; renamed by the 1.0 rewrite. Use OutcomeUnknown.

type RetrievalOption added in v0.8.0 

type RetrievalOption func(*retrievalToolConfig)

RetrievalOption configures a RetrievalTool.

func RetrievalDescription added in v0.8.0 

func RetrievalDescription(description string) RetrievalOption

RetrievalDescription sets the description the model reads to decide when to call the tool, such as what the store holds.

func RetrievalName added in v0.8.0 

func RetrievalName(name string) RetrievalOption

RetrievalName names the tool (the default is "retrieve"), for an agent that searches more than one store, each through its own RetrievalTool.

type Retriever 

type Retriever interface {
	Retrieve(ctx context.Context, query string, k int) ([]Doc, error)
}

Retriever is the bring-your-own-RAG port: given a query, return the top-k relevant documents from YOUR store (pgvector, Pinecone, a file index — anything). Bide ships no vector store and no embedder; you implement Retrieve against infrastructure you already run, and wire it in with RetrievalTool (agentic — the model searches on demand) or WithRetrieval (classic — top-k auto-injected as context on each user turn). Sessions already give conversational memory; this is the seam for semantic / long-term memory.

Retrieve must be safe for concurrent use: parallel tool calls in one turn, concurrent runs on one Agent, and sub-agents sharing a Retriever call it at once.

type Role 

type Role string

Role identifies the author of a Message. 

const (
	RoleSystem    Role = "system"
	RoleUser      Role = "user"
	RoleAssistant Role = "assistant"
	RoleTool      Role = "tool"
)

type RunFilter added in v0.9.0 

type RunFilter struct {
	// After is a cursor: only run IDs greater than it (in byte order) are yielded.
	After string
	// Prefix admits only run IDs that start with it: a tenant or namespace, by the convention
	// that a deployment prefixes its run IDs.
	Prefix string
	// ExcludeHolding drops every run that holds an entry with any of these names. Recover passes
	// the terminal markers (run:complete, run:aborted, run:cancelled).
	ExcludeHolding []string
}

RunFilter selects the runs a Lister yields. A SQL store evaluates it in its query, so a recovery pass over many finished runs reads none of them.

func (RunFilter) Admits added in v0.9.0 

func (f RunFilter) Admits(runID string, holds func(name string) bool) bool

Admits reports whether f admits runID, given holds, which reports whether the run holds an entry with a given name. It is the filter's meaning, for a store that evaluates it in Go.

type RunKind added in v0.9.0 

type RunKind string

RunKind is what drives a run, as its run:start records it (see RunStart). 

const (
	// RunKindAgent is a run an Agent drives (Run, RunSaga, Stream, a sub-agent, a session turn).
	// A run:start with no kind, as agent runs record it, is this kind.
	RunKindAgent RunKind = "agent"
	// RunKindFlow is a run a plan flow drives (plan.Flow.Run). RunStart.Flow names the flow.
	RunKindFlow RunKind = "flow"
)

type RunRef added in v0.9.0 

type RunRef struct {
	RunID     string
	RootRunID string
}

RunRef names a paused run. RunID is the journal the pause lives in: record the answer against it (Approve, AnswerInterrupt, Signal, ResolveHaltRef). RootRunID is the top-level run to re-invoke to continue; it differs from RunID when the pause comes from inside a sub-agent, whose journal is RunID.

func (RunRef) Paused added in v0.9.0 

func (r RunRef) Paused() RunRef

Paused returns r. Every pause type embeds a RunRef and so reports it through Pause.

type RunStart added in v0.8.0 

type RunStart struct {
	Input string   `json:"input"`
	Saga  bool     `json:"saga,omitempty"`
	Kind  RunKind  `json:"kind,omitempty"`
	Flow  *FlowRef `json:"flow,omitempty"`
}

RunStart is how a run was started, as its first drive records it: the input it answers (for a Session turn, the turn's message) and whether it runs as a saga (RunSaga, StreamSaga, RunSagaResult, or a sub-agent called inside a saga). A run's model turns and tool calls answer that input under that entry point's rules, so every later drive is held to it: resuming an unfinished run with another input, or through the other entry point (Run for a saga, RunSaga for a run), is ErrConfig. A finished run returns its recorded answer whatever it is passed, as before.

A run whose earlier drives predate this record gets it on its first drive under this version, with the input and entry point that drive is given.

Kind says what drives the run. A plan flow's run records RunKindFlow, its flow's name in Flow, and its input as JSON text in Input; resuming it with another input (compared as canonical JSON, so the input decoded from Input and encoded again resumes it), under another flow's name, or driving it as an agent run (or an agent run as a flow) is ErrConfig. The flow's topology is held by its own record, flow:digest, not here.

func RecordedStart added in v0.8.0 

func RecordedStart(ctx context.Context, d Durable, runID string) (RunStart, bool, error)

RecordedStart returns how runID was started (see RunStart), and ok=false for a run whose journal holds no such record: one never driven, or one not driven since before the record existed. A session's turn run (IsSessionRun) records its message, but only the session can drive it (it seeds the turn with the transcript before that message), so Recover never hands one to its callback. A recovery callback uses it to re-drive a run with its own input and entry point: 

start, ok, err := agent.RecordedStart(ctx, store, runID)
if err != nil {
    return err
}
if !ok {
    start = startFor(runID) // the deployment's own record, for a run not driven under this version
}
if start.Kind == agent.RunKindFlow {
    return driveFlow(ctx, start.Flow.Name, runID, start.Input) // the flow's Run, with the input decoded
}
if start.Saga {
    _, err = a.RunSaga(ctx, runID, start.Input)
} else {
    _, err = a.Run(ctx, runID, start.Input)
}

type Safety 

type Safety struct {
	// ReadOnly: no external side effects — always safe to re-run.
	ReadOnly bool
	// Idempotent: mutates state but is safe to retry, because a repeat with the same
	// key is a no-op downstream. Set IdempotencyKey to make that concrete.
	Idempotent bool
	// IdempotencyKey derives a stable key from the args so a retried call can be
	// de-duplicated. Empty result => not derivable.
	IdempotencyKey func(args json.RawMessage) string
	// RequiresApproval pauses the run for a durable human decision (HITL) before the
	// tool executes, surfaced as *ApprovalPending; resume after agent.Approve.
	RequiresApproval bool
	// Approval, when non-nil, upgrades the approval gate from one
	// decision to an m-of-n human gate over a bounded, named set of
	// approvers. nil = the existing 1-of-1 RequiresApproval behavior.
	Approval *ApprovalPolicy
}

Safety declares how a tool may be retried when a run resumes after a crash and the call's outcome is unknown (it was invoked, but no result was journaled). Replaces a coarse Read/Write flag. For a trusted MCP server, its tool annotations (readOnlyHint / idempotentHint) map onto it (see mcp.TrustAnnotations).

func ToolSafety added in v0.6.0 

func ToolSafety(ctx context.Context) (s Safety, ok bool)

ToolSafety returns the Safety of the tool that a tool-middleware call is for, so a middleware can tell a side effect from a read before it retries, caches, or skips a call. ok is false outside a tool call, or when the call names no registered tool; a middleware should then treat the call as a side effect.

func (Safety) RetrySafe added in v0.6.0 

func (s Safety) RetrySafe() bool

RetrySafe reports whether a call to the tool may run more than once for one tool call: it is ReadOnly, Idempotent, or carries an IdempotencyKey. It is the test the agent applies on resume, and the one tool middleware applies before retrying a call (see ToolSafety).

type SagaAborted 

type SagaAborted struct {
	RunID         string
	Cause         error
	Compensated   []string
	Uncompensated []string
	CompensateErr error
}

SagaAborted is returned by RunSaga when a step failed and the transaction was rolled back. Compensated lists tools whose side effects were undone (reverse of execution, including sub-agent trees). Uncompensated lists the writes the rollback did not undo: a completed write with no compensator, a call whose outcome is unknown (the rollback halted on it, with a *ResumeHalt in CompensateErr, whether or not it has a compensator), or a call whose tool is no longer registered. They are side effects that may need manual cleanup. CompensateErr is non-nil if the rollback stopped (a compensator failed, or an outcome is unknown), so writes before it remain uncompensated.

func (*SagaAborted) Error 

func (e *SagaAborted) Error() string

func (*SagaAborted) Unwrap 

func (e *SagaAborted) Unwrap() error

type Sampling 

type Sampling struct {
	Temperature *float64 // 0..2 (OpenAI) / 0..1 (Anthropic); determinism at 0
	TopP        *float64 // nucleus sampling
	MaxTokens   *int     // cap on generated tokens; overrides the adapter's construction default
	Stop        []string // stop sequences
	Seed        *int64   // best-effort determinism (OpenAI; ignored where unsupported)
}

Sampling holds provider-neutral generation controls. Pointer fields distinguish "unset" (nil → use the provider/model default) from an explicit value — so a deliberate Temperature of 0 is not confused with "not specified". Each adapter maps the set fields onto its wire format and ignores those it doesn't support (e.g. Anthropic has no Seed). Set it once with Agent.WithSampling.

type SamplingOption 

type SamplingOption func(*Sampling)

SamplingOption sets one field of the Sampling config; see Temperature, TopP, MaxTokens, Stop, Seed.

func MaxTokens 

func MaxTokens(v int) SamplingOption

MaxTokens caps generated tokens, overriding the model adapter's construction default.

func Seed 

func Seed(v int64) SamplingOption

Seed sets a best-effort determinism seed (honored by providers that support it).

func Stop 

func Stop(seqs ...string) SamplingOption

Stop sets stop sequences.

func Temperature 

func Temperature(v float64) SamplingOption

Temperature sets the sampling temperature (0 = most deterministic).

func TopP 

func TopP(v float64) SamplingOption

TopP sets nucleus-sampling top-p.

type ScriptedModel 

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

ScriptedModel is a deterministic, LLM-free Model that replays a fixed script of turns, one turn per model call. It is the batteries-included version of the model fakes people otherwise reverse-engineer from ToolCallDelta + Finish for tests, examples, and local dev, so you can drive the whole agent loop offline.

It is replay-safe: it selects the turn to emit by counting the assistant messages already in the request, not by an internal cursor. On a durable resume the agent loop replays journaled assistant turns without calling the model, so a cursor would drift; counting the transcript instead lines the script back up with the live turn every time, including after a crash and re-run. Once the script is exhausted it repeats its last turn, so an over-long conversation does not panic.

Build turns with ToolTurn, TextTurn, and ErrorTurn: 

m := agent.NewScriptedModel(
    agent.ToolTurn("call-1", "lookup", `{"q":"weather"}`),
    agent.TextTurn("it is sunny"),
)
a := agent.New(m, agent.NewMemStore(), lookupTool)
out, err := a.Run(ctx, "run-1", "what's the weather?")

A ScriptedModel is safe for the sequential agent loop; it is not intended for concurrent Stream calls from multiple runs at once.

func NewScriptedModel 

func NewScriptedModel(turns ...ScriptedTurn) *ScriptedModel

NewScriptedModel builds a ScriptedModel that replays turns in order, one per model call.

func (*ScriptedModel) Stream 

func (m *ScriptedModel) Stream(_ context.Context, req Request) (*Stream, error)

Stream implements Model. It counts the assistant turns already in the request to pick the script entry to emit, so replay after a durable resume lines up with the live turn.

type ScriptedTurn 

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

ScriptedTurn is one scripted model output. Construct one with ToolTurn, TextTurn, or ErrorTurn rather than building it directly.

func ErrorTurn 

func ErrorTurn(err error) ScriptedTurn

ErrorTurn scripts a turn that fails the model call with err, so a test can exercise the loop's error path (a mid-run "crash" before the next step is journaled, say).

func TextTurn 

func TextTurn(text string) ScriptedTurn

TextTurn scripts a turn in which the model emits a final text answer, ending the run.

func ToolTurn 

func ToolTurn(id, name, args string) ScriptedTurn

ToolTurn scripts a turn in which the model calls one tool: id is the tool-use ID the journal keys the call under, name is the tool, and args is its JSON arguments.

type Session 

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

Session is a durable multi-turn conversation. Each Send is one full agent run (tools, resume, side-effect safety and all), seeded with the transcript so far — so the agent remembers earlier turns. The transcript is journaled turn-by-turn in the session's journal, so a Session reloaded (after a restart) from the same store resumes the conversation.

Layering: a turn runs under its own run ID ("<id>>@turn/<n>" for Send, "<id>>@event/<key>" for SendOnce, the key encoded), whose durable journal handles crash resume WITHIN the turn; the session's journal, "<id>>@session", records which message started each Send turn and each completed turn's (input, answer) so the transcript can be rebuilt. A root run ID may not contain '>', so no run started with Run shares a journal with a session, and IsSessionRun tells a session's run IDs apart from every other. Recover skips them: a turn is seeded with the transcript before it, which only the session holds, and its answer is recorded only by the session, so an unfinished turn resumes when its message is sent again (the same Send, or the redelivered SendOnce). Intermediate tool calls stay in the turn's journal and are NOT carried into later turns — the conversational memory is the question/answer transcript, not every tool call. Each turn also journals the transcript it started from (as "from/<turn run>"), so a turn resumed after a crash is seeded with exactly the turns its earlier model calls saw, even if other messages were answered in between.

A turn belongs to the message that started it: while a Send turn is unfinished, Send with a different message is ErrConfig rather than resuming that turn. Several handles on one session (a stale handle, or two workers) never lose a turn or answer one message with another's reply; a handle that finds the journal moved on reloads it.

A Session is safe for concurrent use: callers sharing one handle behave as callers on separate handles do, and their turns run in parallel.

func (*Session) History 

func (s *Session) History() []Message

History returns a copy of the conversation transcript so far (alternating user and final-assistant messages).

func (*Session) Send 

func (s *Session) Send(ctx context.Context, input string) (Message, error)

Send runs one conversation turn: the agent answers `input` with the full prior transcript in context, and the turn is journaled. Returns the assistant's answer.

If the turn pauses (a tool needs approval or Interrupt) or fails, Send returns that error (*ApprovalPending / *InterruptPending / ...) and does NOT advance the transcript; resolve it (Approve / AnswerInterrupt) and call Send again with the SAME input to resume that turn. Until then, Send with a different input is ErrConfig: the open turn belongs to its message.

func (*Session) SendOnce added in v0.6.0 

func (s *Session) SendOnce(ctx context.Context, key, input string) (Message, error)

SendOnce runs one conversation turn for an inbound message identified by key (an event or message id), at most once per key. A key whose turn already completed returns that turn's answer without running anything, so a redelivered message never opens a second turn, even if the process died after the turn was recorded and before the caller replied. A key whose turn was interrupted resumes that same turn: it runs under its own journal, "<session id>>@event/<key>" (the key encoded, so any key is allowed), so a different message arriving in between gets its own turn. Reusing a key with a different input is ErrConfig, whether the key's turn has finished or is still open: the turn's run records the message it answers (see RunStart).

func (*Session) Turns 

func (s *Session) Turns() int

Turns returns the number of completed turns.

type SignalPending added in v0.9.0 

type SignalPending struct {
	RunRef
	Name string
}

SignalPending is returned by Run when a tool called Await, AwaitFor or Receive and nothing has been delivered under Name yet (for Receive, Name is the channel). The run has paused durably at the await point. Deliver with Signal (or Enqueue for a channel), then re-invoke Run with RootRunID to continue. It is the externally-pushed dual of InterruptPending: Interrupt asks a human and resumes with their answer; Await waits for an event an outside system delivers.

func (*SignalPending) Error added in v0.9.0 

func (e *SignalPending) Error() string

Error names the run and the signal or channel it awaits.

type Sleeping deprecated 

type Sleeping = TimerPending

Sleeping is the former name of TimerPending.

Deprecated: transitional; renamed by the 1.0 rewrite. Use TimerPending.

type StepKind 

type StepKind string

StepKind records what a journal entry represents. 

const (
	StepModel      StepKind = "model"       // an assistant message from the model
	StepToolResult StepKind = "tool_result" // a completed tool call + its result
	StepValue      StepKind = "value"       // a user-authored durable step (see Step[T])
	StepSignal     StepKind = "signal"      // an external event delivered into a run (see Signal/Await)
	StepApproval   StepKind = "approval"    // a durable human approve/deny decision (HITL)
	StepAttempt    StepKind = "attempt"     // "about to run a non-retriable side effect" marker
	StepSagaFail   StepKind = "saga_fail"   // a saga step failed → durable abort trigger
	// StepNotStarted records that the attempt whose marker it names never called its side
	// effect, written by the driver that claimed it (see attempt.go). That attempt then no
	// longer halts a resume, and the effect is re-attempted under a new marker.
	StepNotStarted StepKind = "not_started"
	// StepHeader is the journal header: the first record of every run's journal, named "@journal",
	// whose Format names the journal format the run is written in (see JournalFormat). The
	// journal writes it before a run's first record and checks it before reading any other.
	StepHeader StepKind = "header"
)

type StepOption added in v0.6.0 

type StepOption func(*stepConfig)

StepOption configures Step.

func StepSafety added in v0.6.0 

func StepSafety(s Safety) StepOption

StepSafety declares how safe a step is to re-run, as Safety does for a tool. A step that is RetrySafe (ReadOnly, Idempotent, or keyed) re-runs after a crash; any other step halts.

type Store added in v0.9.0 

type Store interface {
	// Insert stores data under (runID, name) if no entry has that name, and returns the stored
	// entry (the caller's, or the one already there) and whether this call stored it.
	Insert(ctx context.Context, runID, name string, data []byte) (Entry, bool, error)
	// Get returns runID's entry named name, if there is one.
	Get(ctx context.Context, runID, name string) (Entry, bool, error)
	// Load yields runID's entries whose Seq is greater than after (-1 for all of them), in
	// ascending Seq. It yields an error at most once, as its last element.
	Load(ctx context.Context, runID string, after int64) iter.Seq2[Entry, error]
}

Store is the storage port: an append-only set of named entries per run. It stores bytes and knows nothing of what they mean; the Journal above it encodes records, memoizes steps, writes the format header and runs attempt claims. A backend implements Store and passes the storetest suite (agent/storetest), which checks every requirement below.

Requirements (A1 to A8):

  • A1, unique names and a linearizable insert: a run holds at most one entry per name. Concurrent Inserts of one name, from any number of processes, have exactly one winner, and every caller and every later reader sees the winner's bytes. Which entry a (runID, name) names does not depend on the context a method is called with: a store that scopes keys by a tenant carried in the context breaks the journal (the Journal remembers runs by run ID), so a tenant belongs in the run ID (see RunFilter.Prefix).
  • A2, commit-ordered, prefix-closed visibility: Seq is an opaque int64, strictly increasing in commit order within a run. Every read of a run returns a prefix of the run's final order: no entry ever becomes visible with a lower Seq than an entry already visible. Gaps in Seq are allowed. Nothing reads Seq as a count; positions are computed when a run is read, in Load order.
  • A3, durable before return: Insert reports inserted only after the entry is committed. On an error the entry is either absent or complete, and the same bytes may be inserted again.
  • A4, read-your-writes and monotone visibility: once visible, an entry stays visible with the same bytes and position (the one exception is a redaction, which keeps the position).
  • A5, byte fidelity: Data comes back exactly as inserted, and the caller may keep or modify the slices it passes and receives. That includes every record's salt: the engine tells its own model record from another driver's by the salt it drew (see JournalEntry), so a store that rewrites or drops salt bytes makes it take its own record for another's.
  • A6, immutable: the port has no update or delete. Only a redaction may replace an entry's Data, with a tombstone (see Record.Redacted), and only in a run that is over.
  • A7, context: every method honors ctx. Recording a step's outcome after ctx is cancelled is the Journal's job, not the store's.
  • A8, iterator hygiene: breaking out of Load's iterator releases every resource it holds, and a store holds no connection, transaction or lock across a yield, so a caller may Insert into the run inside its Load loop without deadlocking.

A store that wraps another store may implement 

Unwrap() Store

so Capability finds the capabilities of the store it wraps (Lister, Leaser), but only if it passes run IDs and names through unchanged. A wrapper that rewrites keys (a tenant prefix, say) must not implement Unwrap, and implements each capability itself; storetest.CheckWrapper checks this.

type Stream 

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

Stream is a live model response. Range over Events() for a streaming UI, OR call Message() to drain it into the fully-assembled assistant Message — not both (a Stream is consumed once).

A consumer that stops early releases the producer: breaking out of Events() does it, and so does Close. Cancelling the context passed to Model.Stream also releases it.

func NewStream 

func NewStream(ch <-chan Emit) *Stream

NewStream wraps an event channel the caller fills and closes. It suits producers that buffer their whole response up front. A producer that sends as it reads (a network adapter) should use NewStreamFunc, so it is released when the consumer stops reading; a goroutine blocked sending on ch is not.

func NewStreamFunc added in v0.6.0 

func NewStreamFunc(ctx context.Context, produce func(send func(Emit) bool)) *Stream

NewStreamFunc runs produce in its own goroutine and returns the Stream it feeds. produce delivers each event through send, which blocks until the consumer takes it and returns false once the consumer has stopped reading (it broke out of Events or called Close) or ctx is cancelled; produce should then return promptly, releasing whatever it holds, such as a response body. The stream ends when produce returns.

If ctx cancels an event the consumer has not yet taken, the stream ends with ctx's error, so a consumer never mistakes a response cut short by cancellation for a complete one.

func (*Stream) Close added in v0.6.0 

func (s *Stream) Close()

Close tells the producer the consumer will read no further events, so it can stop and release its resources. It is safe to call more than once, and after the stream ends.

func (*Stream) Events 

func (s *Stream) Events() iter.Seq2[Event, error]

Events returns a range-over-func iterator (Go 1.23+) over streamed events. Breaking out of the loop closes the stream. A stream that ends without a Finish event yields ErrIncompleteResponse last: the response stopped partway through the turn. A Finish is the turn's last event: an event after it yields ErrStreamProtocol and ends the stream.

func (*Stream) Message 

func (s *Stream) Message() (Message, Usage, error)

Message drains the stream and returns the assembled assistant Message + usage. When the stream fails, the usage is what it reported before failing, which the provider may still bill.

type Task 

type Task[T any] struct {
	Name string
	Fn   func(context.Context) (T, error)
	// Safety declares whether the task may re-run after a crash, as StepSafety does for a Step.
	// The zero value treats the task as a side effect: it runs at most once, and a crash after
	// its effect halts the resumed call with *OutcomeUnknown. Mark a check or a lookup ReadOnly.
	Safety Safety
}

Task is one unit of durable parallel work: a named step returning a T. Name is the durable memoization key within the run, so it must be unique across the tasks in one Parallel call. Because each task is a journaled step, a resumed run returns a completed task's recorded result without re-running it, and that result is an independently provable record (see audit.ProveStep).

type Text 

type Text struct {
	Text string `json:"text"`
}

Text is plain assistant/user text.

type TextDelta 

type TextDelta struct {
	Text string `json:"text"`
}

type TimerPending added in v0.9.0 

type TimerPending struct {
	RunRef
	Name   string
	FireAt time.Time
}

TimerPending is returned by Run when a tool called Sleep or WaitUntil and the wake time has not yet passed. The run has paused durably at the timer: its wake time is journaled, so the pause survives a restart. Re-invoke Run with RootRunID at or after FireAt to resume (a Waker does this automatically; otherwise the deployment re-invokes on its own schedule).

func (*TimerPending) Error added in v0.9.0 

func (e *TimerPending) Error() string

Error names the run, the timer, and its wake time.

type Tool 

type Tool interface {
	Name() string
	Description() string
	// ArgsSchema is the provider-NEUTRAL argument schema. The schema/ package emits
	// per-provider dialects (OpenAI-strict / Gemini / Anthropic) from it — so this is
	// never a single frozen blob handed straight to a provider.
	ArgsSchema() json.RawMessage
	// Safety declares how the tool may be retried on resume.
	Safety() Safety
	Call(ctx context.Context, args json.RawMessage) (json.RawMessage, error)
}

Tool is an action the agent can take. The interface is untyped (json.RawMessage) so heterogeneous tools share one type — including runtime MCP tools whose schema is only known at runtime and can't be a Go struct. Native tools get compile-time typing via Func (below); that's the primary ergonomic.

func CompensatedFunc 

func CompensatedFunc[In, Out any](
	name, description string,
	safety Safety,
	do func(context.Context, In) (Out, error),
	undo func(context.Context, In, Out) error,
) Tool

CompensatedFunc is a typed tool that declares both its forward action and its compensator. undo receives the same typed input and the output do produced: the call's recorded arguments decoded as the forward call decoded them, strictly (see Func). A record written before tool arguments decoded strictly may hold arguments only encoding/json accepts; the forward call of that time decoded them with encoding/json, so compensation does too, and undoes the value that call acted on. Arguments neither decodes are ErrProtocol.

The arguments are the ones the tool accepted. When a tool middleware changes a compensable call's arguments in a saga, the arguments the tool receives are journaled before it runs (under a key of their own), and compensation undoes those; otherwise the model's arguments are the ones the tool received. A journal written before accepted arguments were journaled has no such record, so compensation there falls back to the model's arguments. A middleware that rewrites a retry-safe compensable call's arguments must rewrite them the same way every time: the first record is kept when the call runs again.

func Func 

func Func[In, Out any](name, description string, safety Safety, fn func(context.Context, In) (Out, error)) Tool

Func wraps a typed Go function into a Tool. In is decoded from the args strictly, so the tool reads exactly what the model sent: a missing required field (one schema.For lists as required), null for a required field whose schema does not admit null, an unknown name or a case variant of a field's name, a duplicate name, data after the value, invalid UTF-8, or an escaped lone surrogate is ErrToolArgs, which goes back to the model as a tool error to correct. Empty args are the empty object. The return value is JSON-encoded. This is the compile-time-typed ergonomic: change In and the handler won't compile. The Tool interface itself stays untyped so a map of mixed tools (and runtime MCP tools) works.

Func panics, as New does for a missing model, if schema.For cannot describe In: such a type (a field reached through an embedded pointer to an unexported struct) could never be decoded from a call's arguments, so the tool would fail every call.

func RetrievalTool 

func RetrievalTool(r Retriever, k int, opts ...RetrievalOption) Tool

RetrievalTool exposes a Retriever as a tool the model can call to search on demand (agentic RAG): the model decides when to retrieve and with what query. It returns the top-k documents: a Retriever that returns more is cut to its first k. Read-only (retry-safe). The tool is named "retrieve" unless RetrievalName says otherwise; an agent's tools need distinct names, so give each its own when one agent searches several stores. It panics if k is below 1 or a name is empty.

Example

ExampleRetrievalTool shows bring-your-own-RAG: implement Retriever against your store, wire it in as a tool (or with WithRetrieval), and the loop handles the rest. 

// Your store, behind the Retriever port (here a trivial in-memory one).
docs := map[string]string{"france": "Paris is the capital of France."}
r := retrieverFunc(func(_ context.Context, query string, _ int) ([]Doc, error) {
	if d, ok := docs[strings.ToLower(query)]; ok {
		return []Doc{{Text: d}}, nil
	}
	return nil, nil
})

tool := RetrievalTool(r, 1)
out, _ := tool.Call(context.Background(), json.RawMessage(`{"query":"france"}`))
fmt.Println(string(out))

Output:
[{"text":"Paris is the capital of France."}]

func SubAgent 

func SubAgent(name, description string, sub *Agent) Tool

SubAgent wraps an agent as a Tool so a parent can delegate to it — the multi-agent primitive, as a durable tree rather than a fragile handoff.

The sub-run journals under a hierarchical, deterministic ID (SubRunID(parentRunID, toolUseID)), so give `sub` the SAME Durable store as the parent for a unified journal. Then a crash ANYWHERE in the tree resumes the whole tree precisely: completed sub-agents are reused, the in-flight one resumes from its own journal, and OutcomeUnknown / ApprovalPending from deep in the tree propagate up (approve, re-run the root, and it resumes down the path).

This is what the incumbents can't do: ADK/agenticenv can't recover sub-agents across a restart, and Eino doesn't unify nested state into the parent checkpoint.

type ToolCallDelta 

type ToolCallDelta struct {
	Index        int             `json:"index"`
	ID           string          `json:"id"`
	Name         string          `json:"name"`
	ArgsFragment json.RawMessage `json:"args_fragment"`
	Signature    string          `json:"signature"` // opaque provider token for this call, kept on ToolUse.Signature (Gemini thoughtSignature)
}

ToolCallDelta is an incremental fragment of a tool call, keyed by Index. Fragments are concatenated and gated by json.Valid before use (the UTF-8/partial-JSON safety the research flagged as universally missing).

type ToolChoice 

type ToolChoice struct {
	Mode string // "", "auto", "none", "required", or "tool"
	Name string // tool name to force; only meaningful when Mode == "tool"
}

ToolChoice controls whether and how the model may call tools on a request. Mode is one of "" or "auto" (model decides, the default), "none" (never call a tool), "required" (must call some tool), or "tool" (must call the specific tool named in Name). A nil *ToolChoice on a Request means the provider default (auto).

CAVEAT: in the multi-turn agent loop, forcing "required" or "tool" on EVERY turn prevents the model from ever emitting a final text answer, so the loop never terminates. Those two modes are intended for single-turn or typed/structured calls where exactly one tool round-trip is expected. Use "auto" (the default) for the loop.

type ToolCompleted 

type ToolCompleted struct {
	ToolUseID string          `json:"tool_use_id"`
	Name      string          `json:"name"`
	Result    json.RawMessage `json:"result"`
	IsError   bool            `json:"is_error"`
}

ToolCompleted carries a tool call's result. Emitted for live executions, for a human-denied call, and for each result reconstructed from the journal on resume.

type ToolHandler 

type ToolHandler func(context.Context, ToolUse) (json.RawMessage, error)

ToolHandler executes one tool call: given the ToolUse (ID, Name, Args), it returns the raw JSON result or an error. It is the tool-side analogue of ModelHandler.

type ToolMiddleware 

type ToolMiddleware func(ToolHandler) ToolHandler

ToolMiddleware wraps a ToolHandler — the same func(Handler) Handler idiom as model Middleware, but around TOOL execution. First added = outermost. A middleware can:

  • observe/log/trace a call (its name, args, duration, error);
  • MUTATE the outgoing args (validate, redact, inject defaults) by calling next with a modified ToolUse;
  • transform the result before it is journaled;
  • SHORT-CIRCUIT — return a result (a cache hit) or an error (a policy denial) WITHOUT calling next, so the tool never runs.

A middleware that may call next more than once (a retry) or not at all (a cache) must check the tool's Safety first, with ToolSafety(ctx): repeating or skipping a side effect is not its call to make. The agent enforces the first half itself: a tool that is not RetrySafe runs at most once per tool call, and a second call to next for it returns ErrToolReinvoked without running the tool.

The chain runs INSIDE the durable, memoized step, so a short-circuit result or a transformed result is what gets journaled — resume replays it and never re-runs the middleware or the tool. Batteries live in the middleware/ package (ToolLog, ToolCache).

type ToolResult 

type ToolResult struct {
	ToolUseID string          `json:"tool_use_id"`
	Result    json.RawMessage `json:"result,omitempty"`
	IsError   bool            `json:"is_error,omitempty"`
}

ToolResult carries a tool's output back to the model (tool turn).

type ToolStarted 

type ToolStarted struct {
	ToolUseID string          `json:"tool_use_id"`
	Name      string          `json:"name"`
	Args      json.RawMessage `json:"args"`
}

ToolStarted fires when a tool call begins executing: after any approval gate and, for non-idempotent tools, after the durable attempt marker is written, immediately before the tool is called. A call that does not start (cancelled after its attempt marker and before the call, and recorded as not started) emits neither ToolStarted nor ToolCompleted.

type ToolUse 

type ToolUse struct {
	ID        string          `json:"id"`
	Name      string          `json:"name"`
	Args      json.RawMessage `json:"args,omitempty"`
	Signature string          `json:"signature,omitempty"`
}

ToolUse is the model's request to call a tool (assistant turn). Signature is an opaque provider token bound to this call that must be sent back with it on later turns (a Gemini thoughtSignature); it is journaled with the call, and adapters that have no such token leave it empty and ignore it.

func FindToolCall added in v0.4.0 

func FindToolCall(recs []Record, toolUseID string) (idx int, call ToolUse, ok bool)

FindToolCall returns the journal index of the model turn that requested toolUseID, and the call itself (its name and arguments), or ok=false if no recorded turn requested it.

type TurnRestarted added in v0.8.0 

type TurnRestarted struct {
	Seq int `json:"seq"`
}

TurnRestarted fires when the model call for turn Seq starts over after an attempt that had already streamed ModelEvent deltas: a middleware such as Retry called the model again after that attempt failed, or delivered a different response in its place. The deltas received since TurnStarted{Seq} (or the previous TurnRestarted) belong to the discarded attempt and are not part of the recorded turn, so a consumer rendering the turn should clear them. Not emitted when the discarded attempt streamed nothing.

type TurnStarted 

type TurnStarted struct {
	Seq int `json:"seq"`
}

TurnStarted marks the beginning of a fresh model turn (Seq is the model-call sequence number within the run). Not emitted for turns replayed from the journal.

type Usage 

type Usage struct {
	// InputTokens is input tokens neither read from nor written to the prompt cache.
	InputTokens  int `json:"input_tokens,omitempty"`
	OutputTokens int `json:"output_tokens,omitempty"`
	// CacheReadTokens is input tokens served from the provider's prompt cache (billed at
	// a discount). CacheWriteTokens is input tokens written to the cache on this call
	// (Anthropic cache creation; 0 for providers that cache implicitly, like OpenAI).
	CacheReadTokens  int `json:"cache_read_tokens,omitempty"`
	CacheWriteTokens int `json:"cache_write_tokens,omitempty"`
}

Usage is token accounting for a call; middleware turns it into cost. The four counts are disjoint, whatever the provider: each input token is counted in exactly one of InputTokens, CacheReadTokens, or CacheWriteTokens, so each is billed once at its own rate and TotalInputTokens is the whole prompt. Providers differ on the wire (Anthropic reports uncached input separately; OpenAI and Gemini report a prompt total that includes cached tokens), and adapters normalize to this form.

func (Usage) TotalInputTokens added in v0.6.0 

func (u Usage) TotalInputTokens() int

TotalInputTokens is every input token the call processed, cached or not.

func (Usage) TotalTokens added in v0.6.0 

func (u Usage) TotalTokens() int

TotalTokens is every token the call processed: all input plus output.

func (Usage) Validate added in v0.8.0 

func (u Usage) Validate() error

Validate reports an error wrapping ErrNegativeUsage when any count in u is negative.

type Wake added in v0.9.0 

type Wake struct {
	RunID     string
	RootRunID string
	Name      string
	FireAt    time.Time
}

Wake is one scheduled wake: re-invoke RootRunID at or after FireAt. RunID and Name identify the timer (a sub-agent's timer lives in its sub-run, RunID); the top-level run is what a deployment re-invokes, since re-running it re-enters any sub-agent down to the timer.

type Waker 

type Waker interface {
	// Schedule registers w, idempotent per (w.RunID, w.Name): scheduling the same timer again
	// replaces its fire time. An error means the wake may not be registered; the run then fails
	// with an error wrapping ErrStorage, records nothing, and schedules again when re-driven.
	Schedule(ctx context.Context, w Wake) error
}

Waker is the pluggable trigger that re-invokes a sleeping run when its durable timer is due. Sleep registers a wake with the Waker bound to the run's context (WithWaker); the Waker later calls back to resume the run. The SDK provides the durable, at-most-once timer and its resume safety; what re-invokes the run at the wake time is deployment policy (an in-process loop, a cron, a queue), exactly as the inbound trigger for an event-driven run is (see docs/guides/messaging.md). MemWaker is the reference in-process implementation.

Source Files

View all Source files

Directories

Path Synopsis
Package agenttest holds test doubles for code built on package agent: a Store that counts its round trips (CountingStore).
 Package agenttest holds test doubles for code built on package agent: a Store that counts its round trips (CountingStore).
Package durabletest is the record-fidelity suite under its former name.
 Package durabletest is the record-fidelity suite under its former name.
Package storetest checks that an agent.Store meets the requirements the journal builds on (A1 to A8, documented on agent.Store), that a Journal over it writes and checks the journal format header, and that records round-trip byte for byte.
 Package storetest checks that an agent.Store meets the requirements the journal builds on (A1 to A8, documented on agent.Store), that a Journal over it writes and checks the journal format header, and that records round-trip byte for byte.

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.