venat

README

Venat

Venat is a Go library for multi-agent workloads that need crash recovery, human approvals, and idempotent side effects. Its Runner records typed Run, Task, Event, Lease, Approval, and ActionAttempt state through configurable stores. Applications embed the runner and provide the production storage implementation.

Status: The latest release is v0.14.0. The public API may still change before v1.0.

Why Venat

Agent workloads must preserve state across crashes, prevent duplicate side effects, and record decisions that cross process or human boundaries. Venat handles those concerns in four concrete ways:

• Recovery: commands are persisted, leased, audited, and resumable; guarded side effects use an action-attempt ledger.
• Type safety: CI rejects exported []any results and loose any fields in the public API.
• Storage ownership: applications implement the storage contract; the kernel ships no production backend or domain schema.
• Embedding: Venat runs as a library in a normal Go program. It ships no UI or hosted service.

At a glance

• Strong bounded agent loop (agent/): one agent does one task well: step trace, schema repair, tool safety, context management, typed failure, and budget enforcement. Reusable skill/ instruction bundles support explicit and model-driven activation there; they do not grant host tools or create a second runtime.
• Explicit multi-agent scheduler (multiagent/): first-class AgentClass, Team, Scheduler, Dispatch, typed Handoff, Blackboard, Voting, and Supervisor, instead of ad-hoc helpers.
• Durable runner (root + internal/): runs, tasks, events, leases, approvals, an outbox / action-attempt ledger for idempotent side effects, and handoffs, all persisted and replayable.
• Workflow modeling (workflow/): a declarative Definition compiles to a multiagent graph; it adds no second runtime.
• Durable triggers (transport/cron, transport/webhook, and others): schedule and event entry points, with per-trigger timezone support on the cron driver.
• Product packs (packs/): vertical skeletons free to encode domain vocabulary the kernel never touches.

Install

go get github.com/Viking602/venat@latest

Requires Go 1.25+. Repository development and release gates pin Go 1.25.12.

Agent Skills

Venat supports the complete local Agent Skills lifecycle: trusted directory discovery, standards-compatible SKILL.md parsing, explicit activation, metadata-only catalog disclosure, model-driven activation, and bounded on-demand reads from scripts/, references/, and assets/.

found, err := skill.Discover(skill.DiscoveryOptions{
	ProjectDir:   projectRoot,
	TrustProject: true,
})
if err != nil {
	return err
}
skills := skill.NewRegistry()
for _, current := range found.Skills {
	if err := skill.Register(skills, current); err != nil {
		return err
	}
}
engine, err := agent.Build(agent.Spec{
	Model:           "gpt-5",
	Skills:          []string{"always-on-review"},
	AvailableSkills: []string{"pdf-processing"},
}, agent.BuildDeps{Providers: providers, Skills: skills})

Skills are injected immediately. AvailableSkills disclose only names and descriptions until the model calls the framework-owned activation tool. Project skills are ignored unless TrustProject is true; additional discovery roots are explicit trust grants. Bundled files are listed without being loaded, and reads are limited to the captured manifest. Skills never grant host tools and scripts are never executed automatically: allowed-tools remains advisory while Spec.Tools, the tool bus, hooks, and policy remain authoritative.

Run the complete local example with go run ./_examples/skills.

Quickstart

Queue a run on the default in-memory runner and read the append-only event stream:

package main

import (
	"context"
	"fmt"

	"github.com/Viking602/venat"
	"github.com/Viking602/venat/api"
)

func main() {
	runner := venat.NewDevelopment()

	run, err := runner.QueueRun(context.Background(), api.StartRunCommand{
		Request: "compare options for a Go research assistant",
	})
	if err != nil {
		panic(err)
	}

	events, err := runner.RunEvents(context.Background(), run.ID)
	if err != nil {
		panic(err)
	}
	fmt.Println(run.ID, len(events))
}

Override defaults via api.Config:

runner := venat.NewDevelopment(api.Config{
	PolicyEngine: policy.DenySideEffectsByDefault(),
})

Continue with the full quickstart, or run the workflow example for a declarative multi-step flow.

How it works

Venat is four layers. Their critical import seams are explicit and checked in CI: api/ imports no Venat package, agent/ never imports multiagent/, multiagent/ never imports the root Runner, worker/, or internal/, and the root facade never imports multiagent/. The shared stream/ package is intentionally available to multiagent/; it is a runtime-neutral collaboration primitive, not an upward dependency on Runner.

┌─────────────────────────────────────────────┐
│ Packs / Workflow / Examples                 │
│ research, customer-support, devops, aiops,  │
│ workflow modeling                           │
└─────────────────────────────────────────────┘
                    ↓
┌─────────────────────────────────────────────┐
│ Multi-Agent Layer  (multiagent/)            │
│ AgentClass, AgentInstance, Team, Scheduler, │
│ Dispatch, typed Handoff, Blackboard, Voting │
└─────────────────────────────────────────────┘
                    ↓
┌─────────────────────────────────────────────┐
│ Agent Loop Layer  (agent/)                  │
│ Step, OutputPolicy, ToolSafety,             │
│ ContextManager, AgentFailure, LoopPolicy    │
└─────────────────────────────────────────────┘
                    ↓
┌─────────────────────────────────────────────┐
│ Durable Runner  (root + internal/)          │
│ Run / Task / Event / Lease / Approval /     │
│ Outbox / ActionAttempt / Handoff stores     │
└─────────────────────────────────────────────┘

