A parallel plan crashed the scaffold: shared fields become reducers (#91)

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

Author: Shashankss1205

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 ────────────────────────────────────────────────────

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

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(

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"]