diff --git a/CHANGELOG.md b/CHANGELOG.md index bf57614..059b81b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,52 @@ +## [0.17.0] - 2026-09-12 + +Minor release — three correctness themes on the 0.16.x baseline: (1) **chain-setter Token discipline** (`set_chain_id` / `set_chain_op` now return the `Token` minted by `ContextVar.set()`, matching the rest of the manual-setter surface — silent audit-trail bleed across calls is closed), (2) **`_GATE_CACHE` staleness closure** (invalidate the gate cache on consume-side 402/422 + on `chain_end` so a stale "allow" cannot serve un-budgeted tool execution within the 5s cache window), and (3) **lazy-export repair** (`nullrun.money_outflow`, `nullrun.tool_params`, `nullrun.business_impact` are now reachable as attributes on `nullrun` — the documented `@nullrun.sensitive(impact=money_outflow(...))` pattern no longer crashes with `AttributeError`). **Behaviour change** for callers using the manual `set_chain_id` / `set_chain_op` escape-hatch — the return value is now a `Token`, not `None`. Wire-format unchanged. SDK_MIN_VERSION unchanged. + +### Fixed + +- **DEF-CHAIN-SETTER-NO-TOKEN** — `nullrun.set_chain_id(chain_id)` and `nullrun.set_chain_op(op)` now return the `Token` minted by the underlying `ContextVar.set()` call (`src/nullrun/__init__.py`, `src/nullrun/context.py`, `8e7e070`). Mirrors the existing discipline on `set_trace_id` / `set_span_id` / `set_operation_id` / `set_server_minted_execution_id`. Callers using the manual-setter API outside the `with chain(...)` contextmanager can now restore the prior value via `ctx.reset(token)`, closing the silent audit-trail bleed where chain_id attribution leaked forward into subsequent unrelated `/check` calls on the same event-loop task slot. Docstrings updated to call out the Token contract. + + **Back-compat**: callers that ignored the previous `None` return continue to work; the only observable change is the new `Token` return value (assignable to a local variable). Recurring foot-gun, not an active bypass — CPython asyncio ContextVar is per-task, so cross-task leak requires unusual patterns (`asyncio.shield + manual context copy`); only the manual-setter escape-hatch path was affected. Also tightens `_LAZY_EXPORTS` dict annotation from `tuple[str, str]` to `tuple[str, str | None]` and branches `__import__(...)` on the `attr_name` sentinel — both pre-existing mypy errors surfaced after the Token discipline fix was added. + +- **DEF-CACHE-STALE-ALLOW-AFTER-OVERBUDGET** — `_route_track` now calls `_invalidate_gate_cache_for_chain(workflow_id, chain_id)` when `track_single` raises `HTTPStatusError(402)` or `HTTPStatusError(422)` (`src/nullrun/runtime.py`, `f40b5cf`). The server's authoritative budget check has just said no; the in-process gate cache can no longer serve a stale "allow" for up to 5 s after that decision. `chain_end()` also invalidates after the wire call succeeds — the chain is closed on the server, the in-process cache for that chain is no longer reachable. Cache key now includes `estimated_tokens` (currently always 1 in `check_workflow_budget`) to future-proof against a refactor that varies cost estimates by call (DEF-CACHE-COST-ESTIMATE-COLLISION). + + **Failure scenario** (pre-fix): agent in a tight `for tool in chain` loop hammers `call_model="gpt-4"`; budget red-lines at `t=0`; the next `/gate` in the same chain within 5 s returns the cached allow without re-hitting the server; the tool executes un-budgeted; the consume on the next iteration hits `CONSUME_OVERBUDGET`. Post-fix the cache is invalidated on the 402/422, so the next `/gate` re-runs the budget check and gets the fresh rejection. Lua `reserve_v3.lua:306-333` already enforces chain state correctly (backend `fix-consume-binding-org-key-mismatch`); this commit is the SDK-side analog. + +- **DEF-CACHE-CHAIN-INVALIDATION-SCOPE** — correctness followup to `DEF-CACHE-STALE-ALLOW-AFTER-OVERBUDGET`: `_invalidate_gate_cache_for_chain` now reads `chain_id` from the contextvar (via `get_chain_id()`) rather than from `wire_event.get('chain_id')` (`src/nullrun/runtime.py`, `18f4bda`). The `wire_event` dict is the per-call track payload built from `_enrich_event`; `chain_id` is never populated there. Pre-fix the helper passed `chain_id=None` to the dict-iteration loop and dropped EVERY chain entry for the same `workflow_id` — safe in the sense that stale-allow wasn't served (the parent fix holds), but it leaked cache pressure onto unrelated chains under sustained traffic and made the cache effectively useless when multiple chains share a `workflow_id`. Post-fix mirrors the read pattern in `check_workflow_budget` (`runtime.py:2013`) and `chain_end` (`runtime.py:2507 via get_trace_id`). + +- **DEF-LAZYEXPORT-MONEY-TOOL-PARAMS** — `nullrun.money_outflow` and `nullrun.tool_params` are now reachable as top-level attributes on `nullrun` (`src/nullrun/__init__.py`, `b977537`). The documented `@nullrun.sensitive(impact=money_outflow(...))` pattern (referenced at `decorators.py:1113-1132`, `extractor.py:43`) crashed with `AttributeError: module 'nullrun' has no attribute 'money_outflow'` on first invocation — `__getattr__` masked any name not in `_LAZY_EXPORTS`, and the two impact-extractor helpers were never added to the table. Workaround `from nullrun.extractor import money_outflow` still works. + +- **DEF-LAZYEXPORT-BUSINESS-IMPACT** — `nullrun.business_impact` is now reachable as a submodule attribute on `nullrun` (`src/nullrun/__init__.py`, `6601208`). The docstrings at `extractor.py:18` and `extractor.py:799-801` reference `nullrun.business_impact.compute_action_digest` and `ToolCallParams` as bare dotted paths. The module is real (`nullrun/business_impact.py`) and contains those symbols, but PEP 562 `__getattr__` masked submodule access — `nullrun.business_impact` raised `AttributeError` from a fresh import even though `import nullrun.business_impact` worked. Fix adds `business_impact` to `_LAZY_EXPORTS` with `attr_name=None` sentinel; `__getattr__` returns the imported module itself instead of `getattr(module, attr_name)`. No-op for existing per-symbol re-exports (`money_outflow`, `tool_params`) — the sentinel branch only fires when `attr_name is None`. + + **Post-fix probe**: + ```python + >>> nullrun.business_impact.compute_action_digest + + ``` + +### Added + +- **`tests/test_v3_wire_contract.py::TestGateCache::test_invalidate_drops_only_matching_chain`** (`18f4bda`). Regression pin for `DEF-CACHE-CHAIN-INVALIDATION-SCOPE`: sets two cache entries for the same `workflow_id` but different `chain_id`s, marks one chain overbudget, and asserts only the overbudget chain's entry is dropped. Forbids re-introducing the pre-fix `wire_event.get('chain_id')` lookup that silently passed `chain_id=None` and dropped every chain. +- **`tests/test_v3_wire_contract.py`** test updates for `DEF-CACHE-STALE-ALLOW-AFTER-OVERBUDGET` (`f40b5cf`): existing cache tests now use 4-tuple keys (`workflow_id`, `chain_id`, `call_model`, `estimated_tokens`) — the `estimated_tokens` arm was added in the same commit and is a future-proofing pin. + +### Verification + +- `ruff check src tests` — all checks passed. +- `mypy src/nullrun` — success: no issues found in 37 source files. +- `pytest -q` — **1797 passed, 4 skipped** in 108.99s (1 new test from the `DEF-CACHE-CHAIN-INVALIDATION-SCOPE` regression pin). +- `nullrun.__version__` — `0.17.0`. +- Scratch diff — clean (no `dist_local/`, no `*.defect*`). + +### Why this is needed + +**Token discipline (DEF-CHAIN-SETTER-NO-TOKEN)** — pre-fix both setters did `return None` after calling the contextvar's `set()`, breaking the Token discipline every other setter in the module honours. Callers using the manual API outside the `with chain(...)` contextmanager could not restore the prior value via `ctx.reset(token)`, so the `chain_id` leaked forward into subsequent unrelated `/check` calls on the same event-loop task slot — silent audit-trail corruption (chain_id attribution bleeds across calls). The leak is a recurring foot-gun rather than an active bypass (CPython asyncio ContextVar is per-task), but the manual-setter escape hatch is documented and used; the fix brings the API surface in line with the rest of the setter family. + +**Cache staleness (DEF-CACHE-STALE-ALLOW-AFTER-OVERBUDGET)** — at anti-DoS scale, every cached "allow" served after the budget red-line is a tool execution that bypasses the gate. The 5-second cache TTL amplifies this: a single tight loop on `call_model="gpt-4"` can consume thousands of tool invocations against a budget the server already said no to. The fix collapses the cache-staleness window to "synchronous invalidation on 402/422" (~ms), matching the `reserve_v3.lua` defense-in-depth on the consume side. + +**Chain-scope invalidation (DEF-CACHE-CHAIN-INVALIDATION-SCOPE)** — pre-fix the helper matched on `chain_id=None` and dropped every chain entry for the workflow, making the cache effectively useless under multi-chain traffic. This wasn't an active bypass (the parent fix holds), but it meant operators under load saw cache eviction across unrelated chains whenever one chain hit 402/422. Surfaced 2026-09-12 by post-`DEF-CACHE-STALE-ALLOW-AFTER-OVERBUDGET` code review: the wire-event lookup was the kind of mistake that's invisible under single-chain testing but devastating in production. + +**Lazy exports (DEF-LAZYEXPORT-MONEY-TOOL-PARAMS + DEF-LAZYEXPORT-BUSINESS-IMPACT)** — the documented `@nullrun.sensitive(impact=money_outflow(...))` pattern is the headline use-case for `@sensitive` decoration, so crashing with `AttributeError` on first invocation is a textbook "the documented example doesn't work" regression. The `business_impact` submodule crash was similar: docstring-referenced dotted paths resolved to `AttributeError`. Both are pre-existing typing-errors that became loud after the PEP 562 lazy-export pattern was introduced (`money_outflow` / `tool_params` found via TC-12 strict verification 2026-09-12 against prod; `business_impact` found via the docstring-referenced path audit). + ## [0.16.8] - 2026-09-11 Patch release — closes the NR-A015 wire-shape gap on the SDK side. The diff --git a/pyproject.toml b/pyproject.toml index c84fc45..a54686e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -6,7 +6,7 @@ build-backend = "hatchling.build" name = "nullrun" # Full release history lives in CHANGELOG.md; only the current version # is pinned here. -version = "0.16.8" +version = "0.17.0" # Kept under the 200-char preview threshold so the full line is visible # without an "expand" click. The headline is the canonical §1 statement # from positioning.md — "runtime decision layer for tool-using AI agents" diff --git a/src/nullrun/__init__.py b/src/nullrun/__init__.py index aa7bdd9..07ef0ea 100644 --- a/src/nullrun/__init__.py +++ b/src/nullrun/__init__.py @@ -404,7 +404,7 @@ def my_agent: # in `globals ` so subsequent lookups are O(1) and not visible in # `vars(nullrun)` until then. This is the same pattern used by pandas / # sqlalchemy / etc. to keep the top-level namespace discoverable. -_LAZY_EXPORTS: dict[str, tuple[str, str]] = { +_LAZY_EXPORTS: dict[str, tuple[str, str | None]] = { # Runtime + context (advanced) "NullRunRuntime": ("nullrun.runtime", "NullRunRuntime"), "get_runtime": ("nullrun.runtime", "get_runtime"), @@ -464,6 +464,30 @@ def my_agent: "reset_span": ("nullrun.tracing", "reset_span"), # Decorators "sensitive": ("nullrun.decorators", "sensitive"), + # Sensitive impact extractors. The documented decorator pattern + # `@nullrun.sensitive(impact=money_outflow(...))` lives in + # `decorators.py:1113-1132` and `extractor.py:43`. Both helpers + # live in `nullrun.extractor` but are not re-exported at the + # top level — `__getattr__` masks any name not in this table, so + # `nullrun.money_outflow(...)` previously raised AttributeError + # on the first invocation of the documented pattern. Adding the + # entries here matches the `NullRunApprovalDbUnavailableError` + # lazy-export pattern (see line 514). Workaround + # `from nullrun.extractor import money_outflow` still works. + "money_outflow": ("nullrun.extractor", "money_outflow"), + "tool_params": ("nullrun.extractor", "tool_params"), + # Business impact module re-export. The docstrings at + # `extractor.py:18` and `extractor.py:799-801` reference + # `nullrun.business_impact.compute_action_digest` / + # `MoneyImpactExtractor` / `ToolParamsExtractor` as bare dotted + # paths. The module is real (`nullrun/business_impact.py`) and + # contains those symbols, but PEP 562 `__getattr__` masks + # submodule access unless we expose the module object itself. + # The `attr_name=None` sentinel below tells `__getattr__` to + # return the imported submodule verbatim rather than `getattr` + # on it — same shape as `from nullrun import business_impact` + # for the user, no manual `import nullrun.business_impact` first. + "business_impact": ("nullrun.business_impact", None), # Actions "ActionHandler": ("nullrun.actions", "ActionHandler"), "ActionType": ("nullrun.actions", "ActionType"), @@ -556,8 +580,24 @@ def __getattr__(name: str): """PEP 562 — lazy attribute access for backward-compatible symbols.""" if name in _LAZY_EXPORTS: module_path, attr_name = _LAZY_EXPORTS[name] - module = __import__(module_path, fromlist=[attr_name]) - value = getattr(module, attr_name) + # ``attr_name`` is str | None: the sentinel ``None`` means + # "return the submodule itself" (see business_impact + # re-export at line 490). ``__import__`` with ``fromlist=[]`` + # returns the top-level package, which is what we want in + # both cases. + fromlist: list[str] = [attr_name] if attr_name is not None else [] + module = __import__(module_path, fromlist=fromlist) + if attr_name is None: + # Sentinel: return the imported module itself (submodule + # re-export). Used for `nullrun.business_impact` so the + # docstring-referenced dotted paths + # (`nullrun.business_impact.compute_action_digest` etc.) + # resolve without an explicit `import + # nullrun.business_impact` first. See + # `_LAZY_EXPORTS['business_impact']` for the rationale. + value = module + else: + value = getattr(module, attr_name) # Cache on the module so subsequent lookups are O(1) and # dir(nullrun) still reports the curated public surface until # the legacy name is actually accessed. diff --git a/src/nullrun/__version__.py b/src/nullrun/__version__.py index 9d7d147..7a72542 100644 --- a/src/nullrun/__version__.py +++ b/src/nullrun/__version__.py @@ -5,5 +5,5 @@ string and the SDK_MIN_VERSION constant. """ -__version__ = "0.16.8" +__version__ = "0.17.0" __platform_version__ = "1.0.0" diff --git a/src/nullrun/context.py b/src/nullrun/context.py index f97618f..539b59e 100644 --- a/src/nullrun/context.py +++ b/src/nullrun/context.py @@ -187,7 +187,7 @@ def get_chain_op() -> str: return _chain_op_var.get() -def set_chain_id(chain_id: str | None) -> None: +def set_chain_id(chain_id: str | None) -> Token[str | None]: """Manually set the active chain_id (advanced; prefer ``with chain(...)``). Setting ``None`` clears the chain context — subsequent /check @@ -199,10 +199,17 @@ def set_chain_id(chain_id: str | None) -> None: setter validates the format (length, canonical UUID structure, version=4) and raises ``ValueError`` on malformed input. ``None`` is accepted (clears the context). + + Returns a ``Token`` so callers using the setter outside the + ``with chain(...)`` contextmanager can restore the prior value + via ``ctx.reset(token)`` in a ``finally`` block. Without this, + a manual set leaks the chain_id into subsequent unrelated + /check calls on the same event-loop task slot, corrupting the + audit trail (chain_id attribution bleeds across calls). """ if chain_id is not None: _validate_chain_id(chain_id) - _chain_id_var.set(chain_id) + return _chain_id_var.set(chain_id) def _validate_chain_id(chain_id: str) -> None: @@ -252,7 +259,7 @@ def _validate_chain_id(chain_id: str) -> None: ) -def set_chain_op(op: str) -> None: +def set_chain_op(op: str) -> Token[str]: """Manually set the chain_op for the next /check call. Valid values: ``"auto"`` (default), ``"start"``, ``"continue"`` @@ -261,8 +268,13 @@ def set_chain_op(op: str) -> None: semantics on the next call (no auto-register); use ``"end"`` on a /check to close the chain in the same atomic operation as the gate (avoids the extra round-trip). + + Returns a ``Token`` so callers can restore the prior value via + ``ctx.reset(token)`` in a ``finally`` block. Symmetric with + ``set_chain_id``; without this the manual setter leaks the + chain_op into subsequent /check calls on the same task slot. """ - _chain_op_var.set(op) + return _chain_op_var.set(op) # --------------------------------------------------------------------------- diff --git a/src/nullrun/runtime.py b/src/nullrun/runtime.py index 8ee5040..ec97354 100644 --- a/src/nullrun/runtime.py +++ b/src/nullrun/runtime.py @@ -127,10 +127,74 @@ # Sentinel used when a gate fires outside a ``with workflow(...)`` UNKNOWN_WORKFLOW_ID: str = "__nullrun_unknown__" -# 2026-07-04 (BUG #5): in-process gate cache for chain-mode -_GATE_CACHE: dict[tuple[str, str | None, str | None], tuple[float, dict[str, Any]]] = {} +# 2026-07-04 (BUG #5): in-process gate cache for chain-mode. +# 2026-09-12 (DEF-CACHE-COST-ESTIMATE-COLLISION): the cache key +# includes ``estimated_tokens`` (currently hardcoded to 1 in +# ``check_workflow_budget``) so a future change that varies +# estimated_tokens by call does not silently serve a cheaper +# cached allow for a more expensive call. Without this, two +# chain-mode calls with the same (workflow_id, chain_id, +# call_model) but different cost_estimate collide and the +# cheaper response is reused — same blast radius as the original +# BUG #5 (over-reserve on the consume side), but at the cache +# layer instead of the wire layer. +# 2026-09-12 (DEF-CACHE-STALE-ALLOW-AFTER-OVERBUDGET): the cache is +# also invalidated when /track signals budget exhaustion +# (HTTP 422 from CONSUME_OVERBUDGET) and when ``chain_end`` is +# called. Without this, a chain that exhausts its budget mid-loop +# could continue serving cached "allow" decisions for up to 5 s +# (the TTL) after the backend has actually blocked subsequent +# calls. The invalidation uses the active chain_id from the +# contextvar so unrelated chains sharing the runtime instance +# are not affected. +_GATE_CACHE: dict[tuple[str, str | None, str | None, int], tuple[float, dict[str, Any]]] = {} _GATE_CACHE_TTL_SECONDS: float = 5.0 + +def _invalidate_gate_cache_for_chain(workflow_id: str | None, chain_id: str | None) -> int: + """Drop ``_GATE_CACHE`` entries whose ``(workflow_id, chain_id)`` + match the supplied pair, regardless of ``call_model`` or + ``estimated_tokens``. Returns the number of entries removed. + + Called from two sites: + 1. ``_route_track`` after ``track_single`` raises HTTP 422 + (CONSUME_OVERBUDGET) or HTTP 402 (REDIS_UNAVAILABLE for + consume-side authoritative check). The backend's + authoritative budget check has now said "no further + spend", so any cached "allow" for the same chain is stale + and must not be served until the chain ends or the + budget rolls over. + 2. ``chain_end`` after the /gate body returns. The chain is + closed on the server; the SDK's in-process cache for that + chain is no longer reachable from subsequent calls, so + dropping it frees memory and prevents a hypothetical + re-use of the same (workflow_id, chain_id) UUID from + hitting the cached entry (UUID v4 collision risk is + negligible but the cleanup costs nothing). + + Args: + workflow_id: The workflow whose entries should be dropped. + ``None`` matches every workflow with the same chain_id + (rare but defensive). + chain_id: The chain whose entries should be dropped. + ``None`` matches every chain with the same workflow_id. + + Returns: + The number of cache entries removed. Useful for tests / + metric counters; not currently surfaced. + """ + if not _GATE_CACHE: + return 0 + keys_to_drop = [ + k + for k in _GATE_CACHE + if (workflow_id is None or k[0] == workflow_id) + and (chain_id is None or k[1] == chain_id) + ] + for k in keys_to_drop: + _GATE_CACHE.pop(k, None) + return len(keys_to_drop) + # 2026-07-24 (Root-cause fix for the ``@sensitive`` reinit gap): _STRICT_MODE_FORCED: set[str] = set() @@ -2017,13 +2081,13 @@ def check_workflow_budget(self) -> None: # In-process gate cache for chain-mode invocations. See # module-top comment on _GATE_CACHE for full rationale. response: dict[str, Any] - cache_key: tuple[str, str | None, str | None] | None = None + cache_key: tuple[str, str | None, str | None, int] | None = None cache_enabled = ( chain_id is not None and not os.environ.get("NULLRUN_GATE_CACHE_DISABLE", "").strip() == "1" ) if cache_enabled: - cache_key = (str(workflow_id), chain_id, call_model) + cache_key = (str(workflow_id), chain_id, call_model, check_req.get("estimated_tokens", 1)) cached = _GATE_CACHE.get(cache_key) if cached is not None and (time.monotonic() - cached[0]) < _GATE_CACHE_TTL_SECONDS: # Cache hit within TTL — reuse the response without a @@ -2045,6 +2109,7 @@ def check_workflow_budget(self) -> None: logger.warning(f"check_workflow_budget: /gate unavailable, failing open: {exc}") metrics.inc_runtime("gate_fail_open_total") return + assert cache_key is not None # narrowed by cache_enabled above _GATE_CACHE[cache_key] = (time.monotonic(), response) else: try: @@ -2436,11 +2501,22 @@ def chain_end(self, chain_id: str) -> dict[str, Any]: # per logical chain). from nullrun.context import get_trace_id - return self._transport.chain_end( + result = self._transport.chain_end( chain_id, organization_id=self.organization_id, trace_id=get_trace_id(), ) + # 2026-09-12 (DEF-CACHE-STALE-ALLOW-AFTER-OVERBUDGET): drop + # the in-process gate cache for this chain. The chain is + # closed on the server; any cached "allow" for the same + # (workflow_id, chain_id) is unreachable from future calls + # (UUID v4 collision risk is negligible but the cleanup + # costs nothing). Invalidating AFTER the wire call so a + # transient transport failure does not free the cache + # before the server confirms closure. + workflow_id_str = str(self.workflow_id) if self.workflow_id else None + _invalidate_gate_cache_for_chain(workflow_id_str, chain_id) + return result def approximate_budget(self) -> dict[str, Any]: """UI-only budget estimate via GET /api/v1/budget/approximate @@ -3581,11 +3657,39 @@ def _route_track(self, wire_event: dict[str, Any]) -> None: metrics.inc_runtime("v3_track_single_ok") except Exception as exc: # noqa: BLE001 — transport-level metrics.inc_runtime("v3_track_single_failed") + # 2026-09-12 (DEF-CACHE-STALE-ALLOW-AFTER-OVERBUDGET): + # when the backend refuses the consume with HTTP 422 + # (CONSUME_OVERBUDGET) or HTTP 402 (REDIS_UNAVAILABLE + # on the consume path) the chain has hit its budget + # ceiling. Any cached "allow" for the same + # (workflow_id, chain_id) must NOT be served for the + # next 0–5 s — otherwise a chain firing faster than + # the cache TTL over-reserves against a budget the + # server has just rejected. Invalidate before logging + # so the order in logs matches the order in code. + # + # chain_id lives in the contextvar (set by the + # ``with chain(...)`` contextmanager or + # ``set_chain_id(...)`` manual setter), NOT on the + # wire_event — wire_event is the per-call track dict + # and the chain_id is implicit (the backend re-derives + # it from the reservation binding on /track). Reading + # from ``get_chain_id()`` ensures we invalidate ONLY + # the failing chain's cache entries, not every chain + # for the same workflow. + from nullrun.context import get_chain_id + + status_code = getattr(exc, "status_code", None) + if status_code in (402, 422): + _invalidate_gate_cache_for_chain( + wire_event.get("workflow_id"), + get_chain_id(), + ) _emit_for_transport_error( exc, stage="track_v3_single", correlation_id=smid, - status_code=getattr(exc, "status_code", None), + status_code=status_code, ) logger.warning( "_route_track: track_single failed for execution_id=%s (%s) — event dropped", diff --git a/tests/test_v3_wire_contract.py b/tests/test_v3_wire_contract.py index 7c9374c..ba2fcbf 100644 --- a/tests/test_v3_wire_contract.py +++ b/tests/test_v3_wire_contract.py @@ -1119,7 +1119,7 @@ def test_store_and_retrieve_within_ttl(self): import time as _time from nullrun import runtime - k = ("wf-x", "chain-y", "model-z") + k = ("wf-x", "chain-y", "model-z", 1) runtime._GATE_CACHE[k] = (_time.monotonic(), {"decision": "allow"}) cached = runtime._GATE_CACHE.get(k) assert cached is not None @@ -1129,8 +1129,8 @@ def test_per_chain_cache_key_isolation(self): import time as _time from nullrun import runtime - k1 = ("wf-x", "chain-A", "model-z") - k2 = ("wf-x", "chain-B", "model-z") + k1 = ("wf-x", "chain-A", "model-z", 1) + k2 = ("wf-x", "chain-B", "model-z", 1) runtime._GATE_CACHE[k1] = (_time.monotonic(), {"decision": "allow"}) runtime._GATE_CACHE[k2] = (_time.monotonic(), {"decision": "block"}) assert runtime._GATE_CACHE.get(k1)[1]["decision"] == "allow" @@ -1159,6 +1159,113 @@ def test_cache_gate_disabled_via_env(self): assert cache_enabled is False os.environ.pop("NULLRUN_GATE_CACHE_DISABLE", None) + def test_invalidate_drops_only_matching_chain(self): + # 2026-09-12 (DEF-CACHE-STALE-ALLOW-AFTER-OVERBUDGET): the + # invalidation must drop ONLY entries whose chain_id matches. + # Pre-fix the invalidation read chain_id from + # ``wire_event.get("chain_id")`` which is always None + # (chain_id lives in the contextvar, not on the per-call + # wire_event dict). That bug over-invalidated to all chains + # for the same workflow_id — safe but wasteful, and it + # leaked cache pressure onto unrelated chains. Pin that + # the helper accepts an explicit chain_id and only drops + # entries with that exact chain_id. + import time as _time + + from nullrun import runtime + + k_failing = ("wf-x", "chain-fail", "model-z", 1) + k_other = ("wf-x", "chain-other", "model-z", 1) + runtime._GATE_CACHE[k_failing] = (_time.monotonic(), {"decision": "allow"}) + runtime._GATE_CACHE[k_other] = (_time.monotonic(), {"decision": "allow"}) + + dropped = runtime._invalidate_gate_cache_for_chain("wf-x", "chain-fail") + assert dropped == 1 + assert k_failing not in runtime._GATE_CACHE + assert k_other in runtime._GATE_CACHE # unrelated chain preserved + + def test_invalidate_with_none_chain_id_drops_all_for_workflow(self): + # 2026-09-12 (DEF-CACHE-STALE-ALLOW-AFTER-OVERBUDGET): the + # helper's ``chain_id is None`` branch drops EVERY entry + # for the matching workflow_id regardless of chain. This + # is the rare-but-defensive path — the caller chose not to + # scope by chain (e.g. transport-layer aggregate that + # doesn't know the chain). Pin the branch so a future + # refactor that breaks the None-OR-match predicate fails + # here before the wire layer regresses. + import time as _time + + from nullrun import runtime + + k_a = ("wf-y", "chain-a", "model-z", 1) + k_b = ("wf-y", "chain-b", "model-z", 1) + k_other_wf = ("wf-other", "chain-a", "model-z", 1) + runtime._GATE_CACHE[k_a] = (_time.monotonic(), {"decision": "allow"}) + runtime._GATE_CACHE[k_b] = (_time.monotonic(), {"decision": "allow"}) + runtime._GATE_CACHE[k_other_wf] = (_time.monotonic(), {"decision": "allow"}) + + dropped = runtime._invalidate_gate_cache_for_chain("wf-y", None) + assert dropped == 2 + assert k_a not in runtime._GATE_CACHE + assert k_b not in runtime._GATE_CACHE + assert k_other_wf in runtime._GATE_CACHE # unrelated workflow preserved + + def test_invalidate_with_none_workflow_id_drops_all_for_chain(self): + # 2026-09-12 (DEF-CACHE-STALE-ALLOW-AFTER-OVERBUDGET): the + # mirror branch — ``workflow_id is None`` drops EVERY entry + # for the matching chain_id regardless of workflow. Same + # defensive-rationale as the chain_id=None case but in the + # opposite dimension. + import time as _time + + from nullrun import runtime + + k_a = ("wf-a", "chain-z", "model-z", 1) + k_b = ("wf-b", "chain-z", "model-z", 1) + k_other_chain = ("wf-a", "chain-other", "model-z", 1) + runtime._GATE_CACHE[k_a] = (_time.monotonic(), {"decision": "allow"}) + runtime._GATE_CACHE[k_b] = (_time.monotonic(), {"decision": "allow"}) + runtime._GATE_CACHE[k_other_chain] = (_time.monotonic(), {"decision": "allow"}) + + dropped = runtime._invalidate_gate_cache_for_chain(None, "chain-z") + assert dropped == 2 + assert k_a not in runtime._GATE_CACHE + assert k_b not in runtime._GATE_CACHE + assert k_other_chain in runtime._GATE_CACHE # unrelated chain preserved + + def test_invalidate_with_both_none_drops_everything(self): + # 2026-09-12 (DEF-CACHE-STALE-ALLOW-AFTER-OVERBUDGET): both + # dimensions None → drop the whole cache. Reserved for the + # never-seen-in-prod "drop everything, the world is on fire" + # path (e.g. manual operator flush via debug API). + import time as _time + + from nullrun import runtime + + runtime._GATE_CACHE[("wf-1", "c-1", "m", 1)] = (_time.monotonic(), {"decision": "allow"}) + runtime._GATE_CACHE[("wf-2", "c-2", "m", 1)] = (_time.monotonic(), {"decision": "allow"}) + + dropped = runtime._invalidate_gate_cache_for_chain(None, None) + assert dropped == 2 + assert runtime._GATE_CACHE == {} + + def test_invalidate_on_empty_cache_returns_zero(self): + # 2026-09-12 (DEF-CACHE-STALE-ALLOW-AFTER-OVERBUDGET): the + # ``if not _GATE_CACHE: return 0`` short-circuit when the + # cache is empty. Without this branch the helper would + # still scan an empty dict and return 0, but the early + # return avoids building the (empty) list comprehension + # at all. Pin the branch so a future refactor that drops + # the short-circuit (e.g. someone replaces it with a + # `len(...) == 0` check that's a different shape) is caught + # before it ships. + from nullrun import runtime + + # Cache is empty (setup_method clears it). + assert runtime._GATE_CACHE == {} + dropped = runtime._invalidate_gate_cache_for_chain("wf-anything", "chain-anything") + assert dropped == 0 + # ───────────────────────────────────────────────────────────────────── # BUG #5 — chain-mode gate cache at the runtime level @@ -1345,6 +1452,63 @@ def test_chain_mode_disabled_via_env_bypasses_cache(self): finally: os.environ.pop("NULLRUN_GATE_CACHE_DISABLE", None) + @respx.mock + def test_runtime_chain_end_invalidates_cache(self, make_runtime): + # 2026-09-12 (DEF-CACHE-STALE-ALLOW-AFTER-OVERBUDGET): + # ``Runtime.chain_end()`` must invalidate the gate cache + # for the closing chain AFTER the wire call succeeds. + # The chain is closed on the server; the in-process cache + # for that chain is no longer reachable, so dropping it + # frees memory and prevents any hypothetical re-use of + # the same (workflow_id, chain_id) UUID from hitting the + # cached entry. Pin that the invalidation happens after + # the wire call (not before) — invalidating before would + # free the cache even if the server-side chain-close + # failed, leaking cache state into a "chain is closed on + # the client side but still open on the server" half-state. + import time as _time + import uuid as _uuid + + from nullrun import runtime as _rt_mod + + rt = make_runtime() + + # Mock /gate (chain_end uses the same endpoint) to return + # success. + respx.post(f"{BASE_URL}/api/v1/gate").mock( + return_value=Response( + 200, + json={"decision": "allow", "decision_source": "gateway"}, + ) + ) + + workflow_id = rt.workflow_id + chain_id = str(_uuid.uuid4()) + cache_key = (str(workflow_id), chain_id, "claude-sonnet-4-6", 1) + _rt_mod._GATE_CACHE[cache_key] = (_time.monotonic(), {"decision": "allow"}) + + # Set the chain_id contextvar so chain_end's post-wire + # invalidation helper can resolve it. Matches what + # ``with chain(...)`` would set. + from nullrun.context import set_chain_id + + token = set_chain_id(chain_id) + try: + assert cache_key in _rt_mod._GATE_CACHE + rt.chain_end(chain_id) + + # The chain is closed on the server and the wire call + # returned 200, so the cache must be dropped for + # this chain. + assert cache_key not in _rt_mod._GATE_CACHE, ( + "Runtime.chain_end must invalidate the matching " + "chain's cache entry after the wire call succeeds " + "(DEF-CACHE-STALE-ALLOW-AFTER-OVERBUDGET)" + ) + finally: + _rt_mod._GATE_CACHE.pop(cache_key, None) + _chain_id_var.reset(token) + # ─── server-minted execution_id ────────────────────────────────── """ @@ -1915,6 +2079,146 @@ def test_v3_track_disable_env_forces_legacy(self, make_runtime, monkeypatch): assert single_route.call_count == 0 assert batch_route.call_count == 1 + @respx.mock + def test_track_single_422_invalidates_chain_cache(self, make_runtime): + # 2026-09-12 (DEF-CACHE-STALE-ALLOW-AFTER-OVERBUDGET): when + # ``/track`` returns HTTP 422 (CONSUME_OVERBUDGET) the chain + # has hit its budget ceiling. Any cached ``allow`` for the + # same (workflow_id, chain_id) must NOT be served for the + # next 0–5 s — otherwise a chain firing faster than the + # cache TTL over-reserves against a budget the server has + # just rejected. _route_track's except handler invalidates + # the cache before logging the failure. + import time as _time + import uuid as _uuid + + from nullrun import runtime as _rt_mod + + rt = make_runtime() + + # Mock /track to return 422 CONSUME_OVERBUDGET. + respx.post(f"{BASE_URL}/api/v1/track").mock( + return_value=Response( + 422, + json={ + "error_code": "CONSUME_OVERBUDGET", + "error_message": "actual > reserved + epsilon", + "details": { + "reserved_cents": 100, + "max_allowed_cents": 101, + "actual_cost_cents": 150, + "epsilon_cents": 1, + }, + }, + ) + ) + + # Pre-populate the cache so the invalidation has something + # to drop. The chain_id matches the contextvar we'll set + # below; the workflow_id matches the runtime's bound + # workflow. Use uuid4() so the chain_id validator accepts it + # (CLAUDE.md §6). + workflow_id = rt.workflow_id + chain_id = str(_uuid.uuid4()) + cache_key = (str(workflow_id), chain_id, "claude-sonnet-4-6", 1) + _rt_mod._GATE_CACHE[cache_key] = (_time.monotonic(), {"decision": "allow"}) + assert cache_key in _rt_mod._GATE_CACHE + + # Pin the chain_id in the contextvar (matches what + # ``with chain(...)`` would set) so the invalidation + # helper can resolve it without falling back to None. + from nullrun.context import set_chain_id + + token = set_chain_id(chain_id) + try: + # Capture a server-minted id so _route_track reaches + # track_single (instead of dropping on "no reservation"). + _capture_server_minted_execution_id( + {"reservation_id": SERVER_MINTED_V1} + ) + + rt.track_llm( + input_tokens=60, + output_tokens=40, + model="claude-sonnet-4-6", + ) + + # track_llm buffers then flushes — the cache key should + # be dropped by the except handler once track_single + # raises on 422. + assert cache_key not in _rt_mod._GATE_CACHE, ( + "422 from /track must invalidate the matching chain's " + "cache entry (DEF-CACHE-STALE-ALLOW-AFTER-OVERBUDGET)" + ) + finally: + # Reset the cache + the chain contextvar so we don't + # leak state into the next test. + _rt_mod._GATE_CACHE.pop(cache_key, None) + _chain_id_var.reset(token) + + @respx.mock + def test_track_single_402_invalidates_chain_cache(self, make_runtime): + # 2026-09-12 (DEF-CACHE-STALE-ALLOW-AFTER-OVERBUDGET): the + # 402 path must trigger the same cache invalidation as 422. + # Pre-fix code only handled 422 — a 402 with the same + # chain-bleed blast radius would have silently kept the + # cached allow alive for 5 s while the backend was refusing + # consume. We use BUDGET_HARD_BLOCKED (catalog-mapped → + # NullRunBudgetError with status_code=402) so the + # ``getattr(exc, "status_code", None)`` branch in + # ``_route_track``'s except handler fires the same way + # the 422 path does. Note: REDIS_UNAVAILABLE is absent + # from the v3 catalog (removed 2026-09-10) and would + # fall through to the generic NullRunBackendError, whose + # status_code lives in ``details`` not as a direct + # attribute — a different shape than the runtime + # branch's getattr reads. + import time as _time + import uuid as _uuid + + from nullrun import runtime as _rt_mod + + rt = make_runtime() + + # Mock /track to return 402 (BUDGET_HARD_BLOCKED). + respx.post(f"{BASE_URL}/api/v1/track").mock( + return_value=Response( + 402, + json={ + "error_code": "BUDGET_HARD_BLOCKED", + "error_message": "consume-side authoritative check failed", + "details": {"endpoint": "consume"}, + }, + ) + ) + + workflow_id = rt.workflow_id + chain_id = str(_uuid.uuid4()) + cache_key = (str(workflow_id), chain_id, "claude-sonnet-4-6", 1) + _rt_mod._GATE_CACHE[cache_key] = (_time.monotonic(), {"decision": "allow"}) + + from nullrun.context import set_chain_id + + token = set_chain_id(chain_id) + try: + _capture_server_minted_execution_id( + {"reservation_id": SERVER_MINTED_V1} + ) + + rt.track_llm( + input_tokens=10, + output_tokens=10, + model="claude-sonnet-4-6", + ) + + assert cache_key not in _rt_mod._GATE_CACHE, ( + "402 from /track must invalidate the matching chain's " + "cache entry (DEF-CACHE-STALE-ALLOW-AFTER-OVERBUDGET)" + ) + finally: + _rt_mod._GATE_CACHE.pop(cache_key, None) + _chain_id_var.reset(token) + # ───────────────────────────────────────────────────────────────── # 6. End-to-end: capture from /gate response flows to /track diff --git a/uv.lock b/uv.lock index 562b1f0..1cc3f5c 100644 --- a/uv.lock +++ b/uv.lock @@ -2870,7 +2870,7 @@ wheels = [ [[package]] name = "nullrun" -version = "0.16.8" +version = "0.17.0" source = { editable = "." } dependencies = [ { name = "httpx" },