The durable runner is what makes the upper layers safe to operate. The root Runner façade exposes typed commands for the full lifecycle: QueueRun / RunEvents, task leasing (AcquireTaskExecution), human approvals (RequestApproval / DecideApproval with resume tokens), idempotent side effects (StartActionAttempt / CompleteActionAttempt), trace spans, the blackboard, and the task-envelope mailbox. Storage sits behind a contract (ADR-012); the default runner keeps everything in memory so you can start without provisioning anything, and you swap in your own backend for production by implementing the store interfaces.

Workflow layer

workflow/ is a modeling layer, not a runtime. A workflow.Definition (built fluently with New().Step().Then().Branch().Map()) compiles to a multiagent.CompiledGraph:

Definition ──Compile──▶ Compiled ──Scheduler()──▶ multiagent.Scheduler
                                └──Graph()──────▶ multiagent.CompiledGraph

Because it lowers to the existing graph, workflow execution still flows through multiagent.Scheduler decisions and multiagent.Dispatch values. It does not create a second durable runtime and does not bypass Runner-owned Run, Task, Event, Lease, Policy, or Outbox behavior. Engine is an in-process convenience over multiagent.Drive; for durable execution, supply a multiagent.Executor that persists each dispatch through the root Runner before running agent work.

Two constraints worth knowing:

• Branch conditions must be pure functions of api.TypedReport. The compiled graph may evaluate them during replay or recovery, so they must not read clocks, mutate state, call providers, or depend on process-local counters.
• flow / api.Flow (preset adapter metadata) is a separate concept from workflow/, despite sharing the same English word.

Examples

Examples live under _examples/ (the leading underscore keeps them out of go build ./...). Run one with go run ./_examples/<name>.

• incident_response: fan-out, blackboard, review, approval, and action in one flow.
• workflow: a declarative workflow compiled to a multi-agent graph.
• subagent: agent-as-tool delegation across models and providers.
• evaluation: the evaluation harness.

Browse the _examples directory for the complete set.

Documentation

• Quickstart: install and run the first task.
• Public API: exported contracts and stability.
• Documentation index: runtime concepts, extensions, compatibility, architecture, and the package map.
• Contributing: development workflow and verification gates.

Scope & status

By design the kernel ships no:

• Built-in storage backend
• Built-in memory backend
• Domain vocabulary
• UI / console
• Hosted observability backends
• Provider-specific deep integrations

These belong in application code or optional plugins.

Development

See CONTRIBUTING.md for coding standards and the architectural gates CI enforces (sentrux, check-public-any.sh, check-business-words.sh, check-import-boundaries.sh). The fast local gate is make verify; run make ci-local before substantial changes.

Documentation

Overview

Package venat is the root facade for the Venat runner. It owns construction and the Runner wrapper:

runner := venat.NewDevelopment()

Import api for public contracts such as Config, commands, store interfaces, policy requests, and Run/Task value types.

Public packages are grouped by stable extension concern:

• venat — Runner construction, methods, and error re-exports
• api — Config, commands, interfaces, Run/Task data contracts
• [agent] — agent engine and profile contracts
• [tool] — tool contract, effect types, and tooltest helpers
• [flow] — flow preset contracts
• [policy] — unified authorization contract
• [blackboard] — blackboard item and selector contracts
• [provider] — LLM provider drivers (anthropic, openai, scripted)
• [message] — shared message/content data types
• [hook] — pre/post-turn hook contracts
• [transport] — integration transports such as MCP
• [worker] — optional Runner-to-agent.Engine execution glue

Types under internal/ are implementation details. Core storage, mailbox, transition, command-handler, and replay internals are not public extension points and cannot be imported by consumers.

Package venat is the public façade for the Venat runner.

Import github.com/Viking602/venat/api for public contracts such as Config, commands, interfaces, and value types. The root package owns only construction, Runner methods, and sentinel error re-exports.

Index

