runner

package
v0.32.1 Latest Latest
Warning

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

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

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var ErrHostAlreadyRunning = errors.New("runner: host invocation is already running")

ErrHostAlreadyRunning is returned when another outer invocation is active.

View Source 
var ErrHostInvocationRunning = errors.New("runner: host invocation is still running")

ErrHostInvocationRunning is returned when shutdown is requested before the outer invocation callback has completed.

View Source 
var ErrHostShutdownStarted = errors.New("runner: host shutdown has started")

ErrHostShutdownStarted is returned when resources are adopted after the host has begun shutting down, or when Run is called on a closed host.

Functions

func EvaluateAssertions added in v0.27.0 

func EvaluateAssertions(result any, asserts []string, localSymbols ...map[string]compiler.SymbolNamespace) []error

EvaluateAssertions evaluates assertions without rendering CLI output. It is shared with self-test so both paths use the same expression and symbol rules.

func EvaluateAssertionsContext added in v0.27.0 

func EvaluateAssertionsContext(ctx context.Context, result any, asserts []string, localSymbols ...map[string]compiler.SymbolNamespace) []error

EvaluateAssertionsContext evaluates assertions without rendering output and preserves the caller's context.

func FormatDuration added in v0.10.0 

func FormatDuration(d time.Duration) string

FormatDuration handles Windows clock resolution limits cleanly without misleading '0ns' outputs.

func JSONMapView added in v0.10.0 

func JSONMapView(value any) any

JSONMapView applies JSON tags to struct outputs before expr evaluation.

func ParseApprovalOption added in v0.30.2 

func ParseApprovalOption(args []string) (token string, remaining []string, err error)

ParseApprovalOption removes at most one --approval token from args and returns the remaining arguments unchanged and in order.

func Preflight added in v0.17.0 

func Preflight(ctx context.Context, cfg Config, src, name string) (*compiler.PreparedSource, error)

func PrintInfo added in v0.10.0 

func PrintInfo(w io.Writer, path, _ string, meta map[string]any, ast compiler.Expr) int

PrintInfo describes a .nflow file without executing it. It shows directives from meta, the sequence of atoms from the AST, and expected input.

func PrintInfoJSON added in v0.14.0 

func PrintInfoJSON(w io.Writer, path string, meta map[string]any, ast compiler.Expr) error

func Run added in v0.10.0 

func Run(ctx context.Context, cfg Config, path string, payload map[string]any, opts Opts) int

Run executes a .nflow file from disk.

func RunAssertions 

func RunAssertions(w io.Writer, result any, durationMS int64, actions, asserts []string, verbosity int, localSymbols ...map[string]compiler.SymbolNamespace) int

RunAssertions evaluates every @assert expression against the run result and prints a per-assert pass/fail line.

Failures are tinted red when the output writer is a terminal. When at least one assert fails, a trailing "exit status: N" line reports the process exit code the CLI is about to return, so a caller scanning the tail of the output sees the outcome without inferring it from the individual ✗ marks.

func RunAssertionsContext added in v0.27.0 

func RunAssertionsContext(ctx context.Context, w io.Writer, result any, durationMS int64, actions, asserts []string, verbosity int, localSymbols ...map[string]compiler.SymbolNamespace) int

RunAssertionsContext evaluates every @assert expression using the caller's execution context.

func RunLean added in v0.23.0 

func RunLean(ctx context.Context, source, name string, bundles []compiler.Bundle, args []string) int

RunLean executes an embedded flow source using only the provided bundles. It contains zero CLI, testkit, or development dependencies.

func RunSource added in v0.10.0 

func RunSource(ctx context.Context, cfg Config, src, name string, payload map[string]any, opts Opts) int

RunSource executes workflow DSL directly from in-memory text.

It is a thin wrapper over Execute: it handles --info, installs the observer, prints the result, and evaluates @assert: directives. Every other piece of orchestration lives in Execute so the same pipeline runs across cli, selftest, and embedded binaries.

func SplitPipelineModifiers added in v0.14.0 

func SplitPipelineModifiers(table *compiler.ModifierTable, mods []string) (wrapper, body []string)

Types

type ApprovalSuspension added in v0.30.2 

type ApprovalSuspension struct {
	ExecutionID string    `json:"execution_id"`
	Gate        string    `json:"gate"`
	Role        string    `json:"role"`
	Reason      string    `json:"reason"`
	SourceName  string    `json:"source_name"`
	SourceHash  string    `json:"source_hash"`
	InputHash   string    `json:"input_hash"`
	SuspendedAt time.Time `json:"suspended_at"`
	ExpiresAt   time.Time `json:"expires_at"`
	Persisted   bool      `json:"-"`
}

