graph

type Checkpoint ¶

type Checkpoint[S any] struct {
	// RunID uniquely identifies the execution this checkpoint belongs to.
	RunID string `json:"run_id"`

	// StepID is the execution step number at checkpoint time.
	// Monotonically increasing within a run.
	StepID int `json:"step_id"`

	// State is the current accumulated state after applying all deltas up to StepID.
	// Must be JSON-serializable for persistence.
	State S `json:"state"`

	// Frontier contains the work items ready to execute at this checkpoint.
	// These represent the nodes queued for execution when resuming from this checkpoint.
	Frontier []WorkItem[S] `json:"frontier"`

	// RNGSeed is the seed for deterministic random number generation.
	// Computed from RunID to ensure consistent random values across replays.
	RNGSeed int64 `json:"rng_seed"`

	// RecordedIOs contains all captured external interactions up to this checkpoint.
	// Indexed by (NodeID, Attempt) for lookup during replay.
	RecordedIOs []RecordedIO `json:"recorded_ios"`

	// IdempotencyKey is a hash of (RunID, StepID, State, Frontier) that prevents.
	// duplicate checkpoint commits. Format: "sha256:hex_encoded_hash".
	IdempotencyKey string `json:"idempotency_key"`

	// Timestamp records when this checkpoint was created.
	Timestamp time.Time `json:"timestamp"`

	// Label is an optional user-defined name for this checkpoint, useful for.
	// debugging or creating named save points (e.g., "before_summary", "after_validation").
	// Empty string for automatic checkpoints.
	Label string `json:"label,omitempty"`
}

Checkpoint represents a durable snapshot of execution state, enabling. resumption and deterministic replay of graph executions.

Checkpoints are created atomically after each execution step and contain. all the information needed to resume from that point: - Current accumulated state. - Work items ready to execute (frontier). - Recorded I/O for replay. - RNG seed for deterministic random number generation. - Idempotency key for preventing duplicate commits.

Checkpoints support both automatic resumption after failures and. user-initiated labeled snapshots for debugging or branching workflows.

type ConflictPolicy ¶

type ConflictPolicy int

ConflictPolicy defines how concurrent state updates are handled when multiple branches. modify the same state fields.

This is a placeholder for future CRDT (Conflict-free Replicated Data Type) support. Currently only ConflictFail is supported, which returns an error on detected conflicts.

Planned policies: - ConflictFail: Return error on conflict (default, implemented). - LastWriterWins: Last merge wins based on OrderKey. - CRDT: Use conflict-free replicated data types for automatic resolution.

const (
	// ConflictFail returns an error when concurrent branches modify the same state field.
	// This is the safest default, requiring explicit conflict resolution.
	ConflictFail ConflictPolicy = iota

	// LastWriterWins uses the delta with the highest OrderKey when conflicts occur.
	// WARNING: Not yet implemented. Will return error if specified.
	LastWriterWins

	// ConflictCRDT uses CRDT semantics for automatic conflict resolution.
	// WARNING: Not yet implemented. Will return error if specified.
	ConflictCRDT
)

type CostTracker ¶

type CostTracker struct {
	// RunID associates costs with a specific workflow execution.
	RunID string

	// Currency is the cost unit (e.g., "USD").
	Currency string

	// Pricing maps model names to their input/output token costs.
	Pricing map[string]ModelPricing

	// Calls records all LLM invocations with full details.
	Calls []LLMCall

	// TotalCost accumulates all costs in the specified currency.
	TotalCost float64

	// ModelCosts tracks costs per model for attribution.
	ModelCosts map[string]float64

	// InputTokens counts total input tokens across all calls.
	InputTokens int64

	// OutputTokens counts total output tokens across all calls.
	OutputTokens int64

	// CreatedAt marks when cost tracking began.
	CreatedAt time.Time
	// contains filtered or unexported fields
}

CostTracker (T040) tracks financial costs associated with LLM API calls, providing detailed token usage and cost attribution for production monitoring.

Features: - Per-model token counting (input/output separate). - Accurate cost calculation using static pricing tables. - Cumulative cost tracking across multiple calls. - Per-model cost breakdown for attribution. - Thread-safe concurrent recording.

Pricing is based on static tables (defaultModelPricing) for major providers: - OpenAI: GPT-4o, GPT-4o-mini, GPT-4-turbo, GPT-3.5-turbo. - Anthropic: Claude 3.5 Sonnet, Claude 3 Opus/Sonnet/Haiku. - Google: Gemini 1.5 Pro/Flash, Gemini 1.0 Pro.

Usage:

// Create tracker for a run. tracker := NewCostTracker("run-123", "USD").

// Record LLM calls. tracker.RecordLLMCall("gpt-4o", 1000, 500, "nodeA"). tracker.RecordLLMCall("claude-3-sonnet", 2000, 800, "nodeB").

// Get total cost. total := tracker.GetTotalCost() // e.g., $0.0345.

// Get per-model breakdown. costs := tracker.GetCostByModel() // map[string]float64{"gpt-4o": 0.0125, "claude-3-sonnet": 0.0220}.

Thread-safe: All methods use mutex protection for concurrent access.

func NewCostTracker(runID, currency string) *CostTracker

func (ct *CostTracker) Disable()

