flow

module

Go to main page

Versions in this module

v0

v0.20.3

Oct 8, 2026

v0.20.2

Oct 8, 2026

v0.20.1

Oct 8, 2026

v0.20.0

Oct 8, 2026

v0.19.0

Oct 8, 2026

v0.18.0

Oct 7, 2026

v0.17.1

Oct 7, 2026

v0.17.0

Oct 6, 2026

v0.16.0

Oct 5, 2026

v0.15.0

Oct 5, 2026

v0.14.0

Oct 4, 2026

v0.13.1

Oct 2, 2026

v0.13.0

Oct 2, 2026

v0.12.1

Oct 2, 2026

v0.12.0

Oct 2, 2026

v0.11.0

Oct 1, 2026

v0.10.2

Oct 1, 2026

v0.10.1

Oct 1, 2026

v0.10.0

Oct 1, 2026

v0.9.0

Sep 25, 2026

Changes in this version

type CompileOption

+ func WithCompileContext(ctx context.Context) CompileOption

type Preprocessed

+ func PreprocessFS(fsys fs.FS, path string) (*Preprocessed, error)

v0.8.0

Sep 23, 2026

Changes in this version

+ var DefaultProfiles = map[Profile]ProfilePolicy

+ func BaseLibrary() action.Library

+ func BuildResolvedConfig(pre *core.Preprocessed, cliArgs []string) map[string]string

+ func CoerceLiteralValue(trimmed string) any

+ func ContributeRegistry(pre *core.Preprocessed, base *action.Registry) (*action.Registry, error)

+ func CountStreamResult(res any) int

+ func GateRules(pre *directives.Preprocessed) []directives.GateRule

+ func HasStreamAtoms(dsl string, registry *Registry) bool

+ func RegisterBoundary(spec BoundarySpec)

+ func RegisterResolver(r ServiceResolver)

+ func ResetResolvers() int

+ func ResolveParamRef(raw string, opts *compileOptions) any

+ func SetDefaultRegistry(r *Registry)

+ func WithGates(ctx context.Context, gates []directives.GateRule) context.Context

+ func WithSecurityObserver(o SecurityObserver) func(*Compiler)

+ type BoundarySpec struct

+ Consume func(upstream action.AnyStreamAction) *action.Builder[any, any]

+ Description string

+ Name string

+ func BoundaryByName(name string) (BoundarySpec, bool)

+ func NamedBoundaries() []BoundarySpec

+ type CompileOption func(*compileOptions)

+ func WithAtomAdvisors(pre *core.Preprocessed) CompileOption

+ func WithCLIArgs(args []string) CompileOption

+ func WithConfig(cfg map[string]string) CompileOption

+ func WithFlowRegistry(reg *Registry) CompileOption

+ func WithSchemas(pre *core.Preprocessed) CompileOption

type GraphDefinition

+ Profile Profile

type GraphExecReq

+ Name string

+ Preprocessed *core.Preprocessed

+ Profile Profile

type GraphExecRes

+ Result any

+ type Kind uint8

+ const KindOperator

+ const KindSource

+ const KindUnary

+ func (k Kind) String() string

+ type NamedOperator = action.NamedOperator

type Preprocessed

+ func PreprocessBytes(source []byte, name string) (*Preprocessed, error)

+ type Profile string

+ const ProfileNetworkIsolated

+ const ProfileTrusted

+ const ProfileUntrustedInput

+ type ProfilePolicy struct

+ AllowedCapabilities []string

+ DefaultHooks []string

+ MaxRetries int

+ func LookupProfile(name string) (ProfilePolicy, error)

+ func (p ProfilePolicy) Allows(capability string) bool

type Registry

+ func DefaultRegistry() *Registry

+ func RegistryFromActionRegistry(actionRegistry *action.Registry) *Registry

+ func (r *Registry) GetOperator(name string) (NamedOperator, bool)

+ func (r *Registry) GetStream(name string) (action.AnyStreamAction, bool)

+ func (r *Registry) Names() []string

+ func (r *Registry) Register(lib Library) error

+ func (r *Registry) Resolve(name string) (Kind, bool)

+ type SecurityEvent struct

+ Elapsed time.Duration

+ Err error

+ Hook string

+ Node string

+ Phase string

+ type SecurityObserver interface

+ OnSecurity func(ctx context.Context, ev SecurityEvent)

+ type ServiceResolver interface

+ Match func(actionName string, mods dslparse.Modifiers) bool

+ Resolve func(actionName string, mods dslparse.Modifiers, act action.AnyAction) (action.AnyAction, error)

+ type StreamDrainResult struct

+ Count int

+ type StreamOperator = action.StreamOperator

+ func ComposeOperators(ops ...StreamOperator) (StreamOperator, error)

+ func NewTypedStreamOperator[In, Out any](name string, op action.StreamOp[In, Out]) StreamOperator

