flow

package module
v0.2.0 Latest Latest
Warning

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

Go to latest
Published: Sep 9, 2026 License: Apache-2.0 Imports: 30 Imported by: 0

README ΒΆ

Nexss Flow (nexssp/flow)

Go Version License CI

nexssp/flow is a universal, zero-infrastructure, Go-native Action Orchestration Engine & Arrow DSL.

It enables developers and AI agents to weave isolated, typed Go actions (github.com/nexssp/kernel/action) into high-throughput DAGs, scatter-gather parallel groups, resilient fallback chains, and dynamic data-shaping pipelines with zero Go glue code.


⚑ Key Invariants

  • Universal Action Engine: Orchestrates standard Go functions, microservices, DB queries, and AI Agents indiscriminately.
  • Inline Data Projections ({ ... }): Reshapes data between nodes on the fly using expr-lang/expr. No manual DTO adapters needed.
  • Compact Arrow DSL: Express complex execution topographies using concise inline text strings (->, |, &, ||, ?).
  • Zero-Allocation Hot Paths: Memory-pooled state management and pre-compiled expression bytecode execution.
  • Durable Branch Journaling: SQLite/SQL-backed branch recording for deterministic execution replay.
  • Cost & Budget Governance: Exact microdollar ledger tracking with hard-stop budget boundaries.

πŸ—οΈ Architecture

                  β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”
                  β”‚      Input Payload           β”‚
                  β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”¬β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜
                                 β”‚
                                 β–Ό
                  β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”
                  β”‚   github.get_issue           β”‚
                  β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”¬β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜
                                 β”‚
                                 β–Ό
                  β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”
                  β”‚ Inline Projection ({ ... })  β”‚
                  β”‚ Reshapes output for AI Agent β”‚
                  β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”¬β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜
                                 β”‚
                                 β–Ό
                  β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”
                  β”‚   ai.summarizer              β”‚
                  β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”¬β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜
                                 β”‚
                                 β–Ό
                  β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”
                  β”‚ Inline Projection ({ ... })  β”‚
                  β”‚ Reshapes output for Slack    β”‚
                  β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”¬β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜
                                 β”‚
                                 β–Ό
                  β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”
                  β”‚   slack.send                 β”‚
                  β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜

πŸ“¦ Installation

go get github.com/nexssp/flow@latest

πŸš€ Quickstart: Data-Shaping Pipeline

package main

import (
	"context"
	"fmt"

	"github.com/nexssp/flow"
	"github.com/nexssp/kernel/action"
)

func main() {
	ctx := context.Background()

	// 1. Define isolated Actions
	getIssue := action.New("github.get_issue", func(_ context.Context, id int) (map[string]any, error) {
		return map[string]any{
			"issue": map[string]any{"id": id, "title": "Nil pointer in scheduler", "body": "Line 42 panics"},
		}, nil
	}).Build()

	summarize := action.New("ai.summarizer", func(_ context.Context, req map[string]any) (map[string]any, error) {
		return map[string]any{"summary": "Critical panic in scheduler", "urgency": "HIGH"}, nil
	}).Build()

	sendSlack := action.New("slack.send", func(_ context.Context, req map[string]any) (string, error) {
		return fmt.Sprintf("Posted to %s: %s", req["channel"], req["text"]), nil
	}).Build()

	// 2. Register Actions
	registry := flow.NewRegistry(getIssue, summarize, sendSlack)

	// 3. Express Workflow in Arrow DSL with Inline Shaping
	dsl := `
		github.get_issue
		-> { prompt: "Summarize: " + issue.title + " - " + issue.body }
		-> ai.summarizer
		-> { channel: "#alerts", text: "🚨 [" + urgency + "] " + summary }
		-> slack.send
	`

	// 4. Compile & Execute
	builder, err := flow.CompilePipeline(dsl, registry)
	if err != nil {
		panic(err)
	}

	result, err := builder.Build().Do(ctx, 999)
	if err != nil {
		panic(err)
	}

	fmt.Println(result)
	// Output: Posted to #alerts: 🚨 [HIGH] Critical panic in scheduler
}

πŸ“– Arrow DSL Syntax Reference

Operator Syntax Description
Sequential Pipe A -> B or A | B Passes output of A as input to B.
Inline Projection { x: .y, z: "val" } Reshapes incoming data on the fly using expr.
Parallel Scatter-Gather (A & B & C) Executes A, B, and C concurrently; aggregates output.
Fallback Chain A || B Tries A. If A returns an error, executes B (FirstSuccess).
Conditional Branch gate ? target Evaluates boolean input from gate. If true, runs target.