func (ct *CostTracker) Enable()

func (ct *CostTracker) GetCallHistory() []LLMCall

func (ct *CostTracker) GetCostByModel() map[string]float64

// Output:

func (ct *CostTracker) GetTokenUsage() (inputTokens, outputTokens int64)

func (ct *CostTracker) GetTotalCost() float64

func (ct *CostTracker) RecordLLMCall(model string, inputTokens, outputTokens int, nodeID string) error

func (ct *CostTracker) Reset()

func (ct *CostTracker) SetCustomPricing(model string, inputPer1M, outputPer1M float64)

func (ct *CostTracker) String() string

type Edge ¶

type Edge[S any] struct {
	// From is the source node ID.
	From string

	// To is the destination node ID.
	To string

	// When is an optional predicate that determines if this edge should be traversed.
	// If nil, the edge is unconditional (always traverse).
	// If non-nil, the edge is only traversed when When(state) returns true.
	When Predicate[S]
}

Edge represents a connection between two nodes in the workflow graph.

Edges define the control flow between nodes. They can be: - Unconditional: Always traverse (When = nil). - Conditional: Only traverse if predicate returns true (When != nil).

Edges are used during graph construction to define possible transitions. At runtime, the Engine evaluates predicates to determine which edge to follow.

For explicit routing, nodes can return Next in NodeResult which overrides. edge-based routing.

Type parameter S is the state type used for predicate evaluation.

type Engine ¶

type Engine[S any] struct {
	// contains filtered or unexported fields
}

Engine orchestrates stateful workflow execution with checkpointing support.

The Engine is the core runtime that:

• Manages workflow graph topology (nodes and edges)
• Executes nodes in sequence or parallel
• Merges state updates via the reducer
• Persists state at each step via the store
• Emits observability events via the emitter
• Enforces execution limits (MaxSteps, Retries)
• Supports checkpoint save/resume

Concurrency Model ¶

The Engine supports two execution modes:

**Sequential Mode** (MaxConcurrentNodes = 0):

• Nodes execute one at a time in strict order
• Deterministic execution: Same inputs → Same outputs
• Best for workflows requiring strict ordering or replay guarantees
• Lower resource usage, simpler debugging

**Concurrent Mode** (MaxConcurrentNodes > 0):

• Multiple nodes execute simultaneously (bounded by MaxConcurrentNodes)
• Worker pool with FIFO work stealing from priority heap
• Higher throughput for parallel workflows
• Recommended: MaxConcurrentNodes = NumCPU for CPU-bound tasks

Concurrency Safety Mechanisms (Bug Fixes 2025-10-29) ¶

The Engine implements four critical concurrency patterns:

**1. Per-Worker RNG Derivation (BUG-002 Fix)**

• Each worker receives a unique deterministic RNG instance
• Derived from RunID + WorkerID to ensure thread safety
• Enables deterministic random values in concurrent execution
• Context key: RNGKey (*rand.Rand)

**2. Heap-Based Priority Ordering (BUG-003 Fix)**

• Frontier uses heap as single source of truth
• Channel provides notifications only (empty struct{})
• Ensures deterministic OrderKey-based merge ordering
• Prevents out-of-order processing in concurrent scenarios

**3. Buffered Error Delivery (BUG-001 Fix)**

• Results channel sized at MaxConcurrentNodes*2
• sendErrorAndCancel blocks on error delivery (errors are rare + critical)
• Guarantees 100% error delivery rate
• Prevents deadlocks under simultaneous error conditions

**4. Atomic Completion Detection (BUG-004 Fix)**

• CompareAndSwap-based completion flag (race-free)
• Workers check completion after dequeue failure and node execution
• 290x faster than polling (36µs vs 10.5ms)
• Zero premature or delayed termination scenarios

Deterministic Replay Guarantees ¶

When using the same RunID across multiple executions:

• Sequential mode (MaxConcurrentNodes=0): 100% deterministic (byte-identical states)
• Concurrent mode: Deterministic when using OrderKey for merge ordering
• RNG sequences identical across replays (per-worker seed derivation from RunID)
• Retry delays and backoff jitter deterministic (RNG-based with deterministic seed)

Validation Results (see CHANGELOG.md "Deterministic Replay Validation"):

• 1000-iteration stress test: byte-identical final states across all runs
• RNG sequence validation: 100 runs with identical random value sequences
• OrderKey merge consistency: 50 runs with identical parallel branch merge order
• Throughput: 9,750 executions/sec with zero determinism failures

Prerequisites for deterministic replay:

1. Use consistent RunID across replay attempts
2. Nodes must use context-provided RNG (ctx.Value(RNGKey)), not global rand
3. External I/O (network, filesystem) must be mocked or use deterministic responses
4. System time calls (time.Now()) should use context-provided clock for testing

See graph/replay_validation_test.go for comprehensive validation test suite.

Type parameter S is the state type shared across the workflow.

Example:

reducer := func(prev, delta MyState) MyState {
    if delta.Query != "" {
        prev.Query = delta.Query
    }
    prev.Steps++
    return prev
}

store := store.NewMemStore[MyState]()
emitter := emit.NewLogEmitter()
opts := Options{
    MaxSteps: 100,
    MaxConcurrentNodes: 4,  // 4 parallel workers
}

