engine

type Board ¶

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

Board is the engine execution blackboard: typed message channels plus untyped control vars, both protected by the same mutex.

Thread safety: every public method takes a mutex. Concurrent reads use RLock; mutations use Lock. The contract matches the original graph.Board it replaces — callers that previously held graph.Board across goroutines do not need to add any new locking.

Board is intentionally ignorant of agent concepts. It does not know what "messages", "answer" or "run id" mean; those names are established by callers. Per-execution metadata (ID, Attributes, Deps) belongs on Run, not here.

func NewBoard() *Board

func RestoreBoard(snap *BoardSnapshot) *Board

func (b *Board) AppendChannelMessage(name string, msg model.Message)

func (b *Board) AppendSliceVar(key string, value any) error

func (b *Board) Channel(name string) []model.Message

func (b *Board) ChannelsCopy() map[string][]model.Message

func (b *Board) GetVar(key string) (any, bool)

func (b *Board) GetVarString(key string) string

func (b *Board) RestoreFrom(snap *BoardSnapshot)

func (b *Board) SetChannel(name string, msgs []model.Message)

func (b *Board) SetVar(key string, value any)

func (b *Board) Snapshot() *BoardSnapshot

func (b *Board) UpdateSliceVarItem(key string, match func(any) bool, update func(any) any)

func (b *Board) Vars() map[string]any

type BoardSnapshot ¶

type BoardSnapshot struct {
	Vars     map[string]any             `json:"vars"`
	Channels map[string][]model.Message `json:"channels,omitempty"`
}

BoardSnapshot is a serialisable representation of board state, used for resume / checkpoint flows. It carries no live mutex and is safe to JSON-encode.

type Cause ¶

type Cause string

Cause classifies why a run was asked to stop. The agent layer maps these onto its higher-level commit/discard policy (e.g. discard the partial output on user_input, commit it on host_shutdown).

Engines should NEVER branch on Cause for control flow — Cause is metadata for the host, not a directive for the engine. The engine's only correct response to any cause is "stop cleanly and return".

const (
	// CauseUnknown is the zero value. Hosts should avoid sending it;
	// it exists so a zero-value Interrupt is recognisable.
	CauseUnknown Cause = ""

	// CauseUserCancel is a user-initiated cancel ("stop talking",
	// "abort this turn"). Output is typically discarded.
	CauseUserCancel Cause = "user_cancel"

	// CauseUserInput is a barge-in: the user spoke / typed and the
	// agent should yield to fresh input. Output is typically discarded.
	CauseUserInput Cause = "user_input"

	// CauseHostShutdown is a graceful shutdown from the host.
	// Output should typically be committed if any was produced.
	CauseHostShutdown Cause = "host_shutdown"

	// CauseCustom carries a host-defined cause in [Interrupt.Detail].
	CauseCustom Cause = "custom"
)

type Checkpoint ¶

type Checkpoint struct {
	// ExecID identifies the engine execution this checkpoint belongs
	// to. MUST equal the producing [Run.ID].
	ExecID string `json:"exec_id"`

	// Step is an opaque, engine-defined marker that locates "where"
	// the run is. For graph it is typically the next node id; for a
	// script engine it might be a continuation id. The host treats
	// this as opaque bytes.
	Step string `json:"step,omitempty"`

	// Iteration is an optional monotonic counter for engines that
	// loop (graph re-entry counter, scheduler tick, …). Zero is fine
	// when the engine doesn't track iterations.
	Iteration int `json:"iteration,omitempty"`

	// Board is the Board state at the boundary. Always non-nil.
	Board *BoardSnapshot `json:"board"`

	// Payload is engine-specific extra state the engine wants to
	// persist alongside the Board. Treated as opaque JSON by the
	// store; the producing engine is the only consumer that knows
	// how to decode it.
	Payload json.RawMessage `json:"payload,omitempty"`

	// Attributes mirrors [Run.Attributes] at the time the checkpoint
	// was produced (run id at the agent layer, tenant, graph id, …).
	// Stores may use these for indexing/lookup.
	Attributes map[string]string `json:"attributes,omitempty"`

	// Timestamp is the wall-clock time the engine produced the
	// checkpoint. Hosts may overwrite when they actually persist.
	Timestamp time.Time `json:"timestamp"`
}

