From febca6fcf4445097f3987f97340d7303330c808f Mon Sep 17 00:00:00 2001 From: Shashank Shekhar Singh Date: Tue, 4 Aug 2026 00:49:58 +0530 Subject: [PATCH] Isolate and type fan-out payloads, check input at the front door, trace a Ctrl-C'd node MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- CHANGELOG.md | 3 + grapharc/runtime/graph.py | 117 ++++++++++++++++++++++++---- tests/test_async_kernel.py | 44 +++++++++++ tests/test_runtime_discipline.py | 128 ++++++++++++++++++++++++++++++- 4 files changed, 278 insertions(+), 14 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index fe78119..c74becf 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,3 +20,6 @@ Entries are newest-last within a release, matching the order they were written. - a **NUL byte in a path came back as silence**, the worst answer a chat bot can give: `Path(raw).resolve()` raises `ValueError`, `handle_text_live` catches only `SlackCommandError`, so `trace a\x00b` escaped the bolt listener as an unhandled exception and the requester saw no reply at all — indistinguishable from the bot being down. A NUL anywhere in the request is now a refusal in the same voice the core tools already use ("cannot name a file"), and `_confined` turns any `ValueError`/`OSError` out of the filesystem into a refusal too, for callers of its own. Folded in from the same report: the flag allowlist tested `token.startswith("--")`, so a single-dash token slipped it and was spent as a positional — `trace -h` was admitted with `-h` as the path. Any leading dash is a flag now, and one not on the list is refused like any other. - 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. +- **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. diff --git a/grapharc/runtime/graph.py b/grapharc/runtime/graph.py index 9058646..71708c0 100644 --- a/grapharc/runtime/graph.py +++ b/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,7 +333,15 @@ 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: @@ -327,6 +349,7 @@ def add_node( 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 diff --git a/tests/test_async_kernel.py b/tests/test_async_kernel.py index 7d1e3c6..d04b45a 100644 --- a/tests/test_async_kernel.py +++ b/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']" diff --git a/tests/test_runtime_discipline.py b/tests/test_runtime_discipline.py index 045bee8..00a31e6 100644 --- a/tests/test_runtime_discipline.py +++ b/tests/test_runtime_discipline.py @@ -4,7 +4,7 @@ from typing import Annotated import pytest -from pydantic import BaseModel, Field +from pydantic import BaseModel, Field, ValidationError from grapharc.runtime.budget import Budget, BudgetExceeded, BudgetMeter from grapharc.runtime.graph import ( @@ -155,6 +155,132 @@ def observe(state: NestedState) -> dict: assert result["items"][0].text == "original" +# -- fan-out payloads: the same isolation, and the declared input contract --- + + +class FanState(GraphARCState): + ran: Annotated[list[str], operator.add] = [] + + +class Shard(BaseModel): + who: str = "" + hits: list[int] = [] + + +def _fanout(payloads, *, worker, input_schema=None): + g = GraphARC(FanState, name="fan") + g.add_node("plan", lambda s: None, writes=set()) + g.add_node("worker", worker, writes={"ran"}, input_schema=input_schema) + g.add_edge(START, "plan") + g.add_fanout_edge("plan", lambda s: [("worker", p) for p in payloads]) + g.add_edge("worker", END) + return g.compile() + + +def test_two_workers_fanned_out_with_one_payload_do_not_share_it(): + """The deep copy is not only for BaseModel state. Two Sends built from one + dict used to hand both workers the same live object — parallel nodes reading + each other's mutations, through a channel no node declared a write to.""" + + def worker(payload: dict) -> dict: + payload["hits"].append(1) + return {"ran": [f"hits={len(payload['hits'])}"]} + + shared = {"who": "orig", "hits": []} + result = _fanout([shared, shared], worker=worker).invoke({}) + assert result["ran"] == ["hits=1", "hits=1"] + assert shared == {"who": "orig", "hits": []} + + +def test_a_basemodel_payload_is_still_isolated(): + """The path that already worked has to keep working.""" + + def worker(payload: Shard) -> dict: + payload.hits.append(1) + return {"ran": [f"hits={len(payload.hits)}"]} + + shard = Shard(who="orig") + result = _fanout([shard, shard], worker=worker, input_schema=Shard).invoke({}) + assert result["ran"] == ["hits=1", "hits=1"] + assert shard.hits == [] + + +def test_a_payload_contradicting_input_schema_is_refused_at_dispatch(): + """`input_schema` is a claim about what the worker is handed; LangGraph + passes Send.arg through untouched, so GraphARC checks it where Send.node is + checked rather than letting the worker's body discover it.""" + compiled = _fanout( + [{"who": "orig"}], worker=lambda p: {"ran": [p.who]}, input_schema=Shard + ) + with pytest.raises(StateTypeError) as caught: + compiled.invoke({}) + message = str(caught.value) + assert "fan-out dispatcher on node 'plan'" in message + assert "node 'worker'" in message + assert "expected Shard" in message and "got dict" in message + + +def test_a_payload_of_the_wrong_model_class_is_refused_too(): + """It used to reach the worker and surface as a bare AttributeError.""" + + class Other(BaseModel): + nope: int = 0 + + compiled = _fanout( + [Other()], worker=lambda p: {"ran": [p.who]}, input_schema=Shard + ) + with pytest.raises(StateTypeError, match="got Other"): + compiled.invoke({}) + + +def test_a_worker_declaring_no_input_schema_accepts_any_payload(): + """No schema is no claim — the payload is still copied, but not typed.""" + compiled = _fanout([{"who": "a"}, {"who": "b"}], worker=lambda p: {"ran": [p["who"]]}) + assert sorted(compiled.invoke({})["ran"]) == ["a", "b"] + + +# -- the front door: input keys are checked like any other write ------------ + + +class Front(GraphARCState): + question: str = "" + out: str = "" + + +def _front_door(): + g = GraphARC(Front, name="front") + g.add_node("n", lambda s: {"out": f"saw:{s.question!r}"}, writes={"out"}) + g.add_edge(START, "n") + g.add_edge("n", END) + return g.compile() + + +def test_invoke_refuses_an_input_key_the_schema_does_not_have(): + """LangGraph filters input down to known channels before the state model is + constructed, so `extra="forbid"` never saw the typo and the graph ran to + completion on defaults — a plausible-looking answer to an empty question.""" + with pytest.raises(WritePermissionError) as caught: + _front_door().invoke({"quesiton": "typo"}) + assert str(caught.value) == "invoke() targets unknown state fields: ['quesiton']" + + +def test_stream_refuses_it_in_the_same_words(): + with pytest.raises(WritePermissionError) as caught: + list(_front_door().stream({"quesiton": "typo"})) + assert str(caught.value) == "stream() targets unknown state fields: ['quesiton']" + + +def test_a_wrongly_typed_input_value_is_still_pydantics_to_report(): + """Only the unknown-key case changed; a type error was already loud.""" + with pytest.raises(ValidationError): + _front_door().invoke({"out": 5}) + + +def test_legal_input_reaches_the_graph_untouched(): + assert _front_door().invoke({"question": "q"})["out"] == "saw:'q'" + assert _front_door().invoke(Front(question="q"))["out"] == "saw:'q'" + + class Typed(GraphARCState): count: int = 0 label: str = ""