Run the release gate
yourself.

#!/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())

#!/usr/bin/env python3
"""
agent_graph_demo.py: a working orchestrator-worker agent graph in LangGraph,
and the one edge that quietly breaks it.

Companion to the walkthrough "The Graph Looks Right. The Merge Is Where It Breaks."

    pip install langgraph langchain-core
    python agent_graph_demo.py            # the SAFE graph, with a human gate
    python agent_graph_demo.py --trap     # the SAME graph wired the unsafe way

There is no API key and no network here. `call_model` is a deterministic stub
so the graph's SHAPE is what you are watching, not a model's mood. The last
function in this file shows the two lines you change to use a real model.

Task: three read-only workers each check one source about a release (the test
run, the error metrics, the changelog). A single writer reads all three and
makes ONE call: ship, hold, or roll back.
"""
import argparse
import operator
from typing import Annotated, List, Literal, TypedDict

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command

CALLS = {"n": 0}   # a crude cost meter: how many model calls the graph made


def call_model(role: str, source: str) -> str:
    """A stand-in for a real LLM. Deterministic so the graph is what varies."""
    CALLS["n"] += 1
    verdict = {
        "tests": "hold",        # a flaky test failed
        "metrics": "ship it",   # error rate is flat
        "changelog": "roll back",  # an undocumented breaking change
    }[source]
    return verdict


# ---------------------------------------------------------------------------
# THE SAFE GRAPH
# GATHER state has many writers (workers append structured findings). DECISION
# state has one owner (the writer applies an explicit policy). Different jobs,
# different channels.
# ---------------------------------------------------------------------------
Verdict = Literal["ship it", "hold", "roll back"]


class Finding(TypedDict):
    reviewer: str
    verdict: Verdict


class State(TypedDict):
    findings: Annotated[List[Finding], operator.add]   # gather: many writers append
    decision: Verdict                                  # decision: one owner writes


def orchestrator(state: State) -> dict:
    # In a real build this would PLAN the workers from the question. Here the
    # plan is fixed so you can see the topology. It writes nothing a worker writes.
    return {}


def make_worker(source: str):
    def worker(state: State) -> dict:
        verdict = call_model(role="reviewer", source=source)
        # Each worker emits its OWN structured finding. It never writes `decision`.
        return {"findings": [{"reviewer": source, "verdict": verdict}]}
    return worker


def decide(verdicts: List[Verdict]) -> Verdict:
    # The decision POLICY, named and owned. This one is any-veto by severity;
    # swap it for majority / unanimous / weighted / threshold as your gate needs.
    if "roll back" in verdicts:
        return "roll back"
    if "hold" in verdicts:
        return "hold"
    return "ship it"


EXPECTED = {"tests", "metrics", "changelog"}


def writer(state: State) -> dict:
    # The single owner reads ALL findings and applies the policy. It is the only
    # node that writes `decision`. Ownership (one writer) is only part of the gate:
    # completeness (every reviewer reported) and uniqueness (each reported exactly
    # once) matter too, or a missing verdict silently ships and a duplicate one can
    # tip a majority rule. Fail closed on either.
    reviewers = [f["reviewer"] for f in state["findings"]]
    if sorted(reviewers) != sorted(EXPECTED):   # missing or duplicate -> not a ship
        return {"decision": "hold"}
    return {"decision": decide([f["verdict"] for f in state["findings"]])}


def human_gate(state: State):
    # interrupt() PAUSES the graph and hands control back to you. Resuming
    # re-runs this node FROM THE TOP, so keep side effects below the interrupt.
    approved = interrupt({"proposed": state["decision"],
                          "ask": "approve this decision? (resume with True/False)"})
    if not approved:
        return {"decision": state["decision"] + "  [VETOED by human]"}
    return {"decision": state["decision"] + "  [approved]"}


def build_safe_graph():
    g = StateGraph(State)
    g.add_node("orchestrator", orchestrator)
    g.add_node("writer", writer)
    g.add_node("human_gate", human_gate)
    for src in ("tests", "metrics", "changelog"):
        g.add_node(src, make_worker(src))
        g.add_edge("orchestrator", src)   # fan out (reads run in parallel)
        g.add_edge(src, "writer")         # fan in to the one writer
    g.add_edge(START, "orchestrator")
    g.add_edge("writer", "human_gate")
    g.add_edge("human_gate", END)
    # A checkpointer is REQUIRED for interrupt(): no saved state, nothing to resume.
    return g.compile(checkpointer=MemorySaver())


