graph

Documentation

Overview

Package graph builds, validates, compiles, and executes graph topology.

Index

• Constants
• func LoadGraphDefinitionFile(path string) (dsl.GraphDefinition, error)
• func NewGraphRunner(targetGraph *Graph, executionStore fruntime.ExecutionStore, ...) (*fruntime.GraphRunner, error)
• func SetLogger(l *zap.Logger)
• type Builder
  • func NewBuilder(reg *registry.Registry) *Builder
  • func (builder *Builder) Build(def dsl.GraphDefinition, ctx *registry.BuildContext) (*Graph, error)
  • func (builder *Builder) BuildFile(path string, ctx *registry.BuildContext) (*Graph, error)
  • func (builder *Builder) BuildInstance(def dsl.GraphDefinition, instance dsl.GraphInstanceConfig, ...) (*Graph, error)
• type ConditionError
  • func (conditionErr *ConditionError) Class() core.ErrorClass
  • func (conditionErr *ConditionError) Details() map[string]any
  • func (conditionErr *ConditionError) Error() string
  • func (conditionErr *ConditionError) RetryAfter() time.Duration
  • func (conditionErr *ConditionError) Unwrap() error
• type Graph
  • func NewGraph(reg *registry.Registry) *Graph
  • func (g *Graph) AddConditionalEdge(from, to string, condition registry.EdgeCondition) error
  • func (g *Graph) AddEdge(from, to string) error
  • func (g *Graph) AddFailureRoute(from, to string, route dsl.FailureRouteSpec) error
  • func (g *Graph) AddNode(targetNode core.Node) error
  • func (g *Graph) AddResolvedConditionalEdge(from, to string, condition registry.EdgeCondition, contract state.Contract) error
  • func (g *Graph) Compile() (*Runnable, error)
  • func (g *Graph) ContractDiagnostics() []core.ContractDiagnostic
  • func (g *Graph) Definition() (dsl.GraphDefinition, error)
  • func (g *Graph) DrawMermaid() (string, error)
  • func (g *Graph) ExecutionPolicy() fruntime.GraphExecutionPolicy
  • func (g *Graph) InitialStateRequirements() core.InitialStateRequirements
  • func (g *Graph) InitialStateRequirementsFor(provider *core.EntryStateProvider) core.InitialStateRequirements
  • func (g *Graph) NodeSpecs() map[string]dsl.GraphNodeSpec
  • func (g *Graph) Run(ctx context.Context, initialState *state.State) (*state.State, error)
  • func (g *Graph) SemanticHash() (string, error)
  • func (g *Graph) SetEntryPoint(ref string) error
  • func (g *Graph) SetExecutionPolicy(policy fruntime.GraphExecutionPolicy) error
  • func (g *Graph) SetFinishPoint(ref string) error
  • func (g *Graph) SetNodeExecutionPolicy(nodeID string, policy fruntime.ExecutionPolicy) error
  • func (g *Graph) SetNodeSpec(spec dsl.GraphNodeSpec) error
  • func (g *Graph) SnapshotHash() (string, error)
  • func (g *Graph) Validate() error
  • func (g *Graph) ValidateInitialState(initial *state.State) error
  • func (g *Graph) WriteToFile(path string) error
• type NodeEvent
• type Runnable
  • func (r *Runnable) Invoke(ctx context.Context, initialState *state.State) (*state.State, error)
  • func (r *Runnable) InvokeWithConfig(ctx context.Context, initialState *state.State, ...) (*state.State, error)
  • func (r *Runnable) Stream(ctx context.Context, initialState *state.State) <-chan StreamEvent
• type StreamEvent

Constants

const (
	EndNodeRef = dsl.EndNodeRef
)

Functions

func LoadGraphDefinitionFile

func LoadGraphDefinitionFile(path string) (dsl.GraphDefinition, error)

func NewGraphRunner

func NewGraphRunner(targetGraph *Graph, executionStore fruntime.ExecutionStore, checkpointStore fruntime.CheckpointStore, codec state.Codec, eventSink fruntime.EventSink, options ...fruntime.GraphRunnerOption) (*fruntime.GraphRunner, error)

func SetLogger

func SetLogger(l *zap.Logger)

Types

type Builder

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

func NewBuilder

func NewBuilder(reg *registry.Registry) *Builder

func (*Builder) Build

func (builder *Builder) Build(def dsl.GraphDefinition, ctx *registry.BuildContext) (*Graph, error)

func (*Builder) BuildFile

func (builder *Builder) BuildFile(path string, ctx *registry.BuildContext) (*Graph, error)

func (*Builder) BuildInstance

func (builder *Builder) BuildInstance(def dsl.GraphDefinition, instance dsl.GraphInstanceConfig, ctx *registry.BuildContext) (*Graph, error)

type ConditionError

