From 0309ede8e590fd86740e9ea37b87604174ad988a Mon Sep 17 00:00:00 2001 From: Shashank Shekhar Singh Date: Wed, 5 Aug 2026 00:22:26 +0530 Subject: [PATCH] A routing typo waited for a run to fail, and a reused run id merged two of them MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two declaration-time gaps, one in the kernel and one in the CLI, both of the same shape: something knowable before a run started was left for the run to discover, or not to. `add_conditional_edge` handed the router and its mapping to LangGraph untouched, so a mapping pointing at a node nobody added was accepted and the first branch to take it died on `self.ends[key]` — a bare `KeyError` from inside LangGraph's branch machinery, naming neither the graph, the source node nor the router. The mapping is topology and was checkable all along: an empty mapping is refused now, every unreachable target is named alongside the key that leads to it, and a router annotated with what it returns (a `Literal`, an `Enum`) has those members held against the mapping's keys with the same hash lookup LangGraph will use. A router that annotates nothing is still not second-guessed — but the key it returns is checked when it returns one, and raises `GraphRoutingError` naming the node, the key and the keys declared. The wrapper keeps the router's name and annotations: LangGraph branches by the one and infers the branch's input schema from the other. Every executing command appends to its `--trace`, which is right — `diff` reads two runs out of one file — but nothing checked whether the `--run-id` the operator passed was already in there. Two runs then merged under one name, and `metrics` summed both runs' tokens, `viz` welded the second path onto the first, `replay` reconstructed a chimera, with no signal at any point. The appendable file was never the defect; the reused id was. `plan`, `run` and `agent` (both executors) refuse an explicit `--run-id` that already has events in the target trace, exit 2, before a single event is written. Fail closed rather than auto-rename: the id is the name an operator looks the run up under later. Generated ids pay for no scan, and different ids in one file are untouched. Fixes #4 Fixes #29 Co-Authored-By: Claude Fable 5 --- CHANGELOG.md | 2 + README.md | 2 +- docs/cookbook/01-basics.md | 20 +++-- grapharc/cli/agent.py | 6 ++ grapharc/cli/delegate.py | 6 ++ grapharc/cli/graphrun.py | 9 ++ grapharc/cli/main.py | 14 ++- grapharc/cli/plan.py | 10 +++ grapharc/cli/runid.py | 88 ++++++++++++++++++ grapharc/runtime/graph.py | 149 ++++++++++++++++++++++++++++++- tests/test_async_kernel.py | 41 +++++++++ tests/test_cli.py | 113 +++++++++++++++++++++++ tests/test_runtime_discipline.py | 125 +++++++++++++++++++++++++- 13 files changed, 570 insertions(+), 15 deletions(-) create mode 100644 grapharc/cli/runid.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 0b15fcf..16a7f3f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,8 @@ Entries are newest-last within a release, matching the order they were written. - a run **stopped for overspending reported spending nothing**. Tokens were attributed from `end` events, and a node the budget interrupts emits `error` instead — so `grapharc metrics` answered `tokens: 0` for a run whose own enforcement message named the figure that stopped it (`max_tokens reached (51/5)`). The audit trail lost precisely the number the stop was about, and per-node attribution dropped the most expensive node in the run. Every `error` event is now stamped with what its node spent, exactly as `end` is, and both `summarize` and the cost report count it; sub-events inside a node remain a breakdown of its total rather than an addition, so the disjointness that kept `ends + orphans` from double-counting is unchanged, and `RunCost.tokens == RunMetrics.tokens` still holds. - the `.env` credential loader **walked up parent directories to `/`**, while the config layer next door refuses exactly that on principle — so the file that *spends money* was discovered more eagerly than the one that *constrains* a run. A run started in a scratch subdirectory picked up an `OPENROUTER_API_KEY` from any ancestor: a `.env` in `$HOME` billed every user's experiment on a shared box to that key, a demo checked out under a client project quietly used the client's key, and since `redact()` is the only thing that ever prints a key, nothing in normal operation said *which file paid*. The rationale `cli/config.py` wrote down for `grapharc.toml` — "a run must never be silently governed by a file in a directory you didn't know about" — applies with more force to the file that pays than to the file that restrains, so `find_env_file` now reads the start directory (default: the working directory) and no ancestor of it. **This is a behaviour change:** anyone relying on a parent-directory `.env` must move it into the directory they run from, `export` the variable, or pass `env_file=` naming the file. Neither escape hatch moved — a real environment variable still beats any file, and an explicit `env_file=` still reads a file anywhere on disk — and no "search boundary" was added in place of the walk, because stopping at a git root is still an upward search. +- **the one edge-declaration path that still deferred its error.** `add_conditional_edge` passed the router and its mapping straight through to LangGraph, so a mapping pointing at a node nobody added was accepted, an empty mapping was accepted, and the first run to take that branch died on `self.ends[key]` — a bare `KeyError` raised from inside LangGraph's branch machinery, naming neither the graph, the source node, nor the router that produced the key. Everywhere else this kernel fails at declaration: an undeclared write raises at `add_node`, a write to a field the schema does not have raises at `add_node`, a cycle is refused at `compile()`. The mapping's targets were knowable all along. They are checked now, at `add_conditional_edge`, with an empty mapping refused and every unreachable target named alongside the key that leads to it; a router that annotates what it returns — a `Literal`, an `Enum` — has those members held against the mapping's keys, using the same hash lookup LangGraph will use, so the check predicts the failure rather than approximating it. A router that annotates nothing is still not second-guessed: predicting an arbitrary function's return value is not a check, and inventing a requirement would be worse than the gap. That last case is no longer a `KeyError`, though — the router is wrapped so an unmapped key raises `GraphRoutingError` naming the node, the key and the keys that were declared, which is what the rest of the kernel raises for a transition it cannot make. The wrapper keeps the router's name and annotations, because LangGraph names the branch after the one and infers the branch's input schema from the other. +- a **reused `--run-id` silently welded two runs into one record.** Every executing command appends to its `--trace` file — by design, since `grapharc diff` reads two runs out of one file — and nothing checked whether the id the operator passed was already in there. Running the same `plan` twice with one `--trace`/`--run-id` pair produced a single "run" whose `metrics` summed both runs' tokens and node counts, whose `viz` drew the second path welded onto the end of the first, and whose `replay` reconstructed a chimera; the operator got no signal at any point, and the trace is documented as the record the metrics cannot disagree with. The file being appendable was never the defect — the id being reused was, so the guard sits at the start of the run rather than in the recorder: `plan`, `run` and `agent` (both executors) refuse an explicit `--run-id` that already has events in the target trace, with exit 2 naming the id, the count and the file, before a single event is written. Fail closed rather than auto-renaming, because a run id is the name an operator will look the run up under later and picking a different one silently is the same class of surprise. Generated ids are untouched — fresh by construction, so they pay for no scan — and different ids in one file stay exactly as they were. ## 0.1.3 diff --git a/README.md b/README.md index 5f7dbec..6cb6caa 100644 --- a/README.md +++ b/README.md @@ -269,7 +269,7 @@ Three of those need their edges stated, because the gap is where people get hurt **Budgets.** Tokens are charged without the node's cooperation: a LangChain callback is installed for the duration of every node, so any chat model invoked on that thread reports usage to the run's meter — including calls buried inside library code the node merely calls — and the ceiling is enforced at the node boundary. `max_seconds` is an interrupt, not a poll: SIGALRM on the main thread, an asynchronous exception otherwise, so a node parked in `time.sleep` or on a provider's socket is cut off at the deadline. Where it stops short: spend a provider never reports cannot be charged, a model invoked on a thread the node started itself is outside the callback's context, and an async exception cannot unwind a thread sitting inside a C call — it lands when that call returns. Even then the deadline holds at the node boundary: a node that overran does not get its writes into state. -**Routing.** The routers are code, which is the property that matters: no model output is ever consulted to pick an edge. But `add_conditional_edge` passes the router and its mapping straight through to LangGraph — GraphARC does not verify that the router's return value is a key in the mapping, so a typo surfaces as a `KeyError` at run time rather than when the edge is added. +**Routing.** The routers are code, which is the property that matters: no model output is ever consulted to pick an edge. `add_conditional_edge` checks the mapping where it is declared — an empty mapping is refused, every target must name a node the graph has or `END`, and a router annotated with what it returns (a `Literal`, an `Enum`) has those members held against the mapping's keys. Where it stops short: a router that declares nothing is not second-guessed, so the key it returns is only known when it returns one. That case is no longer a bare `KeyError` from inside LangGraph's branch machinery — it raises `GraphRoutingError` naming the node, the key and the keys there were — but it is still discovered by a run rather than by `add_conditional_edge`. **Typing.** Writes are checked in both directions: the dict a node returns is validated field by field against the state schema before it lands, and the state is validated again when the next node receives it. A value that doesn't fit raises `StateTypeError` naming the node, the field, the declared type and what arrived — and that includes the last node before `END`, so a bad type no longer escapes into the result. The validated value is what gets written, so a schema that says `int` means the result holds an `int`. The remaining gap is narrow and worth stating exactly: write-time validation is built from each field's *annotation*, so constraints carried in the annotation (`Annotated[int, Field(gt=0)]`) do bite, but a validator the state model declares for itself — `@field_validator`, `@model_validator` — is not run on a write. A node returning `{"slug": "NOT-LOWER"}` into a field whose validator demands lowercase is accepted, even though constructing the model directly with that value raises; the violation surfaces only when a later node receives the state and the whole model is rebuilt, which means one written by the last node before `END` still reaches the result. The write *allowlist* is GraphARC's; the *types* are Pydantic's. diff --git a/docs/cookbook/01-basics.md b/docs/cookbook/01-basics.md index 28a0dc2..5e94704 100644 --- a/docs/cookbook/01-basics.md +++ b/docs/cookbook/01-basics.md @@ -741,11 +741,15 @@ No model output is ever consulted to pick an edge, so no amount of prose in a mo reply can steer the graph — a model that writes `ROUTE TO: all_verified` into a state field is writing a string, not choosing a branch. -One gap to know: `add_conditional_edge` passes your router and mapping straight -through to LangGraph, and GraphARC does **not** check that the router's return -values are keys of the mapping. A typo shows up as a `KeyError` at run time, not -when the edge is added. `StopReason` is a `StrEnum`, so using its members as your -mapping keys is a cheap way to make that typo impossible. +What `add_conditional_edge` checks, and when: the mapping is read at declaration +time, so an empty one is refused, a target naming a node you never added raises +there and then, and a router annotated `-> Literal["again", "stop"]` (or with an +`Enum` return type) has those members held against your mapping's keys. A router +that declares nothing is left alone — the key it returns is only knowable when it +returns one — but that case is no longer a bare `KeyError` from inside LangGraph: +it raises `GraphRoutingError` naming the node, the key and the keys you declared. +`StopReason` is a `StrEnum`, so annotating your router with it moves that last +check to declaration time too. --- @@ -1523,5 +1527,7 @@ but repeated here because they are the ones that surprise people: node writes — only the field's annotation is enforced, which does include a nested model's own validators. Rebuild the model at your program's boundary, or keep the invariant one level down. -2. `add_conditional_edge` does not verify that your router's return values are keys - of your mapping. A typo is a run-time `KeyError`. +2. `add_conditional_edge` checks its mapping when the edge is added — the targets, + and a router that annotates what it returns. A router that annotates nothing is + not second-guessed, so the key it returns is checked when it returns one; that + is a `GraphRoutingError` naming the router, not a bare `KeyError`. diff --git a/grapharc/cli/agent.py b/grapharc/cli/agent.py index 952ac36..7a7b74b 100644 --- a/grapharc/cli/agent.py +++ b/grapharc/cli/agent.py @@ -22,6 +22,7 @@ from grapharc.cli import optional, style from grapharc.cli.output import EXIT_FAILED, EXIT_OK, emit, fail +from grapharc.cli.runid import refuse_reused_run_id # Entry points accepted from `grapharc.tools`, in preference order: a registrar # that fills a registry, and a factory that returns specs. Both are supported @@ -173,6 +174,11 @@ def run_agent( workspace = Path(workspace).expanduser().resolve() workspace.mkdir(parents=True, exist_ok=True) trace_path = Path(trace_path) if trace_path else workspace / "trace.jsonl" + # While `run_id` still says whether the operator chose one: the generated + # id below is fresh by construction and has nothing to collide with. + reused = refuse_reused_run_id(trace_path, run_id, command="agent", as_json=as_json, task=task) + if reused is not None: + return reused run_id = run_id or f"agent-{uuid.uuid4().hex[:8]}" try: diff --git a/grapharc/cli/delegate.py b/grapharc/cli/delegate.py index 472b459..9178f97 100644 --- a/grapharc/cli/delegate.py +++ b/grapharc/cli/delegate.py @@ -30,6 +30,7 @@ from grapharc.cli import style from grapharc.cli.output import EXIT_FAILED, EXIT_OK, emit, fail +from grapharc.cli.runid import refuse_reused_run_id #: What a bare run may use, mirroring the harness default of "the core tools, #: shell included". An explicit `--allow` replaces this outright. @@ -209,6 +210,11 @@ def run_delegated( workspace = Path(workspace).expanduser().resolve() workspace.mkdir(parents=True, exist_ok=True) trace_path = Path(trace_path) if trace_path else workspace / "trace.jsonl" + # `--executor claude-cli` resolves its own trace path, so it owns the same + # guard `run_agent` applies to the sandboxed path. + reused = refuse_reused_run_id(trace_path, run_id, command="agent", as_json=as_json, task=task) + if reused is not None: + return reused run_id = run_id or f"agent-{uuid.uuid4().hex[:8]}" allowed = list(allow) if allow and allow != ["*"] else list(DEFAULT_DELEGATED_TOOLS) diff --git a/grapharc/cli/graphrun.py b/grapharc/cli/graphrun.py index fa2263b..8b01a2e 100644 --- a/grapharc/cli/graphrun.py +++ b/grapharc/cli/graphrun.py @@ -44,6 +44,7 @@ from grapharc.cli.generate import resolve_or_generate_policy from grapharc.cli.output import EXIT_FAILED, EXIT_OK, emit, fail from grapharc.cli.plan import PlanSetupError, resolve_registry +from grapharc.cli.runid import refuse_reused_run_id #: Stage names `demo` owns. Kept so `grapharc run stage0` — which worked before #: the split — fails with a redirection rather than an argparse complaint about @@ -162,6 +163,14 @@ def run_graph( schema = state_schema or IncidentState trace_path = trace_path or Path(tempfile.mkdtemp(prefix="grapharc-run-")) / "trace.jsonl" + # Ahead of the recorder, and ahead of the admission check that writes the + # first event: an id already in this file would merge this run with an + # earlier one, and `--check-only` writes its verdict under that id too. + reused = refuse_reused_run_id( + trace_path, run_id, command="run", as_json=as_json, graph=graph_path + ) + if reused is not None: + return reused trace = TraceRecorder(trace_path) checker = AdmissionChecker( registry=registry, diff --git a/grapharc/cli/main.py b/grapharc/cli/main.py index 71fbef4..b59e029 100644 --- a/grapharc/cli/main.py +++ b/grapharc/cli/main.py @@ -703,7 +703,9 @@ def build_parser() -> argparse.ArgumentParser: "--tenant", default=None, metavar="NAME", help="tenant to compile --policy for" ) run.add_argument("--trace", type=Path, default=None, help="trace JSONL output path") - run.add_argument("--run-id", default=None, help="name this run") + run.add_argument( + "--run-id", default=None, help="name this run; refused if --trace already holds it" + ) run.add_argument( "--max-tokens", type=int, @@ -773,7 +775,9 @@ def build_parser() -> argparse.ArgumentParser: "--tenant", default=None, metavar="NAME", help="tenant to compile --policy for" ) plan.add_argument("--trace", type=Path, default=None, help="trace JSONL output path") - plan.add_argument("--run-id", default=None, help="name this run") + plan.add_argument( + "--run-id", default=None, help="name this run; refused if --trace already holds it" + ) plan.add_argument( "--max-rounds", type=int, default=None, help="planning rounds the loop may take (default: 8)", @@ -822,7 +826,11 @@ def build_parser() -> argparse.ArgumentParser: help="directory the tools work in (default: a fresh temp dir)", ) agent.add_argument("--trace", type=Path, default=None, help="default: /trace.jsonl") - agent.add_argument("--run-id", default=None, help="name this run (default: agent-)") + agent.add_argument( + "--run-id", + default=None, + help="name this run, refused if --trace already holds it (default: agent-)", + ) agent.add_argument( "--allow", action="append", diff --git a/grapharc/cli/plan.py b/grapharc/cli/plan.py index 10643a3..5ca5ca4 100644 --- a/grapharc/cli/plan.py +++ b/grapharc/cli/plan.py @@ -39,6 +39,7 @@ from grapharc.cli.config import load as load_settings from grapharc.cli.generate import resolve_or_generate_policy from grapharc.cli.output import EXIT_FAILED, EXIT_OK, emit, fail +from grapharc.cli.runid import refuse_reused_run_id DEFAULT_REGISTRY = "grapharc.examples.plan_incident:build_registry" @@ -261,6 +262,15 @@ def plan( trace_path = trace_path or Path(tempfile.mkdtemp(prefix="grapharc-plan-")) / "trace.jsonl" + # Before the setup, because this one is about the file the setup would start + # writing into: a run id already in that file merges this plan with an + # earlier one under a single name. + reused = refuse_reused_run_id( + trace_path, run_id, command="plan", as_json=as_json, goal=goal + ) + if reused is not None: + return reused + # Everything that can be wrong about the setup is decided before a model is # asked anything, so a bad flag cannot half-execute a plan. try: diff --git a/grapharc/cli/runid.py b/grapharc/cli/runid.py new file mode 100644 index 0000000..b05e440 --- /dev/null +++ b/grapharc/cli/runid.py @@ -0,0 +1,88 @@ +"""One guard, shared by every command that writes a trace under a chosen id. + +`TraceRecorder` opens its file in append mode, and that is right: several runs +in one file is the pattern `grapharc diff` is built on. What append mode cannot +notice is that the id being appended under is already in the file. Two runs then +merge under one name, and every reader downstream presents the blend as a single +coherent run — `metrics` sums both runs' tokens and nodes, `viz` welds the second +path onto the end of the first, `replay` reconstructs a chimera. The trace is +documented as the one record everything else agrees with; a reused id is what +makes that one record lie. + +So the file staying appendable is correct and the id being reused is the defect, +which puts the refusal here, at the start of a run, rather than in the recorder. +Only an id the operator actually typed is checked: a generated one is fresh by +construction and must not make every run pay for a file scan. +""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any + +from grapharc.cli.output import fail + + +def count_events(path: Path, run_id: str) -> int: + """How many events in `path` already carry `run_id`. + + Reads the lines itself instead of going through `TraceRecorder.read_events`: + this runs before a guarded command starts, so it reads one field per line + rather than validating a whole event model, and a line it cannot parse is + skipped rather than raised over. The guard's job is to spot a collision, not + to be the file's validator — the readers still report a torn trace, and an + unreadable file is a failure the run's own writer will report in its own + words. + """ + if not path.is_file(): + return 0 + seen = 0 + try: + with path.open(encoding="utf-8", errors="replace") as handle: + for line in handle: + if not line.strip(): + continue + try: + record = json.loads(line) + except ValueError: + continue + if isinstance(record, dict) and record.get("run_id") == run_id: + seen += 1 + except OSError: + return 0 + return seen + + +def refuse_reused_run_id( + trace_path: Path | str | None, + run_id: str | None, + *, + command: str, + as_json: bool, + **extra: Any, +) -> int | None: + """The exit code to return, or None when this run may go ahead. + + `run_id is None` means the caller will generate one, which cannot collide. + """ + if run_id is None or trace_path is None: + return None + path = Path(trace_path) + events = count_events(path, run_id) + if not events: + return None + return fail( + f"run id {run_id!r} already has {events} event{'' if events == 1 else 's'} in " + f"{path}; pick a new --run-id or a new --trace file. Appending a second run " + f"under one id merges the two, and metrics, replay and viz would then report " + f"the blend as one run", + as_json=as_json, + command=command, + run_id=run_id, + trace=str(path), + **extra, + ) + + +__all__ = ["count_events", "refuse_reused_run_id"] diff --git a/grapharc/runtime/graph.py b/grapharc/runtime/graph.py index dd1e984..a7b722f 100644 --- a/grapharc/runtime/graph.py +++ b/grapharc/runtime/graph.py @@ -37,6 +37,7 @@ import asyncio import copy +import functools import inspect import threading import time @@ -44,7 +45,8 @@ from collections.abc import AsyncIterator, Callable, Iterable, Iterator from contextlib import asynccontextmanager from dataclasses import replace -from typing import Any, Literal +from enum import Enum +from typing import Any, Literal, get_args, get_origin, get_type_hints from langchain_core.runnables import RunnableConfig from langgraph.graph import END, START, StateGraph @@ -287,6 +289,41 @@ def _short_repr(value: Any, limit: int = 120) -> str: return text if len(text) <= limit else text[:limit] + "…" +def _declared_return_values(router: Callable[..., Any]) -> tuple[Any, ...] | None: + """The values a router *declares* it returns, or None when it declares none. + + Two annotations say something a mapping can be held against: a `Literal` of + the event names, and an `Enum` whose members are them. `str`, no annotation + at all, a forward reference that will not resolve, a callable object with no + annotations to read — none of those is an opinion about the mapping, so they + are skipped rather than guessed at. A check that invented a requirement + would be worse than the gap it closed. + """ + try: + annotation = get_type_hints(router).get("return") + except Exception: # noqa: BLE001 — an unresolvable annotation is simply no answer + return None + if get_origin(annotation) is Literal: + return get_args(annotation) + if isinstance(annotation, type) and issubclass(annotation, Enum): + return tuple(annotation) + return None + + +def _in_mapping(key: Any, mapping: dict[str, str]) -> bool: + """Whether `mapping[key]` would resolve — the lookup LangGraph itself does. + + Hashability is the question, not equality: an unhashable key raises on the + lookup rather than missing it, and `Enum` hashes by member name while + `StrEnum` compares by value, so a member can equal a key it does not find. + Both are answered the way the branch machinery will answer them. + """ + try: + return key in mapping + except TypeError: + return False + + def _first_error(field: str, exc: ValidationError) -> str: """Pydantic's first complaint, with the path inside the value it happened at.""" first = exc.errors()[0] @@ -368,12 +405,31 @@ def add_conditional_edge( router: Callable[[Any], str], mapping: dict[str, str], ) -> GraphARC: - """Route on a validated event name returned by deterministic `router` code.""" + """Route on a validated event name returned by deterministic `router` code. + + The mapping is topology, and topology is checked where it is declared: + an empty mapping is refused, every target must name a node this graph + has (or `END`), and a router that says what it returns — a `Literal` or + an `Enum` return annotation — has those members held against the + mapping's keys. A router that declares nothing is left alone; guessing + what a function might return is not a check. + + What cannot be settled here is the key an undeclared router actually + returns at run time. LangGraph resolves it with a plain `mapping[key]` + lookup inside its branch machinery, so a typo used to surface as a bare + `KeyError` several frames from the router that caused it. The router is + wrapped so it raises `GraphRoutingError` instead, naming the node, the + key and the keys there are — the same failure the kernel raises for + every other transition it cannot make. + """ if self.dag: raise GraphCycleError( f"graph {self.name!r} is dag=True: conditional edges are not allowed" ) - self._graph.add_conditional_edges(source, router, mapping) + self._check_mapping(source, router, mapping) + self._graph.add_conditional_edges( + source, self._checked_router(source, router, mapping), mapping + ) self._conditional_edges.extend( (source, target) for target in dict.fromkeys(mapping.values()) ) @@ -477,6 +533,93 @@ 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_mapping( + self, source: str, router: Callable[..., Any], mapping: dict[str, str] + ) -> None: + """Everything about a conditional edge that is knowable when it is added.""" + who = f"the conditional edge on node {source!r}" + if not mapping: + raise GraphRoutingError( + f"{who} was given an empty mapping, so there is no key its router " + f"could return that leads anywhere; LangGraph would accept the edge " + f"and fail on the first branch taken. Valid destinations: " + f"{self._destinations()}" + ) + unreachable = [ + (key, target) + for key, target in mapping.items() + if not (target == END or target in self._nodes) + ] + if unreachable: + pairs = ", ".join( + f"{_short_repr(key)} -> {_short_repr(target)}" for key, target in unreachable + ) + raise GraphRoutingError( + f"{who} maps to {'a destination' if len(unreachable) == 1 else 'destinations'} " + f"graph {self.name!r} does not have: {pairs}. LangGraph resolves a branch " + f"target when a run reaches it, so this would surface mid-run rather than " + f"here. Valid destinations: {self._destinations()}" + ) + declared = _declared_return_values(router) or () + unmapped = [value for value in declared if not _in_mapping(value, mapping)] + if unmapped: + raise GraphRoutingError( + f"{who} has a router declaring it returns " + f"{', '.join(_short_repr(value) for value in declared)}, but " + f"{', '.join(_short_repr(value) for value in unmapped)} " + f"{'is' if len(unmapped) == 1 else 'are'} not " + f"{'a key' if len(unmapped) == 1 else 'keys'} of its mapping; a return " + f"the mapping has no entry for is a branch the run cannot take. Keys: " + f"{', '.join(_short_repr(key) for key in mapping)}" + ) + + def _checked_router( + self, source: str, router: Callable[..., Any], mapping: dict[str, str] + ) -> Callable[..., Any]: + """Wrap `router` so an unmapped return is a GraphARC error, not a `KeyError`. + + `functools.wraps` is not decoration here: LangGraph names the branch + after the callable's `__name__` and reads its annotations to infer the + branch's input schema, so an anonymous wrapper would quietly rename the + branch and drop that inference. The signature is `*args`/`**kwargs` for + the same reason — `wraps` makes `inspect.signature` report the router's + own parameters, and LangGraph passes `config` to a router that asks for + one. + """ + who = f"the router on node {source!r}" + + def check(result: Any) -> Any: + # LangGraph accepts one key or a sequence of them, and lets a `Send` + # through to its own dispatch untouched; every other element is + # resolved against the mapping, which is where the KeyError is. + for key in result if isinstance(result, (list, tuple)) else [result]: + if isinstance(key, Send): + continue + if not _in_mapping(key, mapping): + raise GraphRoutingError( + f"{who} returned {_short_repr(key)}, which is not a key of the " + f"mapping the edge was declared with; LangGraph looks a branch " + f"key up in that mapping, so this would surface as a bare " + f"KeyError from inside the branch machinery instead of naming " + f"the router that produced it. Keys: " + f"{', '.join(_short_repr(k) for k in mapping)}" + ) + return result + + if inspect.iscoroutinefunction(router): + + @functools.wraps(router) + async def routed(*args: Any, **kwargs: Any) -> Any: + return check(await router(*args, **kwargs)) + + else: + + @functools.wraps(router) + def routed(*args: Any, **kwargs: Any) -> Any: + return check(router(*args, **kwargs)) + + return routed + def _check_send_payload(self, who: str, send: Send) -> None: """Hold a `Send.arg` to the worker's declared `input_schema`. diff --git a/tests/test_async_kernel.py b/tests/test_async_kernel.py index d04b45a..8d6f211 100644 --- a/tests/test_async_kernel.py +++ b/tests/test_async_kernel.py @@ -168,6 +168,47 @@ async def loop(state: S) -> dict: pass +@pytest.mark.asyncio +async def test_an_async_router_returning_an_unmapped_key_raises_a_grapharc_error(): + """The declaration-time mapping check is shape-blind; this one is not. + + An `async def` router has to be wrapped in an async wrapper or LangGraph + would await the check's return value instead of the router's. + """ + + async def step(state: S) -> dict: + await asyncio.sleep(0) + return {"a": state.a + 1} + + async def route(state: S) -> str: + await asyncio.sleep(0) + return "dnoe" + + g = GraphARC(S, name="loop") + g.add_node("step", step, writes={"a"}) + g.add_edge(START, "step") + g.add_conditional_edge("step", route, {"again": "step", "done": END}) + with pytest.raises(GraphRoutingError, match="'dnoe'"): + await g.compile().ainvoke({"a": 0}) + + +@pytest.mark.asyncio +async def test_an_async_router_that_maps_still_routes(): + async def step(state: S) -> dict: + await asyncio.sleep(0) + return {"a": state.a + 1} + + async def route(state: S) -> str: + await asyncio.sleep(0) + return "done" if state.a >= 2 else "again" + + g = GraphARC(S, name="loop") + g.add_node("step", step, writes={"a"}) + g.add_edge(START, "step") + g.add_conditional_edge("step", route, {"again": "step", "done": END}) + assert (await g.compile().ainvoke({"a": 0}))["a"] == 2 + + @pytest.mark.asyncio async def test_astream_yields_one_update_per_node(): async def one(state: S) -> dict: diff --git a/tests/test_cli.py b/tests/test_cli.py index ae5019c..0ef548d 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -1582,6 +1582,119 @@ def test_run_honours_the_run_id_it_was_given(tmp_path, capsys): assert payload["nodes_executed"] == 3 +# -- a run id names one run --------------------------------------------------- + + +def test_a_reused_run_id_is_refused_before_the_second_run_writes_anything(tmp_path, capsys): + """Two runs under one id merge, and every reader then reports the blend as + one run: doubled tokens from `metrics`, a welded path from `viz`.""" + trace = tmp_path / "t.jsonl" + first, _, _ = call( + ["plan", "goal one", "--trace", str(trace), "--run-id", "r1"], capsys + ) + assert first == 0 + before = TraceRecorder(trace).read_events("r1") + + code, _, err = call(["plan", "goal two", "--trace", str(trace), "--run-id", "r1"], capsys) + + assert code == 2 + assert "r1" in err and str(trace) in err + assert "--run-id" in err, "the message has to say how to proceed" + after = TraceRecorder(trace).read_events("r1") + assert [e.model_dump() for e in after] == [e.model_dump() for e in before] + + +def test_a_reused_run_id_fails_as_one_json_document(tmp_path, capsys): + trace = tmp_path / "t.jsonl" + call(["plan", "goal one", "--trace", str(trace), "--run-id", "r1"], capsys) + + code, payload, err = call_json( + ["plan", "goal two", "--trace", str(trace), "--run-id", "r1"], capsys + ) + + assert code == 2 + assert payload["ok"] is False + assert payload["command"] == "plan" + assert payload["run_id"] == "r1" + assert payload["trace"] == str(trace) + assert err == "" + + +def test_run_refuses_a_reused_run_id_before_the_admission_event(tmp_path, capsys): + """`run` writes its verdict under the id, so `--check-only` collides too.""" + graph = _write_graph(tmp_path, _LEGAL_GRAPH) + trace = tmp_path / "t.jsonl" + call(["run", str(graph), "--trace", str(trace), "--run-id", "chosen"], capsys) + before = len(TraceRecorder(trace).read_events("chosen")) + + code, _, err = call( + ["run", str(graph), "--check-only", "--trace", str(trace), "--run-id", "chosen"], capsys + ) + + assert code == 2 + assert "chosen" in err + assert len(TraceRecorder(trace).read_events("chosen")) == before + + +def test_agent_refuses_a_reused_run_id(tmp_path, capsys): + """Checked before the model is built, so the refusal costs no backend call.""" + trace = tmp_path / "t.jsonl" + TraceRecorder(trace).event( + run_id="a1", graph="cli-agent", node="agent", phase="start", step=1 + ) + + code, _, err = call( + ["agent", "do the thing", "--workspace", str(tmp_path), + "--trace", str(trace), "--run-id", "a1"], + capsys, + ) + + assert code == 2 + assert "a1" in err and "1 event" in err + + +def test_different_run_ids_in_one_trace_stay_supported(tmp_path, capsys): + """`grapharc diff` reads two runs out of one file; that is the pattern the + guard must not touch. The file being appendable is correct — the id being + reused is the defect.""" + trace = tmp_path / "t.jsonl" + + assert call(["plan", "goal one", "--trace", str(trace), "--run-id", "r1"], capsys)[0] == 0 + assert call(["plan", "goal two", "--trace", str(trace), "--run-id", "r2"], capsys)[0] == 0 + + assert TraceRecorder(trace).run_ids() == ["r1", "r2"] + + +def test_a_generated_run_id_is_never_guarded(tmp_path, capsys): + """Fresh by construction, so it must not pay for a scan of the file either.""" + trace = tmp_path / "t.jsonl" + + assert call(["plan", "goal one", "--trace", str(trace)], capsys)[0] == 0 + assert call(["plan", "goal two", "--trace", str(trace)], capsys)[0] == 0 + + assert len(TraceRecorder(trace).run_ids()) == 2 + + +def test_the_guard_counts_a_line_it_can_read_and_skips_the_rest(tmp_path): + """A torn line is the readers' business to report. The guard looks for one + id, so it skips what it cannot parse rather than raising over it — and it + reports nothing at all for a file that is not there.""" + from grapharc.cli.runid import count_events + + trace = tmp_path / "t.jsonl" + assert count_events(trace, "r1") == 0 + assert count_events(tmp_path, "r1") == 0, "a directory is not a trace" + + TraceRecorder(trace).event(run_id="r1", graph="g", node="n", phase="start", step=1) + with trace.open("a", encoding="utf-8") as handle: + handle.write("not json at all\n\n") + TraceRecorder(trace).event(run_id="r2", graph="g", node="n", phase="start", step=1) + + assert count_events(trace, "r1") == 1 + assert count_events(trace, "r2") == 1 + assert count_events(trace, "r3") == 0 + + def test_check_only_refuses_a_topology_that_passes_the_gate_but_cannot_be_built( tmp_path, capsys ): diff --git a/tests/test_runtime_discipline.py b/tests/test_runtime_discipline.py index 00a31e6..49d924d 100644 --- a/tests/test_runtime_discipline.py +++ b/tests/test_runtime_discipline.py @@ -1,7 +1,8 @@ """Unit tests for the runtime discipline layer: write permissions, DAG mode, budgets, traces.""" import operator -from typing import Annotated +from enum import StrEnum +from typing import Annotated, Literal import pytest from pydantic import BaseModel, Field, ValidationError @@ -12,6 +13,7 @@ START, GraphARC, GraphCycleError, + GraphRoutingError, MissingRunContextError, StateTypeError, WritePermissionError, @@ -69,6 +71,127 @@ def test_dag_mode_rejects_conditional_edges(): g.add_conditional_edge("x", lambda s: "x", {"x": "x"}) +# -- conditional edges are checked where they are declared --------------------- + + +def _spinner(name: str = "t") -> GraphARC: + g = GraphARC(S, name=name) + g.add_node("spin", lambda s: {"a": s.a + 1}, writes={"a"}) + g.add_edge(START, "spin") + return g + + +def test_a_mapping_target_nobody_added_raises_when_the_edge_is_added(): + """The whole point: this used to be a KeyError three nodes into a run.""" + g = _spinner() + with pytest.raises(GraphRoutingError) as exc: + g.add_conditional_edge("spin", lambda s: "go", {"go": "sipn"}) + message = str(exc.value) + assert "'spin'" in message, "the error has to name the edge's source" + assert "'sipn'" in message, "and the target it cannot reach" + assert "'go' -> 'sipn'" in message + + +def test_every_unreachable_target_is_named_at_once(): + g = _spinner() + with pytest.raises(GraphRoutingError, match="destinations") as exc: + g.add_conditional_edge("spin", lambda s: "go", {"go": "nope", "stop": "also_nope"}) + assert "'go' -> 'nope'" in str(exc.value) + assert "'stop' -> 'also_nope'" in str(exc.value) + + +def test_an_empty_mapping_is_refused(): + g = _spinner() + with pytest.raises(GraphRoutingError, match="empty mapping"): + g.add_conditional_edge("spin", lambda s: "go", {}) + + +def test_a_correct_mapping_is_unaffected(): + g = _spinner() + g.add_conditional_edge( + "spin", lambda s: "stop" if s.a >= 3 else "again", {"again": "spin", "stop": END} + ) + assert g.compile().invoke({"a": 0})["a"] == 3 + + +def test_a_router_declaring_a_literal_has_its_members_checked(): + def route(s) -> Literal["again", "stop"]: + return "again" + + g = _spinner() + with pytest.raises(GraphRoutingError, match="declaring it returns") as exc: + g.add_conditional_edge("spin", route, {"again": "spin"}) + assert "'stop' is not a key" in str(exc.value) + + +def test_a_router_declaring_an_enum_has_its_members_checked(): + class Stop(StrEnum): + again = "again" + stop = "stop" + + def route(s) -> Stop: + return Stop.again + + g = _spinner() + with pytest.raises(GraphRoutingError, match="declaring it returns"): + g.add_conditional_edge("spin", route, {Stop.again: "spin"}) + + ok = _spinner() + ok.add_conditional_edge("spin", route, {Stop.again: "spin", Stop.stop: END}) + + +def test_a_router_declaring_a_literal_that_matches_is_accepted(): + def route(s) -> Literal["again", "stop"]: + return "stop" if s.a >= 2 else "again" + + g = _spinner() + g.add_conditional_edge("spin", route, {"again": "spin", "stop": END}) + assert g.compile().invoke({"a": 0})["a"] == 2 + + +def test_a_router_that_declares_nothing_is_left_alone(): + """A `str` return says nothing about the mapping; guessing is not checking.""" + + def route(s) -> str: + return "stop" if s.a >= 1 else "again" + + g = _spinner() + g.add_conditional_edge("spin", route, {"again": "spin", "stop": END}) + assert g.compile().invoke({"a": 0})["a"] == 1 + + +def test_an_unmapped_router_return_is_a_grapharc_error_not_a_keyerror(): + """What cannot be settled at declaration time still must not be a bare KeyError.""" + g = _spinner() + g.add_conditional_edge("spin", lambda s: "stpo", {"again": "spin", "stop": END}) + with pytest.raises(GraphRoutingError) as exc: + g.compile().invoke({"a": 0}) + message = str(exc.value) + assert "'stpo'" in message + assert "'spin'" in message, "the router that produced it has to be named" + assert "'again', 'stop'" in message, "and the keys there were" + + +def test_an_unmapped_key_inside_a_returned_list_is_caught_too(): + g = _spinner() + g.add_node("other", lambda s: {"b": 1}, writes={"b"}) + g.add_edge("other", END) + g.add_conditional_edge("spin", lambda s: ["stop", "elsewhere"], {"stop": "other"}) + with pytest.raises(GraphRoutingError, match="elsewhere"): + g.compile().invoke({"a": 0}) + + +def test_the_wrapped_router_keeps_the_name_langgraph_branches_by(): + """`functools.wraps` is load-bearing: LangGraph names the branch after it.""" + + def pick_a_branch(s) -> str: + return "stop" + + g = _spinner() + g.add_conditional_edge("spin", pick_a_branch, {"stop": END}) + assert "pick_a_branch" in g._graph.branches["spin"] + + def test_budget_hard_ceiling_cannot_be_looped_past(): g = GraphARC(S, name="t", budget=Budget(max_iterations=5)) g.add_node("loop", lambda s: {"a": s.a + 1}, writes={"a"})