flow

func (ChoosePlanner) Plan ¶

func (ChoosePlanner) Plan(
	invocator op.PlanInvocator,
	receiverType op.ProviderReceiverType,
	method *op.Method,
	args []any,
	kwargs map[string]any,
	annotations map[string]any,
	onError *op.Subgraph,
	onRetry *op.Subgraph,
	retryPolicy *op.RetryPolicy,
	transitionPolicy *op.TransitionPolicy,
) (op.ExecutableUnit, error)

Plan implements op.Planner for flow.Provider.Choose.

Consumes the `default=` kwarg (a body — a list of invocations, the plan.subgraph construction) and the positional cases (each a `*Case` from `plan.case(when=, then=)`); every other kwarg lands in the subgraph's unified slot map as a frame binding, like SubgraphPlanner. Children are laid out when₀, then₀, …, default and the decision tree's guarded edges are emitted alongside; op.NewSubgraph seals the topology and op.ValidateGraph enforces the guarded-subgraph invariant at the boundaries.

Parameters:

• `invocator`: the session host; consulted only to desugar a lambda `default=` into a function.call invocation.
• `receiverType`: the flow planning provider.
• `method`: the registered descriptor for Choose.
• `args`: positional arguments converted starlark → Go; every entry must be a `*Case`.
• `kwargs`: keyword arguments converted starlark → Go; `default=` is required and reserved (a body, or a lambda desugared to one), the rest are frame bindings.
• `annotations`: plan-time annotations applied to the subgraph at construction.
• `onError`: the failure-handler subgraph applied to the subgraph at construction, or nil.
• `onRetry`: the per-attempt retry-handler subgraph applied to the subgraph at construction, or nil.
• `retryPolicy`: the retry policy applied to the subgraph at construction, or nil.
• `transitionPolicy`: the transition policy applied to the subgraph at construction, or nil.

Returns:

• `op.ExecutableUnit`: the constructed choose-shaped *op.Subgraph.
• `error`: non-nil when `receiverType` or `method` is nil, `default=` is missing or malformed, or a positional argument is not a `*Case`.

func (GatherPlanner) Plan ¶

func (GatherPlanner) Plan(
	_ op.PlanInvocator,
	receiverType op.ProviderReceiverType,
	method *op.Method,
	args []any,
	kwargs map[string]any,
	annotations map[string]any,
	onError *op.Subgraph,
	onRetry *op.Subgraph,
	retryPolicy *op.RetryPolicy,
	transitionPolicy *op.TransitionPolicy,
) (op.ExecutableUnit, error)

Plan implements op.Planner for flow.Provider.Gather.

Reserves `body=` (intercepted before the param walk and adopted via [addBodyChildren], not stamped as a slot), then walks the method's declared parameter list and maps positional `args` and named `kwargs` onto the subgraph's slot map: the items parameter consumes a matching kwarg or positional, the `**kwargs` sink collects every unconsumed kwarg (including `limit=`), and Variable / Invocation / Promise values route through [projectKwargValue] before stamping. After the walk, stamps `item` as a frame-local sentinel (`nil` value) so child slot-references to `plan.variable("item")` are satisfied by the per-iteration frame rather than the bubble-up surface.

Parameters:

• `invocator`: the session host (unused — Gather constructs its subgraph from `args` / `kwargs` alone).
• `receiverType`: the flow planning provider.
• `method`: the registered descriptor for Gather.
• `args`: positional arguments converted starlark → Go.
• `kwargs`: keyword arguments converted starlark → Go (reserved entries removed).
• `onError`: the failure-handler subgraph applied to the subgraph at construction, or nil.
• `onRetry`: the per-attempt retry-handler subgraph applied to the subgraph at construction, or nil.
• `retryPolicy`: the retry policy applied to the subgraph at construction, or nil.
• `transitionPolicy`: the transition policy applied to the subgraph at construction, or nil.

Returns:

• op.ExecutableUnit: the constructed gather-shaped *op.Subgraph.
• `error`: non-nil when `receiverType` or `method` is nil, when `body=` is malformed, or a required parameter is missing.

func (*Provider) Choose ¶

func (p *Provider) Choose(
	activation *op.ActivationRecord,
	kwargs map[string]any,
) (any, *op.RecoveryStack, error)

Choose executes the choose subgraph — and nothing else.

