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)
https://durabilitycurve.com/tools/agent-graph/

It measures, from a real LangGraph 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 goes SILENT: a last-write-wins reducer keeps one of the three
writes and drops the other two with no error. The same reducer knob on a gather
channel is correct, and a single writer then reconciles 3 of 3.

Every number in the headline is counted from the run: how many of the three
verdicts the channel still holds, how many findings reached the writer, and the
recursion limit the error reports. If LangGraph's behaviour changes, the
headline changes with it and the script exits 1.

Checked on langgraph 0.6.11 (Python 3.9.6 and 3.12.12) and langgraph 1.2.14
(Python 3.12.12), 7 Oct 2026. The recursion limit is 25 on 0.6.11 and 10,007 on
1.2.14; everything else is identical.

Without langgraph installed it prints a REPLAY of the recorded 0.6.11 run and
exits 2, so it can never pass a gate without the real library.

No network, no API key, no LLM call. Exit 0 when the run reproduces the claim.
"""
import operator
import re
import sys
import warnings
from typing import Annotated, List, TypedDict

# 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.*")

WORKERS = ("w1", "w2", "w3")   # Anthropic's research lead agent "spins up 3-5 subagents in parallel"
                               # (anthropic.com/engineering/multi-agent-research-system, checked 7 Oct 2026)
DECISIONS = ("ship it", "hold", "roll back")   # three plausible, mutually exclusive calls

# Verbatim strings recorded from the real run (langgraph 0.6.11, 2026-08-29), used
# only by the replay when langgraph is not installed:
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)


def count_survivors(kept):
    """How many of the three verdicts the channel still holds, and which were lost."""
    survived = sum(1 for d in DECISIONS if d == kept)
    dropped = [d for d in DECISIONS if d != kept]
    return survived, dropped


# ---------------------------------------------------------------------------
# Real LangGraph path.
# ---------------------------------------------------------------------------
def real_run():
    from langgraph.graph import StateGraph, START, END
    from langgraph.errors import GraphRecursionError, InvalidUpdateError

    ok = True
    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")
        ok = False
    except InvalidUpdateError as e:          # only the expected refusal; anything else fails
        print("   raised", type(e).__name__ + ":", str(e).splitlines()[0])
        if "'decision'" not in str(e):
            print("   unexpected: the error names a different key")
            ok = False

    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"]
    survived, dropped = count_survivors(kept)
    print("   reducer = keep the latest write (a common way to make the error stop)")
    print("   the channel now holds:", repr(kept))
    if survived == 1:
        print("   silently dropped:", dropped, "(no error was raised)")
    else:
        print(f"   unexpected: the channel holds {survived} of the three verdicts")
        ok = False
    print("   note: with an operator.add reducer here the three strings run together")
    print("         into one: garbage you can see, where this drop is silent.")

    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 channel)")
    print("   findings reaching the single writer =", reconciled, "of 3")
    print("  ", out3["answer"])
    if reconciled != 3:
        ok = False

    rule("4. a cycle with no stop condition  ->  stopped by the recursion limit")
    g4 = StateGraph(SLoop)
    g4.add_node("loop", lambda s: {"n": 1})
    g4.add_edge(START, "loop")
    g4.add_edge("loop", "loop")
    limit = None
    try:
        g4.compile().invoke({"n": 0})
        print("   unexpected: terminated")
        ok = False
    except GraphRecursionError as e:
        msg = str(e).splitlines()[0]
        print("   raised GraphRecursionError:", msg)
        m = re.search(r"limit of (\d+)", msg)
        limit = int(m.group(1)) if m else None
        print(f"   the default limit on this install is {limit}" if limit else
              "   could not read the limit from the error")

    return ok, survived, reconciled, limit


# ---------------------------------------------------------------------------
# Replay of the recorded 0.6.11 run, for a machine without langgraph.
# It runs no framework code, so it proves nothing and always exits 2.
# ---------------------------------------------------------------------------
def replay():
    rule("REPLAY of the recorded langgraph 0.6.11 run (langgraph is not installed)")
    print("   1. plain channel, three parallel writes: InvalidUpdateError:", ERR_INVALID)
    print("   2. last-write-wins reducer: the channel held 'roll back';"
          " dropped ['ship it', 'hold'], no error")
    print("   3. operator.add gather channel: the writer reconciled 3 of 3")
    print("   4. a cycle with no stop condition: GraphRecursionError:", ERR_RECURSION)
    print("\n   This is a recording. To measure it yourself:")
    print('   pip install "langgraph==0.6.11" "langchain-core==0.3.86" && python3 verify.py')


def main() -> int:
    try:
        from importlib.metadata import version
        import langgraph  # noqa: F401
        lg = version("langgraph")
    except Exception:
        replay()
        return 2

    ok, survived, reconciled, limit = real_run()
    rule("HEADLINE")
    print(f"   mode: REAL langgraph {lg}")
    print(f"   gather channel: one writer reconciled {reconciled} of 3 worker results.")
    if survived == 1:
        print(f"   decision channel: the reducer kept {survived} of 3 writes and dropped the rest, "
              f"with no error.")
    else:
        print(f"   decision channel: the reducer left a value that matches {survived} of the 3 writes.")
    if limit:
        print(f"   recursion limit on this install: {limit}")
    print()
    print(f"   The number the piece is built on:  {survived} of 3")
    if not (ok and survived == 1 and reconciled == 3):
        print("   This run did NOT reproduce the claim. See the unexpected lines above.")
        return 1
    return 0


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."
https://durabilitycurve.com/tools/agent-graph/

    pip install "langgraph==0.6.11" "langchain-core==0.3.86"
    python agent_graph_demo.py            # the SAFE graph; the human gate vetoes
    python agent_graph_demo.py --approve  # the SAFE graph; the human gate approves
    python agent_graph_demo.py --trap     # the SAME graph wired the unsafe way
    python agent_graph_demo.py --trap --rename   # rename one node, watch the survivor change

There is no API key and no network here. `call_model` is a deterministic stub,
so what you are watching is the graph's SHAPE. The last function in this file
shows the two lines you change to use a real model. Every reply passes through
normalise_verdict(), so free text from a model ("Hold.", "Roll back - ...") maps
to one of the three verdicts, and anything it cannot map becomes "hold".

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
import warnings
from typing import Annotated, List, Literal, TypedDict

# 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.*")

from langgraph.graph import StateGraph, START, END          # noqa: E402
from langgraph.checkpoint.memory import MemorySaver          # noqa: E402
from langgraph.types import interrupt, Command               # noqa: E402

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

VERDICTS = {
    "tests": "hold",           # a flaky test failed
    "metrics": "ship it",      # error rate is flat
    "changelog": "roll back",  # an undocumented breaking change
}


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
    return VERDICTS[source]


# ---------------------------------------------------------------------------
# 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
    approved: bool                                     # the human's call, kept out of `decision`


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 {}


VALID = ("ship it", "hold", "roll back")


def normalise_verdict(reply) -> Verdict:
    # A model answers in free text. Read the reply's opening words; a reply that
    # does not start with one of the three verdicts is a hold (fail closed).
    text = " ".join(str(reply).strip().lower().split())
    for v in VALID:
        if text.startswith(v):
            return v
    return "hold"


def make_worker(source: str):
    def worker(state: State) -> dict:
        verdict = normalise_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, and it
    # fails closed: it ships only when every verdict is exactly "ship it", and any
    # verdict it does not recognise counts as a hold. Swap it for majority /
    # unanimous / weighted / threshold as your gate needs.
    if "roll back" in verdicts:
        return "roll back"
    if verdicts and all(v == "ship it" for v in verdicts):
        return "ship it"
    return "hold"


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: hold
        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)"})
    return {"approved": bool(approved)}   # `decision` stays a valid Verdict


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(approve=False):
    app = build_safe_graph()
    cfg = {"configurable": {"thread_id": "release-1"}}
    result = app.invoke({"findings": []}, 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=approve), cfg)   # veto unless --approve
    print("FINAL decision:", final["decision"],
          "| human:", "approved" if final["approved"] else "VETOED")
    print("model calls made:", CALLS["n"])
    human = "a human approved it" if final["approved"] else "a human vetoed it"
    print(f"\nSAFE: three workers, three findings, ONE writer. 3 of 3 findings"
          f" reached the decision, and {human}.")


# ---------------------------------------------------------------------------
# THE TRAP
# The SAME three workers, but each writes `decision` directly. `decision` is a
# DECISION channel (one answer should win); it is no place for a merge. 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 build_trap_graph(schema=TrapStateReducer, names=None):
    """Three workers that each write `decision`. `names` maps node name -> source."""
    names = names or {src: src for src in VERDICTS}
    g = StateGraph(schema)
    g.add_node("orchestrator", lambda s: {})
    for node, src in names.items():
        g.add_node(node, (lambda src=src: (lambda s: {"decision": call_model("r", src)}))())
        g.add_edge("orchestrator", node)
        g.add_edge(node, END)
    g.add_edge(START, "orchestrator")
    return g.compile()


def release_action(decision: str) -> str:
    # The code downstream acts on the one value it is given. It fails closed: only
    # an exact "ship it" ships.
    if decision == "ship it":
        return "SHIP"
    if decision == "roll back":
        return "roll back the release"
    return "hold the release"


def run_trap(rename=False):
    # Step 1: the naive version. The framework STOPS you, loudly.
    try:
        build_trap_graph(TrapState).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".
    names = {"tests": "tests", "zz_metrics": "metrics", "changelog": "changelog"} if rename else None
    kept = build_trap_graph(TrapStateReducer, names).invoke({"decision": ""})["decision"]
    survivors = [src for src, v in VERDICTS.items() if v == kept]
    dropped = [v for v in VERDICTS.values() if v != kept]
    print("STEP 2 (reducer added): no error. the decision channel now holds:", repr(kept))
    others = "the other two reviewers'" if len(dropped) == 2 else f"{len(dropped)} reviewers'"
    print(f"        {others} verdicts were silently dropped: {dropped}")

    # Step 3: the code downstream acts on the one value it was given.
    print(f"STEP 3 (downstream): ship only on an exact 'ship it'  ->  {release_action(kept)}")
    stops = sum(v != "ship it" for v in VERDICTS.values())
    if kept == "ship it":
        print(f"        {stops} of 3 reviewers said stop; the gate saw only 'ship it', so it ships.")
    elif "roll back" in dropped:
        print(f"        {stops} of 3 reviewers said stop, one of them 'roll back'; the gate saw only"
              f" {kept!r}, so it holds where the policy says roll back.")
    else:
        print(f"        the gate saw only {kept!r}.")

    print(f"\nTRAP: {len(survivors)} of 3 verdicts survives; {len(dropped)} vanish, no error. The error"
          " could not tell a decision channel from a gather channel, and neither could the reducer"
          " you added.")
    if rename:
        print("      (renamed `metrics` to `zz_metrics`: now its verdict survives. In this graph the"
              " write from the node whose name sorts last wins, an undocumented detail.)")
    else:
        print("      (why 'hold'? here the write from the node whose name sorts last wins, an"
              " undocumented detail. Run with --rename and 'ship it' survives: the release"
              " ships although 2 of 3 reviewers said stop.)")
    print("      (with an operator.add reducer here the three strings run together into"
          " 'roll backship ithold': garbage you can see, where the drop above is silent.)")


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-5-5")    # current Sonnet as of 7 Oct 2026; reads ANTHROPIC_API_KEY

    then inside the worker, replace the stub call call_model(role="reviewer", source=source)
    with:

        _llm.invoke(f"Review the {source}. Answer with one of: ship it, hold, roll back.").content

    The worker already passes the reply through normalise_verdict(), so "Hold." and
    "Roll back - a breaking change" map to their verdicts and an answer it cannot
    map becomes "hold". 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")
    ap.add_argument("--rename", action="store_true",
                    help="with --trap: rename one node and watch a different verdict survive")
    ap.add_argument("--approve", action="store_true",
                    help="resume the human gate with approval (default: veto)")
    args = ap.parse_args()
    run_trap(args.rename) if args.trap else run_safe(args.approve)

#!/usr/bin/env python3
"""
audit.py: the schema half of The State Channel Audit. Point it at your own
LangGraph State schema (or your compiled graph) and it names every channel where
writes from parallel nodes can merge silently, then gives a PASS or FAIL you can
put in a build.

    python3 audit.py                                         # audit the demo's two schemas
    python3 audit.py my_graph.py:State --gather findings     # gate your own schema
    python3 audit.py my_graph.py:graph --gather findings     # gate a compiled graph: also
                                                             # reads nodes' private schemas
    python3 audit.py my_graph.py:State --gather findings --json
    python3 audit.py my_package.graph:State --gather messages,findings

    exit 0  PASS: every channel that carries a reducer is one you declared a gather
    exit 1  FAIL: a reducer channel you have not declared (it may be a decision)
    exit 2  nothing to audit: the file or name could not be loaded, the target is
            neither a schema class nor a graph, it has no fields, LangGraph could not
            build a graph from it, or --gather names a missing field

