#!/usr/bin/env python3 """ check_owners.py: the runtime half of The State Channel Audit. It runs your compiled LangGraph graph once, records which node wrote each decision channel, and fails when more than one node did. python3 check_owners.py # demo: one safe graph passes, two broken ones fail In a test: from check_owners import assert_single_writer assert_single_writer(app, {"findings": []}, ["decision"]) # a graph that pauses at interrupt(): pass every step of the run, same config from langgraph.types import Command assert_single_writer(app, [{"findings": []}, Command(resume=True)], ["decision"], config={"configurable": {"thread_id": "test-1"}}) Companion to "The Graph Looks Right. The Merge Is Where It Breaks." https://durabilitycurve.com/tools/agent-graph/ Why a run and a schema both. audit.py reads the State type and finds every channel where parallel writes can merge. It cannot see which node writes what, because that is the dict a node returns at runtime. So a plain decision channel written by two nodes in sequence (A writes "roll back", B then writes "ship it") passes the schema audit and ships B's answer with no error. This check watches the run itself: every write to a decision channel must come from one node. The owner may write many times (a loop); no other node may write it at all. How it watches. graph.stream(..., stream_mode="debug") reports every task that finished, with its node name, its step number and the writes it made. So the check sees a second node writing in a later step, two nodes writing in parallel, and one node run several times in parallel (a Send fan-out: three copies of `judge` in one step), and it fails on any of them: more than one node writes a decision channel during the run, or a decision channel receives more than one write in the same step. The owner may write many times in different steps (a loop). Checked on langgraph 0.6.11 and 1.2.14 (7 Oct 2026). If the graph refuses a parallel write (InvalidUpdateError), that error is raised to your test unchanged: the framework caught it first. """ import sys import warnings # Two harmless import notices, silenced: urllib3's NotOpenSSLWarning on macOS system # Python, and a langchain-core notice. langchain-core resets its own warning filters # when it is imported, so it is imported first and quietened after. warnings.filterwarnings("ignore", module="urllib3") try: import langchain_core # noqa: F401,E402 except ImportError: pass warnings.filterwarnings("ignore", message=".*allowed_objects.*") def write_log(app, inputs, keys, config=None): """Run the graph and return [(run, step, node, key), ...] for every write to a channel in `keys`, in the order the tasks finished. `inputs` is one graph input, or a list of inputs run in turn on the same config (for example the first input, then Command(resume=...)).""" runs = inputs if isinstance(inputs, list) else [inputs] log = [] for i, run in enumerate(runs): for ev in app.stream(run, config, stream_mode="debug"): if not isinstance(ev, dict) or ev.get("type") != "task_result": continue payload = ev.get("payload") or {} node = str(payload.get("name", "")) if node.startswith("__"): # __interrupt__ and other markers continue result = payload.get("result") or {} pairs = result.items() if isinstance(result, dict) else result for pair in pairs: key = pair[0] if isinstance(pair, (list, tuple)) else pair if key in keys: log.append((i, ev.get("step"), node, key)) return log def writers(app, inputs, keys, config=None): """Run the graph and return {key: [node, node, ...]} in the order the writes happened (one entry per write).""" seen = {k: [] for k in keys} for _, _, node, key in write_log(app, inputs, keys, config): seen[key].append(node) return seen def assert_single_writer(app, inputs, decision_keys, config=None, require_write=True): """Raise AssertionError unless each decision channel was written by exactly one node, at most once per step, during the run. With require_write (the default) a decision channel the run never wrote also fails, because then ownership was never tested. Returns {key: [node, ...]}, one entry per write.""" log = write_log(app, inputs, decision_keys, config) seen = {k: [n for _, _, n, key in log if key == k] for k in decision_keys} problems = [] for key, nodes in seen.items(): owners = list(dict.fromkeys(nodes)) if len(owners) > 1: problems.append(f"'{key}' was written by {len(owners)} nodes " f"({', '.join(owners)}, in that order); one node should own it") continue per_step = {} for run, step, node, k in log: if k == key: per_step.setdefault((run, step), []).append(node) crowded = [(rs, ns) for rs, ns in per_step.items() if len(ns) > 1] if crowded: (run, step), ns = crowded[0] problems.append(f"'{key}' received {len(ns)} writes in one step (step {step}, from " f"{', '.join(ns)}): parallel copies of one node, such as a Send " f"fan-out, merged by a reducer; one write per step should decide it") elif not nodes and require_write: problems.append(f"'{key}' was never written in this run, so its owner was not tested") if problems: raise AssertionError("single-writer check failed: " + "; ".join(problems)) return seen # --------------------------------------------------------------------------- # Demo: the safe release gate passes; the sequential overwrite and the trap fail. # --------------------------------------------------------------------------- def _demo(): import operator from typing import Annotated, List, TypedDict from langgraph.graph import StateGraph, START, END def report(label, app, inputs): try: seen = assert_single_writer(app, inputs, ["decision"]) print(f"PASS {label}: decision written by {seen['decision']}") return True except AssertionError as e: print(f"FAIL {label}: {e}") return False # 1. safe: three reviewers append to a gather; one writer owns the decision class Safe(TypedDict): findings: Annotated[List[str], operator.add] decision: str g = StateGraph(Safe) g.add_node("orchestrator", lambda s: {}) for w in ("tests", "metrics", "changelog"): g.add_node(w, (lambda w=w: (lambda s: {"findings": [w]}))()) g.add_edge("orchestrator", w) g.add_edge(w, "writer") g.add_node("writer", lambda s: {"decision": "hold"}) g.add_edge(START, "orchestrator") g.add_edge("writer", END) ok1 = report("safe release gate", g.compile(), {"findings": []}) # 2. sequential overwrite: a plain channel, two writers in different steps. # audit.py passes this schema; the run ships the second answer. class Seq(TypedDict): decision: str g = StateGraph(Seq) g.add_node("reviewer_a", lambda s: {"decision": "roll back"}) g.add_node("reviewer_b", lambda s: {"decision": "ship it"}) g.add_edge(START, "reviewer_a") g.add_edge("reviewer_a", "reviewer_b") g.add_edge("reviewer_b", END) seq = g.compile() final = seq.invoke({"decision": ""})["decision"] ok2 = not report(f"sequential overwrite (final decision {final!r}, no error)", seq, {"decision": ""}) # 3. the trap: a last-write-wins reducer on the decision channel class Trap(TypedDict): decision: Annotated[str, lambda old, new: new] g = StateGraph(Trap) g.add_node("orchestrator", lambda s: {}) for w, v in (("tests", "hold"), ("metrics", "ship it"), ("changelog", "roll back")): g.add_node(w, (lambda v=v: (lambda s: {"decision": v}))()) g.add_edge("orchestrator", w) g.add_edge(w, END) g.add_edge(START, "orchestrator") ok3 = not report("last-write-wins reducer", g.compile(), {"decision": ""}) print("\nAs expected: the safe gate passes, both broken gates fail." if (ok1 and ok2 and ok3) else "\nUNEXPECTED: a demo graph did not behave as described above.") return 0 if (ok1 and ok2 and ok3) else 1 if __name__ == "__main__": sys.exit(_demo())