πŸ› οΈ Capability Discovery for AI Agents

Expose registered capabilities so AI agents can dynamically inspect actions and synthesize Flow DSLs:

// Extract machine-readable capability specifications
capabilities := flow.ExtractCapabilities(registry)

// Expose /flow/catalog over HTTP/A2A
catalogAction := flow.BuildCatalogAction(registry)

Contributing

Read CONTRIBUTING.md before opening a pull request. Changes must be typed, covered by tests, and include benchmarks for performance-critical paths.

Security

Report suspected vulnerabilities privately according to SECURITY.md.

License

Apache License 2.0. See LICENSE. Copyright Β© 2018–2026 Marcin Polak and Contributors.

Documentation ΒΆ

Index ΒΆ

Constants ΒΆ

View Source
const (
	BranchSelected = journal.BranchSelected
	BranchSkipped  = journal.BranchSkipped
)
View Source
const APIVersion = "nexss.ai/v1"
View Source
const EdgeRingSize = 1024

Variables ΒΆ

This section is empty.

Functions ΒΆ

func AcquireStateFromGraphState ΒΆ

func AcquireStateFromGraphState(s *State) *dag.State

AcquireStateFromGraphState converts a graph.State into a pooled dag.State without manual map copying.

func ApprovalTokenFrom ΒΆ

func ApprovalTokenFrom(ctx context.Context) string

func AsCostHook ΒΆ

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

func BuildCatalogAction ΒΆ

func BuildCatalogAction(reg Registry) action.AnyAction

BuildCatalogAction returns a system action exposing the capability catalog over HTTP/A2A.

func BuildEdgeMeshAction ΒΆ

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

func ChargeBranch ΒΆ

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

func CompilePipeline ΒΆ

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

CompilePipeline parses the Arrow DSL into an AST and compiles it into an executable Builder.

func CompileSaga ΒΆ

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

CompileSaga parses Arrow DSL with embedded transaction rollbacks into a Saga Node.

func EvaluateCondition ΒΆ

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

func Execute ΒΆ

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

Execute is a type-safe generic invoker helper.

func NewExecuteAction ΒΆ

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

NewExecuteAction exposes dynamic graph execution over CLI, HTTP REST, MCP, and A2A.

func NewPromptNode ΒΆ

func NewPromptNode(cfg nodes.PromptConfig) action.AnyAction

func NewSupervisorNode ΒΆ

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

func WithApprovalGate ΒΆ

func WithApprovalGate(g ApprovalGate) func(*Compiler)

func WithApprovalToken ΒΆ

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

func WithJournal ΒΆ

func WithJournal(j BranchJournal) func(*Compiler)

func WithLedger ΒΆ

func WithLedger(l CostLedger) func(*Compiler)

Types ΒΆ

type ApprovalGate ΒΆ

type ApprovalGate interface {
	Check(ctx context.Context, actionName string, payload string, token string) error
}

ApprovalGate defines an interface for human-in-the-loop or policy approvals.

type BranchJournal ΒΆ

type BranchJournal = journal.BranchJournal

type BranchMode ΒΆ

type BranchMode string
const (
	BranchFirstMatch BranchMode = "first_match"
	BranchAllMatches BranchMode = "all_matches"
)

type BranchRecord ΒΆ

type BranchRecord = journal.BranchRecord

type BranchStatus ΒΆ

type BranchStatus = journal.BranchStatus

type BudgetCheck ΒΆ

type BudgetCheck = governance.BudgetCheck

type BudgetPolicy ΒΆ

type BudgetPolicy struct {
	MaxCostUSD          *float64 `json:"max_cost_usd,omitempty" yaml:"max_cost_usd,omitempty"`
	MaxCostUSDPerRun    *float64 `json:"max_cost_usd_per_run,omitempty" yaml:"max_cost_usd_per_run,omitempty"`
	MaxCostUSDPerDay    *float64 `json:"max_cost_usd_per_day,omitempty" yaml:"max_cost_usd_per_day,omitempty"`
	MaxCostUSDPerMonth  *float64 `json:"max_cost_usd_per_month,omitempty" yaml:"max_cost_usd_per_month,omitempty"`
	MaxCostUSDPerTenant *float64 `json:"max_cost_usd_per_tenant,omitempty" yaml:"max_cost_usd_per_tenant,omitempty"`
}

type CapabilitySpec ΒΆ

