diff --git a/CHANGELOG.md b/CHANGELOG.md index 41a1188..7c7a30c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,65 @@ Versioning: [Semantic Versioning](https://semver.org/spec/v2.0.0.html). ## [Unreleased] +## [0.3.0] — 2026-10-01 + +### ⛔ Breaking — `check()` is renamed to `coherence_check()` + +**Step-by-step upgrade: [`docs/migration-0.3.md`](docs/migration-0.3.md).** One line: + +```python +spec.check(strategy) # 0.2.0 +spec.coherence_check(strategy) # 0.3.0 +``` + +No alias. A missed call site is an `AttributeError` at the call, not a silent change of +behaviour — same choice 0.2.0 made when `NodeSpec(...)` became a `TypeError`. + +**Why.** `check()` did not say what it checks, and the type it returns says `Coherence` — a word +that appeared nowhere else in the API. One concept was wearing two names, which is the thing the +house naming rule exists to prevent. `coherence_check()` grounds it. + +`design_check()` was considered and rejected: `spec` *is* the design, so `spec.design_check()` +restates its own receiver. + +**Unchanged:** the eleven `check_*` functions, and the `check` field on a finding — both name an +individual check, which is what they still are. + +### Added — `coherence_check()` returns `CoherenceFinding`, not a bare `str` + +**Backward compatible. No call site needs editing** — `CoherenceFinding` is a `str` subclass, so +`"x" in f`, `f.startswith(...)`, `"\n".join(findings)`, `f == "the message"`, sorting, hashing +and `repr()` in a printed list all behave exactly as before. Verified byte-for-byte across all 64 +findings the test designs produce: nothing in the text moved. + +```python +f = spec.coherence_check(strategy)[0] +f.check # 'check_bindings' — the function that produced it +f.about # 'compose' — a node name; 'source->target' for an edge; '' for the whole design +f.blocking # True — False only for a `NOT CHECKED — …` stated gap + +from workflow_workbench import blocking +blocking(findings) # the filter `render()` uses; replaces startswith("NOT CHECKED") +``` + +**Why.** The findings were sentences, so the structure a caller needs was encoded in the prose. +`[f for f in findings if not f.startswith("NOT CHECKED")]` was load-bearing control flow in three +production call sites here and in both downstream repos — two different kinds of finding wearing +one type, told apart by a prefix match. `.claude/rules/checks.md`: *NOT CHECKED and 0 FOUND must +never render the same.* An agent using `coherence_check()` as its acceptance test could only regex it. + +A frozen dataclass is tidier and costs a second breaking migration one release after `StepSpec`; +that is why the subclass wins. `blocking` is a bool rather than a severity enum — two states, and +no third has been observed. + +- `CoherenceFinding`, `blocking()` and `NOT_CHECKED` are exported from the package root. +- `coherence_check()` and every `check_*` function are now annotated `list[CoherenceFinding]`. + +### Upgrading + +Nothing to do. `uv lock --upgrade-package workflow-workbench` when you want the fields; until +then a pinned consumer is unaffected. + ## [0.2.0] — 2026-09-30 ### ⛔ Breaking — `NodeSpec` is renamed to `StepSpec` diff --git a/README.md b/README.md index 5faad33..2cb6c14 100644 --- a/README.md +++ b/README.md @@ -77,7 +77,7 @@ uv run python3 -m examples.greeting Everything it produces goes to the terminal; no files are written. Excerpt: ``` -1. check() with nothing implemented: clean +1. coherence_check() with nothing implemented: clean ... 3. what varies between the two strategies: {'normalize': ('trim', 'trim_and_collapse')} ... @@ -98,7 +98,7 @@ viewer is available as a separate process — `uv run python3 -m workflow_workbe | | step | what you can inspect | |---|---|---| | 1 | specify the workflow | the nodes, named values and edges, as data | -| 2 | check and draw it | `check()` findings and `diagram()` mermaid, with nothing implemented | +| 2 | check and draw it | `coherence_check()` findings and `diagram()` mermaid, with nothing implemented | | 3 | implement the steps | ordinary Pydantic Graph step bodies | | 4 | bind a named strategy | `diagram(strategy)` — the design with each role's implementation named | | 5 | check the strategy | missing bindings, wrong return types, and `render()` refusing outright | @@ -138,7 +138,7 @@ slots are one transposition away from a graph that is wrong and runs. ```python spec = Greeting() -spec.check() # -> [] — no strategy, no implementations, no engine +spec.coherence_check() # -> [] — no strategy, no implementations, no engine spec.diagram() # -> mermaid for the specification ``` @@ -170,13 +170,31 @@ says so; `render()` refuses rather than building a graph with a hole in it: ```python unfinished = StrategySpec("unfinished", {normalize: trim_and_collapse}) -spec.check(unfinished) +spec.coherence_check(unfinished) # ["strategy 'unfinished' does not bind node 'compose'. Every one is bound explicitly, # including unchanged ones — a partial strategy makes 'what varies between these arms' # unanswerable without reading both files."] spec.render(unfinished) # raises SpecError with the same finding ``` +A finding is a sentence, and it is also **structured**. `CoherenceFinding` is a `str` subclass, so +everything above reads exactly as it looks — and an agent driving this as its acceptance test can +branch on fields instead of matching on prose: + +```python +f = spec.coherence_check(unfinished)[0] +f.check # 'check_bindings' — which check produced it +f.about # 'compose' — the node; 'source->target' for an edge; '' for the design +f.blocking # True — False only for a `NOT CHECKED — …` stated gap + +from workflow_workbench import blocking +blocking(spec.coherence_check(unfinished)) # what `render()` refuses on, gaps excluded +``` + +`blocking` is a bool rather than a severity enum because there are two states and no third has +turned up. A stated gap and a clean pass must never read the same — that is the one distinction +`coherence_check()` has always made, and it used to be recoverable only with `startswith("NOT CHECKED")`. + Which is what makes growing a workflow safe: add a node and every existing strategy fails loudly rather than skipping a step it never heard of ([`stage3_new_node.py`](examples/ladder/stage3_new_node.py)). @@ -228,7 +246,7 @@ specification guarantees; behaviour is what the battle is for. The specification is the reviewable artifact. Review the diagram and the contracts, and the agent's job narrows to step bodies satisfying a declared input and output type for a named role, -with `check()` as the acceptance test. +with `coherence_check()` as the acceptance test. A proposed change to the workflow itself is then a diff to `nodes` and `edges` — one small place, reviewed on its own, not a behaviour change buried in a function body. diff --git a/docs/ladder.md b/docs/ladder.md index 90f61f9..400dd70 100644 --- a/docs/ladder.md +++ b/docs/ladder.md @@ -49,14 +49,14 @@ class HelloWorld(GraphSpec): ``` Both print `'Hello, Ada!'` and both have the node ids `pick`, `compose`. On this rung the -declaration buys you `check()` and `diagram()` before any implementation exists, and nothing +declaration buys you `coherence_check()` and `diagram()` before any implementation exists, and nothing else — it starts paying on rung 2, when `pick` has two implementations and something has to hold them to one shape. | rung | adds | source | |---|---|---| | 0 | nothing — Pydantic Graph alone, the control | [`their_hello.py`](../examples/ladder/their_hello.py) | -| 1 | the design as data; `check()` and `diagram()` with nothing implemented | [`stage1_bare.py`](../examples/ladder/stage1_bare.py) | +| 1 | the design as data; `coherence_check()` and `diagram()` with nothing implemented | [`stage1_bare.py`](../examples/ladder/stage1_bare.py) | | 2 | **two strategies over one design**, with identical node ids | [`stage2_strategies.py`](../examples/ladder/stage2_strategies.py) | | 3 | a new node — and a strategy that predates it is refused | [`stage3_new_node.py`](../examples/ladder/stage3_new_node.py) | | 4 | one node implemented by a **whole child design** | [`stage4_subgraph.py`](../examples/ladder/stage4_subgraph.py) | diff --git a/docs/migration-0.3.md b/docs/migration-0.3.md new file mode 100644 index 0000000..0bfd66a --- /dev/null +++ b/docs/migration-0.3.md @@ -0,0 +1,65 @@ +# Upgrading to 0.3.0 + +Two changes. One is a rename you must make; the other needs nothing from you. + +## 1. `check()` → `coherence_check()` — required + +```python +spec.check() # 0.2.0 +spec.coherence_check() # 0.3.0 + +spec.check(strategy) # 0.2.0 +spec.coherence_check(strategy) # 0.3.0 +``` + +**There is no alias.** A missed call site raises `AttributeError: 'YourSpec' object has no +attribute 'check'` at the call — loud, and at the line that needs editing. 0.2.0 made the same +choice when `NodeSpec(...)` became a `TypeError`: a silent narrowing would be worse than a stop. + +```bash +grep -rn "\.check(" --include=*.py . # every site, and there is nothing else named .check( +sed -i 's/\.check(/.coherence_check(/g' +pytest -q +``` + +**Unchanged, and deliberately so:** + +| | | +|---|---| +| `check_names`, `check_reachable`, … the eleven functions | unchanged — each *is* one check | +| `CoherenceFinding.check` | unchanged — it names which of those eleven produced the finding | +| `render()`, `diagram()`, `diff_diagram()`, `varies()`, `eval_battle()` | unchanged | + +### Why + +`check()` did not say what it checks, and the type it returns said `Coherence` — a word that +appeared nowhere else in the API. One concept, two names. + +`design_check()` was considered and rejected: `spec` *is* the design, so `spec.design_check()` +restates its own receiver, the way `file.file_close()` would. + +## 2. `coherence_check()` returns `CoherenceFinding` — nothing to do + +`CoherenceFinding` is a `str` subclass, so every string operation on a finding behaves exactly as +it did. Verified byte-for-byte across all 64 findings this repo's designs produce. + +```python +f = spec.coherence_check(strategy)[0] + +f == "the raw message" # True, as before +"unreachable" in f # as before +"\n".join(findings) # as before +f.startswith("NOT CHECKED") # as before — and `f.blocking` now says the same thing as a field + +f.check # 'check_bindings' — which check produced it +f.about # 'compose'; 'source->target' for an edge; '' for the whole design +f.blocking # False only for a `NOT CHECKED — …` stated gap +``` + +The prefix match is still correct and still supported. `blocking(findings)` is the shared filter +`render()` uses, if you would rather not spell it out: + +```python +from workflow_workbench import blocking +blocking(spec.coherence_check(strategy)) +``` diff --git a/docs/parity.md b/docs/parity.md index 0e78d25..a3e2f10 100644 --- a/docs/parity.md +++ b/docs/parity.md @@ -1,6 +1,6 @@ # What a `GraphSpec` can express — every Pydantic Graph builder feature, enumerated -`GraphSpec` declares a workflow as DATA, because data is the only form `check()` and `diagram()` +`GraphSpec` declares a workflow as DATA, because data is the only form `coherence_check()` and `diagram()` can read before any implementation exists. That buys the checks and the diagrams, and it costs expressiveness: a few things Pydantic Graph lets you write in code cannot be written down. diff --git a/docs/probe_builder_features.py b/docs/probe_builder_features.py index 79548c9..1621355 100644 --- a/docs/probe_builder_features.py +++ b/docs/probe_builder_features.py @@ -6,10 +6,10 @@ does it RUN? can the feature be reached at all from a GraphSpec, if necessary through `build_pydantic_structure()` is it DECLARED? is it in `nodes`/`joins`/`decisions`/`edges` as DATA — which is the only - form `check()`, `diagram()`, `diff_diagram()` and `varies()` can read + form `coherence_check()`, `diagram()`, `diff_diagram()` and `varies()` can read ⚠️ The second is the whole product. An escape-hatch topology runs perfectly and is invisible to -every check this library exists to provide — `check()` says so out loud (`NOT CHECKED — ... +every check this library exists to provide — `coherence_check()` says so out loud (`NOT CHECKED — ... overrides build_pydantic_structure()`), and the middle section measures exactly that. ⛔ THE TABLE IS `workflow_workbench/parity.py`, AND IT IS CHECKED AGAINST THE REAL API. It was @@ -224,7 +224,7 @@ async def dbl(ctx) -> int: only = StrategySpec("only", {double: dbl}) -print(f"check() -> {Declarative().check(only) or 'clean, reachability VERIFIED'}") +print(f"coherence_check() -> {Declarative().coherence_check(only) or 'clean, reachability VERIFIED'}") print(f"hook to override the wiring? " f"{hasattr(GraphSpec, 'build_pydantic_structure')}") print(" ⛔ There was one. It was the ONLY way a built graph could differ from its declaration,") diff --git a/examples/counter.py b/examples/counter.py index 270955b..e65f5c6 100644 --- a/examples/counter.py +++ b/examples/counter.py @@ -70,8 +70,8 @@ async def times_three(ctx) -> int: def main() -> None: spec = Counter() - findings = spec.check() - print(f"check() with no strategy at all: {findings or 'clean'}") + findings = spec.coherence_check() + print(f"coherence_check() with no strategy at all: {findings or 'clean'}") for strategy in (modest, aggressive): graph = spec.render(strategy) diff --git a/examples/greeting.py b/examples/greeting.py index d7d299b..2712b0d 100644 --- a/examples/greeting.py +++ b/examples/greeting.py @@ -117,7 +117,7 @@ def evaluate(self, ctx: EvaluatorContext) -> float: def main() -> None: spec = Greeting() - print("1. check() with nothing implemented:", spec.check() or "clean") + print("1. coherence_check() with nothing implemented:", spec.coherence_check() or "clean") print("\n2. the specification, drawn from the declaration:") print(spec.diagram()) @@ -127,7 +127,7 @@ def main() -> None: print("\n4. an incomplete strategy — `compose` left unbound:") unfinished = StrategySpec("unfinished", {normalize: trim_and_collapse}) - for finding in spec.check(unfinished): + for finding in spec.coherence_check(unfinished): print(f" check finding: {finding}") try: spec.render(unfinished) diff --git a/examples/ladder/stage10_no_basenode.py b/examples/ladder/stage10_no_basenode.py index 408c086..81b95e7 100644 --- a/examples/ladder/stage10_no_basenode.py +++ b/examples/ladder/stage10_no_basenode.py @@ -159,7 +159,7 @@ async def do_triage_permissive(ctx) -> object: def main() -> None: spec = Intake() - print(f"check(): {spec.check(careful) or 'clean — gate, loop and dispatch, all declared'}\n") + print(f"coherence_check(): {spec.coherence_check(careful) or 'clean — gate, loop and dispatch, all declared'}\n") graph = spec.render(careful) for text in ("my cat is unwell", "metformin 1000 mg daily"): diff --git a/examples/ladder/stage1_bare.py b/examples/ladder/stage1_bare.py index 619557d..de264f3 100644 --- a/examples/ladder/stage1_bare.py +++ b/examples/ladder/stage1_bare.py @@ -13,7 +13,7 @@ What you get already, and cannot get from a built Graph: - HelloWorld().check() runs with NO strategy and NO implementations + HelloWorld().coherence_check() runs with NO strategy and NO implementations HelloWorld().diagram() draws the design before anything is written uv run python3 -m examples.ladder.stage1_bare @@ -88,7 +88,7 @@ def main() -> None: # ⚠️ No strategy, no implementations, no engine. This is the thing a built Graph cannot do, # because a built Graph cannot exist until every function is written. - print(f"check() with nothing implemented: {spec.check() or 'clean'}") + print(f"coherence_check() with nothing implemented: {spec.coherence_check() or 'clean'}") graph = spec.render(formal) state = Guest() diff --git a/examples/ladder/stage8_join.py b/examples/ladder/stage8_join.py index c45fdb1..ff3e322 100644 --- a/examples/ladder/stage8_join.py +++ b/examples/ladder/stage8_join.py @@ -120,7 +120,7 @@ class BrokenGreetings(Greetings): def main() -> None: spec = Greetings() - print(f"check(): {spec.check(greet) or 'clean'}") + print(f"coherence_check(): {spec.coherence_check(greet) or 'clean'}") state = Guest() print(f"run('Ada') -> {spec.render(greet).run_sync(inputs='Ada', state=state)!r}") @@ -132,7 +132,7 @@ async def collect_step(ctx) -> list: broken = StrategySpec("broken", {say_formal: formal, say_casual: casual, collect_as_step: collect_step, announce: announce_both}) - for finding in BrokenGreetings().check(broken): + for finding in BrokenGreetings().coherence_check(broken): print(f" refused: {finding[:110]}...") try: BrokenGreetings().render(broken) diff --git a/examples/ladder/stage9_decision.py b/examples/ladder/stage9_decision.py index fb785f0..a9760fa 100644 --- a/examples/ladder/stage9_decision.py +++ b/examples/ladder/stage9_decision.py @@ -116,7 +116,7 @@ async def do_report(ctx) -> str: def main() -> None: spec = Triage() - print(f"check(): {spec.check(careful) or 'clean — including reachability, through branches'}\n") + print(f"coherence_check(): {spec.coherence_check(careful) or 'clean — including reachability, through branches'}\n") graph = spec.render(careful) for text in ("chest pain since this morning", "dry skin on my elbow"): @@ -150,7 +150,7 @@ class NoWhen(Triage): EdgeSpec(source=research, target=report, carries=handled), EdgeSpec(source=report, target=END, carries=report_out)) - for finding in NoWhen().check(careful): + for finding in NoWhen().coherence_check(careful): print(f" {finding[:118]}...") class StrayWhen(Triage): @@ -163,7 +163,7 @@ class StrayWhen(Triage): EdgeSpec(source=research, target=report, carries=handled), EdgeSpec(source=report, target=END, carries=report_out)) - for finding in StrayWhen().check(careful): + for finding in StrayWhen().coherence_check(careful): print(f" {finding[:118]}...") try: diff --git a/examples/local/extraction.py b/examples/local/extraction.py index 83e959e..2cdf260 100644 --- a/examples/local/extraction.py +++ b/examples/local/extraction.py @@ -72,7 +72,7 @@ async def keep_confident(ctx) -> list[Fact]: def main() -> None: spec = Extraction() - print(f"check(): {spec.check(greedy) or 'clean'}") + print(f"coherence_check(): {spec.coherence_check(greedy) or 'clean'}") for strategy in (greedy, strict): graph = spec.render(strategy) diff --git a/examples/parallel.py b/examples/parallel.py index 310a9a9..66218e4 100644 --- a/examples/parallel.py +++ b/examples/parallel.py @@ -73,8 +73,8 @@ async def cube(ctx) -> int: def main() -> None: spec = ParallelProcessing() - print(f"check() with no strategy: {spec.check() or 'clean'}") - print(f"check(squares): {spec.check(squares) or 'clean'}") + print(f"coherence_check() with no strategy: {spec.coherence_check() or 'clean'}") + print(f"check(squares): {spec.coherence_check(squares) or 'clean'}") print(" ⚠️ neither says NOT CHECKED. A fan-out design is now checked like any other.\n") for strategy in (squares, cubes): diff --git a/examples/subgraph.py b/examples/subgraph.py index a9dcecc..677ee32 100644 --- a/examples/subgraph.py +++ b/examples/subgraph.py @@ -139,7 +139,7 @@ def main() -> None: deps = ExtractionDeps() print(f"design checks clean with no strategy at all: " - f"{ExtractionWorkflow().check() or 'yes'}") + f"{ExtractionWorkflow().coherence_check() or 'yes'}") # The child stands on its own. If it did not, it would be a fragment, not a design. child_state = ExtractionState() diff --git a/pyproject.toml b/pyproject.toml index 85bcf96..e8fc714 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "workflow-workbench" -version = "0.2.0" +version = "0.3.0" description = "One fixed graph design, many competing implementations — checked, diagrammed, and battled on Pydantic Evals" readme = "README.md" requires-python = ">=3.12" diff --git a/tests/test_fan_out.py b/tests/test_fan_out.py index 9781dae..dc3f0c2 100644 --- a/tests/test_fan_out.py +++ b/tests/test_fan_out.py @@ -17,9 +17,9 @@ def test_a_declared_fan_out_is_checked_and_runs() -> None: from examples.parallel import ParallelProcessing, cubes, squares spec = ParallelProcessing() - assert spec.check() == [] - assert spec.check(squares) == [] - assert not any("NOT CHECKED" in f for f in spec.check(squares)) + assert spec.coherence_check() == [] + assert spec.coherence_check(squares) == [] + assert not any("NOT CHECKED" in f for f in spec.coherence_check(squares)) assert spec.render(squares).run_sync(inputs=[1, 2, 3, 4]) == 30 assert spec.render(cubes).run_sync(inputs=[1, 2, 3, 4]) == 100 @@ -59,7 +59,7 @@ class Mismatched(GraphSpec): async def double(ctx) -> int: return ctx.inputs * 2 - findings = Mismatched().check(StrategySpec("s", {step: double})) + findings = Mismatched().coherence_check(StrategySpec("s", {step: double})) assert any("wrong_item" in f and "one item per run" in f for f in findings), findings @@ -103,7 +103,7 @@ async def by_space(ctx): spec = Splitter() strategy = StrategySpec("by_space", {split: by_space}) - assert not [f for f in spec.check(strategy) if not f.startswith("NOT CHECKED")] + assert not [f for f in spec.coherence_check(strategy) if not f.startswith("NOT CHECKED")] graph = spec.render(strategy) assert sorted(graph.run_sync(inputs="a b c")) == ["a", "b", "c"] @@ -171,7 +171,7 @@ async def look_up(ctx) -> float: strategy = StrategySpec("lookup", {price: look_up}) spec = Shop() - assert spec.check(strategy) == [] + assert spec.coherence_check(strategy) == [] assert round(spec.render(strategy).run_sync(inputs=list(prices)), 2) == 4.80 # the point of the whole construct: `price` never sees the list @@ -183,7 +183,7 @@ def test_a_fan_out_that_never_rejoins_is_refused() -> None: """⛔ The mirror of `check_step_arity`, and it was missing until someone asked why a join is always needed. Measured before the check existed: - check() -> clean + coherence_check() -> clean run -> 1.2 price ran 3 times, with ['milk', 'eggs', 'bread'] @@ -240,5 +240,33 @@ async def double(ctx) -> int: spec = TwoStepsThenJoin() strategy = StrategySpec("s", {first: keep, second: double}) - assert spec.check(strategy) == [] + assert spec.coherence_check(strategy) == [] assert spec.render(strategy).run_sync(inputs=[1, 2, 3]) == 12 + + +def test_the_fan_out_finding_is_about_the_EDGE_that_fans_out() -> None: + """`about` is the handle a caller filters on, and for this defect it is the edge. + + Neither endpoint alone names the problem: START is fine and `price` is fine — it is the + `.map()` between them with nothing downstream to divide it back. An `about` naming either + node would send an agent to fix something that is not broken. + """ + shopping = VariableSpec("shopping", list) + item = VariableSpec("item", str) + cost = VariableSpec("cost", float) + price = StepSpec("price", inputs=(item,), outputs=(cost,)) + + class NoJoin(GraphSpec): + name = "no_join" + input_type, output_type = list, float + nodes = (price,) + edges = (MapEdgeSpec(source=START, target=price, carries=shopping, delivers=item), + EdgeSpec(source=price, target=END, carries=cost)) + + async def look_up(ctx) -> float: + return 1.0 + + fan = [f for f in NoJoin().coherence_check(StrategySpec("s", {price: look_up})) + if "without passing a join" in f] + assert [(f.check, f.about, f.blocking) for f in fan] == \ + [("check_fan_out_rejoins", "START->price", True)] diff --git a/tests/test_greeting.py b/tests/test_greeting.py index 207c6ff..2661bea 100644 --- a/tests/test_greeting.py +++ b/tests/test_greeting.py @@ -29,7 +29,7 @@ def test_the_specification_checks_clean_with_nothing_implemented() -> None: """Stage 2 of the walkthrough, and the thing a built `Graph` cannot reach.""" - assert Greeting().check() == [] + assert Greeting().coherence_check() == [] def test_both_strategies_satisfy_the_specification() -> None: @@ -37,7 +37,7 @@ def test_both_strategies_satisfy_the_specification() -> None: structural consistency does not imply correct behaviour.""" spec = Greeting() for strategy in (trim_only, normalize_spaces): - assert spec.check(strategy) == [], strategy.name + assert spec.coherence_check(strategy) == [], strategy.name def test_both_arms_have_the_same_node_ids_so_the_comparison_aligns() -> None: @@ -72,7 +72,7 @@ def test_an_incomplete_strategy_is_reported_and_then_refused() -> None: spec = Greeting() unfinished = StrategySpec("unfinished", {normalize: trim_and_collapse}) - findings = spec.check(unfinished) + findings = spec.coherence_check(unfinished) assert len(findings) == 1 and "'compose'" in findings[0] with pytest.raises(SpecError) as exc: diff --git a/tests/test_ladder.py b/tests/test_ladder.py index 4d78f10..7db0f7d 100644 --- a/tests/test_ladder.py +++ b/tests/test_ladder.py @@ -42,7 +42,7 @@ def test_rung1_checks_before_anything_is_implemented() -> None: """The capability a built Graph cannot have: it cannot exist until every body is written.""" from examples.ladder.stage1_bare import HelloWorld - assert HelloWorld().check() == [] + assert HelloWorld().coherence_check() == [] assert "flowchart" in HelloWorld().diagram() @@ -259,7 +259,7 @@ def test_rung8_a_join_actually_combines_both_arrivals() -> None: from examples.ladder.stage8_join import Greetings, Guest, greet spec = Greetings() - assert spec.check(greet) == [] + assert spec.coherence_check(greet) == [] result = spec.render(greet).run_sync(inputs="Ada", state=Guest()) assert result == "Hello, Ada! / Yo, Ada!" @@ -287,7 +287,7 @@ def test_rung8_a_join_is_reachability_checked_which_it_could_not_be_before() -> on every node around it.""" from examples.ladder.stage8_join import Greetings, greet - findings = Greetings().check(greet) + findings = Greetings().coherence_check(greet) assert not any("NOT CHECKED" in f for f in findings), findings @@ -296,7 +296,7 @@ def test_rung8_a_join_binds_nothing_and_never_appears_in_varies() -> None: from examples.ladder.stage8_join import Greetings, collect, greet assert collect not in greet.bindings - assert Greetings().check(greet) == [] + assert Greetings().coherence_check(greet) == [] assert "collect" not in Greetings().varies(greet, greet) @@ -322,7 +322,7 @@ def test_rung9_each_branch_routes_and_only_one_fires() -> None: from examples.ladder.stage9_decision import Log, Triage, careful spec = Triage() - assert spec.check(careful) == [] + assert spec.coherence_check(careful) == [] graph = spec.render(careful) urgent_log = Log() @@ -349,7 +349,7 @@ def test_rung9_converging_branches_are_not_a_fan_in() -> None: incoming = [e for e in spec.edges if e.target is report] assert len(incoming) == 2, "the test's premise is gone; report is no longer a convergence" - assert spec.check(careful) == [], "a converging branch was reported as a fan-in" + assert spec.coherence_check(careful) == [], "a converging branch was reported as a fan-in" for text in ("chest pain now", "dry elbow"): log = Log() @@ -372,7 +372,7 @@ class RealFanIn(Triage): EdgeSpec(source=intake, target=sneak, carries=verdict), # NOT behind the decision EdgeSpec(source=sneak, target=report, carries=handled)) # a third, unconditional arrival - findings = RealFanIn().check() + findings = RealFanIn().coherence_check() assert any("invoked once PER EDGE" in f for f in findings), findings @@ -433,7 +433,7 @@ def test_rung9_reachability_runs_through_branches() -> None: design — so every branching workflow was entirely unchecked.""" from examples.ladder.stage9_decision import Triage, careful - assert not any("NOT CHECKED" in f for f in Triage().check(careful)) + assert not any("NOT CHECKED" in f for f in Triage().coherence_check(careful)) def test_rung9_the_diagram_labels_branches_by_type_not_variable() -> None: @@ -494,7 +494,7 @@ def test_rung10_a_retry_loop_and_a_dispatch_in_the_same_design() -> None: assert out == "audited: draft-2" assert [s for s in log.steps if s.startswith("propose")] == ["propose#1", "propose#2"] - assert Intake().check(careful) == [] + assert Intake().coherence_check(careful) == [] def test_rung10_two_arms_gate_differently_without_moving_the_topology() -> None: diff --git a/tests/test_parity.py b/tests/test_parity.py index 1fca99c..c633469 100644 --- a/tests/test_parity.py +++ b/tests/test_parity.py @@ -121,7 +121,7 @@ def test_the_probe_reads_parity_rather_than_keeping_its_own_copy() -> None: #: spelled out rather than globbed: `docs/migration-*.md` would silently cover a new file nobody #: reviewed. `test_the_retirement_exemptions_all_exist` fails if either path is renamed, so the #: exemption cannot outlive the document it was written for. -_NAMES_THE_OLD_API = ("CHANGELOG.md", "docs/migration-0.2.md") +_NAMES_THE_OLD_API = ("CHANGELOG.md", "docs/migration-0.2.md", "docs/migration-0.3.md") def _prose_docs(*, include_migration: bool = True) -> list[tuple[str, str]]: diff --git a/tests/test_subgraph.py b/tests/test_subgraph.py index bc619fa..2b2fd18 100644 --- a/tests/test_subgraph.py +++ b/tests/test_subgraph.py @@ -265,7 +265,7 @@ def test_a_node_wired_to_the_sentinels_is_checked_like_any_other() -> None: graph = StartEndParent().render(strategy) assert graph.run_sync(inputs=" hi ", state=state, deps=Deps(prefix="ok:")) == "ok:HI" assert state.calls == ["first", "second"] - assert StartEndParent().check(strategy) == [] + assert StartEndParent().coherence_check(strategy) == [] def test_a_node_that_declares_nothing_is_now_refused() -> None: @@ -282,7 +282,7 @@ class Silent(GraphSpec): async def anything(ctx) -> str: return ctx.inputs - findings = Silent().check(StrategySpec("s", {silent: anything})) + findings = Silent().coherence_check(StrategySpec("s", {silent: anything})) assert any("does not declare it as an input" in f for f in findings), findings @@ -339,7 +339,7 @@ async def one(ctx) -> str: strategy = StrategySpec("sub", {mid_first: one, mid_second: SubgraphBinding(Child(), child_strategy)}) - assert MidParent().check(strategy) == [] + assert MidParent().coherence_check(strategy) == [] MidParent().render(strategy) @@ -366,7 +366,7 @@ def test_a_multi_port_node_is_rejected_rather_than_guessed() -> None: def test_recursive_subgraph_binding_is_rejected() -> None: """A design implementing one of its own nodes with itself builds children until the stack - ends. Caught in `_check`, which is the single owner of the ancestry path.""" + ends. Caught in `_coherence_check`, which is the single owner of the ancestry path.""" bindings: dict = {} recursive = StrategySpec("recursive", bindings) bindings[transform] = SubgraphBinding(graph=Parent(), strategy=recursive) @@ -447,7 +447,7 @@ class FanIn(GraphSpec): def test_a_node_that_cannot_receive_both_its_inputs_is_refused() -> None: """⛔ Measured before this check existed: it rendered, ran, called `merge` TWICE with one - value each, and returned one result while discarding the other. `check()` said clean. + value each, and returned one result while discarding the other. `coherence_check()` said clean. Every other check passes on it — both variables are declared on both ends, everything reaches END. Only arity sees it. @@ -456,7 +456,7 @@ async def one(ctx) -> str: return ctx.inputs strategy = StrategySpec("s", {split_a: one, split_b: one, merge: one}) - findings = FanIn().check(strategy) + findings = FanIn().coherence_check(strategy) assert len(findings) == 2, findings assert "declares 2 inputs" in findings[0] @@ -468,8 +468,8 @@ async def one(ctx) -> str: def test_a_linear_chain_is_not_flagged() -> None: """The check must not fire on the ordinary shape, or it is noise nobody reads.""" - assert Parent().check(direct_strategy) == [] - assert Child().check(child_strategy) == [] + assert Parent().coherence_check(direct_strategy) == [] + assert Child().coherence_check(child_strategy) == [] # ── a loop-back is not a fan-in ───────────────────────────────────────────────────────────── @@ -547,7 +547,7 @@ async def do_finish(ctx) -> str: unwrap: do_unwrap, finish: do_finish}) spec = WithRetry() - assert spec.check(strategy) == [], "a loop-back was reported as a fan-in" + assert spec.coherence_check(strategy) == [], "a loop-back was reported as a fan-in" log = Log() assert spec.render(strategy).run_sync(inputs="a plan", state=log) == "done(draft-3)" @@ -573,4 +573,45 @@ class RealFanIn(GraphSpec): EdgeSpec(source=two, target=sink, carries=a_var), EdgeSpec(source=sink, target=END, carries=text)) - assert any("invoked once PER EDGE" in f for f in RealFanIn().check()), RealFanIn().check() + assert any("invoked once PER EDGE" in f for f in RealFanIn().coherence_check()), RealFanIn().coherence_check() + + +def test_a_subgraph_finding_is_about_the_PARENT_node_the_child_is_bound_to() -> None: + """The child graph is coherent on its own — `WrongInput` renders and runs. What is wrong is + the pairing, and the parent node is the only name that exists in the design the caller + handed in. Naming the child's node would point at another design entirely. + """ + other = StepSpec("other", inputs=(number,), outputs=(text,)) + + class WrongInput(GraphSpec): + name = "wrong_input" + state_type, deps_type = State, Deps + input_type, output_type = int, str + nodes = (other,) + edges = (EdgeSpec(source=START, target=other, carries=number), + EdgeSpec(source=other, target=END, carries=text)) + + async def run(ctx) -> str: + return str(ctx.inputs) + + bad = StrategySpec("bad", {transform: SubgraphBinding( + graph=WrongInput(), strategy=StrategySpec("inner", {other: run}))}) + + mismatch = [f for f in Parent().coherence_check(bad) if "input_type" in f] + assert [(f.check, f.about) for f in mismatch] == [("check_subgraphs", "transform")] + + +def test_a_recursive_binding_is_about_the_STRATEGY_not_the_design() -> None: + """⚠️ The one finding produced outside `checks.py` — `GraphSpec._check` owns cycle detection + because it is the only thing holding the `ancestry`. It is therefore the one most likely to + be left untagged, and nothing else here would notice. + + `about` is the strategy because the design is fine: `Parent` and `Child` are both coherent, + and swapping the strategy is the move that fixes it. + """ + loop = StrategySpec("loop", {transform: direct}) + loop.bindings[transform] = SubgraphBinding(graph=Parent(), strategy=loop) + + recursive = [f for f in Parent().coherence_check(loop) if "recursive subgraph binding" in f] + assert recursive, "the cycle was not detected at all" + assert all((f.check, f.about) == ("GraphSpec._coherence_check", "loop") for f in recursive) diff --git a/tests/test_transform_edge.py b/tests/test_transform_edge.py index b639661..9b1693b 100644 --- a/tests/test_transform_edge.py +++ b/tests/test_transform_edge.py @@ -63,7 +63,7 @@ def test_a_fixed_transform_runs_and_creates_no_node() -> None: reader would count it as a stage of the workflow. It is not one — it is an accessor. """ spec = Fixed() - assert spec.check(fixed_s) == [] + assert spec.coherence_check(fixed_s) == [] graph = spec.render(fixed_s) assert graph.run_sync(inputs="x") == "2 edges cited" @@ -111,7 +111,7 @@ def test_two_arms_can_reshape_differently_and_varies_says_so() -> None: """⛔ The reason this is bindable at all. A difference nobody can see is the one that ruins a comparison — two arms that pruned differently would otherwise look identical.""" spec = Varying() - assert spec.check(arm_all) == [] + assert spec.coherence_check(arm_all) == [] assert spec.render(arm_all).run_sync(inputs="x") == "2 edges cited" assert spec.render(arm_first).run_sync(inputs="x") == "1 edges cited" @@ -161,7 +161,7 @@ class Mismatched(Fixed): name = "mismatched" edges = (EdgeSpec(source=START, target=propose, carries=plan), bad_edge, EdgeSpec(source=cite, target=END, carries=report)) - findings = Mismatched().check(fixed_s) + findings = Mismatched().coherence_check(fixed_s) assert any("reshaped on the wire" in f and "wrong" in f for f in findings), findings @@ -187,3 +187,31 @@ def test_fan_out_and_reshape_are_separate_types() -> None: assert not issubclass(TransformEdgeSpec, MapEdgeSpec) assert not hasattr(TransformEdgeSpec(source=propose, target=cite, carries=draft, delivers=edge_list, apply=take_edges), "map_over") + + +def test_a_transform_finding_is_about_the_WIRE_it_sits_on() -> None: + """A transform edge has no name of its own, so `about` is its two endpoints. + + ⚠️ The assertion worth having is that TWO different checks hand back the SAME handle for the + same wire. `check_bindings` sees an unbound variation point and `check_transform_edges` sees + a transform that reshapes nothing — one defect, two checks, and a caller grouping findings by + `about` must get one group rather than two. + """ + incomplete = StrategySpec("incomplete", {propose: do_propose, cite: do_cite}) + findings = Varying().coherence_check(incomplete) + + unbound = [f for f in findings if "no `apply=` and no binding" in f] + assert [(f.check, f.about) for f in unbound] == [("check_transform_edges", "propose->cite")] + + not_bound = [f for f in findings if "does not bind" in f] + assert [(f.check, f.about) for f in not_bound] == [("check_bindings", "propose->cite")] + + +def test_an_async_transform_finding_is_about_that_wire_too() -> None: + async def slow(ctx) -> list: + return ctx.inputs.edges + + s = StrategySpec("bad", {propose: do_propose, cite: do_cite, shape: slow}) + async_findings = [f for f in Varying().coherence_check(s) if "cannot await" in f] + assert [(f.check, f.about) for f in async_findings] == \ + [("check_transform_edges", "propose->cite")] diff --git a/tests/test_workflow_spec.py b/tests/test_workflow_spec.py index b789cb0..810ebd8 100644 --- a/tests/test_workflow_spec.py +++ b/tests/test_workflow_spec.py @@ -17,7 +17,12 @@ SpecError, StrategySpec, VariableSpec, + CoherenceFinding, + DecisionSpec, + NOT_CHECKED, + blocking, check_bindings, + check_decisions, check_implementations, check_names, check_reachable, @@ -191,7 +196,7 @@ def test_varies_names_only_what_differs(): def test_check_with_no_strategy_needs_no_implementations(): - assert Linear().check() == [] + assert Linear().coherence_check() == [] # ── there is exactly one way to wire a graph ──────────────────────────────────────────────── @@ -214,7 +219,7 @@ class TriesToOverride(Linear): def build_pydantic_structure(self, g, nodes): # noqa: ARG002 — deliberately ignored raise AssertionError("this must never be called") - assert TriesToOverride().check(arm_a) == [] + assert TriesToOverride().coherence_check(arm_a) == [] assert TriesToOverride().render(arm_a).run_sync(inputs="hi") == "A:HI" @@ -268,3 +273,191 @@ def test_a_plain_edge_cannot_deliver_something_else(): mechanism to convert, so declaring a different arrival would be a claim it cannot honour.""" with pytest.raises(SpecError, match="cannot deliver something other than it carries"): EdgeSpec(source=load, target=parse, carries=text, delivers=VariableSpec("other", int)) + + +# ── CoherenceFinding ──────────────────────────────────────────────────────────────────────── +# +# ⚠️ The 221 tests above are a real oracle for the MESSAGES — they assert substrings, so a +# reworded finding fails loudly. They cannot say anything about `check` and `about`, which are +# new and which nothing else would notice being wrong. That is what this section is for. + +def test_a_finding_is_still_a_string_everywhere_it_was_one(): + """⛔ THE COMPATIBILITY ORACLE. This is why the design is a `str` subclass and not a + dataclass: ~30 call sites here and in two downstream repos do these five things to a finding + and none of them was edited. If this test fails, the subclass stopped being additive. + """ + msg = "node 'orphan' is unreachable from START — it never runs." + f = CoherenceFinding(msg, check="check_reachable", about="orphan") + + assert f == msg and str(f) == msg # value equality, byte-for-byte text + assert "unreachable" in f # substring containment + assert not f.startswith(NOT_CHECKED) # the prefix match, still the same answer + assert "\n ".join([f, f]) == f"{msg}\n {msg}" + assert hash(f) == hash(msg) and {f} == {msg} + assert repr([f]) == repr([msg]), "repr must not move — examples print whole lists of these" + assert isinstance(f, str) + + +def test_blocking_is_derived_and_not_checked_is_the_only_non_blocking_kind(): + assert CoherenceFinding("anything at all", check="c").blocking + assert not CoherenceFinding(f"{NOT_CHECKED} — we could not look", check="c").blocking + # A stated gap and a clean pass must not read the same — `.claude/rules/checks.md`. + assert blocking([CoherenceFinding(f"{NOT_CHECKED} — x", check="c")]) == [] + + +def test_blocking_agrees_with_the_comprehension_it_replaces(): + """`render()`, `eval_battle` and the devserver all used the same `startswith` comprehension. + They call `blocking()` now, so the two must give the identical verdict on the same input — + including on a PLAIN string a caller mixed in, which has no `.blocking` to read.""" + findings = [*Linear().coherence_check(StrategySpec("partial", {load: load_a})), + f"{NOT_CHECKED} — a plain string from somewhere else", + "a plain string that is a real defect"] + assert blocking(findings) == [f for f in findings if not f.startswith(NOT_CHECKED)] + + +def test_every_check_tags_its_findings_with_its_own_name(): + """`check` must name the function that produced the finding — the whole point is that a + caller can branch on it. A typo'd or copy-pasted name is invisible to every other test.""" + import workflow_workbench.checks as c + + produced = {f.check for f in _every_finding_we_can_provoke()} + assert produced, "no findings were provoked — the assertions below would pass vacuously" + for name in produced: + if name == "GraphSpec._coherence_check": + continue # cycle detection needs `ancestry`; it has no `check_*` function + assert callable(getattr(c, name, None)), f"`check={name!r}` names no function in checks" + + +def test_about_names_something_the_caller_can_look_up(): + """⛔ THE `about` ORACLE, and the reason it is structural rather than a list of expected + strings: a hand-written table of 27 answers is as likely to be wrong as the code it checks. + + This asserts the INVARIANT instead — an `about` is empty, or it is a name the caller can + resolve against the design it just handed in. An `about` that names nothing is worse than an + empty one, because it reads as a handle and is not. + """ + spec = _Broken() + resolvable = {n.name for n in (*spec.nodes, *spec.joins, *spec.decisions)} + resolvable |= {"START", "END", _broken_arm.name} + + findings = spec.coherence_check(_broken_arm) + assert len(findings) > 5, f"only {len(findings)} findings — not enough to be a real sweep" + for f in findings: + if not f.about: + continue # a whole-design finding, stated as such + for part in f.about.split("->"): + assert part in resolvable, f"`about={f.about!r}` names {part!r}, which is not in {spec.name}" + + +def test_about_follows_the_subject_of_the_sentence_not_the_loop_variable(): + """The two cases where the obvious answer is the wrong one. + + An undeclared node cannot be looked up — so that finding is about the EDGE that references + it. And a decision's branch missing a `when=` is about that one branch, not about the + decision: a caller filtering on the decision name would be handed a finding it cannot act on + at the granularity it asked for. + """ + ghost = StepSpec("ghost") + edges = (*Linear.edges, EdgeSpec(source=parse, target=ghost, carries=text)) + undeclared = [f for f in check_reachable((load, parse), edges) if "not in `nodes`" in f] + assert [f.about for f in undeclared] == ["parse->ghost"], "named the ghost, not the edge" + + route = DecisionSpec("route") + branch = EdgeSpec(source=route, target=parse, carries=text) # no `when=` + no_when = [f for f in check_decisions((route,), (branch,)) if "without a `when=`" in f] + assert [f.about for f in no_when] == ["route->parse"] + + +def test_about_is_the_node_for_a_node_finding_and_the_edge_for_an_edge_finding(): + """One assertion per check that can produce a finding from a two-node design, because a + plausible-looking `about` on the wrong axis is exactly what nothing else here would see.""" + dup = StepSpec("load", (text,), (text,)) + assert [f.about for f in check_names((load, dup))] == ["load"] + + orphan = StepSpec("orphan") + unreachable = [f for f in check_reachable((load, parse, orphan), Linear.edges) + if "unreachable" in f] + assert [f.about for f in unreachable] == ["orphan"] + + wrong = EdgeSpec(source=load, target=parse, carries=other) + assert [f.about for f in check_variables((load, parse), (wrong,))] == \ + ["load->parse", "load->parse"] # neither end declares it: source AND target + + partial = StrategySpec("partial", {load: load_a}) + assert [f.about for f in check_bindings(Linear.nodes, partial)] == ["parse"] + + async def two_args(ctx, extra) -> str: + return "" + assert [f.about for f in check_implementations(StrategySpec("bad", {load: two_args}))] \ + == ["load"] + + +def test_a_whole_design_finding_says_so_with_an_empty_about(): + """⚠️ `""` is a VALUE here, not a missing one. "no edge leaves START" is about the design; + inventing a node name for it would make a filter on that node return a finding it did not + cause.""" + stranded = StepSpec("stranded") + assert [f.about for f in check_reachable((stranded,), ())] == ["", ""] # no START, no END + + +def test_every_append_site_is_tagged_even_the_ones_no_test_provokes(): + """⛔ THE RATCHET, and it is deliberately a source scan rather than a dynamic sweep. + + `test_every_check_tags_its_findings_with_its_own_name` is the stronger test — it reads real + `check` values off real findings — but it can only cover branches something provokes. A + `findings.append("...")` on a rare branch would return a plain `str`, and the FIRST caller to + read `.check` off it gets an `AttributeError` in production rather than a red test here. + + So this one asserts the shape of every append site, including the ones nothing reaches. + """ + import inspect + + import workflow_workbench.checks as c + + src = inspect.getsource(c) + assert src.count("findings.append(") > 20, "checks.py read as empty or tiny — vacuous" + assert src.count("findings.append(") == src.count("findings.append(CoherenceFinding("), \ + "a findings.append() in checks.py does not build a CoherenceFinding" + + +# ── the broken design these sweep over ────────────────────────────────────────────────────── +# +# One design that trips as many checks at once as possible. Deliberately NOT a list of expected +# messages: it exists so the structural assertions above run over real output from most of the +# check surface rather than over one hand-picked finding. + +_a, _b = VariableSpec("a", str), VariableSpec("b", int) +_split = StepSpec("split", outputs=(_a, _b)) +_merge = StepSpec("merge", inputs=(_a, _b), outputs=(_a,)) # two inputs: not a step +_lost = StepSpec("lost", inputs=(_a,)) # cannot reach END +_route = DecisionSpec("route") # no branches + + +class _Broken(GraphSpec): + name = "broken" + input_type, output_type = str, str + nodes = (_split, _merge, _lost) + decisions = (_route,) + edges = (EdgeSpec(source=START, target=_split, carries=_a), + EdgeSpec(source=_split, target=_merge, carries=_a), + EdgeSpec(source=_split, target=_merge, carries=_b), # fan-in onto a step + EdgeSpec(source=_split, target=_lost, carries=_a), + EdgeSpec(source=_merge, target=END, carries=_a)) + + +async def _untyped(ctx): # no return annotation + return "" + + +_broken_arm = StrategySpec("broken_arm", {_split: _untyped, _merge: _untyped, _lost: _untyped}) + + +def _every_finding_we_can_provoke(): + """Findings from across the check surface — the broken design, plus the cases its shape + cannot reach.""" + out = list(_Broken().coherence_check(_broken_arm)) + out += check_names((load, StepSpec("load", (text,), (text,)))) + out += check_implementations(StrategySpec("x", {load: "nope"})) + out += check_reachable((StepSpec("stranded"),), ()) + out += check_decisions((_route,), (EdgeSpec(source=_route, target=parse, carries=text),)) + return out diff --git a/uv.lock b/uv.lock index ae7d7ba..0d93b7f 100644 --- a/uv.lock +++ b/uv.lock @@ -633,7 +633,7 @@ wheels = [ [[package]] name = "workflow-workbench" -version = "0.2.0" +version = "0.3.0" source = { editable = "." } dependencies = [ { name = "pydantic" }, diff --git a/workflow_workbench/__init__.py b/workflow_workbench/__init__.py index 72b7639..a00d076 100644 --- a/workflow_workbench/__init__.py +++ b/workflow_workbench/__init__.py @@ -6,12 +6,17 @@ Bindable StepSpec | TransformEdgeSpec — everything a strategy must bind StrategySpec one complete set of implementations for it SubgraphBinding a whole child design, used as ONE node's implementation + CoherenceFinding what a check found — a `str`, with `check` / `about` / `blocking` on it + blocking the findings that stop a render, i.e. all but the stated gaps spec.render(strategy) -> a real pydantic_graph.Graph `evals` is imported separately (`from workflow_workbench.evals import eval_battle`) so that `render()` stays usable without an evaluation framework installed. """ from workflow_workbench.checks import ( + NOT_CHECKED, + CoherenceFinding, + blocking, check_bindings, check_fan_out_rejoins, check_decisions, @@ -51,6 +56,7 @@ "SubgraphBinding", "SpecError", "START", "END", + "CoherenceFinding", "blocking", "NOT_CHECKED", "check_names", "check_reachable", "check_variables", "check_bindings", "check_implementations", "check_subgraphs", "check_step_arity", "check_decisions", "check_variable_types", "check_transform_edges", "check_fan_out_rejoins", diff --git a/workflow_workbench/checks.py b/workflow_workbench/checks.py index d4eda01..5107099 100644 --- a/workflow_workbench/checks.py +++ b/workflow_workbench/checks.py @@ -1,7 +1,8 @@ """Every check, as pure data. No `pydantic_graph` import anywhere in this module. -Each returns a list of findings — strings a human can act on — and never raises. An empty list is -a pass; `GraphSpec.check()` is what turns a non-empty list into an exception. +Each returns a list of `CoherenceFinding` — sentences a human can act on, carrying the structure +an agent would otherwise have to regex back out — and never raises. An empty list is a pass; +`GraphSpec.coherence_check()` is what turns a non-empty list into an exception. ⚠️ Findings say what is wrong AND what it costs. "node 'x' is unreachable" is a fact; "…so its implementation never runs, and a strategy that binds it will look like it works" is a reason to @@ -10,7 +11,7 @@ from __future__ import annotations import inspect -from collections.abc import Callable +from collections.abc import Callable, Iterable from typing import Any from workflow_workbench.spec import ( @@ -29,16 +30,83 @@ is_sentinel, ) -__all__ = ["check_names", "check_reachable", "check_variables", "check_bindings", +__all__ = ["CoherenceFinding", "blocking", "NOT_CHECKED", + "check_names", "check_reachable", "check_variables", "check_bindings", "check_implementations", "check_subgraphs", "check_step_arity", "check_decisions", "check_variable_types", "check_transform_edges", "check_fan_out_rejoins"] +NOT_CHECKED = "NOT CHECKED" +"""The prefix that marks a STATED GAP rather than a defect. `.claude/rules/checks.md`: NOT CHECKED +and 0 FOUND must never render the same — this is the one spelling of that distinction, and +`CoherenceFinding.blocking` is how a caller reads it without matching on text.""" + + +def _is_blocking(message: str) -> bool: + """The ONE definition. `CoherenceFinding.blocking` and `blocking()` both call it, so the field + and the filter cannot drift apart into two slightly different ideas of what stops a render.""" + return not message.startswith(NOT_CHECKED) + + +class CoherenceFinding(str): + """One thing a check found, carrying what a caller needs instead of making them regex it out. + + A `str` SUBCLASS, deliberately. `"x" in f`, `f.startswith(...)`, `"\\n".join(findings)`, + `f == "the raw message"`, sorting, hashing and `repr()` in a printed list all behave exactly + as they did when these were plain strings — so this is additive, and the ~30 existing string + call sites here and in both downstream repos need no edit. A frozen dataclass is tidier and + costs a second breaking migration one release after `StepSpec`; that is why it loses. + + ⛔ `str(finding)` must reproduce the message BYTE FOR BYTE, and `__repr__` is deliberately NOT + overridden. The 221 existing tests assert finding substrings, and the examples print whole + lists of them — that is a real oracle for this change only as long as neither rendering moves. + + check the producing function's name, e.g. "check_variables" + about a node / join / decision `name`; "source->target" for an edge; the strategy + `name`; "" for a finding about the whole design + blocking derived: False for a stated gap (`NOT CHECKED — …`), True for a defect. + A bool, not an enum — two states, and no third has been observed. + """ + + __slots__ = ("check", "about", "blocking") + + check: str + about: str + blocking: bool + + def __new__(cls, message: str, *, check: str, about: str = "") -> CoherenceFinding: + self = super().__new__(cls, message) + self.check = check + self.about = about + self.blocking = _is_blocking(message) + return self + + +def blocking(findings: Iterable[str]) -> list[str]: + """The findings that STOP a render — everything that is not a stated gap. + + ⚠️ Takes `Iterable[str]`, not `Iterable[CoherenceFinding]`, and matches on the prefix rather + than reading `.blocking`. A caller that has mixed in a plain string of its own gets the same + verdict either way, and `_is_blocking` is what keeps the two readings identical. + """ + return [f for f in findings if _is_blocking(f)] + + def _name(ep: Any) -> str: return "START" if isinstance(ep, _Start) else "END" if isinstance(ep, _End) else ep.name +def _about_edge(e: EdgeSpec) -> str: + """An edge has no name of its own, so it is identified by the two it joins.""" + return f"{_name(e.source)}->{_name(e.target)}" + + +def _about(spec: Any) -> str: + """`about` for anything bindable — a node has a name, a transform edge has two endpoints.""" + return _about_edge(spec) if isinstance(spec, EdgeSpec) else getattr(spec, "name", "") + + def _type_name(t: Any) -> str: """A stable name for a type in a finding. `__name__` misses generic aliases like `list[Fact]`, which have none — and printing `` for one and a bare name for the @@ -46,7 +114,7 @@ def _type_name(t: Any) -> str: return getattr(t, "__name__", None) or repr(t) -def check_names(nodes: tuple[NodeSpec, ...]) -> list[str]: +def check_names(nodes: tuple[NodeSpec, ...]) -> list[CoherenceFinding]: """Node names must be unique — `render()` uses them as graph node ids. ⚠️ This check exists BECAUSE `StepSpec` is `eq=False`. Identity keying is what stops a @@ -54,19 +122,22 @@ def check_names(nodes: tuple[NodeSpec, ...]) -> list[str]: distinct nodes may share a name, and pydantic-graph would then refuse with a message about node ids that points at the render, not at the declaration. """ - findings, seen = [], {} + findings: list[CoherenceFinding] = [] + seen: dict[str, list[NodeSpec]] = {} for n in nodes: seen.setdefault(n.name, []).append(n) for name, group in seen.items(): if len(group) > 1: - findings.append( + findings.append(CoherenceFinding( f"{len(group)} different nodes are named {name!r}. Node names become graph node " f"ids, so this cannot be rendered — and because StepSpec is identity-keyed these " - f"really are separate nodes, not one node declared twice.") + f"really are separate nodes, not one node declared twice.", + check="check_names", about=name)) return findings -def check_reachable(nodes: tuple[NodeSpec, ...], edges: tuple[EdgeSpec, ...]) -> list[str]: +def check_reachable(nodes: tuple[NodeSpec, ...], + edges: tuple[EdgeSpec, ...]) -> list[CoherenceFinding]: """Every node reachable from START, and every node able to reach END. Pure Python over the declaration — no graph is built, so this runs before a single @@ -76,7 +147,7 @@ def check_reachable(nodes: tuple[NodeSpec, ...], edges: tuple[EdgeSpec, ...]) -> · unreachable from START — the node never runs; a strategy binding it looks like it works · cannot reach END — the work is done and thrown away, which reads as a silent drop """ - findings: list[str] = [] + findings: list[CoherenceFinding] = [] fwd: dict[Any, list[Any]] = {} bwd: dict[Any, list[Any]] = {} for e in edges: @@ -96,10 +167,16 @@ def walk(start: Any, adj: dict[Any, list[Any]]) -> set[int]: starts = [e.source for e in edges if isinstance(e.source, _Start)] ends = [e.target for e in edges if isinstance(e.target, _End)] + # ⚠️ `about=""` — these two are about the DESIGN, not about any one node. An `about` naming + # some arbitrary node would be a worse answer than an honest empty one. if not starts: - findings.append("no edge leaves START — nothing in this design can ever run.") + findings.append(CoherenceFinding( + "no edge leaves START — nothing in this design can ever run.", + check="check_reachable")) if not ends: - findings.append("no edge reaches END — this design produces no output.") + findings.append(CoherenceFinding( + "no edge reaches END — this design produces no output.", + check="check_reachable")) from_start: set[int] = set() for s in starts: @@ -110,25 +187,32 @@ def walk(start: Any, adj: dict[Any, list[Any]]) -> set[int]: for n in nodes: if starts and id(n) not in from_start: - findings.append( + findings.append(CoherenceFinding( f"node {n.name!r} is unreachable from START — its implementation never runs, so a " - f"strategy that binds it will appear to work while doing nothing.") + f"strategy that binds it will appear to work while doing nothing.", + check="check_reachable", about=n.name)) if ends and id(n) not in to_end: - findings.append( + findings.append(CoherenceFinding( f"node {n.name!r} cannot reach END — whatever it produces is discarded, which is " - f"indistinguishable from a step that was never wired.") + f"indistinguishable from a step that was never wired.", + check="check_reachable", about=n.name)) declared = {id(n) for n in nodes} for e in edges: for ep in (e.source, e.target): if not is_sentinel(ep) and id(ep) not in declared: - findings.append( + # ⚠️ `about` is the EDGE, not the undeclared node. The node is not in `nodes`, so + # naming it would point a caller at something it cannot look up; the edge is the + # thing that exists and the thing to delete or re-wire. + findings.append(CoherenceFinding( f"edge {e!r} references node {_name(ep)!r}, which is not in `nodes`. " - f"An undeclared node is invisible to every other check and to any strategy.") + f"An undeclared node is invisible to every other check and to any strategy.", + check="check_reachable", about=_about_edge(e))) return findings -def check_variables(nodes: tuple[NodeSpec, ...], edges: tuple[EdgeSpec, ...]) -> list[str]: +def check_variables(nodes: tuple[NodeSpec, ...], + edges: tuple[EdgeSpec, ...]) -> list[CoherenceFinding]: """Per edge: the variable it carries must be an output of its source and an input of its target. ⛔ PER EDGE, never aggregated across a node's edges. The aggregate form — "is the set of a @@ -145,15 +229,16 @@ def check_variables(nodes: tuple[NodeSpec, ...], edges: tuple[EdgeSpec, ...]) -> Edges touching START/END are skipped on the sentinel side: sentinels declare no variables, and the graph's own `input_type`/`output_type` is what constrains them. """ - findings: list[str] = [] + findings: list[CoherenceFinding] = [] for e in edges: if not is_sentinel(e.source): if e.carries not in e.source.outputs: declared = ", ".join(v.name for v in e.source.outputs) or "nothing" - findings.append( + findings.append(CoherenceFinding( f"edge {e!r} carries {e.carries.name!r}, but {e.source.name!r} does not " f"declare it as an output (it declares: {declared}). Either the edge is wired " - f"to the wrong variable or the node's contract is out of date.") + f"to the wrong variable or the node's contract is out of date.", + check="check_variables", about=_about_edge(e))) if not is_sentinel(e.target): # ⚠️ `delivers`, not `carries`. On a fan-out or a transform the two ends of one wire # carry DIFFERENT variables, and the target must be checked against what ARRIVES. @@ -165,13 +250,15 @@ def check_variables(nodes: tuple[NodeSpec, ...], edges: tuple[EdgeSpec, ...]) -> declared = ", ".join(v.name for v in e.target.inputs) or "nothing" how = (" (reshaped on the wire)" if isinstance(e, TransformEdgeSpec) else " (one item per run)" if isinstance(e, MapEdgeSpec) else "") - findings.append( + findings.append(CoherenceFinding( f"edge {e!r} delivers {arrives.name!r}{how} to {e.target.name!r}, which does " - f"not declare it as an input (it declares: {declared}).") + f"not declare it as an input (it declares: {declared}).", + check="check_variables", about=_about_edge(e))) return findings -def check_bindings(bindables: tuple[Bindable, ...], strategy: StrategySpec) -> list[str]: +def check_bindings(bindables: tuple[Bindable, ...], + strategy: StrategySpec) -> list[CoherenceFinding]: """The strategy binds exactly the declared VARIATION POINTS — no missing, no extra. ⚠️ `bindables`, not `nodes`. A `TransformEdgeSpec` with no `apply=` is a variation point too, @@ -182,27 +269,29 @@ def check_bindings(bindables: tuple[Bindable, ...], strategy: StrategySpec) -> l binding keyed on a look-alike node from another design, which is the failure identity keying exists to prevent. """ - findings: list[str] = [] + findings: list[CoherenceFinding] = [] declared = {id(n): n for n in bindables} bound = {id(n): n for n in strategy.bindings} for nid, n in declared.items(): if nid not in bound: - findings.append( + findings.append(CoherenceFinding( f"strategy {strategy.name!r} does not bind {_bindable_name(n)}. Every one is " f"bound " f"explicitly, including unchanged ones — a partial strategy makes 'what varies " - f"between these arms' unanswerable without reading both files.") + f"between these arms' unanswerable without reading both files.", + check="check_bindings", about=_about(n))) for nid, n in bound.items(): if nid not in declared: - findings.append( + findings.append(CoherenceFinding( f"strategy {strategy.name!r} binds {_bindable_name(n)}, which this design does " f"not declare. " f"Most likely it was written against a different GraphSpec that has a node of the " - f"same name.") + f"same name.", + check="check_bindings", about=_about(n))) return findings -def check_implementations(strategy: StrategySpec) -> list[str]: +def check_implementations(strategy: StrategySpec) -> list[CoherenceFinding]: """Each bound CALLABLE is callable and takes exactly one positional argument (`ctx`). Caught here rather than inside `GraphBuilder`, so a strategy's fault is reported against the @@ -213,13 +302,14 @@ def check_implementations(strategy: StrategySpec) -> list[str]: at the wrong file. Whether a subgraph fits is a RELATIONAL fact about it and the parent node, so `check_subgraphs`, which is handed the parent, owns it. """ - findings: list[str] = [] + findings: list[CoherenceFinding] = [] for node, impl in strategy.bindings.items(): if isinstance(impl, SubgraphBinding): continue if not callable(impl): - findings.append( - f"{strategy.name!r} binds {node.name!r} to {impl!r}, which is not callable.") + findings.append(CoherenceFinding( + f"{strategy.name!r} binds {node.name!r} to {impl!r}, which is not callable.", + check="check_implementations", about=_about(node))) continue try: sig = inspect.signature(impl) @@ -229,11 +319,12 @@ def check_implementations(strategy: StrategySpec) -> list[str]: if p.kind in (p.POSITIONAL_ONLY, p.POSITIONAL_OR_KEYWORD) and p.default is p.empty] if len(positional) != 1: - findings.append( + findings.append(CoherenceFinding( f"{strategy.name!r} binds {node.name!r} to " f"{getattr(impl, '__qualname__', impl)}{sig}, which takes {len(positional)} " f"required positional arguments. A pydantic-graph step body takes exactly one " - f"(`ctx`).") + f"(`ctx`).", + check="check_implementations", about=_about(node))) return findings @@ -287,7 +378,7 @@ def _port_type(parent: Any, node: StepSpec, side: str) -> tuple[Any, str | None] def check_subgraphs(parent: Any, strategy: StrategySpec, - *, ancestry: tuple[tuple[type, int], ...] = ()) -> list[str]: + *, ancestry: tuple[tuple[type, int], ...] = ()) -> list[CoherenceFinding]: """Every child design used as a node implementation fits the node it is bound to. ⚠️ `parent` is typed `Any` on purpose: `graph_spec` imports this module, so this module cannot @@ -305,7 +396,7 @@ def check_subgraphs(parent: Any, strategy: StrategySpec, Cycles are NOT checked here. `GraphSpec._check` owns that, so there is exactly one place that decides whether a chain has closed on itself. """ - findings: list[str] = [] + findings: list[CoherenceFinding] = [] declared_nodes = {id(n) for n in parent.nodes} for node, binding in strategy.bindings.items(): @@ -321,32 +412,39 @@ def check_subgraphs(parent: Any, strategy: StrategySpec, ("output", "output_type", child.output_type)): want, note = _port_type(parent, node, side) if note is not None: - findings.append(note if note.startswith("NOT CHECKED") else f"{where}, but {note}") + findings.append(CoherenceFinding( + note if note.startswith(NOT_CHECKED) else f"{where}, but {note}", + check="check_subgraphs", about=_about(node))) continue if child_type is not want: verb = "accepts" if side == "input" else "produces" - findings.append( + findings.append(CoherenceFinding( f"{where}, but the node {verb} {_type_name(want)} and the child graph " f"declares {port} {_type_name(child_type)}. A subgraph is a valid " - f"implementation only when its public boundary matches the role it fills.") + f"implementation only when its public boundary matches the role it fills.", + check="check_subgraphs", about=_about(node))) for attr in ("state_type", "deps_type"): mine, theirs = getattr(parent, attr), getattr(child, attr) if theirs is not mine: - findings.append( + findings.append(CoherenceFinding( f"{where}, but the parent declares {attr} {_type_name(mine)} and the child " f"declares {_type_name(theirs)}. A subgraph runs on the parent's exact " f"{attr.split('_')[0]} object, so the declared types must be identical — " f"there is no conversion, and inventing one would make it ambiguous who owns " - f"a mutation.") + f"a mutation.", + check="check_subgraphs", about=_about(node))) - findings += child._check(child_strategy, ancestry=ancestry) + # ⚠️ NOT re-tagged. A child's findings already name the check that produced them and the + # node inside the CHILD they are about; overwriting either with the parent's node would + # replace a precise answer with a vaguer one. + findings += child._coherence_check(child_strategy, ancestry=ancestry) return findings def check_decisions(decisions: tuple[DecisionSpec, ...], - edges: tuple[EdgeSpec, ...]) -> list[str]: + edges: tuple[EdgeSpec, ...]) -> list[CoherenceFinding]: """`when` appears exactly on the edges leaving a decision, and nowhere else. Both directions are real mistakes with different consequences: @@ -360,37 +458,44 @@ def check_decisions(decisions: tuple[DecisionSpec, ...], a decision with no branches at all routes nowhere; everything downstream is unreachable and the value is dropped """ - findings: list[str] = [] + findings: list[CoherenceFinding] = [] declared = {id(d) for d in decisions} + # ⚠️ `about` follows the SUBJECT of each sentence, not the loop variable. Two of these four + # are about the decision and two are about one edge of it — a caller filtering on a decision + # name would otherwise be handed findings it cannot act on at that granularity. for d in decisions: branches = [e for e in edges if e.source is d] if not branches: - findings.append( + findings.append(CoherenceFinding( f"decision {d.name!r} has no branches — no edge leaves it. It would route nothing " - f"and everything it was meant to reach is unreachable.") + f"and everything it was meant to reach is unreachable.", + check="check_decisions", about=d.name)) for e in branches: if e.when is None: - findings.append( + findings.append(CoherenceFinding( f"edge {e!r} leaves decision {d.name!r} without a `when=` type. A branch is " f"chosen by the type of the routed value; without one there is nothing to " - f"match on and the branch cannot be built.") + f"match on and the branch cannot be built.", + check="check_decisions", about=_about_edge(e))) seen: dict[Any, int] = {} for e in branches: if e.when is not None: seen[e.when] = seen.get(e.when, 0) + 1 for typ, n in seen.items(): if n > 1: - findings.append( + findings.append(CoherenceFinding( f"decision {d.name!r} has {n} branches matching {_type_name(typ)}. Only the " - f"first can ever be taken; the rest are dead and read as coverage.") + f"first can ever be taken; the rest are dead and read as coverage.", + check="check_decisions", about=d.name)) for e in edges: if e.when is not None and id(e.source) not in declared: - findings.append( + findings.append(CoherenceFinding( f"edge {e!r} carries `when={_type_name(e.when)}` but its source is not a " f"DecisionSpec, so the condition is IGNORED — the declaration reads as " - f"conditional and the graph routes unconditionally.") + f"conditional and the graph routes unconditionally.", + check="check_decisions", about=_about_edge(e))) return findings @@ -427,7 +532,8 @@ def reach(start: Any) -> set[int]: def check_step_arity(nodes: tuple[StepSpec, ...], edges: tuple[EdgeSpec, ...], - *, decisions: tuple[DecisionSpec, ...] = ()) -> list[str]: + *, decisions: tuple[DecisionSpec, ...] = () + ) -> list[CoherenceFinding]: """A step body receives exactly ONE value, so a node cannot consume two inputs at once. ⚠️ Takes `nodes` ONLY, never joins. A `JoinSpec` exists precisely to receive several arrivals @@ -470,7 +576,7 @@ def check_step_arity(nodes: tuple[StepSpec, ...], edges: tuple[EdgeSpec, ...], The lesson is in the rule's shape: it is stated in terms of edge COUNT, which is easy to compute and is not the question. Concurrency is. """ - findings: list[str] = [] + findings: list[CoherenceFinding] = [] groups = _exclusive_groups(decisions, edges) back = _back_edges(edges) incoming: dict[int, list[EdgeSpec]] = {} @@ -483,22 +589,24 @@ def check_step_arity(nodes: tuple[StepSpec, ...], edges: tuple[EdgeSpec, ...], for n in nodes: if len(n.inputs) > 1: names = ", ".join(v.name for v in n.inputs) - findings.append( + findings.append(CoherenceFinding( f"node {n.name!r} declares {len(n.inputs)} inputs ({names}), but a pydantic-graph " f"step body receives exactly one value — there is no invocation in which both " f"arrive. Combining two arrivals is what a join is for; a step cannot express it, " - f"and the declaration reads as though it can.") + f"and the declaration reads as though it can.", + check="check_step_arity", about=n.name)) arrivals = incoming.get(id(n), []) if len(arrivals) > 1 and _mutually_exclusive(arrivals, groups): continue if len(arrivals) > 1: froms = ", ".join(sorted(_name(e.source) for e in arrivals)) - findings.append( + findings.append(CoherenceFinding( f"node {n.name!r} is fed by {len(arrivals)} edges ({froms}), so it is invoked " f"once PER EDGE with one value each time, and all but one result is discarded. " f"Measured on exactly this shape: the step ran twice and the graph returned only " - f"the first. If the intent is to combine them, this is a join, not a step.") + f"the first. If the intent is to combine them, this is a join, not a step.", + check="check_step_arity", about=n.name)) return findings @@ -572,7 +680,7 @@ def _produces(annotation: Any, declared: Any) -> bool | None: return None # generic aliases, TypeVars, exotic forms -def check_variable_types(parent: Any, strategy: StrategySpec) -> list[str]: +def check_variable_types(parent: Any, strategy: StrategySpec) -> list[CoherenceFinding]: """Each implementation returns the type its role is declared to produce. ⛔ WHY THIS EXISTS, measured before it was written: @@ -580,10 +688,10 @@ def check_variable_types(parent: Any, strategy: StrategySpec) -> list[str]: wrong = StepSpec("wrong", inputs=(text,), outputs=(number,)) # declares int async def returns_a_string(ctx) -> str: ... # returns str - check() -> clean + coherence_check() -> clean run('x') -> "got 'not an int: x' (str)" - Nothing objected — not `check()`, not `build(validate_graph_structure=True)`, not the run. + Nothing objected — not `coherence_check()`, not `build(validate_graph_structure=True)`, not the run. ⚠️ And this gap is WORSE here than in raw pydantic-graph, which is the uncomfortable part. Their API never asks you to write the type down, so it promises nothing. This library invites @@ -599,7 +707,7 @@ async def returns_a_string(ctx) -> str: ... # returns str check that emits a finding per unannotated function is noise, and noise is how a check stops being read. """ - findings: list[str] = [] + findings: list[CoherenceFinding] = [] unchecked: list[str] = [] for node in parent.nodes: @@ -621,18 +729,23 @@ async def returns_a_string(ctx) -> str: ... # returns str unchecked.append( f"{node.name} ({_type_name(annotation)} vs {_type_name(declared)}: not decidable)") elif verdict is False: - findings.append( + findings.append(CoherenceFinding( f"{strategy.name!r} binds {node.name!r} to " f"{getattr(impl, '__qualname__', impl)}, which returns " f"{_type_name(annotation)} — but {node.name!r} is declared to produce " f"{_type_name(declared)}. The declaration is what the diagram draws and what a " - f"reader of this design believes; one of the two is wrong.") + f"reader of this design believes; one of the two is wrong.", + check="check_variable_types", about=_about(node))) + # ⚠️ `about=""` because this one line covers SEVERAL nodes — which is the whole reason it is + # aggregated. Naming one of them would be a worse answer than naming none; the node names are + # in the text, where a reader needs them and a filter cannot be misled by them. if unchecked: - findings.append( + findings.append(CoherenceFinding( "NOT CHECKED — return types were not compared for: " + "; ".join(sorted(unchecked)) + ". An unannotated or unresolvable implementation cannot be checked against its " - "declared output, and saying nothing would make that look like a pass.") + "declared output, and saying nothing would make that look like a pass.", + check="check_variable_types")) return findings @@ -673,7 +786,8 @@ def _bindable_name(spec: Any) -> str: return f"node {getattr(spec, 'name', spec)!r}" -def check_transform_edges(edges: tuple[EdgeSpec, ...], strategy: StrategySpec | None) -> list[str]: +def check_transform_edges(edges: tuple[EdgeSpec, ...], + strategy: StrategySpec | None) -> list[CoherenceFinding]: """A transform edge is fixed (`apply=`) or a variation point (bound) — exactly one. ⛔ Neither is a silently missing transform: the value would cross unchanged while the @@ -687,37 +801,41 @@ def check_transform_edges(edges: tuple[EdgeSpec, ...], strategy: StrategySpec | coroutine object and warns "never awaited". A value that is a coroutine flows on to the next step and fails there, attributed to the wrong place. """ - findings: list[str] = [] + findings: list[CoherenceFinding] = [] for e in edges: if not isinstance(e, TransformEdgeSpec): continue bound = strategy is not None and e in strategy.bindings if e.apply is not None and bound: - findings.append( + findings.append(CoherenceFinding( f"{e!r} declares `apply=` AND is bound by strategy {strategy.name!r}. Exactly one " - f"— otherwise which of the two runs is a coin toss.") + f"— otherwise which of the two runs is a coin toss.", + check="check_transform_edges", about=_about_edge(e))) if e.apply is None and strategy is not None and not bound: - findings.append( + findings.append(CoherenceFinding( f"{e!r} has no `apply=` and no binding, so nothing reshapes the value. It would " f"cross unchanged while the declaration says it becomes " - f"{e.delivers.name!r} — a lie the diagram would repeat.") + f"{e.delivers.name!r} — a lie the diagram would repeat.", + check="check_transform_edges", about=_about_edge(e))) fn = e.apply if e.apply is not None else (strategy[e] if bound else None) if fn is not None and inspect.iscoroutinefunction(fn): - findings.append( + findings.append(CoherenceFinding( f"{e!r} is bound to an ASYNC function. A transform runs on the wire and cannot " f"await — pydantic-graph would not reject it, it would quietly pass a coroutine " f"object to the next step. If it needs to await, it is a stage: give it a " - f"StepSpec.") + f"StepSpec.", + check="check_transform_edges", about=_about_edge(e))) return findings -def check_fan_out_rejoins(nodes: tuple[NodeSpec, ...], edges: tuple[EdgeSpec, ...]) -> list[str]: +def check_fan_out_rejoins(nodes: tuple[NodeSpec, ...], + edges: tuple[EdgeSpec, ...]) -> list[CoherenceFinding]: """Everything a fan-out produces must reach a join before it reaches END. ⛔ THE MIRROR OF `check_step_arity`, and it was missing. Measured on a three-item shopping list with `map -> price -> END` and no join: - check() -> clean + coherence_check() -> clean run -> 1.2 price ran 3 times, with ['milk', 'eggs', 'bread'] @@ -734,7 +852,7 @@ def check_fan_out_rejoins(nodes: tuple[NodeSpec, ...], edges: tuple[EdgeSpec, .. `map -> a -> b -> join -> END` is fine, and so is a branch, as long as every path from the fanned target to END passes through one. """ - findings: list[str] = [] + findings: list[CoherenceFinding] = [] fans = [e for e in edges if isinstance(e, MapEdgeSpec)] if not fans: return findings @@ -760,10 +878,11 @@ def reaches_end_without_a_join(start: Any) -> bool: for e in fans: if reaches_end_without_a_join(e.target): - findings.append( + findings.append(CoherenceFinding( f"edge {e!r} fans out, but a path from {_name(e.target)!r} reaches END without " f"passing a join. Every item produces its own result and a step cannot merge " f"them, so all but one are discarded — silently, with the right answer's shape. " f"Add a JoinSpec: a reducer `(current, input) -> current` is the only thing that " - f"can put them back together.") + f"can put them back together.", + check="check_fan_out_rejoins", about=_about_edge(e))) return findings diff --git a/workflow_workbench/devserver.py b/workflow_workbench/devserver.py index 4eccba3..071df70 100644 --- a/workflow_workbench/devserver.py +++ b/workflow_workbench/devserver.py @@ -19,6 +19,7 @@ import inspect from typing import Any +from workflow_workbench.checks import blocking from workflow_workbench.diagram import impl_name from workflow_workbench.graph_spec import GraphSpec from workflow_workbench.spec import StrategySpec, SubgraphBinding, is_sentinel @@ -118,12 +119,12 @@ def spec_payload(spec: GraphSpec, strategies: list[StrategySpec]) -> dict[str, A "unbound": False, **_source_of(impl), } - findings = spec.check(s) + findings = spec.coherence_check(s) layers.append({ "name": s.name, "bindings": bindings, "findings": findings, - "ok": not [f for f in findings if not f.startswith("NOT CHECKED")], + "ok": not blocking(findings), }) return { @@ -133,7 +134,7 @@ def spec_payload(spec: GraphSpec, strategies: list[StrategySpec]) -> dict[str, A "nodes": nodes, "edges": edges, "layers": layers, - "design_findings": spec.check(), + "design_findings": spec.coherence_check(), "mermaid": spec.diagram(), } diff --git a/workflow_workbench/evals.py b/workflow_workbench/evals.py index c6502ba..e37ac82 100644 --- a/workflow_workbench/evals.py +++ b/workflow_workbench/evals.py @@ -24,6 +24,7 @@ from dataclasses import dataclass, field from typing import Any +from workflow_workbench.checks import blocking from workflow_workbench.graph_spec import GraphSpec from workflow_workbench.spec import SpecError, StrategySpec @@ -108,7 +109,7 @@ def eval_battle(spec: GraphSpec, strategy_a: StrategySpec, strategy_b: StrategyS # never silently reported as though two different things were compared. pass for s in (strategy_a, strategy_b): - findings = [f for f in spec.check(s) if not f.startswith("NOT CHECKED")] + findings = blocking(spec.coherence_check(s)) if findings: raise SpecError(f"strategy {s.name!r} does not satisfy " f"{spec.name or type(spec).__name__}:\n " + "\n ".join(findings)) diff --git a/workflow_workbench/graph_spec.py b/workflow_workbench/graph_spec.py index 835ac31..072edca 100644 --- a/workflow_workbench/graph_spec.py +++ b/workflow_workbench/graph_spec.py @@ -13,6 +13,7 @@ from pydantic_graph import GraphBuilder from workflow_workbench import checks +from workflow_workbench.checks import CoherenceFinding, blocking from workflow_workbench.diagram import diagram as _diagram, diff_diagram as _diff_diagram from workflow_workbench.spec import ( Bindable, @@ -45,7 +46,7 @@ class Extraction(GraphSpec): EdgeSpec(source=load, target=extract, carries=raw_text), EdgeSpec(source=extract, target=END)) - The DAG is DATA, not code — which is what lets `check()` and `diagram()` run with zero + The DAG is DATA, not code — which is what lets `coherence_check()` and `diagram()` run with zero implementations and no engine. ⛔ There is exactly ONE way a graph comes into existence here: `edges` is compiled by `_wire`. @@ -84,7 +85,7 @@ class Extraction(GraphSpec): # ── checking ──────────────────────────────────────────────────────────────────────────── - def check(self, strategy: StrategySpec | None = None) -> list[str]: + def coherence_check(self, strategy: StrategySpec | None = None) -> list[CoherenceFinding]: """Every applicable check, as findings. Never raises. With no strategy: the design's own coherence — names, reachability, variables. Usable the @@ -95,14 +96,18 @@ def check(self, strategy: StrategySpec | None = None) -> list[str]: ⚠️ One method, not two. There is no "check the design" / "check the strategy" split because the no-strategy case needs no placeholder graph — the declaration is already data. - Recursion into subgraph bindings lives in `_check`, so nested designs can carry an + Recursion into subgraph bindings lives in `_coherence_check`, so nested designs can carry an ancestry path without that bookkeeping showing up in the public signature. + + ⚠️ Each finding is a `CoherenceFinding` — still a `str`, and still the same sentence, with + `check`, `about` and `blocking` on it so a caller can branch on structure instead of + matching on text. `checks.blocking(spec.coherence_check(s))` is the filter `render()` itself uses. """ - return self._check(strategy, ancestry=()) + return self._coherence_check(strategy, ancestry=()) - def _check(self, strategy: StrategySpec | None, - *, ancestry: tuple[tuple[type, int], ...]) -> list[str]: - """`check()`, plus the path of (design, strategy) pairs already open above this one. + def _coherence_check(self, strategy: StrategySpec | None, + *, ancestry: tuple[tuple[type, int], ...]) -> list[CoherenceFinding]: + """`coherence_check()`, plus the path of (design, strategy) pairs already open above this one. ⚠️ This is the ONE place a cycle is detected. `check_subgraphs` deliberately does not also check — two owners of one rule is how a chain ends up either reported twice or, worse, @@ -114,18 +119,24 @@ def _check(self, strategy: StrategySpec | None, """ key = (type(self), id(strategy)) if strategy is not None else None if key is not None and key in ancestry: - return [ + # ⚠️ `check="GraphSpec._coherence_check"` names the producing function, like every other + # finding — and here that is honestly not a `check_*` in `checks.py`. Cycle detection + # needs the `ancestry` only this method carries, which is why it lives here and why + # `check_subgraphs` deliberately does not also do it. `about` is the STRATEGY: the + # design is fine, and swapping the strategy is the move. + return [CoherenceFinding( f"recursive subgraph binding: {self.name or type(self).__name__!r} with strategy " f"{strategy.name!r} appears inside its own subgraph chain. Rendering it would " f"build child graphs until the stack ran out — a design cannot implement one of " - f"its own nodes with itself."] + f"its own nodes with itself.", + check="GraphSpec._coherence_check", about=strategy.name)] # ⚠️ `nodes` is STEPS ONLY — the roles a strategy fills. `NodeSpec` is every declared # box. Conflating the two is how a join ends up demanding an implementation, or an # unreachable join goes unreported. Both axes now have a name, so the annotations below # are true rather than merely conventional. declared: tuple[NodeSpec, ...] = (*self.nodes, *self.joins, *self.decisions) - findings = list(checks.check_names(declared)) + findings: list[CoherenceFinding] = list(checks.check_names(declared)) findings += checks.check_variables(declared, self.edges) findings += checks.check_decisions(self.decisions, self.edges) findings += checks.check_step_arity(self.nodes, self.edges, @@ -280,8 +291,8 @@ def render(self, strategy: StrategySpec) -> Any: built graph. Nothing else should compare two bare `Graph` objects — there is no way to tell whether they came from one design. """ - findings = self.check(strategy) - hard = [f for f in findings if not f.startswith("NOT CHECKED")] + findings = self.coherence_check(strategy) + hard = blocking(findings) if hard: raise SpecError( f"{self.name or type(self).__name__} cannot be rendered with strategy "