v0.7.0

Sep 16, 2026

Changes in this version

type Compiler

+ func (c *Compiler) CompilePipeline(expr string) (action.Executable, error)

v0.6.0

Sep 15, 2026

Changes in this version

+ func ExponentialJitterOr(base, maxDelay time.Duration) func(attempt int) time.Duration

+ func LibraryNames(libs []Library) []string

+ func MaxTokensFromCtx(ctx context.Context, def int) int

+ func RegisterPipelines(reg *MapRegistry, pipelines []Pipeline) error

+ func WithHooks(hooks ...action.AnyHook) func(*Compiler)

+ func WithMaxTokens(ctx context.Context, n int) context.Context

+ type ActionMeta struct

+ Description string

+ Name string

+ type Alias struct

+ Canonical string

+ Short []string

+ type Config struct

+ Approval string

+ BudgetMicros int64

+ MaxTokens int

+ Model string

+ Observe string

+ OutputDir string

+ OutputFormat string

+ Provider string

+ Sandbox string

+ Verbosity int

+ type Layer uint8

+ const LayerCLI

+ const LayerDSL

+ const LayerDefault

+ const LayerEnv

+ func (l Layer) String() string

+ type Library struct

+ Actions []action.AnyAction

+ Aliases []Alias

+ Description string

+ Hooks []action.AnyHook

+ Name string

+ Overrides []string

+ func StandardLibrary() Library

type MapRegistry

+ func BuildRegistry(libs ...Library) (*MapRegistry, error)

+ type Pipeline struct

+ Body string

+ Name string

+ type Preprocessed struct

+ Action *ActionMeta

+ DSL string

+ Includes []string

+ Pipelines []Pipeline

+ Requires []Requirement

+ func Preprocess(path string) (*Preprocessed, error)

+ type Requirement struct

+ Import string

+ LocalPath string

+ ModulePath string

+ ModuleRoot string

+ Version string

+ func (r Requirement) IsLocal() bool

+ type Resolved struct

+ Config Config

+ Provenance map[string]Layer

+ func ResolveConfig(dslText string, cliArgs []string) Resolved

+ type Runner interface

+ RunFlow func(ctx context.Context, path string, payload map[string]any, args []string, ...) int

v0.5.0

Sep 14, 2026

Changes in this version

+ var DefaultAliases = map[string][]string

+ func RegisterAliases(reg *MapRegistry)

+ func SanitizeDSL(rawContent string) string

+ type PipelineCompiler = contracts.PipelineCompiler

v0.4.0

Sep 13, 2026

Changes in this version

+ var ApprovalTokenKey = xctx.NewKey[string]("nexss.approval.token")

type DynamicPolicy

+ A2ARole string

+ CLICommand string

+ CLIDesc string

+ CustomName string

+ Description string

+ HTTPMethod string

+ HTTPPath string

+ StatusCode int

+ type SystemAssembler struct

+ func NewAssembler(capabilities ...action.AnyAction) *SystemAssembler

+ func (s *SystemAssembler) Actions() []action.AnyAction

+ func (s *SystemAssembler) AssembleFile(path string) ([]action.AnyAction, error)

+ func (s *SystemAssembler) AssembleManifest(manifestDSL string) ([]action.AnyAction, error)

+ func (s *SystemAssembler) Register(name string, act action.AnyAction) *SystemAssembler

+ type WorkflowFixture struct

+ func NewWorkflowTest(t testing.TB, reg Registry, dsl string) *WorkflowFixture

+ func (wf *WorkflowFixture) Execute(input any) *WorkflowResult

+ func (wf *WorkflowFixture) WithTimeout(d time.Duration) *WorkflowFixture

+ type WorkflowResult struct

+ func (r *WorkflowResult) Duration() time.Duration

+ func (r *WorkflowResult) ExpectError() *WorkflowResult

+ func (r *WorkflowResult) ExpectSuccess() *WorkflowResult

+ func (r *WorkflowResult) Output() any

v0.3.1

Sep 10, 2026

v0.3.0

Sep 10, 2026

Changes in this version

+ func GuardCost(reserver cost.Reserver, estimateMicros int64) action.AnyHook

+ func WithReserver(r cost.Reserver) func(*Compiler)

type DynamicPolicy

+ CacheTTL time.Duration

+ Coalesce bool

+ Debug bool

+ Dedup bool

+ Validate bool

type GraphPolicy

+ BudgetMicros int64

type NodeSpec

+ EstimateMicros int64

v0.2.0

Sep 9, 2026

v0.1.2

Sep 9, 2026

v0.1.1

Sep 8, 2026

v0.1.0

Sep 8, 2026

Changes in this version

+ const APIVersion

+ const BranchSelected

+ const BranchSkipped

+ const EdgeRingSize

