Isolate and type fan-out payloads, check input at the front door, trace a Ctrl-C'd node (#76)

Three defects in `grapharc/runtime/graph.py`, all of the same family: a
contract the module documents, enforced everywhere except at one boundary.

**Fan-out handed every worker the same object, and never typed it.** `_enter`'s
comment states the rule — "Nodes get a deep copy: the returned dict is the
*only* write channel" — but the copy was guarded by
`isinstance(state, BaseModel)`, and a `Send` payload can be anything. Two Sends
built from one dict therefore gave both parallel workers the same live dict:

    workers reported: ['dict hits=1 id=139338015848512',
                       'dict hits=2 id=139338015848512']   # same id
    shared dict after run: {'who': 'orig', 'hits': [1, 1]}

That is a data race between nodes that are supposed to be isolated, and the
mutations travel through a channel no node declared a write to and no trace
event records — so the write-permission model and the audit trail both miss it,
and which worker ran first decides the answer. Fan-out is the one place the
isolation matters most; it was the one place that skipped it.

The other half of the same boundary: `_check_goto_target` validated `Send.node`
and left `Send.arg` alone, so `input_schema` — whose docstring says it "types a
fan-out worker's Send payload" — checked nothing. A dict where a model was
declared reached the worker and failed as a bare `AttributeError` several frames
from the dispatcher that produced it; a wrong model class sharing a field name
did not fail at all. That `Send.node` is checked and `Send.arg` is not reads as
an oversight rather than a decision, so `Send.arg` now gets the same treatment,
with `StateTypeError` naming the node, the schema and what arrived.

Every payload is deep-copied now, whatever its type. Declaring no `input_schema`
stays legal — a worker that names no schema is claiming nothing, so there is
nothing to check — but the copy is unconditional, and `add_node`'s docstring now
says so instead of leaving it to be inferred. A `BaseModel` payload was already
isolated correctly and still is, asserted by its own test.

**The front door was the one door the state contract did not hold.**
`state.py` opens by promising "a typo'd update key fails loudly at the edge
instead of silently polluting downstream nodes". `update_state` keeps that
promise and so does constructing the model; the entry points did not. They
handed `input` to LangGraph, which filters a dict down to known channels
*before* the state model is ever constructed, so `extra="forbid"` never got a
chance:

    invoke({'quesiton': 'typo'})       -> {'out': "saw:''"}     # ran on defaults
    update_state({'quesiton': 'typo'}) -> WritePermissionError: unknown state fields

Misspell the field carrying the question and you get a complete, plausible run
against an empty question and are told nothing — the quietest failure in the
runtime, on the door every user goes through first. `invoke`, `stream`,
`ainvoke`, `astream` and `astream_events` now refuse an unknown input key, in
the same words `update_state` uses, from the same helper so the two cannot
drift. A wrongly *typed* input value was already loud and still raises
Pydantic's `ValidationError`; a state model or `None` is passed through
untouched.

**A Ctrl-C'd sync node left no ending in the trace.** The sync wrapper caught
`Exception`; the async twin catches `BaseException`, and its comment gives the
reason — "a stop with no trace line is a stop nobody can audit afterwards". So
the same node body, run each way:

    sync   trace phases: [topology, start]
    async  trace phases: [topology, start, error]

`metrics.summarize` then reported `errors: 0` for the sync run, and an audit
reads it as having simply stopped between nodes. Ctrl-C is not an exotic
ending; it is the commonest way a human stops a long run. The sync wrapper now
matches its twin. The exception is re-raised untouched — only the trace write
is new.

Tests: twelve, each failing on main. Fan-out — two workers sharing one dict
cannot see each other's mutations, a payload contradicting `input_schema` is
refused at dispatch, a wrong model class likewise, an untyped payload stays
legal, and a `BaseModel` payload is still isolated. Front door — `invoke` and
`stream` reject the typo verbatim, `ainvoke`/`astream` in a parametrised twin,
with the type-error and valid-input paths pinned unchanged. Ctrl-C — one test
asserts both wrappers write the `error` event for `KeyboardInterrupt` and
`SystemExit`, so the two paths cannot drift again.

Deep-copying every payload costs nothing measurable: 200 fan-out workers run in
41.1ms with the change and 41.6ms without (best of five, dict payloads, well
inside run-to-run noise), because 200 `copy.deepcopy` calls on a realistic
shard total 0.5ms against ~41ms of graph execution.

Fixes #67
Fixes #68
Fixes #70

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

Author: Shashankss1205

CHANGELOG.md