• Variables
• func DefaultConfig() api.Config
• type Runner
  • func New(configs ...api.Config) *Runner
  • func NewDevelopment(configs ...api.Config) *Runner
  • func NewProduction(cfg api.Config) (*Runner, error)
  • func (r *Runner) AckEnvelope(ctx context.Context, cmd api.AckEnvelopeCommand) error
  • func (r *Runner) AcquireResourceClaims(ctx context.Context, request api.ResourceClaimRequest) (api.ResourceClaimDecision, error)
  • func (r *Runner) AcquireTaskExecution(ctx context.Context, cmd api.AcquireTaskExecutionCommand) (api.TaskExecutionLease, bool, error)
  • func (r *Runner) AcquireTaskExecutionWithClaims(ctx context.Context, cmd api.AcquireTaskExecutionCommand) (api.AcquireTaskExecutionResult, error)
  • func (r *Runner) ActiveLeaseCount(runID, taskID string) int
  • func (r *Runner) ActiveLeaseCountContext(ctx context.Context, runID, taskID string) int
  • func (r *Runner) AdvanceRun(ctx context.Context, cmd api.AdvanceRunCommand) (api.Run, error)
  • func (r *Runner) Agents() []api.AgentProfile
  • func (r *Runner) AppendEvent(ctx context.Context, event api.Event) error
  • func (r *Runner) AppendTaskExecutionEvent(ctx context.Context, cmd api.AppendTaskExecutionEventCommand) error
  • func (r *Runner) AppendUsage(ctx context.Context, record api.UsageRecord) error
  • func (r *Runner) Begin(ctx context.Context) (api.UnitOfWork, error)
  • func (r *Runner) Close(ctx context.Context) error
  • func (r *Runner) CompleteActionAttempt(ctx context.Context, cmd api.CompleteActionAttemptCommand) (api.ActionAttempt, error)
  • func (r *Runner) CreateTask(ctx context.Context, cmd api.CreateTaskCommand) (api.Task, error)
  • func (r *Runner) DeadLetter(ctx context.Context, cmd api.DeadLetterCommand) error
  • func (r *Runner) DecideApproval(ctx context.Context, cmd api.DecideApprovalCommand) error
  • func (r *Runner) DispatchTask(ctx context.Context, cmd api.DispatchTaskCommand) (api.TaskEnvelope, error)
  • func (r *Runner) DispatchTaskFanOut(ctx context.Context, cmd api.FanOutDispatchTaskCommand) ([]api.TaskEnvelope, error)
  • func (r *Runner) DrainResponseOutbox(ctx context.Context) (int, error)
  • func (r *Runner) EndTraceSpan(ctx context.Context, cmd api.EndTraceSpanCommand) error
  • func (r *Runner) EnforceToolResult(ctx context.Context, request api.ToolResultEnforcementRequest) (api.ToolResultEnforcementResult, error)
  • func (r *Runner) Events(runID string) []api.Event
  • func (r *Runner) ExecuteCommand(ctx context.Context, command api.Command) (any, error)
  • func (r *Runner) HeartbeatTaskExecution(ctx context.Context, cmd api.HeartbeatTaskExecutionCommand) error
  • func (r *Runner) InvokeTool(ctx context.Context, cmd api.ToolInvocation) (api.ToolInvocationResult, error)
  • func (r *Runner) ListActionAttempts(ctx context.Context, selector api.ActionAttemptSelector) ([]api.ActionAttempt, error)
  • func (r *Runner) ListAdmissionReservations(ctx context.Context, selector api.AdmissionReservationSelector) ([]api.AdmissionReservation, error)
  • func (r *Runner) ListAgentDefinitionSnapshots(ctx context.Context, selector api.AgentDefinitionSnapshotSelector) ([]api.AgentDefinitionSnapshot, error)
  • func (r *Runner) ListEnvelopes(ctx context.Context, runID string) ([]api.TaskEnvelope, error)
  • func (r *Runner) ListEvents(ctx context.Context, runID string) ([]api.Event, error)
  • func (r *Runner) ListMessages(ctx context.Context, runID string) ([]api.UserMessage, error)
  • func (r *Runner) ListQueuedMessages(ctx context.Context) ([]api.UserMessage, error)
  • func (r *Runner) ListResourceClaims(ctx context.Context, selector api.ResourceClaimSelector) ([]api.ResourceClaim, error)
  • func (r *Runner) ListTasks(ctx context.Context, runID string) ([]api.Task, error)
  • func (r *Runner) ListTraceSpans(ctx context.Context, runID string) ([]api.TraceSpan, error)
  • func (r *Runner) LoadAdmissionReservation(ctx context.Context, id string) (api.AdmissionReservation, error)
  • func (r *Runner) LoadAgentDefinitionSnapshot(ctx context.Context, definitionID, version string) (api.AgentDefinitionSnapshot, error)
  • func (r *Runner) LoadEnvelope(ctx context.Context, envelopeID string) (api.TaskEnvelope, error)
  • func (r *Runner) LoadMessage(ctx context.Context, runID, messageID string) (api.UserMessage, error)
  • func (r *Runner) LoadResourceClaim(ctx context.Context, id string) (api.ResourceClaim, error)
  • func (r *Runner) LoadRun(ctx context.Context, runID string) (api.Run, error)
  • func (r *Runner) LoadTask(ctx context.Context, runID, taskID string) (api.Task, error)
  • func (r *Runner) Mode() api.RuntimeMode
  • func (r *Runner) PendingResumeTokens(ctx context.Context, selector api.ResumeTokenSelector) ([]api.ResumeToken, error)
  • func (r *Runner) PreviewAdmission(ctx context.Context, request api.AdmissionRequest) (api.AdmissionDecision, error)
  • func (r *Runner) PublishResponse(ctx context.Context, cmd api.PublishResponseCommand) error
  • func (r *Runner) QueryUsage(ctx context.Context, selector api.UsageSelector) ([]api.UsageRecord, error)
  • func (r *Runner) QueueEnvelope(ctx context.Context, env api.TaskEnvelope) error
  • func (r *Runner) QueueMessage(ctx context.Context, message api.UserMessage) error
  • func (r *Runner) QueueRun(ctx context.Context, cmd api.StartRunCommand) (api.Run, error)
  • func (r *Runner) ReadyTasks(runID string) []api.Task
  • func (r *Runner) ReadyTasksContext(ctx context.Context, runID string) []api.Task
  • func (r *Runner) Recover(ctx context.Context, runID string) (api.Projection, error)
  • func (r *Runner) RecoverResumeToken(ctx context.Context, cmd api.RecoverResumeTokenCommand) (api.ResumeToken, error)
  • func (r *Runner) RegisterAgent(profile api.AgentProfile)
  • func (r *Runner) RegisterFlow(flow api.Flow) error
  • func (r *Runner) RegisterTool(tool api.Tool)
  • func (r *Runner) RegisterToolForInvocation(runID, taskID string, holderType api.HolderType, holderID string, ...)
  • func (r *Runner) ReleaseTaskExecution(ctx context.Context, cmd api.ReleaseTaskExecutionCommand) error
  • func (r *Runner) RemoveToolsForInvocation(runID, taskID string, holderType api.HolderType, holderID string)
  • func (r *Runner) Replay(runID string, mode api.ReplayMode) (api.Projection, error)
  • func (r *Runner) ReplayContext(ctx context.Context, runID string, mode api.ReplayMode) (api.Projection, error)
  • func (r *Runner) ReplayRunState(runID string) (api.Projection, error)
  • func (r *Runner) ReplayRunStateContext(ctx context.Context, runID string) (api.Projection, error)
  • func (r *Runner) RequestApproval(ctx context.Context, cmd api.RequestApprovalCommand) (api.ApprovalRequest, api.ResumeToken, error)
  • func (r *Runner) RequestHandoff(ctx context.Context, cmd api.HandoffCommand) error
  • func (r *Runner) ReserveAdmission(ctx context.Context, request api.AdmissionRequest) (api.AdmissionDecision, error)
  • func (r *Runner) ResolveActionAttempt(ctx context.Context, cmd api.ResolveActionAttemptCommand) (api.ActionAttempt, error)
  • func (r *Runner) ResponseOutbox(runID string) []api.UserMessage
  • func (r *Runner) ResponseOutboxContext(ctx context.Context, runID string) []api.UserMessage
  • func (r *Runner) ResumeTokens() map[string]api.ResumeToken
  • func (r *Runner) Run(ctx context.Context, runID string) (api.Run, error)
  • func (r *Runner) RunEvents(ctx context.Context, runID string) ([]api.Event, error)
  • func (r *Runner) RunTimeline(ctx context.Context, runID string) ([]api.RunTimelineItem, error)
  • func (r *Runner) SaveAgentDefinitionSnapshot(ctx context.Context, snapshot api.AgentDefinitionSnapshot) error
  • func (r *Runner) SaveRun(ctx context.Context, run api.Run) error
  • func (r *Runner) SaveTask(ctx context.Context, task api.Task) error
  • func (r *Runner) SaveTraceSpan(ctx context.Context, span api.TraceSpan) error
  • func (r *Runner) SelectItems(ctx context.Context, runID string, selector api.BlackboardSelector) ([]api.BlackboardItem, error)
  • func (r *Runner) SetMessagePolicy(policy api.MessagePolicyChecker)
  • func (r *Runner) SetOutputGateway(gateway api.OutputGateway)
  • func (r *Runner) SetPipeline(components api.PipelineComponents)
  • func (r *Runner) SetPolicyEngine(policy api.PolicyEngine)
  • func (r *Runner) StartActionAttempt(ctx context.Context, cmd api.StartActionAttemptCommand) (api.ActionAttempt, error)
  • func (r *Runner) StartRun(ctx context.Context, cmd api.StartRunCommand) (api.Run, api.Task, error)
  • func (r *Runner) StartRunWithResult(ctx context.Context, cmd api.StartRunCommand) (api.StartRunResult, error)
  • func (r *Runner) StartTraceSpan(ctx context.Context, cmd api.StartTraceSpanCommand) (api.TraceSpan, error)
  • func (r *Runner) StoreCapabilities(ctx context.Context) (api.StoreCapabilities, error)
  • func (r *Runner) StoreProvider() api.StoreProvider
  • func (r *Runner) SubmitResponseOutput(ctx context.Context, cmd api.SubmitResponseOutputCommand) error
  • func (r *Runner) SubmitTypedReport(ctx context.Context, cmd api.SubmitTypedReportCommand) error
  • func (r *Runner) SubmitUserInput(ctx context.Context, cmd api.SubmitUserInputCommand) error
  • func (r *Runner) Subscribe(ctx context.Context, runID string, filter api.BlackboardFilter) (<-chan api.BlackboardItem, func() error, error)
  • func (r *Runner) SumUsageCredits(ctx context.Context, selector api.UsageSelector) (int64, error)
  • func (r *Runner) Task(ctx context.Context, runID, taskID string) (api.Task, error)
  • func (r *Runner) TraceSpans(runID string) []api.TraceSpan
  • func (r *Runner) TransitionAdmission(ctx context.Context, transition api.AdmissionTransition) (api.AdmissionDecision, error)
  • func (r *Runner) TransitionResourceClaims(ctx context.Context, request api.ResourceClaimTransitionRequest) (api.ResourceClaimDecision, error)
  • func (r *Runner) TransitionRun(ctx context.Context, cmd api.TransitionRunCommand) error
  • func (r *Runner) TransitionTask(ctx context.Context, cmd api.TransitionTaskCommand) error
  • func (r *Runner) UpdateEnvelope(ctx context.Context, env api.TaskEnvelope) error
  • func (r *Runner) UpdateMessage(ctx context.Context, message api.UserMessage) error
  • func (r *Runner) WaitForBlackboard(ctx context.Context, runID string, filter api.BlackboardFilter, ...) ([]api.BlackboardItem, error)
  • func (r *Runner) WriteItem(ctx context.Context, item api.BlackboardItem) error