+ func AcquireStateFromGraphState(s *State) *dag.State

+ func ApprovalTokenFrom(ctx context.Context) string

+ func AsCostHook(ledger CostLedger, estimatedCostMicros int64, budgetLimitMicros int64) action.AnyHook

+ func BuildCatalogAction(reg Registry) action.AnyAction

+ func BuildEdgeMeshAction(region LocationRegion, nodeID string, ring *EdgeTelemetryRing) action.AnyAction

+ func ChargeBranch(ctx context.Context, ledger CostLedger, g *CompiledGraph, event CostEvent) error

+ func CompilePipeline(expr string, reg Registry) (*action.Builder[any, any], error)

+ func CompileSaga(expr string, reg Registry) (*action.Builder[any, any], error)

+ func EvaluateCondition(condition string, state *State) (bool, error)

+ func Execute[Req, Res any](ctx context.Context, act *action.BuiltAction[Req, Res], req Req) (Res, error)

+ func NewExecuteAction(compiler *Compiler) *action.BuiltAction[GraphExecReq, GraphExecRes]

+ func NewPromptNode(cfg nodes.PromptConfig) action.AnyAction

+ func NewSupervisorNode(name string, compiler nodes.PipelineCompiler) action.AnyAction

+ func WithApprovalGate(g ApprovalGate) func(*Compiler)

+ func WithApprovalToken(ctx context.Context, token string) context.Context

+ func WithJournal(j BranchJournal) func(*Compiler)

+ func WithLedger(l CostLedger) func(*Compiler)

+ type ApprovalGate interface

+ Check func(ctx context.Context, actionName string, payload string, token string) error

+ type BranchJournal = journal.BranchJournal

+ type BranchMode string

+ const BranchAllMatches

+ const BranchFirstMatch

+ type BranchRecord = journal.BranchRecord

+ type BranchStatus = journal.BranchStatus

+ type BudgetCheck = governance.BudgetCheck

+ type BudgetPolicy struct

+ MaxCostUSD *float64

+ MaxCostUSDPerDay *float64

+ MaxCostUSDPerMonth *float64

+ MaxCostUSDPerRun *float64

+ MaxCostUSDPerTenant *float64

+ type CapabilitySpec struct

+ Description string

+ InputSchema map[string]any

+ IsSystem bool

+ Method string

+ Name string

+ OutputSchema map[string]any

+ Route string

+ Tags []string

+ func ExtractCapabilities(reg Registry) []CapabilitySpec

+ type ChildResult = nodes.ChildResult

+ type ChildTask = nodes.ChildTask

+ type CompiledBudget struct

+ MaxCostMicrosPerDay int64

+ MaxCostMicrosPerMonth int64

+ MaxCostMicrosPerRun int64

+ MaxCostMicrosPerTenant int64

+ type CompiledEdge struct

+ Budget CompiledBudget

+ From string

+ Otherwise bool

+ Priority int

+ To string

+ When string

+ type CompiledGraph struct

+ Budget CompiledBudget

+ Definition GraphDefinition

+ EdgeKeys []string

+ Edges []CompiledEdge

+ Layers [][]string

+ NodeByID map[string]NodeSpec

+ Outgoing map[string][]CompiledEdge

+ func Compile(def GraphDefinition) (*CompiledGraph, error)

+ func LoadYAML(data []byte) (*CompiledGraph, error)

+ func LoadYAMLFile(path string) (*CompiledGraph, error)

+ func (g *CompiledGraph) SelectOutgoing(source string, matches func(condition string) (bool, error)) ([]CompiledEdge, error)

+ func (g *CompiledGraph) SelectOutgoingDurable(ctx context.Context, j journal.BranchJournal, runID, source string, ...) ([]CompiledEdge, error)

+ type Compiler struct

+ func NewCompiler(reg Registry, opts ...func(*Compiler)) *Compiler

+ func (c *Compiler) Compile(ctx context.Context, def GraphDefinition) (*dag.DAG, *CompiledGraph, error)

+ type CostEvent = governance.CostEvent

+ type CostLedger = governance.CostLedger

+ type CostReporter interface

+ CostMicros func() int64

+ type CostUsage = governance.CostUsage

+ type DynamicPolicy struct

+ Breaker bool

+ Idempotent bool

+ RetryCount int

+ Timeout time.Duration

+ type EdgeEventSlot struct

+ EdgeID [16]byte

+ LatencyNs int64

+ Seq uint64

+ StatusCode uint16

+ type EdgeNodeReq struct

+ Payload map[string]any

+ Region LocationRegion

+ type EdgeNodeRes struct

+ Data map[string]any

+ LatencyMs int64

+ NodeID string

+ Region LocationRegion

+ StatusCode uint16

+ type EdgeSpec struct

+ Budget BudgetPolicy

+ From string