type CapabilitySpec struct {
	Name         string         `json:"name"`
	Description  string         `json:"description"`
	Route        string         `json:"route,omitempty"`
	Method       string         `json:"method,omitempty"`
	InputSchema  map[string]any `json:"input_schema"`
	OutputSchema map[string]any `json:"output_schema"`
	Tags         []string       `json:"tags,omitempty"`
	IsSystem     bool           `json:"is_system"`
}

CapabilitySpec represents a machine-readable action contract for AI agents & graph builders.

func ExtractCapabilities ΒΆ

func ExtractCapabilities(reg Registry) []CapabilitySpec

ExtractCapabilities converts a Registry into an AI-friendly CapabilitySpec catalog.

type ChildResult ΒΆ

type ChildResult = nodes.ChildResult

type ChildTask ΒΆ

type ChildTask = nodes.ChildTask

type CompiledBudget ΒΆ

type CompiledBudget struct {
	MaxCostMicrosPerRun    int64
	MaxCostMicrosPerDay    int64
	MaxCostMicrosPerMonth  int64
	MaxCostMicrosPerTenant int64
}

type CompiledEdge ΒΆ

type CompiledEdge struct {
	From      string
	To        string
	When      string
	Otherwise bool
	Priority  int
	Budget    CompiledBudget
}

type CompiledGraph ΒΆ

type CompiledGraph struct {
	Definition GraphDefinition
	Layers     [][]string
	NodeByID   map[string]NodeSpec
	Edges      []CompiledEdge
	Outgoing   map[string][]CompiledEdge
	EdgeKeys   []string
	Budget     CompiledBudget
}

func Compile ΒΆ

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

func LoadYAML ΒΆ

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

func LoadYAMLFile ΒΆ

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

func (*CompiledGraph) SelectOutgoing ΒΆ

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

func (*CompiledGraph) SelectOutgoingDurable ΒΆ

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

type Compiler ΒΆ

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

func NewCompiler ΒΆ

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

func (*Compiler) Compile ΒΆ

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

type CostEvent ΒΆ

type CostEvent = governance.CostEvent

type CostLedger ΒΆ

type CostLedger = governance.CostLedger

type CostReporter ΒΆ

type CostReporter interface {
	CostMicros() int64
}

CostReporter is an interface that nodes can implement to report runtime costs to the ledger.

type CostUsage ΒΆ

type CostUsage = governance.CostUsage

type DynamicPolicy ΒΆ

type DynamicPolicy struct {
	RetryCount int
	Timeout    time.Duration
	Idempotent bool
	Breaker    bool
}

type EdgeEventSlot ΒΆ

type EdgeEventSlot struct {
	Seq        uint64
	LatencyNs  int64
	StatusCode uint16

	EdgeID [16]byte
	// contains filtered or unexported fields
}

type EdgeNodeReq ΒΆ

type EdgeNodeReq struct {
	Region  LocationRegion `json:"region"`
	Payload map[string]any `json:"payload"`
}

type EdgeNodeRes ΒΆ

type EdgeNodeRes struct {
	Region     LocationRegion `json:"region"`
	NodeID     string         `json:"node_id"`
	LatencyMs  int64          `json:"latency_ms"`
	Data       map[string]any `json:"data"`
	StatusCode uint16         `json:"status_code"`
}

type EdgeSpec ΒΆ

type EdgeSpec struct {
	From      string       `json:"from" yaml:"from"`
	To        string       `json:"to" yaml:"to"`
	When      string       `json:"when,omitempty" yaml:"when,omitempty"`
	Otherwise bool         `json:"otherwise,omitempty" yaml:"otherwise,omitempty"`
	Priority  int          `json:"priority,omitempty" yaml:"priority,omitempty"`
	Budget    BudgetPolicy `json:"budget,omitempty" yaml:"budget,omitempty"`
}

type EdgeTelemetryRing ΒΆ

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

func NewEdgeTelemetryRing ΒΆ

func NewEdgeTelemetryRing() *EdgeTelemetryRing

func (*EdgeTelemetryRing) BatchDrain ΒΆ

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

func (*EdgeTelemetryRing) Push ΒΆ

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

type EffectClass ΒΆ

type EffectClass string
const (
	EffectReadOnly   EffectClass = "read_only"
	EffectSideEffect EffectClass = "side_effect"
	EffectHighRisk   EffectClass = "high_risk"
)

type EventSlot ΒΆ

type EventSlot = telemetry.EventSlot

type FanInPolicy ΒΆ

type FanInPolicy struct {
	RequireAll   bool
	AllowPartial bool
	FailOnEmpty  bool
}

