diff --git a/.conceptlint-baseline b/.conceptlint-baseline index cf39f3b..4bdd30e 100644 --- a/.conceptlint-baseline +++ b/.conceptlint-baseline @@ -1,14 +1,15 @@ -# Findings in plan_types/ — the package this repo now ships. CI fails only if this rises. +# Findings across the WHOLE REPO. CI fails only if this rises. # -# ⚠️ WAS 4, POINTED AT THE WRONG DIRECTORY. The baseline counted duplicates between -# `conceptlint/dataflow/` and `conceptlint/ontologies/pplan/concepts.py` — Step, Plan, Activity and -# Entity each declared twice, once as a P-Plan Concept and once as the implementation. +# ⚠️ WAS `plan_types/` ONLY, AND THAT IS HOW A DUPLICATE SHIPPED (2026-08-20). +# `Square` was declared twice with different shapes, in examples/pydantic_graph_docs/ — outside the +# gate — and sat there unreported until someone asked whether the linter was earning its place. The +# narrowing was deliberate, to dodge the intentional before/after pairs in evals/minimal/, and it +# silently took examples/ and tests/ with it. # -# The PlanTypes refactor RESOLVED all four: `Concept` is gone, and Plan/Step/Variable exist once, -# in plan_types/plan/. The duplicate the tool reported about itself is the duplicate the redesign -# removed — which is the outcome, arriving quietly as a number going down. +# This file already carried that exact warning about a DIFFERENT directory: "a gate pointed at the +# wrong directory reads exactly like a clean codebase." It happened again, one directory over. # -# But CI was still linting `conceptlint/`, now a near-empty shell, and would have reported 0 -# forever whatever landed in plan_types/. A gate pointed at the wrong directory reads exactly like -# a clean codebase. -0 +# So: lint `.`, and carry the known-deliberate duplicates in the number rather than by excluding a +# path. The six are 4 ontology-vs-implementation pairs (Plan, Step, Activity, Entity) and the 2 +# before/after pairs in evals/minimal/sibling_refinement, which exist to BE duplicates. +6 diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 1ce4b97..4b9d838 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -37,10 +37,13 @@ jobs: # Dogfood. If this tool cannot survive being pointed at its own source, it has no business # being pointed at anyone else's. # - # `conceptlint/` and not `.` on purpose: evals/minimal/ holds deliberate before/after pairs, - # so linting them measures the fixtures rather than the code. Measured — 5 findings against - # `.`, 3 against the package, and both extras were eval twins. - - name: lint plan_types with itself + # `.` — the WHOLE repo. This said "`conceptlint/` and not `.` on purpose" until 2026-08-20, + # narrowed so evals/minimal/'s deliberate before/after pairs would not be counted. That + # narrowing also excluded examples/ and tests/, and a duplicate `Square` sat in + # examples/pydantic_graph_docs/ unreported until someone asked whether the linter earned its + # place. The intentional pairs are carried IN the baseline number instead, where they are + # visible, rather than by excluding a path, where the exclusion is invisible. + - name: lint the repo with itself run: | BASELINE=$(grep -vE '^\s*(#|$)' .conceptlint-baseline | head -1) @@ -50,9 +53,14 @@ jobs: # --import, or `declared()` returns [] and grounded-citation checks NOTHING while # reporting green. Registration is subclassing, so a Concept in an unimported # module does not exist — measured: without this the concept list was empty. - OUT=$(uv run conceptlint plan_types/ 2>&1); RC=$? - set -e + OUT=$(uv run conceptlint . 2>&1); RC=$? + # ⚠️ `set -e` stays OFF across the count. `grep -c` exits 1 when it counts ZERO, so with + # `bash -e` this line killed the job before its first echo — the gate went red precisely + # when plan_types/ was clean, and had done since the baseline reached 0 on 2026-08-17. + # Three red runs on main, none of them about the code. Still not `|| true`: that is the + # nobsmed failure the comment above names, where a usage message counted as 0 and passed. FOUND=$(printf '%s\n' "$OUT" | grep -cE '^(naming|typing|topology|provenance)\.[a-z_]+:') + set -e if [ "$RC" -ne 0 ] && [ "$FOUND" -eq 0 ]; then echo "::error::conceptlint failed to run — this gate was measuring nothing." printf '%s\n' "$OUT" | tail -20 diff --git a/README.md b/README.md index 145c690..939690b 100644 --- a/README.md +++ b/README.md @@ -1,66 +1,130 @@ # PlanTypes -**From Claude Code Plan Mode to Plan Types.** +**Declarative, typed workflow plans, separated from execution.** -Turn an agent's plan into a typed process specification — one you can validate, draw, and hold it to. +LangGraph, Temporal and [Pydantic Graph](https://pydantic.dev/docs/ai/graph/builder/) *execute* a +workflow, and each is good at it. This is the layer above: the workflow plan you settle — and can +validate, draw and argue about — **before** you pick an engine, or instead of picking one, since +plenty of workflows never need retries or durability. -Claude Code's Plan Mode says *"here's what I intend to do."* It's prose, it's gone when the conversation moves on, -and nothing checks that the code matches it. PlanTypes makes the same intent an artifact: +## The plan exists before the code does +That is the whole claim, and it is one snippet: + +```python +document = Variable("document", Document) +outline = Variable("outline", Outline) +summary = Variable("summary", Summary) + +class MakeOutline(Step): + inputs, outputs = (document,), (outline,) + +class Summarize(Step): + inputs, outputs = (document, outline), (summary,) # fans in — needs both + +plan = Plan(name="summarize_document", steps=(MakeOutline(), Summarize()), + declared_inputs=(document,)) + +validate(plan, [*topology.ALL, *typing.ALL]) # → [] +print(render_mermaid(plan)) +print(plan.shape()) # → ((Document,), (Summary,)) ``` -Plan -├── typed Steps -├── typed Variables -├── explicit dependencies -├── invariants -└── visualization + +```mermaid +flowchart TD + IN_document(["document: Document"]) + summarize_document_0["Make Outline"] + summarize_document_1["Summarize"] + OUT_summary(["summary: Summary"]) + IN_document -- document --> summarize_document_0 + IN_document -- document --> summarize_document_1 + summarize_document_0 -- outline --> summarize_document_1 + summarize_document_1 --> OUT_summary + classDef port fill:#fff,stroke:#333,stroke-width:1px,color:#333; + class IN_document,OUT_summary port; ``` -> **Plan Mode describes intent. PlanTypes makes the plan typed, inspectable, and testable.** +**Not one line of that workflow is implemented.** No prompt, no model, no function body, no runtime. +The process is checked, drawn and type-checked anyway — because a Step declares *what a +transformation is*, and nothing more. -This is not a replacement for Claude Code or Cursor. It's the thing their plans should produce. +Every arrow is read from the bindings, so adding an input changes the picture with no edit to the +renderer. A hand-drawn diagram is a claim about the code that stops being true the moment a Step +moves, and nothing tells you. -## 60 seconds +## How it is performed comes later, and is chosen per run ```python -from plan_types import Plan, Step, Variable, render_mermaid, validate -from plan_types.invariants import topology, typing +def summarize_fast(document: Document, outline: Outline) -> Summary: ... +def summarize_precise(document: Document, outline: Outline) -> Summary: ... -PAPER = Variable("paper", ClinicalStudy) -FINDINGS = Variable("findings", list[Finding]) -SUMMARY = Variable("summary", str) +fast = {MakeOutline: outline_by_sentence, Summarize: summarize_fast} +precise = {MakeOutline: outline_by_sentence, Summarize: summarize_precise} -class Extract(Step): - inputs, outputs = (PAPER,), (FINDINGS,) +execute(plan, {"document": doc}, LocalRunner(fast)) +execute(plan, {"document": doc}, LocalRunner(precise)) +``` -class Summarize(Step): - inputs, outputs = (PAPER, FINDINGS), (SUMMARY,) # fans in — needs both +**The `plan` object is not touched between those two lines.** An experiment can therefore say *the +logical process was held constant, only the implementation of `Summarize` changed* — and mean it, +because the same declaration served both arms. -plan = Plan( - name="extract_and_summarize", - steps=(Extract(), Summarize()), - declared_inputs=(PAPER,), # what the Plan expects to be handed -) +The alternative is to declare `SummarizeV1` and `SummarizeV2` as separate Steps, and that is not a +workaround: it is two names for one concept, the `naming.naming_drift` this package reports. -validate(plan, [*topology.ALL, *typing.ALL]) # → [] -print(render_mermaid(plan)) +## How is this different from Pydantic Graph? + +It is not a competitor and does not want to be one. Pydantic Graph already owns steps, typed +execution, edges, branching, `map`, `join`, reducers, state, dependency injection, execution and +rendering — all of it well. `plan_types.execution.pydantic_graph` **compiles a Plan onto it**, using +its real primitives, and `tests/test_pydantic_graph.py` asserts both runtimes return the same +answer. + +The difference is what a Step *is*. Theirs is executable — the function is both the node and the +implementation. Ours is a declaration, so several implementations are peers rather than one being +privileged and the rest overrides. + +That shows up in the diagram. Same workflow — their `parallel_processing.py`, `map` → `square` → +join → total — rendered by each: + +``` +THEIRS — graph.render() OURS — render_mermaid(plan) + +stateDiagram-v2 flowchart TD + state map <> IN_numbers(["numbers: list"]) + square Square + state reduce_list_append <> Total + total OUT_total(["total: int"]) + + [*] --> map IN_numbers -- numbers --> Square + map --> square Square -- squares --> Total + square --> reduce_list_append Total --> OUT_total + reduce_list_append --> total + total --> [*] ``` -```mermaid -flowchart TD - IN_paper(["paper: ClinicalStudy"]) - s0["Extract"] - s1["Summarize"] - OUT_summary(["summary: str"]) - IN_paper -- paper --> s0 - IN_paper -- paper --> s1 - s0 -- findings --> s1 - s1 --> OUT_summary -``` - -Every arrow is read from the bindings. Add an input and the picture changes with no edit to the -renderer — a hand-drawn diagram is a claim about the code that stops being true the moment a Step -moves, and nothing tells you. +Two real differences, and neither is a rendering gap: + +- **Theirs has no types on its edges.** It cannot: an edge carries whatever the function returned, + and there is no name for it. Ours labels every edge with the Variable that flows. +- **Theirs draws the machinery** — `map <>` and `reduce_list_append <>` are nodes. Ours + draws the process; the fan-out is a property of `Square` and does not appear. Their picture + answers *how will this execute*, ours answers *what is this workflow*. Both are correct. + +And the fan-out is one declaration on the Step, in their vocabulary: + +```python +class Square(Step): + inputs, outputs = (number,), (squared,) # int -> int, exactly their `square` + map_over = (numbers, squares) # list[int] in, list[int] out +``` + +Under `LocalRunner` that is a sequential loop. Compiled, it is their real `.map()` and +`join(reduce_list_append)`. **Same declaration, nothing edited.** Their documented output, +`[1, 4, 9, 16, 25]`, is asserted on both. + +**→ [`examples/pydantic_graph_docs/`](examples/pydantic_graph_docs/)** — their own docs examples, +copied verbatim, run through this layer, with a control arm that uses no Plan at all. ## Why this exists @@ -135,22 +199,40 @@ hidden the fact that the signature was wrong. ## The logical process first, the runtime later — or never +Four layers, and each one only knows about the one above it: + ``` -domain types - ↓ -Plan / Step / Variable - ↓ -invariants - ↓ -visualization - ↓ -optional execution adapters ── plain Python │ Temporal │ LangGraph +WHAT EXISTS Variable ── typed slot ┐ + Step ── one operation │ plan-time + Plan ── how Steps compose │ imports nothing else + Service ── what must be up ┘ + +HOW IT IS PERFORMED Strategy ── {Step: implementation} chosen per execution + several per Step, none privileged + +HOW IT IS RUN StepRunner ── Protocol LocalRunner ── here + Pydantic Graph ── here, optional extra + Temporal ── not built + LangGraph ── not built + +WHAT ACTUALLY RAN prov:Activity, prov:Entity ⚠️ NOT modelled by this package ``` -Most workflow systems fuse process design with orchestration semantics from the first line. PlanTypes -separates them, and the separation is the point: simpler debugging, fewer irrelevant runtime -concerns, and a specification a coding agent can change safely. Plenty of processes never need -retries or durability at all. +The arrows only point down. `plan_types.plan` imports nothing from `plan_types.execution`, which is +what lets a specification be read, validated and drawn with no execution backend in the room — and +`scripts/check_wheel.py` imports both from a built wheel outside the source tree, because a layering +claim that only holds in an editable install is not a layering claim. + +⚠️ **The bottom row is deliberately absent.** `p-plan:Step` is the intended operation; +`prov:Activity` is one execution of it. A runtime may map them one to one, and the moment code +believes that, *"the definition is wrong"* and *"that run failed"* become the same sentence with +opposite fixes. + +Most workflow systems fuse process design with orchestration semantics from the first line. +PlanTypes separates them, and the separation is the point: simpler debugging, fewer irrelevant +runtime concerns, and a specification a coding agent can change safely. Plenty of processes never +need retries or durability at all — `LocalRunner` is sequential, in-process, and has no retries by +decision, not by omission. ⚠️ **A Plan is not a DAG.** A Plan *may* be acyclic — that's `topology.acyclic`, an invariant you opt into. Building it into the type would rule out iterative processes before anyone asked for one, @@ -208,26 +290,90 @@ cited IRI names a real term. That rule exists because this package once cited `p memory and implemented something P-Plan doesn't describe. A citation nobody can follow is decoration with the authority of a fact. +### A borrowed vocabulary, so a Plan can be handed over + +`p-plan:Step` has a definition we did not write, at an address anyone can resolve — so the Plan a +product owner argues about in a PRD is the same object an engineer compiles onto Pydantic Graph, and +anything else speaking P-Plan (or a model asked for one) means the same thing by `Step`. + +⚠️ A door, not a feature: there is **no RDF import or export** yet. + ## Status — honest -**Works today:** typed Plans, the four invariant categories, `render_mermaid`, P-Plan/PROV-O -grounding with vendored ontologies, 116 tests. +**Works today:** typed Plans, the four invariant categories, `render_mermaid` (with an optional +Strategy overlay), P-Plan/PROV-O grounding with vendored ontologies, `Strategy` + `check_strategy`, +`LocalRunner`, `map_over` fan-out, and `to_pydantic_graph` — which emits Pydantic Graph's real +`.map()` and `join`, not a loop wearing the name. 158 tests. + +### What it does NOT do, stated plainly -**Not built:** execution adapters (Temporal, LangGraph), persistence, retries, scheduling, embedding -based similarity, and agent-hook integration. The pattern above describes what the artifact is -*for*; the hooks that would put it in an agent's loop are not wired yet. +These are the first four objections anyone will raise, so they are answered here rather than waited +for: -Not on PyPI. From source: +| | | +|---|---| +| **cycles do not run** | A Plan may BE cyclic — it constructs, wires, renders, and `topology.acyclic` reports it. But it cannot execute, and the reason is not the toposort: **a Plan carries no termination predicate**. Their `Feedback.run()` returns `WriteEmail \| End[Email]` and the predicate lives in the node body, which is the right place for it and is theirs. `MultiStep.until` declares an iterative region; collapsing a strongly connected component into one is designed, not built. | +| **`Fork` / `Join` are not first-class** | Only `map_over`. Pydantic Graph 2.x has both as concepts; a Plan can currently declare a fan-out over a list and nothing else. | +| **no async runner** | `LocalRunner` is synchronous by decision and REFUSES an `async def` rather than returning an un-awaited coroutine. A compiled graph awaits it fine. | +| **at one arm, this is overhead** | Measured, not conceded reluctantly: with a single implementation and two steps, a Plan costs a declaration and buys nothing. It starts paying when there is a second arm, or a second reader. | + +**Not built:** RDF import/export (no Turtle, no JSON-LD, no `rdflib` — the P-Plan grounding is a +checked vocabulary, not a serialization format), Temporal and LangGraph adapters, persistence, +retries, scheduling, concurrency, embedding-based similarity, agent-hook integration. + +⚠️ **And the honest scale caveat.** Everything demonstrable here is small. The failure this is built +for shows up at volume: one codebase downstream has **18 variants of one extraction pipeline, 23 +distinct step names, 47% of them used exactly once**, with `enrich` and `enrich_one_finding` +coexisting in the same file. A toy cannot show that, and this README will not pretend the toy does. + +## Optional: interrupt the agent before it writes a duplicate — EXPERIMENTAL + +A linter reports after the fact. `conceptlint/integrations/pre_write.py` is a Claude Code +`PreToolUse` hook that asks the question **before** the file is written: -```bash -git clone https://github.com/borisdev/plan-types && cd plan-types && uv sync -uv run pytest -q +``` +⛔ `EvidenceLookup` looks like `EvidenceSearch`, which already exists at + libs/.../find_evidence.py:86 + Reuse it, or say what distinction `EvidenceLookup` carries that it does not. ``` -## Open question, deliberately +Install into `~/.claude/settings.json`, merging with any hooks already there: + +```json +{"hooks": {"PreToolUse": [{"matcher": "Write|Edit", + "hooks": [{"type": "command", + "command": "/bin/python3 -m conceptlint.integrations.pre_write 2>/dev/null || true"}]}]}} +``` + +Use the interpreter of an environment where `conceptlint` and `plan_types` are importable — a bare +`python3` will not find them. It reads the repo root from the hook payload's `cwd`, so one install +covers every project. + +### ⚠️ Four things it does not do, measured rather than assumed + +- **It only sees NEW Pydantic models.** A new `Step` subclass produces nothing — verified by piping + one in. Same root cause as the discovery gap in `naming/records.py`: a class is only recognised if + its base was already found. +- **It cannot interrupt in an auto-accept permission mode.** It returns `permissionDecision: "ask"`, + which `default` mode turns into a prompt and permissive modes answer for you. The check still + runs; nobody reads the answer. +- **It is silent on almost every write, by design.** A hook that comments on every edit is removed + within a day, and then the useful interrupt is gone too. +- **It fails open on everything** — unparseable source, missing import, timeout. A convenience must + never be able to block work. + +## Settled: PlanTypes is the product Are semantic invariants the general product, with PlanTypes as the first ontology-grounded use case — or is PlanTypes the product, with invariants as a subsystem inside it? -Not decided. The code is arranged so invariants *could* split into a standalone package later, and -that split has not been made. Worth arguing about in the open rather than settling early. +**Decided 2026-08-20: PlanTypes is the product.** + +Not because the first framing is wrong, but because it cannot be argued yet. A general +semantic-invariant product needs a corpus of trial and error that does not exist, and it is hard to +grasp before it is demonstrated — which is a bad combination for the thing you lead with. *"Settle +the workflow plan before the execution details"* is one sentence, and it works today. + +So `conceptlint` stays, as **a supporting semantic linter rather than a second product**. The code +is still arranged so it could split into its own package, and that option is worth keeping. What +changes is which one gets explained first, and which one the repo is named after. diff --git a/docs/handoffs-consolidated.md b/docs/handoffs-consolidated.md new file mode 100644 index 0000000..8ecd1e2 --- /dev/null +++ b/docs/handoffs-consolidated.md @@ -0,0 +1,293 @@ +# A Step declares, a Strategy implements, a Runner executes + +Four ChatGPT handoffs about this repo, consolidated. They converge on one decision, contradict each +other in three places, and one of them contradicts this repo's own vocabulary. This doc is the +decision, the disagreements resolved, and what was measured rather than argued. + +--- + +## The decision, in code + +```diff + class Step(Generic[InputT, OutputT]): + inputs: ClassVar[tuple[Variable[Any], ...]] = () + outputs: ClassVar[tuple[Variable[Any], ...]] = () + +- def run(self, **values: Any) -> Any: +- """Execute. Keyword arguments are the input Variables, by name.""" +- raise NotImplementedError +``` + +```python +# how it is performed — outside the declaration, chosen per execution +fast = {MakeOutline: outline_by_sentence, Summarize: summarize_fast} +precise = {MakeOutline: outline_by_sentence, Summarize: summarize_precise} + +execute(plan, {"document": doc}, LocalRunner(fast)) +execute(plan, {"document": doc}, LocalRunner(precise)) +``` + +**The `plan` object is not touched between those two lines.** + +--- + +## Why — the argument that actually carries it + +The weak argument is YAGNI: nothing called `run()`, so delete it. True, and it sounds like tidying. + +The real argument is that **the signature was a false claim about the domain**: + +```python +def run(self, **values: Any) -> Any: # the implementation is A METHOD ON THIS CLASS + raise NotImplementedError # so there is one, and any second is an override +``` + +`ExtractClaims` is one stable operation with several ways to perform it — a cheap model, an +expensive one, an ensemble — and none of them is the *real* one that the others override. A method +on the class privileges whichever implementation got there first. + +The alternative people reach for when the method is in the way is worse: + +```python +class ExtractClaimsV1(Step): ... # not three operations. +class ExtractClaimsV2(Step): ... # three names for one concept — +class ExtractClaimsEnsemble(Step): ... # `naming.naming_drift`, committed inside the + # package built to report it. +``` + +So the implementation moves out of the declaration, and *which* implementation runs becomes a +property of an execution rather than of the process. + +--- + +## ⚠️ One word was already taken, and three of the handoffs used it wrong + +Every handoff calls the Step→implementation mapping **"bindings"**. This repo already has that word: + + binding how VARIABLES connect STEPS plan_types/plan/bindings.py — producers, consumers, + edges, execution_order + ??? which IMPLEMENTATION performs the new thing + a Step + +Two concepts, one word, in one package. That is `naming.ambiguous_reference` — *one name, many +concepts* — and it would have shipped inside the module written to catch it. + +**Shipped as `Strategy`**, which is the handoffs' own gloss (*"different strategies for performing +the same operation"*) and is what makes `plan_types/plan/bindings.py` still mean exactly what it +meant. One rename reverses it if you disagree; the thing that must not happen is the collision. + +Grounding, stated plainly: P-Plan and PROV-O have **no term** for *"the code that will perform this +Step if it runs."* `p-plan:Step` is the intended operation, `prov:Activity` is one execution of it, +and this is neither. So `Strategy` is ours, deliberately uncited — the same call as `Step.uses`, and +for the same reason. + +--- + +## Where the four handoffs disagree + +| question | the positions | resolution, and why | +|---|---|---| +| where the implementation attaches | **A**: `Step.run()`, "portable domain vocabulary" · **B**: not on Step, `bind()` + callable · **C**: `step.impl`, called by a Protocol · **D**: a `bindings` map, resolved by a runner | **D.** A privileges one implementation. C puts strategy back on the declarative graph, which C's own author warns against three paragraphs later. B and D are the same idea; D is the one with a worked eval-arm example. | +| sync or async | **B**: "PlanTypes should remain sync/async-neutral" · **C, D**: `async def run(...)` in the Protocol | **Sync.** An `async def` on the Protocol is not neutral — it forces every implementation of every Step to be a coroutine, including `lambda x: x + 1`. An `AsyncStepRunner` when a real async implementation exists. | +| how the implementation is called | **D**: `extract_claims_v1(paper: Paper)` — positional · **D, same page**: `impl(**inputs)` — keyword | **Keyword**, and `Variable.name` therefore becomes part of the contract. A Step with three inputs called positionally is one reorder away from a mis-wire the types cannot catch when two inputs share a type. `check_strategy` reports a parameter-name mismatch at declaration time. | +| repo layout | **A**: top-level `plan/` + `execution/` | Adopted as `plan_types/plan/` + `plan_types/execution/`, which is where `plan/` already was. | +| runtime state and deps | **A**: `EvidenceBuildState`, `NoBSmedDeps` injected per execution | **Not built.** No caller. Note that `plan_types/plan/service.py` is already the *plan-time* half — what a Step needs reachable — and `typing.plan_time_only` refuses the runtime half onto plan types by rule. | + +--- + +## Evidence — measured before the change, not argued after + +**Nothing called `run()`.** Not in `plan_types/`, `tests/`, `evals/` or `examples/`. + +**The only implementation of it in this repo did not import.** +`examples/evidence_case_graph/flow.py` still declared `consumes`/`produces`, retired 2026-08-16: + +``` +TypeError: ParseStudyStep.consumes is retired — use `inputs`, a TUPLE of Variables. +``` + +It had been dead since the P-Plan DAG correction, and **no test imported it**, so no check could +have caught that or the second fault on the line below — `def run(self, value: Study)`, positional, +against a base class whose docstring warns positional args silently mis-wire. An example nobody +imports is not an example, it is a claim. `tests/test_execution.py::test_examples_import_and_run` +now runs both examples end to end. + +**The type contradicted its own module.** `plan_types/plan/plan.py` has said this the whole time: + +> ⚠️ `run()` is not the centre of this … An execution adapter wraps Steps from outside. Never the +> reverse. + +**Downstream, 15 overrides that all raise.** nobsmed's `plans.py`, every one +`raise NotImplementedError(_DECLARATION_ONLY)`. + +--- + +## What shipped + +``` +plan_types/plan/step.py run() removed; `run` added to _RETIRED +plan_types/execution/strategy.py Strategy, Implementation, check_strategy +plan_types/execution/runner.py StepRunner Protocol — sync, runtime_checkable +plan_types/execution/local.py LocalRunner, execute, ExecutionError +examples/hello/flow.py NEW — domain-free, two arms, the README's 60 seconds +examples/evidence_case_graph/ rewritten to nobsmed's real fan-in shape, two arms +tests/test_execution.py NEW — 20 tests +scripts/check_wheel.py plan_types.execution added to REQUIRED +README.md 60 seconds, the four-layer diagram, Status +``` + +142 tests pass, up from 122. + +**`_RETIRED`, not deletion.** Deleting `run()` would let a subclass declare it again and get the +privileged implementation back with nothing to say so. The error names its replacement: + +``` +TypeError: AskModelForCausalMap.run is retired — use a Strategy. a Step DECLARES a +transformation; it does not perform one. A method here is one implementation privileged over +every other, so a second way of doing the same operation becomes an override rather than a peer. +``` + +⚠️ The old `__init_subclass__` built its message with `'Input' if old == 'consumes' else 'Output'`, +so retiring a **third** name would have explained it as being about `hasOutputVar`. Each retired +name now carries its own sentence. + +**What the runner refuses, loudly, rather than accepting quietly:** + +| | why | +|---|---| +| an `async def` implementation | returns a coroutine — truthy, with a repr, flowing onward as though it were data | +| a Step with 2 outputs whose impl returns 1 value | else one Variable holds the whole tuple and the mismatch surfaces three Steps later | +| a Step with 0 outputs whose impl returns something | discarding it hides a wrong declaration *or* a wrong implementation | +| a produced value of the wrong declared type | plain classes only — parameterized generics are reported **NOT CHECKED**, never as a pass | +| two Variables sharing a name | a Variable is `(name, type)`, so two can differ in type and collide in a name-keyed environment | + +--- + +## ⚠️ Breaking change for nobsmed — measured, and it corrects my own earlier note + +I previously wrote that nobsmed's 15 `run()` overrides *"still import and work as ordinary methods."* +**That is wrong.** It described deleting `run()`; what shipped retires it, which is stricter. Run +against the built wheel, using `plans.py`'s exact class shape: + +``` +BREAKS AT IMPORT: TypeError: AskModelForCausalMap.run is retired — use a Strategy... +without run(): Plan('arm', 1 steps, 2 variables) +``` + +**Contained by the pin.** nobsmed has `plan-types = { git = ..., tag = "v0.6.0" }` +(`pyproject.toml:68`), so nothing moves until that tag is bumped — deliberately, not by luck. + +On bump, one nobsmed commit fixes all of it: + +| site | what happens | fix | +|---|---|---| +| `plans.py` — 15 `def run(...)` overrides | **ImportError at module import** | delete all 15 | +| `plans.py` — `_DECLARATION_ONLY` | orphaned | delete | +| `test_plans.py:56-65` — asserts `run()` raises | tests a method that no longer exists | delete the test | +| `plan_diagram.py:190` — "Every `run()` raises" | stale prose | reword | +| anything **calling** `Step.run` | none exist — verified by grep | — | + +**A loud break is the right one here.** The quiet alternative leaves 15 methods in the file that a +reader reasonably takes as evidence that Steps execute — which is the thing that was false. And the +fix is mechanical: delete, do not rewrite. + +--- + +## Advice — where I would push back on the handoffs + +**"A dictionary plus a LocalRunner is sufficient to prove the architecture."** It proves the +architecture; it does not give it a consumer. nobsmed's builders are not decomposed into Steps +(`plans.py` says so and calls it separate work with its own risk), so **nothing in production binds +a Strategy yet.** The examples and tests are real; the production caller is not there. Worth +knowing before this reads as load-bearing. + +**Handoff D §1 asks to edit nobsmed's `plans.py`.** Different repo, different PR, and gated on the +tag bump above. + +**`Step(Generic[InputT, OutputT])` is still declared and is now entirely vestigial.** `InputT` and +`OutputT` are unused — the README already says that one-in-one-out signature "died to a real +builder." Left alone here to keep this diff about one thing; it should go. + +**Do not add the Temporal / LangGraph / Pydantic Graph adapters yet.** Each is a real dependency +bought to serve zero callers. The Protocol is three lines; an adapter can be written the day someone +has a workflow that needs durability. + +--- + +## Not built, and what would justify each + +- **`AsyncStepRunner`** — one real async implementation to run. Today the honest state is that async + is *refused*, not *supported*, and it says so at the call site. +- **Strategy per Step *instance*** — a Plan holding two instances of one Step class that need + different implementations. Keyed on the class today. +- **Return-type checking for generics** — `list[Finding]` cannot be tested by `isinstance`. Reported + NOT CHECKED. A real mis-wire that slipped through would justify a deeper check. +- **Runtime state / dependency injection** (`EvidenceBuildState`, `NoBSmedDeps`) — a Step + implementation that genuinely needs per-execution context beyond its closure. + +--- + +## ⚠️ The experiment corrected itself once, and that is the most useful thing in it + +The first stage 3 invented three *topology* variants — `Square → Total`, `Square → DropOutliers → +TotalKept`, `Square → DropOutliers → Weight → TotalWeighted` — to demonstrate structural diffing. +It produced three Step classes doing one job: + + class Total(Step): inputs, outputs = (squares,), (total,) + class TotalKept(Step): inputs, outputs = (kept,), (total,) + class TotalWeighted(Step): inputs, outputs = (weighted,), (total,) + +I read that as a design flaw — a Step's port binds to a Variable's identity, so moving a step means +a new class — and drafted a fix. **Boris read it correctly:** all three were one Plan, + + numbers ──> [transform] ──> transformed ──> [sum] ──> total + +with three implementations of the transform. Two Step classes, not six. The duplication was not the +design failing; it was **the naming rule reporting that I had drawn the Step boundaries in the wrong +place** — three names for one concept, appearing exactly where the mis-modelling was. + +Two things follow, and both are load-bearing: + +- **The proposed fix was cancelled.** Its motivating case evaporated. `docs/design.md` §6 — add the + second implementation first — applied to a change I was about to make on one bad example. +- **The invented variants were not in their docs at all.** `DropOutliers` and `Weight` are mine. + Stage 3 is now their actual `parallel_processing.py`: `.map()`, a join, and `reduce_list_append`, + in their vocabulary. + +The question the layer does NOT answer, and should not pretend to: *when is a composite a Step and +when is it an implementation?* The usable test is **can you evaluate the parts separately, and do +you need to** — "dropping outliers is what fixed `[1, 2, 50]`" requires `DropOutliers` to be a +declared Step, because you can only score what is declared. "Arm B beat arm A" does not. Choosing is +the modelling act; the Plan makes you write the choice down somewhere checkable. + +## ⚠️ A measurement retracted + +This section reported "5 lines with a Plan, 37 for the control". **The 37 was three copy-pasted +builders.** The obvious control is a parameterised one, which handles all three arms in twelve lines +and was verified to produce identical results. Measuring against code nobody would write is not a +measurement, and a reader who noticed would have been right to discard everything around it. + +There is no line-count claim here now. At this scale there is no volume advantage worth naming, and +saying so is worth more than the number was. + +What survives a fair control — checked, not asserted: + +- **The Plan validates and renders with NOTHING implemented.** `build(square_impl)` cannot produce a + graph, a diagram or a type check until an implementation exists. That is the "agree the process + first" claim, and it is the only one that does not shrink when the control is written properly. +- **Their diagram has no types on its edges**, because an edge carries whatever the function + returned and there is no name for it. Ours labels every edge with the Variable that flows. +- **Their diagram draws the machinery**: `map <>` and `reduce_list_append <>` are nodes + in it. Ours draws the process and does not mention them. + +And the honest other end: at one arm and two steps the Plan layer is pure overhead. + +## Deliberately out of scope + +The **A/B counterfactual experiment** from the process-spec handoff: fork an official Pydantic Graph +example, evolve it with Claude Code alone vs Claude Code + PlanTypes, same requirement changes to +both arms, and measure. That is GTM evidence, not this decision, and it deserves its own issue. + +`docs/counterfactual.md`'s rule applies to it in advance: **`no_effect` and `refutes` are results.** +The claim to test is *does a separate typed process specification help a coding agent preserve +coherence as requirements evolve* — not *PlanTypes makes Claude better.* diff --git a/examples/evidence_case_graph/flow.py b/examples/evidence_case_graph/flow.py index 054db4d..9322457 100644 --- a/examples/evidence_case_graph/flow.py +++ b/examples/evidence_case_graph/flow.py @@ -1,60 +1,115 @@ -"""The first real Plan: a study becomes findings, findings become an evidence graph. +"""The shape of a real builder: nobsmed's simple causal-map arm, declared and run. - Study -> ParseStudyStep -> Findings -> BuildEvidenceGraphStep -> EvidenceGraph + plan_text ──> AskModelForCausalMap ──> draft ──┐ + │ ├──> DropWhatThePasteDoesNotSupport ──> case_graph + └───────────────────────────────────────────┘ -⚠️ The EvidenceGraph produced here is the PRODUCT — the canonical IR for one case. The Plan above is -how it gets built. §10: do not collapse those two graphs. Mermaid, PDF and JSON are projections of -the EvidenceGraph, never of the Plan. +⚠️ **The fan-in is the point.** The grounding step needs the DRAFT and the ORIGINAL PASTE at once — +a model asked for a causal map will invent an intervention nobody wrote down, and the only way to +drop those is to compare against what the patient actually said. A Step modelled as a function of +its predecessor's output cannot say this, and the earlier version of this file papered over it by +pretending the second step consumed only the draft. -Domain types are deliberately thin. This is an example inside a generic package, and a realistic -Finding model would drag medical vocabulary into code that must stay domain-free. +Two strategies are bound to the first Step, because "ask a model for a map" is one operation with +several ways to do it — that is the case the `Strategy` layer exists for, and inventing +`AskModelForCausalMapV2` would be the naming drift this package reports. + +Domain types are deliberately thin. This is an example inside a package that must not learn a +domain; the real `CaseGraph` lives in nobsmed. + + uv run python3 -m examples.evidence_case_graph.flow """ from __future__ import annotations -from dataclasses import dataclass, field +from dataclasses import dataclass -from plan_types.plan import Plan, Step, Variable +from plan_types import Plan, Service, Step, Variable, render_mermaid, validate +from plan_types.execution import LocalRunner, check_strategy, execute +from plan_types.invariants import topology, typing @dataclass(frozen=True) -class Study: - pmid: str - text: str +class CausalMap: + """A set of claims of the form `intervention -> outcome`. The product, not the Plan.""" + edges: tuple[tuple[str, str], ...] = () -@dataclass(frozen=True) -class Findings: - pmid: str - claims: tuple[str, ...] = () + def __str__(self) -> str: + return ", ".join(f"{a} -> {b}" for a, b in self.edges) or "(empty)" -@dataclass(frozen=True) -class EvidenceGraph: - """The canonical case IR. Projections read this; nothing reads the Plan.""" +plan_text = Variable("plan_text", str) +draft = Variable("draft", CausalMap) +case_graph = Variable("case_graph", CausalMap) - nodes: tuple[str, ...] = () - edges: tuple[tuple[str, str], ...] = field(default_factory=tuple) +#: What the arm needs REACHABLE, as opposed to what flows through it. Declared at the top level and +#: referenced by name, docker-compose style: a Step reaching for an undeclared Service is refused. +llm = Service("llm_service", kind="api", why="every arm calls a model; runs anywhere with network") -class ParseStudyStep(Step[Study, Findings]): - consumes = Variable("study", Study) - produces = Variable("findings", Findings) +class AskModelForCausalMap(Step): + """One call: the paste in, a whole causal map out. Nothing retrieved, nothing cited.""" - def run(self, value: Study) -> Findings: - claims = tuple(s.strip() for s in value.text.split(".") if s.strip()) - return Findings(pmid=value.pmid, claims=claims) + inputs, outputs = (plan_text,), (draft,) + uses = (llm,) -class BuildEvidenceGraphStep(Step[Findings, EvidenceGraph]): - consumes = Variable("findings", Findings) - produces = Variable("graph", EvidenceGraph) +class DropWhatThePasteDoesNotSupport(Step): + """Remove edges the paste never mentioned. Needs the draft AND the paste.""" - def run(self, value: Findings) -> EvidenceGraph: - nodes = (value.pmid, *value.claims) - return EvidenceGraph(nodes=nodes, edges=tuple((value.pmid, c) for c in value.claims)) + inputs, outputs = (draft, plan_text), (case_graph,) -EVIDENCE_CASE_GRAPH = Plan( - name="evidence_case_graph", - steps=(ParseStudyStep(), BuildEvidenceGraphStep()), +plan = Plan( + name="llm_causal_map", + steps=(AskModelForCausalMap(), DropWhatThePasteDoesNotSupport()), + declared_inputs=(plan_text,), + services=(llm,), ) + + +# ── two ways to ask a model, neither of them the "real" one ────────────────────────────────────── + +def ask_one_shot(plan_text: str) -> CausalMap: + """Cheap: one edge per mentioned drug, straight to the stated goal.""" + goal = "symptom control" + return CausalMap(edges=tuple((w.strip(".,"), goal) for w in plan_text.split() + if w.istitle() and len(w) > 4)) + + +def ask_with_mechanism(plan_text: str) -> CausalMap: + """Expensive: routes each intervention through a mechanism node, and invents one edge.""" + edges = [(w.strip(".,"), f"{w.strip('.,').lower()} mechanism") for w in plan_text.split() + if w.istitle() and len(w) > 4] + edges.append(("Acupuncture", "symptom control")) # ← nobody wrote this down + return CausalMap(edges=tuple(edges)) + + +def drop_ungrounded(draft: CausalMap, plan_text: str) -> CausalMap: + """The grounding step. THIS is why the fan-in exists.""" + said = plan_text.lower() + return CausalMap(edges=tuple((a, b) for a, b in draft.edges if a.lower() in said)) + + +one_shot = {AskModelForCausalMap: ask_one_shot, + DropWhatThePasteDoesNotSupport: drop_ungrounded} +with_mechanism = {AskModelForCausalMap: ask_with_mechanism, + DropWhatThePasteDoesNotSupport: drop_ungrounded} + + +def main() -> None: + print("invariants:", validate(plan, [*topology.ALL, *typing.ALL]) or "[]") + print(render_mermaid(plan)) + + paste = "Metformin and Spironolactone for PCOS, plus Inositol daily." + + for arm, strategy in (("one_shot", one_shot), ("with_mechanism", with_mechanism)): + problems = check_strategy(plan, strategy) + if problems: + raise SystemExit("\n".join(problems)) + env = execute(plan, {"plan_text": paste}, LocalRunner(strategy)) + print(f"{arm:>14}: draft {len(env['draft'].edges)} edges -> kept {env['case_graph']}") + + +if __name__ == "__main__": + main() diff --git a/examples/hello/__init__.py b/examples/hello/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/examples/hello/flow.py b/examples/hello/flow.py new file mode 100644 index 0000000..04f17f1 --- /dev/null +++ b/examples/hello/flow.py @@ -0,0 +1,114 @@ +"""One Plan, two ways of performing it — the whole idea, in a domain nobody has to learn. + + document ──> Outline ──> outline ──┐ + │ ├──> Summarize ──> summary + └────────────────────────────────┘ + +`Summarize` fans in: it needs the outline AND the original document. That is the ordinary case, not +an advanced one, and it is what a Step modelled as a function of its predecessor's output cannot +express. + +Run it: + + uv run python3 -m examples.hello.flow +""" +from __future__ import annotations + +from dataclasses import dataclass + +from plan_types import Plan, Step, Variable, render_mermaid, validate +from plan_types.execution import LocalRunner, check_strategy, execute +from plan_types.invariants import topology, typing + + +# ── what flows ─────────────────────────────────────────────────────────────────────────────────── + +@dataclass(frozen=True) +class Document: + title: str + body: str + + +@dataclass(frozen=True) +class Outline: + points: tuple[str, ...] + + +@dataclass(frozen=True) +class Summary: + text: str + + +# ── the declaration ────────────────────────────────────────────────────────────────────────────── +# +# Lowercase, because these are names bound to Variables, not conventional constants — and +# `SUMMARY = Variable("summary", ...)` says the word twice in two cases for no gain. + +document = Variable("document", Document) +outline = Variable("outline", Outline) +summary = Variable("summary", Summary) + + +class MakeOutline(Step): + """Pull the points out of a document.""" + + inputs, outputs = (document,), (outline,) + + +class Summarize(Step): + """Write the summary. Needs the outline and the document it came from.""" + + inputs, outputs = (document, outline), (summary,) + + +plan = Plan( + name="summarize_document", + steps=(MakeOutline(), Summarize()), + declared_inputs=(document,), +) + + +# ── how it is performed — several ways, none of them privileged ────────────────────────────────── +# +# Ordinary functions. Parameter names match the Variable names because the runner calls by keyword: +# a Step with two inputs called positionally is one reorder away from a mis-wire that the types +# cannot catch when both inputs are strings. + +def outline_by_sentence(document: Document) -> Outline: + return Outline(points=tuple(s.strip() for s in document.body.split(".") if s.strip())) + + +def summarize_fast(document: Document, outline: Outline) -> Summary: + """The cheap one: the first point, and stop.""" + first = outline.points[0] if outline.points else document.title + return Summary(text=f"{document.title}: {first}.") + + +def summarize_precise(document: Document, outline: Outline) -> Summary: + """The careful one: every point, in order.""" + return Summary(text=f"{document.title}: " + "; ".join(outline.points) + ".") + + +#: Two arms. Note what is NOT here: a second Plan, and a `SummarizeV2` Step. There is one operation +#: called `Summarize` and two ways of doing it, which is what the words already meant. +fast = {MakeOutline: outline_by_sentence, Summarize: summarize_fast} +precise = {MakeOutline: outline_by_sentence, Summarize: summarize_precise} + + +def main() -> None: + findings = validate(plan, [*topology.ALL, *typing.ALL]) + print("invariants:", findings or "[]") + print(render_mermaid(plan)) + + doc = Document(title="Ninety seconds", body="Name the unit. Then run it. Record the result") + + for arm, strategy in (("fast", fast), ("precise", precise)): + problems = check_strategy(plan, strategy) + if problems: + raise SystemExit("\n".join(problems)) + result = execute(plan, {"document": doc}, LocalRunner(strategy)) + print(f"{arm:>8}: {result['summary'].text}") + + +if __name__ == "__main__": + main() diff --git a/examples/pydantic_graph_demo/__init__.py b/examples/pydantic_graph_demo/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/examples/pydantic_graph_demo/flow.py b/examples/pydantic_graph_demo/flow.py new file mode 100644 index 0000000..93a0c37 --- /dev/null +++ b/examples/pydantic_graph_demo/flow.py @@ -0,0 +1,134 @@ +"""One Plan. Two strategies. Two runtimes. Nothing edited in between. + +The workflow is Pydantic Graph's own email-feedback example, in its acyclic core — and with the +one thing that example does not have: + + user ──> WriteEmail ──> draft ──> Critique ──> feedback ──┐ + │ ├──> Revise ──> email + └────────────────────────────────┘ + +**`Revise` needs the draft AND the critique.** That is a fan-in, and it is not an exotic case: you +cannot revise a draft from the critique alone. In a node-returns-the-next-node model the draft has +to be carried in mutable state to be available two hops later; here it is an edge, so the +requirement is visible in the declaration and `topology.bound_inputs` can check it. + +Run it: + + uv sync --extra pydantic-graph + uv run python3 -m examples.pydantic_graph_demo.flow +""" +from __future__ import annotations + +import asyncio +from dataclasses import dataclass + +from plan_types import Plan, Step, Variable, render_mermaid, validate +from plan_types.execution import LocalRunner, check_strategy, execute +from plan_types.execution.pydantic_graph import to_pydantic_graph +from plan_types.invariants import topology, typing + + +@dataclass(frozen=True) +class User: + name: str + interests: tuple[str, ...] + + +@dataclass(frozen=True) +class Email: + subject: str + body: str + + +@dataclass(frozen=True) +class Critique: + notes: tuple[str, ...] + + +user = Variable("user", User) +draft = Variable("draft", Email) +feedback = Variable("feedback", Critique) +email = Variable("email", Email) + + +class WriteEmail(Step): + """Write the first draft from what we know about the reader.""" + + inputs, outputs = (user,), (draft,) + + +class CritiqueDraft(Step): + """Say what is wrong with it.""" + + inputs, outputs = (draft,), (feedback,) + + +class Revise(Step): + """⚠️ THE FAN-IN. Needs the draft and the critique of it, at once.""" + + inputs, outputs = (draft, feedback), (email,) + + +plan = Plan( + name="email_with_feedback", + steps=(WriteEmail(), CritiqueDraft(), Revise()), + declared_inputs=(user,), +) + + +# ── implementations. Ordinary functions; no framework in sight ─────────────────────────────────── + +def write_terse(user: User) -> Email: + return Email(subject="Hello", body=f"Hi {user.name}. {user.interests[0]}?") + + +def write_warm(user: User) -> Email: + return Email(subject=f"{user.name}, something for you", + body=f"Hi {user.name} — we thought of you because you like " + f"{' and '.join(user.interests)}.") + + +def critique(draft: Email) -> Critique: + notes = [] + if len(draft.body) < 60: + notes.append("too short to be worth sending") + if draft.subject.lower() in {"hello", "hi"}: + notes.append("subject says nothing") + return Critique(notes=tuple(notes)) + + +def revise(draft: Email, feedback: Critique) -> Email: + if not feedback.notes: + return draft + return Email(subject=draft.subject, body=draft.body + f" [revised: {len(feedback.notes)} note(s)]") + + +terse = {WriteEmail: write_terse, CritiqueDraft: critique, Revise: revise} +warm = {WriteEmail: write_warm, CritiqueDraft: critique, Revise: revise} + + +async def main() -> None: + print("invariants:", validate(plan, [*topology.ALL, *typing.ALL]) or "[]") + print(render_mermaid(plan)) + + reader = User(name="Samuel", interests=("type safety", "graphs")) + + for arm, strategy in (("terse", terse), ("warm", warm)): + assert check_strategy(plan, strategy) == () + + local = execute(plan, {"user": reader}, LocalRunner(strategy))["email"] + + graph = to_pydantic_graph(plan, strategy) + on_graph = (await graph.run(state={}, inputs={"user": reader}))["email"] + + assert local == on_graph, ( + f"{arm}: the two runtimes disagreed — {local!r} vs {on_graph!r}. That would mean the " + f"Plan does not determine the result, which is the claim this file exists to check.") + print(f"\n{arm}:") + print(f" LocalRunner {local.body}") + print(f" Pydantic Graph {on_graph.body}") + print(" identical ✓") + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/examples/pydantic_graph_docs/__init__.py b/examples/pydantic_graph_docs/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/examples/pydantic_graph_docs/control_no_plan.py b/examples/pydantic_graph_docs/control_no_plan.py new file mode 100644 index 0000000..d47a041 --- /dev/null +++ b/examples/pydantic_graph_docs/control_no_plan.py @@ -0,0 +1,73 @@ +"""CONTROL ARM — stage 3's three arms with NO Plan layer. Just GraphBuilder. + +⚠️ **This is the FAIR version, and the first one was not.** It originally had three copy-pasted +builders, which made the Plan layer look good by comparison and would not have survived ten seconds +of scrutiny: the obvious move is to parameterise, and a parameterised builder handles all three arms +in a dozen lines. Measuring against code nobody would write is not a measurement. + +So there is no line-count claim here any more. What is left is the difference that survives a fair +control, and it is not about volume: + + build(square_impl) you cannot get a graph, a diagram, or a type check + until an implementation EXISTS + + Plan(...) validates and renders with nothing implemented at all + + uv run python3 -m examples.pydantic_graph_docs.control_no_plan +""" +from __future__ import annotations + +import asyncio +from typing import Callable + +from pydantic_graph import GraphBuilder, StepContext, reduce_list_append + + +def build(square_impl: Callable[[int], int]): + """Their `parallel_processing.py`, with the varying step passed in.""" + g = GraphBuilder(name="arm", input_type=list, output_type=int) + + @g.step + async def square(ctx: StepContext[None, None, int]) -> int: + return square_impl(ctx.inputs) + + collect = g.join(reduce_list_append, initial_factory=list) + + @g.step + async def total(ctx: StepContext[None, None, list]) -> int: + return sum(ctx.inputs) + + g.add(g.edge_from(g.start_node).map().to(square), g.edge_from(square).to(collect), + g.edge_from(collect).to(total), g.edge_from(total).to(g.end_node)) + return g.build() + + +ARMS = { + "exact": lambda n: n * n, + "by_addition": lambda n: sum(abs(n) for _ in range(abs(n))), + "cheap": lambda n: n * n if abs(n) <= 10 else abs(n) * 10, +} +CORPUS = [[1, 2, 3, 4, 5], [12], [3, 20]] + + +async def main() -> None: + print(f" {'input':<16} {'expected':>9} " + " ".join(f"{a:>14}" for a in ARMS)) + for case in CORPUS: + expected = sum(n * n for n in case) + cells = [] + for impl in ARMS.values(): + got = await build(impl).run(inputs=case) + cells.append(f"{got:>8} {'ok' if got == expected else 'WRONG':>5}") + print(f" {str(case):<16} {expected:>9} " + " ".join(cells)) + + print("\n Same numbers as stage 3, in twelve lines. This is a GOOD control.") + print(" The differences that survive it:") + print(" - `build()` needs an implementation before it can produce anything at all.") + print(" No graph, no diagram, no type check, until someone has written square_impl.") + print(" - its diagram has no types on the edges, because an edge carries whatever the") + print(" function returned and there is no name for it.") + print(" - its diagram shows map/join as NODES: it draws the machinery, not the process.") + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/examples/pydantic_graph_docs/stage1_counter.py b/examples/pydantic_graph_docs/stage1_counter.py new file mode 100644 index 0000000..ff161ff --- /dev/null +++ b/examples/pydantic_graph_docs/stage1_counter.py @@ -0,0 +1,109 @@ +"""STAGE 1 — their `simple_counter.py`, declared as a Plan and run on their runtime. + +Their example, copied verbatim from +https://pydantic.dev/docs/ai/graph/builder/ (Quick Start), is `their_version()` below. + +The question this file answers is only: **does the round trip work at all?** + + Plan ──> render_mermaid ──> to_pydantic_graph ──> graph.run() + +No claim about value yet. Stage 2 varies the implementation, stage 3 varies the shape. + + uv run python3 -m examples.pydantic_graph_docs.stage1_counter +""" +from __future__ import annotations + +import asyncio +from dataclasses import dataclass + +from plan_types import Plan, Step, Variable, render_mermaid, validate +from plan_types.execution import LocalRunner, execute +from plan_types.execution.pydantic_graph import to_pydantic_graph +from plan_types.invariants import topology, typing + + +# ── theirs, verbatim from the docs ─────────────────────────────────────────────────────────────── + +@dataclass +class CounterState: + """State for tracking a counter value.""" + + value: int = 0 + + +async def their_version() -> int: + from pydantic_graph import GraphBuilder, StepContext + + g = GraphBuilder(state_type=CounterState, output_type=int) + + @g.step + async def increment(ctx: StepContext[CounterState, None, None]) -> int: + """Increment the counter and return its value.""" + ctx.state.value += 1 + return ctx.state.value + + @g.step + async def double_it(ctx: StepContext[CounterState, None, int]) -> int: + """Double the input value.""" + return ctx.inputs * 2 + + g.add( + g.edge_from(g.start_node).to(increment), + g.edge_from(increment).to(double_it), + g.edge_from(double_it).to(g.end_node), + ) + return await g.build().run(state=CounterState()) + + +# ── ours: the same process, said once, with no runtime in it ───────────────────────────────────── + +count = Variable("count", int) +doubled = Variable("doubled", int) + + +class Increment(Step): + inputs, outputs = (), (count,) + + +class DoubleIt(Step): + inputs, outputs = (count,), (doubled,) + + +plan = Plan(name="counter", steps=(Increment(), DoubleIt())) + + +def increment() -> int: + return 1 + + +def double_it(count: int) -> int: + return count * 2 + + +strategy = {Increment: increment, DoubleIt: double_it} + + +async def main() -> None: + print("invariants:", validate(plan, [*topology.ALL, *typing.ALL]) or "[]") + print(render_mermaid(plan)) + + theirs = await their_version() + ours_local = execute(plan, {}, LocalRunner(strategy))["doubled"] + ours_on_their_runtime = (await to_pydantic_graph(plan, strategy).run(state={}, inputs={}))["doubled"] + + print(f" their GraphBuilder, hand-wired {theirs}") + print(f" our Plan, LocalRunner {ours_local}") + print(f" our Plan, compiled onto theirs {ours_on_their_runtime}") + assert theirs == ours_local == ours_on_their_runtime == 2 + print(" all three agree ✓") + + # ⚠️ The one real difference, and it is not cosmetic. Their `increment` reads and mutates + # `ctx.state`; it declares its input as None and takes the value from a mutable object that + # outlives the step. Ours makes the same value an EDGE. Same answer, and the dependency is + # visible in the diagram rather than in the body of a function. + print("\n theirs threads the counter through mutable state (ctx.state.value += 1)") + print(" ours makes it an edge: Increment -> count -> DoubleIt") + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/examples/pydantic_graph_docs/stage2_strategies.py b/examples/pydantic_graph_docs/stage2_strategies.py new file mode 100644 index 0000000..8a59232 --- /dev/null +++ b/examples/pydantic_graph_docs/stage2_strategies.py @@ -0,0 +1,64 @@ +"""STAGE 2 — three implementations of ONE logical Step, evaluated against each other. + +Domain is theirs: `parallel_processing.py` squares numbers. The Plan: + + numbers ──> Square ──> squares ──> Total ──> total + +Three strategies vary ONE Step. `Square` stands in for the step you would really want to vary — +a different model, a different prompt, a different agent. Deterministic stand-ins here so the file +runs with no API key and the same numbers come out every time; the STRUCTURE is what is being +shown, and it is identical either way. + +What to look at: + + the Plan object is constructed ONCE and never touched + three diagrams, identical topology, different implementation labels + an eval table whose rows are comparable BECAUSE the Plan is the same object + + uv run python3 -m examples.pydantic_graph_docs.stage2_strategies +""" +from __future__ import annotations + +import asyncio + +from plan_types import Plan, Step, Variable, render_mermaid, validate +from plan_types.execution import LocalRunner, check_strategy, execute +from plan_types.execution.pydantic_graph import to_pydantic_graph +from plan_types.invariants import topology, typing + +from examples.pydantic_graph_docs.their_example import (ARMS, Square, Total, numbers, + plan, squares) + +#: Stage 2's own corpus — small, and the last two rows are what separate the arms. +CORPUS = [[1, 2, 3], [4, 5], [12], [3, 20]] + + +async def main() -> None: + print("invariants:", validate(plan, [*topology.ALL, *typing.ALL]) or "[]") + + print("\nSAME PLAN, THREE STRATEGIES — identical topology, different labels:\n") + for arm, strategy in ARMS.items(): + print(f"### {arm}") + print(render_mermaid(plan, strategy)) + print() + + print("EVAL — every row is comparable because the Plan is the same object\n") + print(f" {'input':<12} {'expected':>9} " + " ".join(f"{a:>20}" for a in ARMS)) + for case in CORPUS: + expected = sum(n * n for n in case) + cells = [] + for strategy in ARMS.values(): + assert check_strategy(plan, strategy) == () + got = execute(plan, {"numbers": case}, LocalRunner(strategy))["total"] + cells.append(f"{got:>13} {'ok' if got == expected else 'WRONG':>6}") + print(f" {str(case):<12} {expected:>9} " + " ".join(cells)) + + print("\nand the winning arm compiled onto THEIR runtime, unchanged:") + graph = to_pydantic_graph(plan, ARMS['exact']) + got = (await graph.run(state={}, inputs={"numbers": [3, 20]}))["total"] + print(f" pydantic-graph, numbers=[3, 20] -> {got} (expected 409)") + assert got == 409 + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/examples/pydantic_graph_docs/stage3_map_join.py b/examples/pydantic_graph_docs/stage3_map_join.py new file mode 100644 index 0000000..216cddc --- /dev/null +++ b/examples/pydantic_graph_docs/stage3_map_join.py @@ -0,0 +1,73 @@ +"""STAGE 3 — their `parallel_processing.py`: map, join, reducer. + +Their example, from https://pydantic.dev/docs/ai/graph/builder/ ("A More Complex Example"): + + g.add( + g.edge_from(g.start_node).map().to(square), + g.edge_from(square).to(collect_results), + g.edge_from(collect_results).to(g.end_node), + ) + +`square` is `int -> int`. The `.map()` fans each item out to its own execution; `collect_results` is +a join with `reduce_list_append`. Their words, kept: **map**, **join**, **reducer**. + +A Plan says the same thing by declaring the per-item Step and what it maps over: + + class Square(Step): + inputs, outputs = (number,), (squared,) # int -> int, exactly theirs + map_over = (numbers, squares) # list[int] in, list[int] out + +⚠️ Read what that buys and what it does not. The Plan declares that a fan-out EXISTS. It says +nothing about workers, ordering or concurrency — so the same declaration is a sequential loop under +LocalRunner and an actual parallel `.map()` + join on their engine, with no edit. That is the split +doing its job, and it is the whole claim. + + uv run python3 -m examples.pydantic_graph_docs.stage3_map_join +""" +from __future__ import annotations + +import asyncio + +from plan_types import Plan, Step, Variable, render_mermaid, validate +from plan_types.execution import LocalRunner, check_strategy, execute +from plan_types.execution.pydantic_graph import to_pydantic_graph +from plan_types.invariants import topology, typing + +from examples.pydantic_graph_docs.their_example import (ARMS, CORPUS, Square, Total, + numbers, plan, squares) + +__all__ = ["ARMS", "CORPUS", "Square", "Total", "numbers", "plan", "squares"] + + +async def main() -> None: + print("invariants:", validate(plan, [*topology.ALL, *typing.ALL]) or "[]") + print(render_mermaid(plan, ARMS["exact"])) + + print("\nTHEIR DOCS' OWN CASE — inputs=[1, 2, 3, 4, 5]\n") + strategy = ARMS["exact"] + local = execute(plan, {"numbers": [1, 2, 3, 4, 5]}, LocalRunner(strategy)) + on_theirs = await to_pydantic_graph(plan, strategy).run( + state={}, inputs={"numbers": [1, 2, 3, 4, 5]}) + print(f" their docs say Results: [1, 4, 9, 16, 25]") + print(f" LocalRunner (a loop) {local['squares']}") + print(f" compiled to .map()/join {on_theirs['squares']}") + assert local["squares"] == on_theirs["squares"] == [1, 4, 9, 16, 25] + print(" identical ✓ — one declaration, sequential here, parallel there") + + print("\nTHREE STRATEGIES OVER THE ONE PLAN\n") + print(f" {'input':<16} {'expected':>9} " + " ".join(f"{a:>14}" for a in ARMS)) + for case in CORPUS: + expected = sum(n * n for n in case) + cells = [] + for s in ARMS.values(): + assert check_strategy(plan, s) == () + got = execute(plan, {"numbers": case}, LocalRunner(s))["total"] + cells.append(f"{got:>8} {'ok' if got == expected else 'WRONG':>5}") + print(f" {str(case):<16} {expected:>9} " + " ".join(cells)) + + print("\n The Plan object is never edited between arms. `Square` is ONE logical Step with") + print(" three implementations — not SquareV1/V2/V3, which would be three names for one thing.") + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/examples/pydantic_graph_docs/their_example.py b/examples/pydantic_graph_docs/their_example.py new file mode 100644 index 0000000..cad8d85 --- /dev/null +++ b/examples/pydantic_graph_docs/their_example.py @@ -0,0 +1,66 @@ +"""Their `parallel_processing.py`, declared ONCE and used by every stage. + + numbers ──> Square ──> squares ──> Total ──> total + +Stage 2 varies the implementation of `Square`; stage 3 shows the same declaration compiling to +their real `.map()` and join. Both import from here. + +⚠️ **This module exists because conceptlint reported it.** Stage 2 and stage 3 each declared their +own `Square` and `Total`, and `naming.ambiguous_reference` fired: *"Square is declared twice with +different shapes"*. Two names for what a reader takes to be one operation is exactly what this +package reports, and it had happened inside the examples FOR this package, within an hour of them +being written. One declaration, two demos — which is also the thing the package claims you get. +""" +from __future__ import annotations + +from plan_types import Plan, Step, Variable + +numbers = Variable("numbers", list) +number = Variable("number", int) +squared = Variable("squared", int) +squares = Variable("squares", list) +total = Variable("total", int) + + +class Square(Step): + """Theirs, unchanged: one number in, one number out, applied to each item.""" + + inputs, outputs = (number,), (squared,) + map_over = (numbers, squares) + + +class Total(Step): + """Their example stops at the joined list; summing gives us something to eval.""" + + inputs, outputs = (squares,), (total,) + + +plan = Plan(name="parallel_processing", steps=(Square(), Total()), + declared_inputs=(numbers,)) + + +def square_exact(number: int) -> int: + return number * number + + +def square_by_addition(number: int) -> int: + return sum(abs(number) for _ in range(abs(number))) + + +def square_cheap(number: int) -> int: + """Wrong above 10 — the arm an eval has to catch.""" + return number * number if abs(number) <= 10 else abs(number) * 10 + + +def total_impl(squares: list) -> int: + return sum(squares) + + +#: Three peers. Not SquareV1/V2/V3, which would be three names for one operation. +ARMS = { + "exact": {Square: square_exact, Total: total_impl}, + "by_addition": {Square: square_by_addition, Total: total_impl}, + "cheap": {Square: square_cheap, Total: total_impl}, +} + +CORPUS = [[1, 2, 3, 4, 5], [12], [3, 20]] diff --git a/plan_types/execution/__init__.py b/plan_types/execution/__init__.py new file mode 100644 index 0000000..c2abfc7 --- /dev/null +++ b/plan_types/execution/__init__.py @@ -0,0 +1,16 @@ +"""Execution: how a declared Plan is actually performed. + +Nothing in `plan_types.plan` imports this. The dependency runs one way — a specification must be +readable, validatable and drawable without an execution backend in the room, which is the whole +claim the package makes. + + Step what transformation exists plan/ + Strategy how it is performed here + StepRunner the mechanics of performing it here +""" +from plan_types.execution.local import ExecutionError, LocalRunner, execute +from plan_types.execution.runner import StepRunner +from plan_types.execution.strategy import Implementation, Strategy, check_strategy + +__all__ = ["Strategy", "Implementation", "check_strategy", + "StepRunner", "LocalRunner", "execute", "ExecutionError"] diff --git a/plan_types/execution/local.py b/plan_types/execution/local.py new file mode 100644 index 0000000..8e1af9f --- /dev/null +++ b/plan_types/execution/local.py @@ -0,0 +1,184 @@ +"""`LocalRunner` and `execute` — the smallest thing that can actually run a Plan. + + execute(plan, inputs, LocalRunner(strategy)) + +In-process, sequential, no retries, no concurrency, no durability. It exists to answer one +question — *does the declared process, plus these implementations, produce the expected result?* — +and to be the thing a Plan is proved against before anyone decides whether it needs a workflow +engine. Plenty of Plans never will. + +## The two failures this refuses to have + +**A coroutine returned instead of a result.** An `async def` bound into a synchronous Strategy +returns a coroutine object, which is truthy, has a repr, and flows into the next Step as though it +were data. Refused at the call site with the Step's name. + +**A silently discarded or mis-shaped output.** A Step declaring two outputs whose implementation +returns one value has to fail here, because the alternative is one Variable holding the whole tuple +and the mismatch surfacing three Steps later as a type error about something unrelated. +""" +from __future__ import annotations + +from typing import Any, Mapping, get_origin + +from plan_types.execution.runner import StepRunner +from plan_types.execution.strategy import Strategy +from plan_types.plan.bindings import execution_order +from plan_types.plan.plan import Plan +from plan_types.plan.step import Step, wired_inputs + + +class ExecutionError(RuntimeError): + """An execution that cannot proceed. Never raised for a Step's own failure. + + A Step's implementation raising is that Step's business and propagates untouched — wrapping it + would bury the traceback the caller needs. This is raised only when the RUNNER cannot do its + job: nothing bound, a coroutine it cannot await, an output shape that does not fit the + declaration. + """ + + +class LocalRunner: + """Perform a Step by calling whatever the Strategy bound to it. Satisfies `StepRunner`. + + Deliberately does not inherit from the Protocol — structural conformance is the property being + demonstrated, and a test asserts `isinstance(LocalRunner({}), StepRunner)` holds anyway. + """ + + def __init__(self, strategy: Strategy) -> None: + self.strategy = strategy + + def run(self, step: Step, inputs: dict[str, Any]) -> dict[str, Any]: + cls = type(step) + impl = self.strategy.get(cls) + if impl is None: + raise ExecutionError( + f"no implementation bound for {cls.__name__}. Its Strategy has " + f"{sorted(c.__name__ for c in self.strategy)} — run check_strategy(plan, strategy) " + f"before executing and this is reported for every Step at once.") + + if cls.map_over is not None: + return _run_mapped(cls, impl, inputs) + + result = impl(**inputs) + + if hasattr(result, "__await__"): + result.close() # else "coroutine was never awaited" fires far from the real error + raise ExecutionError( + f"{cls.__name__} is bound to an async implementation, which returned a coroutine " + f"instead of a result. LocalRunner is synchronous by decision — see runner.py. " + f"Nothing awaits this, so it would flow into the next Step as data.") + + return _outputs_by_name(cls, result) + + +def _run_mapped(cls: type[Step], impl: Any, inputs: dict[str, Any]) -> dict[str, Any]: + """Their `.map()` + `join`, run sequentially. + + The implementation is the PER-ITEM operation — `square: int -> int`, exactly theirs — so it is + called once per element and the results are collected in order. That collection is their + `reduce_list_append`, which is the reducer their own example uses. + + ⚠️ Sequential here, and that is not a limitation being hidden: `LocalRunner` has no concurrency + by decision (see runner.py), so a mapped Step under it is a loop. Actual parallelism is what + their engine is for, and `to_pydantic_graph` is where it belongs. + """ + source, collected = cls.map_over + item_var, out_var = cls.inputs[0], cls.outputs[0] + + items = inputs[source.name] + try: + iter(items) + except TypeError: + raise ExecutionError( + f"{cls.__name__} maps over {source.name!r}, which arrived as " + f"{type(items).__name__} and is not iterable. A mapped Step fans out over a " + f"sequence.") from None + + results = [] + for i, item in enumerate(items): + out = impl(**{item_var.name: item}) + if hasattr(out, "__await__"): + out.close() + raise ExecutionError( + f"{cls.__name__} is bound to an async implementation; LocalRunner is synchronous " + f"by decision. See runner.py.") + _check_type(cls, out_var, out) + results.append(out) + del i + return {collected.name: results} + + +def _outputs_by_name(cls: type[Step], result: Any) -> dict[str, Any]: + """Map what an implementation returned onto what its Step declared it produces.""" + outs = cls.outputs + if not outs: + if result is not None: + raise ExecutionError( + f"{cls.__name__} declares no outputs but its implementation returned " + f"{type(result).__name__}. Discarding it silently would hide either a wrong " + f"declaration or a wrong implementation.") + return {} + + if len(outs) == 1: + _check_type(cls, outs[0], result) + return {outs[0].name: result} + + if not isinstance(result, tuple) or len(result) != len(outs): + got = f"a {len(result)}-tuple" if isinstance(result, tuple) else type(result).__name__ + raise ExecutionError( + f"{cls.__name__} declares {len(outs)} outputs " + f"({', '.join(v.name for v in outs)}) so its implementation must return a tuple of " + f"{len(outs)}; got {got}.") + for var, value in zip(outs, result): + _check_type(cls, var, value) + return {v.name: r for v, r in zip(outs, result)} + + +def _check_type(cls: type[Step], var: Any, value: Any) -> None: + """Does the produced value match the Variable's declared type? + + ⚠️ Plain classes only. `list[Finding]` is a parameterized generic and `isinstance` cannot test + its contents, so those are NOT CHECKED — which is not the same as checked and fine, and is why + this says so here rather than leaving a reader to assume the types are enforced end to end. + """ + declared = var.type + if get_origin(declared) is not None or not isinstance(declared, type): + return + if not isinstance(value, declared): + raise ExecutionError( + f"{cls.__name__} declares {var.name!r} as {declared.__name__} and its implementation " + f"produced {type(value).__name__}. The declaration and the code disagree about what " + f"flows here; the next Step would receive the wrong thing.") + + +def execute(plan: Plan, inputs: Mapping[str, Any], runner: StepRunner) -> dict[str, Any]: + """Run every Step in dependency order. Returns every Variable produced, keyed by name. + + Everything, not only `plan.outputs` — an intermediate is what you want when a run went wrong, + and hiding it would make the runner's own output less useful than a print statement. The Plan's + terminal Variables are named by `plan.outputs`, so taking that projection is one line. + + ⚠️ `execution_order` RAISES on a cycle rather than inventing an order. A Plan may legally be + cyclic — `topology.acyclic` reports it, it is not part of what a Plan means — but nothing here + can linearise one. + """ + names = [v.name for v in plan.variables] + if len(names) != len(set(names)): + clashing = sorted({n for n in names if names.count(n) > 1}) + raise ExecutionError( + f"Plan {plan.name!r} has different Variables sharing the name(s) {clashing}. Values " + f"flow by name here, so one would overwrite the other. A Variable is (name, type), so " + f"two of them can differ in type and collide in this dict.") + + missing = [v.name for v in plan.inputs if v.name not in inputs] + if missing: + raise ExecutionError( + f"Plan {plan.name!r} expects {missing}, which was not supplied. Its signature is " + f"{[v.name for v in plan.inputs]}.") + + env: dict[str, Any] = dict(inputs) + for step in execution_order(plan): + gathered = {v.name: env[v.name] for v in wired_inputs(step)} + env.update(runner.run(step, gathered)) + return env diff --git a/plan_types/execution/pydantic_graph.py b/plan_types/execution/pydantic_graph.py new file mode 100644 index 0000000..98e4996 --- /dev/null +++ b/plan_types/execution/pydantic_graph.py @@ -0,0 +1,220 @@ +"""Compile a Plan onto [Pydantic Graph](https://pydantic.dev/docs/ai/graph/graph/). + + Plan + Strategy --> pydantic_graph.Graph --> await graph.run(inputs=...) + +**This is the direction the whole package points.** PlanTypes is not another workflow engine and +does not want to be one: LangGraph, Temporal and Pydantic Graph execute workflows, and they are good +at it. What is missing is the layer where you decide whether the process is RIGHT — before retries, +workers, state and serialization are in the room — and then hand it to one of them unchanged. + +So the same `Plan` and the same `Strategy` run under `LocalRunner` while you are still arguing about +the shape, and under Pydantic Graph when you want its runtime. Neither the Plan nor any +implementation is edited in between. `examples/pydantic_graph_demo/` runs both and asserts the +results are identical, because a claim like that one has to be executable. + +## What this compile does, exactly + +Steps in `execution_order`, chained, threading a dict of Variable-name -> value: + + start --> step_0 --> step_1 --> ... --> end + +Each compiled node calls the SAME `Strategy` implementation `LocalRunner` would call, through the +same `_outputs_by_name`, so a Step cannot behave differently on the two runtimes. + +## What it deliberately does NOT do yet + +**No `Fork` / `Join`.** Pydantic Graph 2.x has both, and a Plan's bindings already say which Steps +are independent — so parallelising is DERIVABLE rather than declarable, which is the interesting +version. It is not built, and this docstring is not going to imply it is. + +**Acyclic only.** `execution_order` raises on a cycle rather than inventing an order. That is not a +gap in the compile, it is where the split earns its keep: Pydantic Graph's own examples — the fives +graph, the vending machine, the email-feedback loop — are cyclic state machines, and a cyclic Plan +is exactly the case where you SHOULD reach for that engine. PlanTypes says so instead of pretending +to run it. + +**No state or deps.** `state_type` and `deps_type` stay `None`. A Step implementation's +dependencies live in its closure — see `strategy.py`. + +## On the two `Step` classes + +`pydantic_graph.Step` and `plan_types.Step` are different things and both are correctly named: +theirs is an executable node, ours is a declaration — `p-plan:Step` against, in effect, +`prov:Activity`. This module is the one place both are in scope, and it never imports theirs. +""" +from __future__ import annotations + +from typing import TYPE_CHECKING, Any + +from plan_types.execution.local import ExecutionError, _check_type, _outputs_by_name +from plan_types.execution.strategy import Strategy, check_strategy +from plan_types.plan.bindings import execution_order +from plan_types.plan.plan import Plan + +if TYPE_CHECKING: # pragma: no cover + from pydantic_graph import Graph + + +def to_pydantic_graph(plan: Plan, strategy: Strategy, *, name: str | None = None) -> "Graph": + """Build a `pydantic_graph.Graph` that runs `plan` under `strategy`. + + Raises BEFORE building rather than mid-run: an unbound Step or an uncallable signature is a fact + about the Strategy, and discovering it inside a graph run would attribute it to the runtime. + + ## Shape of the compiled graph + + The Plan's environment — Variable name -> value — lives in their `state`, which is what state is + for and is how their own examples carry values that outlive one edge. Edges then carry only what + a `.map()` needs to fan out. + + seed ──> step ──> step ──> ... ──> finish ──> end + + A MAPPED Step compiles to their four-part fan-out, using their primitives, not a loop wearing + their name: + + emit ──.map()──> ──> join(reduce_list_append) ──> store + + So a Plan that declares `map_over` becomes an actually-parallel map on their engine, and the + same declaration is a sequential loop under `LocalRunner`. That is the split working: the Plan + says a fan-out exists; how many workers run it is the runtime's business. + """ + try: + from pydantic_graph import GraphBuilder, reduce_list_append + except ImportError as exc: # pragma: no cover - depends on the extra + raise ExecutionError( + "pydantic-graph is not installed. It is an OPTIONAL extra, on purpose: plan_types.plan " + "imports no execution framework, and this adapter is the only module in the package " + "that imports one. Install with: uv add 'plan-types[pydantic-graph]'") from exc + + problems = check_strategy(plan, strategy) + if problems: + raise ExecutionError( + f"Plan {plan.name!r} cannot be compiled — the Strategy is incomplete:\n " + + "\n ".join(problems)) + + _refuse_cycles_for_the_right_reason(plan) + order = execution_order(plan) + g = GraphBuilder(name=name or plan.name, state_type=dict, input_type=dict, output_type=dict) + + async def seed(ctx: Any) -> None: + ctx.state.update(ctx.inputs) + + async def finish(ctx: Any) -> dict[str, Any]: + return dict(ctx.state) + + seed_node = g.step(seed, label="seed") + finish_node = g.step(finish, label="finish") + + edges = [g.edge_from(g.start_node).to(seed_node)] + previous: Any = seed_node + + for i, step in enumerate(order): + cls = type(step) + if cls.map_over is None: + node = g.step(_plain(cls, strategy[cls], i), label=cls.__name__) + edges.append(g.edge_from(previous).to(node)) + previous = node + continue + + emit, per_item, store = _mapped(g, cls, strategy[cls], i) + collect = g.join(reduce_list_append, initial_factory=list) + edges += [ + g.edge_from(previous).to(emit), + g.edge_from(emit).map().to(per_item), + g.edge_from(per_item).to(collect), + g.edge_from(collect).to(store), + ] + previous = store + + edges += [g.edge_from(previous).to(finish_node), g.edge_from(finish_node).to(g.end_node)] + g.add(*edges) + return g.build() + + +def _plain(cls: Any, impl: Any, i: int) -> Any: + """An ordinary Step: read its inputs from state, write its outputs back. + + Calls the same `_outputs_by_name` `LocalRunner` does, so a Step cannot mean one thing here and + another there — a second copy of that logic is a second thing to drift. + """ + async def run_step(ctx: Any) -> None: + env = ctx.state + missing = [v.name for v in cls.inputs if v.name not in env] + if missing: + raise ExecutionError( + f"{cls.__name__} needs {missing}, which nothing upstream produced. On a compiled " + f"graph that is a WIRING fault, not a runtime one — " + f"validate(plan, topology.ALL) reports it before you compile.") + result = impl(**{v.name: env[v.name] for v in cls.inputs}) + if hasattr(result, "__await__"): + result = await result # async IS fine here — a graph run awaits. LocalRunner cannot. + env.update(_outputs_by_name(cls, result)) + + run_step.__name__ = f"{cls.__name__}_{i}" + return run_step + + +def _mapped(g: Any, cls: Any, impl: Any, i: int) -> tuple[Any, Any, Any]: + """`emit -> .map() -> per_item -> join -> store`, in their vocabulary. + + `emit` puts the list on the edge, because `.map()` fans out an EDGE's value and our environment + lives in state. `store` puts the joined list back. Neither is a Step in the Plan — they are the + seam between our environment and their edges, and they exist because the two models differ, not + because the Plan has extra nodes in it. + """ + source, collected = cls.map_over + item_var, out_var = cls.inputs[0], cls.outputs[0] + + async def emit(ctx: Any) -> list: + items = ctx.state.get(source.name) + try: + return list(items) + except TypeError: + raise ExecutionError( + f"{cls.__name__} maps over {source.name!r}, which arrived as " + f"{type(items).__name__} and is not iterable.") from None + + async def per_item(ctx: Any) -> Any: + out = impl(**{item_var.name: ctx.inputs}) + if hasattr(out, "__await__"): + out = await out + _check_type(cls, out_var, out) + return out + + async def store(ctx: Any) -> None: + ctx.state[collected.name] = list(ctx.inputs) + + emit.__name__ = f"emit_{source.name}_{i}" + per_item.__name__ = f"{cls.__name__}_{i}" + store.__name__ = f"store_{collected.name}_{i}" + return (g.step(emit, label=f"map {source.name}"), + g.step(per_item, label=cls.__name__), + g.step(store, label=f"join -> {collected.name}")) + + +def _refuse_cycles_for_the_right_reason(plan: Plan) -> None: + """A cyclic Plan cannot compile — and the reason is NOT that toposort fails. + + `execution_order` raises anyway, saying "there is no execution order to return", which reads as + a limitation of the algorithm. It is not. Given a perfect cyclic scheduler this would still loop + forever, because **a Plan does not say when to stop**: their `Feedback.run()` returns + `WriteEmail | End[Email]` and the predicate lives in the node body, where we cannot see it. + + Fixing that in general means adding Decision/branch/End here, which is rebuilding what + GraphBuilder already owns. The route that does not: collapse each strongly connected component + into a `MultiStep` and declare `until` — an SCC condensation is ALWAYS a DAG, so the outer Plan + compiles and the loop inside is theirs. Partly built: `MultiStep.until` exists, the condensation + does not. + """ + from plan_types.plan.plan import PlanError + + try: + execution_order(plan) + except PlanError as exc: + raise PlanError( + f"{exc} — and note the reason is NOT that this cannot be ordered. A Plan does not " + f"carry a termination predicate, so no scheduler could run it either: pydantic-graph " + f"puts that predicate in the node body (`-> Next | End[T]`), which is the right place " + f"for it and is theirs. An iterative region belongs in a MultiStep with `until` set, " + f"whose implementation is one of their graphs; the outer Plan is then a DAG." + ) from exc diff --git a/plan_types/execution/runner.py b/plan_types/execution/runner.py new file mode 100644 index 0000000..2e41dd8 --- /dev/null +++ b/plan_types/execution/runner.py @@ -0,0 +1,53 @@ +"""`StepRunner` — the boundary between a Plan and whatever performs it. + + Plan / Step / Variable / Strategy what to do, and how it is implemented + | + v + StepRunner Protocol: one Step, in and out + | + +-- LocalRunner here. In-process, sequential, no retries. + +-- (Temporal, PydanticGraph, LangGraph) NOT BUILT. See below. + +## Protocol, not a base class + +Nothing has to inherit from anything. `LocalRunner` satisfies this by having the method, and so +would an adapter living in another package that has never heard of `plan_types` — which is the +point, since an execution backend should not have to import a specification library to be usable +with one. A base class would also accumulate helpers, state and hooks over time, and every one of +them would be execution semantics leaking back toward the declaration. + +## ⚠️ Synchronous, and that is a decision rather than an oversight + +Two of the four handoffs that produced this module disagreed: one argued PlanTypes must stay +sync/async-neutral, the other wrote `async def run(...)` into the Protocol, which is not neutral — +it forces every implementation of every Step to be a coroutine, including `lambda x: x + 1`. + +So: sync now. An `AsyncStepRunner` lands when there is a real async implementation to run, and not +before — `docs/design.md` §6. What must NOT happen in the meantime is a sync runner silently +accepting an `async def` and returning an un-awaited coroutine, which is a result-shaped object that +is not a result. `check_strategy` reports it and `LocalRunner` refuses it. + +## What this deliberately does not do + +No retries, no timeouts, no concurrency, no durability, no checkpointing. Those are the reasons to +reach for Temporal or LangGraph, and a Plan that never needs them should never pay for them. An +exception means the execution failed and stops. +""" +from __future__ import annotations + +from typing import Any, Protocol, runtime_checkable + +from plan_types.plan.step import Step + + +@runtime_checkable +class StepRunner(Protocol): + """Perform one Step with its inputs already gathered, and return its outputs by name. + + `inputs` and the return value are both keyed by Variable NAME, so a runner never has to know + what a Plan is — only what a Step declares. That is what keeps `execute()` (which does know + about Plans) separable from the mechanics of running one Step. + """ + + def run(self, step: Step, inputs: dict[str, Any]) -> dict[str, Any]: + ... diff --git a/plan_types/execution/strategy.py b/plan_types/execution/strategy.py new file mode 100644 index 0000000..430e2e1 --- /dev/null +++ b/plan_types/execution/strategy.py @@ -0,0 +1,132 @@ +"""`Strategy` — which implementation performs each Step, for one execution. + + Step WHAT transformation exists plan-time, declared once + Strategy HOW it is performed, this time chosen per execution + StepRunner the mechanics of performing it sync, retries, durability, workers + +A Strategy is an ordinary mapping. That is deliberate and it is the whole API: + + fast = {Summarize: summarize_fast} + slow = {Summarize: summarize_precise} + +The Plan is not edited between those two lines. That is the property worth having — an experiment +can then say *the logical process was held constant; only the implementation of `Summarize` +changed*, and mean it, because the same `Plan` object served both arms. + +## ⚠️ `Strategy`, not `bindings` — the word was already taken + +`plan_types.plan.bindings` means **how Variables connect Steps**: two Steps are bound when they +share a `Variable`. The handoffs that produced this module used "bindings" for **which +implementation satisfies a Step**. Two concepts, one word, in one package — the exact thing +`naming.ambiguous_reference` reports, and it would have shipped inside the module built to prevent +it. + + binding Variable wiring plan_types/plan/bindings.py + Strategy Step -> implementation here + +## Grounding, stated plainly + +P-Plan has no term for this, and neither does PROV-O. `p-plan:Step` is the intended operation and +`prov:Activity` is one execution of it; *"the code that will perform this Step if it runs"* is +neither. So `Strategy` is **ours, deliberately uncited** — the same call as `Step.uses`, and for the +same reason: citing a term that does not say this would be the failure +`provenance.grounded_citation` exists to catch. +""" +from __future__ import annotations + +import inspect +from typing import Any, Callable, Mapping + +from plan_types.plan.plan import Plan +from plan_types.plan.step import Step + +#: One way of performing a Step. Called by keyword with the Step's input Variable NAMES. +Implementation = Callable[..., Any] + +#: Step class -> the implementation that performs it. Keyed on the CLASS, since the class is the +#: declared operation; a Plan holding two instances of one Step class cannot give them different +#: implementations, which has not been needed and is recorded here rather than designed around. +Strategy = Mapping[type[Step], Implementation] + + +def check_strategy(plan: Plan, strategy: Strategy) -> tuple[str, ...]: + """Findings about a Strategy against a Plan. Empty means every Step can be called. + + Static, and honest about its limits: it reads signatures, so it can tell you that + `summarize(doc)` will not accept the keyword `document`, and it CANNOT tell you what that + function returns. Return shape is checked by the runner at execution — see `local.py`. + + ⚠️ Returns findings rather than raising, matching `validate()`: one wrong signature usually + produces several complaints, and seeing all of them is how you tell one root cause from three + problems. + """ + findings: list[str] = [] + for step in plan.steps: + cls = type(step) + impl = strategy.get(cls) + if impl is None: + findings.append( + f"{cls.__name__} has no implementation in this Strategy. A Step is a declaration; " + f"something has to say how it is performed.") + continue + if not callable(impl): + findings.append(f"{cls.__name__} is bound to {impl!r}, which is not callable.") + continue + if inspect.iscoroutinefunction(impl): + findings.append( + f"{cls.__name__} is bound to async {impl.__name__}. Every runner here is " + f"synchronous, and calling it would return a coroutine that nobody awaits — a " + f"result-shaped object that is not the result. Wrap it, or add an async runner.") + continue + findings.extend(_signature_findings(cls, impl)) + return tuple(findings) + + +def _signature_findings(cls: type[Step], impl: Implementation) -> list[str]: + """Can `impl(**{name: value for each of cls.inputs})` actually be called? + + ⚠️ This is where `Variable.name` stops being a label and becomes part of the contract. The + runner calls by KEYWORD — a Step with three inputs called positionally is one argument reorder + away from a silent mis-wire that the types cannot catch when two inputs share a type. The cost + is that renaming a Variable renames a parameter, and that cost is deliberate: it is visible + here, at import, rather than at 3am as a value in the wrong slot. + """ + # `cls.inputs` deliberately, not the wired ports: a mapped Step's implementation is called + # with ONE ITEM, so its parameter is the item's name. The list never reaches it. + wanted = [v.name for v in cls.inputs] + try: + sig = inspect.signature(impl) + except (TypeError, ValueError): # builtins and C callables have no introspectable signature + return [f"{cls.__name__}: cannot read the signature of {impl!r} — NOT CHECKED, " + f"which is not the same as checked and fine."] + + params = sig.parameters + takes_kwargs = any(p.kind is inspect.Parameter.VAR_KEYWORD for p in params.values()) + + findings: list[str] = [] + if not takes_kwargs: + unaccepted = [n for n in wanted if n not in params + or params[n].kind is inspect.Parameter.POSITIONAL_ONLY] + if unaccepted: + findings.append( + f"{cls.__name__} declares input(s) {unaccepted} that {_name(impl)}{sig} will not " + f"accept by keyword. The runner calls by Variable name, so the parameter has to be " + f"spelled the same.") + + required = [n for n, p in params.items() + if p.default is inspect.Parameter.empty + and p.kind in (inspect.Parameter.POSITIONAL_OR_KEYWORD, + inspect.Parameter.KEYWORD_ONLY, + inspect.Parameter.POSITIONAL_ONLY)] + unfed = [n for n in required if n not in wanted] + if unfed: + findings.append( + f"{_name(impl)}{sig} requires {unfed}, which {cls.__name__} does not declare as an " + f"input. Either it is a Variable the Step should consume — in which case the Plan is " + f"missing an edge — or it is a dependency, which belongs in the implementation's " + f"closure, not in the dataflow.") + return findings + + +def _name(impl: Implementation) -> str: + return getattr(impl, "__name__", repr(impl)) diff --git a/plan_types/plan/bindings.py b/plan_types/plan/bindings.py index 477bbc9..be34777 100644 --- a/plan_types/plan/bindings.py +++ b/plan_types/plan/bindings.py @@ -22,7 +22,7 @@ from dataclasses import dataclass from typing import TYPE_CHECKING, Any -from plan_types.plan.step import Step +from plan_types.plan.step import Step, wired_inputs, wired_outputs from plan_types.plan.variable import Variable if TYPE_CHECKING: @@ -52,7 +52,7 @@ def producers(plan: Plan) -> dict[Variable[Any], list[Step]]: """ out: dict[Variable[Any], list[Step]] = {} for s in plan.steps: - for v in s.outputs: + for v in wired_outputs(s): out.setdefault(v, []).append(s) return out @@ -61,7 +61,7 @@ def consumers(plan: Plan) -> dict[Variable[Any], list[Step]]: """Variable -> the Steps that consume it. Many is legal: `isInputVarOf` is not functional.""" out: dict[Variable[Any], list[Step]] = {} for s in plan.steps: - for v in s.inputs: + for v in wired_inputs(s): out.setdefault(v, []).append(s) return out @@ -85,7 +85,7 @@ def unbound_inputs(plan: Plan) -> tuple[tuple[Step, Variable[Any]], ...]: """ prod = producers(plan) declared = set(plan.declared_inputs) - return tuple((s, v) for s in plan.steps for v in s.inputs + return tuple((s, v) for s in plan.steps for v in wired_inputs(s) if v not in prod and v not in declared) @@ -96,7 +96,7 @@ def orphans(plan: Plan) -> tuple[Variable[Any], ...]: Variable that is neither — which today means a Step declaring an output nothing reads while the Plan does not expose it either. Kept as its own function so the invariant can say WHICH. """ - consumed = {v for s in plan.steps for v in s.inputs} + consumed = {v for s in plan.steps for v in wired_inputs(s)} terminal = set(plan.outputs) return tuple(v for v, _ in producers(plan).items() if v not in consumed and v not in terminal) @@ -115,7 +115,7 @@ def execution_order(plan: Plan) -> tuple[Step, ...]: from plan_types.plan.plan import PlanError # noqa: PLC0415 — avoids an import cycle prod = producers(plan) - pending = {s: {p for v in s.inputs for p in prod.get(v, ()) if p is not s} + pending = {s: {p for v in wired_inputs(s) for p in prod.get(v, ()) if p is not s} for s in plan.steps} ordered: list[Step] = [] while pending: diff --git a/plan_types/plan/plan.py b/plan_types/plan/plan.py index 72b1f79..22ab7a5 100644 --- a/plan_types/plan/plan.py +++ b/plan_types/plan/plan.py @@ -41,7 +41,7 @@ from dataclasses import dataclass from typing import Any, ClassVar, Sequence -from plan_types.plan.step import Step +from plan_types.plan.step import Step, wired_inputs, wired_outputs from plan_types.plan.service import Service from plan_types.plan.variable import Variable @@ -106,7 +106,7 @@ def variables(self) -> tuple[Variable[Any], ...]: """Every Variable any Step consumes or produces, in first-seen order.""" seen: dict[Variable[Any], None] = {} for s in self.steps: - for v in (*s.inputs, *s.outputs): + for v in (*wired_inputs(s), *wired_outputs(s)): seen.setdefault(v, None) return tuple(seen) @@ -118,16 +118,16 @@ def inputs(self) -> tuple[Variable[Any], ...]: """ if self.declared_inputs: return self.declared_inputs - produced = {v for s in self.steps for v in s.outputs} + produced = {v for s in self.steps for v in wired_outputs(s)} return tuple(v for v in self.variables if v not in produced - and any(v in s.inputs for s in self.steps)) + and any(v in wired_inputs(s) for s in self.steps)) @property def outputs(self) -> tuple[Variable[Any], ...]: """The Plan's TERMINAL variables — produced here, consumed by nothing here.""" - consumed = {v for s in self.steps for v in s.inputs} + consumed = {v for s in self.steps for v in wired_inputs(s)} return tuple(v for v in self.variables if v not in consumed - and any(v in s.outputs for s in self.steps)) + and any(v in wired_outputs(s) for s in self.steps)) @property def used_services(self) -> tuple[Service, ...]: @@ -169,15 +169,65 @@ def check_arms(arms: Sequence[Plan]) -> None: @dataclass(frozen=True) -class MultiStep(Plan): +class MultiStep(Plan, Step): """[p-plan:MultiStep](http://purl.org/net/p-plan#MultiStep) — a Plan that appears as a Step. `rdfs:subClassOf` **both** `p-plan:Plan` and `p-plan:Step`, bound to its definition by `isDecomposedAsPlan`. This is how a Plan nests: one node in an outer Plan, a whole Plan inside. + + ## ⚠️ It did not inherit from `Step` until 2026-08-20, so it could not BE one + + It subclassed `Plan` alone, and `Plan.__post_init__` requires every entry in `steps` to be a + `Step` — so putting a MultiStep inside a Plan raised. The one thing the class exists for did not + work, for as long as the class existed, because nothing ever nested a Plan. The ontology + citation was right and the code did not implement it: the failure + `provenance.grounded_citation` exists to catch, in the class that carries the IRI. + + Ports are DERIVED — `Plan.inputs` and `Plan.outputs` already mean free and terminal Variables, + which is exactly what this node consumes and produces. Declaring them again would be a second + source of truth that goes stale the moment a nested Step moves. + + ## `until` — the termination predicate, declared but not implemented here + + An iterative region cannot be linearised, and the reason is not the toposort: a Plan that says + `WriteEmail -> Feedback -> WriteEmail` never says when to STOP, so no scheduler could run it + either. `until` names the Variable whose value decides. + + class ReviseUntilApproved(MultiStep): ... + ReviseUntilApproved(name="revise", steps=(...), until=approved) + + Plan-time: it declares WHICH Variable governs termination. The test itself is a function and + therefore an implementation, so it lives in the `Strategy` alongside every other one — the same + split as `Step`, for the same reason. This keeps `Decision`, branching and `End` out of the Plan + layer, where pydantic-graph and LangGraph already own them and do them well. + + ⚠️ `until` is OURS, deliberately uncited. P-Plan's 18 terms are Plan/Step/Variable structure and + PROV-O describes executions; neither has a word for "this planned region repeats until". Citing + one that does not say it is the failure this package was built to report. """ + #: The Variable whose value ends the iteration. `None` means this MultiStep is not iterative — + #: it is plain nesting, and `topology.terminating_iteration` refuses a cyclic inner Plan without + #: one, because a non-terminating declaration is not a specification. + until: Variable[Any] | None = None + ONTOLOGY_IRI: ClassVar[str] = "http://purl.org/net/p-plan#MultiStep" def decomposed_as_plan(self) -> Plan: """p-plan:isDecomposedAsPlan — the Plan this step expands into.""" return Plan(name=self.name, steps=self.steps) + + @property + def iterative(self) -> bool: + """Does the inner Plan contain a cycle? Read from the bindings, never declared.""" + from plan_types.plan.bindings import execution_order # noqa: PLC0415 — import cycle + + try: + execution_order(self.decomposed_as_plan()) + except PlanError: + return True + return False + + def __repr__(self) -> str: + tail = f", until={self.until.name!r}" if self.until else "" + return f"MultiStep({self.name!r}, {len(self.steps)} steps{tail})" diff --git a/plan_types/plan/step.py b/plan_types/plan/step.py index 3b2f87b..bae77e4 100644 --- a/plan_types/plan/step.py +++ b/plan_types/plan/step.py @@ -29,6 +29,34 @@ Not `consumes`/`produces`, which were ours and which drifted; not `scope`, which would describe an executor's name lookup rather than the relation — a Step input is **bound to a typed Variable**, not resolved from an environment. + +## ⚠️ `run()` was here, and removing it is the point + +A Step DECLARES a transformation. It does not perform one. `run()` said otherwise for as long as it +existed, and said it in the place that makes the claim structural: + + def run(self, **values: Any) -> Any: # the implementation is A METHOD ON THIS CLASS + raise NotImplementedError # so there is one, and any second is an override + +That signature asserts one implementation per Step. There is not one. `ExtractClaims` is a stable +operation with several ways to perform it — a cheap model, an expensive one, an ensemble — and none +of them is the real one the others override. Which one runs is chosen by a `Strategy`, outside the +declaration and per execution: `plan_types.execution`. + +Declaring `ExtractClaimsV1` and `ExtractClaimsV2` as separate Steps instead is not a workaround. It +is `naming.naming_drift` — several names for one concept — which is the failure this package exists +to report. + +What the removal cost, measured before it was made rather than argued afterwards: + + nothing called it not in plan_types/, tests/, evals/ or examples/ + the sole "impl" examples/evidence_case_graph/flow.py took a POSITIONAL `value` against + the keywords-only contract below — and did not import at all, so + neither fault was reachable by any check + downstream 15 overrides in nobsmed's plans.py, every one of them raising + +`run` is RETIRED rather than merely deleted — see `_RETIRED`. Deleting it would let a subclass +declare `run` again and get the privileged implementation back with nothing to say so. """ from __future__ import annotations @@ -42,11 +70,15 @@ class Step(Generic[InputT, OutputT]): - """Subclass it, declare `inputs` and `outputs`, implement `run`. + """Subclass it and declare `inputs` and `outputs`. That is the whole of it. Both are class-level because they are the DECLARATION: the shape of a pipeline must be readable without constructing anything, which is what lets a Plan be validated before a single Step has been implemented. + + ⚠️ There is no `run`. A Step says WHAT transformation exists; a `Strategy` says HOW it is + performed, and a `StepRunner` performs it — `plan_types.execution`. A Step you cannot execute + is not an unfinished Step, it is a Step nobody has bound yet, and the two must not look alike. """ #: P-Plan grounding. Checked — see `ontologies/invariants.GroundedCitation`, which exists @@ -59,31 +91,78 @@ class Step(Generic[InputT, OutputT]): #: p-plan:hasOutputVar — 0..N. outputs: ClassVar[tuple[Variable[Any], ...]] = () + #: Their `.map()`, declared. `(source, collected)` — two LIST Variables. The Step itself stays + #: the per-item operation, so `Square` is `int -> int` exactly as their `square` is, and the + #: Plan wires `numbers -> Square -> squares`. + #: + #: class Square(Step): + #: inputs, outputs = (number,), (squared,) # int -> int, ONE item + #: map_over = (numbers, squares) # list[int] in, list[int] out + #: + #: The vocabulary is theirs — `map`, `join`, `reducer` — because a Plan that fans out is + #: describing the same thing pydantic-graph and LangGraph already have words for, and inventing + #: a third word for it would be the drift this package reports. + #: + #: ⚠️ Why not put it on the edge, where they put it: our edges are DERIVED. Two Steps are + #: connected when they share a Variable, so there is no edge object to hang `.map()` on. The + #: Step is the only declared thing in the neighbourhood. + map_over: ClassVar[tuple[Variable[Any], Variable[Any]] | None] = None + #: Services this Step needs REACHABLE. Not values, not edges — see `service.py`. Every entry #: must appear in the owning Plan's `services`, enforced by `topology.declared_services`, #: which is the docker-compose property that stops the name meaning three things. uses: ClassVar[tuple["Service", ...]] = () - #: The names this class used before 2026-08-16. A Step still declaring them is not a Step with - #: an extra attribute — it is a Step whose inputs and outputs silently default to (), which - #: makes its Plan report no ports and `check_arms` agree that two empty shapes match. + #: Names this class used to carry, mapped to what replaced them. A subclass still declaring one + #: fails AT IMPORT — because in every case here the name still parses, and a name that parses + #: while meaning nothing is worse than one that breaks the build. + #: + #: `consumes`/`produces` were singular and retired 2026-08-16. Observed downstream the moment + #: v0.3.0 landed: both nobsmed arms became `shape=((), ())` and nothing raised, because `inputs` + #: and `outputs` simply defaulted to (). #: - #: Observed downstream the moment v0.3.0 landed: both nobsmed arms became `shape=((), ())` and - #: nothing raised. A retired name that still parses is worse than one that breaks the import. - _RETIRED: ClassVar[dict[str, str]] = {"consumes": "inputs", "produces": "outputs"} + #: `run` is the same shape of failure with a different consequence — it does not silently empty + #: the declaration, it silently re-privileges one implementation. See the module docstring. + _RETIRED: ClassVar[dict[str, str]] = { + "consumes": "inputs", "produces": "outputs", "run": "a Strategy", + } + + #: Why each retired name is retired, in the error a human has to act on. Kept beside the mapping + #: rather than generated from it: the previous version built the sentence with + #: `'Input' if old == 'consumes' else 'Output'`, which silently told anyone retiring a THIRD + #: name that it was about `hasOutputVar`. + _RETIRED_WHY: ClassVar[dict[str, str]] = { + "consumes": + "`inputs` is a TUPLE of Variables. p-plan:hasInputVar carries no cardinality " + "restriction, so a Step has 0..N. Leaving `consumes` in place does not error at " + "import: `inputs` defaults to () and the Plan reports no ports at all.", + "produces": + "`outputs` is a TUPLE of Variables. p-plan:hasOutputVar carries no cardinality " + "restriction, so a Step has 0..N. Leaving `produces` in place does not error at " + "import: `outputs` defaults to () and the Plan reports no ports at all.", + "run": + "a Step DECLARES a transformation; it does not perform one. A method here is one " + "implementation privileged over every other, so a second way of doing the same " + "operation becomes an override rather than a peer. Bind implementations with a " + "Strategy instead — `from plan_types.execution import LocalRunner, execute` — which " + "is per execution, so one Plan can run several ways without being edited.", + } def __init_subclass__(cls, **kw: Any) -> None: super().__init_subclass__(**kw) for old, new in cls._RETIRED.items(): if old in cls.__dict__: raise TypeError( - f"{cls.__name__}.{old} is retired — use `{new}`, a TUPLE of Variables. " - f"p-plan:has{'Input' if old == 'consumes' else 'Output'}Var carries no " - f"cardinality restriction, so a Step has 0..N of each. Leaving `{old}` in place " - f"does not error at import: `{new}` defaults to () and the Plan reports no " - f"ports at all.") + f"{cls.__name__}.{old} is retired — use {new}. {cls._RETIRED_WHY[old]}") for field in ("inputs", "outputs"): value = getattr(cls, field, ()) + # ⚠️ DERIVED ports are legal and are not tuples. `MultiStep` is a Plan that is also a + # Step, and its ports are its inner Plan's free and terminal Variables — computed, not + # declared, because declaring them a second time is the two-sources-of-truth this + # package refuses everywhere else. A descriptor here means "derived"; leave it alone. + if isinstance(getattr(cls, "__dict__", {}).get(field, None), property) or isinstance( + value, property): + continue if isinstance(value, Variable): raise TypeError( f"{cls.__name__}.{field} is a single Variable; it must be a tuple. P-Plan puts " @@ -93,22 +172,35 @@ def __init_subclass__(cls, **kw: Any) -> None: raise TypeError( f"{cls.__name__}.{field} must be a tuple of Variables, got {value!r}") + if cls.__dict__.get("map_over") is not None: + m = cls.map_over + if not (isinstance(m, tuple) and len(m) == 2 + and all(isinstance(v, Variable) for v in m)): + raise TypeError( + f"{cls.__name__}.map_over must be (source, collected) — two list Variables, " + f"the one fanned out and the one collected into. Got {m!r}.") + if len(cls.inputs) != 1 or len(cls.outputs) != 1: + raise TypeError( + f"{cls.__name__} maps over {m[0].name!r}, so it is the PER-ITEM operation and " + f"must declare exactly one input and one output — theirs is `square: int -> " + f"int`. Got {len(cls.inputs)} in, {len(cls.outputs)} out. A mapped Step with a " + f"fan-in has no meaning: there is no second list to zip against.") + if m[0] in cls.inputs or m[1] in cls.outputs: + raise TypeError( + f"{cls.__name__}.map_over names the LIST Variables, and inputs/outputs name the " + f"ITEM. Using the same Variable for both says the step consumes the whole list " + f"and one of its items at once.") + # Duplicate names within one side would make a binding ambiguous, and the ambiguity would # surface as the wrong value arriving rather than as an error here. for field in ("inputs", "outputs"): - names = [v.name for v in getattr(cls, field, ())] + value = getattr(cls, field, ()) + if isinstance(value, property): # derived — see the note above + continue + names = [v.name for v in value] if len(names) != len(set(names)): raise TypeError(f"{cls.__name__}.{field} names a Variable twice: {names}") - def run(self, **values: Any) -> Any: - """Execute. Keyword arguments are the input Variables, by name. - - ⚠️ Keywords, not positional. A Step with three inputs called positionally is one argument - reorder away from a silent mis-wire, and the type check that justifies this package would - not catch it when two inputs share a type. - """ - raise NotImplementedError - @classmethod def shape(cls) -> tuple[tuple[type, ...], tuple[type, ...]]: """`(input types, output types)`. Two Steps with equal shapes are substitutable.""" @@ -118,3 +210,21 @@ def __repr__(self) -> str: ins = ", ".join(v.name for v in self.inputs) or "-" outs = ", ".join(v.name for v in self.outputs) or "-" return f"{type(self).__name__}({ins} -> {outs})" + + +def wired_inputs(step: Any) -> tuple[Variable[Any], ...]: + """What the PLAN connects to this Step's input side. + + Equal to `step.inputs` unless the Step maps, in which case the Plan sees the LIST it fans out + from while the implementation sees one item. Everything structural — bindings, every topology + and typing invariant, the diagram, the execution order — reads this, so a mapped Step is an + ordinary node in all of them and no invariant has to learn about `map_over`. + """ + m = getattr(step, "map_over", None) + return (m[0],) if m else tuple(step.inputs) + + +def wired_outputs(step: Any) -> tuple[Variable[Any], ...]: + """What the PLAN connects to this Step's output side. See `wired_inputs`.""" + m = getattr(step, "map_over", None) + return (m[1],) if m else tuple(step.outputs) diff --git a/plan_types/plan/visualization.py b/plan_types/plan/visualization.py index 29c7f37..e32c253 100644 --- a/plan_types/plan/visualization.py +++ b/plan_types/plan/visualization.py @@ -17,6 +17,7 @@ from plan_types.plan import bindings from plan_types.plan.plan import Plan +from plan_types.plan.step import wired_inputs, wired_outputs from plan_types.plan.variable import Variable _SPLIT = re.compile(r"(?<=[a-z0-9])(?=[A-Z])|(?<=[A-Z])(?=[A-Z][a-z])") @@ -41,16 +42,36 @@ def _node(plan: Plan, index: int) -> str: return f"{re.sub(r'[^A-Za-z0-9_]', '_', plan.name)}_{index}" +def _impl_label(step: object, strategy: Any) -> str: + """The bound implementation's name, on a second line, or nothing. + + ⚠️ Nothing when a Step is UNBOUND — deliberately not "unbound" or "?". A diagram that invents a + label for a Step nobody has implemented is asserting something about the Strategy; the absence + says only that this render was given no Strategy for it, which is the truth. + """ + if strategy is None: + return "" + impl = strategy.get(type(step)) + if impl is None: + return "" + return f"
{getattr(impl, '__name__', repr(impl))}" + + def _port(prefix: str, v: Variable[Any]) -> str: return f"{prefix}_{re.sub(r'[^A-Za-z0-9_]', '_', v.name)}" -def render_mermaid(plan: Plan) -> str: +def render_mermaid(plan: Plan, strategy: Any = None) -> str: """A mermaid `flowchart TD` of the Plan's actual structure. Free Variables enter as ports, terminal Variables leave as ports, and every internal arrow is a producer→consumer edge labelled with the Variable that flows. A Step with three inputs shows three arrows — which is the whole reason this reads from bindings rather than from step order. + + Pass a `Strategy` and each node also names the implementation bound to it. Render the same Plan + under two Strategies and the diagrams are IDENTICAL except for those names — which is the claim + "the logical process was held constant, only the implementation changed", drawn instead of + asserted. A reader can check it by looking, which is not true of two hand-made diagrams. """ idx = {id(s): i for i, s in enumerate(plan.steps)} lines = ["```mermaid", "flowchart TD"] @@ -58,13 +79,13 @@ def render_mermaid(plan: Plan) -> str: for v in plan.inputs: lines.append(f' {_port("IN", v)}(["{v.name}: {_type_name(v.type)}"])') for i, s in enumerate(plan.steps): - lines.append(f' {_node(plan, i)}["{_label(s)}"]') + lines.append(f' {_node(plan, i)}["{_label(s)}{_impl_label(s, strategy)}"]') for v in plan.outputs: lines.append(f' {_port("OUT", v)}(["{v.name}: {_type_name(v.type)}"])') free = set(plan.inputs) for i, s in enumerate(plan.steps): - for v in s.inputs: + for v in wired_inputs(s): if v in free: lines.append(f' {_port("IN", v)} -- {v.name} --> {_node(plan, i)}') @@ -75,7 +96,7 @@ def render_mermaid(plan: Plan) -> str: terminal = set(plan.outputs) for i, s in enumerate(plan.steps): - for v in s.outputs: + for v in wired_outputs(s): if v in terminal: lines.append(f' {_node(plan, i)} --> {_port("OUT", v)}') diff --git a/pyproject.toml b/pyproject.toml index 13dac5c..8cb63d4 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -10,6 +10,11 @@ dependencies = ["pydantic>=2.0,<3"] [project.scripts] conceptlint = "conceptlint.core.lint:main" +[project.optional-dependencies] +pydantic-graph = [ + "pydantic-graph>=2.0", +] + [dependency-groups] dev = ["pytest>=8.0"] diff --git a/scripts/check_wheel.py b/scripts/check_wheel.py index f47c6fd..c0168b8 100644 --- a/scripts/check_wheel.py +++ b/scripts/check_wheel.py @@ -18,6 +18,7 @@ "plan_types", "plan_types.invariants", "plan_types.plan", + "plan_types.execution", "conceptlint.core.lint", ] diff --git a/tests/test_eval_trial.py b/tests/test_eval_trial.py index ca311c5..82f65c6 100644 --- a/tests/test_eval_trial.py +++ b/tests/test_eval_trial.py @@ -28,7 +28,6 @@ class Other: def _plan(name: str, out=Graph) -> Plan: step = type(f"S{name}", (Step,), { "inputs": (Variable("paste", Paste),), "outputs": (Variable("out", out),), - "run": lambda self, v: out(), }) return Plan(name=name, steps=(step(),)) diff --git a/tests/test_execution.py b/tests/test_execution.py new file mode 100644 index 0000000..652416b --- /dev/null +++ b/tests/test_execution.py @@ -0,0 +1,222 @@ +"""A Step declares; a Strategy implements; a Runner performs. + +Each test names the failure it would catch. The ones worth reading twice are `test_async_refused` +and `test_examples_import` — both exist because the thing they check was silently broken and no +test could see it. +""" +from __future__ import annotations + +import pytest + +from plan_types import Plan, Step, Variable +from plan_types.execution import (ExecutionError, LocalRunner, StepRunner, check_strategy, + execute) + +document = Variable("document", str) +outline = Variable("outline", tuple) +summary = Variable("summary", str) + + +class BuildIndex(Step): + inputs, outputs = (document,), (outline,) + + +class Condense(Step): + inputs, outputs = (document, outline), (summary,) + + +plan = Plan(name="t", steps=(BuildIndex(), Condense()), + declared_inputs=(document,)) + + +def index_impl(document: str) -> tuple: + return tuple(document.split()) + + +def condense_short(document: str, outline: tuple) -> str: + return outline[0] if outline else document + + +def condense_long(document: str, outline: tuple) -> str: + return " ".join(outline) + + +short = {BuildIndex: index_impl, Condense: condense_short} +long_ = {BuildIndex: index_impl, Condense: condense_long} + + +# ── the declaration carries no implementation ──────────────────────────────────────────────────── + +def test_step_has_no_run() -> None: + """The removal itself. A `run` reachable on Step is one implementation privileged over peers.""" + assert not hasattr(Step, "run") + assert not hasattr(BuildIndex(), "run") + + +def test_declaring_run_fails_at_import() -> None: + """Retired, not merely deleted — otherwise the privileged implementation comes back silently.""" + with pytest.raises(TypeError, match="retired"): + class Rebel(Step): + inputs, outputs = (document,), (summary,) + + def run(self, **values): # noqa: ANN001, ANN201 + return "x" + + +def test_retired_message_names_its_own_replacement() -> None: + """The old error built its text with `'Input' if old == 'consumes' else 'Output'`, so a THIRD + retired name would have been explained as being about hasOutputVar.""" + with pytest.raises(TypeError, match="Strategy"): + class Rebel2(Step): + inputs, outputs = (document,), (summary,) + + def run(self, **values): # noqa: ANN001, ANN201 + return "x" + + with pytest.raises(TypeError, match="hasInputVar"): + class Old(Step): + consumes = document + + +# ── one Plan, several strategies ───────────────────────────────────────────────────────────────── + +def test_same_plan_two_strategies_differ() -> None: + """The capability the whole change exists for: the Plan object is not touched between arms.""" + a = execute(plan, {"document": "one two three"}, LocalRunner(short)) + b = execute(plan, {"document": "one two three"}, LocalRunner(long_)) + assert a["summary"] == "one" + assert b["summary"] == "one two three" + + +def test_execute_returns_intermediates_too() -> None: + env = execute(plan, {"document": "a b"}, LocalRunner(short)) + assert env["outline"] == ("a", "b") + + +def test_local_runner_satisfies_the_protocol_structurally() -> None: + """No inheritance. An adapter in another package conforms without importing plan_types.""" + assert isinstance(LocalRunner({}), StepRunner) + + +# ── what the runner refuses, loudly ────────────────────────────────────────────────────────────── + +def test_missing_binding_names_the_step() -> None: + with pytest.raises(ExecutionError, match="Condense"): + execute(plan, {"document": "a"}, LocalRunner({BuildIndex: index_impl})) + + +def test_async_refused() -> None: + """A coroutine is truthy, has a repr, and flows onward as though it were data.""" + async def condense_async(document: str, outline: tuple) -> str: + return "x" + + strategy = {BuildIndex: index_impl, Condense: condense_async} + with pytest.raises(ExecutionError, match="coroutine"): + execute(plan, {"document": "a"}, LocalRunner(strategy)) + + +def test_wrong_output_type_refused() -> None: + strategy = {BuildIndex: index_impl, Condense: lambda document, outline: 42} + with pytest.raises(ExecutionError, match="declares 'summary' as str"): + execute(plan, {"document": "a"}, LocalRunner(strategy)) + + +def test_multi_output_arity_refused() -> None: + left, right = Variable("left", str), Variable("right", str) + + class Split(Step): + inputs, outputs = (document,), (left, right) + + two = Plan(name="two", steps=(Split(),), declared_inputs=(document,)) + with pytest.raises(ExecutionError, match="must return a tuple of 2"): + execute(two, {"document": "a"}, LocalRunner({Split: lambda document: "only one"})) + + env = execute(two, {"document": "a"}, LocalRunner({Split: lambda document: ("l", "r")})) + assert (env["left"], env["right"]) == ("l", "r") + + +def test_output_from_a_step_declaring_none_refused() -> None: + class Audit(Step): + inputs, outputs = (document,), () + + p = Plan(name="a", steps=(Audit(),), declared_inputs=(document,)) + with pytest.raises(ExecutionError, match="declares no outputs"): + execute(p, {"document": "a"}, LocalRunner({Audit: lambda document: "leaked"})) + + assert execute(p, {"document": "a"}, LocalRunner({Audit: lambda document: None})) == { + "document": "a"} + + +def test_missing_plan_input_named() -> None: + with pytest.raises(ExecutionError, match=r"expects \['document'\]"): + execute(plan, {}, LocalRunner(short)) + + +def test_two_variables_one_name_refused() -> None: + """A Variable is (name, type), so two can differ in type and collide in a name-keyed env.""" + same_name_other_type = Variable("outline", list) + + class Other(Step): + inputs, outputs = (document,), (same_name_other_type,) + + p = Plan(name="clash", steps=(BuildIndex(), Other()), declared_inputs=(document,)) + with pytest.raises(ExecutionError, match="sharing the name"): + execute(p, {"document": "a"}, LocalRunner({})) + + +# ── check_strategy: static, and honest about what it cannot see ────────────────────────────────── + +def test_check_strategy_clean() -> None: + assert check_strategy(plan, short) == () + + +def test_check_strategy_reports_every_problem_not_the_first() -> None: + findings = check_strategy(plan, {}) + assert len(findings) == 2 + assert any("BuildIndex" in f for f in findings) + assert any("Condense" in f for f in findings) + + +def test_check_strategy_catches_a_parameter_name_mismatch() -> None: + """`Variable.name` is part of the contract, because the runner calls by keyword.""" + findings = check_strategy(plan, {BuildIndex: index_impl, + Condense: lambda doc, outline: "x"}) + assert any("will not accept by keyword" in f for f in findings) + + +def test_check_strategy_catches_an_undeclared_required_parameter() -> None: + findings = check_strategy(plan, {BuildIndex: index_impl, + Condense: lambda document, outline, model: "x"}) + assert any("'model'" in f and "does not declare" in f for f in findings) + + +def test_check_strategy_catches_async_before_execution() -> None: + async def condense_async(document: str, outline: tuple) -> str: + return "x" + + findings = check_strategy(plan, {BuildIndex: index_impl, Condense: condense_async}) + assert any("async" in f for f in findings) + + +def test_check_strategy_says_not_checked_when_it_cannot_read_a_signature() -> None: + """NOT CHECKED and "checked, fine" must never render the same. + + `min` is the stand-in because C builtins are the real case: `inspect.signature(min)` raises + ValueError, and the tempting `return []` there would report a clean Strategy for a callable + nobody had looked at. + """ + findings = check_strategy(plan, {BuildIndex: index_impl, Condense: min}) + assert any("NOT CHECKED" in f for f in findings) + + +# ── the examples are code, so they have to run ─────────────────────────────────────────────────── + +def test_examples_import_and_run() -> None: + """`examples/evidence_case_graph/flow.py` raised TypeError on import for two months — it still + declared `consumes`, retired 2026-08-16 — and nothing noticed, because no test touched it. An + example nobody imports is not an example, it is a claim.""" + from examples.evidence_case_graph import flow as evidence + from examples.hello import flow as hello + + hello.main() + evidence.main() diff --git a/tests/test_plan_types.py b/tests/test_plan_types.py index a46851b..cfd2a90 100644 --- a/tests/test_plan_types.py +++ b/tests/test_plan_types.py @@ -30,14 +30,14 @@ class Extract(Step): inputs, outputs = (PAPER,), (FINDINGS,) -class Summarize(Step): +class SummarizePaper(Step): """Fans in: needs the paper AND the findings.""" inputs, outputs = (PAPER, FINDINGS), (SUMMARY,) def a_plan() -> Plan: - return Plan(name="extract_and_summarize", steps=(Extract(), Summarize())) + return Plan(name="extract_and_summarize", steps=(Extract(), SummarizePaper())) # ── the Plan answers the eight questions ───────────────────────────────────────────────────────── @@ -56,13 +56,13 @@ def test_inputs_are_the_free_variables_not_the_first_step() -> None: def test_a_step_may_consume_many_variables() -> None: """p-plan:hasInputVar carries no cardinality restriction — fan-in is the ordinary case.""" - assert len(Summarize.inputs) == 2 + assert len(SummarizePaper.inputs) == 2 def test_declaration_order_carries_no_meaning() -> None: """Order is derived from the bindings; declaring backwards must not change it.""" forward = bindings.execution_order(a_plan()) - backward = bindings.execution_order(Plan(name="x", steps=(Summarize(), Extract()))) + backward = bindings.execution_order(Plan(name="x", steps=(SummarizePaper(), Extract()))) assert [type(s).__name__ for s in forward] == [type(s).__name__ for s in backward] @@ -148,7 +148,7 @@ class Needs(Step): def test_a_DECLARED_plan_input_is_NOT_unbound() -> None: """The PASS twin. A Plan's own signature must not read as an error.""" - plan = Plan(name="ok", steps=(Extract(), Summarize()), declared_inputs=(PAPER,)) + plan = Plan(name="ok", steps=(Extract(), SummarizePaper()), declared_inputs=(PAPER,)) assert validate(plan, [topology.BOUND_INPUTS]) == [] @@ -157,7 +157,7 @@ def test_two_producers_for_one_variable_is_reported() -> None: class AlsoExtract(Step): inputs, outputs = (PAPER,), (FINDINGS,) - found = validate(Plan(name="dup", steps=(Extract(), AlsoExtract(), Summarize())), + found = validate(Plan(name="dup", steps=(Extract(), AlsoExtract(), SummarizePaper())), [topology.SINGLE_PRODUCER]) assert found and "findings" in found[0].message @@ -228,7 +228,7 @@ class PICOSetBuilder(Step): # ── p-plan:MultiStep — a Plan that is also a Step ──────────────────────────────────────────────── def test_multistep_is_both_a_plan_and_decomposable() -> None: - ms = MultiStep(name="inner", steps=(Extract(), Summarize())) + ms = MultiStep(name="inner", steps=(Extract(), SummarizePaper())) assert isinstance(ms, Plan) assert ms.decomposed_as_plan().shape() == ms.shape() @@ -243,21 +243,27 @@ def test_the_readme_example_runs_and_returns_what_it_claims() -> None: package is about — a confident claim with nothing checking it. """ import pathlib - import re + + from examples.hello.flow import fast, plan, precise readme = (pathlib.Path(__file__).resolve().parents[1] / "README.md").read_text() - assert "declared_inputs=(PAPER,)" in readme, ( + assert "declared_inputs=(document,)" in readme, ( "the 60-second example must declare its inputs, or the validate() it claims is wrong") - plan = Plan(name="extract_and_summarize", steps=(Extract(), Summarize()), - declared_inputs=(PAPER,)) assert validate(plan, [*topology.ALL, *typing_inv.ALL]) == [], ( "the README claims this returns []") - # And the shape of the diagram it prints. - out = render_mermaid(plan) - for edge in ("IN_paper -- paper -->", "-- findings -->", "--> OUT_summary"): - assert edge in out, f"README shows {edge!r}, renderer does not produce it" + # ⚠️ The diagram must be BYTE-IDENTICAL to what the renderer produces. The previous version of + # this test rebuilt an equivalent Plan from local fixtures and grepped for three edges, so a + # hand-edited README block passed as long as it contained them — and it was hand-edited: the + # committed block used node ids `s0`/`s1`, which this renderer has never emitted. + assert render_mermaid(plan) in readme, ( + "the README's mermaid block is not what render_mermaid(plan) returns — regenerate it " + "rather than editing it, since a hand-drawn diagram stops being true silently") + + # And the claim that matters: one Plan, two strategies, without touching the Plan. + assert fast is not precise + assert set(fast) == set(precise), "both arms must implement the same Steps" def test_the_readme_does_not_claim_unbuilt_integrations() -> None: @@ -273,3 +279,78 @@ def test_the_readme_does_not_claim_unbuilt_integrations() -> None: for unbuilt in ("Temporal", "persistence", "agent-hook"): assert unbuilt in readme.split("Not built:")[1][:400], ( f"{unbuilt} is not implemented; Status must say so") + + +# --- MultiStep: a Plan that is also a Step ------------------------------------------------------- + +def test_a_multistep_can_actually_be_a_step() -> None: + """It could not until 2026-08-20 — the one thing the class exists for. + + It subclassed Plan alone, so `Plan.__post_init__`'s isinstance(s, Step) check rejected it. The + ontology citation said rdfs:subClassOf BOTH; the code implemented one. Nothing caught it + because nothing had ever nested a Plan. + """ + from plan_types.plan.plan import MultiStep + + a, b, c = Variable("a", str), Variable("b", str), Variable("c", str) + + class In1(Step): + inputs, outputs = (a,), (b,) + + class In2(Step): + inputs, outputs = (b,), (c,) + + class After(Step): + inputs, outputs = (c,), (Variable("d", str),) + + nested = MultiStep(name="inner", steps=(In1(), In2())) + assert isinstance(nested, Step) + + outer = Plan(name="outer", steps=(nested, After())) + assert len(outer.steps) == 2 + + # Ports are DERIVED from the inner Plan, never declared twice. + assert [v.name for v in nested.inputs] == ["a"] + assert [v.name for v in nested.outputs] == ["c"] + + +def test_a_multistep_knows_whether_its_inner_plan_iterates() -> None: + """`until` names the Variable that ends the loop. Read the cycle, do not declare it.""" + from plan_types.plan.plan import MultiStep + + draft, critique = Variable("draft", str), Variable("critique", str) + + class Write(Step): + inputs, outputs = (critique,), (draft,) + + class Review(Step): + inputs, outputs = (draft,), (critique,) + + loop = MultiStep(name="revise", steps=(Write(), Review()), until=critique) + assert loop.iterative is True + assert loop.until is critique + assert "until='critique'" in repr(loop) + + straight = MultiStep(name="chain", steps=(Write(),)) + assert straight.iterative is False + + +def test_the_readme_does_not_promise_rdf_it_does_not_have() -> None: + """The grounding is a CHECKED VOCABULARY, not a serialization format. + + The README says a Plan is legible to anything else that speaks P-Plan, and immediately says + there is no import or export. That second sentence is the one that stops the first from being + an overclaim, so both halves are asserted here: the disclaimer must be present, and no RDF + machinery may quietly appear that would make it stale in the other direction. + """ + import pathlib + + root = pathlib.Path(__file__).resolve().parents[1] + readme = (root / "README.md").read_text() + assert "no RDF import or export" in readme + + sources = " ".join(f.read_text() for f in (root / "plan_types").rglob("*.py")) + for term in ("rdflib", "to_turtle", "to_jsonld"): + assert term not in sources, ( + f"{term} exists now — the README's 'a door, not a feature' paragraph is stale and " + f"understates what ships") diff --git a/tests/test_pplan_grounding.py b/tests/test_pplan_grounding.py index 24c8a44..503dbc1 100644 --- a/tests/test_pplan_grounding.py +++ b/tests/test_pplan_grounding.py @@ -180,11 +180,9 @@ def test_our_plan_does_not_assume_a_linear_chain(): class First(Step): # needs TWO inputs, so first/last cannot describe the Plan inputs, outputs = (A, B), (C,) - def run(self, **v): return 1.0 class Second(Step): inputs, outputs = (C,), (D,) - def run(self, **v): return {} plan = Plan(name="fanin", steps=(First(), Second())) assert {v.name for v in plan.inputs} == {"a", "b"}, \ diff --git a/tests/test_pydantic_graph.py b/tests/test_pydantic_graph.py new file mode 100644 index 0000000..d69286e --- /dev/null +++ b/tests/test_pydantic_graph.py @@ -0,0 +1,180 @@ +"""The layering claim, executed. + +`plan_types.plan` must import no execution framework, and the same Plan under the same Strategy must +produce the same answer on `LocalRunner` and on Pydantic Graph. Both are assertions this repo makes +in prose; neither is worth anything unless something runs it. +""" +from __future__ import annotations + +import asyncio + +import pytest + +pytest.importorskip("pydantic_graph", reason="optional extra: uv sync --extra pydantic-graph") + +from plan_types.execution import (ExecutionError, LocalRunner, check_strategy, # noqa: E402 + execute) +from plan_types.execution.pydantic_graph import to_pydantic_graph # noqa: E402 + + +def test_both_runtimes_agree() -> None: + from examples.pydantic_graph_demo.flow import User, plan, terse, warm + + reader = User(name="Samuel", interests=("type safety", "graphs")) + for strategy in (terse, warm): + local = execute(plan, {"user": reader}, LocalRunner(strategy))["email"] + graph = to_pydantic_graph(plan, strategy) + assert asyncio.run(graph.run(state={}, inputs={"user": reader}))["email"] == local + + +def test_the_demo_runs() -> None: + from examples.pydantic_graph_demo.flow import main + + asyncio.run(main()) + + +def test_an_incomplete_strategy_is_refused_before_the_graph_is_built() -> None: + """Mid-run would attribute a Strategy fault to the runtime.""" + from examples.pydantic_graph_demo.flow import WriteEmail, plan, terse + + with pytest.raises(ExecutionError, match="Strategy is incomplete"): + to_pydantic_graph(plan, {WriteEmail: terse[WriteEmail]}) + + +def test_a_cyclic_plan_is_refused_rather_than_linearised() -> None: + """The case where you SHOULD use their engine, said out loud instead of faked.""" + from plan_types import Plan, Step, Variable + from plan_types.plan.plan import PlanError + + a, b = Variable("a", int), Variable("b", int) + + class Up(Step): + inputs, outputs = (a,), (b,) + + class Down(Step): + inputs, outputs = (b,), (a,) + + cyclic = Plan(name="loop", steps=(Up(), Down())) + with pytest.raises(PlanError, match="cycle"): + to_pydantic_graph(cyclic, {Up: lambda a: a, Down: lambda b: b}) + + +def test_the_plan_layer_imports_no_execution_framework() -> None: + """The layering claim itself. `plan/` must be readable with no engine in the room.""" + import pathlib + + plan_dir = pathlib.Path(__file__).resolve().parents[1] / "plan_types" / "plan" + for path in plan_dir.glob("*.py"): + source = path.read_text() + for engine in ("pydantic_graph", "temporal", "langgraph"): + assert engine not in source, f"{path.name} references {engine}" + + +# ── the staged comparison against their docs (examples/pydantic_graph_docs/) ───────────────────── + +def test_stage1_their_docs_example_round_trips() -> None: + """Their `simple_counter.py` verbatim, our Plan, and our Plan on their runtime all give 2.""" + from examples.pydantic_graph_docs import stage1_counter as s1 + + asyncio.run(s1.main()) + + +def test_stage2_three_strategies_over_one_plan() -> None: + from examples.pydantic_graph_docs import stage2_strategies as s2 + + asyncio.run(s2.main()) + + +def test_the_plan_object_is_shared_across_arms_not_copied() -> None: + """The claim an eval depends on: arms differ ONLY in the Strategy. + + Identity, not equality — a Plan rebuilt per arm could drift a Step and still compare equal + under a weaker check, which is the failure `AIEvalTrial` exists to prevent. + """ + from examples.pydantic_graph_docs.stage2_strategies import ARMS, plan + + for strategy in ARMS.values(): + assert check_strategy(plan, strategy) == () + assert len({id(plan) for _ in ARMS}) == 1 + + +def test_the_diagrams_differ_only_in_the_implementation_labels() -> None: + """The visual claim, checked: strip the labels and the three renders are byte-identical.""" + import re + + from plan_types import render_mermaid + from examples.pydantic_graph_docs.stage2_strategies import ARMS, plan + + strip = lambda t: re.sub(r"
[^<]+", "", t) # noqa: E731 + renders = [render_mermaid(plan, s) for s in ARMS.values()] + assert len({strip(r) for r in renders}) == 1, "topology differed between arms" + assert len(set(renders)) == len(ARMS), "the arms rendered identically — labels missing" + + +def test_stage3_and_the_control_arm_agree() -> None: + """The comparison is only worth reading if both arms compute the same thing.""" + from examples.pydantic_graph_docs import control_no_plan as ctrl + from examples.pydantic_graph_docs import stage3_map_join as s3 + from plan_types.execution import LocalRunner, execute + + for case in s3.CORPUS: + with_plan = [execute(s3.plan, {"numbers": case}, LocalRunner(a))["total"] + for a in s3.ARMS.values()] + control = [asyncio.run(ctrl.build(impl).run(inputs=case)) + for impl in ctrl.ARMS.values()] + assert with_plan == control, f"arms disagree on {case}: {with_plan} vs {control}" + + +def test_stage3_runs() -> None: + from examples.pydantic_graph_docs import stage3_map_join as s3 + + asyncio.run(s3.main()) + + +def test_a_mapped_step_gives_their_documented_answer() -> None: + """Their docs print `Results: [1, 4, 9, 16, 25]`. Both our runtimes must too — and the + compiled one gets there through their real `.map()` and join, not a loop wearing the name.""" + from examples.pydantic_graph_docs.stage3_map_join import ARMS, plan + from plan_types.execution import LocalRunner, execute + from plan_types.execution.pydantic_graph import to_pydantic_graph + + case = {"numbers": [1, 2, 3, 4, 5]} + assert execute(plan, case, LocalRunner(ARMS["exact"]))["squares"] == [1, 4, 9, 16, 25] + compiled = asyncio.run(to_pydantic_graph(plan, ARMS["exact"]).run(state={}, inputs=case)) + assert compiled["squares"] == [1, 4, 9, 16, 25] + + +def test_a_mapped_step_is_an_ordinary_node_to_every_invariant() -> None: + """Why `map_over` went into `wired_inputs` instead of into each invariant. + + The Plan sees `numbers -> Square -> squares`; only the runner sees the per-item call. The ITEM + Variable must never surface as a Plan port, or every topology rule would need to learn about + mapping and they would each learn it slightly differently. + """ + from plan_types import validate + from plan_types.invariants import topology, typing as typing_inv + from examples.pydantic_graph_docs.stage3_map_join import numbers, plan, squares + + assert validate(plan, [*topology.ALL, *typing_inv.ALL]) == [] + assert [v.name for v in plan.inputs] == ["numbers"] + assert numbers in plan.variables and squares in plan.variables + assert "number" not in [v.name for v in plan.variables], "the ITEM is not a Plan port" + + +def test_a_mapped_step_must_be_the_per_item_operation() -> None: + """A mapped fan-in has no meaning: there is no second list to zip against.""" + from plan_types import Step, Variable + + xs, x, y, ys = (Variable("xs", list), Variable("x", int), + Variable("y", int), Variable("ys", list)) + other = Variable("other", int) + + with pytest.raises(TypeError, match="exactly one input"): + class TwoIn(Step): + inputs, outputs = (x, other), (y,) + map_over = (xs, ys) + + with pytest.raises(TypeError, match="names the LIST"): + class Confused(Step): + inputs, outputs = (xs,), (y,) + map_over = (xs, ys) diff --git a/uv.lock b/uv.lock index ce04d05..c755add 100644 --- a/uv.lock +++ b/uv.lock @@ -11,6 +11,19 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/99/91/8acff4f5e50511b911bbccb72b8628a49c68ce14148cd9f6431094859a90/annotated_types-0.8.0-py3-none-any.whl", hash = "sha256:f072f4d804ea359e4eaf198b1af7a8b0943881a87f31bb764f8bf219bb9419e0", size = 13427, upload-time = "2026-07-23T20:16:12.938Z" }, ] +[[package]] +name = "anyio" +version = "4.14.2" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "idna" }, + { name = "typing-extensions", marker = "python_full_version < '3.13'" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/61/cc/a381afa6efea9f496eff839d4a6a1aed3bfafc7b3ab4b0d1b243a12573dd/anyio-4.14.2.tar.gz", hash = "sha256:cfa139f3ed1a23ee8f88a145ddb5ac7605b8bbfd8592baacd7ce3d8bb4313c7f", size = 260176, upload-time = "2026-07-12T20:29:07.082Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/da/35/f2287558c17e29fafc8ef3daf819bb9834061cfa43bff8014f7df7f63bdc/anyio-4.14.2-py3-none-any.whl", hash = "sha256:9f505dda5ac9f0c8309b5e8bd445a8c2bf7246f3ce950121e45ea15bc41d1494", size = 125813, upload-time = "2026-07-12T20:29:05.763Z" }, +] + [[package]] name = "colorama" version = "0.4.6" @@ -20,6 +33,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/d1/d6/3965ed04c63042e047cb6a3e6ed1a63a35087b6a609aa3a15ed8ac56c221/colorama-0.4.6-py2.py3-none-any.whl", hash = "sha256:4f1d9991f5acc0ca119f9d443620b77f9d6b33703e51011c16baf57afb285fc6", size = 25335, upload-time = "2022-10-25T02:36:20.889Z" }, ] +[[package]] +name = "idna" +version = "3.19" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/5f/f7/abb373e5757eaec4b922b92f97ec8d6d7e057cf06778247604fbc4e7c3f3/idna-3.19.tar.gz", hash = "sha256:5e0811a4383b21dc5838069f801c4fb62113b7447663d2530d2bd6e77b49bf15", size = 215237, upload-time = "2026-08-18T05:14:24.27Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/57/b0/0e52c878c53f245edd3a11020f20979b3f490f245af532c7cae3027754b5/idna-3.19-py3-none-any.whl", hash = "sha256:815e7be7a7806d54abb586dc943addc79e8b2ee16915059658cbeff4b1b43bf4", size = 68550, upload-time = "2026-08-18T05:14:22.343Z" }, +] + [[package]] name = "iniconfig" version = "2.3.0" @@ -29,6 +51,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/cb/b1/3846dd7f199d53cb17f49cba7e651e9ce294d8497c8c150530ed11865bb8/iniconfig-2.3.0-py3-none-any.whl", hash = "sha256:f631c04d2c48c52b84d0d0549c99ff3859c98df65b3101406327ecc7d53fbf12", size = 7484, upload-time = "2025-10-18T21:55:41.639Z" }, ] +[[package]] +name = "logfire-api" +version = "4.41.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/41/83/a2e7de43bb092ffaad904b5756cfc1e0ea4a8d79fdacd24cd55e60790585/logfire_api-4.41.0.tar.gz", hash = "sha256:ec39252acac38b5b50d60cfb9cc62f0ea10c841345fc59692af32dbe3de4a140", size = 92818, upload-time = "2026-08-20T17:42:24.302Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/c8/96/97552d0d742866719b3a6e8fd7e68dd2877804f4c3b10c44a19a1f99b0d6/logfire_api-4.41.0-py3-none-any.whl", hash = "sha256:c71010d086c0211b04b4640181e836e8a75cc635fcfa7f712557f7b0676c1413", size = 143003, upload-time = "2026-08-20T17:42:21.764Z" }, +] + [[package]] name = "packaging" version = "26.3" @@ -40,19 +71,28 @@ wheels = [ [[package]] name = "plan-types" -version = "0.5.0" +version = "0.6.0" source = { editable = "." } dependencies = [ { name = "pydantic" }, ] +[package.optional-dependencies] +pydantic-graph = [ + { name = "pydantic-graph" }, +] + [package.dev-dependencies] dev = [ { name = "pytest" }, ] [package.metadata] -requires-dist = [{ name = "pydantic", specifier = ">=2.0,<3" }] +requires-dist = [ + { name = "pydantic", specifier = ">=2.0,<3" }, + { name = "pydantic-graph", marker = "extra == 'pydantic-graph'", specifier = ">=2.0" }, +] +provides-extras = ["pydantic-graph"] [package.metadata.requires-dev] dev = [{ name = "pytest", specifier = ">=8.0" }] @@ -156,6 +196,21 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/fa/c3/7c8b240552251faf6b3a957db200fcfbbcec36763c050428b601e0c9b83b/pydantic_core-2.46.4-graalpy312-graalpy250_312_native-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:00c603d540afdd6b80eb39f078f33ebd46211f02f33e34a32d9f053bba711de0", size = 2147590, upload-time = "2026-05-06T13:39:29.883Z" }, ] +[[package]] +name = "pydantic-graph" +version = "2.32.1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "anyio" }, + { name = "logfire-api" }, + { name = "pydantic" }, + { name = "typing-inspection" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/0f/75/9c0d96a0d99a5e71a1ef1f93d974efcc0e34d0c806f4788adc0abb19e213/pydantic_graph-2.32.1.tar.gz", hash = "sha256:881ef6e5115cceedfbc03befa9ec127c32c5d41d6b8b8aa4a950459fb913d848", size = 45162, upload-time = "2026-08-20T02:26:57.404Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/48/ef/d5ebf857fab20746462976a646a866dae25a05fee46ea1f98283487a339d/pydantic_graph-2.32.1-py3-none-any.whl", hash = "sha256:29a0cd70914db95b7b1f8a2ab19888f17757784e8b41ac100b894de2eed27993", size = 52647, upload-time = "2026-08-20T02:26:49.484Z" }, +] + [[package]] name = "pygments" version = "2.20.0"