The choose's logic lives entirely in the graph ChoosePlanner builds from the case statements: a conditional ladder in which each `when`-subgraph's truthy edge routes to its own `then`-subgraph and its falsy edge to the next case's `when`, with the last falsy edge routing to the default. Executing that topology is what produces the first-truthy short-circuit; Choose carries **no** selection logic of its own — it walks the children exactly as Provider.Subgraph does. See the step-10 design doc (`docs/plans/extract-starlark-from-op/phase-8/steps/10-plan-choose.md`).

A zero-case choose (default only) carries no guarded edges and takes the ordinary run-all walk — defined behavior, per the switch-statement precedent: the single default child runs and its result is the choose result.

Parameters:

• `activation`: the per-dispatch record; `activation.CallerID` is the choose `*op.Subgraph` and `activation.Stack` is the executor-owned recovery stack.
• `kwargs`: frame-binding kwargs, layered onto the inherited variable frame like Provider.Subgraph.

Returns:

• `any`: the result of whichever branch the graph walk reached.
• *op.RecoveryStack: `activation.Stack`, carrying the branches that ran.
• `error`: non-nil on a child failure or when `activation.CallerID` is not a `*op.Subgraph`.

func (*Provider) CompensateChoose ¶

func (p *Provider) CompensateChoose(activation *op.ActivationRecord, stack *op.RecoveryStack) error

CompensateChoose unwinds the recovery state captured by a successful Provider.Choose call.

Today this is structurally a delegation to op.RecoveryStack.Unwind — the stack is empty until phase-8 / step 16 lands the executor-side traversal that pushes the chosen branch's compensation entries into it. Once that's wired, this body still does the same thing: unwinds whatever the executor populated.

Parameters:

• `activation`: the per-dispatch record; supplies the *op.RuntimeEnvironment passed to op.RecoveryStack.Unwind.
• `stack`: the op.RecoveryStack returned by the forward Choose call.

Returns:

• `error`: non-nil if the unwind fails.

func (*Provider) CompensateGather ¶

func (p *Provider) CompensateGather(activation *op.ActivationRecord, stack *op.RecoveryStack) error

CompensateGather unwinds the per-iteration recovery stacks accumulated by a successful Gather.

Called by the executor when the parent stack unwinds and hits gather's compensable entry. The single returned op.RecoveryStack holds one nested substack per iteration in completion order; op.RecoveryStack.Unwind walks the entries LIFO so the iteration that finished last (and therefore produced the freshest side effects) undoes first, mirroring standard compensation semantics.

Parameters:

• `activation`: the per-dispatch record; supplies the *op.RuntimeEnvironment passed to op.RecoveryStack.Unwind.
• `stack`: the gather stack as returned by Gather (one nested substack per iteration in completion order).

Returns:

• `error`: a joined error across any substack that failed to unwind; nil on total success.

func (*Provider) CompensateSubgraph ¶

func (p *Provider) CompensateSubgraph(activation *op.ActivationRecord, stack *op.RecoveryStack) error

CompensateSubgraph unwinds the subgraph's local saga stack as a single transactional unit.

The stack carries one entry per compensable child (a Receipt or a deeper nested substack). Unwind walks LIFO and dispatches per entry kind, recursing into nested substacks. Until phase-8 / step 16 wires the executor-side population, the stack is empty and Unwind is a no-op.

Parameters:

• `activation`: the per-dispatch record; supplies the *op.RuntimeEnvironment passed to op.RecoveryStack.Unwind.
• `stack`: the op.RecoveryStack returned by the forward Subgraph call.

Returns:

• `error`: non-nil if any entry fails to unwind.

func (*Provider) CompensateWaitUntil ¶

func (p *Provider) CompensateWaitUntil(activation *op.ActivationRecord, stack *op.RecoveryStack) error

CompensateWaitUntil unwinds the recovery state captured by a successful Provider.WaitUntil call.

The stack carries the truthy run's stamped substack (falsy polls left nothing behind); unwinding it rolls back whatever that run did.

Parameters:

• `activation`: the per-dispatch record; supplies the *op.RuntimeEnvironment passed to op.RecoveryStack.Unwind.
• `stack`: the op.RecoveryStack returned by the forward WaitUntil call.

Returns:

• `error`: non-nil if the unwind fails.