type GraphDefinition ΒΆ

type GraphDefinition struct {
	APIVersion string      `json:"apiVersion" yaml:"apiVersion"`
	Kind       string      `json:"kind" yaml:"kind"`
	Metadata   Metadata    `json:"metadata" yaml:"metadata"`
	Policy     GraphPolicy `json:"policy,omitempty" yaml:"policy,omitempty"`
	Nodes      []NodeSpec  `json:"nodes" yaml:"nodes"`
	Edges      []EdgeSpec  `json:"edges" yaml:"edges"`
}

func ParseArrowDSL ΒΆ

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

ParseArrowDSL converts a compact arrow pipeline expression into a standard declarative GraphDefinition using the unified AST parser.

type GraphExecReq ΒΆ

type GraphExecReq struct {
	DSL            string         `json:"dsl,omitempty" cli:"dsl,d" usage:"Compact arrow pipeline expression"`
	YAML           string         `json:"yaml,omitempty" cli:"yaml,y" usage:"Declarative YAML graph manifest"`
	InitialPayload map[string]any `json:"initial_payload,omitempty" usage:"Initial state values passed to root nodes"`
}

GraphExecReq defines the omni-protocol input payload.

type GraphExecRes ΒΆ

type GraphExecRes struct {
	GraphName  string         `json:"graph_name"`
	Outputs    map[string]any `json:"outputs"`
	LayersRun  int            `json:"layers_run"`
	DurationMS int64          `json:"duration_ms"`
}

GraphExecRes defines the structured execution audit output.

type GraphPolicy ΒΆ

type GraphPolicy struct {
	Budget              BudgetPolicy   `json:"budget,omitempty" yaml:"budget,omitempty"`
	MaxParallelNodes    int            `json:"max_parallel_nodes,omitempty" yaml:"max_parallel_nodes,omitempty"`
	MaxContextBytes     int64          `json:"max_context_bytes,omitempty" yaml:"max_context_bytes,omitempty"`
	ApprovalRequiredFor []EffectClass  `json:"approval_required_for,omitempty" yaml:"approval_required_for,omitempty"`
	FanInRecovery       RecoveryPolicy `json:"fan_in_recovery,omitempty" yaml:"fan_in_recovery,omitempty"`
}

type HotPathExecutor ΒΆ

type HotPathExecutor = telemetry.HotPathExecutor

func NewHotPathExecutor ΒΆ

func NewHotPathExecutor(ring *LockFreeRingBuffer) *HotPathExecutor

type LocationRegion ΒΆ

type LocationRegion string
const (
	RegionEU   LocationRegion = "eu-central-1"
	RegionUS   LocationRegion = "us-east-1"
	RegionAsia LocationRegion = "ap-southeast-1"
)

type LockFreeRingBuffer ΒΆ

type LockFreeRingBuffer = telemetry.LockFreeRingBuffer

func NewLockFreeRingBuffer ΒΆ

func NewLockFreeRingBuffer() *LockFreeRingBuffer

type MapRegistry ΒΆ

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

func NewRegistry ΒΆ

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

func (*MapRegistry) Actions ΒΆ

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

func (*MapRegistry) CompilePipeline ΒΆ

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

func (*MapRegistry) Get ΒΆ

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

func (*MapRegistry) Register ΒΆ

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

type MemoryBranchJournal ΒΆ

type MemoryBranchJournal = journal.MemoryBranchJournal

func NewMemoryBranchJournal ΒΆ

func NewMemoryBranchJournal() *MemoryBranchJournal

type MemoryCostLedger ΒΆ

type MemoryCostLedger = governance.MemoryCostLedger

func NewMemoryCostLedger ΒΆ

func NewMemoryCostLedger() *MemoryCostLedger

type Metadata ΒΆ

type Metadata struct {
	Name        string `json:"name" yaml:"name"`
	Version     string `json:"version" yaml:"version"`
	Description string `json:"description,omitempty" yaml:"description,omitempty"`
}

type NodeKind ΒΆ

type NodeKind string
const (
	NodeDeterministic NodeKind = "deterministic"
	NodeLLM           NodeKind = "llm"
	NodeTool          NodeKind = "tool"
	NodeSubgraph      NodeKind = "subgraph"
	NodeHuman         NodeKind = "human"
	NodeApproval      NodeKind = "approval"
)

type NodeResult ΒΆ

type NodeResult[Res any] struct {
	Spawned SpawnedNode `json:"spawned"`
	Value   Res         `json:"value"`
	Err     error       `json:"error,omitempty"`
	Usage   CostUsage   `json:"usage"`
}

