Conditions¶

zeroth.contracts.conditions ¶

Conditional execution contracts.

This package handles deciding which path to take next in an agent workflow graph. It evaluates conditions on edges, figures out which branches are active, and describes the outcome. Recording outcomes onto a run is runtime-owned: see zeroth.runtime.runs.condition_recorder.

ConditionBinder ¶

Creates runtime bindings from graph edges.

Use this when you need to convert a graph (or a set of edges) into ConditionBinding objects that the evaluator and branch resolver can process.

bind_graph ¶

bind_graph(graph: Graph) -> list[ConditionBinding]

Create bindings for every edge in the given graph.

bind_edges ¶

bind_edges(
    graph_id: str, edges: Iterable[Edge]
) -> list[ConditionBinding]

Create bindings for a specific set of edges within a graph.

bind_edge ¶

bind_edge(graph_id: str, edge: Edge) -> ConditionBinding

Convert a single edge into a ConditionBinding ready for evaluation.

ConditionBinding ¶

Bases: BaseModel

Links a graph edge to its condition so it can be evaluated at runtime.

Contains everything the evaluator needs: which graph and edge this came from, the source and target nodes, the condition to check, and whether the edge is enabled.

BranchResolution ¶

Bases: BaseModel

The result of figuring out which outgoing edges from a node are active.

After evaluation, this tells you which edges passed their conditions (active), which did not (suppressed), and which nodes to visit next. If no edges are active, terminal_reason explains why execution stopped.

BranchResolver ¶

BranchResolver(
    *,
    binder: ConditionBinder | None = None,
    evaluator: ConditionEvaluator | None = None,
)

Figures out which outgoing edges from a node should be followed.

For a given node, it looks at all outgoing edges, evaluates their conditions, checks visit limits, and returns a BranchResolution that says which edges are active and which nodes to visit next.

resolve ¶

resolve(
    graph: Graph,
    source_node_id: str,
    context: ConditionContext
    | Mapping[str, Any]
    | None = None,
    *,
    traversal_state: TraversalState | None = None,
) -> BranchResolution

Evaluate all outgoing edges from a node and return which ones are active.

Edges can be suppressed because they are disabled, their condition evaluated to false, or a visit limit would be exceeded.

NextStepPlan ¶

Bases: BaseModel

Describes what the workflow should do next after the current node.

Contains the list of next node IDs to execute, the full branch resolution details, and the current traversal state. If there are no next nodes, terminal_reason explains why execution is finished.

NextStepPlanner ¶

NextStepPlanner(resolver: BranchResolver | None = None)

Determines which node(s) to execute next in a workflow.

This is a higher-level wrapper around BranchResolver. Given the current node, it resolves branches and produces a NextStepPlan that the runtime can use to decide what to do next.

plan ¶

plan(
    graph: Graph,
    current_node_id: str,
    context: ConditionContext
    | Mapping[str, Any]
    | None = None,
    *,
    traversal_state: TraversalState | None = None,
) -> NextStepPlan

Resolve branches from the current node and return a plan for what to do next.

ConditionContext ¶

Bases: BaseModel

Holds all the runtime data that conditions can reference during evaluation.

Think of this as the "world state" that condition expressions can read from. It includes the current payload, workflow state, user-defined variables, visit counts, and the traversal path so far.

namespace ¶

namespace() -> dict[str, Any]

Build the dictionary of names that condition expressions can access.

For example, an expression like "payload.score > 10" will look up "payload" in this namespace to find the actual payload dict.

ConditionEvaluator ¶

Evaluates graph conditions against runtime context data.

This is the main entry point for condition evaluation. Give it a condition (from a graph edge) and a context (the current runtime data), and it will tell you whether the condition matched and what value it produced.

evaluate ¶

evaluate(
    condition: Condition,
    context: ConditionContext | Mapping[str, Any],
    *,
    condition_id: str | None = None,
    edge_id: str | None = None,
    source_node_id: str | None = None,
    target_node_id: str | None = None,
    metadata: Mapping[str, Any] | None = None,
) -> RunConditionResult

Evaluate a single condition and return a result with match info and details.

The returned RunConditionResult tells you whether the condition matched, which edge was selected, and includes debugging details like the expression and operand references.

evaluate_operand_ref ¶

evaluate_operand_ref(
    ref: str, context: ConditionContext | Mapping[str, Any]
) -> Any

Look up a single operand reference (like "payload.score") in the context.

ConditionOutcome ¶

Bases: BaseModel

The full result of evaluating a condition, including where it came from.

Wraps a RunConditionResult with extra info about which graph, edge, and nodes were involved. Useful for debugging and auditing why a particular branch was taken.

RunConditionResult ¶

Bases: BaseModel

A record of a condition (branching decision) that was evaluated during a run.

When a graph has conditional edges, this captures which condition was checked, whether it matched, and which edge was selected. The run domain republishes it as part of the persisted run surface.

TraversalState ¶

Bases: BaseModel

Tracks how many times each node and edge has been visited during a run.

The branch resolver uses this to enforce visit limits and prevent infinite loops. The 'path' list records the order nodes were visited.