@@ -21,3 +21,6 @@ Entries are newest-last within a release, matching the order they were written.
 - the `/live` **token check crashed on the strangers it exists to refuse**. `secrets.compare_digest` rejects `str` outside ASCII, and `_authorized` handed it the raw query parameter, so `?token=café` raised `TypeError` through the handler: an unauthenticated 500 with a traceback in the log on all four `/live` routes, where every ASCII guess correctly got a 401. The 500-vs-401 split was itself an oracle about how the token is compared. Both sides are encoded to UTF-8 now, which drops the ASCII restriction and keeps the constant-time comparison that is the whole reason `compare_digest` is there. A NUL byte in `?trace=` was the same shape one function over — `resolve_trace` raises `ValueError`, not the `LivePathError` the route caught — and is a 404 like any other malformed path now.
 - the `/live` **index advertised traces the reader refuses to serve**. `scan_traces` walked the live root with `rglob("*.jsonl")`, which matches a symlinked file by name, then parsed it and published its name, size, mtime and **run ids** on `GET /live/api/runs` and the HTML index — for a file outside the root that `/live/api/stream` then 404s, the 404 being the proof of intent. One contract, two code paths, and only the reader enforced it; the live root is documented as the Slack bot's working directory, i.e. somewhere other things write. `scan_traces` routes every candidate through `resolve_trace` now and skips symlinks outright, so a refactor of either check cannot reopen the leak. The reader's confinement — `../`, `%2e%2e%2f`, absolute paths, `sub/../../`, symlinked directories — is unchanged.
 - a **`deny` rule naming a tool literally failed open** when the name carried fnmatch metacharacters. `PermissionPolicy.decide` matched with `fnmatch(name, pattern)` alone, so `DENY "exfil[all]"` read as a character class, did not match the tool it spells, and evaluation fell through to whatever came next — typically a broad `ALLOW "*"`. The operator got no error, no warning and no deny; worse, `visible()` decides the same way, so the tool the operator had just forbidden was described to the model as available and then ran when it asked. The failure was inconsistent as well as silent: `DENY "tool?x"` happened to hold, because a `?` glob matches a literal `?`. This was the one place in the tree where a deny failed open — an unmatched tool, an unregistered kind and an unreachable backend all refuse. A `deny` or `ask` rule now also fires on an exact literal match. The widening is bound to those two tiers on purpose: equality can only add a rule that refuses or gates a call, never one that permits it, so it cannot loosen a policy the way the same change on `allow` could. For the `allow` case there is `PermissionRule.literal(action, name)`, which `glob.escape`s the name rather than widening the match, and which `default_harness` now uses for the registry names it allows. Glob semantics are untouched: `rm*` still spans `rmdir`, `*` still matches everything, the tier order and the `deny` default are unchanged.
+- **fan-out handed every worker the same payload object**, and never held it to the schema the worker declared. `_enter` deep-copied only a `BaseModel`, so two `Send`s built from one dict gave both parallel workers the *same live dict* — each reading the other's mutations, through a channel no node declared a write to and no trace event records, in the one place the isolation matters most. `_check_goto_target` validated `Send.node` against exactly this class of silent failure and left `Send.arg` alone, so `input_schema` — documented as typing a worker's payload — enforced nothing: a dict where a model was declared reached the worker and surfaced as a bare `AttributeError` frames away from the dispatcher that produced it, and a wrong model class sharing a field name never surfaced at all. Every payload is deep-copied now whatever its type, and one contradicting a declared `input_schema` is refused at dispatch with `StateTypeError` naming the node, the schema and what arrived. Declaring no `input_schema` stays legal — no claim, nothing to check — but the copy is unconditional.
+- **the front door was the one door the state contract did not hold.** `update_state` refuses an unknown field and `GraphARCState` forbids extras, but `invoke`/`stream`/`ainvoke`/`astream` handed `input` straight to LangGraph, which filters a dict down to known channels *before* the state model is ever constructed — so `extra="forbid"` never saw the typo. `invoke({"quesiton": …})` ran the whole graph on default values and returned a complete, plausible answer to an empty question, with nothing said to the caller: the quietest failure in the runtime, on the door every user goes through first. All four entry points, and `astream_events`, now refuse an unknown input key in the same words `update_state` uses. A wrongly *typed* input value was already loud and still raises Pydantic's `ValidationError`.
+- a node stopped by **Ctrl-C left no ending in the trace**. The sync wrapper caught `Exception` while its async twin catches `BaseException` for the reason its own comment gives — "a stop with no trace line is a stop nobody can audit afterwards" — so a `KeyboardInterrupt` or `SystemExit` inside a sync node escaped with no terminal `error` event, and `metrics.summarize` then reported `errors: 0` for a run an audit reads as having simply stopped between nodes. Ctrl-C is not an exotic ending; it is the commonest way a human stops a long run. The sync wrapper catches `BaseException` now and re-raises it untouched: only the record is new.

