exec

func NewAgentHandle ¶

func NewAgentHandle(id string) (*AgentHandle, error)

NewAgentHandle constructs an AgentHandle with both internal channels (`done` for the external AgentResult reader + `finished` close-only signal for internal lifecycle goroutines) already initialized. Use this in tests that construct standalone AgentHandles outside of RunAgentAsync - direct struct literals risk skipping the `finished` field, which would nil-channel-hang any future internal reader that waits on it. Returns ErrInvalidAgentHandleID when id is empty so callers see the misuse via the error path rather than a runtime panic. Production code should go through Orchestrator.RunAgentAsync, not NewAgentHandle directly.

func NewAgentRunner ¶

func NewAgentRunner(opts ...RunnerOption) *AgentRunner

NewAgentRunner constructs an AgentRunner with the given options. Callers that prefer struct-literal initialization may continue to use &AgentRunner{...} directly - both forms are supported. Stable.

func NewDefaultCoordRunner ¶

func NewDefaultCoordRunner(llm provider.LanguageModel, opts ...CoordOption) *AgentRunner

NewDefaultCoordRunner returns a pre-configured *AgentRunner suitable for use as a workflow coordinator. The returned runner has: - StepID = "coordinator" (matches Executor's reverse-reply inbox key) - Mailbox = a fresh in-memory MailboxStore (events from the executor land here; the runner's Run loop drains them on each wake) - Wake = a freshly allocated cap-1 buffered channel so the executor can signal mailbox-driven re-entry of the coord's goai tool loop after each lifecycle event push. AgentRunner.Run only enters mailbox-mode when BOTH Mailbox AND Wake are non-nil; without Wake the coord drains its mailbox once at Run-start and never re-enters to react to subsequent EventStepStart / EventStepEnd / EventError pushes. Stable. - Model = the supplied LanguageModel (may be nil if the caller is constructing a routing-only runner for tests) - Tools = the three defaults (forward_to_agent, narrate, finalize), plus any extras from WithCoordTools, minus narrate when SynthesizeOnly is set - SystemPrompt = DefaultCoordSystemPrompt - WakeContextProvider = supplied via WithCoordContextProvider, if any; nil disables per-wake context injection. The factory does NOT start runner.Run - caller owns the lifecycle.

func WithCleanupTimeout ¶

func WithCleanupTimeout(d time.Duration) CoordLoopOption

WithCleanupTimeout overrides the default cleanup wall-clock cap (DefaultCoordCleanupTimeout, 2s) used by the cleanup func returned from RunCoordinatorLoop. Useful in tests that want a sub-second cap to exercise the timer-fired branch quickly. Zero or negative values fall back to DefaultCoordCleanupTimeout. Stable.

func SynthesizeOnly ¶

func SynthesizeOnly() CoordOption

SynthesizeOnly returns a CoordOption that drops the `narrate` tool from the default coord tool set. The coord LLM still has forward_to_agent (so it can curate context for in-flight steps) and finalize (mandatory exit signal with optional synthesis text), but can no longer interject mid-workflow narrations. Use this for the CLI `--summary-only` mode where the user wants a single final synthesis rather than per-step commentary. Stable.

func WithCoordContextProvider ¶

func WithCoordContextProvider(fn func() string) CoordOption

WithCoordContextProvider returns a CoordOption that installs a callback the coord runner invokes before every GenerateText call (initial + each wake-driven re-entry). The callback's return string is appended as a fresh user-role message wrapped in <dynamic-context> tags so the LLM can distinguish ambient state from in-band conversation. Empty / whitespace-only returns are skipped so an idle provider does not pollute the message stream. Use this hook when a chat-driven UX needs ambient context refreshed every wake without re-engineering the system prompt: currently-open files, repo metadata, session topic, recent user actions. Pass nil to disable. The callback runs synchronously on the runner goroutine; keep it cheap (microseconds) and goroutine-safe; it must not call back into the runner. Stable.

func WithCoordMaxWakeCycles ¶

func WithCoordMaxWakeCycles(n int) CoordOption

WithCoordMaxWakeCycles returns a CoordOption that overrides the coord runner's wake-cycle cap. Zero AND negative values both keep coordDefaultMaxWakeCycles (100) - the factory only honours strictly- positive overrides (see NewDefaultCoordRunner: `if cfg.maxWakeCycles > 0`). The coord runner is long-lived; the per-Run cap limits wake- driven re-entries inside one Run invocation. Set higher (e.g. 250) for workflows with many long-running parallel steps + frequent send_message bridging that could exceed the 100 default in one Run window. Stable. There is intentionally no path through this option to the package- level `defaultMaxWakeCycles` (10) - the coord cap is set inside NewDefaultCoordRunner with `coordDefaultMaxWakeCycles` (100). Callers who want the smaller per-step cap should set the runner's MaxWakeCycles field directly after factory construction.

func WithCoordSystemPrompt ¶

func WithCoordSystemPrompt(prompt string) CoordOption

WithCoordSystemPrompt returns a CoordOption that overrides the default coord system prompt. Empty string is a no-op (default prompt is used). Exposed for consumers that want stricter or looser narration cadence than the defaults provide. The override fully replaces the default; callers who want to extend rather than replace should concatenate DefaultCoordSystemPrompt with extra text or use the convenience option WithCoordSystemPromptSuffix. Stable.

func WithCoordSystemPromptSuffix ¶

func WithCoordSystemPromptSuffix(extra string) CoordOption

WithCoordSystemPromptSuffix returns a CoordOption that appends extra guidance to the default coord system prompt. Convenience for the common "keep tested baseline + add integration-specific guidance" pattern (e.g. a chat consumer appending session-addressing hints, a reviewer agent appending policy reminders). Empty string is a no-op. Last-write-wins with WithCoordSystemPrompt: if both are supplied, whichever appears later in the option list takes effect. Stable.

func WithCoordTools ¶

func WithCoordTools(tools ...goai.Tool) CoordOption

WithCoordTools returns a CoordOption that APPENDS the supplied tools to the default coord tool set. Multiple WithCoordTools calls chain; each call's tools are appended after any previously accumulated tools in registration order. Stable. This is the SDK-flexibility hook: callers (CLI plugins, embedded SDK consumers, downstream embedders) can extend the coord with bespoke tools (e.g. a "request_user_input" tool that surfaces a prompt to the human, or a "consult_runbook" tool that looks up an SOP) without giving up the standard forward / narrate / finalize defaults. To replace the default tools entirely instead of appending, mutate the runner returned by NewDefaultCoordRunner directly:

runner := NewDefaultCoordRunner(llm)
runner.Tools = []goai.Tool{myReplacementSet...}

- but doing so removes finalize, so the Run loop will not exit without a manual replacement signal.

func NewDeliveryEngine ¶

func NewDeliveryEngine(source EngineActiveStepsSource, mailbox MailboxStore, registry EngineWakeRegistry, opts ...router.EngineOption) *DeliveryEngine

NewDeliveryEngine is re-exported from internal/router.

func NewChanWakeTarget ¶

func NewChanWakeTarget(ch chan struct{}) EngineWakeTarget

func BuildEvalContext ¶

func BuildEvalContext(results map[string]*StepResult) *EvalContext

BuildEvalContext creates an EvalContext from the executor's results map. Must be called with the executor's mu lock held.

func NewFactoryCache ¶

func NewFactoryCache(inner func(sessionID string) *Orchestrator) (*FactoryCache, error)

