#!/usr/bin/env python3 """ verify.py: the runnable object behind the walkthrough "The Graph Looks Right. The Merge Is Where It Breaks." (slug: where-agents-disagree) It reproduces, FROM A RUN, the number the piece is built on: 1 of 3. The claim, precisely. When several agents run in parallel and write the same LangGraph state channel, the framework STOPS you with a loud error and tells you to add a reducer. That advice is right for a GATHER channel (collect a partial from each worker) and wrong for a DECISION channel (one answer should win), and the error cannot tell the two apart. Add a reducer to a decision channel and the collision does not go away, it goes SILENT: a last-write-wins reducer keeps one of the three writes and drops the other two with no error. The same reducer on a gather channel is correct, and a single writer then reconciles 3 of 3. Runs two ways, same headline either way: * with `langgraph` installed, it runs the REAL graphs (verified on langgraph 0.6.11 / langchain-core 0.3.86 / CPython 3.9.6, 2026-08-29). * otherwise it runs a dependency-free model of the identical channel semantics. The verbatim framework error strings are in ERR_* below. No network, no API key, no LLM call. Exit 0 on success. """ import operator import sys from typing import Annotated, List, TypedDict WORKERS = ("w1", "w2", "w3") # Anthropic's research system fans out 3-5 subagents DECISIONS = ("ship it", "hold", "roll back") # three plausible, mutually exclusive calls # Verbatim error strings recorded from the real run (langgraph 0.6.11, 2026-08-29): ERR_INVALID = ("At key 'decision': Can receive only one value per step. " "Use an Annotated key to handle multiple values.") ERR_RECURSION = ("Recursion limit of 25 reached without hitting a stop condition.") # State schemas for the real langgraph path (module level so get_type_hints resolves them). class SBare(TypedDict): decision: str # a plain DECISION channel class SDecideReducer(TypedDict): decision: Annotated[str, lambda old, new: new] # last-write-wins: silently keeps one class SGather(TypedDict): findings: Annotated[List[str], operator.add] # gather channel: the SAME knob, used right answer: str class SLoop(TypedDict): n: Annotated[int, operator.add] def rule(t: str) -> None: print("\n" + "-" * 68 + "\n" + t + "\n" + "-" * 68) # --------------------------------------------------------------------------- # Dependency-free model of LangGraph channel reduction, one superstep at a time. # --------------------------------------------------------------------------- class InvalidUpdate(Exception): pass def reduce_lastvalue(writes): if len(writes) > 1: # a plain channel refuses > 1 write per step raise InvalidUpdate(ERR_INVALID) return writes[0] if writes else None def reduce_lastwins(writes): # a "just make the error stop" reducer return writes[-1] if writes else None # keeps the last write, drops the rest, silently def reduce_add(existing, writes): # operator.add on a list: concatenate every write out = list(existing) for w in writes: out += w return out def model_run(): ok = True rule("1. parallel write to a plain DECISION channel -> the framework STOPS you") try: reduce_lastvalue([d for d in DECISIONS]) print(" unexpected: no error"); ok = False except InvalidUpdate as e: print(" raised InvalidUpdateError:", e) rule("2. silence it with a reducer on the DECISION channel -> the merge is SILENT") kept = reduce_lastwins(list(DECISIONS)) dropped = [d for d in DECISIONS if d != kept] print(" reducer = keep the latest write (a common way to make the error stop)") print(" the channel now holds:", repr(kept)) print(" silently dropped:", dropped, "(no error was raised)") survived = 1 print(" note: an operator.add reducer here would instead turn the scalar into a list,") print(" and a consumer that expects a scalar mishandles it just as silently.") rule("3. the SAME reducer knob, on a GATHER channel -> correct: reconcile 3 of 3") findings = reduce_add([], [[f"{w}: read result"] for w in WORKERS]) reconciled = len(findings) print(" reducer = operator.add on an isolated findings channel (a gather, not a decision)") print(" findings reaching the single writer =", reconciled, "of 3") print(" the one writer reconciles all of them into one decision") if reconciled != 3: print(" unexpected: writer did not see all findings"); ok = False rule("4. a cycle with no stop condition -> capped at 25, not infinite") steps = 0 for steps in range(1, 1000): if steps >= 25: print(" raised GraphRecursionError:", ERR_RECURSION) break if steps != 25: print(" unexpected cap"); ok = False return ok, survived, reconciled # --------------------------------------------------------------------------- # Real LangGraph path (used automatically when the library is installed). # --------------------------------------------------------------------------- def real_run(): from langgraph.graph import StateGraph, START, END from langgraph.errors import GraphRecursionError rule("1. parallel write to a plain DECISION channel -> the framework STOPS you") g = StateGraph(SBare) g.add_node("orch", lambda s: {}) for i, d in enumerate(DECISIONS): g.add_node(f"n{i}", (lambda d=d: (lambda s: {"decision": d}))()) g.add_edge("orch", f"n{i}"); g.add_edge(f"n{i}", END) g.add_edge(START, "orch") try: g.compile().invoke({"decision": ""}) print(" unexpected: no error"); return False, -1, -1 except Exception as e: print(" raised", type(e).__name__ + ":", str(e).splitlines()[0]) rule("2. silence it with a reducer on the DECISION channel -> the merge is SILENT") g2 = StateGraph(SDecideReducer) g2.add_node("orch", lambda s: {}) for i, d in enumerate(DECISIONS): g2.add_node(f"n{i}", (lambda d=d: (lambda s: {"decision": d}))()) g2.add_edge("orch", f"n{i}"); g2.add_edge(f"n{i}", END) g2.add_edge(START, "orch") kept = g2.compile().invoke({"decision": ""})["decision"] dropped = [d for d in DECISIONS if d != kept] print(" reducer = keep the latest write (a common way to make the error stop)") print(" the channel now holds:", repr(kept)) print(" silently dropped:", dropped, "(no error was raised)") survived = 1 print(" note: an operator.add reducer here would instead turn the scalar into a list,") print(" and a consumer that expects a scalar mishandles it just as silently.") rule("3. the SAME reducer knob, on a GATHER channel -> correct: reconcile 3 of 3") g3 = StateGraph(SGather) g3.add_node("orch", lambda s: {}) g3.add_node("writer", lambda s: {"answer": f"reconciled {len(s['findings'])} of 3"}) for w in WORKERS: g3.add_node(w, (lambda w=w: (lambda s: {"findings": [f"{w}: read result"]}))()) g3.add_edge("orch", w); g3.add_edge(w, "writer") g3.add_edge(START, "orch"); g3.add_edge("writer", END) out3 = g3.compile().invoke({"findings": [], "answer": ""}) reconciled = len(out3["findings"]) print(" reducer = operator.add on an isolated findings channel (a gather, not a decision)") print(" findings reaching the single writer =", reconciled, "of 3") print(" ", out3["answer"]) rule("4. a cycle with no stop condition -> capped at 25, not infinite") g4 = StateGraph(SLoop) g4.add_node("loop", lambda s: {"n": 1}) g4.add_edge(START, "loop"); g4.add_edge("loop", "loop") try: g4.compile().invoke({"n": 0}) print(" unexpected: terminated"); return False, survived, reconciled except GraphRecursionError as e: print(" raised GraphRecursionError:", str(e).splitlines()[0]) return True, survived, reconciled def main() -> int: try: import langgraph # noqa: F401 mode = "REAL langgraph" ok, survived, reconciled = real_run() except ImportError: mode = "dependency-free model (langgraph not installed)" ok, survived, reconciled = model_run() rule("HEADLINE") print(f" mode: {mode}") print(f" gather channel: one writer reconciled {reconciled} of 3 worker results.") print(f" decision channel: the reducer kept {survived} of 3 writes and dropped the rest, " f"with no error.") print() print(f" The number the piece is built on: {survived} of 3") return 0 if (ok and survived == 1 and reconciled == 3) else 1 if __name__ == "__main__": sys.exit(main())