Variables

var (
	ErrNotFound                = api.ErrNotFound
	ErrTerminalState           = api.ErrTerminalState
	ErrStaleTaskVersion        = api.ErrStaleTaskVersion
	ErrLeaseHolderMismatch     = api.ErrLeaseHolderMismatch
	ErrLeaseNotActive          = api.ErrLeaseNotActive
	ErrOwnerMismatch           = api.ErrOwnerMismatch
	ErrActionTaskRequired      = api.ErrActionTaskRequired
	ErrActionReconcileRequired = api.ErrActionReconcileRequired
	ErrIdempotencyConflict     = api.ErrIdempotencyConflict
	ErrResponseTaskRequired    = api.ErrResponseTaskRequired
	ErrPolicyDenied            = api.ErrPolicyDenied
	ErrPolicyObligationFailed  = api.ErrPolicyObligationFailed
	ErrHandoffCycle            = api.ErrHandoffCycle
	ErrHandoffDepthExceeded    = api.ErrHandoffDepthExceeded
	ErrInvalidCommand          = api.ErrInvalidCommand
	ErrInvalidTransition       = api.ErrInvalidTransition
	ErrCompletionCriteriaUnmet = api.ErrCompletionCriteriaUnmet
	ErrDependencyUnmet         = api.ErrDependencyUnmet
	ErrDependencyFailed        = api.ErrDependencyFailed
	ErrInvalidConfiguration    = api.ErrInvalidConfiguration
	ErrInvalidAddress          = api.ErrInvalidAddress
	ErrNoRecipients            = api.ErrNoRecipients
	ErrSubscriptionClosed      = api.ErrSubscriptionClosed
	ErrWaitTimeout             = api.ErrWaitTimeout
)