ApprovalSuspension is the persisted description of a workflow paused at an approval gate. It stores hashes, not the source or input payload.

type Compilation added in v0.14.0 

type Compilation struct {
	Program  action.AnyAction
	Meta     map[string]any
	Resolver compiler.CapabilityResolver
}

func Compile added in v0.14.0 

func Compile(ctx context.Context, cfg Config, src, name string) (Compilation, error)

func CompileWithOptions added in v0.23.0 

func CompileWithOptions(ctx context.Context, cfg Config, src, name string, opts ...compiler.CompileOption) (Compilation, error)

type Config added in v0.10.0 

type Config struct {
	Resolver compiler.CapabilityResolver
	// FileResolver routes typed file references registered by bundles. BuildConfig
	// composes provider bindings and Execute exposes the resolver to actions.
	FileResolver  xfile.Resolver
	Directives    *compiler.DirectiveTable
	Modifiers     *compiler.ModifierTable
	Operators     *compiler.OperatorTable
	Primaries     *compiler.PrimaryExtensionTable
	Materializers []compiler.Materializer
	CompileOpts   []compiler.CompileOption
	// Journal, when non-nil, makes each source-positioned capability call
	// durable through Kernel's action.Durable middleware.
	Journal action.Journal
	// IdempotencyStore replaces Kernel's default in-memory store for actions
	// carrying :idempotent. It may be the same object as Journal when the
	// backend implements both contracts.
	IdempotencyStore action.IdempotencyStore

	// Hooks are applied to every resolver action and to the top-level
	// program before execution. Callers that only want to see the
	// result set this to nil.
	Hooks []action.AnyHook

	// EventSink, when non-nil, is wrapped by observe.Hook and appended
	// to Hooks. Both paths end up on the execution-local resolver's
	// actions; use either, or both.
	EventSink observe.Sink

	// MeasureAllocs wraps the compile and run phases of Execute with
	// MemStats deltas so callers can attribute allocations to the
	// compiler versus the runtime. ReadMemStats triggers a brief
	// stop-the-world pause; leave false on production hot paths. This
	// mirrors the same precedent as CompileDuration and RunDuration:
	// Execute already reports per-phase diagnostics, and this is the
	// allocation counterpart.
	MeasureAllocs bool

	// ArgSchemas is the merged per-action argument schema published by
	// every mounted bundle. It is exposed so diagnostics (nflow lint
	// today) can validate @{...} capability references without building
	// the action graph. Build-time advisors are the runtime counterpart;
	// this field makes the same checks reachable from a pure parse.
	ArgSchemas map[string][]compiler.ArgFieldSpec
}

Config bundles all compiler tables, capability resolvers, and materializers for a run. It is passed explicitly to prevent mutable package-level state.

func BuildConfig added in v0.10.0 

func BuildConfig(bundles []compiler.Bundle) (Config, error)

BuildConfig compiles bundles into execution configuration. It has ZERO knowledge of @require, URLs or module paths.

func (Config) PrimariesFor added in v0.14.0 

func (c Config) PrimariesFor(
	meta map[string]any,
	parent ...compiler.PrimaryExtension,
) *compiler.PrimaryExtensionTable

PrimariesFor returns the parser primary table the compiler installs for a source that produced meta. Parent primaries, when supplied, are inserted between the base table and the source's own contributions, mirroring CompileReq.InheritedPrimaries.

Every call site that parses a source outside of CompileAction — nflow lint, nflow expand, and any future diagnostic tool — must go through this function. Reading cfg.Primaries directly is only correct for the base table, and doing so silently drops per-source primaries such as the macro engine. A linter that does this rejects a file the runtime accepts, which is exactly the class of "test and production disagree" bug the runner package exists to prevent.

Order of contribution:

  1. c.Primaries — the base table built at BuildConfig
  2. parent... — inherited from an outer source
  3. meta's OnPreprocess output — this source's own contributions

A MergeablePrimary in step 2 or 3 collapses with the same token in step 1 or 2. A plain primary collides and panics, which is the "duplicates fail loudly" rule.

type DurabilityOptions added in v0.30.2 

type DurabilityOptions struct {
	JournalDir  string
	ExecutionID string
}

DurabilityOptions configure the FileJournal used for a single run and the execution identity to use when replaying that run.

func ParseDurabilityOptions added in v0.30.2 

func ParseDurabilityOptions(args []string) (DurabilityOptions, []string, error)