From Python:

    from audit import audit_schema, assert_schema_gathers
    audit_schema(MyState)                                # prints the table, returns reducer channels
    assert_schema_gathers(MyState, gather={"findings"})  # raises AssertionError on FAIL

Companion to "The Graph Looks Right. The Merge Is Where It Breaks."
https://durabilitycurve.com/tools/agent-graph/

How it reads a schema. LangGraph refuses two writes to a PLAIN channel in the
same step, loudly (InvalidUpdateError), so a plain channel is guarded: a fan-out
into it crashes the run where you can see it. The only place parallel writes
merge without a word is a channel that carries a REDUCER (or a channel object
that collects writes), because that tells the framework "merging is fine here".
So those channels are the whole silent-merge surface.

With langgraph installed, the audit asks LangGraph itself: it builds
StateGraph(your schema) and reads the channel LangGraph made for each field
(BinaryOperatorAggregate for a reducer, LastValue for a plain field, and so on).
Pointed at a compiled graph or a StateGraph, it reads graph.builder.channels,
which also holds the fields declared on nodes' private input and output schemas.
Without langgraph it falls back to a parser of LangGraph's rule (checked against
langgraph 0.6.11 and 1.2.14, 7 Oct 2026): strip one Required[...] or
NotRequired[...] wrapper, then the field is a reducer when the last item in
Annotated[...] is callable. LangGraph does not strip ReadOnly[...], so a reducer
under ReadOnly is a plain channel there and here. A doc string or label in
Annotated leaves the channel plain. The report says which way it was read.