def run_safe():
    app = build_safe_graph()
    cfg = {"configurable": {"thread_id": "release-1"}}
    result = app.invoke({"question": "Do we ship?", "findings": [], "decision": ""}, cfg)
    # The run paused at the human gate. `result` carries the interrupt payload.
    payload = result["__interrupt__"][0].value
    print("PAUSED at human gate. Proposed:", payload["proposed"])
    final = app.invoke(Command(resume=False), cfg)   # human vetoes
    print("FINAL decision:", final["decision"])
    print("model calls made:", CALLS["n"])
    print("\nSAFE: three workers, three findings, ONE writer. 3 of 3 findings"
          " reached the decision, and a human held the veto.")


# ---------------------------------------------------------------------------
# THE TRAP
# The SAME three workers, but each writes `decision` directly. `decision` is a
# DECISION channel (one answer should win), not a gather channel. Watch the
# framework refuse, then watch the reducer you add to silence it hide the conflict.
# ---------------------------------------------------------------------------
class TrapState(TypedDict):
    decision: str   # a plain DECISION channel; three parallel writers collide here


class TrapStateReducer(TypedDict):
    # The reducer a hurried dev adds to make the error stop: keep the latest write.
    # It silences the error and silently drops the losers.
    decision: Annotated[str, lambda old, new: new]


def run_trap():
    # Step 1: the naive version. The framework STOPS you, loudly.
    g = StateGraph(TrapState)
    g.add_node("orchestrator", lambda s: {})
    for src in ("tests", "metrics", "changelog"):
        g.add_node(src, (lambda src=src: (lambda s: {"decision": call_model("r", src)}))())
        g.add_edge("orchestrator", src); g.add_edge(src, END)
    g.add_edge(START, "orchestrator")
    try:
        g.compile().invoke({"decision": ""})
    except Exception as e:
        print("STEP 1 (plain channel): the framework refused ->",
              type(e).__name__ + ":", str(e).splitlines()[0])

    # Step 2: you take the error's advice and add a reducer. Now it "works".
    g2 = StateGraph(TrapStateReducer)
    g2.add_node("orchestrator", lambda s: {})
    for src in ("tests", "metrics", "changelog"):
        g2.add_node(src, (lambda src=src: (lambda s: {"decision": call_model("r", src)}))())
        g2.add_edge("orchestrator", src); g2.add_edge(src, END)
    g2.add_edge(START, "orchestrator")
    kept = g2.compile().invoke({"decision": ""})["decision"]
    print("STEP 2 (reducer added): no error. the decision channel now holds:", repr(kept))
    print("        the other two reviewers' verdicts were silently dropped")
    print("\nTRAP: the error could not tell a decision channel from a gather channel, and"
          " neither could the reducer you added. 1 of 3 verdicts survives; two vanish, no error.")
    print("      (an operator.add reducer here would instead change the channel from a string"
          " to a list, and a consumer expecting a string mishandles it just as quietly.)")


def use_a_real_model_instead():
    """Not called. The only change from the stub to a real model:

        from langchain_anthropic import ChatAnthropic     # pip install langchain-anthropic
        _llm = ChatAnthropic(model="claude-sonnet-4-5")    # reads ANTHROPIC_API_KEY

    then inside a worker, replace the stub line with:

        verdict = _llm.invoke(f"Review the {source} and answer ship it/hold/roll back").content

    The graph, the channels, and the read/write rule do not change. The model
    is a node; the safety is in the wiring.
    """


if __name__ == "__main__":
    ap = argparse.ArgumentParser()
    ap.add_argument("--trap", action="store_true", help="wire the same graph the unsafe way")
    args = ap.parse_args()
    run_trap() if args.trap else run_safe()