type ConditionError struct {
	ConditionID   string
	ConditionType string
	SourceNodeID  string
	TargetNodeID  string
	StatePaths    []string
	Err           error
}

func (*ConditionError) Class

func (conditionErr *ConditionError) Class() core.ErrorClass

func (*ConditionError) Details

func (conditionErr *ConditionError) Details() map[string]any

func (*ConditionError) Error

func (conditionErr *ConditionError) Error() string

func (*ConditionError) RetryAfter

func (conditionErr *ConditionError) RetryAfter() time.Duration

func (*ConditionError) Unwrap

func (conditionErr *ConditionError) Unwrap() error

type Graph

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

Graph owns WeaveFlow topology, state contracts, and scheduler configuration. It centralizes project-level conventions such as: - registering nodes via Node - resolving nodes refs by ID - copy-on-write nodes invocation - serializable conditional edges

func NewGraph

func NewGraph(reg *registry.Registry) *Graph

func (*Graph) AddConditionalEdge

func (g *Graph) AddConditionalEdge(from, to string, condition registry.EdgeCondition) error

func (*Graph) AddEdge

func (g *Graph) AddEdge(from, to string) error

func (*Graph) AddFailureRoute

func (g *Graph) AddFailureRoute(from, to string, route dsl.FailureRouteSpec) error

func (*Graph) AddNode

func (g *Graph) AddNode(targetNode core.Node) error

func (*Graph) AddResolvedConditionalEdge

func (g *Graph) AddResolvedConditionalEdge(from, to string, condition registry.EdgeCondition, contract state.Contract) error

func (*Graph) Compile

func (g *Graph) Compile() (*Runnable, error)

func (*Graph) ContractDiagnostics

func (g *Graph) ContractDiagnostics() []core.ContractDiagnostic

func (*Graph) Definition

func (g *Graph) Definition() (dsl.GraphDefinition, error)

func (*Graph) DrawMermaid

func (g *Graph) DrawMermaid() (string, error)

func (*Graph) ExecutionPolicy

func (g *Graph) ExecutionPolicy() fruntime.GraphExecutionPolicy

func (*Graph) InitialStateRequirements

func (g *Graph) InitialStateRequirements() core.InitialStateRequirements

func (*Graph) InitialStateRequirementsFor

func (g *Graph) InitialStateRequirementsFor(provider *core.EntryStateProvider) core.InitialStateRequirements

func (*Graph) NodeSpecs

func (g *Graph) NodeSpecs() map[string]dsl.GraphNodeSpec

func (*Graph) Run

func (g *Graph) Run(ctx context.Context, initialState *state.State) (*state.State, error)

func (*Graph) SemanticHash

func (g *Graph) SemanticHash() (string, error)

func (*Graph) SetEntryPoint

func (g *Graph) SetEntryPoint(ref string) error

func (*Graph) SetExecutionPolicy

func (g *Graph) SetExecutionPolicy(policy fruntime.GraphExecutionPolicy) error

func (*Graph) SetFinishPoint

func (g *Graph) SetFinishPoint(ref string) error

func (*Graph) SetNodeExecutionPolicy

func (g *Graph) SetNodeExecutionPolicy(nodeID string, policy fruntime.ExecutionPolicy) error

func (*Graph) SetNodeSpec

func (g *Graph) SetNodeSpec(spec dsl.GraphNodeSpec) error

func (*Graph) SnapshotHash

func (g *Graph) SnapshotHash() (string, error)

func (*Graph) Validate

func (g *Graph) Validate() error

func (*Graph) ValidateInitialState

func (g *Graph) ValidateInitialState(initial *state.State) error

ValidateInitialState verifies that concrete invocation state satisfies every required read that is not produced by the Graph itself.

func (*Graph) WriteToFile

func (g *Graph) WriteToFile(path string) error

type NodeEvent

type NodeEvent string

const (
	EventChainStart   NodeEvent = "chain_start"
	EventChainEnd     NodeEvent = "chain_end"
	EventNodeStart    NodeEvent = "start"
	EventNodeComplete NodeEvent = "complete"
	EventNodeError    NodeEvent = "error"
)

type Runnable

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

func (*Runnable) Invoke

func (r *Runnable) Invoke(ctx context.Context, initialState *state.State) (*state.State, error)

func (*Runnable) InvokeWithConfig

func (r *Runnable) InvokeWithConfig(ctx context.Context, initialState *state.State, schedulerConfig fruntime.SchedulerConfig) (*state.State, error)

func (*Runnable) Stream

func (r *Runnable) Stream(ctx context.Context, initialState *state.State) <-chan StreamEvent

type StreamEvent

type StreamEvent struct {
	Timestamp time.Time
	NodeName  string
	Event     NodeEvent
	State     *state.State
	Error     error
	Duration  time.Duration
}