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

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']"