NewFactoryCache wraps inner so repeat calls for the same sessionID return the same Orchestrator. inner is invoked at most once per sessionID under normal conditions; concurrent first-callers may race, but only one winner is retained and the losers are Closed before being discarded. inner may return nil (e.g. when model resolution fails). A nil result is NOT cached - the next For call with the same sessionID retries construction. This matches the pre-cache "graceful degradation" behavior of both zenflowFactory closures. Returns ErrNilFactoryInner when inner is nil so callers see the misuse via the error path rather than a runtime panic.

func NewFileStorage ¶

func NewFileStorage(baseDir string) *FileStorage

NewFileStorage creates a new FileStorage that persists data under baseDir.

func NewInMemoryMailboxStore ¶

func NewInMemoryMailboxStore() *InMemoryMailboxStore

func LoadMCPConfig ¶ added in v0.2.0

func LoadMCPConfig(path string) (*MCPConfig, error)

LoadMCPConfig reads and parses an MCP settings file. A missing file returns an error satisfying errors.Is(err, os.ErrNotExist) so callers can treat "no config" as a no-op without a separate stat. A present but malformed file returns a parse error.

func WithMCPClientInfo ¶ added in v0.2.0

func WithMCPClientInfo(name, version string) MCPOption

WithMCPClientInfo sets the client name/version advertised to MCP servers during the initialize handshake. Defaults to "zenflow"/"1.0.0".

func WithMCPRequestTimeout ¶ added in v0.2.0

func WithMCPRequestTimeout(d time.Duration) MCPOption

WithMCPRequestTimeout bounds each MCP request, including the connect handshake and every subsequent tool call. Zero leaves goai's default (60s). Note: for stdio servers this does NOT bound the subprocess lifetime - that is governed by the context passed to ConnectMCPConfig and by MCPToolset.Close.

func WithMCPStderr ¶ added in v0.2.0

func WithMCPStderr(w io.Writer) MCPOption

WithMCPStderr routes the stderr of stdio MCP subprocesses to w (e.g. for verbose diagnostics). Default discards it.

func ConnectMCPConfig ¶ added in v0.2.0

func ConnectMCPConfig(ctx context.Context, cfg *MCPConfig, opts ...MCPOption) (*MCPToolset, error)

ConnectMCPConfig connects to every (enabled) server in cfg and returns a toolset aggregating their tools. Servers are processed in deterministic (sorted) order. A failure connecting to one server does not abort the others: the returned toolset contains the tools from all servers that connected, and the returned error is the join of per-server failures (nil when all succeed). Callers should use the toolset even on a non-nil error and always Close it.

For stdio servers, ctx governs the subprocess lifetime (goai starts them with exec.CommandContext): pass a context that stays valid for as long as the tools may be invoked, and rely on MCPToolset.Close - not context cancellation - for orderly shutdown. A short-lived/timeout context would kill the subprocess immediately after the handshake.

func NewMemoryStorage ¶

func NewMemoryStorage() *MemoryStorage

NewMemoryStorage creates an empty in-memory storage.

func NewMessageRouter ¶

func NewMessageRouter() *MessageRouter

func WithAdditionalTools ¶ added in v0.2.0

func WithAdditionalTools(tools ...goai.Tool) Option

WithAdditionalTools appends tools to the orchestrator catalog instead of replacing it. Unlike WithTools (which overwrites the whole slice), this composes onto whatever WithTools installed earlier in the option list - the canonical way to layer MCP-discovered tools (see ConnectMCPConfig) or plugin tools on top of a built-in set without rebuilding the slice. Stable.

func WithAgentHandleTTL ¶

func WithAgentHandleTTL(d time.Duration) Option

WithAgentHandleTTL bounds the start-to-finish wall-clock cap on a RunAgentAsync handle. When the TTL is exceeded the handle is force-completed with AgentError{Sentinel: ErrAgentHandleTimeout} and the inner context is cancelled. Zero or negative falls back to DefaultAgentHandleTTL (30m). The library does not consult any environment variables; CLI consumers wire ZENFLOW_AGENT_HANDLE_TTL (or any other source) to this option themselves. Stable.

func WithApproval ¶

func WithApproval(h ApprovalHandler) Option

WithApproval sets the plan approval handler for RunGoal. If set, the coordinator-generated workflow must be approved before execution. Stable.

package main

import (
	"context"
	"errors"

	"github.com/zendev-sh/goai/provider"
	"github.com/zendev-sh/zenflow"
)

// fakeModel is a minimal provider.LanguageModel that returns a canned
// response immediately. Used by examples so they compile and run
// without real API keys.
type fakeModel struct{ reply string }

func (m *fakeModel) ModelID() string { return "example-fake" }

func (m *fakeModel) DoGenerate(_ context.Context, _ provider.GenerateParams) (*provider.GenerateResult, error) {
	return &provider.GenerateResult{
		Text:         m.reply,
		FinishReason: provider.FinishStop,
		Usage:        provider.Usage{InputTokens: 5, OutputTokens: 5},
	}, nil
}

func (m *fakeModel) DoStream(_ context.Context, _ provider.GenerateParams) (*provider.StreamResult, error) {
	return nil, errors.New("streaming not implemented in fakeModel")
}

func main() {
	// autoApprove always approves - replace with an interactive prompt.
	orch := zenflow.New(
		zenflow.WithModel(&fakeModel{reply: "ok"}),
		zenflow.WithApproval(autoApprove{}),
	)
	defer orch.Close() //nolint:errcheck
	_ = orch
}

// autoApprove is a trivial ApprovalHandler that approves every plan.
type autoApprove struct{}

func (autoApprove) ApprovePlan(_ context.Context, _ *zenflow.Workflow) (bool, error) {
	return true, nil
}

Output:

func WithApprovalTimeout ¶

func WithApprovalTimeout(d time.Duration) Option

WithApprovalTimeout bounds how long ApprovalHandler.ApprovePlan may block. Must be applied AFTER WithApproval. Zero or negative is a no-op (no timeout, the default). On timeout, ApprovePlan returns (false, ErrApprovalTimeout) and RunGoal aborts cleanly. Stable.

package main

import (
	"context"
	"errors"
	"time"

	"github.com/zendev-sh/goai/provider"
	"github.com/zendev-sh/zenflow"
)

// fakeModel is a minimal provider.LanguageModel that returns a canned
// response immediately. Used by examples so they compile and run
// without real API keys.
type fakeModel struct{ reply string }

func (m *fakeModel) ModelID() string { return "example-fake" }

func (m *fakeModel) DoGenerate(_ context.Context, _ provider.GenerateParams) (*provider.GenerateResult, error) {
	return &provider.GenerateResult{
		Text:         m.reply,
		FinishReason: provider.FinishStop,
		Usage:        provider.Usage{InputTokens: 5, OutputTokens: 5},
	}, nil
}

func (m *fakeModel) DoStream(_ context.Context, _ provider.GenerateParams) (*provider.StreamResult, error) {
	return nil, errors.New("streaming not implemented in fakeModel")
}

// autoApprove is a trivial ApprovalHandler that approves every plan.
type autoApprove struct{}

func (autoApprove) ApprovePlan(_ context.Context, _ *zenflow.Workflow) (bool, error) {
	return true, nil
}

func main() {
	orch := zenflow.New(
		zenflow.WithModel(&fakeModel{reply: "ok"}),
		zenflow.WithApproval(autoApprove{}),
		zenflow.WithApprovalTimeout(30*time.Second),
	)
	defer orch.Close() //nolint:errcheck
	_ = orch
}

Output:

func WithCoordinator ¶

func WithCoordinator(runner *AgentRunner) Option

WithCoordinator installs a caller-provided AgentRunner as the workflow coordinator. When non-nil, the executor pushes lifecycle events (StepStart, StepEnd, error, etc.) as RouterMessages into runner.Mailbox under the coord step ID: runner.StepID when non-empty, otherwise the constant "coordinator". The runner's lifecycle is the caller's responsibility: the orchestrator does not call runner.Run, does not block on it, and does not check whether anyone drains the mailbox. CLI consumers typically construct an ephemeral coord runner via NewDefaultCoordRunner before RunFlow and dispose it after; embedded consumers reuse their existing primary AgentRunner so workflow events flow into the same chat history as user turns. Dual-ID convention: the "coord" identity is split across two independent string keys, both defaulting to "coordinator" but serving different roles: 1. The coord-runner Mailbox key - resolved by coordStepID(runner): runner.StepID if non-empty, else "coordinator". This is the key under which lifecycle events are Append-ed into runner.Mailbox. Embedded consumers typically pass runner.StepID="primary" so workflow events land in the same chat-history bucket as user turns. 2. The workflow-Router coordinator inbox key - the constant CoordRouterInboxID ("coordinator", in executor.go). Resumed steps reverse-route their replies into this inbox via Router.Send; drainCoordReverseReplies surfaces them as EventCoordinatorInboxMessage. This key is independent of the runner's StepID; even when the caller supplies a custom runner.StepID, reverse replies still flow through the Router's "coordinator" inbox unchanged. The executor's inbox auto-registration loop ALWAYS RegisterStep + RegisterInbox both keys when a Coordinator runner is installed, even when they coincide (de-duped), so a custom-StepID runner never causes resumed-step reverse replies to drop with DropReasonUnknownStep. Tested by TestExecutor_CustomStepIDAutoRegistersCoordRouterInbox. Pass nil to disable the coordinator entirely (workflow runs with no LLM-driven monitoring; messaging tools that target the coord still drop with an explicit reason). Not used by RunAgent - the primary agent IS the coordinator on that path. Stable.

func WithDefaultModel ¶

func WithDefaultModel(model string) Option

WithDefaultModel sets the fallback model name. Stable.

func WithDropCallback ¶

func WithDropCallback(fn func(DropEvent)) Option

WithDropCallback installs a user-supplied observer invoked once per dropped router message (in addition to the EventMessageDropped event already emitted via ProgressSink). Useful for metrics/alerting paths that don't want to subscribe to the full event firehose. The callback runs synchronously by default; set WithDropCallbackBufferSize to a positive value to dispatch asynchronously through a buffered channel (overflow falls back to synchronous so no drop event is itself silently lost). Stable.

func WithDropCallbackBufferSize ¶

func WithDropCallbackBufferSize(n int) Option

WithDropCallbackBufferSize selects the buffer size for asynchronous dispatch of WithDropCallback events. Zero or negative = synchronous dispatch (the default). Stable.

func WithExternalInbox ¶

func WithExternalInbox(ids ...string) Option

WithExternalInbox pre-registers one or more non-step sender inboxes on the Router at Run start. Use this for identities that send RouterMessages but are not workflow steps (e.g. "coordinator") and need to receive reverse-routed responses; notably the reverse message a resumed step sends back to its OriginalSender. Without pre-registration, Router.Send to an unknown non-step target drops as DropReasonUnknownStep. Idempotent and safe to pass the same ID multiple times. Stable.

package main

import (
	"context"
	"errors"

	"github.com/zendev-sh/goai/provider"
	"github.com/zendev-sh/zenflow"
)

// fakeModel is a minimal provider.LanguageModel that returns a canned
// response immediately. Used by examples so they compile and run
// without real API keys.
type fakeModel struct{ reply string }

func (m *fakeModel) ModelID() string { return "example-fake" }

func (m *fakeModel) DoGenerate(_ context.Context, _ provider.GenerateParams) (*provider.GenerateResult, error) {
	return &provider.GenerateResult{
		Text:         m.reply,
		FinishReason: provider.FinishStop,
		Usage:        provider.Usage{InputTokens: 5, OutputTokens: 5},
	}, nil
}

func (m *fakeModel) DoStream(_ context.Context, _ provider.GenerateParams) (*provider.StreamResult, error) {
	return nil, errors.New("streaming not implemented in fakeModel")
}

func main() {
	orch := zenflow.New(
		zenflow.WithModel(&fakeModel{reply: "ok"}),
		zenflow.WithExternalInbox("coordinator", "my-external-sender"),
	)
	defer orch.Close() //nolint:errcheck
	_ = orch
}

Output:

func WithForceModel ¶

func WithForceModel(model string) Option

WithForceModel overrides every per-agent and per-step Model identifier with the given model name during execution. Empty string disables the override (equivalent to leaving the option off). Precedence (high → low) for effective model resolution:

WithForceModel > Step.Model > AgentConfig.Model > WithDefaultModel

Use WithForceModel for cross-provider CLI overrides (e.g. running every step of a workflow through a single test provider regardless of YAML). For ordinary defaults, prefer WithDefaultModel - it lets per-agent and per-step Model declarations win. Stable.

func WithGoAIOptions ¶

func WithGoAIOptions(opts ...goai.Option) Option

WithGoAIOptions passes extra goai options to GenerateText calls. Stable.

func WithHoldTimeout ¶

func WithHoldTimeout(d time.Duration) Option

WithHoldTimeout bounds how long the executor retains a step in StepIdle while messages keep arriving. After the timeout, the executor force-terminates the step and emits one EventMessageDropped{reason: hold-timeout} per remaining mailbox message. Zero falls back to defaultHoldTimeout (30s). Stable.

func WithIsolation ¶

func WithIsolation(iso StepIsolation) Option

WithIsolation sets the step isolation strategy. When set, Setup is called before each step and Cleanup is deferred after. Stable.

func WithMailboxDelivery ¶

func WithMailboxDelivery() Option

WithMailboxDelivery enables the mailbox + delivery engine stack (the default). Use WithoutMailboxDelivery to suppress Router/Mailbox allocation in Executor.Run for tests that exercise the scheduler path without mailbox machinery. Stable.

func WithMailboxStore ¶

func WithMailboxStore(factory func() MailboxStore) Option

WithMailboxStore replaces the default InMemoryMailboxStore with a caller-supplied implementation (e.g. a sqlite/redis-backed store for multi-process workflows). The supplied factory is invoked once per workflow run so each Run gets a fresh store instance. Stable.

package main

import (
	"context"
	"errors"

	"github.com/zendev-sh/goai/provider"
	"github.com/zendev-sh/zenflow"
)

// fakeModel is a minimal provider.LanguageModel that returns a canned
// response immediately. Used by examples so they compile and run
// without real API keys.
type fakeModel struct{ reply string }

func (m *fakeModel) ModelID() string { return "example-fake" }

func (m *fakeModel) DoGenerate(_ context.Context, _ provider.GenerateParams) (*provider.GenerateResult, error) {
	return &provider.GenerateResult{
		Text:         m.reply,
		FinishReason: provider.FinishStop,
		Usage:        provider.Usage{InputTokens: 5, OutputTokens: 5},
	}, nil
}

func (m *fakeModel) DoStream(_ context.Context, _ provider.GenerateParams) (*provider.StreamResult, error) {
	return nil, errors.New("streaming not implemented in fakeModel")
}