Public sentinel errors are owned by api and re-exported here for callers that only need root-level construction and error checks.

Functions

func DefaultConfig

func DefaultConfig() api.Config

DefaultConfig returns an empty api.Config; useful as a baseline before overriding individual fields.

Types

type Runner

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

Runner is the public façade over the internal runtime. All public contract values crossing this boundary use api package types, not internal/core types.

func New

func New(configs ...api.Config) *Runner

New constructs an in-memory development runner.

Deprecated: use NewDevelopment for local/test use or NewProduction for a runner that rejects missing durable storage and policy dependencies.

func NewDevelopment

func NewDevelopment(configs ...api.Config) *Runner

NewDevelopment constructs a runner with in-memory storage and allow-all policy defaults. It is intended for tests, examples, and local development.

func NewProduction

func NewProduction(cfg api.Config) (*Runner, error)

NewProduction constructs a runner only when the host supplies both durable storage and an explicit policy. The framework cannot verify whether a StoreProvider is durable, so production ownership remains with the host.

func (*Runner) AckEnvelope

func (r *Runner) AckEnvelope(ctx context.Context, cmd api.AckEnvelopeCommand) error

func (*Runner) AcquireResourceClaims

func (r *Runner) AcquireResourceClaims(ctx context.Context, request api.ResourceClaimRequest) (api.ResourceClaimDecision, error)

AcquireResourceClaims atomically acquires every requested opaque key or none.

func (*Runner) AcquireTaskExecution

func (r *Runner) AcquireTaskExecution(ctx context.Context, cmd api.AcquireTaskExecutionCommand) (api.TaskExecutionLease, bool, error)

func (*Runner) AcquireTaskExecutionWithClaims

func (r *Runner) AcquireTaskExecutionWithClaims(ctx context.Context, cmd api.AcquireTaskExecutionCommand) (api.AcquireTaskExecutionResult, error)

AcquireTaskExecutionWithClaims atomically acquires a task lease and every requested shared/exclusive resource claim.

func (*Runner) ActiveLeaseCount

func (r *Runner) ActiveLeaseCount(runID, taskID string) int

Deprecated: use ActiveLeaseCountContext.

func (*Runner) ActiveLeaseCountContext

func (r *Runner) ActiveLeaseCountContext(ctx context.Context, runID, taskID string) int

func (*Runner) AdvanceRun

func (r *Runner) AdvanceRun(ctx context.Context, cmd api.AdvanceRunCommand) (api.Run, error)

func (*Runner) Agents

func (r *Runner) Agents() []api.AgentProfile

func (*Runner) AppendEvent

func (r *Runner) AppendEvent(ctx context.Context, event api.Event) error

func (*Runner) AppendTaskExecutionEvent

func (r *Runner) AppendTaskExecutionEvent(ctx context.Context, cmd api.AppendTaskExecutionEventCommand) error

func (*Runner) AppendUsage

func (r *Runner) AppendUsage(ctx context.Context, record api.UsageRecord) error

AppendUsage persists one usage-metering record. The worker runtime calls it after every engine run; hosts may also append their own records (e.g. for non-engine model calls).

func (*Runner) Begin

func (r *Runner) Begin(ctx context.Context) (api.UnitOfWork, error)

func (*Runner) Close

func (r *Runner) Close(ctx context.Context) error

func (*Runner) CompleteActionAttempt

func (r *Runner) CompleteActionAttempt(ctx context.Context, cmd api.CompleteActionAttemptCommand) (api.ActionAttempt, error)

func (*Runner) CreateTask