The type alone cannot tell you what a channel MEANS, so the gate asks you to
declare it. Pass every reducer channel whose value is only read in aggregate
(a list of findings, a message history, a running total) with --gather. Any
reducer channel you leave out FAILS the gate: if a reader branches on it or acts
on its value as one answer, it is a decision channel and needs one writer and
no reducer. The hint column helps you decide:

    LIKELY DECISION  a reducer that keeps one write (last-write-wins and the
                     like), or any reducer on a str / bool / Literal / Enum.
                     This is the shape of the silent drop.
    likely gather    a reducer on a list / dict / set / tuple, or operator.add
                     on a number (a running total). Only operator.add on a list
                     or tuple keeps every write for certain; the note says what
                     a dict merge, a set union or add_messages can lose.
    classify         anything else: read how it is consumed downstream.

A third kind needs no reducer at all: a SNAPSHOT channel, where the latest value
is the contract (a current status, the active request). Keep it plain and give
it one writer.

What the schema cannot show. Which channels a node writes is the dict it returns
at runtime, so a plain channel written by two nodes in DIFFERENT steps
(A -> B, both writing `decision`) overwrites silently and passes this audit.
check_owners.py catches that half: it watches a real run and fails when a
second node writes a decision channel. Audit a State class and you see only the
fields of that class; audit the compiled graph to include reducers that live on
a node's private schema.
"""
import argparse
import importlib
import importlib.util
import json
import operator
import os
import sys
import typing
from typing import Annotated, Literal, get_args, get_origin, get_type_hints

AUDIT_VERSION = "2026-10-07b"


class SchemaReadError(Exception):
    """Nothing could be audited. Never treated as a pass."""


# ---------------------------------------------------------------------------
# Reading one field (the parser, used when langgraph is not installed)
# ---------------------------------------------------------------------------
def _mro_names(m):
    cls = m if isinstance(m, type) else type(m)
    return {c.__name__ for c in getattr(cls, "__mro__", ())}


def _is_channel(m):
    # A LangGraph channel object or class (Topic, EphemeralValue, ...), matched by
    # class name so this file runs without langgraph installed.
    return "BaseChannel" in _mro_names(m)


def _is_managed(m):
    # A managed value such as RemainingSteps: the framework fills it, nodes do not write it.
    return "ManagedValue" in _mro_names(m)


def _strip_required(hint):
    # LangGraph removes one Required[...] or NotRequired[...] wrapper before it reads
    # a field (langgraph/graph/state.py, _get_channel). It leaves ReadOnly[...] alone.
    origin = getattr(hint, "__origin__", None)
    if getattr(origin, "_name", None) in ("Required", "NotRequired"):
        return get_args(hint)[0]
    return hint


def _merger(hint):
    """What LangGraph merges this field with: ('plain', None), ('channel', obj),
    ('managed', obj) or ('reducer', fn)."""
    hint = _strip_required(hint)
    if get_origin(hint) is not Annotated:
        return "plain", None
    meta = get_args(hint)[1:]
    if _is_managed(meta[-1]):
        return "managed", meta[-1]
    for m in meta:
        if _is_channel(m):
            return "channel", m
    if callable(meta[-1]):
        return "reducer", meta[-1]
    return "plain", None


def _value_type(hint):
    hint = _strip_required(hint)
    base = get_args(hint)[0] if get_origin(hint) is Annotated else hint
    base = _strip_required(base)
    # Optional[X] -> X
    if get_origin(base) is typing.Union:
        rest = [a for a in get_args(base) if a is not type(None)]
        if len(rest) == 1:
            base = rest[0]
    return base


def _is_scalar(t):
    if get_origin(t) is Literal:
        return True
    if t in (str, bool, bytes):
        return True
    try:
        import enum
        return isinstance(t, type) and issubclass(t, enum.Enum)
    except Exception:
        return False


def _is_number(t):
    return t in (int, float, complex)


def _collection_kind(t):
    """'sequence', 'mapping', 'set' or None."""
    import collections.abc as cabc
    origin = get_origin(t) or t
    if not isinstance(origin, type) or origin in (str, bytes):
        return None
    if issubclass(origin, (list, tuple)) or issubclass(origin, cabc.Sequence):
        return "sequence"
    if issubclass(origin, dict) or issubclass(origin, cabc.Mapping):
        return "mapping"
    if issubclass(origin, (set, frozenset)) or issubclass(origin, cabc.Set):
        return "set"
    return None


class _A:
    pass


class _B:
    pass


def _keeps_one_write(fn):
    """True when the reducer returns one of its two inputs unchanged (last- or
    first-write-wins). Probed with two private sentinel objects; a reducer that
    combines its inputs raises or returns something new."""
    a, b = _A(), _B()
    try:
        out = fn(a, b)
    except Exception:
        return False
    return out is a or out is b


def _fn_name(m):
    return getattr(m, "__name__", type(m).__name__)


def _type_name(t):
    if get_origin(t) is not None:
        return str(t).replace("typing.", "")
    return getattr(t, "__name__", None) or str(t).replace("typing.", "")


def _reducer_hint(fn, hint):
    """('LIKELY DECISION' | 'likely gather' | 'classify', reason) for a reducer."""
    t = _value_type(hint) if hint is not None else None
    tname = _type_name(t) if t is not None else "this value"
    name = _fn_name(fn)
    if _keeps_one_write(fn):
        return "LIKELY DECISION", f"keeps one write of several ({tname})"
    if t is not None and _is_scalar(t):
        return "LIKELY DECISION", f"merges every write into one {tname}"
    if t is not None and _is_number(t):
        if fn is operator.add:
            return "likely gather", f"a running {tname} total"
        return "classify", f"read how {name} combines each {tname}"
    kind = _collection_kind(t) if t is not None else None
    if kind == "sequence":
        if "add_messages" in name:
            return "likely gather", ("keeps every message, except that a message whose id is "
                                     "already present replaces the old one")
        if fn is operator.add:
            return "likely gather", f"{tname} with operator.add keeps every write"
        return "likely gather", f"{tname} may keep every write: check what {name} does with repeats"
    if kind == "mapping":
        return "likely gather", (f"{tname} may lose writes: two writes to the same key keep only one; "
                                 f"check for key collisions")
    if kind == "set":
        return "likely gather", f"{tname} may lose writes: equal values collapse into one"
    return "classify", f"read how {tname} is used downstream"


# ---------------------------------------------------------------------------
# Reading a target: a schema class, or a StateGraph / compiled graph
# ---------------------------------------------------------------------------
def _target_kind(obj):
    if isinstance(obj, type):
        return "class"
    if hasattr(obj, "channels") and hasattr(obj, "schemas"):
        return "builder"
    b = getattr(obj, "builder", None)
    if b is not None and hasattr(b, "channels") and hasattr(b, "schemas"):
        return "compiled"
    return None


def _not_a_target(obj, name):
    what = "a function" if callable(obj) and not isinstance(obj, type) else f"a {type(obj).__name__}"
    return SchemaReadError(
        f"{name} is {what}, so there is no schema to read. Point at the State class "
        f"(FILE.py:State) or at a compiled graph or StateGraph object (FILE.py:graph).")


def _hints_of(schema, name):
    try:
        return get_type_hints(schema, include_extras=True)
    except Exception as e:
        raise SchemaReadError(f"could not read {name}: {type(e).__name__}: {e}") from e


def _langgraph_view(target, kind, name):
    """Channels as LangGraph builds them: (fields, channels, managed, hints, read_with),
    or None when langgraph is not installed."""
    _quiet_imports()
    try:
        from langgraph.graph import StateGraph
        from importlib.metadata import version
        lg = version("langgraph")
    except Exception:
        return None
    if kind == "class":
        hints = _hints_of(target, name)
        try:
            builder = StateGraph(target)
        except Exception as e:
            raise SchemaReadError(f"LangGraph could not build a graph from {name}: "
                                  f"{type(e).__name__}: {e}") from e
        read_with = f"the channels langgraph {lg} builds for this class"
    else:
        builder = target if kind == "builder" else target.builder
        hints = {}
        for schema in list(getattr(builder, "schemas", {}) or {}):
            try:
                for k, v in get_type_hints(schema, include_extras=True).items():
                    hints.setdefault(k, v)
            except Exception:
                pass
        read_with = (f"the channels langgraph {lg} built for this graph, "
                     f"including nodes' private schemas")
    channels = dict(builder.channels)
    managed = dict(getattr(builder, "managed", {}) or {})
    fields = [k for k in hints if k in channels or k in managed]
    fields += [k for k in list(channels) + list(managed) if k not in fields]
    return fields, channels, managed, hints, read_with


def _row_from_langgraph(field, channels, managed, hints):
    row = {"channel": field}
    if field in managed:
        row.update(kind="managed", status="managed",
                   note=f"filled by the framework ({_fn_name(managed[field])}); nodes do not write it")
        return row
    ch = channels[field]
    cname = type(ch).__name__
    if cname in ("LastValue", "LastValueAfterFinish"):
        row.update(kind="plain", status="guarded",
                   note="a parallel write here raises InvalidUpdateError")
    elif cname == "EphemeralValue" and getattr(ch, "guard", True):
        row.update(kind="channel", status="guarded",
                   note="EphemeralValue(guard=True) refuses two writes in one step")
    elif cname == "EphemeralValue":
        row.update(kind="channel", merge=cname, hint="LIKELY DECISION",
                   note="EphemeralValue(guard=False) keeps one write per step")
    elif cname == "BinaryOperatorAggregate":
        fn = getattr(ch, "operator", None)
        label, why = _reducer_hint(fn, hints.get(field))
        row.update(kind="reducer", merge=_fn_name(fn), hint=label, note=why)
    elif cname == "Topic":
        row.update(kind="channel", merge=cname, hint="likely gather", note="Topic collects every write")
    else:
        row.update(kind="channel", merge=cname, hint="classify",
                   note=f"a {cname} channel: read how it is used downstream")
    return row


def _row_from_parser(field, hint):
    kind, obj = _merger(hint)
    row = {"channel": field, "kind": kind}
    if kind == "plain":
        row.update(status="guarded", note="a parallel write here raises InvalidUpdateError")
    elif kind == "managed":
        row.update(status="managed",
                   note=f"filled by the framework ({_fn_name(obj)}); nodes do not write it")
    elif kind == "channel":
        names = _mro_names(obj)
        if "EphemeralValue" in names and getattr(obj, "guard", True):
            row.update(status="guarded", note="EphemeralValue(guard=True) refuses two writes in one step")
        elif "EphemeralValue" in names:
            row.update(merge="EphemeralValue", hint="LIKELY DECISION",
                       note="EphemeralValue(guard=False) keeps one write per step")
        elif "Topic" in names:
            row.update(merge="Topic", hint="likely gather", note="Topic collects every write")
        else:
            row.update(merge=_fn_name(obj), hint="classify", note="read how it is used downstream")
    else:
        label, why = _reducer_hint(obj, hint)
        row.update(merge=_fn_name(obj), hint=label, note=why)
    return row


def classify_schema(schema, gather=(), name=None):
    """Return the audit as a dict (no printing). `schema` is a State class, a
    StateGraph, or a compiled graph. Raises SchemaReadError."""
    name = name or getattr(schema, "__name__", "State")
    kind = _target_kind(schema)
    if kind is None:
        raise _not_a_target(schema, name)
    view = _langgraph_view(schema, kind, name)
    if view is not None:
        fields, channels, managed, hints, read_with = view
        rows = [_row_from_langgraph(f, channels, managed, hints) for f in fields]
    else:
        hints = _hints_of(schema, name)
        read_with = "the built-in parser of LangGraph's rule (langgraph is not installed)"
        rows = [_row_from_parser(f, h) for f, h in hints.items()]
    if not rows:
        raise SchemaReadError(f"{name} has no fields to audit")
    gather = set(gather or ())
    unknown = sorted(gather - {r["channel"] for r in rows})
    undeclared = []
    for r in rows:
        if "merge" in r:
            if r["channel"] in gather:
                r["status"] = "gather (declared)"
            else:
                r["status"] = "UNDECLARED"
                undeclared.append(r["channel"])
    merge_channels = [r["channel"] for r in rows if "merge" in r]
    verdict = "ERROR" if unknown else ("FAIL" if undeclared else "PASS")
    return {"schema": name, "target": kind, "read_with": read_with, "verdict": verdict,
            "channels": rows, "reducer_channels": merge_channels, "undeclared": undeclared,
            "unknown_gather": unknown, "audit_version": AUDIT_VERSION}


def _langgraph_version():
    try:
        from importlib.metadata import version
        return version("langgraph")
    except Exception:
        return None


def _print_report(rep, gate=True):
    print(f"\naudit of {rep['schema']}")
    print(f"  read with: {rep['read_with']}")
    for r in rep["channels"]:
        status = r["status"]
        detail = r.get("note", "")
        if "merge" in r:
            what = "CHANNEL" if r["kind"] == "channel" else "REDUCER"
            detail = f"{what} ({r['merge']})  {r['hint']}: {r['note']}"
        print(f"  {status:<18} {r['channel']:<16} {detail}")
    if rep["reducer_channels"]:
        print(f"\n  reducer channels (the whole silent-merge surface): {', '.join(rep['reducer_channels'])}")
        print("  for each, ask: does a reader branch on it or act on its value as one answer?")
        print("  yes: a DECISION channel. Give it one writer and drop the reducer.")
        print("  no, it is only read in aggregate: a gather. Declare it with --gather.")
        print("  a snapshot (the latest value is the contract) needs no reducer: plain, one writer.")
    else:
        print("\n  no reducer channels: a fan-out cannot merge silently here; the framework")
        print("  refuses two writes to a plain channel in one step.")
    print("  the schema cannot show a plain channel written by two nodes in sequence:")
    print("  run check_owners.py on a real run for that half.")
    if rep.get("target") == "class":
        print(f"  a reducer declared only on a node's private input or output schema is not in")
        print(f"  {rep['schema']}: audit the compiled graph (FILE.py:graph) to include those.")
    if gate:
        if rep["verdict"] == "ERROR":
            print(f"\nERROR {rep['schema']}: --gather names a field this schema does not have: "
                  f"{', '.join(rep['unknown_gather'])}")
        elif rep["verdict"] == "PASS":
            n = len(rep["reducer_channels"])
            if n == 0:
                print(f"\nPASS {rep['schema']}: no reducer channels; a parallel write to any "
                      f"channel here raises an error you will see")
            else:
                print(f"\nPASS {rep['schema']}: {n} reducer channel{'s' if n != 1 else ''}, "
                      f"all declared gathers ({', '.join(rep['reducer_channels'])})")
        else:
            print(f"\nFAIL {rep['schema']}: reducer channel not declared a gather: "
                  f"{', '.join(rep['undeclared'])}. If a reader acts on it as one answer, give it "
                  f"one writer and remove the reducer; if it is only read in aggregate, "
                  f"add it to --gather.")


def audit_schema(schema, name=None, gather=()):
    """Print the audit table and return the list of reducer channels.
    Raises SchemaReadError when there is nothing to audit."""
    rep = classify_schema(schema, gather=gather, name=name)
    _print_report(rep, gate=bool(gather))
    return rep["reducer_channels"]


def assert_schema_gathers(schema, gather, name=None):
    """For a test: raise AssertionError unless every reducer channel is declared a gather."""
    rep = classify_schema(schema, gather=gather, name=name)
    if rep["unknown_gather"]:
        raise AssertionError(f"{rep['schema']}: gather names a missing field: {rep['unknown_gather']}")
    if rep["verdict"] != "PASS":
        detail = "; ".join(f"{r['channel']} ({r['hint']}: {r['note']})"
                           for r in rep["channels"] if r["status"] == "UNDECLARED")
        raise AssertionError(f"{rep['schema']}: reducer channel not declared a gather: {detail}")
    return rep


# ---------------------------------------------------------------------------
# Loading `path/to/graph.py:State` or `package.module:State`
# ---------------------------------------------------------------------------
def _quiet_imports():
    # Two harmless notices your graph's imports may print: urllib3's NotOpenSSLWarning
    # on macOS system Python, and a langchain-core notice. langchain-core resets its
    # own warning filters when imported, so it is imported first and quietened after.
    import warnings
    warnings.filterwarnings("ignore", module="urllib3")
    try:
        import langchain_core  # noqa: F401
    except ImportError:
        pass
    warnings.filterwarnings("ignore", message=".*allowed_objects.*")


def load_schema(spec):
    _quiet_imports()
    if ":" not in spec:
        raise SchemaReadError(f"expected FILE.py:Name or module:Name, got {spec!r}")
    target, attr = spec.rsplit(":", 1)
    try:
        if target.endswith(".py") or os.sep in target or os.path.exists(target):
            path = os.path.abspath(target)
            if not os.path.exists(path):
                raise SchemaReadError(f"no such file: {target}")
            folder = os.path.dirname(path)
            if folder not in sys.path:
                sys.path.insert(0, folder)          # so the module's own sibling imports work
            modname = os.path.splitext(os.path.basename(path))[0]
            mspec = importlib.util.spec_from_file_location(modname, path)
            module = importlib.util.module_from_spec(mspec)
            sys.modules[modname] = module           # get_type_hints resolves names through this
            mspec.loader.exec_module(module)
        else:
            if os.getcwd() not in sys.path:
                sys.path.insert(0, os.getcwd())
            module = importlib.import_module(target)
    except SchemaReadError:
        raise
    except Exception as e:
        raise SchemaReadError(f"could not import {target}: {type(e).__name__}: {e}") from e
    obj = module
    for part in attr.split("."):
        if not hasattr(obj, part):
            raise SchemaReadError(f"{target} has no attribute {attr!r}")
        obj = getattr(obj, part)
    if _target_kind(obj) is None:
        raise _not_a_target(obj, attr)
    return obj, attr


def main(argv=None):
    ap = argparse.ArgumentParser(
        description="Gate a LangGraph State schema: every reducer channel must be a declared gather.")
    ap.add_argument("schema", nargs="?",
                    help="FILE.py:State, FILE.py:graph (a compiled graph) or package.module:State")
    ap.add_argument("--gather", default="",
                    help="comma-separated channels that are only read in aggregate")
    ap.add_argument("--json", action="store_true", help="print the result as JSON")
    args = ap.parse_args(argv)

    if not args.schema:
        _demo()
        return 0

    gather = [g.strip() for g in args.gather.split(",") if g.strip()]
    try:
        schema, attr = load_schema(args.schema)
        rep = classify_schema(schema, gather=gather, name=attr)
    except SchemaReadError as e:
        if args.json:
            print(json.dumps({"schema": args.schema, "verdict": "ERROR", "error": str(e),
                              "audit_version": AUDIT_VERSION}, indent=2))
        else:
            print(f"ERROR {args.schema}: {e}")
        return 2
    rep["source"] = args.schema
    rep["gather"] = gather
    rep["langgraph"] = _langgraph_version()
    if args.json:
        print(json.dumps(rep, indent=2))
    else:
        _print_report(rep)
    return {"PASS": 0, "FAIL": 1}.get(rep["verdict"], 2)


# ---------------------------------------------------------------------------
# Demo: the two schemas from agent_graph_demo.py, so you can see both verdicts.
# ---------------------------------------------------------------------------
def _demo():
    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:  python3 audit.py agent_graph_demo.py:State --gather findings ===")
    _print_report(classify_schema(State, gather={"findings"}, name="State"))
    print("\n=== the trap:  python3 audit.py agent_graph_demo.py:TrapStateReducer ===")
    _print_report(classify_schema(TrapStateReducer, name="TrapStateReducer"))
    print("\nOn your own graph:  python3 audit.py path/to/graph.py:State --gather <your gathers>")


if __name__ == "__main__":
    sys.exit(main())

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

#!/usr/bin/env python3
"""
test_release_gate.py: the first test in your release gate. Copy this file next
to your graph with audit.py and check_owners.py, change the lines marked YOURS,
and run it in CI.

    python3 test_release_gate.py      # no pytest needed; exits 1 on any failure
    pytest test_release_gate.py       # or under pytest (pip install pytest)

