flow

type AmbiguousRoutingError ¶

type AmbiguousRoutingError struct{ VertexID VertexID }

AmbiguousRoutingError reports that a vertex declares both an unconditional out-edge and a conditional edge, so its routing is ambiguous (§8).

func (e *AmbiguousRoutingError) Error() string

type BuildError ¶

type BuildError struct {
	Op     string // the build call: "AddVertex" | "AddEdge" | "AddConditionalEdge"
	Detail string // the offending parameter, e.g. "nil task" or "zero from vertex"
}

BuildError reports a fail-fast violation of an add-time build invariant that has no more specific typed error: a nil required argument (task, selector, reducer, condition Pick) or a malformed identifier/parameter supplied to a build call (a zero VertexID, an empty Condition.Targets, a second conditional edge on a from that already has one). It is distinct from the structural Compile checks (§8); those keep their own typed errors (UnknownVertexError, AmbiguousRoutingError, …). Op names the failing build call (e.g. "AddVertex", "AddEdge", "AddConditionalEdge") and Detail names the offending parameter so an operator can identify the bad call from a log line.

func (e *BuildError) Error() string

type Checkpoint ¶

type Checkpoint struct {
	Run        GraphRunState     // run-level status + Revision + timestamps
	StepBase   json.RawMessage   `json:",omitempty"` // committed S_N — frozen read snapshot; pending vertices' selectors read THIS
	State      json.RawMessage   `json:",omitempty"` // accumulated S (S_N + reducers of every terminal vertex so far)
	Vertices   []VertexState     // per-vertex records for this step; terminal ones are skipped on resume
	Frontier   []VertexID        // the active vertex set this checkpoint resumes from (meaning depends on Phase)
	Routes     []RouteRecord     // routing decisions that produced Frontier (§9.5)
	Phase      StepPhase         // phase of this checkpoint within the super-step
	Interrupts []InterruptRecord // per-vertex pauses; present in StepRunning AND StepPaused; mutually exclusive with Halt (§9.7)
	Halt       *HaltRecord       // run-level routing/structural halt (StepHalted); mutually exclusive with Interrupts (§9.8)
}

Checkpoint is the engine's durable, append-only record of one unit of execution (§10.1): the run-level state, the frozen read snapshot S_N, the accumulated state S, per-vertex records, the active frontier, the routing decisions, and the phase. Interrupts and Halt are mutually exclusive in a VALID checkpoint (enforced on load in §10.4); the struct can physically hold both, but never does for a well-formed checkpoint.

type CheckpointDecodeError ¶

type CheckpointDecodeError struct {
	Field string // "StepBase" | "State" | "checkpoint"
	Err   error
}

CheckpointDecodeError reports that part of a loaded checkpoint failed to decode. Field names the failing part ("StepBase", "State", or "checkpoint"). It wraps the decode cause (§10.4).

func (e *CheckpointDecodeError) Error() string

func (e *CheckpointDecodeError) Unwrap() error

type CheckpointGranularity ¶

type CheckpointGranularity int

CheckpointGranularity selects WHEN the coordinator writes checkpoints within a super-step (§10.1). PerVertex (the default) appends after each vertex reduces, so a crash mid-step loses at most one vertex's work; PerStep defers all writes to the step boundary for fewer writes at coarser recovery. This sub-task implements PerVertex; the PerStep behavior is a later sub-task.

const (
	PerVertex CheckpointGranularity = iota // append after each vertex (default)
	PerStep                                // defer writes to the step boundary
)

type CheckpointNotFoundError ¶

type CheckpointNotFoundError struct{ GraphRunID GraphRunID }

CheckpointNotFoundError reports that no checkpoint exists for the given run, so there is nothing to resume from (§10.2).

func (e *CheckpointNotFoundError) Error() string

type CheckpointStore ¶

type CheckpointStore interface {
	// Append durably records cp iff cp.Run.Revision is the next revision in
	// sequence for cp.Run.GraphRunID (compare-and-append). Otherwise it returns a
	// *RevisionConflictError. A serialization failure is a *StoreError.
	Append(ctx context.Context, cp *Checkpoint) error
	// Latest returns the highest-revision checkpoint for id (the source of truth),
	// or a *CheckpointNotFoundError if the run has no checkpoints. It MUST return
	// the checkpoint with the HIGHEST Run.Revision for the run: the §10.4 resume
	// contract depends on the loaded checkpoint being genuinely the latest, and a
	// backend that returns a stale revision violates the contract (it would fork or
	// overwrite committed history on the next append).
	Latest(ctx context.Context, id GraphRunID) (*Checkpoint, error)
	// History returns every checkpoint for id ordered by revision (0,1,2,…), or a
	// *CheckpointNotFoundError if the run has no checkpoints.
	History(ctx context.Context, id GraphRunID) ([]*Checkpoint, error)
}

CheckpointStore is the engine's durable, append-only checkpoint history for graph runs (§10.2). Every implementation must honor the same contract: compare-and-append on (GraphRunID, Revision), latest-revision-as-source-of- truth, ordered History, structural immutability of stored checkpoints, and a CheckpointNotFoundError for an unknown run. Every method honors ctx.

SECURITY (durable backends): a backend that decodes UNTRUSTED stored bytes on read (§10.4) MUST bound the payload size and nesting before decoding, to guard against oversized or deeply-nested input (CLAUDE.md: guard against unbounded sizes). MemStore is exempt only because its bytes are self-produced in-memory.

CONTRACT (Latest is genuinely the latest, §10.4): Latest MUST return the checkpoint with the HIGHEST Run.Revision for the run. The resume contract (§10.4) depends on the loaded checkpoint being the true latest — Resume continues the append-only sequence from cp.Run.Revision, so a backend that returns a STALE checkpoint would fork or overwrite committed history. A durable backend's conformance test MUST cover this; MemStore guarantees it structurally (it stores the contiguous 0..len-1 revisions and returns the last).

type CompileOption ¶

type CompileOption func(*compileConfig)

CompileOption configures Compile (§8). It is non-generic so the same option value works for any state type S, since the store is fixed at Compile (§9) and not parameterized by S.

func WithStore(s CheckpointStore) CompileOption

type Condition ¶

type Condition[S any] struct {
	Targets []VertexID                                         // declared possible targets — validated at Compile (§8)
	Pick    func(ctx context.Context, s S) ([]VertexID, error) // choose ≥1 declared target (read-only)
}

Condition is a conditional out-edge from a vertex (§7): a declared set of possible Targets and a Pick that chooses one or more of them from the current state. Pick is read-only and must return at least one declared target (a multi-target return is a fan-out); the empty-set and undeclared-target rules are enforced at runtime (§9.5). Existence of Targets is validated at Compile.

type ConditionError ¶

type ConditionError struct {
	From VertexID
	Err  error
}

ConditionError reports that a condition's Pick returned an error or panicked. It wraps the cause and surfaces as a run-level halt (HaltCondition, §9.5, §9.8).

func (e *ConditionError) Error() string

func (e *ConditionError) Unwrap() error

type ControlPlane ¶

type ControlPlane interface {
	// Submit enqueues w for consumers serving w.Key. It honors ctx and must not
	// block unboundedly.
	Submit(ctx context.Context, w Work) error
	// Consume returns a channel delivering only Work whose Key is in serves. The
	// channel is closed when ctx is done (clean shutdown, no goroutine leak).
	Consume(ctx context.Context, serves []GraphVersionKey) (<-chan Delivery, error)
}

ControlPlane accepts work and distributes it to workers (§18.5). It is SEPARATE from CheckpointStore (transient consume-once dispatch vs durable append-only history). Implementations must honor ctx on every call (no unbounded blocking). Registration is implicit — a worker registers by Consuming the version keys it serves; there is no separate registration RPC.

SINGLE-FLIGHT (best-effort, NOT the correctness boundary). An implementation SHOULD avoid delivering two Works for the same GraphRunID concurrently, to spare duplicate work — but it need not guarantee it. The in-process plane (controlplane.Mem) DOES guarantee it via a central dispatcher; a distributed plane (nats.ControlPlane) provides single-flight at MESSAGE granularity (a work-queue delivers each message once until it is acked), so two DISTINCT messages for the same run can be processed concurrently. That is safe by design: CORRECTNESS — no duplicate COMMITTED effects — is guaranteed by the store's compare-and-append (RevisionConflictError, §10.2) and IdempotencyKey (§4.1), which absorb concurrent duplicate work; the control plane's single-flight is only an efficiency optimization. Providing STRICTER single-flight than required still satisfies this contract (LSP), so controlplane.Mem remains conformant.

type DeadEndError ¶

type DeadEndError struct{ Step StepID }

DeadEndError reports that the frontier drained without the finish vertex ever executing, so the run can make no further progress; a run-level halt (§9.5, §9.8).

func (e *DeadEndError) Error() string

type Delivery ¶

type Delivery struct {
	Work Work
	Ack  func() error
	Nack func() error
}

Delivery wraps Work with explicit ack semantics so durable backends survive worker crashes (§18.5). A worker calls Ack when the work reaches a QUIESCENT result (completed / interrupted / halted / cancelled, or safely requeued) to drop it from the queue; it calls Nack to requeue the work for redelivery (transient failure / shedding load). Exactly one of Ack/Nack should be called per Delivery; the control plane holds the run's single-flight slot until one of them fires (§18.5).

type DuplicateConditionalEdgeError ¶

type DuplicateConditionalEdgeError struct{ From VertexID }

DuplicateConditionalEdgeError reports that a second conditional out-edge was added for a vertex that already has one; a vertex may have at most one conditional edge, since a second would silently overwrite the first (§7).

func (e *DuplicateConditionalEdgeError) Error() string

type DuplicateVertexError ¶

type DuplicateVertexError struct{ VertexID VertexID }

DuplicateVertexError reports that the same VertexID was added to a graph more than once; VertexIDs must be unique within a graph (§8).

func (e *DuplicateVertexError) Error() string

type FuncTask ¶

type FuncTask[I, O any] struct {
	// contains filtered or unexported fields
}

FuncTask adapts a TaskFunc into a Task. It holds only the function — no VertexID, no state — so a single value is safe to reuse across graphs and concurrent runs.

func NewFuncTask[I, O any](fn TaskFunc[I, O]) *FuncTask[I, O]

func (t *FuncTask[I, O]) Execute(ctx context.Context, in I) (O, error)

type Graph ¶

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

Graph is the mutable build-time definition of a workflow over shared state S (§7). It is UNEXPORTED in its fields: callers build it through NewGraph, AddVertex, AddEdge, and AddConditionalEdge, then Compile it (later task) into an immutable Runner. It is not safe for concurrent mutation; build it on one goroutine before compiling.

func NewGraph[S any](id GraphID, opts ...GraphOption) *Graph[S]

func (g *Graph[S]) AddConditionalEdge(from VertexID, c Condition[S]) error

func (g *Graph[S]) AddEdge(from, to VertexID) error

func (g *Graph[S]) Compile(entry, finish VertexID, opts ...CompileOption) (*Runner[S], error)

type GraphID ¶

type GraphID uuid.UUID // stable definition id — pinned as a const by callers.

The identifier types of the engine (design §3). The four UUID-backed types are distinct named types — not aliases — so the compiler rejects passing a GraphID where a VertexID is wanted. Each delegates String/MarshalText/ UnmarshalText to the underlying uuid.UUID by conversion, so they serialize as readable canonical strings (not 16-int arrays) in checkpoints.

GraphID and VertexID are stable DEFINITION ids: a checkpoint frontier references vertices by VertexID and a resume rebuilds the graph from code, so they must be stable across restarts and are pinned as consts by callers via uuid.MustParse. They therefore have no generating constructor here — minting a fresh one per build would break resume. GraphRunID and VertexRunID are runtime instances minted fresh each run/execution via uuid.New (see the NewGraphRunID/NewVertexRunID constructors below).

func (id GraphID) MarshalText() ([]byte, error)

func (id GraphID) String() string

func (id *GraphID) UnmarshalText(b []byte) error

type GraphMismatchError ¶

type GraphMismatchError struct {
	Expected GraphID
	Actual   GraphID
}

GraphMismatchError reports that a loaded checkpoint's GraphID does not match the runner's compiled graph, so it belongs to a different graph (§10.4).

func (e *GraphMismatchError) Error() string

type GraphOption ¶

type GraphOption func(*graphConfig)

GraphOption configures a Graph at NewGraph (§7). It is non-generic so the same option value works for any state type S.

func WithVersion(n uint64) GraphOption

type GraphRunExistsError ¶

type GraphRunExistsError struct{ GraphRunID GraphRunID }

GraphRunExistsError reports an attempt to start a run whose GraphRunID already exists in the store (a duplicate start) (§18.2).

func (e *GraphRunExistsError) Error() string

type GraphRunID ¶

type GraphRunID uuid.UUID // runtime instance — minted per run via NewGraphRunID.

The identifier types of the engine (design §3). The four UUID-backed types are distinct named types — not aliases — so the compiler rejects passing a GraphID where a VertexID is wanted. Each delegates String/MarshalText/ UnmarshalText to the underlying uuid.UUID by conversion, so they serialize as readable canonical strings (not 16-int arrays) in checkpoints.

GraphID and VertexID are stable DEFINITION ids: a checkpoint frontier references vertices by VertexID and a resume rebuilds the graph from code, so they must be stable across restarts and are pinned as consts by callers via uuid.MustParse. They therefore have no generating constructor here — minting a fresh one per build would break resume. GraphRunID and VertexRunID are runtime instances minted fresh each run/execution via uuid.New (see the NewGraphRunID/NewVertexRunID constructors below).

func NewGraphRunID() (GraphRunID, error)

func (id GraphRunID) MarshalText() ([]byte, error)

func (id GraphRunID) String() string

func (id *GraphRunID) UnmarshalText(b []byte) error

type GraphRunMismatchError ¶

type GraphRunMismatchError struct {
	Requested GraphRunID
	Actual    GraphRunID
}

GraphRunMismatchError reports that a loaded checkpoint's embedded Run.GraphRunID does not match the GraphRunID requested on Resume, so the store returned a checkpoint belonging to a DIFFERENT run. Resuming it would write into the embedded run's history; this fails secure before any task runs (§10.4). Requested is the id passed to Resume; Actual is the checkpoint's embedded Run.GraphRunID.

func (e *GraphRunMismatchError) Error() string

type GraphRunState ¶

type GraphRunState struct {
	GraphRunID    GraphRunID
	GraphID       GraphID
	GraphVersion  string // compiled-graph fingerprint (§8.1); a mismatch on resume → GraphVersionMismatchError
	Status        RunStatus
	Step          StepID
	Revision      uint64    // monotonic checkpoint sequence for this run (§10)
	CreatedAt     time.Time // Run() called
	StartedAt     time.Time // first super-step began
	UpdatedAt     time.Time // last checkpoint write
	CompletedAt   time.Time // set when Status == RunCompleted
	InterruptedAt time.Time // set when Status == RunInterrupted (most recent pause)
	CancelledAt   time.Time // set when Status == RunCancelled
	CancelReason  string    // Cancel(id, reason)'s reason, when cancelled
}

GraphRunState is the run-level instrumentation record: identity, status, step, revision, and lifecycle timestamps for one graph run (§4.1).

type GraphVersionKey ¶

type GraphVersionKey struct {
	GraphID      GraphID
	GraphVersion string
}

GraphVersionKey routes work to a worker serving exactly this (GraphID, GraphVersion) — the version IS the route (§18.5). Both fields are comparable value types (GraphID is a [16]byte array, GraphVersion a string), so the struct is usable directly as a map key with no allocation.

type GraphVersionMismatchError ¶

type GraphVersionMismatchError struct {
	Expected string
	Actual   string
}

GraphVersionMismatchError reports that a loaded checkpoint's GraphVersion fingerprint does not match the current compiled graph's, so a changed graph cannot resume an old checkpoint (§8.1, §10.4).

func (e *GraphVersionMismatchError) Error() string

type Halt ¶

type Halt struct {
	GraphRunID GraphRunID
	Kind       HaltKind // defined in checkpoint.go
	Step       StepID
	Cause      error
}

Halt is a RUN-LEVEL routing/structural failure surfaced in a run Result — not a vertex pause (§10.3, §9.8). Kind classifies the structural cause; Step names the super-step at which the run halted; Cause wraps the underlying error.

type HaltKind ¶

type HaltKind int

HaltKind classifies a run-level routing/structural halt (§10.1). Persisted in checkpoints, so the iota ordering is pinned.

const (
	HaltCondition        HaltKind = iota // Condition.Pick returned an error
	HaltUndeclaredTarget                 // Pick returned empty / a target outside Targets
	HaltDeadEnd                          // frontier drained without finish executing
	HaltMaxSteps                         // step budget (WithMaxSteps) exceeded
)

type HaltRecord ¶

type HaltRecord struct {
	Kind  HaltKind
	Step  StepID
	Cause string // error message/type name
}

HaltRecord is the durable record of a run-level halt (§10.1, §9.8). It is set only in StepHalted checkpoints and is mutually exclusive with Interrupts.

type Hooks ¶

type Hooks struct {
	OnRunStart     func(ctx context.Context, ev GraphRunState)
	OnRunFinish    func(ctx context.Context, ev GraphRunState)
	OnVertexStart  func(ctx context.Context, ev VertexState)
	OnVertexFinish func(ctx context.Context, ev VertexState)
	OnEdge         func(ctx context.Context, from, to VertexID, run GraphRunState)
	OnStep         func(ctx context.Context, run GraphRunState, activated int)
	OnCheckpoint   func(ctx context.Context, id GraphRunID, rev uint64, step StepID)
	OnInterrupt    func(ctx context.Context, iv Interruption)
	OnHalt         func(ctx context.Context, h Halt)
}

Hooks is a set of optional, purely observational lifecycle callbacks (§11). Every field is nil-able and independent, so a caller wires only the events it cares about (interface segregation). A callback receives the framework-owned instrumentation records of §4.1 and must not mutate engine state; it is fired best-effort and any panic it raises is recovered and discarded (§12.5).

type IdempotencyKey ¶

type IdempotencyKey string

IdempotencyKey identifies one LOGICAL vertex execution, stable across in-process retries AND crash recovery. Side-effecting tasks pass it to external systems that support idempotency, so a re-run after a pre-checkpoint crash does not duplicate effects. It deliberately EXCLUDES VertexRunID and Attempt, which vary per concrete attempt.

type InterruptKind ¶

type InterruptKind int

InterruptKind classifies a vertex pause (§10.1). It is defined HERE because InterruptRecord needs it; the public Interrupt struct in §10.3 (Phase 5) reuses this SAME type, so its iota ordering is a shared persisted contract.

const (
	Awaiting InterruptKind = iota // user-initiated pause (flow.Interrupt) — Info carries the reason
	Errored                       // failure pause (Pause-policy error) — Cause carries the message
)

type InterruptRecord ¶

type InterruptRecord struct {
	Vertex       VertexID
	Kind         InterruptKind   // Awaiting | Errored
	Info         json.RawMessage `json:",omitempty"` // user-facing reason (Awaiting) — a serialization boundary
	Cause        string          // error message/type name (Errored)
	Continuation json.RawMessage `json:",omitempty"` // optional task continuation (StatefulInterrupt) — a serialization boundary
}

InterruptRecord is the durable record of one paused vertex (§10.1). Info and Continuation are serialization-boundary json.RawMessage fields carrying pre-encoded JSON (the user reason and an optional StatefulInterrupt continuation) that must round-trip untouched.

type Interruption ¶

type Interruption struct {
	GraphRunID GraphRunID
	Vertex     VertexID
	Kind       InterruptKind // Awaiting | Errored (defined in checkpoint.go)
	Info       any           // user reason (Awaiting) — serialization boundary (§10.3)
	Cause      error         // underlying failure (Errored)
}

Interruption is the per-vertex pause surfaced in a run Result and to Hooks.OnInterrupt (§10.3). Info carries the user reason for an Awaiting pause and is an `any` because it is a serialization-boundary value (the same reason passed to Interrupt); Cause carries the underlying failure for an Errored pause. Exactly one is meaningful per Kind.

type MaxStepsExceededError ¶

type MaxStepsExceededError struct {
	Max  int
	Step StepID
}

MaxStepsExceededError reports that the super-step budget (WithMaxSteps) was exhausted before the run completed; a run-level halt, not a vertex pause (§9.5, §9.8).

func (e *MaxStepsExceededError) Error() string

type MemStore ¶

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

MemStore is an in-memory CheckpointStore for development and tests (§10.2). It holds the ENCODED bytes of each checkpoint per run (so stored history is independent of any caller-held *Checkpoint) under a sync.RWMutex. It is the reference implementation of the CheckpointStore contract; a durable backend (SQLite/Postgres/NATS) honors the same behavior.

func NewMemStore() *MemStore

func (s *MemStore) Append(ctx context.Context, cp *Checkpoint) error

func (s *MemStore) History(ctx context.Context, id GraphRunID) ([]*Checkpoint, error)

func (s *MemStore) Latest(ctx context.Context, id GraphRunID) (*Checkpoint, error)

type MissingEntryError ¶

type MissingEntryError struct {
	VertexID VertexID
	Role     string // "entry" or "finish"
}

MissingEntryError reports that a named role vertex is absent from the graph. Role is "entry" or "finish" (§8).

func (e *MissingEntryError) Error() string

type Reducer ¶

type Reducer[S, O any] func(s *S, out O) error

Reducer folds a vertex's output O into the graph state S, returning an error to reject the fold. The coordinator applies it to a clone and commits only on a nil error, so a reducer that mutates then errors leaves S unchanged (§6.2).

type Resolver ¶

type Resolver interface {
	// Resolve returns the handle registered under the exact (id, version) and true,
	// or (nil, false) if none is registered.
	Resolve(id GraphID, version string) (RunnerHandle, bool)
	// Keys returns one GraphVersionKey per registration — the exact set of versions
	// this worker serves, which Serve hands to Consume.
	Keys() []GraphVersionKey
}

Resolver resolves a (GraphID, GraphVersion) to its RunnerHandle and lists the keys this worker serves, so Serve consumes exactly those (§18.5/§18.6 — registration is implicit via Consume). It is the SUBSET of the registry surface Serve needs (interface segregation: Serve never touches Add/Manifest). registry.Registry satisfies it structurally, so no adapter is required.

type Result ¶

type Result[S any] struct {
	Run        GraphRunState  // ids, status, step, revision, timestamps (§4.1)
	State      S              // final accumulated state
	Interrupts []Interruption // per-vertex pauses (§9.7) — nil on the happy path
	Halt       *Halt          // run-level halt (§9.8) — nil on the happy path
}

Result is the outcome of a Run/Resume (§9). Run is the framework-owned run state (ids, status, step, revision, timestamps); State is the final accumulated graph state S. Interrupts (per-vertex pauses, §9.7) and Halt (a run-level routing/structural halt, §9.8) are mutually exclusive and set only when the run is interrupted; on the happy path both are nil. Result.Run.Status is never RunRunning on return.

type ResumeTerminalError ¶

type ResumeTerminalError struct{ Status RunStatus }

ResumeTerminalError reports an attempt to resume a run that is already in a terminal state (RunCompleted or RunCancelled), which cannot continue (§9.3, §12.4).

func (e *ResumeTerminalError) Error() string

type RetryPolicy ¶

type RetryPolicy struct {
	// MaxAttempts bounds the total number of task executions (the first try plus
	// retries). Values <= 1 mean no retry.
	MaxAttempts int

	// Backoff returns the delay to wait before the given 1-based attempt. A nil
	// Backoff means no delay between attempts. The execution wrapper (later phase)
	// invokes it under panic recovery (§12.5).
	Backoff func(attempt int) time.Duration

	// Retryable reports whether err warrants a retry. A nil Retryable means any
	// non-interrupt error is retryable (§12.2). The execution wrapper (later
	// phase) invokes it under panic recovery (§12.5).
	Retryable func(err error) bool
}

RetryPolicy is the per-vertex retry configuration (§12.2). A zero RetryPolicy is meaningful only as the absence of retry (vertexConfig.retry stays nil when WithRetry is not applied); when set, MaxAttempts bounds the re-runs, Backoff computes the delay before attempt n, and Retryable decides whether a given error is worth retrying.

type RevisionConflictError ¶

type RevisionConflictError struct {
	GraphRunID GraphRunID
	Expected   uint64 // the revision the store required next (latest+1)
	Actual     uint64 // the revision actually supplied by the caller
}

RevisionConflictError reports a failed compare-and-append: the supplied revision was not the store's required next revision, because another writer advanced the run (concurrent-resume detection, §10.2). Expected is the revision the store required next (latest+1); Actual is the revision supplied.

func (e *RevisionConflictError) Error() string

type RouteRecord ¶

type RouteRecord struct {
	From        VertexID
	To          []VertexID // chosen target(s); for a Condition, exactly what Pick returned
	Conditional bool       // true if chosen by Condition.Pick; false for a static AddEdge
}

RouteRecord durably records ONE routing decision — which source vertex activated which target(s) this step, and whether via a Condition (§10.1).

type RunInfo ¶

type RunInfo struct {
	GraphID     GraphID
	GraphRunID  GraphRunID
	VertexID    VertexID
	VertexRunID VertexRunID
	Step        StepID
}

RunInfo is the identity an executing vertex reads from its context (§4.2). It names the graph, run, vertex, vertex execution, and super-step, and computes the vertex's IdempotencyKey.

func Info(ctx context.Context) (RunInfo, bool)

func (i RunInfo) IdempotencyKey() IdempotencyKey

type RunOption ¶

type RunOption func(*runConfig)

RunOption configures a single Run/Resume (§9). Options are applied in order over a defaulted runConfig.

func WithCheckpointEvery(g CheckpointGranularity) RunOption

func WithConcurrency(n int) RunOption

func WithGraphRunID(id GraphRunID) RunOption

func WithHooks(h Hooks) RunOption

func WithMaxSteps(n int) RunOption

type RunResult ¶

type RunResult struct {
	Run        GraphRunState   // ids, status, step, revision, timestamps (§4.1)
	State      json.RawMessage // the marshaled final/last-checkpointed S
	Interrupts []Interruption  // per-vertex pauses (§9.7) — nil on the happy path
	Halt       *Halt           // run-level halt (§9.8) — nil on the happy path
}

RunResult is the graph-agnostic, JSON-friendly form of Result[S] (§18.1): State is the marshaled final/last-checkpointed S; Run, Interrupts and Halt are already non-generic and copied through unchanged. As in Result[S], Interrupts and Halt are mutually exclusive and nil on the happy path.

type RunStatus ¶

type RunStatus int

RunStatus is the lifecycle status of a graph run. Its values are persisted in checkpoints, so the iota ordering is pinned (§4.1).

const (
	RunRunning     RunStatus = iota // in flight
	RunCompleted                    // finish executed and the frontier drained
	RunInterrupted                  // Awaiting or Errored — there is no RunFailed (§12)
	RunCancelled                    // terminal: Cancel(id) appended a final checkpoint (§18.2); cannot resume
)

func (s RunStatus) String() string

type Runner ¶

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

Runner is the immutable, validated form of a Graph[S], produced by Compile (§8, §9). It is safe to reuse across concurrent runs. It holds the validated graph, the entry/finish roles, the GraphVersion fingerprint (§8.1), and the single CheckpointStore set at Compile (§9). Per-run behavior (hooks, concurrency, maxSteps, granularity) is supplied per call via RunOptions, not stored on the Runner, so the Runner itself stays immutable and reusable.

The store is the INTERFACE (CheckpointStore), not a concrete type (dependency inversion): the default MemStore is wired at Compile, the composition point, so the Runner never depends on a specific backend. Per §9 it is fixed at Compile — ALL operations (Run, Resume, Status, Get) use this one store; there is no per-run override.

func (r *Runner[S]) Cancel(ctx context.Context, id GraphRunID, reason string, opts ...RunOption) error

func (r *Runner[S]) Get(ctx context.Context, id GraphRunID) (*Result[S], error)

func (r *Runner[S]) GraphID() GraphID

func (r *Runner[S]) GraphVersion() string

func (r *Runner[S]) Resume(ctx context.Context, id GraphRunID, payload any, opts ...RunOption) (*Result[S], error)

func (r *Runner[S]) Run(ctx context.Context, in S, opts ...RunOption) (*Result[S], error)

package main

import (
	"context"
	"fmt"

	"github.com/looprig/core/uuid"
	"github.com/looprig/flow/pkg/flow"
)

var (
	exGraphID = flow.GraphID(uuid.MustParse("a1b2c3d4-0000-4000-8000-000000000000"))

	intakeID = flow.VertexID(uuid.MustParse("11111111-1111-4111-8111-111111111111"))

	sendToSalesID = flow.VertexID(uuid.MustParse("44444444-4444-4444-8444-444444444444"))
)

// Request is the §16 graph state S — the caller's domain blackboard threaded
// through the flow. The only mutation path is a vertex's reducer (the engine is
// the single writer); tasks are pure and never touch it. It must round-trip
// through JSON so it can be checkpointed and resumed, so every field is exported.
type Request struct {
	Question     string
	Draft        string
	Approved     bool
	NeedsExpert  bool
	Refined      int
	MaxRefines   int
	TicketID     string
	ExpertAnswer string
	Sent         bool
	LastErr      string
}

// query is RAGDraft's input; answer is its output.
type query struct{ Text string }
type answer struct{ Text string }

// ragDraft is the "RAG" drafting task: a deterministic stand-in that returns a
// canned answer for the question. The real engine would run retrieval + an LLM
// here; v1 is a plain function so the example is reproducible.
func ragDraft() *flow.FuncTask[query, answer] {
	return flow.NewFuncTask(func(_ context.Context, in query) (answer, error) {
		return answer{Text: "draft answer for: " + in.Text}, nil
	})
}

func main() {
	// buildGraph takes *testing.T for its fail-fast helpers, which a godoc example
	// (no T) cannot supply. So this example is a thin, self-contained mirror of the
	// happy path, built directly with the same exported API and checking errors
	// inline instead.
	g := flow.NewGraph[Request](exGraphID)
	_ = flow.AddVertex(g, intakeID, ragDraft(),
		func(s Request) query { return query{Text: s.Question} },
		func(s *Request, a answer) error { s.Draft = a.Text; s.Approved = true; return nil },
	)
	_ = flow.AddVertex(g, sendToSalesID,
		flow.NewFuncTask(func(_ context.Context, draft string) (bool, error) { return draft != "", nil }),
		func(s Request) string { return s.Draft },
		func(s *Request, ok bool) error { s.Sent = ok; return nil },
	)
	_ = g.AddEdge(intakeID, sendToSalesID)

	runner, err := g.Compile(intakeID, sendToSalesID, flow.WithStore(flow.NewMemStore()))
	if err != nil {
		panic(err)
	}
	res, err := runner.Run(context.Background(), Request{Question: "what is the SLA?"})
	if err != nil {
		panic(err)
	}

	fmt.Println(res.Run.Status)
	fmt.Println(res.State.Draft)
	fmt.Println(res.State.Sent)
}

Output:
Completed
draft answer for: what is the SLA?
true

func (r *Runner[S]) Status(ctx context.Context, id GraphRunID) (GraphRunState, error)

type RunnerHandle ¶

type RunnerHandle interface {
	// GraphID returns the wrapped Runner's stable definition identity (§8.1).
	GraphID() GraphID
	// GraphVersion returns the wrapped Runner's compatibility fingerprint (§8.1).
	GraphVersion() string
	// Run decodes stateJSON into the graph state S and starts a run. A malformed
	// stateJSON is rejected at the decode boundary; an empty/nil stateJSON decodes
	// to the zero S.
	Run(ctx context.Context, stateJSON json.RawMessage, opts ...RunOption) (*RunResult, error)
	// Resume continues run id, passing payloadJSON to the run as the live Resume
	// payload (see the runnerHandle.Resume doc for the payload-typing nuance).
	Resume(ctx context.Context, id GraphRunID, payloadJSON json.RawMessage, opts ...RunOption) (*RunResult, error)
	// Status returns the latest GraphRunState for id without decoding S (§18.2).
	Status(ctx context.Context, id GraphRunID) (GraphRunState, error)
	// Get returns the latest run record with the marshaled current State (§18.2).
	Get(ctx context.Context, id GraphRunID) (*RunResult, error)
	// Cancel appends a terminal RunCancelled checkpoint for id (§18.2).
	Cancel(ctx context.Context, id GraphRunID, reason string, opts ...RunOption) error
}

RunnerHandle is the non-generic, JSON-in/out facade over a typed Runner[S] (§18.1). It lets a graph-agnostic caller (the registry/ingress) start, resume, query, and cancel any compiled graph by JSON alone. It mirrors the in-process control surface (Run/Resume/Status/Get/Cancel + GraphID/GraphVersion, §18.2) with S erased to json.RawMessage at the boundaries. Construct one with NewRunnerHandle.

func NewRunnerHandle[S any](r *Runner[S]) RunnerHandle

type Selector ¶

type Selector[S, I any] func(s S) I

Selector derives a vertex's input I from the graph state S. It is read-only: it must not mutate S (the coordinator passes the immutable step-base snapshot, §6.2).

type StepID ¶

type StepID int // super-step index within a run: 0, 1, 2, …

The identifier types of the engine (design §3). The four UUID-backed types are distinct named types — not aliases — so the compiler rejects passing a GraphID where a VertexID is wanted. Each delegates String/MarshalText/ UnmarshalText to the underlying uuid.UUID by conversion, so they serialize as readable canonical strings (not 16-int arrays) in checkpoints.

GraphID and VertexID are stable DEFINITION ids: a checkpoint frontier references vertices by VertexID and a resume rebuilds the graph from code, so they must be stable across restarts and are pinned as consts by callers via uuid.MustParse. They therefore have no generating constructor here — minting a fresh one per build would break resume. GraphRunID and VertexRunID are runtime instances minted fresh each run/execution via uuid.New (see the NewGraphRunID/NewVertexRunID constructors below).

func (s StepID) String() string

type StepPhase ¶

type StepPhase int

StepPhase is the phase of a checkpoint within a super-step (§10.1). Its values are persisted in checkpoints, so the iota ordering is pinned.

const (
	StepRunning StepPhase = iota // step partly reduced — some vertices terminal, some not
	StepPaused                   // step boundary: ≥1 vertex paused (Interrupts set)
	StepRouted                   // step boundary: routing decisions produced the next Frontier
	StepHalted                   // step boundary: run-level routing/structural halt (Halt set)
)

type StoreError ¶

type StoreError struct {
	Op  string // "Append" | "Latest" | "History"
	Err error
}

StoreError reports a failure of a CheckpointStore operation. Op names the operation ("Append", "Latest", or "History"); it wraps the store cause (§10.2, §12.3).

func (e *StoreError) Error() string

func (e *StoreError) Unwrap() error

type Task ¶

type Task[I, O any] interface {
	Execute(ctx context.Context, in I) (O, error)
}

Task is the reusable, graph-agnostic unit of work (§5). Implementations honor the contract that they are pure with respect to graph state S (they touch no shared state — input is supplied, output is returned) and treat in as read-only (§4.2). The interface is intentionally minimal so new kinds plug in without engine changes.

type TaskFunc ¶

type TaskFunc[I, O any] func(ctx context.Context, in I) (O, error)

TaskFunc is a plain typed In → Out computation. It is the function form a FuncTask wraps; the engine never calls it directly, only via Task.Execute.

type UndeclaredTargetError ¶

type UndeclaredTargetError struct {
	From   VertexID
	Target VertexID // zero value denotes an empty return (no target picked)
}

UndeclaredTargetError reports that a condition's Pick returned a target that is not in its declared Targets set; a zero Target denotes an empty return (Pick returned no target at all), which is equally illegal (§9.5, §9.8).

func (e *UndeclaredTargetError) Error() string

type UnknownVertexError ¶

type UnknownVertexError struct{ VertexID VertexID }

UnknownVertexError reports that an edge, condition, error-route, or checkpoint frontier endpoint references a VertexID that is not in the compiled graph. Used both at build (§8) and at checkpoint validation on load (§10.4).

func (e *UnknownVertexError) Error() string

type UnknownWorkOpError ¶

type UnknownWorkOpError struct{ Op WorkOp }

UnknownWorkOpError reports a Work carrying a WorkOp outside the closed {OpRun, OpResume} domain (§18.5) — only constructible by a corrupt/forged Work crossing the transport. Per CLAUDE.md it is a concrete typed error so a caller can errors.As it; settle treats it as a PERMANENT failure (Ack-and-abandon) — a corrupt op can never dispatch, so Nacking it would be an infinite redelivery loop (the H1 poison-message DoS). The op is named for an operator log line.

func (e *UnknownWorkOpError) Error() string

type UnreachableVertexError ¶

type UnreachableVertexError struct{ VertexID VertexID }

UnreachableVertexError reports that a vertex cannot be reached from the entry vertex (or that finish is unreachable); every vertex must be reachable (§8).

func (e *UnreachableVertexError) Error() string

type VertexError ¶

type VertexError struct {
	VertexID    VertexID
	VertexRunID VertexRunID
	Attempt     int
	Err         error
}

VertexError reports that a vertex's task failed, panicked, or timed out. It carries the logical vertex, the concrete execution, the attempt number, and the wrapped cause, and drives the vertex's error policy (§12.2, §12.4).

func (e *VertexError) Error() string

func (e *VertexError) Unwrap() error

type VertexID ¶

type VertexID uuid.UUID // stable definition id — pinned as a const by callers.

The identifier types of the engine (design §3). The four UUID-backed types are distinct named types — not aliases — so the compiler rejects passing a GraphID where a VertexID is wanted. Each delegates String/MarshalText/ UnmarshalText to the underlying uuid.UUID by conversion, so they serialize as readable canonical strings (not 16-int arrays) in checkpoints.

GraphID and VertexID are stable DEFINITION ids: a checkpoint frontier references vertices by VertexID and a resume rebuilds the graph from code, so they must be stable across restarts and are pinned as consts by callers via uuid.MustParse. They therefore have no generating constructor here — minting a fresh one per build would break resume. GraphRunID and VertexRunID are runtime instances minted fresh each run/execution via uuid.New (see the NewGraphRunID/NewVertexRunID constructors below).

func (id VertexID) MarshalText() ([]byte, error)

func (id VertexID) String() string

func (id *VertexID) UnmarshalText(b []byte) error

type VertexOption ¶

type VertexOption[S any] func(*vertexConfig[S])

VertexOption configures a vertex binding's policy at AddVertex (§6.3). Options are applied in order, so a later option overrides an earlier one (last-wins).

func WithErrorPause[S any]() VertexOption[S]

func WithErrorRoute[S any](handler VertexID, record Reducer[S, error]) VertexOption[S]

func WithRetry[S any](p RetryPolicy) VertexOption[S]

func WithTimeout[S any](d time.Duration) VertexOption[S]

type VertexRunID ¶

type VertexRunID uuid.UUID // runtime instance — minted per vertex execution.

The identifier types of the engine (design §3). The four UUID-backed types are distinct named types — not aliases — so the compiler rejects passing a GraphID where a VertexID is wanted. Each delegates String/MarshalText/ UnmarshalText to the underlying uuid.UUID by conversion, so they serialize as readable canonical strings (not 16-int arrays) in checkpoints.

GraphID and VertexID are stable DEFINITION ids: a checkpoint frontier references vertices by VertexID and a resume rebuilds the graph from code, so they must be stable across restarts and are pinned as consts by callers via uuid.MustParse. They therefore have no generating constructor here — minting a fresh one per build would break resume. GraphRunID and VertexRunID are runtime instances minted fresh each run/execution via uuid.New (see the NewGraphRunID/NewVertexRunID constructors below).

func NewVertexRunID() (VertexRunID, error)

func (id VertexRunID) MarshalText() ([]byte, error)

func (id VertexRunID) String() string

func (id *VertexRunID) UnmarshalText(b []byte) error

type VertexState ¶

type VertexState struct {
	VertexID      VertexID
	VertexRunID   VertexRunID
	Step          StepID
	Status        VertexStatus
	Attempt       int
	CreatedAt     time.Time
	StartedAt     time.Time
	CompletedAt   time.Time // Status == VertexDone
	InterruptedAt time.Time // Status == VertexInterrupted
	FailedAt      time.Time // Status == VertexFailed
	Err           string
}

VertexState is the per-vertex-execution instrumentation record: identity, status, attempt, and lifecycle timestamps (§4.1).

func Self(ctx context.Context) (VertexState, bool)

type VertexStatus ¶

type VertexStatus int

VertexStatus is the lifecycle status of a single vertex execution. Its values are persisted in checkpoints, so the iota ordering is pinned (§4.1).

const (
	VertexPending VertexStatus = iota
	VertexRunning
	VertexDone
	VertexInterrupted
	VertexFailed
)

func (s VertexStatus) String() string

type Work ¶

type Work struct {
	Key        GraphVersionKey
	GraphRunID GraphRunID
	Op         WorkOp
	Input      json.RawMessage
}

Work is one unit of run/resume work submitted to the control plane (§18.5). The GraphRunID is pre-minted by the submitter (ingress, §18.3) so the submit is async-first: the GraphRunID can be returned to the caller immediately while a worker picks the work up later. Input is the initial state JSON for OpRun, or the resume payload JSON for OpResume — a json.RawMessage so the seam stays a pure transport that neither parses nor trusts the payload (the worker's Runner validates it on the way into typed business logic).

type WorkOp ¶

type WorkOp uint8

WorkOp distinguishes the two kinds of work the control plane carries: starting a fresh run (OpRun) versus resuming an interrupted one (OpResume). It is a uint8 so it is a cheap comparable value with a small, closed domain.

const (
	// OpRun starts a fresh run: Work.Input is the initial state JSON.
	OpRun WorkOp = iota
	// OpResume resumes an interrupted run: Work.Input is the resume payload JSON.
	OpResume
)

func (o WorkOp) String() string