#!/usr/bin/env python3
"""
audit.py: point this at your own LangGraph State schema and it names the exact
channels where two agents can silently overwrite each other.

    from audit import audit_schema
    audit_schema(MyState)          # MyState is your graph's State TypedDict

Companion to "The Graph Looks Right. The Merge Is Where It Breaks."

The discriminator, made mechanical, for the fan-out case. You do NOT have to eyeball
every channel and guess which is a decision. LangGraph refuses CONCURRENT writes to a
PLAIN channel, loudly (InvalidUpdateError), so a fan-out cannot silently collide on one.
The only place a fan-out's writes merge SILENTLY is a channel that carries a REDUCER,
because a reducer is you telling the framework "merging is fine here." So for parallel
writes, the reducer channels are your whole silent-merge surface, and this reads them
straight off the type.

Scope, stated honestly. That guarantee is concurrency-only. It does NOT cover a plain
channel written by two nodes in DIFFERENT supersteps (orchestrator -> A -> B, both
writing `decision`): B overwrites A with last-value-wins, no error, no reducer. That
second surface, a plain channel with more than one writer node in sequence, cannot be
read off the schema, because which channels a node writes is decided by the dict it
returns at runtime, not by the type. The rule for it is manual and simple: a channel that
holds a decision should be written by exactly one node. This tool finds the fan-out
surface; you enforce the single-writer rule for the sequential one.

For each reducer channel it asks the one question the schema cannot answer for you,
and gives you the test that answers it:

    GATHER  channel: its value is only ever read in aggregate (summed, listed,
                      reconciled by a downstream node). A reducer is CORRECT here.
    DECISION channel: its value steers control flow (a conditional edge routes on
                      it, or a consumer acts on it as one value). A reducer here
                      SILENTLY drops writes. Give it ONE writer and remove the reducer.

The near-mechanical test for the classification: does any downstream reader BRANCH
on this channel or act on it as a single value? If yes, it is a decision channel.
"""
import sys
from typing import Annotated, get_args, get_origin, get_type_hints


def audit_schema(schema, name=None):
    name = name or getattr(schema, "__name__", "State")
    try:
        hints = get_type_hints(schema, include_extras=True)
    except Exception as e:  # pragma: no cover
        print(f"could not read {name}: {e}")
        return []
    reducer_channels = []
    print(f"\naudit of {name}")
    for field, hint in hints.items():
        if get_origin(hint) is Annotated:
            meta = get_args(hint)[1:]
            red = ", ".join(getattr(m, "__name__", repr(m)) for m in meta)
            reducer_channels.append(field)
            print(f"  {'! classify':<13} {field:14} REDUCER ({red})")
        else:
            print(f"  {'fan-out safe':<13} {field:14} plain")
    if reducer_channels:
        print(f"\n  reducer channels are the whole fan-out silent-merge surface: "
              f"{', '.join(reducer_channels)}")
        print("  for each, ask: does a reader branch on it or act on its value as one answer?")
        print("  if yes it is a DECISION channel -> give it one writer and drop the reducer.")
    else:
        print("\n  no reducer channels: a fan-out cannot merge silently here, the framework")
        print("  refuses concurrent writes to a plain channel.")
    print("  not readable from the schema: a plain channel written by more than one node in")
    print("  sequence overwrites silently. keep every decision channel to one writer.")
    return reducer_channels


# ---------------------------------------------------------------------------
# Demo: audit the two schemas from agent_graph_demo.py so you can see the shape.
# ---------------------------------------------------------------------------
def _demo():
    import operator
    from typing import List, TypedDict

    class State(TypedDict):                       # the SAFE graph's schema
        findings: Annotated[List[str], operator.add]   # gather: many writers append
        decision: str                                  # decision: plain, one owner

    class TrapStateReducer(TypedDict):            # the TRAP's schema
        decision: Annotated[str, lambda old, new: new]  # decision WITH a reducer

    print("=== the safe graph ===")
    audit_schema(State, "State")
    print("\n   verdict: `findings` is a gather (a node reconciles the list); `decision`")
    print("   is plain, so the framework guards it. Clean.")

    print("\n=== the trap ===")
    audit_schema(TrapStateReducer, "TrapStateReducer")
    print("\n   verdict: `decision` carries a reducer but its value IS the decision the")
    print("   graph acts on. The reducer silently keeps one write of several. This is the bug.")


if __name__ == "__main__":
    _demo()
    sys.exit(0)