engine := New(reducer, store, emitter, opts)
engine.Add("process", processNode)
engine.StartAt("process")

final, err := engine.Run(ctx, "run-001", MyState{Query: "hello"})

func New[S any](reducer Reducer[S], st store.Store[S], emitter emit.Emitter, options ...interface{}) *Engine[S]

engine := New(reducer, store, emitter, Options{MaxSteps: 100})

engine := New(
    reducer, store, emitter,
    WithMaxConcurrent(16),
    WithQueueDepth(2048),
    WithDefaultNodeTimeout(10*time.Second),
)

baseOpts := Options{MaxSteps: 100}
engine := New(
    reducer, store, emitter,
    baseOpts,
    WithMaxConcurrent(8), // Overrides baseOpts if specified
)

func (e *Engine[S]) Add(nodeID string, node Node[S]) error

processNode := NodeFunc[MyState](func(ctx context.Context, s MyState) NodeResult[MyState] {
    return NodeResult[MyState]{
        Delta: MyState{Result: "processed"},
        Route: Stop(),
    }
})

err := engine.Add("process", processNode)

func (e *Engine[S]) Connect(from, to string, predicate Predicate[S]) error

// Unconditional edge
engine.Connect("nodeA", "nodeB", nil)

// Conditional edge
engine.Connect("router", "pathA", func(s MyState) bool {
    return s.Score > 0.8
})

func (e *Engine[S]) ReplayRun(ctx context.Context, runID string) (S, error)

// Original execution with I/O recording
opts := Options{ReplayMode: false} // Record mode
engine := New(reducer, store, emitter, opts)
_, err := engine.Run(ctx, "run-001", initialState)

// Later: replay execution for debugging
replayOpts := Options{ReplayMode: true, StrictReplay: true}
replayEngine := New(reducer, store, emitter, replayOpts)
replayedState, err := replayEngine.ReplayRun(ctx, "run-001")
if err != nil {
    log.Fatal(err)
}
fmt.Printf("Replayed state: %+v\n", replayedState)

func (e *Engine[S]) ResumeFromCheckpoint(ctx context.Context, cpID string, newRunID string, startNode string) (S, error)

// Original run with checkpoint
_, _ = engine.Run(ctx, "run-001", initial)
_ = engine.SaveCheckpoint(ctx, "run-001", "after-validation")

// Resume from checkpoint (e.g., after crash or for A/B test)
finalA, _ := engine.ResumeFromCheckpoint(ctx, "after-validation", "run-002-pathA", "pathA")
finalB, _ := engine.ResumeFromCheckpoint(ctx, "after-validation", "run-003-pathB", "pathB")

func (e *Engine[S]) Run(ctx context.Context, runID string, initial S) (S, error)

ctx := context.Background()
final, err := engine.Run(ctx, "run-001", MyState{Query: "hello"})
if err != nil {
    log.Fatal(err)
}
fmt.Printf("Final state: %+v\n", final)

func (e *Engine[S]) RunWithCheckpoint(ctx context.Context, checkpoint store.CheckpointV2[S]) (S, error)

// Original execution with checkpoint
_, _ = engine.Run(ctx, "run-001", initialState)

// Load checkpoint from store
checkpoint, err := store.LoadCheckpointV2(ctx, "run-001", 5)
if err != nil {
    log.Fatal(err)
}

// Resume execution from checkpoint
final, err := engine.RunWithCheckpoint(ctx, checkpoint)
if err != nil {
    log.Fatal(err)
}
fmt.Printf("Resumed execution completed: %+v\n", final)

func (e *Engine[S]) SaveCheckpoint(ctx context.Context, runID string, cpID string) error

// Run workflow
final, _ := engine.Run(ctx, "run-001", initial)

// Save checkpoint at completion
err := engine.SaveCheckpoint(ctx, "run-001", "before-deploy")
if err != nil {
    log.Fatal(err)
}

// Later, can resume from this checkpoint
resumed, _ := engine.ResumeFromCheckpoint(ctx, "before-deploy", "run-002")

func (e *Engine[S]) StartAt(nodeID string) error

engine.Add("start", startNode)
engine.StartAt("start")

type EngineError ¶

type EngineError struct {
	Message string
	Code    string
}

EngineError represents an error from Engine operations.

func (e *EngineError) Error() string

type Frontier ¶

type Frontier[S any] struct {
	// contains filtered or unexported fields
}

Frontier manages the work queue for concurrent graph execution with bounded capacity and deterministic ordering. It combines a priority queue (heap) for ordering with a buffered channel for bounded queue depth and backpressure.

The Frontier ensures that work items are dequeued in deterministic order (by OrderKey) even when they are enqueued concurrently from multiple goroutines. This is critical for deterministic replay of graph executions.

Architecture (BUG-003 fix): - Heap is the single source of truth for work item storage and ordering - Channel carries empty struct notifications only (not data) - Enqueue: Push to heap THEN send notification to channel - Dequeue: Wait for notification from channel THEN pop from heap This prevents heap/channel desynchronization and reduces memory usage.

The bounded channel provides backpressure: when the queue is full, Enqueue will block until capacity becomes available or the context is cancelled. This prevents unbounded memory growth when nodes produce work faster than it can be consumed.