func (r *Runner) CreateTask(ctx context.Context, cmd api.CreateTaskCommand) (api.Task, error)

func (*Runner) DeadLetter

func (r *Runner) DeadLetter(ctx context.Context, cmd api.DeadLetterCommand) error

func (*Runner) DecideApproval

func (r *Runner) DecideApproval(ctx context.Context, cmd api.DecideApprovalCommand) error

func (*Runner) DispatchTask

func (r *Runner) DispatchTask(ctx context.Context, cmd api.DispatchTaskCommand) (api.TaskEnvelope, error)

func (*Runner) DispatchTaskFanOut

func (r *Runner) DispatchTaskFanOut(ctx context.Context, cmd api.FanOutDispatchTaskCommand) ([]api.TaskEnvelope, error)

func (*Runner) DrainResponseOutbox

func (r *Runner) DrainResponseOutbox(ctx context.Context) (int, error)

func (*Runner) EndTraceSpan

func (r *Runner) EndTraceSpan(ctx context.Context, cmd api.EndTraceSpanCommand) error

func (*Runner) EnforceToolResult

func (r *Runner) EnforceToolResult(ctx context.Context, request api.ToolResultEnforcementRequest) (api.ToolResultEnforcementResult, error)

func (*Runner) Events

func (r *Runner) Events(runID string) []api.Event

Deprecated: use RunEvents so cancellation and storage errors are observable.

func (*Runner) ExecuteCommand

func (r *Runner) ExecuteCommand(ctx context.Context, command api.Command) (any, error)

ExecuteCommand dispatches a Command through the internal command bus and returns the typed result. For most use cases prefer the typed methods (QueueRun, StartRun, RequestApproval, AcquireTaskExecution, ...) which avoid the result-type assertion and provide better compile-time signatures. ExecuteCommand exists for tools (replay, migration, admin) that operate generically over Commands.

Result types follow the api.<Command>Result naming convention where a command produces a structured value (StartRunCommand -> api.StartRunResult, RequestApprovalCommand -> api.RequestApprovalResult, AcquireTaskExecutionCommand -> api.AcquireTaskExecutionResult). Commands that produce a single domain value return that value directly.

Deprecated: use the typed Runner methods so result shapes are checked at compile time.

func (*Runner) HeartbeatTaskExecution

func (r *Runner) HeartbeatTaskExecution(ctx context.Context, cmd api.HeartbeatTaskExecutionCommand) error

func (*Runner) InvokeTool

func (r *Runner) InvokeTool(ctx context.Context, cmd api.ToolInvocation) (api.ToolInvocationResult, error)

func (*Runner) ListActionAttempts

func (r *Runner) ListActionAttempts(ctx context.Context, selector api.ActionAttemptSelector) ([]api.ActionAttempt, error)

func (*Runner) ListAdmissionReservations

func (r *Runner) ListAdmissionReservations(ctx context.Context, selector api.AdmissionReservationSelector) ([]api.AdmissionReservation, error)

ListAdmissionReservations lists durable reservations matching selector.

func (*Runner) ListAgentDefinitionSnapshots

func (r *Runner) ListAgentDefinitionSnapshots(ctx context.Context, selector api.AgentDefinitionSnapshotSelector) ([]api.AgentDefinitionSnapshot, error)

ListAgentDefinitionSnapshots lists immutable definition revisions matching selector.

func (*Runner) ListEnvelopes

func (r *Runner) ListEnvelopes(ctx context.Context, runID string) ([]api.TaskEnvelope, error)

func (*Runner) ListEvents

func (r *Runner) ListEvents(ctx context.Context, runID string) ([]api.Event, error)

func (*Runner) ListMessages

func (r *Runner) ListMessages(ctx context.Context, runID string) ([]api.UserMessage, error)

func (*Runner) ListQueuedMessages

func (r *Runner) ListQueuedMessages(ctx context.Context) ([]api.UserMessage, error)

func (*Runner) ListResourceClaims

func (r *Runner) ListResourceClaims(ctx context.Context, selector api.ResourceClaimSelector) ([]api.ResourceClaim, error)

ListResourceClaims lists durable claims matching selector.

func (*Runner) ListTasks

func (r *Runner) ListTasks(ctx context.Context, runID string) ([]api.Task, error)

func (*Runner) ListTraceSpans

func (r *Runner) ListTraceSpans(ctx context.Context, runID string) ([]api.TraceSpan, error)

func (*Runner) LoadAdmissionReservation

func (r *Runner) LoadAdmissionReservation(ctx context.Context, id string) (api.AdmissionReservation, error)

LoadAdmissionReservation loads one durable aggregate-capacity reservation.

func (*Runner) LoadAgentDefinitionSnapshot

func (r *Runner) LoadAgentDefinitionSnapshot(ctx context.Context, definitionID, version string) (api.AgentDefinitionSnapshot, error)

LoadAgentDefinitionSnapshot loads one immutable definition revision.

func (*Runner) LoadEnvelope

func (r *Runner) LoadEnvelope(ctx context.Context, envelopeID string) (api.TaskEnvelope, error)

func (*Runner) LoadMessage

func (r *Runner) LoadMessage(ctx context.Context, runID, messageID string) (api.UserMessage, error)

func (*Runner) LoadResourceClaim

func (r *Runner) LoadResourceClaim(ctx context.Context, id string) (api.ResourceClaim, error)

LoadResourceClaim loads one durable resource claim.

