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