Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 14 additions & 7 deletions grapharc/cli/init_cmd.py
Original file line number Diff line number Diff line change
Expand Up @@ -58,7 +58,8 @@

from __future__ import annotations

from typing import Any
import operator
from typing import Annotated, Any

from pydantic import BaseModel

Expand Down Expand Up @@ -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


Expand All @@ -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:
Expand All @@ -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):
Expand All @@ -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(
Expand All @@ -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

Expand All @@ -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 ────────────────────────────────────────────────────
Expand Down
16 changes: 12 additions & 4 deletions grapharc/examples/plan_incident.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand All @@ -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:
Expand All @@ -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
Expand Down
37 changes: 37 additions & 0 deletions tests/test_cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand Down
29 changes: 29 additions & 0 deletions tests/test_planner_loop.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"]
Loading