func main() {
	// The factory is invoked once per RunFlow so each run gets a fresh store.
	factory := func() zenflow.MailboxStore { return zenflow.NewInMemoryMailboxStore() }

	orch := zenflow.New(
		zenflow.WithModel(&fakeModel{reply: "ok"}),
		zenflow.WithMailboxStore(factory),
	)
	defer orch.Close() //nolint:errcheck
	_ = orch
}

Output:

func WithMaxConcurrency ¶

func WithMaxConcurrency(n int) Option

WithMaxConcurrency sets the maximum parallel step count. Precedence: see `defaultMaxConcurrency` in executor.go - this option is level 2 (Workflow.Options.MaxConcurrency overrides it; if both this option and the YAML default are zero, the executor falls back to defaultMaxConcurrency=5). Stable.

package main

import (
	"context"
	"errors"

	"github.com/zendev-sh/goai/provider"
	"github.com/zendev-sh/zenflow"
)

// fakeModel is a minimal provider.LanguageModel that returns a canned
// response immediately. Used by examples so they compile and run
// without real API keys.
type fakeModel struct{ reply string }

func (m *fakeModel) ModelID() string { return "example-fake" }

func (m *fakeModel) DoGenerate(_ context.Context, _ provider.GenerateParams) (*provider.GenerateResult, error) {
	return &provider.GenerateResult{
		Text:         m.reply,
		FinishReason: provider.FinishStop,
		Usage:        provider.Usage{InputTokens: 5, OutputTokens: 5},
	}, nil
}

func (m *fakeModel) DoStream(_ context.Context, _ provider.GenerateParams) (*provider.StreamResult, error) {
	return nil, errors.New("streaming not implemented in fakeModel")
}

func main() {
	orch := zenflow.New(
		zenflow.WithModel(&fakeModel{reply: "ok"}),
		zenflow.WithMaxConcurrency(8), // allow 8 parallel steps
	)
	defer orch.Close() //nolint:errcheck
	_ = orch
}

Output:

func WithMaxDepth ¶

func WithMaxDepth(n int) Option

WithMaxDepth sets the maximum agent nesting depth. Applies to RunAgent only - RunFlow does not support child agent spawning. Stable.

package main

import (
	"context"
	"errors"

	"github.com/zendev-sh/goai/provider"
	"github.com/zendev-sh/zenflow"
)

// fakeModel is a minimal provider.LanguageModel that returns a canned
// response immediately. Used by examples so they compile and run
// without real API keys.
type fakeModel struct{ reply string }

func (m *fakeModel) ModelID() string { return "example-fake" }

func (m *fakeModel) DoGenerate(_ context.Context, _ provider.GenerateParams) (*provider.GenerateResult, error) {
	return &provider.GenerateResult{
		Text:         m.reply,
		FinishReason: provider.FinishStop,
		Usage:        provider.Usage{InputTokens: 5, OutputTokens: 5},
	}, nil
}

func (m *fakeModel) DoStream(_ context.Context, _ provider.GenerateParams) (*provider.StreamResult, error) {
	return nil, errors.New("streaming not implemented in fakeModel")
}

func main() {
	orch := zenflow.New(
		zenflow.WithModel(&fakeModel{reply: "ok"}),
		zenflow.WithMaxDepth(3),
	)
	defer orch.Close() //nolint:errcheck
	_ = orch
}

Output:

func WithMaxMailboxSize ¶

func WithMaxMailboxSize(n int) Option

WithMaxMailboxSize bounds the per-step in-memory mailbox queue. When the cap is exceeded, Append rejects the newest message and the router emits EventMessageDropped{reason: mailbox-full} via OnDrop. When this option is NOT applied, New installs DefaultMaxMailboxSize (10000) - finite by default so pathological producers cannot exhaust memory in long-running workflows. To opt out and run with an unbounded mailbox, pass WithMaxMailboxSize(0) explicitly: the zero is preserved (it sets an internal flag separate from the size value, so the default install does not overwrite it). Only takes effect when the default InMemoryMailboxStore is used (i.e. no WithMailboxStore override). Custom stores enforce their own caps. Stable.

package main

import (
	"context"
	"errors"

	"github.com/zendev-sh/goai/provider"
	"github.com/zendev-sh/zenflow"
)

// fakeModel is a minimal provider.LanguageModel that returns a canned
// response immediately. Used by examples so they compile and run
// without real API keys.
type fakeModel struct{ reply string }

func (m *fakeModel) ModelID() string { return "example-fake" }

func (m *fakeModel) DoGenerate(_ context.Context, _ provider.GenerateParams) (*provider.GenerateResult, error) {
	return &provider.GenerateResult{
		Text:         m.reply,
		FinishReason: provider.FinishStop,
		Usage:        provider.Usage{InputTokens: 5, OutputTokens: 5},
	}, nil
}

func (m *fakeModel) DoStream(_ context.Context, _ provider.GenerateParams) (*provider.StreamResult, error) {
	return nil, errors.New("streaming not implemented in fakeModel")
}

func main() {
	orch := zenflow.New(
		zenflow.WithModel(&fakeModel{reply: "ok"}),
		zenflow.WithMaxMailboxSize(50),
	)
	defer orch.Close() //nolint:errcheck
	_ = orch
}

Output:

func WithMaxTranscriptBytes ¶

func WithMaxTranscriptBytes(b int64) Option

WithMaxTranscriptBytes overrides the default per-step byte cap for the Day-1 InMemoryTranscriptStore. Zero or negative leaves the default `DefaultMaxTranscriptBytes`. Ignored when a custom store is supplied via WithTranscriptStore.

func WithMaxTranscriptMessages ¶

func WithMaxTranscriptMessages(n int) Option

WithMaxTranscriptMessages overrides the default per-step message count cap for the Day-1 InMemoryTranscriptStore. When exceeded, Append returns ErrTranscriptTooLarge and a subsequent resume on that step emits DropReasonTranscriptTooLarge. Zero or negative leaves the default `DefaultMaxTranscriptMessages`. Ignored when a custom store is supplied via WithTranscriptStore (custom stores enforce their own caps).

func WithMaxTurns ¶

func WithMaxTurns(n int) Option

WithMaxTurns sets the maximum conversation turns. Applies to RunAgent only - RunFlow uses per-agent MaxTurns from workflow YAML. Stable.

package main

import (
	"context"
	"errors"

	"github.com/zendev-sh/goai/provider"
	"github.com/zendev-sh/zenflow"
)

// fakeModel is a minimal provider.LanguageModel that returns a canned
// response immediately. Used by examples so they compile and run
// without real API keys.
type fakeModel struct{ reply string }

func (m *fakeModel) ModelID() string { return "example-fake" }

func (m *fakeModel) DoGenerate(_ context.Context, _ provider.GenerateParams) (*provider.GenerateResult, error) {
	return &provider.GenerateResult{
		Text:         m.reply,
		FinishReason: provider.FinishStop,
		Usage:        provider.Usage{InputTokens: 5, OutputTokens: 5},
	}, nil
}

func (m *fakeModel) DoStream(_ context.Context, _ provider.GenerateParams) (*provider.StreamResult, error) {
	return nil, errors.New("streaming not implemented in fakeModel")
}

func main() {
	orch := zenflow.New(
		zenflow.WithModel(&fakeModel{reply: "ok"}),
		zenflow.WithMaxTurns(10),
	)
	defer orch.Close() //nolint:errcheck
	_ = orch
}

Output:

func WithMaxWakeCycles ¶

func WithMaxWakeCycles(n int) Option