func (*Provider) Complete ¶

func (p *Provider) Complete(activationRecord *op.ActivationRecord, output any) any

Complete is the healthy conclusion of a graph path — an early return from the enclosing body.

It behaves like a `return` statement in a func: it ends the body it executes in with `output` as that body's result. Completion is a Phase event, never a condition flip. The early-return control effect — stopping the enclosing walk so the remaining units never dispatch, with no receipts for the remainder and everything already done kept — is applied by the walk that recognizes this action.

+devlore:defaults output=nil

Parameters:

• `activationRecord`: the per-dispatch record (present for the framework's activation-first convention).
• `output`: optional output value.

Returns:

• `any`: the output value.

func (*Provider) Degraded ¶

func (p *Provider) Degraded(activationRecord *op.ActivationRecord, format string, args []any, kwargs map[string]any) string

Degraded flips the run's condition to op.ConditionDegraded while allowing execution to continue.

A typed condition-flip driver: it submits the flip through the activation's op.ActivationRecord.Transition with op.ReasonDegraded (Phase passes through unchanged — a condition-only move) and returns the rendered message. The submission's error is discarded. This is how a consumer opts a path into degrade-and-continue — place a flow.degraded node in an error action.

Parameters:

• `activationRecord`: the per-dispatch record; supplies the op.ActivationRecord.Transition delegate.
• `format`: format string.
• `args`: positional format arguments.
• `kwargs`: keyword arguments for template rendering.

Returns:

• `string`: the rendered warning message.

func (*Provider) Failed ¶

func (p *Provider) Failed(activationRecord *op.ActivationRecord, format string, args []any, kwargs map[string]any) string

Failed asserts the run's condition to op.ConditionExecutionFailed — a hard failure by the subgraph.

A typed condition-flip driver mirroring Provider.Degraded: it submits the flip through the activation's op.ActivationRecord.Transition with op.ReasonFailed (Phase passes through unchanged) and returns the rendered message as the node's result. It no longer short-circuits with an error — the run's op.TransitionPolicy drives the reaction (stop at the floor), and the execution_failed flip bypasses OnError (a hard assertion is not an incidental failure to absorb). The submission's error is discarded — a rejected flip means the run is already at a worse condition.

Parameters:

• `activationRecord`: the per-dispatch record; supplies the op.ActivationRecord.Transition delegate.
• `format`: format string.
• `args`: positional format arguments.
• `kwargs`: keyword arguments for template rendering.

Returns:

• `string`: the rendered failure message (the node's result), mirroring Provider.Degraded.

func (*Provider) Gather ¶

func (p *Provider) Gather(
	activation *op.ActivationRecord,
	items []any,
	kwargs map[string]any,
) (any, *op.RecoveryStack, error)

Gather runs the activation's subgraph body once per item, concurrently up to `limit`, and nests one stamped substack per iteration onto the gather's own recovery stack.

Gather is a quantifier over Subgraph (phase-8 step 31.2): it runs the same body-walk Subgraph runs ([walkSubgraphChildren]) N times — once per item — each on its own child stack with a frame that binds `item` to the iteration value. The two builtin parameter names, `items` and `limit`, are consumed here and stripped from the per-iteration frame ([buildIterationFrame]); bodies read the iteration value via `plan.variable("item")`. Concurrency is goroutine-per-item throttled by a `limit`-sized semaphore; each goroutine walks the body on its **own** child stack, so `activation.Stack` stays single-writer — stamping and nesting run on the dispatching goroutine, in index order.

Each iteration's child stack is [op.RecoveryStack.Stamp]ed with its identity (`"<gatherID>#<i>"`), result, and status, then nested onto `activation.Stack` (op.RecoveryStack.PushNested). The gather returns that stack as its compensator, exactly as Subgraph returns `activation.Stack`; `CompensateGather` unwinds it. On resume of a *paused* gather the stack comes back carrying the prior iterations' stamped substacks, and Gather classifies each iteration against it (op.RecoveryStack.NestedStackByUnitID): a completed run replays its stamped result and is skipped, a paused run adopts its partial substack and re-enters, a never-run iteration runs fresh.

On any iteration error Gather does not self-unwind: it returns the stamped stack as the compensator alongside the error, so the executor rolls it back (or, on ErrPaused, checkpoints it) — the same contract as Subgraph.

Parameters:

• `activation`: the per-dispatch record; cancellation flows through `activation.Context` and a scoped child is derived for this gather's iterations. `activation.Variables` is the parent frame the per-iteration frames derive from, `activation.Stack` is the gather's own (resume-adopted) stack, and `activation.CallerID` must be a `*op.Subgraph` (the gather's bound unit).
• `items`: the list of items to iterate over.
• `kwargs`: catchall sink — `limit` (max concurrent iterations; defaults to platform concurrency when non-positive) is read here. Other keys are reserved for future extension.

Returns:

• `any`: a []any of terminal results from each iteration, indexed by original item order; nil on error.
• *op.RecoveryStack: `activation.Stack`, carrying one stamped substack per iteration in index order.
• `error`: non-nil if any iteration failed or paused, or the body is malformed.

func (*Provider) Subgraph ¶

func (p *Provider) Subgraph(
	activation *op.ActivationRecord,
	kwargs map[string]any,
) (any, *op.RecoveryStack, error)

Subgraph dispatches the children of a `plan.subgraph(...)` container in declaration order.

Reached from [op.GraphExecutor.executeSubgraph]'s bound-action path: the executor's shim resolves the subgraph's slots, builds the op.ActivationRecord with the subgraph as `Unit`, installs the child-dispatch closure, and calls op.Action.Do. This method walks `activation.CallerID.(*op.Subgraph) .Children()` and dispatches each child via op.ActivationRecord.DispatchChild, which routes through the parent executor (preserving observability hooks, the resolved variable map, and the active results map for promise resolution).

Per-child retry policy: the child's own op.ExecutableUnit.RetryPolicy drives retry attempts (interim; the frame-chain `effectiveRetryPolicy` helper is pending). Nil policy means one attempt with no retry. Delays between retries are computed via op.RetryPolicy.ComputeDelay; cooperative cancellation via `activation.Context` aborts the wait.

Failure handling: a child's OnError / OnRetry handlers are consumed at the child's own dispatch by the executor (the shared dispatchWithPolicy seam), invisible to this walk — so an absorbed failure never reaches here, and a standing failure short-circuits the walk with the child's error.

`items` iteration is not yet implemented; passing a non-empty `items=` to `plan.subgraph(...)` is an error today. The pure-container shape (children walk only) is what this method supports.

Parameters:

• `activation`: the per-dispatch *op.ActivationRecord the executor built. `activation.CallerID` must type-assert to *op.Subgraph; `activation.DispatchChild` must be installed (both invariants are the executor's contract on the bound-action path).
• `items`: the resolved value of the `items=` kwarg from `plan.subgraph(items=[...], body=[...])`. Must be empty for now.
• `kwargs`: frame-binding kwargs (every `plan.subgraph(...)` kwarg except the slot/parameter names declared on this method). Read by children that reference them via `plan.variable(name)`; this method does not consume them directly.

Returns:

• `any`: the last child's terminal result, mirroring the structural-container contract in op.Subgraph.Execute — the leaf unit's output bubbles up through the subgraph to the parent. Nil for a zero-child subgraph. Children's results also flow into the parent results map via op.ActivationRecord.DispatchChild.
• *op.RecoveryStack: the subgraph-local saga stack. Children's compensations accumulated here via the installed `DispatchChild` closure; the executor pushes this nested onto the parent stack as the subgraph's compensator.
• `error`: non-nil on (a) `items` iteration request, (b) `activation.CallerID` not a *op.Subgraph, (c) any child's exhausted-retry failure (with the original child error wrapped).

func (*Provider) WaitUntil ¶

func (p *Provider) WaitUntil(
	activation *op.ActivationRecord,
	timeout, interval time.Duration,
	kwargs map[string]any,
) (any, *op.RecoveryStack, error)

WaitUntil polls the activation's body subgraph until its result is truthy, then returns that result.

WaitUntil is a quantifier over Subgraph (phase-8 step 12): each poll runs the body — the same walk Subgraph runs ([walkSubgraphChildren]) — on its own scratch child stack, and the body-subgraph's result (its last child's) is evaluated with op.IsTruthy. A falsy poll's stack is dropped unrecorded (the body is expected side-effect-free; nothing enforces it — a side-effecting poll is a plan defect, the same by-design stance as gather's concurrency contract), and the walk sleeps `interval` before re-running. The truthy poll's stack is stamped with this unit's ID and nested — the trace reads like a subgraph that ran its body once, because semantically that is what a completed wait_until is. Timeout fails the unit with a plain error carrying the poll count and the last falsy result; a body error fails immediately (a crashed probe is not "not ready"). Resume: a completed wait_until replays its stamped result upstream like any unit; an interrupted one left nothing behind and re-enters fresh with its full budget (settled 2026-07-02).

Parameters:

• `activation`: the per-dispatch record; `activation.CallerID` is the wait-until *op.Subgraph and `activation.Stack` is the executor-owned recovery stack.
• `timeout`: the mandatory polling budget; expiry fails the unit.
• `interval`: the sleep between polls; non-positive falls back to the planner's default.
• `kwargs`: frame-binding kwargs, layered onto the inherited variable frame like Provider.Subgraph.

Returns:

• `any`: the truthy poll's result.
• `*op.RecoveryStack`: `activation.Stack`, carrying the truthy run's stamped substack.
• `error`: non-nil on a body failure, cancellation, or timeout.

func (SubgraphPlanner) Plan ¶

func (SubgraphPlanner) Plan(
	_ op.PlanInvocator,
	receiverType op.ProviderReceiverType,
	method *op.Method,
	_ []any,
	kwargs map[string]any,
	annotations map[string]any,
	onError *op.Subgraph,
	onRetry *op.Subgraph,
	retryPolicy *op.RetryPolicy,
	transitionPolicy *op.TransitionPolicy,
) (op.ExecutableUnit, error)

Plan implements op.Planner for flow.Provider.Subgraph.

Parameters:

• `invocator`: the session host (unused today; future kwarg-classification rules may consult it).
• `receiverType`: the flow planning provider.
• `method`: the registered descriptor for Subgraph.
• `args`: positional arguments; unused — flow.Subgraph has no positional surface today.
• `kwargs`: keyword arguments converted starlark → Go (reserved entries removed); `body=` becomes children, every other entry becomes a slot value.
• `onError`: the failure-handler subgraph applied to the subgraph at construction, or nil.
• `onRetry`: the per-attempt retry-handler subgraph applied to the subgraph at construction, or nil.
• `retryPolicy`: the retry policy applied to the subgraph at construction, or nil.
• `transitionPolicy`: the transition policy applied to the subgraph at construction, or nil.

Returns:

• op.ExecutableUnit: the constructed *op.Subgraph with classified kwargs applied.
• `error`: non-nil if `body=` is not a list, contains a non-invocation element, or `items=` is malformed.

func (WaitUntilPlanner) Plan ¶

func (WaitUntilPlanner) Plan(
	invocator op.PlanInvocator,
	receiverType op.ProviderReceiverType,
	method *op.Method,
	_ []any,
	kwargs map[string]any,
	annotations map[string]any,
	onError *op.Subgraph,
	onRetry *op.Subgraph,
	retryPolicy *op.RetryPolicy,
	transitionPolicy *op.TransitionPolicy,
) (op.ExecutableUnit, error)

Plan implements op.Planner for flow.Provider.WaitUntil.

Parameters:

• `invocator`: the session host; consulted only to desugar a lambda `body=` into a function.call invocation.
• `receiverType`: the flow planning provider.
• `method`: the registered descriptor for WaitUntil.
• `args`: positional arguments; unused — flow.WaitUntil is kwargs-driven.
• `kwargs`: keyword arguments converted starlark → Go; `body=` and `timeout=` are required and reserved, `interval=` is reserved and defaulted, the rest are frame bindings.
• `annotations`: plan-time annotations applied to the subgraph at construction.
• `onError`: the failure-handler subgraph applied to the subgraph at construction, or nil.
• `onRetry`: the per-attempt retry-handler subgraph applied to the subgraph at construction, or nil.
• `retryPolicy`: the retry policy applied to the subgraph at construction, or nil.
• `transitionPolicy`: the transition policy applied to the subgraph at construction, or nil.

Returns:

• `op.ExecutableUnit`: the constructed wait-until-shaped *op.Subgraph.
• `error`: non-nil when `receiverType` or `method` is nil, `body=` or `timeout=` is missing or malformed, or a duration does not parse.