Thread-safety: All methods are safe for concurrent use by multiple goroutines.

func NewFrontier[S any](ctx context.Context, capacity int, runID string, metrics *PrometheusMetrics, emitter emit.Emitter) *Frontier[S]

func (f *Frontier[S]) Dequeue(ctx context.Context) (WorkItem[S], error)

func (f *Frontier[S]) Enqueue(ctx context.Context, item WorkItem[S]) error

func (f *Frontier[S]) Len() int

func (f *Frontier[S]) Metrics() SchedulerMetrics

type LLMCall ¶

type LLMCall struct {
	Model        string    // Model identifier (e.g., "gpt-4o", "claude-3-sonnet")
	InputTokens  int       // Number of input tokens consumed
	OutputTokens int       // Number of output tokens generated
	CostUSD      float64   // Calculated cost in USD
	Timestamp    time.Time // When the call was made
	NodeID       string    // Node that made the call (optional)
}

LLMCall represents a single LLM API invocation with token usage and cost.

type ModelPricing ¶

type ModelPricing struct {
	InputPer1M  float64 // Cost per 1M input tokens in USD
	OutputPer1M float64 // Cost per 1M output tokens in USD
}

ModelPricing defines input and output token costs for LLM models. Prices are in USD per 1M tokens (per million tokens).

type Next ¶

type Next struct {
	// To specifies the next single node to execute.
	// Mutually exclusive with Many and Terminal.
	To string

	// Many specifies multiple nodes to execute in parallel (fan-out).
	// Mutually exclusive with To and Terminal.
	Many []string

	// Terminal indicates workflow execution should stop.
	// Mutually exclusive with To and Many.
	Terminal bool
}

Next specifies the next step(s) in workflow execution after a node completes.

It supports three routing modes:

• Terminal: Stop execution (Route.Terminal = true)
• Single: Go to a specific node (Route.To = "nodeID")
• Fan-out: Go to multiple nodes in parallel (Route.Many = []string{"node1", "node2"})

func Goto(nodeID string) Next

func Many(nodeIDs []string) Next

func Stop() Next

type Node ¶

type Node[S any] interface {
	// Run executes the node's logic with the given context and state.
	// It returns a NodeResult containing state changes, routing decisions,
	// events, and any errors encountered.
	Run(ctx context.Context, state S) NodeResult[S]
}

Node represents a processing unit in the workflow graph. It receives state of type S, performs computation, and returns a NodeResult.

Nodes are the fundamental building blocks of LangGraph workflows. Each node can:

• Access the current state
• Perform computation (call LLMs, tools, or custom logic)
• Return state modifications via Delta
• Control routing via Route
• Emit observability events
• Handle errors

Type parameter S is the state type shared across the workflow.

Optional Methods:

Nodes can optionally implement Policy() NodePolicy to configure execution behavior such as timeouts, retry policies, and idempotency keys. If not implemented, the engine uses default settings from Options.

To maintain backward compatibility, embed DefaultPolicy in your node struct to get a zero-value Policy() implementation:

type MyNode struct {
    DefaultPolicy  // Provides Policy() returning zero value
    // ... your fields
}

For custom policies, implement the method directly:

func (n *MyNode) Policy() NodePolicy {
    return NodePolicy{
        Timeout: 30 * time.Second,
        RetryPolicy: &RetryPolicy{MaxAttempts: 3},
    }
}

Nodes can also optionally implement Effects() SideEffectPolicy to declare their I/O characteristics for deterministic replay. If not implemented, the node is assumed to be pure (no external side effects).

For nodes with external I/O (LLM calls, API requests, database queries):

func (n *MyNode) Effects() SideEffectPolicy {
    return SideEffectPolicy{
        Recordable:          true,  // I/O can be captured for replay
        RequiresIdempotency: true,  // Use idempotency keys
    }
}

Pure nodes (computation only, no I/O) don't need to implement Effects() - the default zero value indicates no side effects.

type NodeError ¶

type NodeError struct {
	// Message is the human-readable error description.
	Message string

	// Code is a machine-readable error code for programmatic handling.
	Code string

	// NodeID identifies which node produced this error.
	NodeID string

	// Cause is the underlying error that caused this NodeError.
	Cause error
}

NodeError represents an error that occurred during node execution. It provides structured error information for better observability and debugging.

func (e *NodeError) Error() string

func (e *NodeError) Unwrap() error

type NodeFunc ¶

type NodeFunc[S any] func(ctx context.Context, state S) NodeResult[S]

NodeFunc is a function adapter that implements the Node interface. It allows using plain functions as nodes without creating custom types.

Example:

processNode := NodeFunc[MyState](func(ctx context.Context, s MyState) NodeResult[MyState] {
    return NodeResult[MyState]{
        Delta: MyState{Result: "processed"},
        Route: Stop(),
    }
})

func (f NodeFunc[S]) Run(ctx context.Context, state S) NodeResult[S]

type NodePolicy ¶