Checkpoint is the engine-agnostic persistence record produced at a safe boundary during execution. Each engine decides what its own step marker / payload looks like; this struct only owns the common envelope shape.

Engines populate Checkpoint and hand it to [Checkpointer.Checkpoint] (the host method). The host is responsible for writing it durably; engines must not assume the call has persisted anything.

type CheckpointDeleter ¶

type CheckpointDeleter interface {
	Delete(ctx context.Context, execID string) error
}

CheckpointDeleter optionally extends CheckpointStore with the ability to delete a single execution's checkpoints. Used by the host when a run completes successfully and its checkpoints are no longer needed.

type CheckpointLister ¶

type CheckpointLister interface {
	List(ctx context.Context) ([]string, error)
}

CheckpointLister optionally extends CheckpointStore with the ability to enumerate persisted exec ids. Stores that support listing satisfy this interface; agent-level resume / dashboard code can type-assert to it.

type CheckpointStore ¶

type CheckpointStore interface {
	Save(ctx context.Context, cp Checkpoint) error
	Load(ctx context.Context, execID string) (*Checkpoint, error)
}

CheckpointStore is the host-side persistence contract. The host's [Checkpointer.Checkpoint] implementation typically delegates to a CheckpointStore. The interface is intentionally narrow: Save persists; Load returns the most-recent persisted record for the given exec id, or (nil, nil) if absent. All methods must be safe for concurrent use.

type Checkpointer ¶

type Checkpointer interface {
	Checkpoint(ctx context.Context, cp Checkpoint) error
}

Checkpointer persists a checkpoint at a safe boundary the engine has reached. The host decides whether to actually write; engines must not assume durability.

Hosts without configured checkpointing should make this a no-op (return nil) rather than an error so engines can call it unconditionally.

type Cloneable ¶

type Cloneable interface {
	Clone() any
}

Cloneable may be implemented by values stored in Board vars to provide a type-safe deep copy instead of the reflection fallback used by Board.Snapshot and RestoreBoard.

type Dependencies ¶

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

Dependencies is a typed dependency-injection container that an engine host (typically the agent runtime) populates at build time and that engine implementations consume at run time.

The container is keyed by an opaque any (use a typed key constant per dependency, never a bare string) to keep the lookup explicit and to discourage stringly-typed coupling. Concurrent reads after Build are safe; mutations through Set/Remove acquire a write lock so a host can adjust the container at any time.

This replaces graph.Dependencies / workflow.ToolDeps and intentionally has no engine-specific knowledge.

func NewDependencies() *Dependencies

func (d *Dependencies) Get(key any) (any, bool)

func (d *Dependencies) Has(key any) bool

func (d *Dependencies) Remove(key any)

func (d *Dependencies) Set(key, value any)

type Engine ¶

type Engine interface {
	Execute(ctx context.Context, run Run, host Host, board *Board) (*Board, error)
}

Engine is the deliberately thin contract every local execution engine satisfies so the agent layer can drive it through a uniform shape. Concrete engines (sdk/graph DAG executor, future script- based engines, …) usually expose richer APIs in addition to this method.

Contract:

• Execute MUST run to completion, until interrupted, or until ctx is cancelled.

• On clean completion, return the final Board (often the same pointer as the input — engines mutate in place by default) and a nil error.

• On cooperative interrupt (host sent through host.Interrupts()), return the (partial) Board together with the result of Interrupted. The error then satisfies errdefs.IsInterrupted(err) and can be destructured into an InterruptedError for the cause.

• On ctx cancellation, return the (partial) Board and ctx.Err().

• On any other failure, return the (partial) Board together with a domain error (preferably classified via errdefs). Returning a non-nil board on error lets the host decide whether to commit / discard / persist.