WithMaxWakeCycles caps the number of wake-driven re-entries into goai per AgentRunner.Run. Zero means "use defaultMaxWakeCycles" (10). When the cap is reached with messages still pending, the runner emits one EventMessageDropped{reason: max-wake-cycles} per remaining message. Stable.

func WithModel ¶

func WithModel(m provider.LanguageModel) Option

WithModel sets the language model. Stable.

package main

import (
	"context"
	"errors"

	"github.com/zendev-sh/goai/provider"
	"github.com/zendev-sh/zenflow"
)

// fakeModel is a minimal provider.LanguageModel that returns a canned
// response immediately. Used by examples so they compile and run
// without real API keys.
type fakeModel struct{ reply string }

func (m *fakeModel) ModelID() string { return "example-fake" }

func (m *fakeModel) DoGenerate(_ context.Context, _ provider.GenerateParams) (*provider.GenerateResult, error) {
	return &provider.GenerateResult{
		Text:         m.reply,
		FinishReason: provider.FinishStop,
		Usage:        provider.Usage{InputTokens: 5, OutputTokens: 5},
	}, nil
}

func (m *fakeModel) DoStream(_ context.Context, _ provider.GenerateParams) (*provider.StreamResult, error) {
	return nil, errors.New("streaming not implemented in fakeModel")
}

func main() {
	model := &fakeModel{reply: "hello"}
	orch := zenflow.New(zenflow.WithModel(model))
	defer orch.Close() //nolint:errcheck
	_ = orch
}

Output:

func WithModelResolver ¶

func WithModelResolver(r ModelResolver) Option

WithModelResolver installs a ModelResolver consulted by the resume path. When a saved transcript's Model identifier differs from the Executor's default runner model, the resolver is called with the transcript's model ID; a nil/empty return fails the resume loudly with ErrModelResolverMissing (DropReasonTargetTerminal). Without a resolver, mismatch also fails loudly; the resume path never silently falls back to the wrong model. Scope: - REQUIRED for cross-run resume with a non-default model (after process restart with a persistent transcript store). The Executor's per-step model-string map is intra-run memory only. - OPTIONAL for intra-run resume: the workflow step's model string is tracked automatically and the default-path match succeeds without a resolver. Stable.

package main

import (
	"context"
	"errors"

	"github.com/zendev-sh/goai/provider"
	"github.com/zendev-sh/zenflow"
)

// fakeModel is a minimal provider.LanguageModel that returns a canned
// response immediately. Used by examples so they compile and run
// without real API keys.
type fakeModel struct{ reply string }

func (m *fakeModel) ModelID() string { return "example-fake" }

func (m *fakeModel) DoGenerate(_ context.Context, _ provider.GenerateParams) (*provider.GenerateResult, error) {
	return &provider.GenerateResult{
		Text:         m.reply,
		FinishReason: provider.FinishStop,
		Usage:        provider.Usage{InputTokens: 5, OutputTokens: 5},
	}, nil
}

func (m *fakeModel) DoStream(_ context.Context, _ provider.GenerateParams) (*provider.StreamResult, error) {
	return nil, errors.New("streaming not implemented in fakeModel")
}

func main() {
	model := &fakeModel{reply: "ok"}

	resolver := zenflow.ModelResolver(func(modelID string) (provider.LanguageModel, error) {
		// In production, switch on modelID to construct the right provider.
		_ = modelID
		return model, nil
	})

	orch := zenflow.New(
		zenflow.WithModel(model),
		zenflow.WithModelResolver(resolver),
	)
	defer orch.Close() //nolint:errcheck
	_ = orch
}

Output:

func WithOutputTransform ¶

func WithOutputTransform(t OutputTransformer) Option

WithOutputTransform sets the output transformer for step output injection. When set, step outputs are transformed before injection into dependent step prompts. Use this to implement smart truncation based on target model context window. Stable.

func WithPermissions ¶

func WithPermissions(h PermissionHandler) Option

WithPermissions sets the permission handler. Stable.

package main

import (
	"context"
	"errors"

	"github.com/zendev-sh/goai/provider"
	"github.com/zendev-sh/zenflow"
)

// fakeModel is a minimal provider.LanguageModel that returns a canned
// response immediately. Used by examples so they compile and run
// without real API keys.
type fakeModel struct{ reply string }

func (m *fakeModel) ModelID() string { return "example-fake" }

func (m *fakeModel) DoGenerate(_ context.Context, _ provider.GenerateParams) (*provider.GenerateResult, error) {
	return &provider.GenerateResult{
		Text:         m.reply,
		FinishReason: provider.FinishStop,
		Usage:        provider.Usage{InputTokens: 5, OutputTokens: 5},
	}, nil
}

func (m *fakeModel) DoStream(_ context.Context, _ provider.GenerateParams) (*provider.StreamResult, error) {
	return nil, errors.New("streaming not implemented in fakeModel")
}

func main() {
	orch := zenflow.New(
		zenflow.WithModel(&fakeModel{reply: "ok"}),
		zenflow.WithPermissions(allowAll{}),
	)
	defer orch.Close() //nolint:errcheck
	_ = orch
}

// allowAll is a trivial PermissionHandler that allows all tool calls.
type allowAll struct{}

func (allowAll) RequestPermission(_ context.Context, _ zenflow.PermissionRequest) (bool, error) {
	return true, nil
}

Output:

func WithProgress ¶

func WithProgress(s ProgressSink) Option

WithProgress sets the progress event sink. Stable.

package main

import (
	"context"
	"errors"
	"fmt"

	"github.com/zendev-sh/goai/provider"
	"github.com/zendev-sh/zenflow"
)

// fakeModel is a minimal provider.LanguageModel that returns a canned
// response immediately. Used by examples so they compile and run
// without real API keys.
type fakeModel struct{ reply string }

func (m *fakeModel) ModelID() string { return "example-fake" }

func (m *fakeModel) DoGenerate(_ context.Context, _ provider.GenerateParams) (*provider.GenerateResult, error) {
	return &provider.GenerateResult{
		Text:         m.reply,
		FinishReason: provider.FinishStop,
		Usage:        provider.Usage{InputTokens: 5, OutputTokens: 5},
	}, nil
}

func (m *fakeModel) DoStream(_ context.Context, _ provider.GenerateParams) (*provider.StreamResult, error) {
	return nil, errors.New("streaming not implemented in fakeModel")
}

func main() {
	sink := &loggingSink{}
	orch := zenflow.New(
		zenflow.WithModel(&fakeModel{reply: "done"}),
		zenflow.WithProgress(sink),
	)
	defer orch.Close() //nolint:errcheck

	wf := &zenflow.Workflow{
		Name:  "observed",
		Steps: []zenflow.Step{{ID: "s1", Instructions: "do work"}},
	}
	_, _ = orch.RunFlow(context.Background(), wf)
}

// loggingSink is a simple ProgressSink that prints each event type.
type loggingSink struct{}

func (s *loggingSink) OnEvent(_ context.Context, e zenflow.Event) {
	fmt.Println("event:", e.Type)
}
func (s *loggingSink) OnOutput(_ context.Context, o zenflow.Output) {
	fmt.Print(o.Delta)
}

Output:

func WithProgressBufferSize ¶

func WithProgressBufferSize(n int) Option