func (*Runner) LoadRun

func (r *Runner) LoadRun(ctx context.Context, runID string) (api.Run, error)

func (*Runner) LoadTask

func (r *Runner) LoadTask(ctx context.Context, runID, taskID string) (api.Task, error)

func (*Runner) Mode

func (r *Runner) Mode() api.RuntimeMode

Mode reports whether the runner was created with development defaults or validated production dependencies.

func (*Runner) PendingResumeTokens

func (r *Runner) PendingResumeTokens(ctx context.Context, selector api.ResumeTokenSelector) ([]api.ResumeToken, error)

PendingResumeTokens lists unconsumed resume tokens matching selector. After a crash, a host enumerates pending tokens with this and recovers each via RecoverResumeToken — the bulk-recovery entry point.

func (*Runner) PreviewAdmission

func (r *Runner) PreviewAdmission(ctx context.Context, request api.AdmissionRequest) (api.AdmissionDecision, error)

PreviewAdmission evaluates aggregate capacity without reserving it.

func (*Runner) PublishResponse

func (r *Runner) PublishResponse(ctx context.Context, cmd api.PublishResponseCommand) error

func (*Runner) QueryUsage

func (r *Runner) QueryUsage(ctx context.Context, selector api.UsageSelector) ([]api.UsageRecord, error)

QueryUsage returns the usage records matching selector.

func (*Runner) QueueEnvelope

func (r *Runner) QueueEnvelope(ctx context.Context, env api.TaskEnvelope) error

func (*Runner) QueueMessage

func (r *Runner) QueueMessage(ctx context.Context, message api.UserMessage) error

func (*Runner) QueueRun

func (r *Runner) QueueRun(ctx context.Context, cmd api.StartRunCommand) (api.Run, error)

func (*Runner) ReadyTasks

func (r *Runner) ReadyTasks(runID string) []api.Task

Deprecated: use ReadyTasksContext.

func (*Runner) ReadyTasksContext

func (r *Runner) ReadyTasksContext(ctx context.Context, runID string) []api.Task

func (*Runner) Recover

func (r *Runner) Recover(ctx context.Context, runID string) (api.Projection, error)

func (*Runner) RecoverResumeToken

func (r *Runner) RecoverResumeToken(ctx context.Context, cmd api.RecoverResumeTokenCommand) (api.ResumeToken, error)

func (*Runner) RegisterAgent

func (r *Runner) RegisterAgent(profile api.AgentProfile)

func (*Runner) RegisterFlow

func (r *Runner) RegisterFlow(flow api.Flow) error

func (*Runner) RegisterTool

func (r *Runner) RegisterTool(tool api.Tool)

func (*Runner) RegisterToolForInvocation

func (r *Runner) RegisterToolForInvocation(runID, taskID string, holderType api.HolderType, holderID string, tool api.Tool)

RegisterToolForInvocation scopes governed tool metadata to one run, task, holder, and tool name. RegisterTool remains the legacy global registration API for direct non-agent callers.

func (*Runner) ReleaseTaskExecution

func (r *Runner) ReleaseTaskExecution(ctx context.Context, cmd api.ReleaseTaskExecutionCommand) error

func (*Runner) RemoveToolsForInvocation

func (r *Runner) RemoveToolsForInvocation(runID, taskID string, holderType api.HolderType, holderID string)

RemoveToolsForInvocation releases all scoped tool metadata for one exact invocation identity.

func (*Runner) Replay

func (r *Runner) Replay(runID string, mode api.ReplayMode) (api.Projection, error)

Deprecated: use ReplayContext.

func (*Runner) ReplayContext

func (r *Runner) ReplayContext(ctx context.Context, runID string, mode api.ReplayMode) (api.Projection, error)

func (*Runner) ReplayRunState

func (r *Runner) ReplayRunState(runID string) (api.Projection, error)

Deprecated: use ReplayRunStateContext.

func (*Runner) ReplayRunStateContext

func (r *Runner) ReplayRunStateContext(ctx context.Context, runID string) (api.Projection, error)

func (*Runner) RequestApproval

func (r *Runner) RequestApproval(ctx context.Context, cmd api.RequestApprovalCommand) (api.ApprovalRequest, api.ResumeToken, error)

func (*Runner) RequestHandoff

func (r *Runner) RequestHandoff(ctx context.Context, cmd api.HandoffCommand) error

func (*Runner) ReserveAdmission

func (r *Runner) ReserveAdmission(ctx context.Context, request api.AdmissionRequest) (api.AdmissionDecision, error)

ReserveAdmission atomically evaluates aggregate limits and records capacity.

func (*Runner) ResolveActionAttempt

func (r *Runner) ResolveActionAttempt(ctx context.Context, cmd api.ResolveActionAttemptCommand) (api.ActionAttempt, error)

func (*Runner) ResponseOutbox

func (r *Runner) ResponseOutbox(runID string) []api.UserMessage

Deprecated: use ResponseOutboxContext.

func (*Runner) ResponseOutboxContext

func (r *Runner) ResponseOutboxContext(ctx context.Context, runID string) []api.UserMessage

func (*Runner) ResumeTokens

func (r *Runner) ResumeTokens() map[string]api.ResumeToken

func (*Runner) Run

func (r *Runner) Run(ctx context.Context, runID string) (api.Run, error)

func (*Runner) RunEvents