• When run.ResumeFrom is non-nil, Execute resumes from that checkpoint instead of starting fresh. See [Run.ResumeFrom] for the resume contract; engines that do not support resume MUST return an errdefs.NotAvailable-classified error rather than silently restarting.

Engines MUST NOT close any host-owned channel and MUST NOT mutate run.Attributes or run.ResumeFrom.

type EngineFunc ¶

type EngineFunc func(ctx context.Context, run Run, host Host, board *Board) (*Board, error)

EngineFunc adapts a plain function to the Engine interface. Useful for test doubles and trivial engines.

func (f EngineFunc) Execute(ctx context.Context, run Run, host Host, board *Board) (*Board, error)

type Host ¶

type Host interface {
	Publisher
	Interrupter
	UserPrompter
	Checkpointer
	UsageReporter
}

Host is the contract a runtime exposes to a running engine.

Host is intentionally a *composition* of small, single-method interfaces — Publisher, Interrupter, UserPrompter, Checkpointer. The composition exists to keep [Engine.Execute] readable; downstream code (graph nodes, tools, …) should depend on the smallest interface it actually needs:

// A pure-mapping node only emits events:
func (n *MapNode) Execute(ctx, board, pub engine.Publisher) error

// A streaming LLM node also wants the interrupt channel:
func (n *LLMNode) Execute(ctx, board,
    pub engine.Publisher, intr engine.Interrupter) error

Host implementations must be safe for concurrent use. The engine may invoke any method from any goroutine.

type Interrupt ¶

type Interrupt struct {
	Cause  Cause
	Detail string
}

Interrupt is the value the host sends through Host.Interrupts() to ask the running engine to stop. It is also the payload of the Interrupted error so the host can introspect why.

Interrupt is a plain value — copy it freely.

type InterruptedError ¶

type InterruptedError = interruptedErr

InterruptedError is the concrete error type returned by Interrupted. It is exported so hosts can use errors.As to destructure it; the preferred way to produce one is Interrupted, not direct construction.

Implements errdefs interrupted-marker so errdefs.IsInterrupted returns true on any error wrapping (or equal to) one of these.

type Interrupter ¶

type Interrupter interface {
	Interrupts() <-chan Interrupt
}

Interrupter exposes the host's cooperative-interrupt channel.

Engines select on the returned channel between steps:

select {
case intr := <-h.Interrupts():
    return engine.Interrupted(intr)
case <-ctx.Done():
    return ctx.Err()
default:
}

A nil channel means "no cooperative interrupt available"; engines should treat it as "never fires" — receiving on nil blocks forever, which is the correct semantic.

type NoopCheckpointStore ¶

type NoopCheckpointStore struct{}

NoopCheckpointStore drops every checkpoint and reports no state. Use as a default when checkpointing is not configured.

func (NoopCheckpointStore) Load(context.Context, string) (*Checkpoint, error)

func (NoopCheckpointStore) Save(context.Context, Checkpoint) error

type NoopHost ¶

type NoopHost struct{}

NoopHost is a zero-cost Host implementation that discards events, never reports interrupts, refuses user prompts, and skips checkpoints. It is meant for tests and embedded scenarios where an engine is invoked outside any real runtime.

func (NoopHost) AskUser(context.Context, UserPrompt) (UserReply, error)

func (NoopHost) Checkpoint(context.Context, Checkpoint) error

func (NoopHost) Interrupts() <-chan Interrupt

func (NoopHost) Publish(context.Context, event.Envelope) error

func (NoopHost) ReportUsage(context.Context, model.TokenUsage)

type Publisher ¶

type Publisher interface {
	Publish(ctx context.Context, env event.Envelope) error
}

Publisher emits a single event envelope.

Subject schema is NOT owned by this package: the host decides what the routing keys look like. Engines simply produce envelopes whose subject they construct from whatever convention their host has agreed with the consumer side.

Publish errors MUST NOT cause the producing engine to fail the run by default; the engine should log/record and continue. Backpressure or transport failures are an observability concern, not a control- flow signal.

type Run ¶