WithProgressBufferSize controls the buffer of the non-blocking progress sink wrapper. Behavior: emits are non-blocking on the critical path while the buffered channel has capacity. On overflow the wrapper applies a bounded back-pressure of up to defaultEventBusTimeout (1s); if the channel is still full at the deadline the event is dropped and the internal dropped counter is incremented (visible via Stats). This bounds worst-case caller latency at 1s rather than blocking indefinitely. Larger buffers tolerate slower downstream sinks at the cost of more buffered memory. Zero or negative falls back to defaultEventBusBuffer (1024).

func WithRouterObserver ¶

func WithRouterObserver(fn func(*MessageRouter)) Option

WithRouterObserver registers a callback invoked once per RunAgent / RunFlow invocation with the freshly-allocated MessageRouter for that run. Intended for observability hooks (telemetry, debug inspectors, integration tests) that need a handle on the per-call router without polling internal state. The callback fires synchronously before any Run-side goroutine consumes the router; it must not block. Production callers typically leave this unset; nil callbacks are ignored. Panic semantics: if the callback panics, the panic IS recovered (the run continues) and logged via slog.Error with key "panic" and hook="routerObserver". The panic is NOT propagated as an error; RunAgent / RunFlow behave as if the observer had returned cleanly. Callers that need to surface observer failures must do so out-of-band (e.g. via their own channel or atomic flag captured in the closure).

func WithRunID ¶

func WithRunID(runID string) Option

WithRunID pins the orchestrator's run identifier for RunFlow / RunAgent / RunGoal. When set, all internally-emitted Event.RunID values use this ID instead of a freshly-generated one. ResumeFlow already takes runID as an argument and ignores this option. Callers that allocate a runID up-front (e.g. HTTP servers that store it in a database row and return it to the client before the workflow starts) use this to guarantee the server-visible ID and the zenflow- emitted Event.RunID match. Without this option, zenflow generates its own ID internally and any pre-allocated ID diverges - persisted events carry the internal ID while DB rows and HTTP responses carry the pre-allocated one, breaking cross-referencing (e.g. /flow-debug). Empty = fall back to internal generation (legacy behavior). Stable.

func WithSharedMemory ¶

func WithSharedMemory(sm *SharedMemory) Option

WithSharedMemory sets the shared memory instance for inter-agent collaboration. If set, shared_memory_read and shared_memory_write tools are auto-injected into agent tool chains during RunFlow and ResumeFlow. Stable.

package main

import (
	"context"
	"errors"

	"github.com/zendev-sh/goai/provider"
	"github.com/zendev-sh/zenflow"
)

// fakeModel is a minimal provider.LanguageModel that returns a canned
// response immediately. Used by examples so they compile and run
// without real API keys.
type fakeModel struct{ reply string }

func (m *fakeModel) ModelID() string { return "example-fake" }

func (m *fakeModel) DoGenerate(_ context.Context, _ provider.GenerateParams) (*provider.GenerateResult, error) {
	return &provider.GenerateResult{
		Text:         m.reply,
		FinishReason: provider.FinishStop,
		Usage:        provider.Usage{InputTokens: 5, OutputTokens: 5},
	}, nil
}

func (m *fakeModel) DoStream(_ context.Context, _ provider.GenerateParams) (*provider.StreamResult, error) {
	return nil, errors.New("streaming not implemented in fakeModel")
}

func main() {
	sm := zenflow.NewSharedMemory()
	orch := zenflow.New(
		zenflow.WithModel(&fakeModel{reply: "ok"}),
		zenflow.WithSharedMemory(sm),
	)
	defer orch.Close() //nolint:errcheck
	_ = orch
}

Output:

func WithStorage ¶

func WithStorage(s Storage) Option

WithStorage sets the persistence backend. Applies to RunFlow only - RunAgent does not persist state. Stable.

package main

import (
	"context"
	"errors"
	"os"

	"github.com/zendev-sh/goai/provider"
	"github.com/zendev-sh/zenflow"
)

// fakeModel is a minimal provider.LanguageModel that returns a canned
// response immediately. Used by examples so they compile and run
// without real API keys.
type fakeModel struct{ reply string }

func (m *fakeModel) ModelID() string { return "example-fake" }

func (m *fakeModel) DoGenerate(_ context.Context, _ provider.GenerateParams) (*provider.GenerateResult, error) {
	return &provider.GenerateResult{
		Text:         m.reply,
		FinishReason: provider.FinishStop,
		Usage:        provider.Usage{InputTokens: 5, OutputTokens: 5},
	}, nil
}

func (m *fakeModel) DoStream(_ context.Context, _ provider.GenerateParams) (*provider.StreamResult, error) {
	return nil, errors.New("streaming not implemented in fakeModel")
}

func main() {
	// MemoryStorage (default) - suitable for short-lived embeddings.
	memOrch := zenflow.New(
		zenflow.WithModel(&fakeModel{reply: "ok"}),
		zenflow.WithStorage(zenflow.NewMemoryStorage()),
	)
	defer memOrch.Close() //nolint:errcheck

	// FileStorage - persists under a temp directory for demo.
	dir, _ := os.MkdirTemp("", "zenflow-example-*")
	defer func() { _ = os.RemoveAll(dir) }()

	fileOrch := zenflow.New(
		zenflow.WithModel(&fakeModel{reply: "ok"}),
		zenflow.WithStorage(zenflow.NewFileStorage(dir)),
	)
	defer fileOrch.Close() //nolint:errcheck

	_, _ = memOrch, fileOrch
}

Output:

func WithStreaming ¶

func WithStreaming() Option

WithStreaming enables streaming mode. When enabled, content that is being displayed (controlled by WithVerbose) is delivered token-by-token via ProgressSink.OnOutput instead of as full text. Stable.

package main

import (
	"context"
	"errors"
	"fmt"

	"github.com/zendev-sh/goai/provider"
	"github.com/zendev-sh/zenflow"
)

// fakeModel is a minimal provider.LanguageModel that returns a canned
// response immediately. Used by examples so they compile and run
// without real API keys.
type fakeModel struct{ reply string }

func (m *fakeModel) ModelID() string { return "example-fake" }

func (m *fakeModel) DoGenerate(_ context.Context, _ provider.GenerateParams) (*provider.GenerateResult, error) {
	return &provider.GenerateResult{
		Text:         m.reply,
		FinishReason: provider.FinishStop,
		Usage:        provider.Usage{InputTokens: 5, OutputTokens: 5},
	}, nil
}

func (m *fakeModel) DoStream(_ context.Context, _ provider.GenerateParams) (*provider.StreamResult, error) {
	return nil, errors.New("streaming not implemented in fakeModel")
}

// loggingSink is a simple ProgressSink that prints each event type.
type loggingSink struct{}

func (s *loggingSink) OnEvent(_ context.Context, e zenflow.Event) {
	fmt.Println("event:", e.Type)
}
func (s *loggingSink) OnOutput(_ context.Context, o zenflow.Output) {
	fmt.Print(o.Delta)
}

func main() {
	orch := zenflow.New(
		zenflow.WithModel(&fakeModel{reply: "streaming reply"}),
		zenflow.WithStreaming(),
		zenflow.WithProgress(&loggingSink{}),
	)
	defer orch.Close() //nolint:errcheck
	_ = orch
}

Output:

func WithTools ¶

func WithTools(tools ...goai.Tool) Option

WithTools sets the tools available to agents. Stable.

package main

import (
	"context"
	"encoding/json"
	"errors"

	"github.com/zendev-sh/goai"
	"github.com/zendev-sh/goai/provider"
	"github.com/zendev-sh/zenflow"
)

// fakeModel is a minimal provider.LanguageModel that returns a canned
// response immediately. Used by examples so they compile and run
// without real API keys.
type fakeModel struct{ reply string }

