lda.chat/lda-wf
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

split .route into + add multiple distinct helpers

Browse Source
This commit is contained in:
lda
2026-05-19 01:16:50 +07:00 Verified
parent 8d9796dd04
commit 8210ae3b85
5 changed files with 347 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/superpowers/specs/2026-05-19-authoring-control-flow-design.md
+200
View File
@@ -0,0 +1,200 @@
# Authoring Control Flow Design
## Goal
Make `wf_authoring` control flow precise enough that future sugar layers can
delegate to it instead of reimplementing graph semantics.
The current problem is `route()`: it accepts both `PathExpr` and `Expr`, then
dispatches to two different behaviors. Both "route" something, but the caller
must understand different case semantics depending on the argument type.
That is too blurry for the core authoring API.
## Design Rule
Each public control-flow method should name one decision mechanism.
| Method | Decision source | Meaning |
| --- | --- | --- |
| `branch` | a node's declared outcome | wire outcome labels to targets |
| `match` | one graph value | compare that value against equality cases |
| `when` | one boolean condition | route through `true` / `false` |
| `choose` | ordered boolean conditions | first true condition wins |
| future `handle` | several source/outcome pairs | send shared outcomes to one target |
Future fluent builders or operator sugar must call these methods rather than
constructing edges/conditions independently.
## Keep `branch`
`branch()` is already good:
```python
g.branch(tool, {
"ok": next_step,
"error": fail_step,
})
```
It has one clear meaning:
```text
route an existing node outcome
```
It should remain named `branch`. Alternatives such as `on_outcome` are more
literal but worse to use, and the current method does not suffer from the
semantic overloading that `route()` does.
## Replace `route`
### `match`
```python
g.match(
state("status"),
{
"done": finish,
"retry": retry,
},
default=fail,
)
```
Meaning:
```text
if state.status == "done": finish
elif state.status == "retry": retry
else: fail
```
This is the existing `PathExpr` branch of `route()`, renamed to say what it
actually does.
### `when`
```python
g.when(
state("count").ge(1),
then=positive,
otherwise=zero,
)
```
Meaning:
```text
if state.count >= 1: positive
else: zero
```
This is the existing `Expr` branch of `route()`, made explicit and easier to
call correctly.
### `choose`
```python
g.choose(
(state("x").gt(10), big),
(state("y").exists(), has_y),
default=fail,
)
```
Meaning:
```text
if state.x > 10: big
elif state.y exists: has_y
else: fail
```
This is an ordered predicate chain. It should lower through the same condition
construction/connect machinery as `when`, repeated for each clause.
`choose` is intentionally one call. Multiple-call fluent syntax can be built
later on top of it if it proves useful.
## Lowering
These methods may generate core condition nodes, but callers should not need to
know the exact node construction details to choose the right API.
Canonical lowering:
- `branch`
- no new condition nodes
- wires declared outcome strings from one source
- `match`
- ordered equality-check condition chain
- one generated condition per case
- `when`
- one condition node
- `true` and `false` edges
- `choose`
- ordered condition chain
- one generated condition per clause
Trace behavior should document that `match` and `choose` expand to generated
condition nodes.
## Outcome Names
Outcome names are strings at the core wire level, but Python authoring should
eventually avoid handwritten strings when a `NodeSpec` already declares them.
Future improvement:
```python
tool.outcomes.ok
tool.outcomes.error
```
derived from `NodeSpec.outcomes`, not duplicated constants that can drift from
the contract.
This is orthogonal to the control-flow rename, but it belongs in the same
authoring quality bar.
## Migration
`route()` should become deprecated compatibility sugar for one release window,
then be removed.
Recommended behavior during compatibility:
- mark `route()` with `@deprecated` so IDEs surface the replacement path
- `route(PathExpr, cases, ...)`
- warns and forwards to `match(...)`
- `route(Expr, {True: a, False: b}, ...)`
- warns and forwards to `when(...)`
The public docs should prefer only:
- `branch`
- `match`
- `when`
- `choose`
## Not In This Pass
- reverse-branch/shared handlers (`handle`)
- fluent/cursor builder APIs
- operator overloading
- graph-as-node/subgraph support
- JSON draft `match` / `when` / `choose` shapes
Those later layers should depend on this API once it is stable.
## Testing
Tests should prove:
1. `branch` still wires outcome labels only
2. `match` reproduces current value-route behavior
3. `when` reproduces current boolean-route behavior
4. `choose` lowers ordered predicates correctly
5. `route` emits deprecation warnings while preserving old behavior
6. later sugar can delegate to these without needing private builder internals
src/wf_authoring/builder/core.py
+83 -41
View File
@@ -4,6 +4,7 @@ from collections.abc import Mapping
from dataclasses import dataclass, field
from typing import Any, Literal, cast
import warnings
from warnings import deprecated
from wf_authoring.ops.values import runtime_error
from wf_core import (
@@ -259,35 +260,15 @@ class WorkflowBuilder:
resolved_targets[outcome] = cast(StepRef, resolved)
return resolved_targets
def route(
def match(
self,
value: PathExpr | Expr,
value: PathExpr,
cases: Mapping[object, BranchRef],
*,
id: str | None = None,
default: BranchRef = runtime_error,
) -> RouteRef:
"""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,
) -> RouteRef:
"""Expand value cases into an ordered chain of equality checks."""
"""Match one graph value against ordered equality cases."""
resolved_targets: dict[object, StepRef] = {}
conditions: list[ConditionNode] = []
default_target = self.use(default) if is_node_spec(default) else default
@@ -306,7 +287,7 @@ class WorkflowBuilder:
previous_condition = condition
resolved_targets[case_value] = cast(StepRef, resolved)
if previous_condition is None:
raise ValueError("WorkflowBuilder.route requires at least one case")
raise ValueError("WorkflowBuilder.match requires at least one case")
self.connect(previous_condition, "false", cast(StepRef, default_target))
resolved_targets["default"] = cast(StepRef, default_target)
return RouteRef(
@@ -315,34 +296,95 @@ class WorkflowBuilder:
targets=resolved_targets,
)
def _route_condition(
def when(
self,
condition: Expr,
cases: Mapping[object, BranchRef],
*,
id: str | None,
default: BranchRef,
then: BranchRef,
otherwise: BranchRef = runtime_error,
id: str | None = None,
) -> RouteRef:
"""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")
"""Route one boolean condition through true and false targets."""
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)
resolved_then = self.use(then) if is_node_spec(then) else then
resolved_otherwise = (
self.use(otherwise) if is_node_spec(otherwise) else otherwise
)
self.connect(condition_node, "true", cast(StepRef, resolved_then))
self.connect(condition_node, "false", cast(StepRef, resolved_otherwise))
return RouteRef(
entry=condition_node,
conditions=(condition_node,),
targets={
True: cast(StepRef, resolved_then),
False: cast(StepRef, resolved_otherwise),
},
)
def choose(
self,
*clauses: tuple[Expr, BranchRef],
default: BranchRef = runtime_error,
id: str | None = None,
) -> RouteRef:
"""Route to the first target whose ordered condition is true."""
if not clauses:
raise ValueError("WorkflowBuilder.choose requires at least one clause")
conditions: list[ConditionNode] = []
resolved_targets: dict[object, StepRef] = {}
previous_condition: ConditionNode | None = None
condition_base = id or "condition"
for index, (condition_expr, target) in enumerate(clauses):
condition = self.condition(
id=self._next_step_id(condition_base),
check=condition_expr,
)
conditions.append(condition)
resolved = self.use(target) if is_node_spec(target) else target
if previous_condition is not None:
self.connect(previous_condition, "false", condition)
self.connect(condition, "true", cast(StepRef, resolved))
previous_condition = condition
resolved_targets[index] = cast(StepRef, resolved)
default_target = self.use(default) if is_node_spec(default) else default
self.connect(cast(ConditionNode, previous_condition), "false", cast(StepRef, default_target))
resolved_targets["default"] = cast(StepRef, default_target)
return RouteRef(
entry=conditions[0],
conditions=tuple(conditions),
targets=resolved_targets,
)
@deprecated("use match(...) or when(...) instead")
def route(
self,
value: PathExpr | Expr,
cases: Mapping[object, BranchRef],
*,
id: str | None = None,
default: BranchRef = runtime_error,
) -> RouteRef:
"""Deprecated compatibility shim for the old overloaded route API."""
warnings.warn(
"WorkflowBuilder.route is deprecated; use match(...) or when(...)",
DeprecationWarning,
stacklevel=2,
)
if isinstance(value, Expr):
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")
return self.when(
value,
then=cases.get(True, default),
otherwise=cases.get(False, default),
id=id,
)
return self.match(value, cases, id=id, default=default)
def compile(self) -> Workflow:
if self.start is None:
raise WorkflowExecutionError(
src/wf_authoring/builder/refs.py
+2 -2
View File
@@ -15,14 +15,14 @@ BranchRef: TypeAlias = StepRef | NodeSpec[Any, Any]
@dataclass(frozen=True, slots=True)
class RouteRef:
"""Reference bundle returned by route() for generated condition nodes."""
"""Reference bundle returned by control-flow helpers with generated conditions."""
entry: ConditionNode
conditions: tuple[ConditionNode, ...]
targets: dict[object, StepRef]
def __getitem__(self, key: object) -> StepRef:
"""Keep route["case"] ergonomic while exposing generated conditions."""
"""Keep helper result lookup ergonomic while exposing generated conditions."""
return self.targets[key]
tests/authoring/test_builder_branch.py
+58 -8
View File
@@ -66,7 +66,7 @@ def test_builder_branch_warns_on_empty_branch_map() -> None:
assert targets == {}
def test_builder_route_expands_state_value_cases_into_condition_chain() -> None:
def test_builder_match_expands_state_value_cases_into_condition_chain() -> None:
builder = WorkflowBuilder(
name="route_demo",
input_schema=AutoBindInput,
@@ -77,7 +77,7 @@ def test_builder_route_expands_state_value_cases_into_condition_chain() -> None:
right = builder.use(auto_bind_node, id="right")
fallback = builder.use(auto_bind_node, id="fallback")
targets = builder.route(
targets = builder.match(
state("value"),
{"left": left, "right": right},
default=fallback,
@@ -99,7 +99,7 @@ def test_builder_route_expands_state_value_cases_into_condition_chain() -> None:
assert targets["default"] is fallback
def test_builder_route_can_name_generated_value_conditions() -> None:
def test_builder_match_can_name_generated_value_conditions() -> None:
builder = WorkflowBuilder(
name="route_named_demo",
input_schema=AutoBindInput,
@@ -109,7 +109,7 @@ def test_builder_route_can_name_generated_value_conditions() -> None:
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")
builder.match(state("value"), {"left": left, "right": right}, id="by_value")
assert [node.id for node in builder.nodes if node.type == "condition"] == [
"by_value",
@@ -123,7 +123,7 @@ def test_builder_route_can_name_generated_value_conditions() -> None:
]
def test_builder_route_accepts_boolean_condition_expression() -> None:
def test_builder_when_routes_one_boolean_condition_expression() -> None:
builder = WorkflowBuilder(
name="route_condition_demo",
input_schema=AutoBindInput,
@@ -133,7 +133,7 @@ def test_builder_route_accepts_boolean_condition_expression() -> None:
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})
targets = builder.when(state("count").ge(1), then=left, otherwise=right)
assert targets.entry.id == "condition"
assert [node.id for node in targets.conditions] == ["condition"]
@@ -145,7 +145,7 @@ def test_builder_route_accepts_boolean_condition_expression() -> None:
assert targets[False] is right
def test_builder_route_can_name_boolean_condition_expression() -> None:
def test_builder_when_can_name_boolean_condition_expression() -> None:
builder = WorkflowBuilder(
name="route_named_condition_demo",
input_schema=AutoBindInput,
@@ -155,9 +155,59 @@ def test_builder_route_can_name_boolean_condition_expression() -> None:
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")
builder.when(
state("count").ge(1),
then=left,
otherwise=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"),
]
def test_builder_choose_routes_first_matching_condition_chain() -> None:
builder = WorkflowBuilder(
name="choose_demo",
input_schema=AutoBindInput,
state_schema=AutoBindState,
output_schema=AutoBindOutput,
)
large = builder.use(auto_bind_node, id="large")
positive = builder.use(auto_bind_node, id="positive")
fallback = builder.use(auto_bind_node, id="fallback")
targets = builder.choose(
(state("count").ge(10), large),
(state("count").ge(1), positive),
default=fallback,
id="count_choice",
)
assert [node.id for node in targets.conditions] == [
"count_choice",
"count_choice_2",
]
assert [(edge.from_, edge.outcome, edge.to) for edge in builder.edges] == [
("count_choice", "true", "large"),
("count_choice", "false", "count_choice_2"),
("count_choice_2", "true", "positive"),
("count_choice_2", "false", "fallback"),
]
def test_builder_route_warns_and_forwards_to_match() -> None:
builder = WorkflowBuilder(
name="route_compat_demo",
input_schema=AutoBindInput,
state_schema=AutoBindState,
output_schema=AutoBindOutput,
)
left = builder.use(auto_bind_node, id="left")
with pytest.warns(DeprecationWarning, match="match"):
builder.route(state("value"), {"left": left})
assert builder.edges[0].to == "left"
tests/rewrite/workflow.py
+4 -9
View File
@@ -62,17 +62,12 @@ 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.connect("counter_up", "ok", "rate_booster")
# gacha.connect("rate_booster", "0", rate_same)
# gacha.connect("rate_booster", "65", rate_up)
rate_route = gacha.route(
rate_route = gacha.when( # condition + branch in one. its so good.
state("counter.c_80").ge(65),
{
True: rate_up,
False: rate_same,
},
then=rate_up,
otherwise=rate_same,
id="rate_booster",
)
@@ -80,7 +75,7 @@ gacha.use(pre_roll_router, id="router")
gacha.connect("rate_up", "ok", "router")
gacha.connect("rate_same", "ok", "router")
preroll_routes = gacha.branch(
preroll_routes = gacha.branch( # `connect`s in one branch. also so good.
"router",
{
"240": "r_gs",