first reducer sweep of failure

i will put second reducer sweep of logic error or second of success

src/wf_authoring/__init__.py

@@ -12,6 +12,7 @@ from .dsl import (
     input,
     input_path,
     merge_maps,
+    not_,
     state,
     state_path,
 )
@@ -119,6 +120,7 @@ __all__ = [
     "rename_fields",
     "runtime_error",
     "node",
+    "not_",
     "outcome",
     "reducer",
     "state",

src/wf_authoring/builder/core.py

@@ -5,6 +5,7 @@ from dataclasses import dataclass, field
 from typing import Any, Literal, cast
 import warnings
 
+from wf_authoring.ops.values import runtime_error
 from wf_core import (
     ConditionNode,
     Edge,
@@ -19,10 +20,12 @@ from wf_core import (
 )
 from wf_core.errors import WorkflowExecutionError
 from wf_core.models.conditions import Condition as CoreCondition
+from wf_core.runtime.ops.merges import ReducerDefinition
 
 from ..dsl import Expr, PathArg, PathExpr, compile_condition
 from ..nodes.callables import SyncRegistryHandler
 from ..nodes.registry import build_registry
+from ..reducers import ReducerCatalog
 from ..schemas import SchemaLike, StateSchemaLike, schema_ref_from, state_schema_from
 from ..nodes import NodeSpec
 from .ids import next_step_id, slug_id
@@ -43,6 +46,7 @@ class WorkflowBuilder:
     state_schema: StateSchemaLike
     output_schema: SchemaLike
     start: str | None = None
+    reducers: ReducerCatalog | Mapping[str, ReducerDefinition] | None = None
     node_specs: dict[str, NodeSpec[Any, Any]] = field(default_factory=dict)
     nodes: list[Any] = field(default_factory=list)
     edges: list[Edge] = field(default_factory=list)
@@ -100,6 +104,19 @@ class WorkflowBuilder:
         """Export handlers for all node specs used by this builder."""
         return build_registry(*self.node_specs.values())
 
+    def reducer_registry(self) -> dict[str, ReducerDefinition]:
+        """Export custom reducers attached to this builder.
+
+        The core runtime always falls back to built-in reducers. This method
+        returns only the authored additions/overrides so callers can pass the
+        same runtime environment that `execute()` uses.
+        """
+        if self.reducers is None:
+            return {}
+        if isinstance(self.reducers, ReducerCatalog):
+            return dict(self.reducers.definitions)
+        return dict(self.reducers)
+
     def execute(self, workflow_input: dict[str, Any]) -> RunState:
         """Compile and execute this workflow with its used node registry.
 
@@ -107,7 +124,12 @@ class WorkflowBuilder:
         callers that need custom registries, persistence, or resume behavior should
         call wf_core execution functions directly.
         """
-        return execute_workflow(self.compile(), workflow_input, self.registry())
+        return execute_workflow(
+            self.compile(),
+            workflow_input,
+            self.registry(),
+            reducers=self.reducer_registry(),
+        )
 
     def condition(
         self, *, id: str | None = None, check: CoreCondition | Expr
@@ -211,24 +233,43 @@ class WorkflowBuilder:
         return resolved_targets
 
     def route(
+        self,
+        value: PathExpr | Expr,
+        cases: Mapping[object, BranchRef],
+        *,
+        id: str | None = None,
+        default: BranchRef = runtime_error,
+    ) -> dict[object, StepRef]:
+        """Route graph data by equality checks or one boolean condition.
+
+        `branch()` wires outcomes already produced by a node. `route()` is the
+        companion for ordinary graph data: with a path, it compares that path
+        against case values; with a condition expression, it expects boolean
+        `True`/`False` cases and emits one condition node.
+        """
+        if isinstance(value, Expr):
+            return self._route_condition(value, cases, id=id, default=default)
+
+        return self._route_value(value, cases, id=id, default=default)
+
+    def _route_value(
         self,
         value: PathExpr,
         cases: Mapping[object, BranchRef],
         *,
+        id: str | None,
         default: BranchRef,
     ) -> dict[object, StepRef]:
-        """Route graph data by equality checks compiled to condition nodes.
-
-        `branch()` wires outcomes already produced by a node. `route()` is the
-        companion for ordinary graph data: it compares one path against case
-        values, expands those checks into a condition chain, and wires the first
-        matching target plus a required fallback.
-        """
+        """Expand value cases into an ordered chain of equality checks."""
         resolved_targets: dict[object, StepRef] = {}
         default_target = self.use(default) if is_node_spec(default) else default
         previous_condition: ConditionNode | None = None
+        condition_base = id or "condition"
         for case_value, target in cases.items():
-            condition = self.condition(check=value.eq(case_value))
+            condition = self.condition(
+                id=self._next_step_id(condition_base),
+                check=value.eq(case_value),
+            )
             resolved = self.use(target) if is_node_spec(target) else target
             if previous_condition is not None:
                 self.connect(previous_condition, "false", condition)
@@ -241,6 +282,30 @@ class WorkflowBuilder:
         resolved_targets["default"] = cast(StepRef, default_target)
         return resolved_targets
 
+    def _route_condition(
+        self,
+        condition: Expr,
+        cases: Mapping[object, BranchRef],
+        *,
+        id: str | None,
+        default: BranchRef,
+    ) -> dict[object, StepRef]:
+        """Route a boolean condition expression through true/false outcomes."""
+        invalid_cases = [case for case in cases if not isinstance(case, bool)]
+        if invalid_cases:
+            raise ValueError("condition route cases must be boolean True/False keys")
+        if not cases:
+            raise ValueError("WorkflowBuilder.route requires at least one case")
+
+        condition_node = self.condition(id=id, check=condition)
+        resolved_targets: dict[object, StepRef] = {}
+        for case_value, outcome in ((True, "true"), (False, "false")):
+            target = cases.get(case_value, default)
+            resolved = self.use(target) if is_node_spec(target) else target
+            self.connect(condition_node, outcome, cast(StepRef, resolved))
+            resolved_targets[case_value] = cast(StepRef, resolved)
+        return resolved_targets
+
     def compile(self) -> Workflow:
         if self.start is None:
             raise WorkflowExecutionError(

src/wf_authoring/dsl/__init__.py

@@ -6,6 +6,7 @@ from .conditions import (
     exists,
     expr,
     input,
+    not_,
     state,
 )
 from .mapping import PathArg, bind_fields, bind_state, merge_maps, normalize_path
@@ -28,6 +29,7 @@ __all__ = [
     "input_path",
     "merge_maps",
     "normalize_path",
+    "not_",
     "state",
     "state_path",
 ]

src/wf_authoring/dsl/conditions.py

@@ -55,7 +55,9 @@ class Expr:
 class PathExpr:
     path: str
 
-    def _binary(self, op: Literal["eq", "ne", "gt", "lt"], other: object) -> Expr:
+    def _binary(
+        self, op: Literal["eq", "ne", "gt", "ge", "lt", "le"], other: object
+    ) -> Expr:
         return Expr(
             BinaryCondition(
                 op=op,
@@ -73,9 +75,15 @@ class PathExpr:
     def gt(self, other: object) -> Expr:
         return self._binary("gt", other)
 
+    def ge(self, other: object) -> Expr:
+        return self._binary("ge", other)
+
     def lt(self, other: object) -> Expr:
         return self._binary("lt", other)
 
+    def le(self, other: object) -> Expr:
+        return self._binary("le", other)
+
     def __eq__(self, other: object) -> Expr:  # pyright: ignore[reportIncompatibleMethodOverride] # type: ignore[override] # ty: ignore[invalid-method-override]
         return self._binary("eq", other)
 
@@ -85,9 +93,15 @@ class PathExpr:
     def __gt__(self, other: object) -> Expr:
         return self.gt(other)
 
+    def __ge__(self, other: object) -> Expr:
+        return self.ge(other)
+
     def __lt__(self, other: object) -> Expr:
         return self.lt(other)
 
+    def __le__(self, other: object) -> Expr:
+        return self.le(other)
+
 
 def expr(value: PathExpr | GraphPath) -> PathExpr:
     if isinstance(value, PathExpr):
@@ -111,6 +125,11 @@ def exists(value: PathExpr | GraphPath) -> Expr:
     return Expr(ExistsCondition(op="exists", path=_path_str(value)))
 
 
+def not_(value: Condition | Expr) -> Expr:
+    """Negate a condition without relying on Python's visually subtle `~expr`."""
+    return Expr(NotCondition(op="not", arg=compile_condition(value)))
+
+
 def compile_condition(value: Condition | Expr) -> Condition:
     if isinstance(value, Expr):
         return value.to_condition()

src/wf_authoring/reducers/catalog.py

@@ -18,13 +18,10 @@ class ReducerCatalog:
     def from_reducers(cls, *reducers: AuthoredReducer) -> "ReducerCatalog":
         return cls(
             definitions={
-                reducer.definition.spec.name: reducer.definition
-                for reducer in reducers
+                reducer.definition.spec.name: reducer.definition for reducer in reducers
             }
         )
 
     @property
     def specs(self) -> dict[str, ReducerSpec]:
-        return {
-            name: definition.spec for name, definition in self.definitions.items()
-        }
+        return {name: definition.spec for name, definition in self.definitions.items()}

src/wf_authoring/schemas.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
-from dataclasses import dataclass
+from collections.abc import Mapping
+from dataclasses import dataclass, field
 from typing import Any, Iterator
 
 from pydantic import BaseModel, TypeAdapter
@@ -9,23 +10,26 @@ from wf_core import ReducerRef, SchemaRef, StateField, StateSchema
 
 SchemaLike = SchemaRef | type[BaseModel] | type[Any] | dict[str, Any]
 StateSchemaLike = StateSchema | type[BaseModel] | type[Any] | dict[str, Any]
+ReducerLike = str | ReducerRef | Mapping[str, Any]
 
 
 @dataclass(frozen=True, slots=True)
 class StateFieldMetadata:
     """Authoring metadata attached to BaseModel state fields."""
 
-    reducer: str = "wf.std.replace"
+    reducer: ReducerRef = field(
+        default_factory=lambda: ReducerRef(name="wf.std.replace")
+    )
     trace: bool = True
 
 
 def state_field(
     *,
-    reducer: str = "wf.std.replace",
+    reducer: ReducerLike = "wf.std.replace",
     trace: bool = True,
 ) -> StateFieldMetadata:
     """Declare workflow state behavior for an Annotated BaseModel field."""
-    return StateFieldMetadata(reducer=reducer, trace=trace)
+    return StateFieldMetadata(reducer=_reducer_ref_from(reducer), trace=trace)
 
 
 def schema_ref_from(value: SchemaLike) -> SchemaRef:
@@ -51,9 +55,7 @@ def state_schema_from(value: StateSchemaLike) -> StateSchema:
     fields = {
         path: StateField(
             type=_state_field_type(property_schema),
-            reducer=ReducerRef(
-                name=metadata_by_name.get(path, StateFieldMetadata()).reducer
-            ),
+            reducer=metadata_by_name.get(path, StateFieldMetadata()).reducer,
             trace=metadata_by_name.get(path, StateFieldMetadata()).trace,
             default=_state_field_default(value, path, property_schema),
         )
@@ -62,6 +64,14 @@ def state_schema_from(value: StateSchemaLike) -> StateSchema:
     return StateSchema(fields=fields)
 
 
+def _reducer_ref_from(value: ReducerLike) -> ReducerRef:
+    if isinstance(value, ReducerRef):
+        return value
+    if isinstance(value, str):
+        return ReducerRef(name=value)
+    return ReducerRef.model_validate(value)
+
+
 def _state_metadata_by_name(value: object) -> dict[str, StateFieldMetadata]:
     if not isinstance(value, type) or not issubclass(value, BaseModel):
         return {}

src/wf_core/conditions.py

@@ -46,8 +46,12 @@ def eval_condition(
             return left != right
         if condition.op == "gt":
             return left > right
+        if condition.op == "ge":
+            return left >= right
         if condition.op == "lt":
             return left < right
+        if condition.op == "le":
+            return left <= right
     raise WorkflowExecutionError(f"unsupported condition operator {condition.op!r}")
 
 

src/wf_core/models/conditions.py

@@ -45,7 +45,7 @@ class VariadicCondition(BaseModel):
 class BinaryCondition(BaseModel):
     """Condition that compares two operands."""
 
-    op: Literal["eq", "ne", "gt", "lt"]
+    op: Literal["eq", "ne", "gt", "ge", "lt", "le"]
     left: PathOperand | LiteralOperand
     right: PathOperand | LiteralOperand
 

src/wf_core/runtime/engine.py

@@ -6,6 +6,7 @@ from typing import Any
 from wf_core.models.workflow import Workflow
 from wf_core.runtime.ops.flow import finalize_run
 from wf_core.runtime.ops.frames import collapse_completed_frames
+from wf_core.runtime.ops.merges import ReducerDefinition
 from wf_core.runtime.ops.nodes import AsyncNodeHandler, NodeHandler
 from wf_core.runtime.ops.runs import create_run_state
 from wf_core.run_state import RunState, RunStatus
@@ -19,13 +20,15 @@ def execute_workflow(
     workflow: Workflow,
     workflow_input: dict[str, Any],
     registry: Mapping[str, NodeHandler],
+    *,
+    reducers: Mapping[str, ReducerDefinition] | None = None,
 ) -> RunState:
     """Create a run and execute a workflow synchronously until it stops."""
     run = create_run_state(workflow, workflow_input)
 
     try:
         run = prepare_new_run(workflow, workflow_input)
-        return resume_workflow(workflow, run, registry)
+        return resume_workflow(workflow, run, registry, reducers=reducers)
     except Exception as exc:
         run.status = RunStatus.FAILED
         run.error = str(exc)
@@ -36,13 +39,15 @@ async def execute_workflow_async(
     workflow: Workflow,
     workflow_input: dict[str, Any],
     registry: Mapping[str, AsyncNodeHandler],
+    *,
+    reducers: Mapping[str, ReducerDefinition] | None = None,
 ) -> RunState:
     """Create a run and execute a workflow asynchronously until it stops."""
     run = create_run_state(workflow, workflow_input)
 
     try:
         run = prepare_new_run(workflow, workflow_input)
-        return await resume_workflow_async(workflow, run, registry)
+        return await resume_workflow_async(workflow, run, registry, reducers=reducers)
     except Exception as exc:
         run.status = RunStatus.FAILED
         run.error = str(exc)
@@ -56,6 +61,7 @@ def resume_workflow(
     *,
     resume_payload: dict[str, Any] | None = None,
     resume_outcome: str = "submitted",
+    reducers: Mapping[str, ReducerDefinition] | None = None,
 ) -> RunState:
     """Resume a synchronous run from its current state."""
     index = prepare_resume(
@@ -63,6 +69,7 @@ def resume_workflow(
         run,
         resume_payload=resume_payload,
         resume_outcome=resume_outcome,
+        reducers=reducers,
     )
     if index is None:
         if run.current_node_id == END:
@@ -78,6 +85,7 @@ def resume_workflow(
             run,
             registry,
             index=index,
+            reducers=reducers,
         )
         if run.status == RunStatus.INTERRUPTED:
             return run
@@ -92,6 +100,7 @@ async def resume_workflow_async(
     *,
     resume_payload: dict[str, Any] | None = None,
     resume_outcome: str = "submitted",
+    reducers: Mapping[str, ReducerDefinition] | None = None,
 ) -> RunState:
     """Resume an async run from its current state."""
     index = prepare_resume(
@@ -99,6 +108,7 @@ async def resume_workflow_async(
         run,
         resume_payload=resume_payload,
         resume_outcome=resume_outcome,
+        reducers=reducers,
     )
     if index is None:
         if run.current_node_id == END:
@@ -114,6 +124,7 @@ async def resume_workflow_async(
             run,
             registry,
             index=index,
+            reducers=reducers,
         )
         if run.status == RunStatus.INTERRUPTED:
             return run

src/wf_core/runtime/ops/interrupts.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from collections.abc import Mapping
 from typing import Any
 
 from wf_core.conditions import safe_resolve_path
@@ -9,6 +10,7 @@ from wf_core.models.workflow import Workflow
 from wf_core.run_state import InterruptRequest, RunState, StepExecutionResult
 from wf_core.runtime.ops.flow import advance_frame, append_step_result_trace
 from wf_core.runtime.ops.index import WorkflowIndex
+from wf_core.runtime.ops.merges import ReducerDefinition
 from wf_core.runtime.ops.state import apply_mapped_state
 
 
@@ -45,6 +47,7 @@ def resume_interrupt(
     index: WorkflowIndex,
     resume_payload: dict[str, Any],
     resume_outcome: str,
+    reducers: Mapping[str, ReducerDefinition] | None = None,
 ) -> None:
     if run.current_frame_id is None:
         raise WorkflowExecutionError("interrupted run has no current frame")
@@ -69,6 +72,7 @@ def resume_interrupt(
         resume_payload,
         step.out_map,
         run.state,
+        reducers=reducers,
         missing_field_message="interrupt resume payload is missing required field {field}",
     )
     next_node_id = index.next_node_id(frame.node_id, resume_outcome)

src/wf_core/runtime/ops/merges.py

@@ -149,10 +149,17 @@ def apply_reducer(
     current_value: Any,
     incoming_value: Any,
     destination_path: str,
-    reducers: Mapping[str, ReducerDefinition] = DEFAULT_REDUCER_DEFINITIONS,
+    reducers: Mapping[str, ReducerDefinition] | None = None,
 ) -> Any:
-    """Apply one named pure reducer to a state write."""
-    definition = reducers.get(reducer.name)
+    """Apply one named pure reducer to a state write.
+
+    Injected reducer definitions are additive over the built-ins so authoring
+    tests and local packages can provide custom reducers without re-registering
+    every `wf.std.*` reducer.
+    """
+    definition = None if reducers is None else reducers.get(reducer.name)
+    if definition is None:
+        definition = DEFAULT_REDUCER_DEFINITIONS.get(reducer.name)
     if definition is None:
         raise WorkflowExecutionError(f"unknown reducer {reducer.name!r}")
     return definition.apply(

src/wf_core/runtime/ops/nodes.py

@@ -12,6 +12,7 @@ from wf_core.models.steps import NodeUse
 from wf_core.models.workflow import Workflow
 from wf_core.run_state import RunState, RuntimeContext, StepExecutionResult
 from wf_core.runtime.ops.frames import frame_context_values
+from wf_core.runtime.ops.merges import ReducerDefinition
 from wf_core.runtime.ops.schemas import validate_payload_against_schema
 from wf_core.runtime.ops.state import apply_output_map
 
@@ -65,6 +66,7 @@ def _finalize_node_execution(
     node_def: NodeDef,
     resolved_input: dict[str, Any],
     raw_result: NodeResult | dict[str, Any],
+    reducers: Mapping[str, ReducerDefinition] | None = None,
 ) -> StepExecutionResult:
     result = coerce_node_result(raw_result)
 
@@ -76,7 +78,13 @@ def _finalize_node_execution(
     validate_payload_against_schema(
         node_def.output_schema, result.output, f"node output for {node.id}"
     )
-    state_changes = apply_output_map(workflow, node, result.output, run.state)
+    state_changes = apply_output_map(
+        workflow,
+        node,
+        result.output,
+        run.state,
+        reducers=reducers,
+    )
     return StepExecutionResult(
         outcome=result.outcome,
         resolved_input=resolved_input,
@@ -91,6 +99,7 @@ def execute_node_use(
     node: NodeUse,
     node_def: NodeDef,
     registry: Mapping[str, NodeHandler],
+    reducers: Mapping[str, ReducerDefinition] | None = None,
 ) -> StepExecutionResult:
     handler = registry.get(node.node)
     if handler is None:
@@ -112,6 +121,7 @@ def execute_node_use(
         node_def=node_def,
         resolved_input=resolved_input,
         raw_result=raw_result,
+        reducers=reducers,
     )
 
 
@@ -121,6 +131,7 @@ async def execute_node_use_async(
     node: NodeUse,
     node_def: NodeDef,
     registry: Mapping[str, AsyncNodeHandler],
+    reducers: Mapping[str, ReducerDefinition] | None = None,
 ) -> StepExecutionResult:
     handler = registry.get(node.node)
     if handler is None:
@@ -146,6 +157,7 @@ async def execute_node_use_async(
         node_def=node_def,
         resolved_input=resolved_input,
         raw_result=cast(NodeResult | dict[str, Any], raw_result),
+        reducers=reducers,
     )
 
 

src/wf_core/runtime/ops/state.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from collections.abc import Mapping
 from typing import Any
 
 from wf_core.errors import WorkflowExecutionError
@@ -13,7 +14,7 @@ from wf_core.paths import (
     set_nested_value,
     split_graph_path,
 )
-from wf_core.runtime.ops.merges import apply_reducer
+from wf_core.runtime.ops.merges import ReducerDefinition, apply_reducer
 
 
 def apply_output_map(
@@ -21,12 +22,14 @@ def apply_output_map(
     node: NodeUse,
     node_output: dict[str, Any],
     state: dict[str, Any],
+    reducers: Mapping[str, ReducerDefinition] | None = None,
 ) -> dict[str, Any]:
     return apply_mapped_state(
         workflow,
         node_output,
         node.out_map,
         state,
+        reducers=reducers,
         missing_field_message=f"node {node.id!r} did not return required mapped field {{field}}",
     )
 
@@ -37,6 +40,7 @@ def apply_mapped_state(
     mapping: dict[str, str],
     state: dict[str, Any],
     *,
+    reducers: Mapping[str, ReducerDefinition] | None = None,
     missing_field_message: str,
 ) -> dict[str, Any]:
     if has_overlapping_paths(mapping.values()):
@@ -55,12 +59,23 @@ def apply_mapped_state(
         patch[destination_path] = value
 
     for destination_path, value in patch.items():
-        write_state_value(workflow, state, destination_path, value)
+        write_state_value(
+            workflow,
+            state,
+            destination_path,
+            value,
+            reducers=reducers,
+        )
     return dict(patch)
 
 
 def write_state_value(
-    workflow: Workflow, state: dict[str, Any], destination_path: str, value: Any
+    workflow: Workflow,
+    state: dict[str, Any],
+    destination_path: str,
+    value: Any,
+    *,
+    reducers: Mapping[str, ReducerDefinition] | None = None,
 ) -> None:
     try:
         root, parts = split_graph_path(destination_path)
@@ -74,7 +89,9 @@ def write_state_value(
 
     declared_path = ".".join(parts)
     declared_field = workflow.state_schema.fields.get(declared_path)
-    reducer = declared_field.reducer if declared_field else ReducerRef(name="wf.std.replace")
+    reducer = (
+        declared_field.reducer if declared_field else ReducerRef(name="wf.std.replace")
+    )
     key_path = parts
     current_value = get_nested_value(state, key_path)
     merged_value = apply_reducer(
@@ -82,6 +99,7 @@ def write_state_value(
         current_value=current_value,
         incoming_value=value,
         destination_path=destination_path,
+        reducers=reducers,
     )
     safe_set_nested_value(state, key_path, merged_value)
 

src/wf_core/runtime/preparation.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from collections.abc import Mapping
 from typing import Any
 
 from wf_core.errors import WorkflowExecutionError
@@ -7,6 +8,7 @@ from wf_core.models.workflow import Workflow
 from wf_core.runtime.ops.frames import collapse_completed_frames
 from wf_core.runtime.ops.index import WorkflowIndex, build_workflow_index
 from wf_core.runtime.ops.interrupts import resume_interrupt
+from wf_core.runtime.ops.merges import ReducerDefinition
 from wf_core.runtime.ops.runs import create_run_state
 from wf_core.runtime.ops.schemas import validate_payload_against_schema
 from wf_core.run_state import FrameStatus, RunState, RunStatus
@@ -29,6 +31,7 @@ def prepare_resume(
     *,
     resume_payload: dict[str, Any] | None,
     resume_outcome: str,
+    reducers: Mapping[str, ReducerDefinition] | None = None,
 ) -> WorkflowIndex | None:
     """Validate and normalize a run state before resume execution."""
     if run.workflow_name != workflow.name:
@@ -57,6 +60,7 @@ def prepare_resume(
             index=index,
             resume_payload=resume_payload,
             resume_outcome=resume_outcome,
+            reducers=reducers,
         )
         collapse_completed_frames(run)
         if run.current_node_id == END:

src/wf_core/runtime/step.py

@@ -20,6 +20,7 @@ from wf_core.runtime.ops.handlers import (
     handle_join_step,
 )
 from wf_core.runtime.ops.index import WorkflowIndex
+from wf_core.runtime.ops.merges import ReducerDefinition
 from wf_core.runtime.ops.nodes import (
     AsyncNodeHandler,
     NodeHandler,
@@ -67,6 +68,7 @@ def step_workflow(
     registry: Mapping[str, NodeHandler],
     *,
     index: WorkflowIndex | None = None,
+    reducers: Mapping[str, ReducerDefinition] | None = None,
 ) -> RunState:
     """Execute at most one synchronous workflow step."""
     prepared = prepare_step(workflow, run, index)
@@ -77,7 +79,14 @@ def step_workflow(
 
     if isinstance(step, NodeUse):
         node_def = index.node_defs[step.node]
-        step_result = execute_node_use(workflow, run, step, node_def, registry)
+        step_result = execute_node_use(
+            workflow,
+            run,
+            step,
+            node_def,
+            registry,
+            reducers=reducers,
+        )
     elif isinstance(step, ConditionNode):
         step_result = handle_condition_step(run, step)
     elif isinstance(step, JoinNode):
@@ -108,6 +117,7 @@ async def step_workflow_async(
     registry: Mapping[str, AsyncNodeHandler],
     *,
     index: WorkflowIndex | None = None,
+    reducers: Mapping[str, ReducerDefinition] | None = None,
 ) -> RunState:
     """Execute at most one async workflow step."""
     prepared = prepare_step(workflow, run, index)
@@ -119,7 +129,12 @@ async def step_workflow_async(
     if isinstance(step, NodeUse):
         node_def = index.node_defs[step.node]
         step_result = await execute_node_use_async(
-            workflow, run, step, node_def, registry
+            workflow,
+            run,
+            step,
+            node_def,
+            registry,
+            reducers=reducers,
         )
     elif isinstance(step, ConditionNode):
         step_result = handle_condition_step(run, step)

tests/authoring/test_builder_branch.py

@@ -96,3 +96,65 @@ def test_builder_route_expands_state_value_cases_into_condition_chain() -> None:
     assert targets["left"] is left
     assert targets["right"] is right
     assert targets["default"] is fallback
+
+
+def test_builder_route_can_name_generated_value_conditions() -> None:
+    builder = WorkflowBuilder(
+        name="route_named_demo",
+        input_schema=AutoBindInput,
+        state_schema=AutoBindState,
+        output_schema=AutoBindOutput,
+    )
+    left = builder.use(auto_bind_node, id="left")
+    right = builder.use(auto_bind_node, id="right")
+
+    builder.route(state("value"), {"left": left, "right": right}, id="by_value")
+
+    assert [node.id for node in builder.nodes if node.type == "condition"] == [
+        "by_value",
+        "by_value_2",
+    ]
+    assert [(edge.from_, edge.outcome, edge.to) for edge in builder.edges] == [
+        ("by_value", "true", "left"),
+        ("by_value", "false", "by_value_2"),
+        ("by_value_2", "true", "right"),
+        ("by_value_2", "false", "authoring_runtime_error"),
+    ]
+
+
+def test_builder_route_accepts_boolean_condition_expression() -> None:
+    builder = WorkflowBuilder(
+        name="route_condition_demo",
+        input_schema=AutoBindInput,
+        state_schema=AutoBindState,
+        output_schema=AutoBindOutput,
+    )
+    left = builder.use(auto_bind_node, id="left")
+    right = builder.use(auto_bind_node, id="right")
+
+    targets = builder.route(state("count").ge(1), {True: left, False: right})
+
+    assert [(edge.from_, edge.outcome, edge.to) for edge in builder.edges] == [
+        ("condition", "true", "left"),
+        ("condition", "false", "right"),
+    ]
+    assert targets[True] is left
+    assert targets[False] is right
+
+
+def test_builder_route_can_name_boolean_condition_expression() -> None:
+    builder = WorkflowBuilder(
+        name="route_named_condition_demo",
+        input_schema=AutoBindInput,
+        state_schema=AutoBindState,
+        output_schema=AutoBindOutput,
+    )
+    left = builder.use(auto_bind_node, id="left")
+    right = builder.use(auto_bind_node, id="right")
+
+    builder.route(state("count").ge(1), {True: left, False: right}, id="count_ge_1")
+
+    assert [(edge.from_, edge.outcome, edge.to) for edge in builder.edges] == [
+        ("count_ge_1", "true", "left"),
+        ("count_ge_1", "false", "right"),
+    ]

tests/authoring/test_conditions.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
-from wf_authoring import exists, expr, state_path
+from wf_authoring import exists, expr, not_, state, state_path
+from wf_core.conditions import eval_condition
 
 
 def test_condition_dsl_compiles_to_core_condition() -> None:
@@ -22,3 +23,18 @@ def test_condition_dsl_compiles_to_core_condition() -> None:
             },
         ],
     }
+
+
+def test_condition_dsl_supports_not_ge_and_ne() -> None:
+    condition = not_(
+        state("score").ge(10) & state("score").le(20) & state("status").ne("blocked")
+    )
+    compiled = condition.to_condition()
+
+    assert compiled.model_dump()["op"] == "not"
+    assert eval_condition(
+        compiled,
+        state={"score": 7, "status": "ready"},
+        workflow_input={},
+        context_data=None,
+    )

tests/authoring/test_reducers.py

@@ -1,8 +1,17 @@
 from __future__ import annotations
 
+from typing import Annotated
+
 from pydantic import BaseModel, Field
 
-from wf_authoring import ReducerCatalog, reducer
+from wf_authoring import (
+    NodeReturn,
+    ReducerCatalog,
+    WorkflowBuilder,
+    node,
+    reducer,
+    state_field,
+)
 from wf_core import ReducerRef
 
 
@@ -10,6 +19,22 @@ class ModuloConfig(BaseModel):
     modulus: int = Field(gt=0)
 
 
+class CounterState(BaseModel):
+    total: Annotated[int, state_field(reducer="wf.std.add")] = 0
+
+
+class EmptyInput(BaseModel):
+    pass
+
+
+class ModuloCounterState(BaseModel):
+    total: int = 0
+
+
+class CounterOutput(BaseModel):
+    total: int
+
+
 @reducer(name="wf.std.add")
 def add(current: int | None, incoming: int) -> int:
     """Add incoming values into integer state."""
@@ -61,3 +86,43 @@ def test_reducer_catalog_exposes_definitions_and_specs() -> None:
     assert set(catalog.definitions) == {"wf.std.add", "wf.std.modulo_add"}
     assert catalog.specs["wf.std.add"].name == "wf.std.add"
     assert catalog.specs["wf.std.modulo_add"].name == "wf.std.modulo_add"
+
+
+def test_builder_executes_with_custom_reducer_catalog() -> None:
+    @node
+    def emit(_: EmptyInput) -> NodeReturn[CounterOutput]:
+        return NodeReturn(outcome="ok", output=CounterOutput(total=4))
+
+    builder = WorkflowBuilder(
+        name="custom_reducer",
+        input_schema=EmptyInput,
+        state_schema=CounterState,
+        output_schema=CounterOutput,
+        reducers=ReducerCatalog.from_reducers(add),
+    )
+    step = builder.use(
+        emit,
+        in_map={},
+        out_map={"total": "state.total"},
+    )
+    builder.set_entry_point(step)
+    builder.connect(step, "ok", "__end__")
+
+    run = builder.execute({})
+
+    assert run.state["total"] == 4
+
+
+def test_state_field_accepts_configured_reducer_reference() -> None:
+    class State(BaseModel):
+        total: Annotated[
+            int,
+            state_field(
+                reducer={"name": "wf.std.modulo_add", "config": {"modulus": 10}}
+            ),
+        ] = 0
+
+    field = State.model_fields["total"].metadata[0]
+
+    assert field.reducer.name == "wf.std.modulo_add"
+    assert field.reducer.config["modulus"] == 10

tests/rewrite/models.py

@@ -17,8 +17,8 @@ class SophisticatedRates(TypedDict):
 
 
 class SophisticatedCounter(TypedDict):
-    c_10: int  # how do i convey "add" sublevel? can we have plugins for this? should we cover this; since langgraph doesnt.
-    c_80: int
+    c_10: Annotated[int, state_field(reducer="wf.std.add")]  # how do i convey "add" sublevel? can we have plugins for this? should we cover this; since langgraph doesnt.
+    c_80: Annotated[int, state_field(reducer="wf.std.add")]
 
 
 class Counters(BaseModel):
@@ -31,7 +31,7 @@ class Counters(BaseModel):
 
 
 class Countdown(BaseModel):
-    countdown: int  # add!
+    countdown: Annotated[int, state_field(reducer="wf.std.add")]  # add!
 
 
 # has to migrate from typeddict for what? for nothing.

tests/rewrite/workflow.py

@@ -43,11 +43,19 @@ gacha.connect("keep_rolling", "true", "tick")
 gacha.connect("keep_rolling", "false", END)
 
 gacha.connect("tick", "ok", "counter_up")
-gacha.use(rate_booster, id="rate_booster")
+# gacha.use(rate_booster, id="rate_booster")
 
 gacha.connect("counter_up", "ok", "rate_booster")
-gacha.connect("rate_booster", "0", rate_same)
-gacha.connect("rate_booster", "65", rate_up)
+# gacha.connect("rate_booster", "0", rate_same)
+# gacha.connect("rate_booster", "65", rate_up)
+gacha.route(
+    state("counter.c_80").ge(65),
+    {
+        True: rate_up,
+        False: rate_same,
+    },
+    id = "rate_booster"
+)
 gacha.use(pre_roll_router, id="router")
 
 gacha.connect("rate_up", "ok", "router")