type NodeSpec ΒΆ

type NodeSpec struct {
	ID                  string            `json:"id" yaml:"id"`
	Kind                NodeKind          `json:"kind" yaml:"kind"`
	Capability          string            `json:"capability" yaml:"capability"`
	Params              map[string]any    `json:"params,omitempty" yaml:"params,omitempty"`
	InputBindings       map[string]string `json:"inputs,omitempty" yaml:"inputs,omitempty"`
	InputSchema         string            `json:"input_schema,omitempty" yaml:"input_schema,omitempty"`
	OutputSchema        string            `json:"output_schema,omitempty" yaml:"output_schema,omitempty"`
	Prompt              string            `json:"prompt,omitempty" yaml:"prompt,omitempty"`
	Retry               RetryPolicy       `json:"retry,omitempty" yaml:"retry,omitempty"`
	TimeoutMS           int64             `json:"timeout_ms,omitempty" yaml:"timeout_ms,omitempty"`
	MaxAttempts         int               `json:"max_attempts,omitempty" yaml:"max_attempts,omitempty"`
	EstimatedCostMicros int64             `json:"estimated_cost_micros,omitempty" yaml:"estimated_cost_micros,omitempty"`
	Effect              EffectClass       `json:"effect,omitempty" yaml:"effect,omitempty"`
	Approval            bool              `json:"approval_required,omitempty" yaml:"approval_required,omitempty"`
	BranchMode          BranchMode        `json:"branch_mode,omitempty" yaml:"branch_mode,omitempty"`
}

type PromptConfig ΒΆ

type PromptConfig = nodes.PromptConfig

type RecoveryPolicy ΒΆ

type RecoveryPolicy struct {
	Strategy           RecoveryStrategy `json:"strategy,omitempty" yaml:"strategy,omitempty"`
	MaxAttempts        int              `json:"max_attempts,omitempty" yaml:"max_attempts,omitempty"`
	BackoffMS          int64            `json:"backoff_ms,omitempty" yaml:"backoff_ms,omitempty"`
	MaxBackoffMS       int64            `json:"max_backoff_ms,omitempty" yaml:"max_backoff_ms,omitempty"`
	RetryTransientOnly bool             `json:"retry_transient_only,omitempty" yaml:"retry_transient_only,omitempty"`
}

type RecoveryStrategy ΒΆ

type RecoveryStrategy string
const (
	RecoveryFailFast        RecoveryStrategy = "fail_fast"
	RecoveryRetryFailed     RecoveryStrategy = "retry_failed"
	RecoveryContinuePartial RecoveryStrategy = "continue_partial"
)

type Registry ΒΆ

type Registry interface {
	Get(name string) (action.AnyAction, bool)
	Actions() []action.AnyAction
}

type RetryPolicy ΒΆ

type RetryPolicy struct {
	MaxAttempts int    `json:"max_attempts,omitempty" yaml:"max_attempts,omitempty"`
	Backoff     string `json:"backoff,omitempty" yaml:"backoff,omitempty"`
}

type SQLBranchJournal ΒΆ

type SQLBranchJournal = journal.SQLBranchJournal

func NewSQLBranchJournal ΒΆ

func NewSQLBranchJournal(db *sql.DB) *SQLBranchJournal

type SpawnedNode ΒΆ

type SpawnedNode struct {
	RunID      string       `json:"run_id"`
	SourceNode string       `json:"source_node"`
	Edge       CompiledEdge `json:"edge"`
	TargetNode string       `json:"target_node"`
	Input      *State       `json:"input"`
	SpawnIndex int          `json:"spawn_index"`
}

func SpawnSelected ΒΆ

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

SpawnSelected creates durable invocation units for each selected edge.

type State ΒΆ

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

func FanIn ΒΆ

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

FanIn deterministically orders parallel child outputs by SpawnIndex and calls reduce.

func NewState ΒΆ

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

func NewStateFromDAG ΒΆ

func NewStateFromDAG(dagState *dag.State) *State

func (*State) Get ΒΆ

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

type SupervisorReq ΒΆ

type SupervisorReq = nodes.SupervisorReq

type SupervisorRes ΒΆ

type SupervisorRes = nodes.SupervisorRes

Directories ΒΆ

Path Synopsis
nexssp/flow/compiler/ast.go
nexssp/flow/compiler/ast.go
examples
04_edge_mesh command
07_strong_typed command

Jump to

Keyboard shortcuts

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