grapharc/runtime/graph.py

@@ -5,6 +5,10 @@
 
 - **Typed edges** — state schemas are Pydantic models; every value a node
   returns is validated against the field's declared type before it is written.
+  The same schema governs both other boundaries: an entry point's input keys are
+  checked rather than silently filtered down to the channels LangGraph knows,
+  and a fan-out worker's declared `input_schema` is enforced on the `Send`
+  payload it is handed.
 - **Write permissions** — every node declares which state fields it may write;
   an undeclared write raises instead of flowing downstream. A node may return a
   plain dict or a `langgraph.types.Command`; the command's `update` is checked
@@ -32,6 +36,7 @@
 from __future__ import annotations
 
 import asyncio
+import copy
 import inspect
 import threading
 import time
@@ -62,12 +67,20 @@ class WritePermissionError(Exception):
 
 
 class StateTypeError(Exception):
-    """A node returned a value that its state field's declared type rejects.
-
-    LangGraph drops unknown keys before validating an update, so a declared
-    field carrying the wrong *value* used to sail through to the graph's output.
-    GraphARC validates each write against the schema at the node boundary
-    instead — the type annotation is the contract, not documentation.
+    """A value contradicted the type its schema declares for it.
+
+    Two boundaries raise this, both for the same reason — a declared type is the
+    contract, not documentation:
+
+    - **A node's write.** LangGraph drops unknown keys before validating an
+      update, so a declared field carrying the wrong *value* used to sail
+      through to the graph's output. GraphARC validates each write against the
+      schema at the node boundary instead.
+    - **A fan-out `Send` payload.** A worker declaring `input_schema` is stating
+      what it is handed; LangGraph passes `Send.arg` to the node untouched, so a
+      payload of the wrong class used to surface deep inside the worker as an
+      `AttributeError` — or not at all, if the shapes happened to overlap.
+      GraphARC checks it at dispatch, next to where `Send.node` is checked.
     """
 
 
@@ -308,6 +321,7 @@ def __init__(
         # recorded and a renderer connects the workers that actually ran.
         self._conditional_edges: list[tuple[str, str]] = []
         self._fanout_sources: list[str] = []
+        self._input_schemas: dict[str, type[BaseModel] | None] = {}
         self._adapters: dict[str, TypeAdapter[Any]] = {}
 
     def add_node(
@@ -319,14 +333,23 @@ def add_node(
         input_schema: type[BaseModel] | None = None,
     ) -> GraphARC:
         """Register a node. `input_schema` types a fan-out worker's Send payload
-        (defaults to the graph state schema)."""
+        (defaults to the graph state schema).
+
+        A declared `input_schema` is enforced: a `Send` carrying anything that is
+        not an instance of it is refused at dispatch with `StateTypeError`, the
+        way an unknown `Send.node` is refused with `GraphRoutingError`. Leaving
+        `input_schema` unset keeps an untyped payload legal — the worker is then
+        saying nothing about what it accepts — but the payload is deep-copied on
+        the way in either way, so parallel workers never share one object.
+        """
         writes_set = set(writes)
         unknown = writes_set - set(self.state_schema.model_fields)
         if unknown:
             raise WritePermissionError(
                 f"node {name!r} declares writes to unknown state fields: {sorted(unknown)}"
             )
         self._nodes[name] = writes_set
+        self._input_schemas[name] = input_schema
         for field in sorted(writes_set):
             self._build_adapter(field)
         self._graph.add_node(
@@ -454,6 +477,32 @@ def _destinations(self) -> str:
         """The destinations this graph can actually route to, for an error message."""
         return ", ".join([*(repr(name) for name in sorted(self._nodes)), "END"])
 
+    def _check_send_payload(self, who: str, send: Send) -> None:
+        """Hold a `Send.arg` to the worker's declared `input_schema`.
+
+        `Send.node` is checked because LangGraph drops an unknown target without
+        an error; `Send.arg` is checked for the mirror-image reason — LangGraph
+        hands the payload to the worker *exactly* as given, `input_schema` or
+        not, so a wrong-class payload is discovered by the worker's own body, as
+        an `AttributeError` several frames from the dispatcher that produced it.
+        A shard is data with a declared shape; it is refused where it is
+        dispatched, not where it is dereferenced.
+
+        No schema declared means no claim was made, so nothing to check — see
+        `add_node`. Payload isolation is separate and unconditional: `_enter`
+        deep-copies whatever arrives.
+        """
+        schema = self._input_schemas.get(send.node)
+        if schema is None or isinstance(send.arg, schema):
+            return
+        raise StateTypeError(
+            f"{who} sent node {send.node!r} a payload its input_schema rejects: "
+            f"expected {schema.__name__}, got {type(send.arg).__name__} "
+            f"({_short_repr(send.arg)}); LangGraph hands a Send payload to the "
+            f"worker unchecked, so this would surface inside the worker's body "
+            f"instead of here"
+        )
+
     def _check_goto_target(self, who: str, target: Any) -> None:
         """Reject one routing destination this graph cannot reach. See `GraphRoutingError`."""
         if isinstance(target, Send):
@@ -469,6 +518,7 @@ def _check_goto_target(self, who: str, target: Any) -> None:
                     f"{', '.join(repr(n) for n in sorted(self._nodes)) or '(none)'} "
                     f"— END is not one, because a Send has to name a node that runs"
                 )