ParseDurabilityOptions removes the runner's durability flags from args and returns their values. Both --flag=value and --flag value forms are accepted. Unrecognized arguments are returned, in order, for the caller to process.

type EntryPoint added in v0.17.0 

type EntryPoint struct {
	Name    string
	Payload any
}

type Execution added in v0.10.0 

type Execution struct {
	ExecutionID     string
	Output          any
	Suspension      *ApprovalSuspension
	Meta            map[string]any
	Resolver        compiler.CapabilityResolver
	CompileDuration time.Duration
	RunDuration     time.Duration

	CompileAllocs     uint64
	CompileAllocBytes uint64
	RunAllocs         uint64
	RunAllocBytes     uint64
}

func Execute added in v0.10.0 

func Execute(ctx context.Context, cfg Config, src, name string, payload map[string]any) (Execution, error)

func ExecutePrepared added in v0.17.0 

func ExecutePrepared(
	ctx context.Context,
	cfg Config,
	prepared *compiler.PreparedSource,
	entry EntryPoint,
	extraOpts ...compiler.CompileOption,
) (Execution, error)

type FileJournal added in v0.30.2 

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

FileJournal persists Kernel durable-step records and idempotency responses below one confined directory. Step and idempotency keys are hashed before they are used as relative paths. The same instance also coordinates idempotency claims among callers in this process; it is not a distributed lock service across separate processes.

func NewFileJournal added in v0.30.2 

func NewFileJournal(dir string) (*FileJournal, error)

NewFileJournal opens or creates a journal directory. It implements both action.Journal and action.IdempotencyCoordinator, so callers may assign the same instance to runner.Config.Journal and runner.Config.IdempotencyStore.

func (*FileJournal) Claim added in v0.30.2 

func (j *FileJournal) Claim(ctx context.Context, key, requestHash string, leaseTTL time.Duration) (action.IdempotencyClaim, error)

Claim coordinates duplicate idempotency keys within this FileJournal instance and checks for previously completed responses.

func (*FileJournal) ClearApprovalSuspension added in v0.30.2 

func (j *FileJournal) ClearApprovalSuspension(ctx context.Context, executionID string) error

ClearApprovalSuspension removes a pending approval record after the resumed execution completes successfully.

func (*FileJournal) Close added in v0.30.2 

func (j *FileJournal) Close() error

Close releases the rooted filesystem handle. The journal must not be used after Close; closing it more than once is safe.

func (*FileJournal) Complete added in v0.30.2 

func (j *FileJournal) Complete(ctx context.Context, key, token string, entry action.IdempotencyEntry, ttl time.Duration) error

Complete persists the response and releases the active claim.

func (*FileJournal) Get added in v0.30.2 

func (j *FileJournal) Get(ctx context.Context, key string) (action.IdempotencyEntry, bool)

Get implements action.IdempotencyStore. The Kernel's legacy Get contract cannot return an error; on a corrupt or unreadable persisted record it returns an invalid body so Kernel fails closed when decoding the cached response, rather than executing the action again.

func (*FileJournal) GetStep added in v0.30.2 

func (j *FileJournal) GetStep(ctx context.Context, executionID, stepName string) (data []byte, found bool, readErr error)

GetStep loads a previously recorded durable result. A missing record is a normal cache miss; filesystem errors are returned to Kernel and do not cause the step to run silently.

func (*FileJournal) LoadApprovalSuspension added in v0.30.2 

func (j *FileJournal) LoadApprovalSuspension(ctx context.Context, executionID string) (ApprovalSuspension, bool, error)

LoadApprovalSuspension returns the pending approval request for an execution.

func (*FileJournal) RecordStep added in v0.30.2 

func (j *FileJournal) RecordStep(ctx context.Context, executionID, stepName string, data []byte) error

RecordStep atomically writes one completed durable result.

func (*FileJournal) Release added in v0.30.2 

func (j *FileJournal) Release(ctx context.Context, key, token string) error

Release drops an uncompleted idempotency claim.

func (*FileJournal) RenewClaim added in v0.30.2 

func (j *FileJournal) RenewClaim(ctx context.Context, key, token string, leaseTTL time.Duration) error

RenewClaim extends an active claim's lease while a long action is running.

func (*FileJournal) SaveApprovalSuspension added in v0.30.2 

func (j *FileJournal) SaveApprovalSuspension(ctx context.Context, suspension ApprovalSuspension) error

SaveApprovalSuspension atomically records the current pending approval gate.

func (*FileJournal) Set added in v0.30.2 

func (j *FileJournal) Set(ctx context.Context, key string, entry action.IdempotencyEntry, ttl time.Duration)