func (m *fakeModel) ModelID() string { return "example-fake" }

func (m *fakeModel) DoGenerate(_ context.Context, _ provider.GenerateParams) (*provider.GenerateResult, error) {
	return &provider.GenerateResult{
		Text:         m.reply,
		FinishReason: provider.FinishStop,
		Usage:        provider.Usage{InputTokens: 5, OutputTokens: 5},
	}, nil
}

func (m *fakeModel) DoStream(_ context.Context, _ provider.GenerateParams) (*provider.StreamResult, error) {
	return nil, errors.New("streaming not implemented in fakeModel")
}

func main() {
	greet := goai.Tool{
		Name:        "greet",
		Description: "Returns a greeting.",
		Execute: func(_ context.Context, _ json.RawMessage) (string, error) {
			return "hello", nil
		},
	}
	orch := zenflow.New(
		zenflow.WithModel(&fakeModel{reply: "ok"}),
		zenflow.WithTools(greet),
	)
	defer orch.Close() //nolint:errcheck
	_ = orch
}

Output:

func WithTracer ¶

func WithTracer(t Tracer) Option

WithTracer sets the tracer for distributed tracing. The OTel sub-module (zenflow/observability/otel) provides a Tracer implementation. When set, workflow and step spans are created automatically. Stable.

func WithTranscriptStore ¶

func WithTranscriptStore(factory func() resume.TranscriptStore) Option

WithTranscriptStore installs a TranscriptStore factory. The factory is invoked once per Executor.Run so each Run gets a fresh store instance (matching WithMailboxStore lifecycle). The default, when this option is not supplied, is a per-Run InMemoryTranscriptStore, which supports intra-run resume only. Supply a persistent-backed store (file / SQLite) to enable cross-run or cross-process resume.

func WithTruncationOnCapReached ¶

func WithTruncationOnCapReached() Option

WithTruncationOnCapReached configures the resume path to fall back to TranscriptTruncatedLoader.LoadTruncated when a sealed transcript's Load returns ErrTranscriptTooLarge. The bound defaults to 1000 messages; stores implementing LoadTruncated may choose their own tail size.

func WithVerbose ¶

func WithVerbose() Option

WithVerbose enables agent output display. When enabled, agent LLM responses are shown in addition to events and narration. Without verbose, only workflow events (▸, ✓) and coordinator narration (≋) are shown. Stable.

func WithoutMailboxDelivery ¶

func WithoutMailboxDelivery() Option

WithoutMailboxDelivery suppresses Router/Mailbox allocation in Executor.Run - useful for tests that exercise the scheduler path without mailbox machinery. Stable.

func WithoutStreaming ¶

func WithoutStreaming() Option

WithoutStreaming disables streaming mode (restores the default batch delivery). Stable.

func WithoutTruncationOnCapReached ¶

func WithoutTruncationOnCapReached() Option

WithoutTruncationOnCapReached restores the default behavior: a sealed transcript fails the resume with DropReasonTranscriptTooLarge so operators can detect and investigate the cap.

func WithoutVerbose ¶

func WithoutVerbose() Option

WithoutVerbose disables agent output display (default: only events and narration). Stable.

func New ¶

func New(opts ...Option) *Orchestrator

New creates an Orchestrator with the given options. Stable. Defaults: MemoryStorage, maxConcurrency=5, OutputTransform=&TokenBudgetTransformer{MaxBytesPerDep: DefaultMaxBytesPerDep}. Effective defaults at call time: maxTurns=50, maxDepth=3. Use WithModel, WithTools, WithMaxTurns, etc. to configure.

package main

import (
	"context"
	"errors"

	"github.com/zendev-sh/goai/provider"
	"github.com/zendev-sh/zenflow"
)

// fakeModel is a minimal provider.LanguageModel that returns a canned
// response immediately. Used by examples so they compile and run
// without real API keys.
type fakeModel struct{ reply string }

func (m *fakeModel) ModelID() string { return "example-fake" }

func (m *fakeModel) DoGenerate(_ context.Context, _ provider.GenerateParams) (*provider.GenerateResult, error) {
	return &provider.GenerateResult{
		Text:         m.reply,
		FinishReason: provider.FinishStop,
		Usage:        provider.Usage{InputTokens: 5, OutputTokens: 5},
	}, nil
}

func (m *fakeModel) DoStream(_ context.Context, _ provider.GenerateParams) (*provider.StreamResult, error) {
	return nil, errors.New("streaming not implemented in fakeModel")
}

func main() {
	model := &fakeModel{reply: "ok"}

	orch := zenflow.New(
		zenflow.WithModel(model),
		zenflow.WithMaxConcurrency(4),
		zenflow.WithMaxTurns(20),
		zenflow.WithStorage(zenflow.NewMemoryStorage()),
	)
	defer orch.Close() //nolint:errcheck
	_ = orch
}

Output:

func LintPortability ¶

func LintPortability(wf *Workflow) (warnings []PortabilityWarning, err error)

LintPortability inspects a parsed workflow and returns: - warnings: absolute paths spotted in instructions / descriptions / default values; non-fatal. - err: the first host-specific env interpolation encountered; fatal - callers should refuse to load the workflow. Both return values are independent: a workflow can have warnings and no error.

func NewResumeStates ¶

func NewResumeStates() *ResumeStates

func WithFlowContext ¶

func WithFlowContext(ctx string) RunFlowOption

WithFlowContext supplies use-case input to be distributed to the workflow steps. When a coordinator runner is installed via WithCoordinator, the context is pushed into the coord's mailbox as the first event (workflow_start) so the coord LLM can curate per-step forwards. When no coordinator is installed (WithCoordinator(nil)), the context is blanket-prepended to every step's effective user prompt as a static fallback. Empty string is a no-op. Stable.

func WithGoalContext ¶

func WithGoalContext(ctx string) RunGoalOption

WithGoalContext supplies additional context (beyond the goal text itself) to the RunGoal coordinator-decomposition prompt. Empty string is a no-op. Stable.

func WithRunnerGoAIOptions ¶

func WithRunnerGoAIOptions(opts ...goai.Option) RunnerOption

WithRunnerGoAIOptions sets extra goai options on the AgentRunner. Stable.

func WithRunnerInitialMessages ¶

func WithRunnerInitialMessages(msgs []provider.Message) RunnerOption

WithRunnerInitialMessages pre-loads conversation history that the runner prepends to the fresh user turn passed to Run. Used by Executor.runResume to replay a saved transcript through the standard AgentRunner.Run machinery. Empty for normal (non-resume) runs. Stable.

func WithRunnerMailbox ¶

func WithRunnerMailbox(m MailboxStore) RunnerOption

WithRunnerMailbox sets the MailboxStore the runner reads inter-agent messages from. Pair with WithRunnerWake to enable mailbox-mode delivery. Stable.

func WithRunnerMaxWakeCycles ¶

func WithRunnerMaxWakeCycles(n int) RunnerOption

WithRunnerMaxWakeCycles caps the number of wake-driven re-entries into goai.GenerateText per Run. Zero or negative means "use the package default" (defaultMaxWakeCycles). Stable.

func WithRunnerModel ¶

func WithRunnerModel(m provider.LanguageModel) RunnerOption

WithRunnerModel sets the language model on the AgentRunner. Stable.

func WithRunnerModelID ¶

func WithRunnerModelID(id string) RunnerOption

WithRunnerModelID sets the model ID string (for transcript metadata) on the AgentRunner. Stable.

