From e1c8a40aba2e545a6799eba948c5ae81b9f9c416 Mon Sep 17 00:00:00 2001 From: Shashank Shekhar Singh Date: Thu, 6 Aug 2026 00:59:34 +0530 Subject: [PATCH] A parallel plan crashed the scaffold: shared fields become reducers Co-Authored-By: Claude Fable 5 --- grapharc/cli/init_cmd.py | 21 +++++++++++------ grapharc/examples/plan_incident.py | 16 +++++++++---- tests/test_cli.py | 37 ++++++++++++++++++++++++++++++ tests/test_planner_loop.py | 29 +++++++++++++++++++++++ 4 files changed, 92 insertions(+), 11 deletions(-) diff --git a/grapharc/cli/init_cmd.py b/grapharc/cli/init_cmd.py index 74c7dd8..47dd068 100644 --- a/grapharc/cli/init_cmd.py +++ b/grapharc/cli/init_cmd.py @@ -58,7 +58,8 @@ from __future__ import annotations -from typing import Any +import operator +from typing import Annotated, Any from pydantic import BaseModel @@ -87,7 +88,13 @@ class State(BaseModel): goal: str = "" # filled from the CLI argument; the planner reads it - notes: list[str] = [] # the working record every kind appends to + # `notes` is a REDUCER (Annotated + operator.add): each writer returns just + # its own lines and LangGraph merges them, so two nodes — or two + # planner-named instances of ONE kind — may write it in the same parallel + # step. A plain `list[str]` here crashes the first time a planner runs two + # writers concurrently (InvalidUpdateError); keep the pattern for any field + # more than one node may write. + notes: Annotated[list[str], operator.add] = [] report: str = "" # the deliverable; the goal check below watches it @@ -111,7 +118,7 @@ def _gather(state: State) -> dict: f"({', '.join(dirs[:12]) or 'none'}) and {len(files)} file(s) " f"({', '.join(files[:12]) or 'none'})" ) - return {"notes": [*state.notes, note]} + return {"notes": [note]} def _analyse(state: State) -> dict: @@ -126,7 +133,7 @@ def _analyse(state: State) -> dict: note = "analyse: file types by count — " + ", ".join( f"{ext} x{count}" for ext, count in top ) - return {"notes": [*state.notes, note]} + return {"notes": [note]} def _report_for(model: Any): @@ -142,7 +149,7 @@ def body(state: State) -> dict: if model is None or scripted or not hasattr(model, "invoke"): return { "report": "report: run with --model SPEC for a model-written report", - "notes": [*state.notes, "report: written without a model"], + "notes": ["report: written without a model"], } try: reply = model.invoke( @@ -153,7 +160,7 @@ def body(state: State) -> dict: text = str(getattr(reply, "content", reply)).strip()[:2000] except Exception as exc: # a failed call is a note, not a crash text = f"report: model call failed ({exc}); notes stand" - return {"report": text, "notes": [*state.notes, "report: written"]} + return {"report": text, "notes": ["report: written"]} return body @@ -164,7 +171,7 @@ def _apply(state: State) -> dict: propose it) and DENIED by the edge policy below (no admitted graph may reach it) until you decide otherwise. Keep the pattern even after you rename it: a gate with nothing to refuse proves nothing.""" - return {"notes": [*state.notes, "apply: this should not have run"]} + return {"notes": ["apply: this should not have run"]} # ── 3. Write permissions ──────────────────────────────────────────────────── diff --git a/grapharc/examples/plan_incident.py b/grapharc/examples/plan_incident.py index bf3f4a5..74289a4 100644 --- a/grapharc/examples/plan_incident.py +++ b/grapharc/examples/plan_incident.py @@ -20,7 +20,8 @@ from __future__ import annotations import json -from typing import Any +import operator +from typing import Annotated, Any from pydantic import BaseModel @@ -44,10 +45,17 @@ class IncidentState(BaseModel): - """One state contract for the whole run, however the topology changes.""" + """One state contract for the whole run, however the topology changes. + + `notes` is a reducer (`Annotated` + `operator.add`): each writer returns + only its own lines and LangGraph merges them, so a planner that runs two + writers in the same parallel step — including two instances of one kind — + composes instead of colliding. A plain `list[str]` here raises + `InvalidUpdateError` the first time that happens. + """ goal: str = "" - notes: list[str] = [] + notes: Annotated[list[str], operator.add] = [] def _step_factory(spec: NodeSpec) -> Any: @@ -59,7 +67,7 @@ def _step_factory(spec: NodeSpec) -> Any: """ def body(state: IncidentState) -> dict: - return {"notes": [*state.notes, f"{spec.name} ran"]} + return {"notes": [f"{spec.name} ran"]} body.writes = {"notes"} return body diff --git a/tests/test_cli.py b/tests/test_cli.py index f3c8ac1..bceaae2 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -2198,6 +2198,43 @@ def test_an_init_scaffold_plans_end_to_end(tmp_path, monkeypatch, capsys): assert "goal_met" in printed +def test_the_scaffold_state_merges_parallel_writers(tmp_path, monkeypatch): + """Two kinds writing `notes` in the same superstep compose via the reducer. + + The shape any real planner eventually proposes: `gather` and `analyse` + both fanned out of START, joining at `report`. With a plain `list[str]` + this run died on LangGraph's InvalidUpdateError before `report` ever ran; + the scaffold's `notes` is a reducer now, and this test is what keeps it + one. + """ + monkeypatch.chdir(tmp_path) + from grapharc.cli.init_cmd import REGISTRY_TEMPLATE + from grapharc.testing import ScriptedChatModel + + module = ModuleType("scaffold_registry") + # The path-form loader registers the module before executing it, and + # pydantic needs that to resolve the template's deferred annotations. + monkeypatch.setitem(sys.modules, "scaffold_registry", module) + exec(compile(REGISTRY_TEMPLATE, "registry.py", "exec"), module.__dict__) + plan = json.dumps( + { + "nodes": [{"name": "gather"}, {"name": "analyse"}, {"name": "report"}], + "edges": [ + {"source": "__start__", "target": "gather"}, + {"source": "__start__", "target": "analyse"}, + {"source": "gather", "target": "report"}, + {"source": "analyse", "target": "report"}, + {"source": "report", "target": "__end__"}, + ], + } + ) + loop = module.build_loop(ScriptedChatModel(responses=[plan])) + result = loop.run("report on this directory, twice over", module.State()) + assert result.stop.value == "goal_met" + assert any(note.startswith("gather:") for note in result.state.notes) + assert any(note.startswith("analyse:") for note in result.state.notes) + + def test_the_path_form_registry_shares_one_module_object(tmp_path, monkeypatch): monkeypatch.chdir(tmp_path) (tmp_path / "reg.py").write_text( diff --git a/tests/test_planner_loop.py b/tests/test_planner_loop.py index 3ff2919..c06783f 100644 --- a/tests/test_planner_loop.py +++ b/tests/test_planner_loop.py @@ -1632,3 +1632,32 @@ def test_the_disclosure_is_not_what_refuses_the_edge(): assert [r.model_dump() for r in with_disclosure.rejections()] == [ r.model_dump() for r in without.rejections() ] + + +def test_the_incident_example_state_merges_parallel_writers(): + """Three kinds writing `notes` in one superstep compose via the reducer. + + The shipped example's state used a plain `list[str]`, so the first plan + that fanned kinds out of START died on LangGraph's InvalidUpdateError. + `IncidentState.notes` is a reducer now; this run is the shape that broke. + """ + from grapharc.examples.plan_incident import IncidentState + from grapharc.examples.plan_incident import build_loop as build_incident_loop + + fan_out = json.dumps( + { + "nodes": [{"name": "triage"}, {"name": "patch"}, {"name": "verify"}], + "edges": [ + {"source": "__start__", "target": "triage"}, + {"source": "__start__", "target": "patch"}, + {"source": "__start__", "target": "verify"}, + {"source": "triage", "target": "__end__"}, + {"source": "patch", "target": "__end__"}, + {"source": "verify", "target": "__end__"}, + ], + } + ) + loop = build_incident_loop(ScriptedChatModel(responses=[fan_out])) + result = loop.run("triage, patch and verify at once", IncidentState()) + assert result.stop.value == "goal_met" + assert sorted(result.state.notes) == ["patch ran", "triage ran", "verify ran"]