type NodePolicy struct {
	// Timeout is the maximum execution time allowed for this node.
	// If zero, Options.DefaultNodeTimeout is used.
	Timeout time.Duration

	// RetryPolicy specifies automatic retry behavior for transient failures.
	// If nil, no retries are attempted.
	RetryPolicy *RetryPolicy

	// IdempotencyKeyFunc generates a custom idempotency key from the state.
	// If nil, a default key based on node ID and step ID is used.
	// This is useful for side-effecting nodes that need exactly-once semantics.
	IdempotencyKeyFunc func(state any) string
}

NodePolicy configures the execution behavior for a specific node, including. timeouts, retry logic, and idempotency key generation.

Policies are attached to nodes and enforced by the scheduler. If not specified, default values from Options are used.

type NodeResult ¶

type NodeResult[S any] struct {
	// Delta is the partial state update produced by this node.
	// It will be merged with the current state using the configured reducer.
	Delta S

	// Route specifies the next step(s) in workflow execution.
	// Use Stop() for terminal nodes, Goto(id) for explicit routing,
	// or set Many for fan-out to multiple nodes.
	Route Next

	// Err contains any error that occurred during node execution.
	// Non-nil errors halt the workflow unless custom error handling is implemented.
	Err error
}

NodeResult represents the output of a node execution.

It contains all information needed to continue workflow execution:

• Delta: Partial state update to be merged via reducer
• Route: Next hop(s) for execution flow
• Events: Observability events emitted during execution
• Err: Node-level error (if any)

type Option ¶

type Option func(*engineConfig) error

Option is a functional option for configuring an Engine.

Functional options provide a clean, extensible API for engine configuration: - Chainable: engine := New(reducer, store, emitter, WithMaxConcurrent(8), WithQueueDepth(1024)). - Self-documenting: Option names clearly describe their purpose. - Optional: Only specify the configuration you need. - Backward compatible: Existing Options struct still works.

Example:

engine := graph.New(.

reducer,
store,
emitter,
graph.WithMaxConcurrent(16),
graph.WithQueueDepth(2048),
graph.WithDefaultNodeTimeout(10*time.Second),

).

Options can be mixed with the Options struct:

opts := graph.Options{MaxSteps: 100}. engine := graph.New(.

reducer,
store,
emitter,
opts,

graph.WithMaxConcurrent(8), // Overrides opts if specified. ).

func WithBackpressureTimeout(d time.Duration) Option

reducer, store, emitter,
graph.WithMaxConcurrent(8),

func WithConflictPolicy(policy ConflictPolicy) Option

reducer, store, emitter,

func WithCostTracker(tracker *CostTracker) Option

tracker := graph.NewCostTracker("run-123", "USD")
engine := graph.New(
    reducer, store, emitter,
    graph.WithCostTracker(tracker),
)

// After execution, get cost summary
totalCost := tracker.GetTotalCost()
modelCosts := tracker.GetCostByModel()

func WithDefaultNodeTimeout(d time.Duration) Option

reducer, store, emitter,

func WithMaxConcurrent(n int) Option

reducer, store, emitter,

func WithMaxSteps(n int) Option

engine := graph.New(
    reducer, store, emitter,
    graph.WithMaxSteps(100), // Limit to 100 execution steps
)

func WithMetrics(metrics *PrometheusMetrics) Option

registry := prometheus.NewRegistry()
metrics := graph.NewPrometheusMetrics(registry)
engine := graph.New(
    reducer, store, emitter,
    graph.WithMetrics(metrics),
)

// Expose metrics endpoint
http.Handle("/metrics", promhttp.HandlerFor(registry, promhttp.HandlerOpts{}))

func WithQueueDepth(n int) Option

reducer, store, emitter,
graph.WithMaxConcurrent(8),

func WithReplayMode(enabled bool) Option

func WithRunWallClockBudget(d time.Duration) Option

reducer, store, emitter,

func WithStrictReplay(enabled bool) Option

reducer, store, emitter,
graph.WithReplayMode(true),

type Options ¶

type Options struct {
	// MaxSteps limits workflow execution to prevent infinite loops.
	// If 0, no limit is enforced (use with caution).
	//
	// Workflow loops (A → B → A) are fully supported. Use MaxSteps to prevent
	// infinite loops when a conditional exit is missing or misconfigured.
	//
	// Loop patterns:
	//   1. Node-based conditional loop:
	//        nodeA returns Goto("B") if shouldContinue(state), else Stop()
	//   2. Edge predicate loop:
	//        Connect("A", "B", loopPredicate)
	//        Connect("A", "exit", exitPredicate)
	//
	// Recommended values:
	//   - Simple workflows (3-5 nodes): MaxSteps = 20
	//   - Workflows with loops: MaxSteps = depth × max_iterations (e.g., 5 nodes × 10 iterations = 50)
	//   - Complex multi-loop workflows: MaxSteps = 100-200
	//
	// When MaxSteps is exceeded, Run() returns EngineError with code "MAX_STEPS_EXCEEDED".
	MaxSteps int

	// Retries specifies how many times to retry a node on transient errors.
	// If 0, nodes are not retried.
	// Transient errors are identified by checking if error implements Retryable interface.
	// Deprecated: Use NodePolicy.RetryPolicy for per-node retry configuration.
	Retries int

	// MaxConcurrentNodes limits the number of nodes executing in parallel.
	// Default: 8. Set to 0 for sequential execution (backward compatible).
	//
	// Tuning guidance:
	//   - CPU-bound workflows: Set to runtime.NumCPU()
	//   - I/O-bound workflows: Set to 10-50 depending on external service limits
	//   - Memory-constrained: Reduce to prevent excessive state copies
	//
	// Each concurrent node holds a deep copy of state, so memory usage scales
	// linearly with MaxConcurrentNodes.
	MaxConcurrentNodes int

	// QueueDepth sets the capacity of the execution frontier queue.
	// Default: 1024. Increase for workflows with large fan-outs.
	//
	// When the queue fills, new work items block until space is available.
	// This provides backpressure to prevent unbounded memory growth.
	//
	// Recommended: MaxConcurrentNodes × 100 for initial estimate.
	QueueDepth int

	// BackpressureTimeout is the maximum time to wait when the frontier queue is full.
	// Default: 30s. If exceeded, Run() returns ErrBackpressureTimeout.
	//
	// Set lower for fast-failing systems, higher for workflows with bursty fan-outs.
	BackpressureTimeout time.Duration

	// DefaultNodeTimeout is the maximum execution time for nodes without explicit Policy().Timeout.
	// Default: 30s. Individual nodes can override via NodePolicy.Timeout.
	//
	// Prevents single slow nodes from blocking workflow progress indefinitely.
	// When exceeded, node execution is cancelled and returns context.DeadlineExceeded.
	DefaultNodeTimeout time.Duration

	// RunWallClockBudget is the maximum total execution time for Run().
	// Default: 10m. If exceeded, Run() returns context.DeadlineExceeded.
	//
	// Use this to enforce hard deadlines on entire workflow execution.
	// Set to 0 to disable (workflow runs until completion or MaxSteps).
	RunWallClockBudget time.Duration

	// ReplayMode enables deterministic replay using recorded I/O.
	// Default: false (record mode - captures I/O for later replay).
	//
	// When true, nodes with SideEffectPolicy.Recordable=true will use recorded
	// responses instead of executing live I/O. This enables:
	//   - Debugging: Replay production executions locally
	//   - Testing: Verify workflow logic without external dependencies
	//   - Auditing: Reconstruct exact execution flow from checkpoints
	//
	// Requires prior execution with ReplayMode=false to record I/O.
	ReplayMode bool

	// StrictReplay controls replay mismatch behavior.
	// Default: true (fail on I/O hash mismatch).
	//
	// When true, replay mode verifies recorded I/O hashes match expected values.
	// If a mismatch is detected (indicating logic changes), Run() returns ErrReplayMismatch.
	//
	// Set to false to allow "best effort" replay that tolerates minor changes.
	// Useful when debugging with modified node logic.
	StrictReplay bool

	// Metrics enables Prometheus metrics collection (T044).
	// If nil, metrics are not collected.
	//
	// Create with NewPrometheusMetrics(registry) to enable production monitoring.
	// All 6 metrics (inflight_nodes, queue_depth, step_latency_ms, retries_total,
	// merge_conflicts_total, backpressure_events_total) are automatically updated.
	//
	// Example:
	//   registry := prometheus.NewRegistry()
	//   metrics := NewPrometheusMetrics(registry)
	//   engine := New(reducer, store, emitter, Options{Metrics: metrics})
	Metrics *PrometheusMetrics

	// CostTracker enables LLM cost tracking with static pricing (T045).
	// If nil, cost tracking is disabled.
	//
	// Create with NewCostTracker(runID, "USD") to track token usage and costs.
	// Static pricing includes OpenAI, Anthropic, and Google models.
	//
	// Example:
	//   tracker := NewCostTracker("run-123", "USD")
	//   engine := New(reducer, store, emitter, Options{CostTracker: tracker})
	CostTracker *CostTracker
}

Options configures Engine execution behavior.

Zero values are valid - the Engine will use sensible defaults.

type Predicate ¶

type Predicate[S any] func(state S) bool

Predicate is a function that evaluates state to determine if an edge should be traversed.

Predicates enable conditional routing based on workflow state. They should be pure functions (deterministic, no side effects).

Common patterns: - Threshold: state.Score > 0.8. - Presence: state.Result != "". - Boolean flag: state.IsReady. - Complex logic: state.Retries < 3 && state.Error == nil.

Type parameter S is the state type to evaluate.

type PrometheusMetrics ¶

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

PrometheusMetrics (T032) provides comprehensive Prometheus-compatible metrics. collection for graph execution monitoring in production environments.

Metrics exposed (all namespaced with "langgraph_"):

1. inflight_nodes (gauge): Current number of nodes executing concurrently. Labels: run_id, graph_id. Use: Monitor concurrency levels and detect bottlenecks.

2. queue_depth (gauge): Number of pending nodes waiting for execution. Labels: run_id, graph_id. Use: Track backpressure and queue saturation.

3. step_latency_ms (histogram): Node execution duration in milliseconds. Labels: run_id, node_id, status (success/error). Buckets: [1, 5, 10, 50, 100, 500, 1000, 5000, 10000]. Use: P50/P95/P99 latency analysis per node.

4. retries_total (counter): Cumulative retry attempts across all nodes. Labels: run_id, node_id, reason. Use: Identify flaky nodes and error patterns.

5. merge_conflicts_total (counter): Concurrent state merge conflicts detected. Labels: run_id, conflict_type. Use: Monitor determinism violations in concurrent execution.

6. backpressure_events_total (counter): Queue saturation events triggering backpressure. Labels: run_id, reason. Use: Track when execution is throttled due to resource limits.

Usage:

// Create metrics with custom registry. registry := prometheus.NewRegistry(). metrics := NewPrometheusMetrics(registry).

// Integrate with engine. engine := New[MyState](.

WithMetrics(metrics),

).

// Metrics automatically update during execution.

// Expose via HTTP for Prometheus scraping:

http.Handle("/metrics", promhttp.HandlerFor(registry, promhttp.HandlerOpts{})).

Thread-safe: All methods use atomic operations or mutex protection.

func NewPrometheusMetrics(registry prometheus.Registerer) *PrometheusMetrics

func (pm *PrometheusMetrics) Disable()

func (pm *PrometheusMetrics) Enable()

func (pm *PrometheusMetrics) IncrementBackpressure(runID, reason string)

func (pm *PrometheusMetrics) IncrementMergeConflicts(runID, conflictType string)

func (pm *PrometheusMetrics) IncrementRetries(runID, nodeID, reason string)

	    // Retry logic...
}.

func (pm *PrometheusMetrics) RecordStepLatency(runID, nodeID string, latency time.Duration, status string)

func (pm *PrometheusMetrics) Reset()

func (pm *PrometheusMetrics) UpdateInflightNodes(count int)

func (pm *PrometheusMetrics) UpdateQueueDepth(depth int)

type RecordedIO ¶

type RecordedIO struct {
	// NodeID identifies the node that performed this I/O operation.
	NodeID string `json:"node_id"`

	// Attempt is the retry attempt number this I/O corresponds to.
	// This allows matching I/O recordings to specific retry attempts.
	Attempt int `json:"attempt"`

	// Request is the serialized request data sent to the external service.
	// Stored as JSON for cross-language compatibility and human readability.
	Request json.RawMessage `json:"request"`

	// Response is the serialized response data received from the external service.
	// Stored as JSON for cross-language compatibility and human readability.
	Response json.RawMessage `json:"response"`

	// Hash is a SHA-256 hash of the response content, used for mismatch detection.
	// during replay. Format: "sha256:hex_encoded_hash".
	Hash string `json:"hash"`

	// Timestamp records when this I/O operation was captured.
	Timestamp time.Time `json:"timestamp"`

	// Duration is how long the I/O operation took to complete.
	// This can be used for performance analysis and replay simulation.
	Duration time.Duration `json:"duration"`
}

RecordedIO captures an external interaction (API call, database query, etc.). for deterministic replay without re-invoking the external service.

During initial execution, RecordedIO instances are created for nodes with. SideEffectPolicy.Recordable=true. During replay, these recorded I/Os are. matched by (NodeID, Attempt) and their responses are returned directly. without executing the node.

The Hash field enables mismatch detection: if a live execution produces. a different response hash than recorded, ErrReplayMismatch is raised, indicating non-deterministic behavior.

type Reducer ¶

type Reducer[S any] func(prev, delta S) S

Reducer is a function that merges a partial state update (delta) into the previous state.

Reducers are responsible for deterministic state composition, enabling:

• Sequential state updates across workflow nodes
• Parallel branch result merging
• Checkpoint-based state reconstruction

The reducer must be:

• **Pure**: Same inputs always produce same output
• **Deterministic**: No randomness or side effects
• **Commutative** (for parallel merges): Order of deltas shouldn't matter for correctness

Example:

reducer := func(prev, delta MyState) MyState {
    if delta.Query != "" {
        prev.Query = delta.Query  // Last write wins
    }
    prev.Counter += delta.Counter  // Accumulate
    return prev
}

type RetryPolicy ¶

type RetryPolicy struct {
	// MaxAttempts is the maximum number of execution attempts (including initial attempt).
	// Must be >= 1. A value of 1 means no retries.
	MaxAttempts int

	// BaseDelay is the base delay for exponential backoff between retries.
	// The actual delay is computed as: min(BaseDelay * 2^attempt + jitter, MaxDelay).
	BaseDelay time.Duration

	// MaxDelay is the maximum delay cap for exponential backoff.
	// Must be >= BaseDelay.
	MaxDelay time.Duration

	// Retryable is a predicate function that determines if an error is retryable.
	// If nil, all errors are considered non-retryable.
	// Common patterns:
	// - Network errors: temporary, connection refused, timeout.
	// - HTTP 429, 503, 504.
	// - Database deadlocks.
	Retryable func(error) bool
}

RetryPolicy defines automatic retry configuration for transient node failures.

When a node execution fails, the retry policy determines whether the failure. is retryable and how long to wait before the next attempt. Exponential backoff. with jitter is used to avoid thundering herd problems.

func (rp *RetryPolicy) Validate() error

type SchedulerMetrics ¶

type SchedulerMetrics struct {
	// ActiveNodes is the current number of nodes executing concurrently.
	// This value should never exceed MaxConcurrentNodes.
	// Use atomic.LoadInt32() to read safely from multiple goroutines.
	ActiveNodes int32

	// QueueDepth is the current number of work items waiting in the frontier.
	// High queue depth indicates work is being produced faster than consumed.
	// When this reaches capacity, backpressure kicks in.
	QueueDepth int32

	// QueueCapacity is the maximum queue depth configured via Options.QueueDepth.
	// This is the capacity at which backpressure will block new work items.
	QueueCapacity int32

	// TotalSteps is the cumulative count of execution steps since run start.
	// Monotonically increasing counter.
	TotalSteps int64

	// TotalEnqueued is the total number of work items enqueued since run start.
	// Useful for calculating throughput (enqueued/second).
	TotalEnqueued int64

	// TotalDequeued is the total number of work items dequeued since run start.
	// Should equal TotalEnqueued when run completes.
	TotalDequeued int64

	// BackpressureEvents counts how many times backpressure was triggered.
	// High count indicates sustained overload or insufficient concurrency.
	BackpressureEvents int32

	// PeakActiveNodes is the maximum concurrent execution observed during this run.
	// Should be <= MaxConcurrentNodes.
	PeakActiveNodes int32

	// PeakQueueDepth is the maximum queue depth observed during this run.
	// If this equals QueueCapacity, backpressure was triggered.
	PeakQueueDepth int32
}

SchedulerMetrics tracks execution metrics for monitoring and observability (T067).

These metrics provide insight into the runtime behavior of the concurrent scheduler, enabling operators to:

• Detect bottlenecks and capacity issues
• Monitor backpressure conditions
• Tune concurrency parameters (MaxConcurrentNodes, QueueDepth)
• Alert on unhealthy execution patterns

Metrics are updated atomically by the scheduler and can be safely read concurrently.

According to spec.md FR-027: System MUST expose metrics for queue_depth, step_latency_ms, active_nodes, and step_count.

type SideEffectPolicy ¶

type SideEffectPolicy struct {
	// Recordable indicates whether this node's I/O can be captured for replay.
	// Examples:
	// - LLM API calls: true (responses are cacheable).
	// - Pure functions: false (no external I/O).
	// - Database queries: false (may be non-deterministic).
	Recordable bool

	// RequiresIdempotency indicates whether this node requires an idempotency key.
	// to prevent duplicate execution. This is important for side-effecting operations.
	// like database writes, payments, or notifications.
	//
	// If true, the node must provide an IdempotencyKeyFunc in its NodePolicy.
	RequiresIdempotency bool
}

SideEffectPolicy declares the external I/O characteristics of a node, informing the replay engine whether the node's interactions should be. recorded and replayed.

This policy affects deterministic replay behavior: - Recordable=true: I/O is captured and can be replayed without re-execution. - RequiresIdempotency=true: Node needs idempotency key to ensure exactly-once semantics.

type StateCopier ¶

type StateCopier interface {
	DeepCopy() (any, error)
}

StateCopier is an optional interface that state types can implement to provide custom deep copy logic. When a state type implements this interface, the engine will use the custom DeepCopy method instead of JSON-based serialization.

This interface allows users to: - Avoid JSON serialization overhead in performance-critical paths - Properly handle unexported fields, channels, or other non-JSON-serializable types - Implement application-specific copying semantics

The interface uses any return type to accommodate both value and pointer receiver implementations, avoiding generic interface instantiation issues where S = *T would require DeepCopy() (*T, error) instead of DeepCopy() (T, error).

Example implementations:

Value receiver (returns value):

type MyState struct {
    Counter int
    Data    []byte
}

func (s MyState) DeepCopy() (any, error) {
    copied := MyState{
        Counter: s.Counter,
        Data:    append([]byte(nil), s.Data...),
    }
    return copied, nil
}

Pointer receiver (returns pointer):

type MyState struct {
    Counter int
    Data    []byte
}

func (s *MyState) DeepCopy() (any, error) {
    copied := &MyState{
        Counter: s.Counter,
        Data:    append([]byte(nil), s.Data...),
    }
    return copied, nil
}

Thread-safety: DeepCopy implementations must be safe to call from multiple goroutines without external synchronization.

type WorkItem ¶

type WorkItem[S any] struct {
	// StepID is the monotonically increasing step number in the run
	StepID int `json:"step_id"`

	// OrderKey is a deterministic sort key computed from hash(path_hash, edge_index).
	// This ensures consistent execution order across replays.
	OrderKey uint64 `json:"order_key"`

	// NodeID identifies the node to execute for this work item
	NodeID string `json:"node_id"`

	// State is the snapshot of state for this work item's execution
	State S `json:"state"`

	// Attempt is the retry counter (0 for first execution, 1+ for retries)
	Attempt int `json:"attempt"`

	// ParentNodeID is the node that spawned this work item, used for path hash computation
	ParentNodeID string `json:"parent_node_id"`

	// EdgeIndex is the index of the edge taken from parent, used for deterministic ordering
	EdgeIndex int `json:"edge_index"`
}

WorkItem represents a schedulable unit of work in the execution frontier. Each WorkItem contains all the context needed to execute a node, including the node's input state, execution metadata, and provenance information for deterministic ordering.

WorkItems are created when nodes produce routing decisions and are queued in the frontier for concurrent execution. The OrderKey ensures deterministic processing order even when nodes complete out of order.