type Run struct {
	// ID is a unique identifier for this engine execution. Engines
	// use it as a correlation key in telemetry and may include it in
	// any subjects/headers their host's subject schema requires.
	//
	// The host generates ID and is responsible for keeping it stable
	// across resume / checkpoint cycles.
	ID string

	// Attributes carries arbitrary host-supplied metadata that should
	// flow into telemetry spans and event headers (run id at the
	// agent layer, tenant id, parent span links, engine kind for
	// observability, …).
	//
	// There is intentionally no dedicated field for "engine kind",
	// "run id (agent layer)" or similar: those are observability
	// concerns whose key conventions the host owns. If the consumer
	// side wants to filter by engine kind, the host sets
	// Attributes["engine_kind"] (or whatever key it agreed on).
	Attributes map[string]string

	// Deps is the typed dependency container the host has populated
	// for this run (LLM clients, tool registries, retrievers, …).
	// May be nil if the engine needs no dependencies; engines that
	// look up dependencies should use [GetDep] which handles nil.
	Deps *Dependencies

	// ResumeFrom, when non-nil, instructs the engine to continue
	// execution from the provided checkpoint instead of starting a
	// fresh run. The engine is the sole interpreter of
	// [Checkpoint.Step] and [Checkpoint.Payload]; the host treats
	// them as opaque.
	//
	// Contract:
	//
	//   - When ResumeFrom is non-nil the engine SHOULD prefer
	//     ResumeFrom.Board over the board parameter passed to
	//     [Engine.Execute]; passing both is allowed but the
	//     checkpoint's board takes precedence as it represents the
	//     state at the boundary the run paused on.
	//
	//   - When ResumeFrom.ExecID differs from [Run.ID] the engine MUST
	//     return an errdefs.Validation-classified error: forking a
	//     run requires a fresh execution id, not a resume.
	//
	//   - Engines that do not support resume MUST return an
	//     errdefs.NotAvailable-classified error when they observe a
	//     non-nil ResumeFrom rather than silently restarting from
	//     scratch.
	//
	// Hosts that drive resumption typically [CheckpointStore.Load]
	// the most recent checkpoint, set ResumeFrom, and call
	// [Engine.Execute] again with the same Run.ID.
	ResumeFrom *Checkpoint
}

Run is the per-execution input bundle an engine receives alongside the host. It is a plain data struct — no methods, no builder, no hidden state — assembled once by the host and passed to [Engine.Execute] read-only.

All fields are conceptually immutable for the duration of the run. Engines may read freely; they MUST NOT mutate the maps in place nor mutate the referenced ResumeFrom checkpoint.

func (r Run) Attribute(key string) string

type StreamDeltaPayload ¶

type StreamDeltaPayload struct {
	// Type discriminates the payload variant. See StreamDeltaType
	// constants for the standard values.
	Type StreamDeltaType `json:"type"`

	// Content carries the assistant text for "token" and the tool
	// output (typically already serialised) for "tool_result".
	Content string `json:"content,omitempty"`

	// ID is the tool-call identifier the model assigned. Set on
	// "tool_call" only; for "tool_result" use ToolCallID instead.
	ID string `json:"id,omitempty"`

	// Name is the tool name. Set on "tool_call"; recommended on
	// "tool_result" so consumers can correlate without a separate
	// dispatch table.
	Name string `json:"name,omitempty"`

	// Arguments is the tool input the model produced. Engines MAY
	// pass it as either a string (raw JSON) or an already-decoded
	// map / slice — consumers should accept both.
	Arguments any `json:"arguments,omitempty"`

	// ToolCallID pairs a "tool_result" with the originating
	// "tool_call". Required on tool_result.
	ToolCallID string `json:"tool_call_id,omitempty"`

	// IsError reports whether the tool dispatch returned an error
	// payload (vs. a successful result). Set on "tool_result".
	IsError bool `json:"is_error,omitempty"`

	// Cancelled reports whether this tool_result is a synthesised
	// cancellation (the call was never dispatched because the round
	// was interrupted). Set on "tool_result" only.
	Cancelled bool `json:"cancelled,omitempty"`
}

StreamDeltaPayload is the canonical decoded shape of a SubjectStreamDelta envelope's payload.

Engines MUST emit payloads that JSON-decode into this struct; the runtime constraint is checked by DecodeStreamDelta. Engines MAY add fields beyond what this struct lists — the JSON decoder is permissive on unknowns — but consumers SHOULD only rely on the fields documented here.

Per-Type field requirements:

Type           Required               Recommended
------------   --------------------   --------------------
token          Content                —
tool_call      ID, Name               Arguments
tool_result    ToolCallID, Content    Name, IsError, Cancelled

func DecodeStreamDelta(env event.Envelope) (StreamDeltaPayload, error)

type StreamDeltaType ¶

type StreamDeltaType string

StreamDeltaType enumerates the kinds of in-flight increments a stream envelope can carry. Engines MUST set [StreamDeltaPayload.Type] to one of these values; consumers SHOULD treat unknown values as forward- compatible additions and skip them.

const (
	// StreamDeltaToken is one piece of generated assistant text.
	// Required field: Content.
	StreamDeltaToken StreamDeltaType = "token"

	// StreamDeltaToolCall is one tool invocation the model just
	// requested. Required fields: ID, Name. Recommended: Arguments.
	StreamDeltaToolCall StreamDeltaType = "tool_call"

	// StreamDeltaToolResult is the outcome of one tool invocation —
	// either the actual result, or a synthesised cancellation when
	// the round was interrupted before the call dispatched.
	// Required fields: ToolCallID, Content. Recommended: Name,
	// IsError, Cancelled.
	StreamDeltaToolResult StreamDeltaType = "tool_result"
)

type UsageReporter ¶

type UsageReporter interface {
	ReportUsage(ctx context.Context, usage model.TokenUsage)
}

UsageReporter accepts incremental LLM token-usage reports an engine observes during a run. Each call adds delta usage; the host is responsible for accumulation, billing, and downstream telemetry.

Engines should call ReportUsage once per LLM invocation that returns usage metadata (typical: streaming nodes call it on completion with the per-call totals). Reports SHOULD be best-effort — engines must not let a slow or failing reporter block forward progress, and must not return an error to the run because of usage reporting.

Hosts without billing or token tracking should make this a no-op.

type UserPrompt ¶

type UserPrompt struct {
	Parts    []model.Part
	Schema   []byte
	Source   string
	Metadata map[string]string
}

UserPrompt describes what the engine is asking the host to relay to the end user. It deliberately stays one level below "chat message":

• Parts carries the multi-modal payload (text, image, audio, file, structured data) using model.Part — the same building block that model.Message uses, minus the chat-specific Role.
• Schema is an optional structured-input hint (JSON-schema-shaped bytes) for cases where the host wants to render a form or validate the response.
• Source identifies the engine step that produced the prompt; useful for trace correlation and resume.
• Metadata is free-form host-passed-through metadata.

Why []model.Part rather than model.Message: a Message also carries Role (system/user/assistant/tool), which is a chat-layer concept the engine has no business naming. Parts give us full multi-modality (image, audio, file, data) without tying the engine to chat semantics — the agent layer wraps Parts back into a Message with the right Role on its way out, and unwraps user-supplied Parts on the way in.

type UserPrompter ¶

type UserPrompter interface {
	AskUser(ctx context.Context, prompt UserPrompt) (UserReply, error)
}

UserPrompter lets an engine ask the host to prompt the end user (chat input, voice DTMF, structured form, …) and block until the reply arrives.

Hosts that don't expose user interaction should return an errdefs.NotAvailable-classified error. Engines that get such an error from a step that strictly needs user input should propagate it so the host can decide whether to fail or fall back.

type UserReply ¶

type UserReply struct {
	Parts    []model.Part
	Metadata map[string]string
}

UserReply is what the host returns to satisfy a UserPrompt. Like UserPrompt it uses []model.Part so the response can carry any modality the host's user interface produced — typed text, an uploaded image, recorded audio, a file attachment, structured form data, …