func WithRunnerPermissions ¶

func WithRunnerPermissions(h PermissionHandler) RunnerOption

WithRunnerPermissions sets the permission handler on the AgentRunner. Stable.

func WithRunnerPreStartDrainGate ¶

func WithRunnerPreStartDrainGate(gate <-chan struct{}) RunnerOption

WithRunnerPreStartDrainGate is a test-only hook: when non-nil, Run blocks on receive from this channel BEFORE the first drainMailboxIntoMessages call. Lets callers hold the pre-start drain while they set up mailbox preconditions. A nil gate preserves production behavior. Stable.

func WithRunnerProgress ¶

func WithRunnerProgress(s ProgressSink) RunnerOption

WithRunnerProgress sets the progress event sink on the AgentRunner. Stable.

func WithRunnerRouter ¶

func WithRunnerRouter(rt *MessageRouter) RunnerOption

WithRunnerRouter wires a shared MessageRouter into the AgentRunner so child spawns inherit a live router for inter-agent messaging. nil is valid (legacy single-call path with no messaging). Stable.

func WithRunnerRunID ¶

func WithRunnerRunID(id string) RunnerOption

WithRunnerRunID sets the workflow run ID on the AgentRunner. Stable.

func WithRunnerSpawnDepth ¶

func WithRunnerSpawnDepth(depth int) RunnerOption

WithRunnerSpawnDepth records the recursion depth of this runner relative to the top-level RunAgent invocation. Used to enrich EventToolCall payloads so TUI consumers can collapse nested-spawn cards under the parent. Stable.

func WithRunnerSpawnParentCallID ¶

func WithRunnerSpawnParentCallID(id string) RunnerOption

WithRunnerSpawnParentCallID records the agent-tool ToolCallID that produced this runner via SpawnChild. Emitted on every EventToolCall in Data["parentCallID"] so consumers can route nested events into the parent's children list. Stable.

func WithRunnerStateRef ¶

func WithRunnerStateRef(s *goai.AgentState) RunnerOption

WithRunnerStateRef wires a goai.AgentState into the AgentRunner so the poller can observe the runner's tool-loop lifecycle without holding a lock. See the StateRef field for the full lifecycle contract. Stable.

func WithRunnerStepID ¶

func WithRunnerStepID(id string) RunnerOption

WithRunnerStepID sets the step ID on the AgentRunner. Stable.

func WithRunnerStreaming ¶

func WithRunnerStreaming() RunnerOption

WithRunnerStreaming enables streaming mode on the AgentRunner. Stable.

func WithRunnerSystemPrompt ¶

func WithRunnerSystemPrompt(prompt string) RunnerOption

WithRunnerSystemPrompt sets the system prompt on the AgentRunner. Stable.

func WithRunnerTools ¶

func WithRunnerTools(tools ...goai.Tool) RunnerOption

WithRunnerTools sets the tools available to the agent. Stable.

func WithRunnerTranscript ¶

func WithRunnerTranscript(ts resume.TranscriptStore) RunnerOption

WithRunnerTranscript wires a TranscriptStore so the runner persists the step's conversation on every goai step-finish hook AND on Run exit. Required for the Resume Mechanism (R2). Stable.

func WithRunnerVerbose ¶

func WithRunnerVerbose() RunnerOption

WithRunnerVerbose enables verbose output on the AgentRunner. Stable.

func WithRunnerWake ¶

func WithRunnerWake(ch chan struct{}) RunnerOption

WithRunnerWake sets the wake-signal channel that the DeliveryEngine pings when the agent's mailbox has unread messages. Pair with WithRunnerMailbox to enable mailbox-mode delivery. Stable.

func WithRunnerWakeContextProvider ¶

func WithRunnerWakeContextProvider(fn func() string) RunnerOption

WithRunnerWakeContextProvider attaches a callback that supplies ambient context the runner injects as a fresh user-role message once before the first LLM call and once after every wake-driven mailbox drain. The returned string is wrapped in <dynamic-context> tags. Empty / whitespace-only returns are skipped. Pass nil to disable (default). Designed for long-running coordinators that need per-wake context refresh (e.g. open-file list, repo metadata, session topic) without re-engineering the system prompt. Coord callers should normally use WithCoordContextProvider instead, which threads through here. Stable.

func NewSharedMemory ¶

func NewSharedMemory() *SharedMemory

NewSharedMemory creates a new empty SharedMemory.

func NewSubmitResultHandler ¶

func NewSubmitResultHandler(schema map[string]any) *SubmitResultHandler

NewSubmitResultHandler creates a handler that validates against the given schema.

func LoadWorkflow ¶

func LoadWorkflow(path string) (*Workflow, error)

LoadWorkflow reads a YAML file, parses it, validates, and resolves @ refs.

func ParseCoordinatorResponse ¶

func ParseCoordinatorResponse(response string) (*Workflow, error)

ParseCoordinatorResponse parses the coordinator LLM output into a Workflow. Returns an error if JSON is invalid or the workflow fails validation. Returns *JSONParseError for malformed JSON, *CoordinatorValidationError for schema issues.

func ParseWorkflow ¶

func ParseWorkflow(data []byte) (*Workflow, error)

ParseWorkflow unmarshals YAML data into a Workflow and validates it. It does NOT resolve @ refs (no baseDir context).

package main

import (
	"errors"
	"fmt"

	"github.com/zendev-sh/zenflow"
)

func main() {
	yaml := `
name: greet
steps:
  - id: hello
    instructions: "Say hello to the user."
`
	wf, err := zenflow.ParseWorkflow([]byte(yaml))
	if err != nil {
		// ParseWorkflow may return a joined error for multi-violation YAML.
		// Unwrap with errors.As to inspect individual *zenflow.ValidationError.
		var ve *zenflow.ValidationError
		if errors.As(err, &ve) {
			fmt.Println("validation error:", ve.Message)
		}
		return
	}

	fmt.Println(wf.Name)
	fmt.Println(len(wf.Steps))
}

Output:
greet
1

package main

import (
	"fmt"

	"github.com/zendev-sh/zenflow"
)

func main() {
	// Two agents both missing the required 'description' field - ValidateWorkflow
	// accumulates one ValidationError per agent and joins them, so the
	// caller sees all violations in one error value rather than only the first.
	invalid := []byte(`
name: test
agents:
  a:
    prompt: "do a"
  b:
    prompt: "do b"
steps:
  - id: s1
    agent: a
    instructions: "run a"
  - id: s2
    agent: b
    instructions: "run b"
    dependsOn: [s1]
`)
	_, err := zenflow.ParseWorkflow(invalid)
	if err == nil {
		return
	}
	// err is errors.Join'd; iterate sub-errors via Unwrap []error.
	if joined, ok := err.(interface{ Unwrap() []error }); ok {
		fmt.Printf("found %d validation errors\n", len(joined.Unwrap()))
	}

}

Output:
found 2 validation errors

func ParseWorkflowJSON ¶

func ParseWorkflowJSON(data []byte) (*Workflow, error)

ParseWorkflowJSON unmarshals JSON data into a Workflow and validates it. Used for coordinator output (JSON format) and .json workflow files. (2026-05-04) - switched from json.Unmarshal to json.Decoder with DisallowUnknownFields so that schema.json's `additionalProperties: false` is enforced symmetrically with the YAML path. Previously YAML rejected unknown fields via yaml.v3 `KnownFields(true)` but JSON silently dropped them, letting a consumer's typo in a `.json` workflow pass without diagnostic.