+            self._check_send_payload(who, target)
             return
         if isinstance(target, str):
             if target == END or target in self._nodes:
@@ -620,8 +670,17 @@ def emit(phase: str, **kw: Any) -> None:
         # Nodes get a deep copy: the returned dict is the *only* write channel.
         # Without this, in-place mutation of nested models would bypass write
         # permissions invisibly (Pydantic passes nested models by reference).
+        #
+        # *Every* input, not only a BaseModel one: a fan-out `Send` payload can
+        # be any object, and two Sends built from one dict used to hand the
+        # workers the same live object — parallel nodes mutating shared state,
+        # which is a data race whose writes appear in nobody's declared writes.
+        # Fan-out is where the isolation matters most, so it cannot be the one
+        # path that skips it.
         if isinstance(state, BaseModel):
             state = state.model_copy(deep=True)
+        else:
+            state = copy.deepcopy(state)
 
         try:
             ctx.meter.check()
@@ -735,7 +794,12 @@ def wrapped(state: Any, config: RunnableConfig) -> Any:
                     deadline_guard(ctx.meter, what=f"node {name!r}"),
                 ):
                     result = fn(state, ctx) if wants_ctx else fn(state)
-            except Exception as exc:
+            except BaseException as exc:
+                # BaseException, not Exception, for the reason the async twin
+                # gives: a sync node is most often stopped by a human hitting
+                # ^C, which is a KeyboardInterrupt and not an Exception, and a
+                # stop with no trace line is a stop nobody can audit afterwards.
+                # The exception is re-raised untouched; only the record is new.
                 emit("error", duration_ms=(time.perf_counter() - t0) * 1000, error=repr(exc))
                 raise
             return self._leave(
@@ -836,6 +900,32 @@ def _thread_config(self, thread_id: str, checkpoint_id: str | None = None) -> di
             configurable["checkpoint_id"] = checkpoint_id
         return {"configurable": configurable}
 
+    def _reject_unknown_fields(self, who: str, values: dict[str, Any]) -> None:
+        """Refuse keys the state schema does not have. One wording, every door."""
+        unknown = set(values) - set(self.arc.state_schema.model_fields)
+        if unknown:
+            raise WritePermissionError(
+                f"{who} targets unknown state fields: {sorted(unknown)}"
+            )
+
+    def _checked_input(self, entry: str, input: Any) -> Any:
+        """Hold an entry point's `input` to the state schema's field names.
+
+        The state schema forbids extra fields, and `update_state` refuses an
+        unknown key — but LangGraph filters a dict input down to the channels it
+        knows *before* the state model is ever constructed, so `extra="forbid"`
+        never sees the typo and the graph runs to completion on default values.
+        A misspelled question field is then a complete, plausible-looking run
+        against an empty question, with nothing said to the caller. The front
+        door gets the same refusal the side door already gives.
+
+        Only a dict is checked: a state model has already been validated by
+        Pydantic, and `None` means "resume from the last checkpoint".
+        """
+        if isinstance(input, dict):
+            self._reject_unknown_fields(f"{entry}()", input)
+        return input
+
     # -- running -----------------------------------------------------------
 
     def invoke(
@@ -854,6 +944,7 @@ def invoke(
         the thread's history so replay points stay unique across resumes.
         """
         self._reject_async_nodes("invoke")
+        input = self._checked_input("invoke", input)
         return self.inner.invoke(input, self._run_config(thread_id, run_id, budget))
 
     def stream(
@@ -871,6 +962,7 @@ def stream(
         closed with MissingRunContextError.
         """
         self._reject_async_nodes("stream")
+        input = self._checked_input("stream", input)
         yield from self.inner.stream(
             input, self._run_config(thread_id, run_id, budget), **stream_kwargs
         )
@@ -889,6 +981,7 @@ async def ainvoke(
         the wrapper's contract does not change. What does change for `async def`
         nodes is how `max_seconds` is delivered: see `_async_deadline`.
         """
+        input = self._checked_input("ainvoke", input)
         return await self.inner.ainvoke(input, self._run_config(thread_id, run_id, budget))
 
     async def astream(
@@ -901,6 +994,7 @@ async def astream(
         **stream_kwargs: Any,
     ) -> AsyncIterator[Any]:
         """Async twin of `stream()`; `.inner.astream()` fails closed."""
+        input = self._checked_input("astream", input)
         async for chunk in self.inner.astream(
             input, self._run_config(thread_id, run_id, budget), **stream_kwargs
         ):
@@ -924,6 +1018,7 @@ async def astream_events(
         """
         if version not in ("v1", "v2"):
             raise ValueError(f"astream_events supports version 'v1' or 'v2', got {version!r}")
+        input = self._checked_input("astream_events", input)
         async for event in self.inner.astream_events(
             input, self._run_config(thread_id, run_id, budget), version=version, **kwargs
         ):
@@ -1005,11 +1100,7 @@ def _checked_values(self, values: Any, as_node: str | None) -> dict[str, Any]:
             raise WritePermissionError(
                 f"update_state takes a dict of field updates, got {type(values)!r}"
             )
-        unknown = set(values) - set(self.arc.state_schema.model_fields)
-        if unknown:
-            raise WritePermissionError(
-                f"update_state targets unknown state fields: {sorted(unknown)}"
-            )
+        self._reject_unknown_fields("update_state", values)
         if as_node is not None and as_node in self.arc._nodes:
             return self.arc._check_update(
                 f"update_state(as_node={as_node!r})", self.arc._nodes[as_node], values

tests/test_async_kernel.py

@@ -804,3 +804,47 @@ async def test_aupdate_state_enforces_declared_writes():
     await compiled.ainvoke({"a": 1}, thread_id="t1")
     with pytest.raises(WritePermissionError, match="undeclared"):
         await compiled.aupdate_state("t1", {"b": 5}, as_node="n")
+
+
+# -- sync/async parity: neither wrapper may lose the ending -----------------
+
+
+# Driven with `asyncio.run` rather than `@pytest.mark.asyncio`: asyncio re-raises
+# KeyboardInterrupt and SystemExit out of the task step and into the loop, so they
+# leave the run at the runner rather than at the await. Both wrappers are asserted
+# in one test so neither can quietly stop matching the other.
+@pytest.mark.parametrize("stopper", [KeyboardInterrupt, SystemExit])
+def test_both_wrappers_record_a_node_stopped_by_a_baseexception(trace, stopper):
+    """Ctrl-C is the commonest way a human ends a long run, and a
+    KeyboardInterrupt is not an Exception. The sync wrapper used to let it past
+    without a trace line, so the audit trail ended mid-node with no reason and
+    `metrics.summarize` reported zero errors for the run."""
+
+    def boom(state: S) -> dict:
+        raise stopper("operator hit ^C inside a node")
+
+    async def aboom(state: S) -> dict:
+        await asyncio.sleep(0)
+        raise stopper("operator hit ^C inside a node")
+
+    with pytest.raises(stopper):
+        _graph(boom, writes={"a"}, trace=trace).invoke({}, run_id="r-stop")
+    with pytest.raises(stopper):
+        asyncio.run(_graph(aboom, writes={"a"}, trace=trace).ainvoke({}, run_id="r-astop"))
+
+    for run_id in ("r-stop", "r-astop"):
+        errors = [e for e in trace.read_events(run_id) if e.phase == "error"]
+        assert errors, f"{run_id} recorded no terminal error event"
+        assert stopper.__name__ in errors[0].error
+
+
+@pytest.mark.asyncio
+async def test_the_async_entry_points_refuse_an_unknown_input_key_too():
+    """`invoke`/`stream` fail loudly on a typo'd input key; so must their twins."""
+    compiled = _graph(lambda s: {"a": 1}, writes={"a"})
+    with pytest.raises(WritePermissionError) as caught:
+        await compiled.ainvoke({"aa": 1})
+    assert str(caught.value) == "ainvoke() targets unknown state fields: ['aa']"
+    with pytest.raises(WritePermissionError) as caught:
+        [chunk async for chunk in compiled.astream({"aa": 1})]
+    assert str(caught.value) == "astream() targets unknown state fields: ['aa']"