As shipped it tests the demo's safe release gate, so you can watch it pass,
and it runs two negative controls that prove each check can fail: a trap graph
built inside this file (a last-write-wins reducer on `decision`) must FAIL both
checks. A check that cannot fail is no check at all. The controls need nothing
from your graph or from the demo.

Companion to "The Graph Looks Right. The Merge Is Where It Breaks."
https://durabilitycurve.com/tools/agent-graph/
"""
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.*")

from typing import Annotated, TypedDict                              # noqa: E402

from langgraph.graph import StateGraph, START, END                   # noqa: E402

from audit import assert_schema_gathers                              # noqa: E402
from check_owners import assert_single_writer                        # noqa: E402

# --- YOURS: five lines. Your State schema and a factory for your compiled graph,
#     its channels, and the input of one run. -----------------------------------
from agent_graph_demo import State, build_safe_graph as build_graph  # noqa: E402  YOURS
GATHERS = {"findings"}      # YOURS: channels only ever read in aggregate
DECISIONS = ["decision"]    # YOURS: channels a reader branches on or acts on as one answer
RUN = {"findings": []}      # YOURS: the input of one run
CONFIG = {"configurable": {"thread_id": "release-gate-test"}}   # YOURS: the run's config
# RUN may also be a list of inputs run in turn on CONFIG. A graph that pauses at
# interrupt() and has a checkpointer can be resumed that way:
#     from langgraph.types import Command
#     RUN = [{"findings": []}, Command(resume=False)]
# Without a checkpointer, keep RUN a single input.


def test_schema_has_no_undeclared_merge():
    # Check one, the schema: every reducer channel is a declared gather.
    assert_schema_gathers(State, gather=GATHERS)


def test_every_decision_has_one_owner():
    # Check two, the run: each decision channel is written by one node, once per step.
    assert_single_writer(build_graph(), RUN, DECISIONS, config=CONFIG)


# --- negative controls: a trap built here must fail both checks ----------------
class _TrapState(TypedDict):
    decision: Annotated[str, lambda old, new: new]     # last-write-wins on a decision


def _build_trap():
    g = StateGraph(_TrapState)
    g.add_node("orchestrator", lambda s: {})
    for node, verdict in (("tests", "hold"), ("metrics", "ship it"), ("changelog", "roll back")):
        g.add_node(node, (lambda v=verdict: (lambda s: {"decision": v}))())
        g.add_edge("orchestrator", node)
        g.add_edge(node, END)
    g.add_edge(START, "orchestrator")
    return g.compile()


def test_control_schema_check_catches_the_trap():
    try:
        assert_schema_gathers(_TrapState, gather=set())
    except AssertionError:
        return
    raise AssertionError("the schema check passed the trap; it cannot fail, so it proves nothing")


def test_control_owner_check_catches_the_trap():
    try:
        assert_single_writer(_build_trap(), {"decision": ""}, ["decision"])
    except AssertionError:
        return
    raise AssertionError("the owner check passed the trap; it cannot fail, so it proves nothing")


if __name__ == "__main__":
    tests = [(n, f) for n, f in list(globals().items()) if n.startswith("test_") and callable(f)]
    failed = 0
    for name, fn in tests:
        try:
            fn()
            print(f"PASS  {name}")
        except AssertionError as e:
            failed += 1
            print(f"FAIL  {name}: {e}")
    print(f"\n{'FAIL' if failed else 'PASS'}: {len(tests) - failed} of {len(tests)} tests passed")
    sys.exit(1 if failed else 0)