func (r *Runner) RunEvents(ctx context.Context, runID string) ([]api.Event, error)

func (*Runner) RunTimeline

func (r *Runner) RunTimeline(ctx context.Context, runID string) ([]api.RunTimelineItem, error)

func (*Runner) SaveAgentDefinitionSnapshot

func (r *Runner) SaveAgentDefinitionSnapshot(ctx context.Context, snapshot api.AgentDefinitionSnapshot) error

SaveAgentDefinitionSnapshot persists one immutable definition revision.

func (*Runner) SaveRun

func (r *Runner) SaveRun(ctx context.Context, run api.Run) error

func (*Runner) SaveTask

func (r *Runner) SaveTask(ctx context.Context, task api.Task) error

func (*Runner) SaveTraceSpan

func (r *Runner) SaveTraceSpan(ctx context.Context, span api.TraceSpan) error

func (*Runner) SelectItems

func (r *Runner) SelectItems(ctx context.Context, runID string, selector api.BlackboardSelector) ([]api.BlackboardItem, error)

func (*Runner) SetMessagePolicy

func (r *Runner) SetMessagePolicy(policy api.MessagePolicyChecker)

func (*Runner) SetOutputGateway

func (r *Runner) SetOutputGateway(gateway api.OutputGateway)

func (*Runner) SetPipeline

func (r *Runner) SetPipeline(components api.PipelineComponents)

func (*Runner) SetPolicyEngine

func (r *Runner) SetPolicyEngine(policy api.PolicyEngine)

func (*Runner) StartActionAttempt

func (r *Runner) StartActionAttempt(ctx context.Context, cmd api.StartActionAttemptCommand) (api.ActionAttempt, error)

func (*Runner) StartRun

func (r *Runner) StartRun(ctx context.Context, cmd api.StartRunCommand) (api.Run, api.Task, error)

func (*Runner) StartRunWithResult

func (r *Runner) StartRunWithResult(ctx context.Context, cmd api.StartRunCommand) (api.StartRunResult, error)

StartRunWithResult starts a run and reports whether this call created it. Created is false for an idempotent retry that returned the existing run.

func (*Runner) StartTraceSpan

func (r *Runner) StartTraceSpan(ctx context.Context, cmd api.StartTraceSpanCommand) (api.TraceSpan, error)

func (*Runner) StoreCapabilities

func (r *Runner) StoreCapabilities(ctx context.Context) (api.StoreCapabilities, error)

func (*Runner) StoreProvider

func (r *Runner) StoreProvider() api.StoreProvider

func (*Runner) SubmitResponseOutput

func (r *Runner) SubmitResponseOutput(ctx context.Context, cmd api.SubmitResponseOutputCommand) error

func (*Runner) SubmitTypedReport

func (r *Runner) SubmitTypedReport(ctx context.Context, cmd api.SubmitTypedReportCommand) error

func (*Runner) SubmitUserInput

func (r *Runner) SubmitUserInput(ctx context.Context, cmd api.SubmitUserInputCommand) error

func (*Runner) Subscribe

func (r *Runner) Subscribe(ctx context.Context, runID string, filter api.BlackboardFilter) (<-chan api.BlackboardItem, func() error, error)

Subscribe streams future blackboard writes for runID that match filter. The subscription ends when the returned cancel func is called or when ctx is cancelled — either way the underlying subscription is released and the channel closes; items in flight at that moment may be dropped. cancel is idempotent and safe to call after ctx cancellation.

func (*Runner) SumUsageCredits

func (r *Runner) SumUsageCredits(ctx context.Context, selector api.UsageSelector) (int64, error)

SumUsageCredits returns the credit sum over records matching selector.

func (*Runner) Task

func (r *Runner) Task(ctx context.Context, runID, taskID string) (api.Task, error)

func (*Runner) TraceSpans

func (r *Runner) TraceSpans(runID string) []api.TraceSpan

Deprecated: use ListTraceSpans so cancellation and storage errors are observable.

func (*Runner) TransitionAdmission

func (r *Runner) TransitionAdmission(ctx context.Context, transition api.AdmissionTransition) (api.AdmissionDecision, error)

TransitionAdmission applies one expected-version reservation transition.

func (*Runner) TransitionResourceClaims

func (r *Runner) TransitionResourceClaims(ctx context.Context, request api.ResourceClaimTransitionRequest) (api.ResourceClaimDecision, error)

TransitionResourceClaims atomically applies expected-version claim changes.

func (*Runner) TransitionRun

func (r *Runner) TransitionRun(ctx context.Context, cmd api.TransitionRunCommand) error

func (*Runner) TransitionTask

func (r *Runner) TransitionTask(ctx context.Context, cmd api.TransitionTaskCommand) error

func (*Runner) UpdateEnvelope

func (r *Runner) UpdateEnvelope(ctx context.Context, env api.TaskEnvelope) error

func (*Runner) UpdateMessage

func (r *Runner) UpdateMessage(ctx context.Context, message api.UserMessage) error

func (*Runner) WaitForBlackboard

func (r *Runner) WaitForBlackboard(ctx context.Context, runID string, filter api.BlackboardFilter, predicate func([]api.BlackboardItem) bool, timeout time.Duration) ([]api.BlackboardItem, error)

func (*Runner) WriteItem

func (r *Runner) WriteItem(ctx context.Context, item api.BlackboardItem) error