Set implements action.IdempotencyStore for compatibility with callers that use the basic store API. Kernel's IdempotencyMiddleware detects the coordinator methods below and uses Complete for normal executions, allowing persistence failures to be returned to the caller.

type Host added in v0.15.0 

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

Host owns bundle cleanup callbacks for one outer synchronous invocation. It is intentionally separate from Config so repeated Compile/Execute calls can safely share a configuration and its resources.

func NewHost added in v0.15.0 

func NewHost(opts ...HostOption) *Host

NewHost creates an invocation host.

func (*Host) Own added in v0.15.0 

func (h *Host) Own(bundle compiler.Bundle) error

Own transfers the bundle's cleanup callbacks to this host. Call it once for each constructed bundle instance, immediately after construction; bundle IDs are not identities and are deliberately not used for deduplication.

func (*Host) OwnAll added in v0.15.0 

func (h *Host) OwnAll(bundles []compiler.Bundle) error

OwnAll transfers every bundle's cleanup callbacks to this host in slice order. Shutdown invokes all callbacks in reverse order, including callbacks on bundles sharing the same ID.

func (*Host) Run added in v0.15.0 

func (h *Host) Run(ctx context.Context, fn func(context.Context) error) (result error)

Run is the outer synchronous lifetime for an invocation. It always shuts down after fn, joins cleanup errors after the primary error, and re-panics with the original panic value if fn panics.

func (*Host) Shutdown added in v0.15.0 

func (h *Host) Shutdown(ctx context.Context) error

Shutdown invokes every owned callback exactly once in reverse ownership order. Callbacks receive a fresh deadline context that retains values from ctx but is unaffected by its cancellation or deadline. Cleanup callbacks must cooperate with that context; Go cannot forcibly stop a callback that ignores cancellation without abandoning a goroutine.

type HostOption added in v0.15.0 

type HostOption func(*hostOptions)

HostOption configures an invocation host.

func WithShutdownTimeout added in v0.15.0 

func WithShutdownTimeout(timeout time.Duration) HostOption

WithShutdownTimeout sets the maximum time available to bundle cleanup. A non-positive timeout selects the conservative default.

type InfoShape added in v0.14.0 

type InfoShape struct {
	Path        string   `json:"path"`
	Description string   `json:"description,omitempty"`
	Asserts     []string `json:"asserts,omitempty"`
	Atoms       []string `json:"atoms"`
}

type Observer added in v0.10.0 

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

Observer tracks every action that passes through the runtime.

func NewObserver added in v0.10.0 

func NewObserver(out io.Writer, verbosity int) *Observer

func (*Observer) ActionNames added in v0.10.0 

func (o *Observer) ActionNames() []string

func (*Observer) Hook added in v0.10.0 

func (o *Observer) Hook() action.AnyHook

func (*Observer) PrintSummary added in v0.10.0 

func (o *Observer) PrintSummary(w io.Writer)

func (*Observer) Records added in v0.10.0 

func (o *Observer) Records() []Record

type Opts added in v0.10.0 

type Opts struct {
	Verbosity int
	Info      bool
	Asserts   []string
	// ApprovalToken grants the named @approval gates, or all gates when set to
	// "all". The caller is responsible for authenticating the approver.
	ApprovalToken string
	// ExecutionID can be set when a run must resume or replay against the same
	// durable journal. An empty value lets the runner generate a new ID.
	ExecutionID string
	// JournalDir opens a FileJournal for this invocation. When omitted, the
	// caller may still provide a journal directly in Config.
	JournalDir string
	Stdout     io.Writer
	Stderr     io.Writer
}

Opts controls execution options for a single runner invocation.

type Record added in v0.10.0 

type Record struct {
	Action   string
	Duration time.Duration
	Err      error
	Start    time.Time
}

type SourcedModifier added in v0.14.0 

type SourcedModifier struct {
	Raw    string
	Source compiler.ModifierSource
}

func SplitPipelineModifiersWithSources added in v0.14.0 

func SplitPipelineModifiersWithSources(
	table *compiler.ModifierTable,
	mods []string,
	sources []compiler.ModifierSource,
) (wrapper, body []SourcedModifier)

type SuspensionStore added in v0.30.2 

type SuspensionStore interface {
	LoadApprovalSuspension(ctx context.Context, executionID string) (ApprovalSuspension, bool, error)
	SaveApprovalSuspension(ctx context.Context, suspension ApprovalSuspension) error
	ClearApprovalSuspension(ctx context.Context, executionID string) error
}

SuspensionStore persists pending approval requests independently of completed durable steps. FileJournal implements this contract.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.