+ Otherwise bool

+ Priority int

+ To string

+ When string

+ type EdgeTelemetryRing struct

+ func NewEdgeTelemetryRing() *EdgeTelemetryRing

+ func (r *EdgeTelemetryRing) BatchDrain(dst []EdgeEventSlot) int

+ func (r *EdgeTelemetryRing) Push(edgeID [16]byte, latencyNs int64, statusCode uint16) bool

+ type EffectClass string

+ const EffectHighRisk

+ const EffectReadOnly

+ const EffectSideEffect

+ type EventSlot = telemetry.EventSlot

+ type FanInPolicy struct

+ AllowPartial bool

+ FailOnEmpty bool

+ RequireAll bool

+ type GraphDefinition struct

+ APIVersion string

+ Edges []EdgeSpec

+ Kind string

+ Metadata Metadata

+ Nodes []NodeSpec

+ Policy GraphPolicy

+ func ParseArrowDSL(name, dsl string) (GraphDefinition, error)

+ type GraphExecReq struct

+ DSL string

+ InitialPayload map[string]any

+ YAML string

+ type GraphExecRes struct

+ DurationMS int64

+ GraphName string

+ LayersRun int

+ Outputs map[string]any

+ type GraphPolicy struct

+ ApprovalRequiredFor []EffectClass

+ Budget BudgetPolicy

+ FanInRecovery RecoveryPolicy

+ MaxContextBytes int64

+ MaxParallelNodes int

+ type HotPathExecutor = telemetry.HotPathExecutor

+ func NewHotPathExecutor(ring *LockFreeRingBuffer) *HotPathExecutor

+ type LocationRegion string

+ const RegionAsia

+ const RegionEU

+ const RegionUS

+ type LockFreeRingBuffer = telemetry.LockFreeRingBuffer

+ func NewLockFreeRingBuffer() *LockFreeRingBuffer

+ type MapRegistry struct

+ func NewRegistry(actions ...action.AnyAction) *MapRegistry

+ func (r *MapRegistry) Actions() []action.AnyAction

+ func (r *MapRegistry) CompilePipeline(expr string) (action.Executable, error)

+ func (r *MapRegistry) Get(capability string) (action.AnyAction, bool)

+ func (r *MapRegistry) Register(capability string, a action.AnyAction)

+ type MemoryBranchJournal = journal.MemoryBranchJournal

+ func NewMemoryBranchJournal() *MemoryBranchJournal

+ type MemoryCostLedger = governance.MemoryCostLedger

+ func NewMemoryCostLedger() *MemoryCostLedger

+ type Metadata struct

+ Description string

+ Name string

+ Version string

+ type NodeKind string

+ const NodeApproval

+ const NodeDeterministic

+ const NodeHuman

+ const NodeLLM

+ const NodeSubgraph

+ const NodeTool

+ type NodeResult struct

+ Err error

+ Spawned SpawnedNode

+ Usage CostUsage

+ Value Res

+ type NodeSpec struct

+ Approval bool

+ BranchMode BranchMode

+ Capability string

+ Effect EffectClass

+ EstimatedCostMicros int64

+ ID string

+ InputBindings map[string]string

+ InputSchema string

+ Kind NodeKind

+ MaxAttempts int

+ OutputSchema string

+ Params map[string]any

+ Prompt string

+ Retry RetryPolicy

+ TimeoutMS int64

+ type PromptConfig = nodes.PromptConfig

+ type RecoveryPolicy struct

+ BackoffMS int64

+ MaxAttempts int

+ MaxBackoffMS int64

+ RetryTransientOnly bool

+ Strategy RecoveryStrategy

+ type RecoveryStrategy string

+ const RecoveryContinuePartial

+ const RecoveryFailFast

+ const RecoveryRetryFailed

+ type Registry interface

+ Actions func() []action.AnyAction

+ Get func(name string) (action.AnyAction, bool)

+ type RetryPolicy struct

+ Backoff string

+ MaxAttempts int

+ type SQLBranchJournal = journal.SQLBranchJournal

+ func NewSQLBranchJournal(db *sql.DB) *SQLBranchJournal

+ type SpawnedNode struct

+ Edge CompiledEdge

+ Input *State

+ RunID string

+ SourceNode string

+ SpawnIndex int

+ TargetNode string

+ func SpawnSelected(runID, source string, selected []CompiledEdge, input *State) ([]SpawnedNode, error)

+ type State struct

+ func FanIn[Res any](ctx context.Context, input *State, results []NodeResult[Res], ...) (*State, error)

+ func NewState(values map[string]any) *State

+ func NewStateFromDAG(dagState *dag.State) *State

+ func (s *State) Get(path string) (any, bool)

+ type SupervisorReq = nodes.SupervisorReq

+ type SupervisorRes = nodes.SupervisorRes

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL