Documentation
¶
Overview ¶
Package flow implements the declarative workflow block: a step graph with dependencies and panels.
Map: step.go = Step, Route, Admission; panel.go = Panel; definition.go = Definition, New, Roots; validate.go = step, panel, payload, and retry validation with Kahn's cycle check; routing.go = admission and panel verdicts plus route application; runner.go = Run, Confirm, and the wave advance; outcome.go = Outcome and Report; failure.go = Failure and its context accessors; retry.go = RetryPolicy and fireWithRetry; loop.go = LoopPolicy and LoopState; wave.go = one concurrent wave of ready steps; resume.go = Resume over a checkpoint; checkpoint.go = Checkpoint and its outcome views; events.go = StepCompletedEvent; wire.go = Decode for a stored Checkpoint; discovery_card.go = Card, Parse, Validate, and Match, a parsed capability card that answers whether an agent can do a task; heartbeat.go = MissedEvent; heartbeat_monitor.go = Monitor, NewMonitor, Beat, Alive, Dead, Forget, and the sentinel errors ErrInvalidOptions, ErrStaleBeat. Monitor tracks the last beat per id and reports which ids have gone silent past a fixed timeout; it holds no clock of its own and never emits MissedEvent itself. The graph is data, not code. A step with Sub runs a nested workflow to completion. Rationale: ../docs/history/flow.md. Contribution rules: ../AGENTS.md.
Index ¶
Constants ¶
const MissedEvent events.Name = "heartbeat.missed"
MissedEvent is the event kind a caller emits after it observes a dead id via Dead. Heartbeat owns the name; it never emits the event.
const StepCompletedEvent events.Name = "flow.step_completed"
StepCompletedEvent is the event kind the runner emits after a step resolves. See machine/events.go for the machine counterpart.
Variables ¶
var ErrInvalidOptions = errors.New("flow: invalid options")
ErrInvalidOptions is the sentinel for a Validate or constructor failure on caller-supplied input. It also covers a call-time argument that fails its rule, such as Monitor.Beat's blank id. Test with errors.Is; the wrapped message names the field and the violated rule.
var ( // ErrStaleBeat is the sentinel for a Beat whose at is before the // id's previously recorded time. A caller that gets ErrStaleBeat // hit a benign race and may retry past it or ignore it. ErrStaleBeat = errors.New("flow: beat is older than last recorded time") )
Sentinel errors for Monitor operations; test with errors.Is.
Functions ¶
This section is empty.
Types ¶
type Admission ¶
type Admission int
Admission is the rule that admits a step after every one of its needs resolves.
const ( // AdmissionOnSucceeded admits a step only when every need ended // OutcomeSucceeded. It is the zero value. A skipped need skips // this step, so route exclusion propagates to an excluded // branch's own dependents by default. AdmissionOnSucceeded Admission = iota // AdmissionOnFinished admits a step when every need ended // OutcomeSucceeded or OutcomeSkipped. It is the explicit opt-in // for skip tolerance, for join steps over optional branches: a // step below an excluded branch runs only by declaring this rule. AdmissionOnFinished // AdmissionOnFailed admits a step once every one of its needs is // terminal and at least one resolved OutcomeFailed. It is an // any-of rule over Needs, unlike the all-of rule the other two // values use. A step with this rule is a fallback: New rejects it // at the root (it would always admit) and inside a panel (a wave // shares one ctx across every member, with no per-member home for // the failure it would catch). AdmissionOnFailed )
type Card ¶ added in v0.5.0
type Card struct {
Name string `json:"name"`
Description string `json:"description,omitempty"`
Capabilities []string `json:"capabilities"`
}
Card holds a parsed capability card: an agent's name, an optional description, and its capability list. Capabilities is an exported slice; Parse does not defensively copy it. This matches envelope.Message's exported slice fields, which carry the same caller-owned mutability with no defensive copy.
func Parse ¶ added in v0.5.0
Parse unmarshals data into a Card, then calls Validate. A JSON decode error, syntax or type mismatch, wraps the decode error with context. An invariant failure returns the Validate error unchanged. Parse ignores an unknown JSON field, matching envelope.Decode's forward-compatibility rule.
func (Card) Match ¶ added in v0.5.0
Match compares need against each capability with strings.EqualFold. It returns the matched capability and true on a hit. It returns an empty string and false when need is blank or no entry matches. Match does not trim need: a padded need, such as a leading space, does not match an entry with no padding. Match never calls Validate: on a Card with a duplicate-case capability entry, it returns the first slice-order match.
func (Card) Validate ¶ added in v0.5.0
Validate checks the card invariants. It rejects a blank Name after TrimSpace. It rejects an empty Capabilities list. It applies TrimSpace to each capability entry before the next three checks. It rejects a capability entry that is blank after trim, including a whitespace-only entry. It rejects a duplicate entry, compared with strings.EqualFold after trim: the same fold Match uses, so a Validate pass guarantees Match never hides a second, equal entry. It rejects a padded entry last, after the duplicate check, because Match compares the stored string and never hits a padded entry.
type Checkpoint ¶
type Checkpoint struct {
Status machine.Status
Record machine.InOut
Done []string
Skipped []string
Failed []string
}
Checkpoint is the full resumable state of a Run: the current machine.Status, the current machine.InOut record, the sorted step IDs of every step that resolved OutcomeSucceeded so far, the sorted step IDs of every step that resolved OutcomeSkipped so far, and the sorted step IDs of every step that resolved OutcomeFailed so far, whether or not a fallback caught that failure. Done's, Skipped's, and Failed's order is a sort, not a completion order: two steps that complete in one order can appear in the opposite order, if their IDs sort the other way. A route exclusion (applyRoute) or an admission skip (nextReadyGroup) is final regardless of the excluding step's later outcome; Skipped preserves that decision across a pause and a Resume the same way Done preserves a success.
Failed preserves only the resolved outcome of an already-caught failure; Resume does not restore the fallback bookkeeping a still- pending handler needs. A fallback step that resolves after a Resume still runs, admitted by AdmissionOnFailed the same way it would without a pause, but FailureFrom returns false inside it: the Failure a pre-pause fallback would have read does not survive the round trip. A Route exclusion that would have emptied a failure's last pending handler set, and so aborted the run with the recorded step error, instead resolves as an ordinary skip after a Resume; no error carries the lost failure across the boundary. Run never checkpoints this loss silently: see Run's doc comment.
func Decode ¶
func Decode(data []byte) (Checkpoint, error)
Decode parses JSON and validates the result. Checkpoint binds no guard or action, so no registry is needed.
func (Checkpoint) Encode ¶
func (c Checkpoint) Encode() ([]byte, error)
Encode serializes the checkpoint to JSON. It validates first. No registry: Record.Input and Record.Output are caller-owned any values. encoding/json decodes an any field back to map[string]interface{}, never the original concrete type. A caller whose Input or Output must survive a Checkpoint round-trip is responsible for using JSON-primitive-compatible types, or for re-hydrating its own concrete type after Decode.
func (Checkpoint) Validate ¶
func (c Checkpoint) Validate() error
Validate rejects an empty Status, a step ID named in more than one of Done, Skipped, and Failed (a step cannot resolve to two different outcomes), and an unsorted Done, Skipped, or Failed. Encode and Decode both call it.
type Confirm ¶
Confirm gates a step's ack. Run calls it after Fire moves the status, for a step named in no panel and for a one-member panel, and again for a chained step after its child workflow completes and the parent transition fires. Run does not call Confirm for a step in a panel of two or more members. A nil return means the ack confirmed; the walk advances.
type Definition ¶
type Definition struct {
// contains filtered or unexported fields
}
Definition holds a validated step graph and its panels. The fields are unexported; the type is immutable after New. roots carries the step IDs with no Needs, in declaration order.
func New ¶
func New(steps []Step, panels []Panel) (*Definition, error)
New builds a Definition and validates the step graph. It rejects an empty ID, a duplicate ID, a missing dependency, a panel that names an unknown step, a panel that names one step twice, a panel whose members disagree on To, and a step ID named in two panels. Kahn's algorithm rejects a cycle. After the graph is proven acyclic, New also rejects a panel where one member's Needs closure reaches a fellow member of the same panel. It rejects a chained step in a panel of two or more members. It rejects a Sub nesting depth above eight. It rejects a step that combines Sub and Route, a branch step with no dependent, a panel that names a branch step, and a panel that names a direct dependent of a branch step. It rejects an AdmissionOnFailed step with no needs, and an AdmissionOnFailed step named in a panel. It rejects an invalid Retry policy, a Retry policy combined with Sub, and a Retry policy on a panel member. It rejects an invalid Loop policy, a Loop policy combined with a nil Sub, and a Loop policy on a panel member. It rejects a step that sets both Payload and PayloadFrom, and a PayloadFrom on a member of a panel of two or more members. It deep-copies the input slices so later caller mutation cannot change the built graph.
func (Definition) Panels ¶
func (d Definition) Panels() []Panel
Panels returns a deep copy of the panel slice: each panel's member slice copies too. The copy keeps the definition immutable; callers cannot mutate the internal graph through it.
func (Definition) Roots ¶
func (d Definition) Roots() []string
Roots returns the root step IDs in declaration order. A root is a step with no Needs. The copy keeps the definition immutable; callers cannot mutate the internal slice.
func (Definition) Steps ¶
func (d Definition) Steps() []Step
Steps returns a deep copy of the step slice: each step's Needs, Retry, Loop, and Sub child copy recursively, at every depth. The copy keeps the definition immutable; mutating the returned graph cannot change the internal one.
type Failure ¶
Failure is the failed step's context a fallback step receives. Step names the failed step. Err is that step's recorded error.
type LoopPolicy ¶
LoopPolicy is a step's loop rule for its Sub child workflow. Guard reuses machine.Guard's exact type; a nil Guard means "always continue," matching machine's own nil convention. Max caps the iteration count; zero means unbounded, bounded only by the caller's own ctx. A negative Max is invalid. A child may end every iteration on one status: when the parent's standing already equals the child final, the re-entry fires no transition row.
func (LoopPolicy) Validate ¶
func (p LoopPolicy) Validate() error
Validate rejects Max < 0 with the pinned message "flow: loop: max must be at least 0". Validate has no step ID to report, so its message names no step; New builds a step-scoped message through loopValidateMessage.
type LoopState ¶
LoopState is the loop context a Guard closure reads. Iteration counts completed iterations, starting at zero before the first Guard call. Record carries the most recent child workflow's output.
type Monitor ¶ added in v0.5.0
type Monitor struct {
// contains filtered or unexported fields
}
Monitor tracks last-seen time per id against a fixed timeout. Mutex-guarded, safe for concurrent use. The zero value is not usable; create a Monitor with New. Monitor holds no clock of its own; every method takes the caller's time.Time.
func NewMonitor ¶ added in v0.5.0
NewMonitor creates a Monitor with a fixed timeout. A non-positive timeout wraps ErrInvalidOptions.
func (*Monitor) Alive ¶ added in v0.5.0
Alive reports whether id has beaten at least once and now.Sub(last) <= timeout. An id with no recorded beat is never alive. A beat timestamped after now (clock skew) makes now.Sub(last) negative, which is always <= timeout, so the id reads as alive; this is deliberate. See docs/history/heartbeat.md.
func (*Monitor) Beat ¶ added in v0.5.0
Beat records at as the last-seen time for id. A blank id after TrimSpace wraps ErrInvalidOptions. An at strictly before the id's previously recorded time wraps ErrStaleBeat and leaves the stored time unchanged. An at equal to or after the previously recorded time overwrites it.
func (*Monitor) Dead ¶ added in v0.5.0
Dead returns the sorted, defensively copied ids that have beaten at least once and are now past the timeout. Dead is level-triggered and at-least-once: it returns the same id on every call until Forget or a new Beat changes the state. Monitor performs no internal dedup.
type Panel ¶
type Panel []string
Panel is a group of step IDs that run together in parallel. The runner of a later step schedules a panel as one wave.
type Report ¶
type Report struct {
// contains filtered or unexported fields
}
Report is the result of one Run call: the final status, the final record, and every resolved step's Outcome.
func Resume ¶
func Resume( ctx context.Context, d *Definition, m *machine.Definition, checkpoint Checkpoint, confirm Confirm, bus *events.Bus, onCheckpoint func(Checkpoint), ) (Report, error)
Resume restarts a graph walk from a stored checkpoint. It seeds outcomes from checkpoint.Done (every listed ID set to OutcomeSucceeded), checkpoint.Skipped (every listed ID set to OutcomeSkipped), and checkpoint.Failed (every listed ID set to OutcomeFailed), cur from checkpoint.Status, and rec from checkpoint.Record, then continues the same graph walk Run uses. A route exclusion or an admission skip captured before the pause stays skipped after Resume; a step already resolved OutcomeFailed before the pause, caught or not, is never re-run after Resume; nextReadyGroup and admissionVerdict never re-evaluate a step already seeded in outcomes. Resume runs five entry checks in order, before any seeding happens: a nil d, a nil m, a nil confirm, a checkpoint that fails Validate, and a checkpoint.Done, checkpoint.Skipped, or checkpoint.Failed entry naming a step ID absent from d. The first failing check returns an error immediately; no step runs.
Resume performs no topology check across Done: it never confirms that a step named in Done has every one of its own Needs also named in Done. A topologically-inconsistent checkpoint is not rejected at entry; nextReadyGroup treats a missing prerequisite as unresolved and selects it to run again, and the resulting pickTransition or machine.Fire call fails, because checkpoint.Status no longer names a status the seeded walk can reach that step from. Resume returns that failure as an ordinary error.
Resume starts pending empty; it does not restore a still-pending fallback's bookkeeping from before the pause. See Checkpoint.
func Run ¶
func Run( ctx context.Context, d *Definition, m *machine.Definition, in machine.InOut, confirm Confirm, bus *events.Bus, onCheckpoint func(Checkpoint), ) (Report, error)
Run walks the step graph in topological order. A step named in no panel runs alone, in declaration order, as it did before panels existed; Run calls confirm for that step. See Confirm. A step named in a panel of one member runs alone the same way, and Run calls confirm for it too. A step named in a panel of two or more members runs as part of that panel's wave, once every member is ready; the wave fires every member's transition concurrently through the one shared row every member's homogeneous To selects. Run does not call confirm for a wave of two or more members. A step with a non-nil Sub runs its child workflow to completion, then uses the child final status as the parent step's target status; the child runs with a nil bus. Run keeps the current status and one record through the walk. Run rejects a nil d, a nil m, and a nil confirm at entry, checking d first, then m, then confirm, so a nil m never panics inside a d-nil or m-nil check.
onCheckpoint, when non-nil, fires immediately after each step or wave resolves OutcomeSucceeded, with a fresh Checkpoint. A nil onCheckpoint skips the call. See Checkpoint and Resume.
Before each step or wave starts, Run checks ctx for cancellation. A canceled ctx stops the walk before the next step starts and returns the pinned pause error, wrapping ctx.Err(); the last Checkpoint onCheckpoint delivered is the resume point. A step already running keeps running to its own completion, unless cancellation is observed inside a retry loop or a looped child workflow; in those cases the current step aborts and returns the context error.
Run returns a Report holding the final status, the final record, and every resolved step's Outcome. On every abort, Run returns the Report built so far, alongside the error. A step whose Fire fails or whose Confirm is rejected is marked OutcomeFailed before the return. A wave's shared, pre-spawn transition failure marks no member of that wave; a per-member Fire failure inside a wave marks every member OutcomeFailed, whether or not a dependent's AdmissionOnFailed rule catches the failure. A step admitted through a failed need (AdmissionOnFailed) is a fallback; Run injects a Failure into its transition's context, and FailureFrom reads it back. A caught Fire or Route failure continues down the fallback path; a Confirm rejection or a missing transition row stays fatal. Checkpoint's Failed field preserves an already-caught failure's outcome across a pause; see Checkpoint for what does not survive.
A panel member's Input and Output must be an immutable value, or a value the caller already cloned per step. Run's copy of each member's InOut is shallow: a map, a slice, or a pointer an Input or Output field holds is not copied. Two members that alias the same data still race if either mutates it in place.
func (Report) Outcome ¶
Outcome returns the outcome of the step named id, and whether that step resolved. The boolean is false when the step never resolved: the run aborted before it, or it sits in an unreached wave.
func (Report) Outcomes ¶
Outcomes returns a copy of every resolved step's Outcome, keyed by step ID. Caller mutation of the returned map cannot change the Report.
type RetryPolicy ¶
type RetryPolicy struct {
MaxAttempts int
BaseDelay time.Duration
MaxDelay time.Duration
Retryable func(error) bool
Jitter func(time.Duration) time.Duration
Sleep func(context.Context, time.Duration)
}
RetryPolicy is a step's retry rule for its Fire call. MaxAttempts counts every attempt, including the first; a value of 1 disables retry. BaseDelay is the first retry's backoff and must not be negative; zero means an immediate retry. MaxDelay clamps every computed backoff, so the exponential term cannot overflow time.Duration's range. Retryable, when non-nil, gates each failure before the next attempt; a nil Retryable retries every error. Jitter and Sleep are determinism hooks: Jitter perturbs NextDelay's clamped result, and Sleep waits between attempts. A nil Sleep defaults to a context-aware sleep. Sleep takes the run's ctx so a caller can cancel a pending backoff.
func (RetryPolicy) NextDelay ¶
func (p RetryPolicy) NextDelay(attempt int) time.Duration
NextDelay returns the backoff before the given retry attempt, one-indexed from the first retry. It doubles delay from BaseDelay one step per attempt above 1, checking the bound before each doubling instead of after: when delay > MaxDelay>>1, one more doubling would reach or overflow MaxDelay, so NextDelay sets delay to MaxDelay and stops doubling, without ever performing the overflow-prone multiply. A final clamp to MaxDelay covers the case where BaseDelay itself already exceeds MaxDelay. The bound holds only for a policy that passes Validate: a negative BaseDelay skips both the pre-doubling guard and the final clamp. Pure; no field mutation, no sleep, no randomness of its own. Jitter, when non-nil, applies to the clamped result last; NextDelay does not re-clamp Jitter's output, so a Jitter closure that returns a value above MaxDelay passes through unclamped. Re-clamping after Jitter is the caller's responsibility.
func (RetryPolicy) Validate ¶
func (p RetryPolicy) Validate() error
Validate enforces MaxAttempts >= 1, MaxDelay > 0, and BaseDelay >= 0. New calls the same check, through retryValidateMessage, for every step whose Retry is non-nil, folding the step's ID into the pinned message.
type Route ¶
Route picks the direct dependents a branch step's run keeps. It receives the branch step's post-fire status and record. It returns the IDs of the direct dependents to admit; every other direct dependent skips at once. A duplicate ID in the return collapses to one admission.
type Step ¶
type Step struct {
ID string
Needs []string
To string
Payload string
PayloadFrom func(rec machine.InOut) string
Sub *Definition
When Admission
Route Route
Retry *RetryPolicy
Loop *LoopPolicy
}
Step is one node in a workflow graph. ID names the step. Needs lists the prerequisite step IDs. To holds the target status a later step binds. Payload carries caller data; PayloadFrom derives it from the live record. New rejects a step with both Payload and PayloadFrom set, and a PayloadFrom on a member of a panel of two or more members; a one-member panel keeps the field. Sub nests a child workflow; when Sub is non-nil, Run ignores To and runs the child workflow to completion. A step with no Needs is a root. When sets the admission rule this step's needs must satisfy; the zero value is AdmissionOnSucceeded, so a skipped need skips this step. Route makes this step a branch step: after it fires, Run calls Route to pick which of this step's direct dependents the run keeps. Retry bounds and paces repeated Fire attempts; a nil Retry keeps the single-attempt behavior. New rejects a non-nil Retry combined with a non-nil Sub or panel membership. Loop, when non-nil, runs Sub more than once, gated by LoopPolicy.Guard, before this step's own transition and Confirm fire; New rejects a non-nil Loop combined with a nil Sub or panel membership. A same-final child re-enters without a row; see LoopPolicy.