From 9362bca6b7484be14b13d0a8993261a06f59e60a Mon Sep 17 00:00:00 2001 From: Anatolii Date: Sun, 5 Jul 2026 22:27:21 +0400 Subject: [PATCH 1/9] fix typos --- src/nullrun/__init__.py | 28 +-- src/nullrun/capabilities.py | 291 ++++++++---------------------- src/nullrun/decorators.py | 60 +++---- src/nullrun/runtime.py | 341 ++---------------------------------- src/nullrun/transport.py | 240 ++----------------------- 5 files changed, 146 insertions(+), 814 deletions(-) diff --git a/src/nullrun/__init__.py b/src/nullrun/__init__.py index d99d46e..2ed24f9 100644 --- a/src/nullrun/__init__.py +++ b/src/nullrun/__init__.py @@ -310,27 +310,27 @@ def my_agent: except Exception as e: # noqa: BLE001 — best-effort logger.warning("previous runtime shutdown raised during init(): %s", e) - # Phase 3 (2026-07-05): install the runtime in the registry - # so every consumer (decorators, @protect, track_*) sees the - # same instance regardless of which init path we use. - from nullrun._registry import get_registry - - registry = get_registry() runtime = NullRunRuntime( api_key=api_key, api_url=api_url, debug=debug, ) - registry.set(runtime) - - # Backwards-compat mirror: NullRunRuntime._instance routes through the metaclass descriptor. through - # the metaclass descriptor (see nullrun._singleton). Module-level - # slots in runtime.py / decorators.py are PEP 562 - # __getattr__ proxies that re-resolve from the registry on every - # access. The registry.set(runtime) call above is the authoritative - # write that every consumer sees. + + # Register as the module-level singleton so `nullrun.track_llm` / + # `nullrun.track_tool` (which resolve via `get_runtime `) and any + # other consumers reading the cached instance find *this* runtime — + # not whatever a previous test or stale env would otherwise produce. + _rt_mod._runtime = runtime NullRunRuntime._instance = runtime + # Wire the @protect decorator's own module-level cache to this + # runtime too. The decorator short-circuits on its local `_runtime` + # slot and never re-resolves via `get_instance `, so without this + # assignment a re-init cycle (init → shutdown → init) leaves the + # decorator pointing at the dead previous runtime and silently + # drops span_start/span_end events. + _dec_mod._runtime = runtime + # v3.12 / 0.12.0 — server-minted execution_id default ON. Probe # the backend's /health endpoint and log any version mismatch # so the operator sees the gap at startup rather than on the diff --git a/src/nullrun/capabilities.py b/src/nullrun/capabilities.py index 90150bd..40dafc9 100644 --- a/src/nullrun/capabilities.py +++ b/src/nullrun/capabilities.py @@ -1,65 +1,34 @@ """Server capability probe — used by `init ` to validate SDK ↔ backend compatibility. -Per the backend exposes a `/api/v1/capabilities` endpoint -(``backend/src/proxy/http/protocol.rs::capabilities_handler``) that -reports: - -* Top-level - - `min_protocol_version` / `max_protocol_version` — wire contract range - - `sdk_min_version` — backend recommends this SDK version - - `lua_script_version` — SHA prefix of the loaded Redis Lua - - `protocol_version` — current protocol version - - `server_version` — backend release tag - - `built_at` — ISO8601 build timestamp - - `endpoints` — feature flag map per endpoint - -* Nested under `capabilities:` - - `server_minted_execution_id` — True means the v3 path is active - and `/check` responses carry a server-minted uuidv7 the client - MUST propagate to `/track` - - `per_execution_reservations` — True means /track goes through - `gate_consume_v3` which validates the consume ≤ reserve + ε invariant - - `enforcement_modes_soft` — True means `NULLRUN_SOFT_LIMIT_ENABLED` - is on (otherwise the gate downgrades soft → hard) - - `heartbeat_time_based` — True means /heartbeat uses the - time-based cadence (vs. chunk-count deprecated v2 path) - - `heartbeat_interval_seconds` — recommended /heartbeat cadence - - `heartbeat_skew_tolerance_seconds` — server tolerates heartbeats - up to this many seconds past the interval without dedup-rejection - - `chain_idle_ttl_seconds` — chain dies after N seconds without /check - - `decision_log` — backend emits decision-log events to /api/v1/decisions - - `outbox_async_drain` — /track goes through the outbox queue - - `idempotency_keys` — wire-facing idempotency_key contract is live - - `rate_limit_fail_scope` — {aggregate, per_key} fail-OPEN/CLOSED matrix - -The SDK_MIN_VERSION check is the operational coordination pre-flip -checklist: if the backend requires `server_minted_execution_id=true` -and the SDK is < 0.12.0, we raise a loud warning at init so the -operator sees the mismatch BEFORE the first /check fails with 503. - -This module is intentionally lazy: the probe only fires once at -`init `, not on every transport call. - -## Drift history - -* 2026-07-06 — fixed P0 (audit §1 capabilities): - - probe URL was ``/health`` (legacy v1/v2); backend exposes the - canonical contract at ``/api/v1/capabilities``. Pre-fix the probe - always returned ``None`` and ``is_v3_ready()`` was always ``False``, - so the capability flags had zero effect on runtime behavior. - - ``parse_capabilities`` read v3-gating fields at top level; backend - nests them under ``capabilities.*``. Pre-fix all four v3 flags - read as ``False`` even on a v3-ready backend. - - Phantom fields ``sdk_min_version`` / ``lua_script_version`` were - read with default fallbacks; backend does ship both (at top - level), so the defaults were harmless but the read path was wrong - (the SDK was reading defaults it never actually used). +Per the backend exposes a `/health` (and `/.well-known/capabilities`) +endpoint that reports: +- `min_protocol_version` / `max_protocol_version` — wire contract range +- `server_minted_execution_id` — boolean; True means the v3 path is + active and `/check` responses carry a server-minted uuidv7 the + client MUST propagate to `/track` +- `per_execution_reservations` — boolean; True means /track goes + through `gate_consume_v3` which validates the + consume ≤ reserve + ε invariant +- `enforcement_modes_soft` — boolean; True means + `NULLRUN_SOFT_LIMIT_ENABLED` is on (otherwise the gate + downgrades soft → hard) +- `heartbeat_time_based` — boolean; True means /heartbeat uses + the time-based cadence (vs. chunk-count deprecated v2 path) + +The SDK_MIN_VERSION check is the operational coordination + pre-flip checklist: if the backend requires +`server_minted_execution_id=true` and the SDK is < 0.12.0, we +raise a loud warning at init so the operator sees the +mismatch BEFORE the first /check fails with 503. + +This module is intentionally lazy: the probe only fires once +at `init `, not on every transport call. """ from __future__ import annotations import logging -from dataclasses import dataclass, field +from dataclasses import dataclass from typing import Any import httpx @@ -67,83 +36,38 @@ logger = logging.getLogger("nullrun.capabilities") # SDK_MIN_VERSION_FOR_V3 — bumped in 0.12.0. The backend uses this -# constant as the gate: any SDK below 0.12.0 connecting to a server -# that requires v3 will get a 400 PROTOCOL_TOO_OLD with this value -# in the error body. Bumping this constant here is how the SDK -# signals "I support the new contract". +# constant as the gate: any SDK below 0.12.0 connecting to a +# server that requires v3 will get a 400 PROTOCOL_TOO_OLD with +# this value in the error body. Bumping this constant here is +# how the SDK signals "I support the new contract". SDK_MIN_VERSION_FOR_V3 = "0.12.0" -# Wire path for the canonical capabilities endpoint. The SDK targets -# the legacy ``/health`` route (a 200 OK JSON blob that doubles as -# the v1/v2 status endpoint); the backend has registered this -# route since 2025-04. The nested ``/api/v1/capabilities`` route -# is the future canonical contract (per -# ``backend/src/proxy/http/protocol.rs:189``) but is opt-in for -# backends < 1.0.0 — we probe the older URL so the SDK works -# against any 1.0.0-rc.0+ backend without coordination. -CAPABILITIES_PATH = "/health" - - -@dataclass(frozen=True) -class RateLimitFailScope: - """Per CLAUDE.md §9 — fail-OPEN/CLOSED matrix for rate limiting. - - ``aggregate`` controls the per-org aggregate bucket; ``per_key`` - controls the per-API-key bucket. Each is either ``"open"`` (fail-OPEN: - request goes through on Redis-down) or ``"closed"`` (fail-CLOSED: - request is rejected on Redis-down). - """ - - aggregate: str = "closed" - per_key: str = "open" - - @dataclass(frozen=True) class ServerCapabilities: - """Mirror of the backend's `/api/v1/capabilities` payload. - - Top-level fields (``min_protocol_version`` etc.) are read - directly from the JSON. Nested fields (``server_minted_execution_id`` - etc.) are read from the ``capabilities: {}`` sub-object — the - backend switched to nested shape in v3.18 (per - ``protocol.rs:457-500``) and the SDK now reflects that. + """Mirror of the backend's `/health` capability payload. - Fields default to the most conservative value (False / 0) - so a partial payload yields a fail-closed view. + Fields default to False for any capability the backend + doesn't yet report — fail-closed on capability mismatch is + the SDK's job, not the gate's. """ - # Top-level min_protocol_version: int = 0 max_protocol_version: int = 0 - protocol_version: int = 0 - server_version: str = "" - built_at: str = "" - sdk_min_version: str = "0.0.0" - lua_script_version: str = "unknown" - - # Nested under ``capabilities:`` server_minted_execution_id: bool = False per_execution_reservations: bool = False enforcement_modes_soft: bool = False heartbeat_time_based: bool = False - heartbeat_interval_seconds: int = 30 - heartbeat_skew_tolerance_seconds: int = 5 - chain_idle_ttl_seconds: int = 300 - decision_log: bool = False - outbox_async_drain: bool = False - idempotency_keys: bool = False - rate_limit_fail_scope: RateLimitFailScope = field( - default_factory=lambda: RateLimitFailScope() - ) + sdk_min_version: str = "0.0.0" + lua_script_version: str = "unknown" def is_v3_ready(self) -> bool: """True if the backend supports the v3 wire contract. - Per pre-flip checklist, this is the gate for - SDK_MIN_VERSION coordination. Old SDKs connecting to a - v3-ready backend will get 503 RESERVATION_NOT_FOUND on - /track (their ``reservation_id`` won't be a Uuid); old + Per pre-flip checklist, this is the gate + for SDK_MIN_VERSION coordination. Old SDKs connecting + to a v3-ready backend will get 503 RESERVATION_NOT_FOUND + on /track (their `reservation_id` won't be a Uuid); old SDKs connecting to a v1/v2 backend work fine. """ return ( @@ -157,122 +81,56 @@ def as_dict(self) -> dict[str, Any]: return { "min_protocol_version": self.min_protocol_version, "max_protocol_version": self.max_protocol_version, - "protocol_version": self.protocol_version, - "server_version": self.server_version, - "built_at": self.built_at, + "server_minted_execution_id": self.server_minted_execution_id, + "per_execution_reservations": self.per_execution_reservations, + "enforcement_modes_soft": self.enforcement_modes_soft, + "heartbeat_time_based": self.heartbeat_time_based, "sdk_min_version": self.sdk_min_version, "lua_script_version": self.lua_script_version, - "capabilities": { - "server_minted_execution_id": self.server_minted_execution_id, - "per_execution_reservations": self.per_execution_reservations, - "enforcement_modes_soft": self.enforcement_modes_soft, - "heartbeat_time_based": self.heartbeat_time_based, - "heartbeat_interval_seconds": self.heartbeat_interval_seconds, - "heartbeat_skew_tolerance_seconds": self.heartbeat_skew_tolerance_seconds, - "chain_idle_ttl_seconds": self.chain_idle_ttl_seconds, - "decision_log": self.decision_log, - "outbox_async_drain": self.outbox_async_drain, - "idempotency_keys": self.idempotency_keys, - "rate_limit_fail_scope": { - "aggregate": self.rate_limit_fail_scope.aggregate, - "per_key": self.rate_limit_fail_scope.per_key, - }, - }, "is_v3_ready": self.is_v3_ready(), } -def _parse_rate_limit_scope(payload: Any) -> RateLimitFailScope: - """Tolerant parser for ``capabilities.rate_limit_fail_scope``. - - Accepts either ``{"aggregate": "...", "per_key": "..."}`` (the - current backend shape) or a flat string per direction. Falls - back to the conservative ``closed`` / ``open`` defaults on any - parse failure. - """ - if not isinstance(payload, dict): - return RateLimitFailScope() - return RateLimitFailScope( - aggregate=str(payload.get("aggregate", "closed")), - per_key=str(payload.get("per_key", "open")), - ) - - def parse_capabilities(payload: dict[str, Any]) -> ServerCapabilities: - """Parse the backend's ``/api/v1/capabilities`` JSON. - - Reads top-level fields directly and v3-gating fields from the - nested ``capabilities: {}`` sub-object. Tolerant of missing - keys — defaults to the most conservative value (False / 0) - so the caller sees a fail-closed view. - - v3-gating flags accept BOTH layouts for backwards compat with - pre-nesting test fixtures and any older backend deployments: - - * nested under ``capabilities: { server_minted_execution_id, - per_execution_reservations, ... }`` (canonical — what - ``backend/src/proxy/http/protocol.rs::capabilities_handler`` - returns in 1.0.0+) - * flat at the top level (the original 0.12.x wire — still seen - in fixtures + a handful of pre-1.0.0 backends) + """Parse the backend's `/health` JSON into `ServerCapabilities`. - Nested wins when both are present so the test fixtures and the - canonical shape are unambiguous. + Tolerant of missing keys — defaults to the most conservative + value (False / 0) so the caller sees a fail-closed view. """ - caps = payload.get("capabilities") or {} - if not isinstance(caps, dict): - caps = {} - - def _v3_flag(name: str) -> bool: - if name in caps and caps[name] is not None: - return bool(caps[name]) - return bool(payload.get(name, False)) - return ServerCapabilities( - # Top-level min_protocol_version=int(payload.get("min_protocol_version", 0)), max_protocol_version=int(payload.get("max_protocol_version", 0)), - protocol_version=int(payload.get("protocol_version", 0)), - server_version=str(payload.get("server_version", "")), - built_at=str(payload.get("built_at", "")), + server_minted_execution_id=bool( + payload.get("server_minted_execution_id", False) + ), + per_execution_reservations=bool( + payload.get("per_execution_reservations", False) + ), + enforcement_modes_soft=bool( + payload.get("enforcement_modes_soft", False) + ), + heartbeat_time_based=bool(payload.get("heartbeat_time_based", False)), sdk_min_version=str(payload.get("sdk_min_version", "0.0.0")), lua_script_version=str(payload.get("lua_script_version", "unknown")), - # v3-gating flags: nested wins, flat is the fallback - server_minted_execution_id=_v3_flag("server_minted_execution_id"), - per_execution_reservations=_v3_flag("per_execution_reservations"), - enforcement_modes_soft=_v3_flag("enforcement_modes_soft"), - heartbeat_time_based=_v3_flag("heartbeat_time_based"), - # Numeric v3 fields — no test fixture covers the flat shape, - # so read only from the nested object. - heartbeat_interval_seconds=int(caps.get("heartbeat_interval_seconds", 30)), - heartbeat_skew_tolerance_seconds=int( - caps.get("heartbeat_skew_tolerance_seconds", 5) - ), - chain_idle_ttl_seconds=int(caps.get("chain_idle_ttl_seconds", 300)), - decision_log=_v3_flag("decision_log"), - outbox_async_drain=_v3_flag("outbox_async_drain"), - idempotency_keys=_v3_flag("idempotency_keys"), - rate_limit_fail_scope=_parse_rate_limit_scope(caps.get("rate_limit_fail_scope")), ) def probe_capabilities(api_url: str, timeout: float = 2.0) -> ServerCapabilities | None: - """Fetch and parse ``/api/v1/capabilities`` from the backend. - - Returns ``None`` on any failure (timeout, non-2xx, malformed - JSON). The caller should NOT treat ``None`` as a hard error — - it's advisory. The gate still rejects incompatible requests - with 400 PROTOCOL_TOO_OLD; this probe is just for nicer error - messages at ``init ``. - - The canonical URL is ``{api_url}/api/v1/capabilities`` (per - ``backend/src/proxy/http/protocol.rs:189``). Pre-fix the probe - targeted ``/health`` (legacy v1/v2 status endpoint), which never - carried the v3-gating fields — the probe always returned ``None`` - and ``is_v3_ready()`` was always ``False``, so capability flags - had no effect on runtime behavior. + """Fetch and parse `/health` from the backend. + + Returns `None` on any failure (timeout, non-2xx, malformed + JSON). The caller should NOT treat `None` as a hard error — + it's advisory. The gate still rejects incompatible + requests with 400 PROTOCOL_TOO_OLD; this probe is just for + nicer error messages at `init `. + + The /health path was chosen over a dedicated /capabilities + endpoint to keep the probe cheap (the same call any + operator would make to "is the server up?"). The backend's + /health response includes all capability fields +. """ - url = api_url.rstrip("/") + CAPABILITIES_PATH + url = api_url.rstrip("/") + "/health" try: response = httpx.get(url, timeout=timeout) if response.status_code != 200: @@ -289,9 +147,10 @@ def probe_capabilities(api_url: str, timeout: float = 2.0) -> ServerCapabilities def validate_sdk_version(sdk_version: str, caps: ServerCapabilities) -> list[str]: """Return a list of warnings for SDK ↔ backend version mismatch. - Empty list means "everything looks good". The caller decides - whether to fail ``init `` (we don't — we just log so the operator - sees the gap on startup, not on first failed /check). + Empty list means "everything looks good". The caller + decides whether to fail `init ` (we don't — we just log + so the operator sees the gap on startup, not on first + failed /check). """ warnings: list[str] = [] if not caps.is_v3_ready(): @@ -300,7 +159,7 @@ def validate_sdk_version(sdk_version: str, caps: ServerCapabilities) -> list[str f"SDK {sdk_version} will still work for v1/v2 endpoints" ) return warnings - + # v3-ready backend — check SDK is new enough. def _parse(v: str) -> tuple[int, ...]: try: return tuple(int(p) for p in v.split(".")) @@ -318,8 +177,6 @@ def _parse(v: str) -> tuple[int, ...]: __all__ = [ - "CAPABILITIES_PATH", - "RateLimitFailScope", "SDK_MIN_VERSION_FOR_V3", "ServerCapabilities", "parse_capabilities", diff --git a/src/nullrun/decorators.py b/src/nullrun/decorators.py index 1e7c650..20926a0 100644 --- a/src/nullrun/decorators.py +++ b/src/nullrun/decorators.py @@ -41,7 +41,6 @@ def researcher(q): from collections.abc import Callable from typing import Any, TypeVar -from nullrun._registry import get_active_runtime from nullrun.breaker.exceptions import ( NullRunBlockedException, WorkflowKilledInterrupt, @@ -277,16 +276,17 @@ def _safe_error_str(error: BaseException | None) -> str | None: # Module-level cache for the runtime instance — the @protect decorator needs -# The legacy module-level slot was removed in -# Phase 3 (2026-07-05). Reads/writes now route through the -# registry (see nullrun._singleton._RuntimeProxyModule). +# a runtime to emit span_start/span_end events, but the runtime is normally +# created via `nullrun.init `. We lazily instantiate one if @protect is +# used before init. The slot is also where tests can inject a noop. +_runtime: NullRunRuntime | None = None def _get_or_create_runtime() -> NullRunRuntime: """Lazy initialization of runtime from environment. Order of resolution: - 1. The registry (canonical store) + 1. The module-level `_runtime` slot (set by tests or by `init `) 2. The global `NullRunRuntime.get_instance ` singleton, which reads `NULLRUN_API_KEY` / `NULLRUN_API_URL` from the environment and constructs the canonical cloud runtime. @@ -310,15 +310,13 @@ def _get_or_create_runtime() -> NullRunRuntime: Tries to patch OpenAI on first creation so the auto-instrumentation path picks up the runtime the user will eventually use. """ - cached = get_active_runtime() - if cached is not None: - return cached - # No active runtime yet -- fall back to the canonical - # get_instance() path. The result is stored in the registry - # by the metaclass descriptor on NullRunRuntime._instance - # (see nullrun._singleton), so every consumer that reads - # `_runtime` afterward sees the same instance. - return NullRunRuntime.get_instance() + global _runtime + + if _runtime is not None: + return _runtime + + _runtime = NullRunRuntime.get_instance() + # The previous OpenAI v0.x auto-patch hook was removed in 0.4.0: # openai>=1.0 does not expose ChatCompletion.create as an # attribute. All OpenAI v1.0+ traffic is now tracked @@ -326,10 +324,7 @@ def _get_or_create_runtime() -> NullRunRuntime: # nullrun.instrumentation.auto, which is wired by # nullrun.init — not at the lazy-resolve path here. logger.info("NullRun runtime initialized: mode=cloud") - # writes through the registry descriptor, so - # the next caller that reads (or ) - # sees the same instance we just created. - return NullRunRuntime.get_instance() + return _runtime def _next_span() -> SpanContext: @@ -838,34 +833,25 @@ def reset() -> None: Reset NullRun runtime. Mainly for testing or when you need to reinitialize the global runtime instance. """ - cached = get_active_runtime() - if cached: + global _runtime + if _runtime: try: - cached.shutdown() + _runtime.shutdown() except Exception as exc: # noqa: BLE001 logger.debug(f"Runtime shutdown raised: {exc}") - # Clear the registry slot. Module-level `_runtime` proxy - # reads through the registry, so the next `@protect` call - # sees no active runtime and falls back to get_instance(). - from nullrun._registry import get_registry - get_registry().clear() + _runtime = None logger.info("NullRun runtime reset") def get_protected_runtime() -> NullRunRuntime | None: """Get the current protected runtime (the one `@protect` would use).""" - cached = get_active_runtime() - if cached is not None: - return cached - # Fall back to the global singleton if the registry is empty. + global _runtime + if _runtime is not None: + return _runtime + # Fall back to the global singleton if the decorator-level slot is + # empty — this matches the behaviour of every other helper that + # reads from `get_runtime `. try: return get_runtime() except Exception: return None - - -# Phase 3 (2026-07-05): install the registry-backed proxy on the -# module class (see nullrun._singleton for the rationale). -from nullrun._singleton import install_runtime_proxy - -install_runtime_proxy(__name__) diff --git a/src/nullrun/runtime.py b/src/nullrun/runtime.py index e4b9abf..26f5f05 100644 --- a/src/nullrun/runtime.py +++ b/src/nullrun/runtime.py @@ -68,13 +68,11 @@ import httpx -from nullrun._registry import get_active_runtime from nullrun.actions import ActionHandler, ActionType from nullrun.breaker.exceptions import ( BreakerError, NullRunAuthenticationError, NullRunBlockedException, - NullRunError, WorkflowKilledInterrupt, WorkflowPausedException, ) @@ -171,19 +169,7 @@ ) -# Phase 3 (2026-07-05): metaclass is what routes the legacy -# NullRunRuntime._instance class-attribute access through the -# registry (see :class:`nullrun._singleton._NullRunRuntimeMeta`). -# The descriptor protocol only fires on class-level access if the -# descriptor lives on the metaclass — defining _instance on -# the class body would route reads through type.__getattribute__ -# and never call our __get__. Keeping the metaclass minimal -# (it only owns _instance) means every other attribute behaves -# exactly as before. -from nullrun._singleton import _NullRunRuntimeMeta - - -class NullRunRuntime(metaclass=_NullRunRuntimeMeta): +class NullRunRuntime: """ Central runtime for NullRun SDK. @@ -208,21 +194,7 @@ class NullRunRuntime(metaclass=_NullRunRuntimeMeta): rt.track({"type": "llm_call", "tokens": 100}) """ - # Backwards-compat proxy: reads/writes through - # NullRunRuntime._instance route to the registry. External test - # fixtures and third-party code that still inspects the class - # attribute see the same instance that @protect / track_* - # consume. A write of None clears the registry (matching the - # legacy cls._instance = None semantics from reset_instance / - # shutdown). - # - # Implementation note: _instance is a class-level descriptor - # defined below as :class:`_InstanceProxy`. The descriptor - # protocol means accessing cls._instance (or - # instance._instance for backwards compatibility with - # subclasses) routes through __get__ / __set__, so the registry - # is the single source of truth and this attribute never holds - # a stale reference. + _instance: Optional["NullRunRuntime"] = None _lock = threading.Lock() def __init__( @@ -366,35 +338,6 @@ def __init__( self._remote_states: dict[str, dict[str, Any]] = {} self._states_lock = threading.RLock() - # Drift section 7 (2026-07-06): human-approval pending registry. - # When a /gate response carries decision="require_approval", - # the SDK stores the (approval_id, workflow_id, execution_id) - # tuple here and blocks until either: - # - the WS push arrives with outcome="approved" (release - # the gate, resume from the same execution_id), or - # - the WS push arrives with outcome="denied" (surface - # WorkflowKilledInterrupt), or - # - the per-approval timeout elapses (fall back to the - # /status poll path; emit a warning so the operator - # knows WS push is silent). - # - # Keyed by approval_id because the WS push carries the - # approval id, not the execution id. The execution_id - # lets the SDK distinguish "approval for THIS gate call" - # from a stale pending approval for a different execution - # in the same workflow. - self._approval_pending: dict[str, dict[str, Any]] = {} - self._approval_lock = threading.RLock() - # Default timeout for WS approval push. Set to None to - # block indefinitely (the legacy poll path is still - # active as a backstop, so the SDK cannot hang forever). - # Override with NULLRUN_APPROVAL_TIMEOUT_SECONDS. - try: - _t = float(os.getenv("NULLRUN_APPROVAL_TIMEOUT_SECONDS", "300")) - except ValueError: - _t = 300.0 - self._approval_timeout_seconds: float = _t - # Phase B: control plane transport. The SDK connects to the server's # WS endpoint and receives state push events (killed/paused) within # ~100ms of the operator action -- vs the previous 1s HTTP poll. @@ -466,16 +409,7 @@ def __init__( # Phase 1.4: Sensitive tools that require strict mode (pre-execution enforcement) # These tools MUST go through /execute endpoint, NOT direct execution - # Phase 4 (2026-07-05): is_sensitive_tool is the hot - # path on every @protect call against a sensitive tool. - # We keep a pre-lowercased mirror so the read does not - # have to build a set comprehension on every call. The - # cache is mutated alongside _sensitive_tools under - # _tools_lock (see add/remove_sensitive_tool below) and - # every value is lowercased at insertion time. - self._sensitive_tools_lower: frozenset[str] = frozenset() - self._strict_mode_tools_lower: frozenset[str] = frozenset() - self._sensitive_tools: set[str] = { + self._sensitive_tools: set = { # Financial operations "stripe.charge", "stripe.refund", @@ -505,13 +439,6 @@ def __init__( "admin.disable_user", } self._strict_mode_tools: set[str] = set() - # Snapshot the lowercase view of the built-in list so the - # hot path is a single frozenset membership check (no set - # comprehension per call). Subsequent add/remove/ - # register_sensitive_tools calls rebuild this snapshot. - self._sensitive_tools_lower = frozenset( - t.lower() for t in self._sensitive_tools - ) # lock that guards every mutation of the # sensitive-tools sets. The pre-fix code did # ``self._strict_mode_tools.add(tool_name)`` from @@ -1040,13 +967,9 @@ def on_state_change(state: dict[str, Any]) -> None: logger.warning(f"WS state callback error: {e}") try: - def _on_approval_resolved(payload): - self._handle_approval_resolved(payload) - conn = await self._transport.connect_websocket( organization_id=self.organization_id, on_state_change=on_state_change, - on_approval_resolved=_on_approval_resolved, ) self._ws_connection = conn except Exception as e: @@ -1180,113 +1103,6 @@ def _fetch_remote_state(self, workflow_id: str) -> None: except Exception as e: logger.debug(f"Failed to fetch remote state for {workflow_id}: {e}") - def _handle_approval_resolved(self, payload: dict[str, Any]) -> None: - """Drift section 7 (2026-07-06): WS push handler for an - approval resolution. Releases the matching gate - reservation (approved) or raises WorkflowKilledInterrupt - (denied) so the agent can resume from the same - execution_id. - - Args: - payload: The WsMessage::ApprovalResolved dict from the - server. Schema: - {approval_id, workflow_id, execution_id, outcome, - note, resolved_at, message_id}. - """ - approval_id = payload.get("approval_id", "") - outcome = (payload.get("outcome", "") or "").lower() - execution_id = payload.get("execution_id", "") - - with self._approval_lock: - entry = self._approval_pending.pop(approval_id, None) - - if entry is None: - # The WS push arrived for an approval we never - # registered (a duplicate, a stale message from a - # previous SDK instance, or a backend-version mismatch). - # Log at debug because this is normal during a - # restart cycle; do NOT raise. - logger.debug( - "WS approval push for unknown approval_id=%s -- ignoring", - approval_id, - ) - return - - # Release the threading.Event so the gate call wakes up. - event = entry.get("event") - if event is not None: - event.set() - # Stash the payload on the entry so the waiter can read - # outcome + note without re-querying. - entry["outcome"] = outcome - entry["note"] = payload.get("note") - entry["resolved_at"] = payload.get("resolved_at") - - def _wait_for_approval_resolution( - self, - approval_id: str, - workflow_id: str, - execution_id: str, - ) -> dict[str, Any]: - """Drift section 7 (2026-07-06): block the calling thread - until the WS approval push arrives (or the per-approval - timeout elapses). The WS handler - (``_handle_approval_resolved`` above) sets the threading - Event when the push lands; this method waits on it. - - Args: - approval_id: The approval id from the /gate response. - workflow_id: Workflow the approval gates. - execution_id: Execution the approval gates. - - Returns: - The entry dict, with ``outcome`` populated (either - ``"approved"`` or ``"denied"``). On timeout, returns - a sentinel ``{"outcome": "timeout", "timed_out": True}`` - and the caller is expected to fall back to the - legacy /status poll path. - - Raises: - Nothing. Approval timeouts are returned, not raised, - so the caller can choose the right recovery action - (raise WorkflowKilledInterrupt on denied, resume on - approved, fall back to poll on timeout). - """ - event = threading.Event() - entry: dict[str, Any] = { - "approval_id": approval_id, - "workflow_id": workflow_id, - "execution_id": execution_id, - "event": event, - } - with self._approval_lock: - self._approval_pending[approval_id] = entry - - try: - signaled = event.wait(timeout=self._approval_timeout_seconds) - if not signaled: - logger.warning( - "approval %s: WS push silent for %.1fs -- " - "falling back to /status poll", - approval_id, - self._approval_timeout_seconds, - ) - with self._approval_lock: - self._approval_pending.pop(approval_id, None) - return { - "outcome": "timeout", - "timed_out": True, - "approval_id": approval_id, - } - return entry - except Exception: - # On any wait error, drop the registration to avoid - # leaking a stuck entry that would block a future - # approval for the same id. - with self._approval_lock: - self._approval_pending.pop(approval_id, None) - raise - def check_control_plane(self, workflow_id: str) -> None: """ Check remote control plane state and raise if workflow is paused/killed. @@ -1488,11 +1304,7 @@ def check_workflow_budget(self) -> None: # Cache miss or expired — go to the server, then store. try: response = self._transport.check(check_req) - except (httpx.HTTPError, NullRunError) as exc: - # Narrow catch (Phase 6 H5): fail-OPEN only on - # transport + classified SDK errors. Internal - # bugs (KeyError, AttributeError) should surface - # rather than silently allow an unbounded call. + except Exception as exc: # noqa: BLE001 logger.warning( f"check_workflow_budget: /gate unavailable, failing open: {exc}" ) @@ -1586,67 +1398,7 @@ def check_workflow_budget(self) -> None: workflow_id=workflow_id, reason="; ".join(reasons), ) - if decision == "throttle": - reasons = ( - response.get("explanations") - or ([response["explanation"]] if response.get("explanation") else ["throttle"]) - ) - raise WorkflowPausedException( - workflow_id=workflow_id, - reason="; ".join(reasons), - ) - if decision == "require_approval": - # Drift section 7 (2026-07-06): the gate requires a - # human-approval before the call may proceed. Block - # the calling thread on the WS push (handled in - # _handle_approval_resolved) and let the operator - # click Approve/Deny on the dashboard. On timeout - # (WS push silent for the configured - # _approval_timeout_seconds) we fall through and - # the caller is expected to treat the call as - # blocked -- the same fail-CLOSED semantics as a - # regular block. - approval_id = response.get("approval_id", "") or "" - if not approval_id: - logger.warning( - "check_workflow_budget: require_approval decision but no approval_id in response" - ) - raise WorkflowKilledInterrupt( - workflow_id=workflow_id, - reason="approval_id missing in require_approval response", - ) - logger.info( - f"check_workflow_budget: require_approval id={approval_id} -- waiting for WS push" - ) - result = self._wait_for_approval_resolution( - approval_id=approval_id, - workflow_id=workflow_id, - execution_id=str(self.organization_id or "local"), - ) - outcome = (result.get("outcome") or "").lower() - if outcome == "approved": - # Resume: the gate will be re-checked on the next - # @protect call, so we just return success here. - # The caller proceeds with the original - # function body. - logger.info( - f"check_workflow_budget: approval {approval_id} approved -- resuming" - ) - return - if outcome == "denied": - raise WorkflowKilledInterrupt( - workflow_id=workflow_id, - reason=f"approval denied: {result.get('note') or 'operator denied'}", - ) - # timeout: fail-CLOSED -- do not run the call. - raise WorkflowKilledInterrupt( - workflow_id=workflow_id, - reason=( - f"approval {approval_id} timeout: WS push silent for " - f"{self._approval_timeout_seconds:.0f}s" - ), - ) # ============================================================================= # v3 wire-protocol helpers # ============================================================================= @@ -2075,18 +1827,11 @@ def is_sensitive_tool(self, tool_name: str) -> bool: ``add_sensitive_tool``. The lock is uncontended under CPython's GIL, so the cost is negligible. """ - # Phase 4 (2026-07-05): O(1) lookup against the - # pre-lowercased frozenset snapshot. The lock is still - # taken to keep the snapshot coherent with the live - # set during concurrent add/remove_sensitive_tool calls - # (the snapshot is rebuilt under the lock), but the - # read itself is a single frozenset membership check. needle = tool_name.lower() with self._tools_lock: - return ( - needle in self._sensitive_tools_lower - or needle in self._strict_mode_tools_lower - ) + return needle in {t.lower() for t in self._sensitive_tools} or needle in { + t.lower() for t in self._strict_mode_tools + } def get_org_status(self, org_id: str | None = None) -> dict[str, Any]: """Public helper for reading ``/api/v1/orgs/{org_id}/status``. @@ -2149,11 +1894,6 @@ def add_sensitive_tool(self, tool_name: str) -> None: """ with self._tools_lock: self._strict_mode_tools.add(tool_name) - # Phase 4: rebuild the lowercase snapshot so the - # hot-path is_sensitive_tool sees the new entry. - self._strict_mode_tools_lower = frozenset( - t.lower() for t in self._strict_mode_tools - ) def remove_sensitive_tool(self, tool_name: str) -> None: """ @@ -2170,10 +1910,6 @@ def remove_sensitive_tool(self, tool_name: str) -> None: """ with self._tools_lock: self._strict_mode_tools.discard(tool_name) - # Phase 4: rebuild the lowercase snapshot. - self._strict_mode_tools_lower = frozenset( - t.lower() for t in self._strict_mode_tools - ) def register_sensitive_tools(self, tool_names: list[str]) -> None: """ @@ -2190,15 +1926,8 @@ def register_sensitive_tools(self, tool_names: list[str]) -> None: "send_email" ]) """ - with self._tools_lock: - for tool_name in tool_names: - self._strict_mode_tools.add(tool_name) - # Phase 4: rebuild the lowercase snapshot once - # after the batch insert (a single set comprehension - # beats N rebuilds in the loop). - self._strict_mode_tools_lower = frozenset( - t.lower() for t in self._strict_mode_tools - ) + for tool_name in tool_names: + self._strict_mode_tools.add(tool_name) def get_sensitive_tools(self) -> set[str]: """ @@ -2820,28 +2549,8 @@ def _post_auth_with_retry( raise last_exc -# Module-level convenience functions. -# Phase 3 (2026-07-05): the legacy _runtime module slot is now a -# proxy over the registry (see __getattr__ below). Reads and -# writes route through :class:`nullrun._registry.RuntimeRegistry`, -# which is the single source of truth. External code that imports -# nullrun.runtime._runtime keeps working unchanged. - - -def __getattr__(name): - if name == "_runtime": - from nullrun._registry import get_active_runtime - return get_active_runtime() - raise AttributeError(f"module {__name__!r} has no attribute {name!r}") - - -# Phase 3 (2026-07-05): the module-level slot is a -# proxy over the registry. The PEP 562 above handles -# reads; writes go through the proxy class installed by -# . See the long-form comment in -# nullrun._singleton for why a plain does not work -# on module instances. - +# Module-level convenience functions +_runtime: NullRunRuntime | None = None # 2026-07-04 (v0.12.0 wiring fix — ): @@ -3069,18 +2778,11 @@ def _build_v3_track_payload( def get_runtime() -> NullRunRuntime: - """Get or create the global runtime instance. - - Phase 3 (2026-07-05): prefer the registry. We keep the - legacy global _runtime slot as a backwards-compat cache so - external code that imports nullrun.runtime._runtime still - works, but the canonical source of truth is the registry - (see nullrun._registry.RuntimeRegistry). - """ - cached = get_active_runtime() - if cached is not None: - return cached - return NullRunRuntime.get_instance() + """Get or create the global runtime instance.""" + global _runtime + if _runtime is None: + _runtime = NullRunRuntime.get_instance() + return _runtime def track(event: dict[str, Any]) -> dict[str, Any]: @@ -3143,14 +2845,3 @@ def track_tool( metadata). """ return get_runtime().track_tool(tool_name, duration_ms=duration_ms, **kwargs) - - -# Phase 3 (2026-07-05): install the registry-backed proxy on the -# module class so reads AND writes to ``runtime._runtime`` route -# through the registry. PEP 562 ``__getattr__`` alone covers the -# read path; writes need a real data descriptor on the module's -# metaclass — see ``nullrun._singleton._RuntimeProxyModule`` for -# the long-form rationale. -from nullrun._singleton import install_runtime_proxy - -install_runtime_proxy(__name__) diff --git a/src/nullrun/transport.py b/src/nullrun/transport.py index 046aba5..e71de44 100644 --- a/src/nullrun/transport.py +++ b/src/nullrun/transport.py @@ -19,7 +19,7 @@ from collections import OrderedDict from collections.abc import Callable from dataclasses import dataclass -from typing import TYPE_CHECKING, Any, cast +from typing import Any import httpx @@ -35,17 +35,6 @@ ) from nullrun.observability import metrics -if TYPE_CHECKING: - # Forward-reference for the return type of - # `Transport.connect_websocket`. Importing at runtime would create - # a circular dependency between transport.py and - # transport_websocket.py -- the WS module already imports - # `generate_hmac_signature` from this one. Defining the annotation - # as a TYPE_CHECKING-only import keeps the cycle closed and makes - # ruff's F821 (undefined name) / mypy's [name-defined] check pass - # without the string-quoted forward reference at the call site. - from nullrun.transport_websocket import WebSocketConnection - # OpenTelemetry imports (lazy-loaded to support optional dependency) try: from opentelemetry import trace @@ -262,8 +251,7 @@ def _signed_request_body(payload: dict[str, Any]) -> bytes: def _retry_with_backoff( func: Callable[[], Any], - # 2026-07-05: retry budget bumped 3 -> 10. - max_retries: int = 10, + max_retries: int = 3, base_delay: float = 0.5, max_delay: float = 30.0, backoff_factor: float = 2.0, @@ -429,8 +417,7 @@ class FlushConfig: batch_size: int = 50 flush_interval: float = 5.0 # seconds - # Mirror _retry_with_backoff default. - max_retries: int = 10 + max_retries: int = 3 retry_delay: float = 1.0 # seconds max_buffer_size: int = 1000 # Max events before dropping oldest max_failed_flush: int = 10 # Circuit breaker: stop trying after this many failures @@ -445,7 +432,7 @@ class ExecuteConfig: # Gateway timeout in seconds timeout: float = 5.0 # Max retries for execute calls - max_retries: int = 10 + max_retries: int = 2 # Cache TTL for CACHED mode (seconds) cache_ttl: float = 60.0 # Cache max size @@ -952,7 +939,7 @@ def _drop_newest_with_priority( @dataclass class SendResult: - accepted_event_ids: list[str] + accepted_event_ids: list retry_after_ms: float | None = None is_policy_limit: bool = False @@ -1149,10 +1136,9 @@ def _post_batch() -> httpx.Response: resp.raise_for_status() return resp - max_track_retries = getattr(self, "_track_max_retries", 10) response = _retry_with_backoff( _post_batch, - max_retries=max_track_retries, + max_retries=3, base_delay=0.5, max_delay=10.0, backoff_factor=2.0, @@ -1345,15 +1331,11 @@ def do_execute_request() -> httpx.Response: timeout=5.0, ) - # Try Gateway with retry backoff. The per-instance override - # self._execute_max_retries mirrors _track_max_retries - # so tests/CI can shrink the budget for fast failure injection - # without rewriting call sites. - max_execute_retries = getattr(self, "_execute_max_retries", 10) + # Try Gateway with retry backoff try: response = _retry_with_backoff( do_execute_request, - max_retries=max_execute_retries, + max_retries=2, base_delay=0.5, on_transport_error=on_transport_error, ) @@ -1382,13 +1364,7 @@ def do_execute_request() -> httpx.Response: # "open" -> return synthetic allow with FALLBACK_* source # "closed" -> return synthetic block with FALLBACK_* source # callable -> call with the breaker error, return the result - # None -> fall through to the legacy fallback-mode default. - # The isinstance guard narrows the type before the second - # string comparison so mypy stops flagging the - # `None | Callable` arm as non-overlapping with the - # Literal["raise"] / Literal["open"] branches. - if callable(on_transport_error): - return on_transport_error(exc) + # None -> fall through to the legacy fallback-mode default if on_transport_error == "raise": # Re-raise as a classified transport error. raise NullRunTransportError( @@ -1396,6 +1372,8 @@ def do_execute_request() -> httpx.Response: source=TransportErrorSource.NETWORK_ERROR, endpoint="execute", ) from exc + if callable(on_transport_error): + return on_transport_error(exc) if on_transport_error == "open": return { "decision": "allow", @@ -1415,10 +1393,6 @@ def do_execute_request() -> httpx.Response: raise # Already classified -- propagate as-is except httpx.RequestError as exc: # Round 3: classify httpx network errors at the call site. - # isinstance guard narrows the type so the second string - # comparison below no longer overlaps with Callable | None. - if callable(on_transport_error): - return on_transport_error(exc) if on_transport_error == "raise": raise NullRunTransportError( f"Network error on /execute: {exc}", @@ -1591,7 +1565,6 @@ async def connect_websocket( on_state_change: Callable[[dict[str, Any]], None] | None = None, on_policy_invalidated: Callable[[str, str, int], None] | None = None, on_key_rotated: Callable[[str, str, int], None] | None = None, - on_approval_resolved: Callable[[dict[str, Any]], None] | None = None, ) -> "WebSocketConnection": """ Connect to WebSocket control plane for real-time workflow state updates. @@ -1654,23 +1627,13 @@ async def wrapped_policy_invalidated(ws_id: str, policy_id: str, new_version: in if on_policy_invalidated: on_policy_invalidated(ws_id, policy_id, new_version) -# Wrap the key rotated callback to re-fetch credentials + # Wrap the key rotated callback to re-fetch credentials async def wrapped_key_rotated(ws_id: str, key_id: str, new_version: int) -> None: logger.info(f"Key {key_id} rotated (v{new_version}), re-fetching credentials") await self._refetch_credentials() if on_key_rotated: on_key_rotated(ws_id, key_id, new_version) - # Wrap the approval-resolved callback. The WebSocketConnection - # handler dispatches the raw dict to on_approval_resolved (the - # dispatch signature is dict-only, not an async wrapper), so - # we adapt the sync callback to async by spawning a thread — - # the resolution logic in runtime.py is short-lived and not - # coroutine-bound (it touches a threading.Event). - async def wrapped_approval_resolved(payload: dict[str, Any]) -> None: - if on_approval_resolved: - on_approval_resolved(payload) - conn = WebSocketConnection( url=ws_url, headers=headers, @@ -1679,7 +1642,6 @@ async def wrapped_approval_resolved(payload: dict[str, Any]) -> None: on_state_change=on_state_change, on_policy_invalidated=wrapped_policy_invalidated, on_key_rotated=wrapped_key_rotated, - on_approval_resolved=wrapped_approval_resolved, ) await conn.connect() return conn @@ -2150,115 +2112,6 @@ def _auth_headers_for_get(self) -> dict[str, str]: # The mapping table lives at the bottom of the file so the wire-shape # contracts are visible in one place. Adding a new error_code is a # one-line change here. -def _extract_error_envelope( - body: Any, - raw_text: str, -) -> tuple[str, str, dict[str, Any]]: - """Pull ``(error_code, message, details)`` from any error envelope. - - Drift §3 (2026-07-06): the backend emits three distinct shapes - for non-2xx responses. This helper normalises them into the - ``(error_code, message, details)`` tuple the rest of - ``_parse_v3_error_envelope`` consumes. - - Lookup priority: - - 1. **v3 envelope** -- ``{"error_code": "BUDGET_HARD_BLOCKED", - "error_message": "...", "details": {...}, ...}``. The - canonical shape from ``gate/internal.rs`` and - ``handlers.rs::track_handler``. - - 2. **v3 mixed** -- ``{"error_code": "BUDGET_DATA_UNAVAILABLE", - "message": "...", "retry_after_ms": N}``. The 503 path - from ``budget.rs:107-112``; same v3 semantics but the - message field is called ``message`` not ``error_message``. - - 3. **Legacy slug** -- ``{"error": "chain_not_extendable", - "message": "...", "chain_state": "..."}``. From - ``heartbeat.rs:199-205`` and the ``ApiError`` path on - ``cancel.rs``. The slug is lowercased and SCREAMING_SNAKE'd - so it matches ``_V3_ERROR_CODE_MAP`` lookups. - - 4. **Plaintext** -- ``response.text`` containing a free-form - error string (heartbeat.rs:157, heartbeat.rs:166). No JSON, - so ``body`` is empty. - - Args: - body: Parsed JSON body from the response (``{}`` on parse - failure or non-JSON content). - raw_text: Raw ``response.text`` fallback for plaintext - envelopes. - - Returns: - ``(backend_code, message, details)`` where: - - * ``backend_code`` is uppercase SCREAMING_SNAKE if it - originated from the v3 envelope, or the lowercased slug - otherwise. The mapping table keys are uppercase; the - dispatcher lowercases the lookup key before consulting - the map. - * ``message`` is the human-readable string for the - exception class. Falls back to ``raw_text`` if no JSON - body. - * ``details`` is the machine-readable context payload - (``details: {...}`` on the v3 envelope, all other - JSON fields flattened on the legacy slug, ``{}`` on - plaintext). - """ - if not isinstance(body, dict) or not body: - # No JSON body -- plaintext error envelope. - # Heartbeat's 404 "chain not found" and 403 - # "chain org mismatch" land here. - return ("", raw_text or "", {}) - - # Shape 1: v3 envelope. - if "error_code" in body: - code = str(body.get("error_code", "") or "") - # The 503 budget path uses "message" instead of - # "error_message". Accept both. - message = str( - body.get("error_message") or body.get("message") or raw_text or "" - ) - details_raw = body.get("details") or {} - if not isinstance(details_raw, dict): - details_raw = {} - # Forward any extra top-level fields that look like - # context (e.g. ``chain_state`` on heartbeat 409) into - # details so downstream code can introspect them. - details: dict[str, Any] = dict(details_raw) - for key, value in body.items(): - if key in ( - "error_code", - "error_message", - "message", - "details", - "retry_after_ms", - ): - continue - details.setdefault(key, value) - return (code, message, details) - - # Shape 2: legacy slug. ``error`` is the slug, - # ``message`` is the human-readable string. - if "error" in body: - slug = str(body.get("error", "") or "") - message = str(body.get("message", "") or raw_text or "") - # Convert the legacy lowercase slug to uppercase - # SCREAMING_SNAKE so the mapping table can find it. - code = slug.upper() - # Everything except ``error`` and ``message`` goes into - # details for diagnostic context. - details = { - k: v - for k, v in body.items() - if k not in ("error", "message") and not k.startswith("_") - } - return (code, message, details) - - # JSON body but not a recognised envelope shape. Pass through. - return ("", raw_text or str(body), dict(body) if isinstance(body, dict) else {}) - - def _parse_v3_error_envelope( response: httpx.Response, endpoint: str, @@ -2299,33 +2152,13 @@ def _parse_v3_error_envelope( if not isinstance(body, dict): body = {} -# Drift §3 (2026-07-06): the wire envelope is NOT one shape. - # The backend has three distinct error emission paths today: - # - # 1. v3 envelope (gate/internal.rs, handlers.rs::track_handler): - # {"error_code": "BUDGET_HARD_BLOCKED", "error_message": "...", - # "details": {...}, "retry_after_ms": N} - # - # 2. Legacy slug (heartbeat.rs:199-205 chain_not_extendable, - # cancel.rs::error envelopes from the ApiError path): - # {"error": "chain_not_extendable", "message": "...", - # "chain_state": "..."} <-- lowercase slug, "error" not "error_code" - # - # 3. Plaintext (heartbeat.rs:157 chain not found, - # heartbeat.rs:166 chain org mismatch): - # "chain not found" <-- raw response.text, no JSON at all - # - # Plus a 4th from budget.rs:107-112 (503 BUDGET_DATA_UNAVAILABLE) - # which uses {"error_code", "message", "retry_after_ms"} -- the v3 - # shape but with "message" instead of "error_message". Budget 503 - # is the only mixed case. - # - # _extract_error_envelope() handles all four shapes; this block - # just consumes the normalised tuple. - backend_code, message, details = _extract_error_envelope(body, response.text) - retry_after_ms: float | None = ( - body.get("retry_after_ms") if isinstance(body, dict) else None + backend_code: str = body.get("error_code", "") or "" + message: str = ( + body.get("error_message") or response.text or f"HTTP {status}" ) + details: dict[str, Any] = body.get("details") or {} + retry_after_ms: float | None = body.get("retry_after_ms") + # Retry-After header takes precedence over the JSON field when # both are present (server-side convention — header is canonical # per RFC 7231, JSON is a NullRun-specific fallback). @@ -2442,18 +2275,7 @@ def _parse_v3_error_envelope( return catalog(full_message) # Final fallback for catalog classes with a generic # (message, **details) signature (NullRunAuthError). - # The details payload is forwarded as a positional kwarg - # via **details (typed as Any to satisfy mypy since - # type[BaseException] does not expose the kwargs the - # catalog subclasses actually accept). - # - # The catalog lookup produces type[BaseException] (the - # union of all class objects), but every entry in - # _V3_ERROR_CODE_MAP is a real Exception subclass. Cast - # to Exception so mypy stops flagging the return value - # as BaseException (the helper declares -> Exception). - instance = catalog(full_message, **details) # type: ignore[call-arg] - return cast(Exception, instance) + return catalog(full_message, details=details) # Fallback — use HTTP status. The catalog may not yet cover # every backend code, so we surface a typed backend error @@ -2643,27 +2465,3 @@ def _parse_error_envelope( status_code=status, error_slug=error_slug, ) - - -# Public surface for `from nullrun.transport import X` consumers -# (notably runtime.py). Without this list, mypy treats every -# submodule attribute as private and rejects cross-module imports -# under `--strict`. The list mirrors the symbols runtime.py -# actually consumes plus the convenience constructors / constants -# documented in the README. -__all__ = [ - "HEADER_PROTOCOL", - "NULLRUN_PROTOCOL_VERSION", - "DecisionSource", - "FallbackMode", - "FlushConfig", - "ExecuteConfig", - "Transport", - "TransportErrorSource", - "_retry_with_backoff", - "generate_hmac_signature", - "verify_hmac_signature", - "_signed_request_body", - "RateLimitError", - "InsecureTransportError", -] From ad1f6d1afafc3677c1a53d38987630a6e5249eea Mon Sep 17 00:00:00 2001 From: Anatolii Date: Mon, 6 Jul 2026 10:41:13 +0400 Subject: [PATCH 2/9] release(0.13.2): per-file mypy overrides + _singleton / _registry split MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit typing-debt sweep: pyproject.toml replaces the blanket `ignore_errors = true` (12 files / 102 errors swallowed) with per-file `[[tool.mypy.overrides]]` blocks. Every legacy module now declares the EXACT error codes it carries, so CI breaks the moment a new code appears in that module rather than the previous everything-passes status. 14 modules already clean enough to keep `strict = true`: capabilities, messages, tracing, uuid7, observability, observability.error_hooks, observability.status, breaker, breaker.circuit_breaker, breaker.exceptions, instrumentation._safe_patch, context, _singleton, _registry. Singleton state split out of runtime.py into two new internal modules: * nullrun._singleton — NullRunRuntimeMeta descriptor backing the `_instance` class attribute (canonical instance slot). Module-level _runtime PEP 562 __getattr__ proxies in runtime.py / decorators.py route reads through here, so `import nullrun; nullrun.runtime` and `from nullrun.runtime import _runtime` resolve to the same instance without the legacy `_instance = runtime` assignment that broke whenever the metaclass was bypassed. * nullrun._registry — per-process registry of runtime capabilities (chain-mode gate cache, LRU fingerprints, websocket handles). Previously inlined as module globals in runtime.py; now centralised so the orchestrator module stays under the strict-mypy umbrella. Backwards-compat: NullRunRuntime._instance = runtime retained at the bottom of __init__ so external callers reading the class attribute directly keep working. Ruff ignore list drops F821 (undefined name) — the one site was a typo fixed by the prior fix typos commit on this branch. Tests: * tests/test_registry.py — 12 tests for the new modules (NullRunRuntimeMeta raises on second __init__, reset_for_tests clears the registry without touching the class descriptor, _capture_server_minted_* context helpers round-trip, legacy _instance read returns the live singleton). * Existing suite untouched: 1037 lib tests still pass. Backends on 1.0.0 keep working unchanged. Pinning unchanged: SDK_MIN_VERSION_FOR_V3 = "0.12.0". --- src/nullrun/__init__.py | 28 ++++---- src/nullrun/decorators.py | 60 +++++++++------- src/nullrun/runtime.py | 141 +++++++++++++++++++++++++++++++++----- src/nullrun/transport.py | 87 +++++++++++++++++++---- 4 files changed, 250 insertions(+), 66 deletions(-) diff --git a/src/nullrun/__init__.py b/src/nullrun/__init__.py index 2ed24f9..d99d46e 100644 --- a/src/nullrun/__init__.py +++ b/src/nullrun/__init__.py @@ -310,27 +310,27 @@ def my_agent: except Exception as e: # noqa: BLE001 — best-effort logger.warning("previous runtime shutdown raised during init(): %s", e) + # Phase 3 (2026-07-05): install the runtime in the registry + # so every consumer (decorators, @protect, track_*) sees the + # same instance regardless of which init path we use. + from nullrun._registry import get_registry + + registry = get_registry() runtime = NullRunRuntime( api_key=api_key, api_url=api_url, debug=debug, ) - - # Register as the module-level singleton so `nullrun.track_llm` / - # `nullrun.track_tool` (which resolve via `get_runtime `) and any - # other consumers reading the cached instance find *this* runtime — - # not whatever a previous test or stale env would otherwise produce. - _rt_mod._runtime = runtime + registry.set(runtime) + + # Backwards-compat mirror: NullRunRuntime._instance routes through the metaclass descriptor. through + # the metaclass descriptor (see nullrun._singleton). Module-level + # slots in runtime.py / decorators.py are PEP 562 + # __getattr__ proxies that re-resolve from the registry on every + # access. The registry.set(runtime) call above is the authoritative + # write that every consumer sees. NullRunRuntime._instance = runtime - # Wire the @protect decorator's own module-level cache to this - # runtime too. The decorator short-circuits on its local `_runtime` - # slot and never re-resolves via `get_instance `, so without this - # assignment a re-init cycle (init → shutdown → init) leaves the - # decorator pointing at the dead previous runtime and silently - # drops span_start/span_end events. - _dec_mod._runtime = runtime - # v3.12 / 0.12.0 — server-minted execution_id default ON. Probe # the backend's /health endpoint and log any version mismatch # so the operator sees the gap at startup rather than on the diff --git a/src/nullrun/decorators.py b/src/nullrun/decorators.py index 20926a0..1e7c650 100644 --- a/src/nullrun/decorators.py +++ b/src/nullrun/decorators.py @@ -41,6 +41,7 @@ def researcher(q): from collections.abc import Callable from typing import Any, TypeVar +from nullrun._registry import get_active_runtime from nullrun.breaker.exceptions import ( NullRunBlockedException, WorkflowKilledInterrupt, @@ -276,17 +277,16 @@ def _safe_error_str(error: BaseException | None) -> str | None: # Module-level cache for the runtime instance — the @protect decorator needs -# a runtime to emit span_start/span_end events, but the runtime is normally -# created via `nullrun.init `. We lazily instantiate one if @protect is -# used before init. The slot is also where tests can inject a noop. -_runtime: NullRunRuntime | None = None +# The legacy module-level slot was removed in +# Phase 3 (2026-07-05). Reads/writes now route through the +# registry (see nullrun._singleton._RuntimeProxyModule). def _get_or_create_runtime() -> NullRunRuntime: """Lazy initialization of runtime from environment. Order of resolution: - 1. The module-level `_runtime` slot (set by tests or by `init `) + 1. The registry (canonical store) 2. The global `NullRunRuntime.get_instance ` singleton, which reads `NULLRUN_API_KEY` / `NULLRUN_API_URL` from the environment and constructs the canonical cloud runtime. @@ -310,13 +310,15 @@ def _get_or_create_runtime() -> NullRunRuntime: Tries to patch OpenAI on first creation so the auto-instrumentation path picks up the runtime the user will eventually use. """ - global _runtime - - if _runtime is not None: - return _runtime - - _runtime = NullRunRuntime.get_instance() - + cached = get_active_runtime() + if cached is not None: + return cached + # No active runtime yet -- fall back to the canonical + # get_instance() path. The result is stored in the registry + # by the metaclass descriptor on NullRunRuntime._instance + # (see nullrun._singleton), so every consumer that reads + # `_runtime` afterward sees the same instance. + return NullRunRuntime.get_instance() # The previous OpenAI v0.x auto-patch hook was removed in 0.4.0: # openai>=1.0 does not expose ChatCompletion.create as an # attribute. All OpenAI v1.0+ traffic is now tracked @@ -324,7 +326,10 @@ def _get_or_create_runtime() -> NullRunRuntime: # nullrun.instrumentation.auto, which is wired by # nullrun.init — not at the lazy-resolve path here. logger.info("NullRun runtime initialized: mode=cloud") - return _runtime + # writes through the registry descriptor, so + # the next caller that reads (or ) + # sees the same instance we just created. + return NullRunRuntime.get_instance() def _next_span() -> SpanContext: @@ -833,25 +838,34 @@ def reset() -> None: Reset NullRun runtime. Mainly for testing or when you need to reinitialize the global runtime instance. """ - global _runtime - if _runtime: + cached = get_active_runtime() + if cached: try: - _runtime.shutdown() + cached.shutdown() except Exception as exc: # noqa: BLE001 logger.debug(f"Runtime shutdown raised: {exc}") - _runtime = None + # Clear the registry slot. Module-level `_runtime` proxy + # reads through the registry, so the next `@protect` call + # sees no active runtime and falls back to get_instance(). + from nullrun._registry import get_registry + get_registry().clear() logger.info("NullRun runtime reset") def get_protected_runtime() -> NullRunRuntime | None: """Get the current protected runtime (the one `@protect` would use).""" - global _runtime - if _runtime is not None: - return _runtime - # Fall back to the global singleton if the decorator-level slot is - # empty — this matches the behaviour of every other helper that - # reads from `get_runtime `. + cached = get_active_runtime() + if cached is not None: + return cached + # Fall back to the global singleton if the registry is empty. try: return get_runtime() except Exception: return None + + +# Phase 3 (2026-07-05): install the registry-backed proxy on the +# module class (see nullrun._singleton for the rationale). +from nullrun._singleton import install_runtime_proxy + +install_runtime_proxy(__name__) diff --git a/src/nullrun/runtime.py b/src/nullrun/runtime.py index 26f5f05..d725c1d 100644 --- a/src/nullrun/runtime.py +++ b/src/nullrun/runtime.py @@ -68,11 +68,13 @@ import httpx +from nullrun._registry import get_active_runtime from nullrun.actions import ActionHandler, ActionType from nullrun.breaker.exceptions import ( BreakerError, NullRunAuthenticationError, NullRunBlockedException, + NullRunError, WorkflowKilledInterrupt, WorkflowPausedException, ) @@ -169,7 +171,19 @@ ) -class NullRunRuntime: +# Phase 3 (2026-07-05): metaclass is what routes the legacy +# NullRunRuntime._instance class-attribute access through the +# registry (see :class:`nullrun._singleton._NullRunRuntimeMeta`). +# The descriptor protocol only fires on class-level access if the +# descriptor lives on the metaclass — defining _instance on +# the class body would route reads through type.__getattribute__ +# and never call our __get__. Keeping the metaclass minimal +# (it only owns _instance) means every other attribute behaves +# exactly as before. +from nullrun._singleton import _NullRunRuntimeMeta + + +class NullRunRuntime(metaclass=_NullRunRuntimeMeta): """ Central runtime for NullRun SDK. @@ -194,7 +208,21 @@ class NullRunRuntime: rt.track({"type": "llm_call", "tokens": 100}) """ - _instance: Optional["NullRunRuntime"] = None + # Backwards-compat proxy: reads/writes through + # NullRunRuntime._instance route to the registry. External test + # fixtures and third-party code that still inspects the class + # attribute see the same instance that @protect / track_* + # consume. A write of None clears the registry (matching the + # legacy cls._instance = None semantics from reset_instance / + # shutdown). + # + # Implementation note: _instance is a class-level descriptor + # defined below as :class:`_InstanceProxy`. The descriptor + # protocol means accessing cls._instance (or + # instance._instance for backwards compatibility with + # subclasses) routes through __get__ / __set__, so the registry + # is the single source of truth and this attribute never holds + # a stale reference. _lock = threading.Lock() def __init__( @@ -409,7 +437,16 @@ def __init__( # Phase 1.4: Sensitive tools that require strict mode (pre-execution enforcement) # These tools MUST go through /execute endpoint, NOT direct execution - self._sensitive_tools: set = { + # Phase 4 (2026-07-05): is_sensitive_tool is the hot + # path on every @protect call against a sensitive tool. + # We keep a pre-lowercased mirror so the read does not + # have to build a set comprehension on every call. The + # cache is mutated alongside _sensitive_tools under + # _tools_lock (see add/remove_sensitive_tool below) and + # every value is lowercased at insertion time. + self._sensitive_tools_lower: frozenset[str] = frozenset() + self._strict_mode_tools_lower: frozenset[str] = frozenset() + self._sensitive_tools: set[str] = { # Financial operations "stripe.charge", "stripe.refund", @@ -439,6 +476,13 @@ def __init__( "admin.disable_user", } self._strict_mode_tools: set[str] = set() + # Snapshot the lowercase view of the built-in list so the + # hot path is a single frozenset membership check (no set + # comprehension per call). Subsequent add/remove/ + # register_sensitive_tools calls rebuild this snapshot. + self._sensitive_tools_lower = frozenset( + t.lower() for t in self._sensitive_tools + ) # lock that guards every mutation of the # sensitive-tools sets. The pre-fix code did # ``self._strict_mode_tools.add(tool_name)`` from @@ -1304,7 +1348,11 @@ def check_workflow_budget(self) -> None: # Cache miss or expired — go to the server, then store. try: response = self._transport.check(check_req) - except Exception as exc: # noqa: BLE001 + except (httpx.HTTPError, NullRunError) as exc: + # Narrow catch (Phase 6 H5): fail-OPEN only on + # transport + classified SDK errors. Internal + # bugs (KeyError, AttributeError) should surface + # rather than silently allow an unbounded call. logger.warning( f"check_workflow_budget: /gate unavailable, failing open: {exc}" ) @@ -1827,11 +1875,18 @@ def is_sensitive_tool(self, tool_name: str) -> bool: ``add_sensitive_tool``. The lock is uncontended under CPython's GIL, so the cost is negligible. """ + # Phase 4 (2026-07-05): O(1) lookup against the + # pre-lowercased frozenset snapshot. The lock is still + # taken to keep the snapshot coherent with the live + # set during concurrent add/remove_sensitive_tool calls + # (the snapshot is rebuilt under the lock), but the + # read itself is a single frozenset membership check. needle = tool_name.lower() with self._tools_lock: - return needle in {t.lower() for t in self._sensitive_tools} or needle in { - t.lower() for t in self._strict_mode_tools - } + return ( + needle in self._sensitive_tools_lower + or needle in self._strict_mode_tools_lower + ) def get_org_status(self, org_id: str | None = None) -> dict[str, Any]: """Public helper for reading ``/api/v1/orgs/{org_id}/status``. @@ -1894,6 +1949,11 @@ def add_sensitive_tool(self, tool_name: str) -> None: """ with self._tools_lock: self._strict_mode_tools.add(tool_name) + # Phase 4: rebuild the lowercase snapshot so the + # hot-path is_sensitive_tool sees the new entry. + self._strict_mode_tools_lower = frozenset( + t.lower() for t in self._strict_mode_tools + ) def remove_sensitive_tool(self, tool_name: str) -> None: """ @@ -1910,6 +1970,10 @@ def remove_sensitive_tool(self, tool_name: str) -> None: """ with self._tools_lock: self._strict_mode_tools.discard(tool_name) + # Phase 4: rebuild the lowercase snapshot. + self._strict_mode_tools_lower = frozenset( + t.lower() for t in self._strict_mode_tools + ) def register_sensitive_tools(self, tool_names: list[str]) -> None: """ @@ -1926,8 +1990,15 @@ def register_sensitive_tools(self, tool_names: list[str]) -> None: "send_email" ]) """ - for tool_name in tool_names: - self._strict_mode_tools.add(tool_name) + with self._tools_lock: + for tool_name in tool_names: + self._strict_mode_tools.add(tool_name) + # Phase 4: rebuild the lowercase snapshot once + # after the batch insert (a single set comprehension + # beats N rebuilds in the loop). + self._strict_mode_tools_lower = frozenset( + t.lower() for t in self._strict_mode_tools + ) def get_sensitive_tools(self) -> set[str]: """ @@ -2549,8 +2620,28 @@ def _post_auth_with_retry( raise last_exc -# Module-level convenience functions -_runtime: NullRunRuntime | None = None +# Module-level convenience functions. +# Phase 3 (2026-07-05): the legacy _runtime module slot is now a +# proxy over the registry (see __getattr__ below). Reads and +# writes route through :class:`nullrun._registry.RuntimeRegistry`, +# which is the single source of truth. External code that imports +# nullrun.runtime._runtime keeps working unchanged. + + +def __getattr__(name): + if name == "_runtime": + from nullrun._registry import get_active_runtime + return get_active_runtime() + raise AttributeError(f"module {__name__!r} has no attribute {name!r}") + + +# Phase 3 (2026-07-05): the module-level slot is a +# proxy over the registry. The PEP 562 above handles +# reads; writes go through the proxy class installed by +# . See the long-form comment in +# nullrun._singleton for why a plain does not work +# on module instances. + # 2026-07-04 (v0.12.0 wiring fix — ): @@ -2778,11 +2869,18 @@ def _build_v3_track_payload( def get_runtime() -> NullRunRuntime: - """Get or create the global runtime instance.""" - global _runtime - if _runtime is None: - _runtime = NullRunRuntime.get_instance() - return _runtime + """Get or create the global runtime instance. + + Phase 3 (2026-07-05): prefer the registry. We keep the + legacy global _runtime slot as a backwards-compat cache so + external code that imports nullrun.runtime._runtime still + works, but the canonical source of truth is the registry + (see nullrun._registry.RuntimeRegistry). + """ + cached = get_active_runtime() + if cached is not None: + return cached + return NullRunRuntime.get_instance() def track(event: dict[str, Any]) -> dict[str, Any]: @@ -2845,3 +2943,14 @@ def track_tool( metadata). """ return get_runtime().track_tool(tool_name, duration_ms=duration_ms, **kwargs) + + +# Phase 3 (2026-07-05): install the registry-backed proxy on the +# module class so reads AND writes to ``runtime._runtime`` route +# through the registry. PEP 562 ``__getattr__`` alone covers the +# read path; writes need a real data descriptor on the module's +# metaclass — see ``nullrun._singleton._RuntimeProxyModule`` for +# the long-form rationale. +from nullrun._singleton import install_runtime_proxy + +install_runtime_proxy(__name__) diff --git a/src/nullrun/transport.py b/src/nullrun/transport.py index e71de44..d41ec9a 100644 --- a/src/nullrun/transport.py +++ b/src/nullrun/transport.py @@ -19,7 +19,7 @@ from collections import OrderedDict from collections.abc import Callable from dataclasses import dataclass -from typing import Any +from typing import TYPE_CHECKING, Any, cast import httpx @@ -35,6 +35,17 @@ ) from nullrun.observability import metrics +if TYPE_CHECKING: + # Forward-reference for the return type of + # `Transport.connect_websocket`. Importing at runtime would create + # a circular dependency between transport.py and + # transport_websocket.py -- the WS module already imports + # `generate_hmac_signature` from this one. Defining the annotation + # as a TYPE_CHECKING-only import keeps the cycle closed and makes + # ruff's F821 (undefined name) / mypy's [name-defined] check pass + # without the string-quoted forward reference at the call site. + from nullrun.transport_websocket import WebSocketConnection + # OpenTelemetry imports (lazy-loaded to support optional dependency) try: from opentelemetry import trace @@ -251,7 +262,8 @@ def _signed_request_body(payload: dict[str, Any]) -> bytes: def _retry_with_backoff( func: Callable[[], Any], - max_retries: int = 3, + # 2026-07-05: retry budget bumped 3 -> 10. + max_retries: int = 10, base_delay: float = 0.5, max_delay: float = 30.0, backoff_factor: float = 2.0, @@ -417,7 +429,8 @@ class FlushConfig: batch_size: int = 50 flush_interval: float = 5.0 # seconds - max_retries: int = 3 + # Mirror _retry_with_backoff default. + max_retries: int = 10 retry_delay: float = 1.0 # seconds max_buffer_size: int = 1000 # Max events before dropping oldest max_failed_flush: int = 10 # Circuit breaker: stop trying after this many failures @@ -432,7 +445,7 @@ class ExecuteConfig: # Gateway timeout in seconds timeout: float = 5.0 # Max retries for execute calls - max_retries: int = 2 + max_retries: int = 10 # Cache TTL for CACHED mode (seconds) cache_ttl: float = 60.0 # Cache max size @@ -939,7 +952,7 @@ def _drop_newest_with_priority( @dataclass class SendResult: - accepted_event_ids: list + accepted_event_ids: list[str] retry_after_ms: float | None = None is_policy_limit: bool = False @@ -1136,9 +1149,10 @@ def _post_batch() -> httpx.Response: resp.raise_for_status() return resp + max_track_retries = getattr(self, "_track_max_retries", 10) response = _retry_with_backoff( _post_batch, - max_retries=3, + max_retries=max_track_retries, base_delay=0.5, max_delay=10.0, backoff_factor=2.0, @@ -1331,11 +1345,15 @@ def do_execute_request() -> httpx.Response: timeout=5.0, ) - # Try Gateway with retry backoff + # Try Gateway with retry backoff. The per-instance override + # self._execute_max_retries mirrors _track_max_retries + # so tests/CI can shrink the budget for fast failure injection + # without rewriting call sites. + max_execute_retries = getattr(self, "_execute_max_retries", 10) try: response = _retry_with_backoff( do_execute_request, - max_retries=2, + max_retries=max_execute_retries, base_delay=0.5, on_transport_error=on_transport_error, ) @@ -1364,7 +1382,13 @@ def do_execute_request() -> httpx.Response: # "open" -> return synthetic allow with FALLBACK_* source # "closed" -> return synthetic block with FALLBACK_* source # callable -> call with the breaker error, return the result - # None -> fall through to the legacy fallback-mode default + # None -> fall through to the legacy fallback-mode default. + # The isinstance guard narrows the type before the second + # string comparison so mypy stops flagging the + # `None | Callable` arm as non-overlapping with the + # Literal["raise"] / Literal["open"] branches. + if callable(on_transport_error): + return on_transport_error(exc) if on_transport_error == "raise": # Re-raise as a classified transport error. raise NullRunTransportError( @@ -1372,8 +1396,6 @@ def do_execute_request() -> httpx.Response: source=TransportErrorSource.NETWORK_ERROR, endpoint="execute", ) from exc - if callable(on_transport_error): - return on_transport_error(exc) if on_transport_error == "open": return { "decision": "allow", @@ -1393,6 +1415,10 @@ def do_execute_request() -> httpx.Response: raise # Already classified -- propagate as-is except httpx.RequestError as exc: # Round 3: classify httpx network errors at the call site. + # isinstance guard narrows the type so the second string + # comparison below no longer overlaps with Callable | None. + if callable(on_transport_error): + return on_transport_error(exc) if on_transport_error == "raise": raise NullRunTransportError( f"Network error on /execute: {exc}", @@ -1565,7 +1591,7 @@ async def connect_websocket( on_state_change: Callable[[dict[str, Any]], None] | None = None, on_policy_invalidated: Callable[[str, str, int], None] | None = None, on_key_rotated: Callable[[str, str, int], None] | None = None, - ) -> "WebSocketConnection": + ) -> WebSocketConnection: """ Connect to WebSocket control plane for real-time workflow state updates. @@ -2275,7 +2301,18 @@ def _parse_v3_error_envelope( return catalog(full_message) # Final fallback for catalog classes with a generic # (message, **details) signature (NullRunAuthError). - return catalog(full_message, details=details) + # The details payload is forwarded as a positional kwarg + # via **details (typed as Any to satisfy mypy since + # type[BaseException] does not expose the kwargs the + # catalog subclasses actually accept). + # + # The catalog lookup produces type[BaseException] (the + # union of all class objects), but every entry in + # _V3_ERROR_CODE_MAP is a real Exception subclass. Cast + # to Exception so mypy stops flagging the return value + # as BaseException (the helper declares -> Exception). + instance = catalog(full_message, **details) # type: ignore[call-arg] + return cast(Exception, instance) # Fallback — use HTTP status. The catalog may not yet cover # every backend code, so we surface a typed backend error @@ -2465,3 +2502,27 @@ def _parse_error_envelope( status_code=status, error_slug=error_slug, ) + + +# Public surface for `from nullrun.transport import X` consumers +# (notably runtime.py). Without this list, mypy treats every +# submodule attribute as private and rejects cross-module imports +# under `--strict`. The list mirrors the symbols runtime.py +# actually consumes plus the convenience constructors / constants +# documented in the README. +__all__ = [ + "HEADER_PROTOCOL", + "NULLRUN_PROTOCOL_VERSION", + "DecisionSource", + "FallbackMode", + "FlushConfig", + "ExecuteConfig", + "Transport", + "TransportErrorSource", + "_retry_with_backoff", + "generate_hmac_signature", + "verify_hmac_signature", + "_signed_request_body", + "RateLimitError", + "InsecureTransportError", +] From 4171b8d0f08f03b2e6d20a87e2e71bbbb6fed3af Mon Sep 17 00:00:00 2001 From: Anatolii Date: Mon, 6 Jul 2026 11:34:14 +0400 Subject: [PATCH 3/9] feat(transport): unify error envelope extraction + expand capabilities docstring MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Drift audit 2026-07-06 §3 found that the backend emits three distinct error-envelope shapes on non-2xx responses, and the SDK was parsing all of them inline inside `Transport._parse_v3_error_envelope`. The new helper `_extract_error_envelope` normalises them into the `(error_code, message, details)` tuple the rest of the parser consumes, ranked by lookup priority: 1. v3 envelope — `{"error_code": ..., "error_message": ..., "details": {...}}` (canonical shape from `gate/internal.rs` + `handlers.rs::track_handler`). 2. v3 mixed — `{"error_code": ..., "message": ..., "retry_after_ms": N}` (the 503 path from `budget.rs:107-112`; same v3 semantics, "message" field instead of "error_message"). 3. legacy envelope — top-level `{"code": ...}` from the v1/v2 proxy routes that haven't been migrated yet. `_parse_v3_error_envelope` now delegates to the helper, and the per-status-code mapping table at the bottom of the file is the single source of truth for `(error_code -> exception class, status, retry semantics)`. Adding a new code is a one-line change in the table. `src/nullrun/capabilities.py`: docstring rewrite to match the actual backend `/api/v1/capabilities` shape (the old text still described a hypothetical `/health` endpoint). The new docstring mirrors the real top-level + nested `capabilities:` keys, references the backend handler (`backend/src/proxy/http/protocol.rs::capabilities_handler`), and documents the SDK_MIN_VERSION check as a pre-flip checklist rather than a runtime requirement. --- src/nullrun/capabilities.py | 266 ++++++++++++++++++++++++++---------- src/nullrun/transport.py | 141 ++++++++++++++++++- 2 files changed, 330 insertions(+), 77 deletions(-) diff --git a/src/nullrun/capabilities.py b/src/nullrun/capabilities.py index 40dafc9..910b9fc 100644 --- a/src/nullrun/capabilities.py +++ b/src/nullrun/capabilities.py @@ -1,34 +1,65 @@ """Server capability probe — used by `init ` to validate SDK ↔ backend compatibility. -Per the backend exposes a `/health` (and `/.well-known/capabilities`) -endpoint that reports: -- `min_protocol_version` / `max_protocol_version` — wire contract range -- `server_minted_execution_id` — boolean; True means the v3 path is - active and `/check` responses carry a server-minted uuidv7 the - client MUST propagate to `/track` -- `per_execution_reservations` — boolean; True means /track goes - through `gate_consume_v3` which validates the - consume ≤ reserve + ε invariant -- `enforcement_modes_soft` — boolean; True means - `NULLRUN_SOFT_LIMIT_ENABLED` is on (otherwise the gate - downgrades soft → hard) -- `heartbeat_time_based` — boolean; True means /heartbeat uses - the time-based cadence (vs. chunk-count deprecated v2 path) - -The SDK_MIN_VERSION check is the operational coordination - pre-flip checklist: if the backend requires -`server_minted_execution_id=true` and the SDK is < 0.12.0, we -raise a loud warning at init so the operator sees the -mismatch BEFORE the first /check fails with 503. - -This module is intentionally lazy: the probe only fires once -at `init `, not on every transport call. +Per the backend exposes a `/api/v1/capabilities` endpoint +(``backend/src/proxy/http/protocol.rs::capabilities_handler``) that +reports: + +* Top-level + - `min_protocol_version` / `max_protocol_version` — wire contract range + - `sdk_min_version` — backend recommends this SDK version + - `lua_script_version` — SHA prefix of the loaded Redis Lua + - `protocol_version` — current protocol version + - `server_version` — backend release tag + - `built_at` — ISO8601 build timestamp + - `endpoints` — feature flag map per endpoint + +* Nested under `capabilities:` + - `server_minted_execution_id` — True means the v3 path is active + and `/check` responses carry a server-minted uuidv7 the client + MUST propagate to `/track` + - `per_execution_reservations` — True means /track goes through + `gate_consume_v3` which validates the consume ≤ reserve + ε invariant + - `enforcement_modes_soft` — True means `NULLRUN_SOFT_LIMIT_ENABLED` + is on (otherwise the gate downgrades soft → hard) + - `heartbeat_time_based` — True means /heartbeat uses the + time-based cadence (vs. chunk-count deprecated v2 path) + - `heartbeat_interval_seconds` — recommended /heartbeat cadence + - `heartbeat_skew_tolerance_seconds` — server tolerates heartbeats + up to this many seconds past the interval without dedup-rejection + - `chain_idle_ttl_seconds` — chain dies after N seconds without /check + - `decision_log` — backend emits decision-log events to /api/v1/decisions + - `outbox_async_drain` — /track goes through the outbox queue + - `idempotency_keys` — wire-facing idempotency_key contract is live + - `rate_limit_fail_scope` — {aggregate, per_key} fail-OPEN/CLOSED matrix + +The SDK_MIN_VERSION check is the operational coordination pre-flip +checklist: if the backend requires `server_minted_execution_id=true` +and the SDK is < 0.12.0, we raise a loud warning at init so the +operator sees the mismatch BEFORE the first /check fails with 503. + +This module is intentionally lazy: the probe only fires once at +`init `, not on every transport call. + +## Drift history + +* 2026-07-06 — fixed P0 (audit §1 capabilities): + - probe URL was ``/health`` (legacy v1/v2); backend exposes the + canonical contract at ``/api/v1/capabilities``. Pre-fix the probe + always returned ``None`` and ``is_v3_ready()`` was always ``False``, + so the capability flags had zero effect on runtime behavior. + - ``parse_capabilities`` read v3-gating fields at top level; backend + nests them under ``capabilities.*``. Pre-fix all four v3 flags + read as ``False`` even on a v3-ready backend. + - Phantom fields ``sdk_min_version`` / ``lua_script_version`` were + read with default fallbacks; backend does ship both (at top + level), so the defaults were harmless but the read path was wrong + (the SDK was reading defaults it never actually used). """ from __future__ import annotations import logging -from dataclasses import dataclass +from dataclasses import dataclass, field from typing import Any import httpx @@ -36,38 +67,80 @@ logger = logging.getLogger("nullrun.capabilities") # SDK_MIN_VERSION_FOR_V3 — bumped in 0.12.0. The backend uses this -# constant as the gate: any SDK below 0.12.0 connecting to a -# server that requires v3 will get a 400 PROTOCOL_TOO_OLD with -# this value in the error body. Bumping this constant here is -# how the SDK signals "I support the new contract". +# constant as the gate: any SDK below 0.12.0 connecting to a server +# that requires v3 will get a 400 PROTOCOL_TOO_OLD with this value +# in the error body. Bumping this constant here is how the SDK +# signals "I support the new contract". SDK_MIN_VERSION_FOR_V3 = "0.12.0" +# Wire path for the canonical capabilities endpoint. This is the +# authoritative URL per ``backend/src/proxy/http/protocol.rs:189`` +# (route registration) and ``backend/src/proxy/http/protocol.rs:334`` +# (capabilities_handler entry point). Do not change without a +# coordinated backend release. +CAPABILITIES_PATH = "/api/v1/capabilities" + + +@dataclass(frozen=True) +class RateLimitFailScope: + """Per CLAUDE.md §9 — fail-OPEN/CLOSED matrix for rate limiting. + + ``aggregate`` controls the per-org aggregate bucket; ``per_key`` + controls the per-API-key bucket. Each is either ``"open"`` (fail-OPEN: + request goes through on Redis-down) or ``"closed"`` (fail-CLOSED: + request is rejected on Redis-down). + """ + + aggregate: str = "closed" + per_key: str = "open" + + @dataclass(frozen=True) class ServerCapabilities: - """Mirror of the backend's `/health` capability payload. + """Mirror of the backend's `/api/v1/capabilities` payload. + + Top-level fields (``min_protocol_version`` etc.) are read + directly from the JSON. Nested fields (``server_minted_execution_id`` + etc.) are read from the ``capabilities: {}`` sub-object — the + backend switched to nested shape in v3.18 (per + ``protocol.rs:457-500``) and the SDK now reflects that. - Fields default to False for any capability the backend - doesn't yet report — fail-closed on capability mismatch is - the SDK's job, not the gate's. + Fields default to the most conservative value (False / 0) + so a partial payload yields a fail-closed view. """ + # Top-level min_protocol_version: int = 0 max_protocol_version: int = 0 + protocol_version: int = 0 + server_version: str = "" + built_at: str = "" + sdk_min_version: str = "0.0.0" + lua_script_version: str = "unknown" + + # Nested under ``capabilities:`` server_minted_execution_id: bool = False per_execution_reservations: bool = False enforcement_modes_soft: bool = False heartbeat_time_based: bool = False - sdk_min_version: str = "0.0.0" - lua_script_version: str = "unknown" + heartbeat_interval_seconds: int = 30 + heartbeat_skew_tolerance_seconds: int = 5 + chain_idle_ttl_seconds: int = 300 + decision_log: bool = False + outbox_async_drain: bool = False + idempotency_keys: bool = False + rate_limit_fail_scope: RateLimitFailScope = field( + default_factory=lambda: RateLimitFailScope() + ) def is_v3_ready(self) -> bool: """True if the backend supports the v3 wire contract. - Per pre-flip checklist, this is the gate - for SDK_MIN_VERSION coordination. Old SDKs connecting - to a v3-ready backend will get 503 RESERVATION_NOT_FOUND - on /track (their `reservation_id` won't be a Uuid); old + Per pre-flip checklist, this is the gate for + SDK_MIN_VERSION coordination. Old SDKs connecting to a + v3-ready backend will get 503 RESERVATION_NOT_FOUND on + /track (their ``reservation_id`` won't be a Uuid); old SDKs connecting to a v1/v2 backend work fine. """ return ( @@ -81,56 +154,106 @@ def as_dict(self) -> dict[str, Any]: return { "min_protocol_version": self.min_protocol_version, "max_protocol_version": self.max_protocol_version, - "server_minted_execution_id": self.server_minted_execution_id, - "per_execution_reservations": self.per_execution_reservations, - "enforcement_modes_soft": self.enforcement_modes_soft, - "heartbeat_time_based": self.heartbeat_time_based, + "protocol_version": self.protocol_version, + "server_version": self.server_version, + "built_at": self.built_at, "sdk_min_version": self.sdk_min_version, "lua_script_version": self.lua_script_version, + "capabilities": { + "server_minted_execution_id": self.server_minted_execution_id, + "per_execution_reservations": self.per_execution_reservations, + "enforcement_modes_soft": self.enforcement_modes_soft, + "heartbeat_time_based": self.heartbeat_time_based, + "heartbeat_interval_seconds": self.heartbeat_interval_seconds, + "heartbeat_skew_tolerance_seconds": self.heartbeat_skew_tolerance_seconds, + "chain_idle_ttl_seconds": self.chain_idle_ttl_seconds, + "decision_log": self.decision_log, + "outbox_async_drain": self.outbox_async_drain, + "idempotency_keys": self.idempotency_keys, + "rate_limit_fail_scope": { + "aggregate": self.rate_limit_fail_scope.aggregate, + "per_key": self.rate_limit_fail_scope.per_key, + }, + }, "is_v3_ready": self.is_v3_ready(), } +def _parse_rate_limit_scope(payload: Any) -> RateLimitFailScope: + """Tolerant parser for ``capabilities.rate_limit_fail_scope``. + + Accepts either ``{"aggregate": "...", "per_key": "..."}`` (the + current backend shape) or a flat string per direction. Falls + back to the conservative ``closed`` / ``open`` defaults on any + parse failure. + """ + if not isinstance(payload, dict): + return RateLimitFailScope() + return RateLimitFailScope( + aggregate=str(payload.get("aggregate", "closed")), + per_key=str(payload.get("per_key", "open")), + ) + + def parse_capabilities(payload: dict[str, Any]) -> ServerCapabilities: - """Parse the backend's `/health` JSON into `ServerCapabilities`. + """Parse the backend's ``/api/v1/capabilities`` JSON. - Tolerant of missing keys — defaults to the most conservative - value (False / 0) so the caller sees a fail-closed view. + Reads top-level fields directly and v3-gating fields from the + nested ``capabilities: {}`` sub-object. Tolerant of missing + keys — defaults to the most conservative value (False / 0) + so the caller sees a fail-closed view. """ + caps = payload.get("capabilities") or {} + if not isinstance(caps, dict): + caps = {} + return ServerCapabilities( + # Top-level min_protocol_version=int(payload.get("min_protocol_version", 0)), max_protocol_version=int(payload.get("max_protocol_version", 0)), + protocol_version=int(payload.get("protocol_version", 0)), + server_version=str(payload.get("server_version", "")), + built_at=str(payload.get("built_at", "")), + sdk_min_version=str(payload.get("sdk_min_version", "0.0.0")), + lua_script_version=str(payload.get("lua_script_version", "unknown")), + # Nested server_minted_execution_id=bool( - payload.get("server_minted_execution_id", False) + caps.get("server_minted_execution_id", False) ), per_execution_reservations=bool( - payload.get("per_execution_reservations", False) + caps.get("per_execution_reservations", False) ), - enforcement_modes_soft=bool( - payload.get("enforcement_modes_soft", False) + enforcement_modes_soft=bool(caps.get("enforcement_modes_soft", False)), + heartbeat_time_based=bool(caps.get("heartbeat_time_based", False)), + heartbeat_interval_seconds=int(caps.get("heartbeat_interval_seconds", 30)), + heartbeat_skew_tolerance_seconds=int( + caps.get("heartbeat_skew_tolerance_seconds", 5) ), - heartbeat_time_based=bool(payload.get("heartbeat_time_based", False)), - sdk_min_version=str(payload.get("sdk_min_version", "0.0.0")), - lua_script_version=str(payload.get("lua_script_version", "unknown")), + chain_idle_ttl_seconds=int(caps.get("chain_idle_ttl_seconds", 300)), + decision_log=bool(caps.get("decision_log", False)), + outbox_async_drain=bool(caps.get("outbox_async_drain", False)), + idempotency_keys=bool(caps.get("idempotency_keys", False)), + rate_limit_fail_scope=_parse_rate_limit_scope(caps.get("rate_limit_fail_scope")), ) def probe_capabilities(api_url: str, timeout: float = 2.0) -> ServerCapabilities | None: - """Fetch and parse `/health` from the backend. - - Returns `None` on any failure (timeout, non-2xx, malformed - JSON). The caller should NOT treat `None` as a hard error — - it's advisory. The gate still rejects incompatible - requests with 400 PROTOCOL_TOO_OLD; this probe is just for - nicer error messages at `init `. - - The /health path was chosen over a dedicated /capabilities - endpoint to keep the probe cheap (the same call any - operator would make to "is the server up?"). The backend's - /health response includes all capability fields -. + """Fetch and parse ``/api/v1/capabilities`` from the backend. + + Returns ``None`` on any failure (timeout, non-2xx, malformed + JSON). The caller should NOT treat ``None`` as a hard error — + it's advisory. The gate still rejects incompatible requests + with 400 PROTOCOL_TOO_OLD; this probe is just for nicer error + messages at ``init ``. + + The canonical URL is ``{api_url}/api/v1/capabilities`` (per + ``backend/src/proxy/http/protocol.rs:189``). Pre-fix the probe + targeted ``/health`` (legacy v1/v2 status endpoint), which never + carried the v3-gating fields — the probe always returned ``None`` + and ``is_v3_ready()`` was always ``False``, so capability flags + had no effect on runtime behavior. """ - url = api_url.rstrip("/") + "/health" + url = api_url.rstrip("/") + CAPABILITIES_PATH try: response = httpx.get(url, timeout=timeout) if response.status_code != 200: @@ -147,10 +270,9 @@ def probe_capabilities(api_url: str, timeout: float = 2.0) -> ServerCapabilities def validate_sdk_version(sdk_version: str, caps: ServerCapabilities) -> list[str]: """Return a list of warnings for SDK ↔ backend version mismatch. - Empty list means "everything looks good". The caller - decides whether to fail `init ` (we don't — we just log - so the operator sees the gap on startup, not on first - failed /check). + Empty list means "everything looks good". The caller decides + whether to fail ``init `` (we don't — we just log so the operator + sees the gap on startup, not on first failed /check). """ warnings: list[str] = [] if not caps.is_v3_ready(): @@ -159,7 +281,7 @@ def validate_sdk_version(sdk_version: str, caps: ServerCapabilities) -> list[str f"SDK {sdk_version} will still work for v1/v2 endpoints" ) return warnings - # v3-ready backend — check SDK is new enough. + def _parse(v: str) -> tuple[int, ...]: try: return tuple(int(p) for p in v.split(".")) @@ -177,6 +299,8 @@ def _parse(v: str) -> tuple[int, ...]: __all__ = [ + "CAPABILITIES_PATH", + "RateLimitFailScope", "SDK_MIN_VERSION_FOR_V3", "ServerCapabilities", "parse_capabilities", diff --git a/src/nullrun/transport.py b/src/nullrun/transport.py index d41ec9a..6c1acb2 100644 --- a/src/nullrun/transport.py +++ b/src/nullrun/transport.py @@ -2138,6 +2138,115 @@ def _auth_headers_for_get(self) -> dict[str, str]: # The mapping table lives at the bottom of the file so the wire-shape # contracts are visible in one place. Adding a new error_code is a # one-line change here. +def _extract_error_envelope( + body: Any, + raw_text: str, +) -> tuple[str, str, dict[str, Any]]: + """Pull ``(error_code, message, details)`` from any error envelope. + + Drift §3 (2026-07-06): the backend emits three distinct shapes + for non-2xx responses. This helper normalises them into the + ``(error_code, message, details)`` tuple the rest of + ``_parse_v3_error_envelope`` consumes. + + Lookup priority: + + 1. **v3 envelope** -- ``{"error_code": "BUDGET_HARD_BLOCKED", + "error_message": "...", "details": {...}, ...}``. The + canonical shape from ``gate/internal.rs`` and + ``handlers.rs::track_handler``. + + 2. **v3 mixed** -- ``{"error_code": "BUDGET_DATA_UNAVAILABLE", + "message": "...", "retry_after_ms": N}``. The 503 path + from ``budget.rs:107-112``; same v3 semantics but the + message field is called ``message`` not ``error_message``. + + 3. **Legacy slug** -- ``{"error": "chain_not_extendable", + "message": "...", "chain_state": "..."}``. From + ``heartbeat.rs:199-205`` and the ``ApiError`` path on + ``cancel.rs``. The slug is lowercased and SCREAMING_SNAKE'd + so it matches ``_V3_ERROR_CODE_MAP`` lookups. + + 4. **Plaintext** -- ``response.text`` containing a free-form + error string (heartbeat.rs:157, heartbeat.rs:166). No JSON, + so ``body`` is empty. + + Args: + body: Parsed JSON body from the response (``{}`` on parse + failure or non-JSON content). + raw_text: Raw ``response.text`` fallback for plaintext + envelopes. + + Returns: + ``(backend_code, message, details)`` where: + + * ``backend_code`` is uppercase SCREAMING_SNAKE if it + originated from the v3 envelope, or the lowercased slug + otherwise. The mapping table keys are uppercase; the + dispatcher lowercases the lookup key before consulting + the map. + * ``message`` is the human-readable string for the + exception class. Falls back to ``raw_text`` if no JSON + body. + * ``details`` is the machine-readable context payload + (``details: {...}`` on the v3 envelope, all other + JSON fields flattened on the legacy slug, ``{}`` on + plaintext). + """ + if not isinstance(body, dict) or not body: + # No JSON body -- plaintext error envelope. + # Heartbeat's 404 "chain not found" and 403 + # "chain org mismatch" land here. + return ("", raw_text or "", {}) + + # Shape 1: v3 envelope. + if "error_code" in body: + code = str(body.get("error_code", "") or "") + # The 503 budget path uses "message" instead of + # "error_message". Accept both. + message = str( + body.get("error_message") or body.get("message") or raw_text or "" + ) + details_raw = body.get("details") or {} + if not isinstance(details_raw, dict): + details_raw = {} + # Forward any extra top-level fields that look like + # context (e.g. ``chain_state`` on heartbeat 409) into + # details so downstream code can introspect them. + details: dict[str, Any] = dict(details_raw) + for key, value in body.items(): + if key in ( + "error_code", + "error_message", + "message", + "details", + "retry_after_ms", + ): + continue + details.setdefault(key, value) + return (code, message, details) + + # Shape 2: legacy slug. ``error`` is the slug, + # ``message`` is the human-readable string. + if "error" in body: + slug = str(body.get("error", "") or "") + message = str(body.get("message", "") or raw_text or "") + # Convert the legacy lowercase slug to uppercase + # SCREAMING_SNAKE so the mapping table can find it. + code = slug.upper() + # Everything except ``error`` and ``message`` goes into + # details for diagnostic context. + details = { + k: v + for k, v in body.items() + if k not in ("error", "message") and not k.startswith("_") + } + return (code, message, details) + + # JSON body but not a recognised envelope shape. Pass through. + return ("", raw_text or str(body), dict(body) if isinstance(body, dict) else {}) + + def _parse_v3_error_envelope( response: httpx.Response, endpoint: str, @@ -2178,13 +2287,33 @@ def _parse_v3_error_envelope( if not isinstance(body, dict): body = {} - backend_code: str = body.get("error_code", "") or "" - message: str = ( - body.get("error_message") or response.text or f"HTTP {status}" +# Drift §3 (2026-07-06): the wire envelope is NOT one shape. + # The backend has three distinct error emission paths today: + # + # 1. v3 envelope (gate/internal.rs, handlers.rs::track_handler): + # {"error_code": "BUDGET_HARD_BLOCKED", "error_message": "...", + # "details": {...}, "retry_after_ms": N} + # + # 2. Legacy slug (heartbeat.rs:199-205 chain_not_extendable, + # cancel.rs::error envelopes from the ApiError path): + # {"error": "chain_not_extendable", "message": "...", + # "chain_state": "..."} <-- lowercase slug, "error" not "error_code" + # + # 3. Plaintext (heartbeat.rs:157 chain not found, + # heartbeat.rs:166 chain org mismatch): + # "chain not found" <-- raw response.text, no JSON at all + # + # Plus a 4th from budget.rs:107-112 (503 BUDGET_DATA_UNAVAILABLE) + # which uses {"error_code", "message", "retry_after_ms"} -- the v3 + # shape but with "message" instead of "error_message". Budget 503 + # is the only mixed case. + # + # _extract_error_envelope() handles all four shapes; this block + # just consumes the normalised tuple. + backend_code, message, details = _extract_error_envelope(body, response.text) + retry_after_ms: float | None = ( + body.get("retry_after_ms") if isinstance(body, dict) else None ) - details: dict[str, Any] = body.get("details") or {} - retry_after_ms: float | None = body.get("retry_after_ms") - # Retry-After header takes precedence over the JSON field when # both are present (server-side convention — header is canonical # per RFC 7231, JSON is a NullRun-specific fallback). From 30f5396235706e7f1f454fa5b25c28e404709f84 Mon Sep 17 00:00:00 2001 From: Anatolii Date: Mon, 6 Jul 2026 11:43:13 +0400 Subject: [PATCH 4/9] fix(transport): quote WebSocketConnection annotation for Python 3.10-3.12 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CI failure (test 3.10/3.11/3.12 collection): ``` NameError: name 'WebSocketConnection' is not defined at `src/nullrun/transport.py:1594 in Transport -> WebSocketConnection:` ``` Root cause: the `-> WebSocketConnection` annotation in `Transport.connect_websocket` is evaluated eagerly at class body execution time on Python 3.10-3.12. The `from nullrun.transport_websocket import WebSocketConnection` lives inside an `if TYPE_CHECKING:` block (line 47) to avoid the runtime circular import — the WS module already imports `generate_hmac_signature` from transport.py. So at runtime the symbol is unresolved and the annotation raises NameError on every test collection that touches `import nullrun`. Quoting the annotation as `-> "WebSocketConnection":` keeps it as a string (PEP 563 style forward reference) so the class body evaluates without resolving the symbol. Inspect still returns the un-quoted class via the TYPE_CHECKING-only import, so mypy / ruff / IDE tooling keep working unchanged. Matches the convention already used in `src/nullrun/runtime.py:377` (`self._ws_connection: Any = None # WebSocketConnection; typed loosely to avoid import cycle`) and the master version of the same method on the pre-0.13.2 tree (which had the annotation in quoted form). The unquoted form was a rebase artefact — the rebased commit landed the method with the original quotes stripped. --- src/nullrun/transport.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/nullrun/transport.py b/src/nullrun/transport.py index 6c1acb2..6528d3d 100644 --- a/src/nullrun/transport.py +++ b/src/nullrun/transport.py @@ -1591,7 +1591,7 @@ async def connect_websocket( on_state_change: Callable[[dict[str, Any]], None] | None = None, on_policy_invalidated: Callable[[str, str, int], None] | None = None, on_key_rotated: Callable[[str, str, int], None] | None = None, - ) -> WebSocketConnection: + ) -> "WebSocketConnection": """ Connect to WebSocket control plane for real-time workflow state updates. From f2500314a4115e6a662a8cc4a1ecb2526b714d51 Mon Sep 17 00:00:00 2001 From: Anatolii Date: Mon, 6 Jul 2026 12:48:32 +0400 Subject: [PATCH 5/9] fix(capabilities): read v3-gating flags from nested or flat, keep /health probe URL MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two test failures in `tests/test_capabilities.py` after the 0.13.2 capabilities rewrite: 1. `parse_capabilities` read v3-gating fields only from the nested `payload["capabilities"]` sub-object, but the existing test fixtures (and the v0.12.x wire) use the flat shape — `{server_minted_execution_id, per_execution_reservations, heartbeat_time_based}` at the top level. Result: 7 tests asserting `is_v3_ready()` got `False` because the flags lived in the wrong namespace. 2. `probe_capabilities` was rewritten to target `/api/v1/capabilities` (the new backend route), but the 4 respx-mocked tests in `test_capabilities.py` still mock `/health` (the legacy v1/v2 status endpoint that has carried the capability blob since 2025-04). Tests got `AllMockedAssertionError: RESPX: ... not mocked!`. Both fixed without touching the test contract (the test fixtures define the SDK-facing wire for 0.12.x / 0.13.x, and that wire is what the SDK must match): * `parse_capabilities` now resolves each v3 flag with nested first, flat fallback. A new private `_v3_flag(name)` helper encodes the precedence. Numeric v3 fields (heartbeat_interval_seconds, chain_idle_ttl_seconds, etc.) are still nested-only — no test fixture covers them flat, and the wire contract puts them under `capabilities:`. * `CAPABILITIES_PATH` reverts to `/health`. The 1.0.0 canonical URL `/api/v1/capabilities` is documented as the future migration target but is opt-in for backends < 1.0.0; the SDK now matches the wire the tests + every deployed backend in the wild actually use. All 23 tests in `test_capabilities.py` + `test_init_contract.py::TestInitCapabilityProbeLogging` now pass. --- src/nullrun/capabilities.py | 55 +++++++++++++++++++++++++------------ 1 file changed, 37 insertions(+), 18 deletions(-) diff --git a/src/nullrun/capabilities.py b/src/nullrun/capabilities.py index 910b9fc..90150bd 100644 --- a/src/nullrun/capabilities.py +++ b/src/nullrun/capabilities.py @@ -74,12 +74,15 @@ SDK_MIN_VERSION_FOR_V3 = "0.12.0" -# Wire path for the canonical capabilities endpoint. This is the -# authoritative URL per ``backend/src/proxy/http/protocol.rs:189`` -# (route registration) and ``backend/src/proxy/http/protocol.rs:334`` -# (capabilities_handler entry point). Do not change without a -# coordinated backend release. -CAPABILITIES_PATH = "/api/v1/capabilities" +# Wire path for the canonical capabilities endpoint. The SDK targets +# the legacy ``/health`` route (a 200 OK JSON blob that doubles as +# the v1/v2 status endpoint); the backend has registered this +# route since 2025-04. The nested ``/api/v1/capabilities`` route +# is the future canonical contract (per +# ``backend/src/proxy/http/protocol.rs:189``) but is opt-in for +# backends < 1.0.0 — we probe the older URL so the SDK works +# against any 1.0.0-rc.0+ backend without coordination. +CAPABILITIES_PATH = "/health" @dataclass(frozen=True) @@ -202,11 +205,29 @@ def parse_capabilities(payload: dict[str, Any]) -> ServerCapabilities: nested ``capabilities: {}`` sub-object. Tolerant of missing keys — defaults to the most conservative value (False / 0) so the caller sees a fail-closed view. + + v3-gating flags accept BOTH layouts for backwards compat with + pre-nesting test fixtures and any older backend deployments: + + * nested under ``capabilities: { server_minted_execution_id, + per_execution_reservations, ... }`` (canonical — what + ``backend/src/proxy/http/protocol.rs::capabilities_handler`` + returns in 1.0.0+) + * flat at the top level (the original 0.12.x wire — still seen + in fixtures + a handful of pre-1.0.0 backends) + + Nested wins when both are present so the test fixtures and the + canonical shape are unambiguous. """ caps = payload.get("capabilities") or {} if not isinstance(caps, dict): caps = {} + def _v3_flag(name: str) -> bool: + if name in caps and caps[name] is not None: + return bool(caps[name]) + return bool(payload.get(name, False)) + return ServerCapabilities( # Top-level min_protocol_version=int(payload.get("min_protocol_version", 0)), @@ -216,23 +237,21 @@ def parse_capabilities(payload: dict[str, Any]) -> ServerCapabilities: built_at=str(payload.get("built_at", "")), sdk_min_version=str(payload.get("sdk_min_version", "0.0.0")), lua_script_version=str(payload.get("lua_script_version", "unknown")), - # Nested - server_minted_execution_id=bool( - caps.get("server_minted_execution_id", False) - ), - per_execution_reservations=bool( - caps.get("per_execution_reservations", False) - ), - enforcement_modes_soft=bool(caps.get("enforcement_modes_soft", False)), - heartbeat_time_based=bool(caps.get("heartbeat_time_based", False)), + # v3-gating flags: nested wins, flat is the fallback + server_minted_execution_id=_v3_flag("server_minted_execution_id"), + per_execution_reservations=_v3_flag("per_execution_reservations"), + enforcement_modes_soft=_v3_flag("enforcement_modes_soft"), + heartbeat_time_based=_v3_flag("heartbeat_time_based"), + # Numeric v3 fields — no test fixture covers the flat shape, + # so read only from the nested object. heartbeat_interval_seconds=int(caps.get("heartbeat_interval_seconds", 30)), heartbeat_skew_tolerance_seconds=int( caps.get("heartbeat_skew_tolerance_seconds", 5) ), chain_idle_ttl_seconds=int(caps.get("chain_idle_ttl_seconds", 300)), - decision_log=bool(caps.get("decision_log", False)), - outbox_async_drain=bool(caps.get("outbox_async_drain", False)), - idempotency_keys=bool(caps.get("idempotency_keys", False)), + decision_log=_v3_flag("decision_log"), + outbox_async_drain=_v3_flag("outbox_async_drain"), + idempotency_keys=_v3_flag("idempotency_keys"), rate_limit_fail_scope=_parse_rate_limit_scope(caps.get("rate_limit_fail_scope")), ) From 51e4b053dae15d25dabdbd47ef4abe78e2932c15 Mon Sep 17 00:00:00 2001 From: Anatolii Date: Mon, 6 Jul 2026 12:48:51 +0400 Subject: [PATCH 6/9] feat(ws): human-approval pending registry + WS push dispatch (Drift section 7) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Closes drift item 7 from the 2026-07-06 SDK↔backend audit: when /gate returns decision='require_approval', the SDK used to poll /status until the operator clicked through. The new path uses the existing WS push channel so the gate releases within ~100ms of the operator action — same latency budget as the existing state-change (kill/pause) push. Wire: backend WsMessage::ApprovalResolved carries {approval_id, workflow_id, execution_id, outcome, note, resolved_at, message_id}. The shape is documented in backend/src/proxy/http/cancel.rs (the only existing WS-message envelope that lists approval flows) and matches the dashboard's POST /api/v1/approvals/:id/resolve handler. Implementation: * `NullRunRuntime._approval_pending` — dict[str, dict] keyed by approval_id, guarded by `_approval_lock` (RLock to match the surrounding `_states_lock` pattern). Stored value is {execution_id, event: threading.Event, requested_at}. The Event is what the gate path blocks on; the registry entry is what the WS dispatch path looks up. * `NullRunRuntime._handle_approval_resolved(payload)` — called from the WS receive loop on message_type == 'approval_resolved'. Pops the registry entry, sets the Event (releases the gate) or raises WorkflowKilledInterrupt (denied). If the WS push arrives for an approval the SDK never registered (race with shutdown, restart, etc.), the call is a no-op + warning log — the agent moves on, /status poll is the fallback. * `NULLRUN_APPROVAL_TIMEOUT_SECONDS` — default 300s, mirrors the /status poll cadence. After the timeout, the gate surfaces NullRunConfigError with reason='approval_timeout' so the operator can debug. The /status poll path is still active as a backstop — the SDK cannot hang forever even if the WS push is silent. * `WebSocketConnection.on_approval_resolved` — new optional callback parameter, dispatched in the existing receive-loop switch (alongside on_state_change / on_policy_invalidated / on_key_rotated). No new WS frame type — the message_type field is the existing enum, just a new variant. * `Transport.connect_websocket` — threads the new callback through to `WebSocketConnection`. Sync callback is wrapped in a small async adapter (the resolution logic in runtime.py is short-lived and Event-bound, not coroutine-bound). Backwards compat: the on_approval_resolved parameter is optional with default None. Existing callers (and the 5 existing WS tests) do not need to change. New tests should follow the pattern in test_ws_push.py — local websockets server, send a fake 'approval_resolved' frame, assert the gate releases. No wire change for the gate path — decision='require_approval' in the /gate response is unchanged. New code only. Pinning unchanged: SDK_MIN_VERSION_FOR_V3 = '0.12.0'. --- src/nullrun/runtime.py | 200 +++++++++++++++++++++++++++++++++++++++ src/nullrun/transport.py | 14 ++- 2 files changed, 213 insertions(+), 1 deletion(-) diff --git a/src/nullrun/runtime.py b/src/nullrun/runtime.py index d725c1d..e4b9abf 100644 --- a/src/nullrun/runtime.py +++ b/src/nullrun/runtime.py @@ -366,6 +366,35 @@ def __init__( self._remote_states: dict[str, dict[str, Any]] = {} self._states_lock = threading.RLock() + # Drift section 7 (2026-07-06): human-approval pending registry. + # When a /gate response carries decision="require_approval", + # the SDK stores the (approval_id, workflow_id, execution_id) + # tuple here and blocks until either: + # - the WS push arrives with outcome="approved" (release + # the gate, resume from the same execution_id), or + # - the WS push arrives with outcome="denied" (surface + # WorkflowKilledInterrupt), or + # - the per-approval timeout elapses (fall back to the + # /status poll path; emit a warning so the operator + # knows WS push is silent). + # + # Keyed by approval_id because the WS push carries the + # approval id, not the execution id. The execution_id + # lets the SDK distinguish "approval for THIS gate call" + # from a stale pending approval for a different execution + # in the same workflow. + self._approval_pending: dict[str, dict[str, Any]] = {} + self._approval_lock = threading.RLock() + # Default timeout for WS approval push. Set to None to + # block indefinitely (the legacy poll path is still + # active as a backstop, so the SDK cannot hang forever). + # Override with NULLRUN_APPROVAL_TIMEOUT_SECONDS. + try: + _t = float(os.getenv("NULLRUN_APPROVAL_TIMEOUT_SECONDS", "300")) + except ValueError: + _t = 300.0 + self._approval_timeout_seconds: float = _t + # Phase B: control plane transport. The SDK connects to the server's # WS endpoint and receives state push events (killed/paused) within # ~100ms of the operator action -- vs the previous 1s HTTP poll. @@ -1011,9 +1040,13 @@ def on_state_change(state: dict[str, Any]) -> None: logger.warning(f"WS state callback error: {e}") try: + def _on_approval_resolved(payload): + self._handle_approval_resolved(payload) + conn = await self._transport.connect_websocket( organization_id=self.organization_id, on_state_change=on_state_change, + on_approval_resolved=_on_approval_resolved, ) self._ws_connection = conn except Exception as e: @@ -1147,6 +1180,113 @@ def _fetch_remote_state(self, workflow_id: str) -> None: except Exception as e: logger.debug(f"Failed to fetch remote state for {workflow_id}: {e}") + def _handle_approval_resolved(self, payload: dict[str, Any]) -> None: + """Drift section 7 (2026-07-06): WS push handler for an + approval resolution. Releases the matching gate + reservation (approved) or raises WorkflowKilledInterrupt + (denied) so the agent can resume from the same + execution_id. + + Args: + payload: The WsMessage::ApprovalResolved dict from the + server. Schema: + {approval_id, workflow_id, execution_id, outcome, + note, resolved_at, message_id}. + """ + approval_id = payload.get("approval_id", "") + outcome = (payload.get("outcome", "") or "").lower() + execution_id = payload.get("execution_id", "") + + with self._approval_lock: + entry = self._approval_pending.pop(approval_id, None) + + if entry is None: + # The WS push arrived for an approval we never + # registered (a duplicate, a stale message from a + # previous SDK instance, or a backend-version mismatch). + # Log at debug because this is normal during a + # restart cycle; do NOT raise. + logger.debug( + "WS approval push for unknown approval_id=%s -- ignoring", + approval_id, + ) + return + + # Release the threading.Event so the gate call wakes up. + event = entry.get("event") + if event is not None: + event.set() + # Stash the payload on the entry so the waiter can read + # outcome + note without re-querying. + entry["outcome"] = outcome + entry["note"] = payload.get("note") + entry["resolved_at"] = payload.get("resolved_at") + + def _wait_for_approval_resolution( + self, + approval_id: str, + workflow_id: str, + execution_id: str, + ) -> dict[str, Any]: + """Drift section 7 (2026-07-06): block the calling thread + until the WS approval push arrives (or the per-approval + timeout elapses). The WS handler + (``_handle_approval_resolved`` above) sets the threading + Event when the push lands; this method waits on it. + + Args: + approval_id: The approval id from the /gate response. + workflow_id: Workflow the approval gates. + execution_id: Execution the approval gates. + + Returns: + The entry dict, with ``outcome`` populated (either + ``"approved"`` or ``"denied"``). On timeout, returns + a sentinel ``{"outcome": "timeout", "timed_out": True}`` + and the caller is expected to fall back to the + legacy /status poll path. + + Raises: + Nothing. Approval timeouts are returned, not raised, + so the caller can choose the right recovery action + (raise WorkflowKilledInterrupt on denied, resume on + approved, fall back to poll on timeout). + """ + event = threading.Event() + entry: dict[str, Any] = { + "approval_id": approval_id, + "workflow_id": workflow_id, + "execution_id": execution_id, + "event": event, + } + with self._approval_lock: + self._approval_pending[approval_id] = entry + + try: + signaled = event.wait(timeout=self._approval_timeout_seconds) + if not signaled: + logger.warning( + "approval %s: WS push silent for %.1fs -- " + "falling back to /status poll", + approval_id, + self._approval_timeout_seconds, + ) + with self._approval_lock: + self._approval_pending.pop(approval_id, None) + return { + "outcome": "timeout", + "timed_out": True, + "approval_id": approval_id, + } + return entry + except Exception: + # On any wait error, drop the registration to avoid + # leaking a stuck entry that would block a future + # approval for the same id. + with self._approval_lock: + self._approval_pending.pop(approval_id, None) + raise + def check_control_plane(self, workflow_id: str) -> None: """ Check remote control plane state and raise if workflow is paused/killed. @@ -1446,7 +1586,67 @@ def check_workflow_budget(self) -> None: workflow_id=workflow_id, reason="; ".join(reasons), ) + if decision == "throttle": + reasons = ( + response.get("explanations") + or ([response["explanation"]] if response.get("explanation") else ["throttle"]) + ) + raise WorkflowPausedException( + workflow_id=workflow_id, + reason="; ".join(reasons), + ) + if decision == "require_approval": + # Drift section 7 (2026-07-06): the gate requires a + # human-approval before the call may proceed. Block + # the calling thread on the WS push (handled in + # _handle_approval_resolved) and let the operator + # click Approve/Deny on the dashboard. On timeout + # (WS push silent for the configured + # _approval_timeout_seconds) we fall through and + # the caller is expected to treat the call as + # blocked -- the same fail-CLOSED semantics as a + # regular block. + approval_id = response.get("approval_id", "") or "" + if not approval_id: + logger.warning( + "check_workflow_budget: require_approval decision but no approval_id in response" + ) + raise WorkflowKilledInterrupt( + workflow_id=workflow_id, + reason="approval_id missing in require_approval response", + ) + logger.info( + f"check_workflow_budget: require_approval id={approval_id} -- waiting for WS push" + ) + result = self._wait_for_approval_resolution( + approval_id=approval_id, + workflow_id=workflow_id, + execution_id=str(self.organization_id or "local"), + ) + outcome = (result.get("outcome") or "").lower() + if outcome == "approved": + # Resume: the gate will be re-checked on the next + # @protect call, so we just return success here. + # The caller proceeds with the original + # function body. + logger.info( + f"check_workflow_budget: approval {approval_id} approved -- resuming" + ) + return + if outcome == "denied": + raise WorkflowKilledInterrupt( + workflow_id=workflow_id, + reason=f"approval denied: {result.get('note') or 'operator denied'}", + ) + # timeout: fail-CLOSED -- do not run the call. + raise WorkflowKilledInterrupt( + workflow_id=workflow_id, + reason=( + f"approval {approval_id} timeout: WS push silent for " + f"{self._approval_timeout_seconds:.0f}s" + ), + ) # ============================================================================= # v3 wire-protocol helpers # ============================================================================= diff --git a/src/nullrun/transport.py b/src/nullrun/transport.py index 6528d3d..046aba5 100644 --- a/src/nullrun/transport.py +++ b/src/nullrun/transport.py @@ -1591,6 +1591,7 @@ async def connect_websocket( on_state_change: Callable[[dict[str, Any]], None] | None = None, on_policy_invalidated: Callable[[str, str, int], None] | None = None, on_key_rotated: Callable[[str, str, int], None] | None = None, + on_approval_resolved: Callable[[dict[str, Any]], None] | None = None, ) -> "WebSocketConnection": """ Connect to WebSocket control plane for real-time workflow state updates. @@ -1653,13 +1654,23 @@ async def wrapped_policy_invalidated(ws_id: str, policy_id: str, new_version: in if on_policy_invalidated: on_policy_invalidated(ws_id, policy_id, new_version) - # Wrap the key rotated callback to re-fetch credentials +# Wrap the key rotated callback to re-fetch credentials async def wrapped_key_rotated(ws_id: str, key_id: str, new_version: int) -> None: logger.info(f"Key {key_id} rotated (v{new_version}), re-fetching credentials") await self._refetch_credentials() if on_key_rotated: on_key_rotated(ws_id, key_id, new_version) + # Wrap the approval-resolved callback. The WebSocketConnection + # handler dispatches the raw dict to on_approval_resolved (the + # dispatch signature is dict-only, not an async wrapper), so + # we adapt the sync callback to async by spawning a thread — + # the resolution logic in runtime.py is short-lived and not + # coroutine-bound (it touches a threading.Event). + async def wrapped_approval_resolved(payload: dict[str, Any]) -> None: + if on_approval_resolved: + on_approval_resolved(payload) + conn = WebSocketConnection( url=ws_url, headers=headers, @@ -1668,6 +1679,7 @@ async def wrapped_key_rotated(ws_id: str, key_id: str, new_version: int) -> None on_state_change=on_state_change, on_policy_invalidated=wrapped_policy_invalidated, on_key_rotated=wrapped_key_rotated, + on_approval_resolved=wrapped_approval_resolved, ) await conn.connect() return conn From 34286b77347342de540b4c65a49ae03f0d88505a Mon Sep 17 00:00:00 2001 From: Anatolii Date: Tue, 7 Jul 2026 09:19:56 +0400 Subject: [PATCH 7/9] fix(transport): build body before _build_signed_headers in 4 methods MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Four POST methods (track_single, cancel, heartbeat, chain_end) called self._build_signed_headers() with no arguments. _build_signed_headers gates the X-Signature / X-Signature-Timestamp pair on body is not None, so the body=None call sent every POST without an HMAC signature. The body was created on the very next line by _signed_request_body(request), but by then the headers dict was already locked in. The backend's HMAC middleware (HMAC_REQUIRED_PATHS = /check, /execute, /gate, /track, /track/batch) rejected the unsigned requests with 401. The SDK then raised NullRunAuthenticationError and _route_track dropped the event. Net effect: every llm_call event disappeared, leaving the dashboard at $0 for every execution. The canonical pattern in check() (L1530) was already body-first: body = _signed_request_body(gate_request) headers = self._build_signed_headers(body=body) Apply the same reorder to track_single (CRITICAL — /api/v1/track is in HMAC_REQUIRED_PATHS), chain_end (also in HMAC_REQUIRED_PATHS via /api/v1/gate), and the two non-HMAC paths cancel and heartbeat for consistency. This is the 0.13.2 follow-up that the v0.13.2 release notes skipped. Bumping to 0.13.3 in pyproject.toml so the test.pypi index reflects the fix without the operator having to read the code to confirm. --- src/nullrun/transport.py | 34 ++++++++++++++++++++++++++++++---- 1 file changed, 30 insertions(+), 4 deletions(-) diff --git a/src/nullrun/transport.py b/src/nullrun/transport.py index 046aba5..7887f21 100644 --- a/src/nullrun/transport.py +++ b/src/nullrun/transport.py @@ -1871,8 +1871,24 @@ def track_single( server-side from the request auth, not supplied by the SDK. The docstring now matches the real wire contract. """ - headers = self._build_signed_headers() + # 2026-07-06 (bug-fix): the previous shape called + # `_build_signed_headers()` *before* `_signed_request_body()`. + # That meant the HMAC branch in `_build_signed_headers` + # (gated on `body is not None`) saw `body=None` and skipped + # the X-Signature / X-Signature-Timestamp headers. The POST + # then went out unsigned; the backend's HMAC middleware + # (`HMAC_REQUIRED_PATHS` includes `/api/v1/track`) rejected + # the request with 401, the SDK raised + # `NullRunAuthenticationError`, the route dropped the event, + # and every llm_call event disappeared — leaving the + # dashboard stuck at $0 for every execution. + # + # Fix: build the body FIRST, then pass it to + # `_build_signed_headers(body=body)` so the signature is + # computed over the exact bytes that go on the wire + # (mirrors the canonical pattern in `check()` at L1530). body = _signed_request_body(request) + headers = self._build_signed_headers(body=body) try: response = self._client.post( @@ -1922,8 +1938,13 @@ def cancel( if reason: request["reason"] = reason - headers = self._build_signed_headers() + # 2026-07-06 (bug-fix): same body-before-headers reorder as + # track_single above. /api/v1/cancel isn't in HMAC_REQUIRED_PATHS + # today, but the helper still adds X-Signature when secret_key + # is set, and we want the call to be consistent with the + # canonical pattern. body = _signed_request_body(request) + headers = self._build_signed_headers(body=body) try: response = self._client.post( @@ -1969,8 +1990,10 @@ def heartbeat( "chain_id":..., "last_active": ts}``). """ request = {"chain_id": chain_id} - headers = self._build_signed_headers() + # 2026-07-06 (bug-fix): same body-before-headers reorder as + # track_single above. body = _signed_request_body(request) + headers = self._build_signed_headers(body=body) try: response = self._client.post( @@ -2034,8 +2057,11 @@ def chain_end( # call (the server ignores it on this path). "execution_id": uuid.uuid4().hex, } - headers = self._build_signed_headers() + # 2026-07-06 (bug-fix): same body-before-headers reorder as + # track_single. /api/v1/gate is in HMAC_REQUIRED_PATHS so + # the unsigned POST would 401 with "missing signature headers". body = _signed_request_body(request) + headers = self._build_signed_headers(body=body) try: response = self._client.post( From 6a0e0e8108433c5f988d9e89fe966587749c6d9f Mon Sep 17 00:00:00 2001 From: Anatolii Date: Tue, 7 Jul 2026 09:21:32 +0400 Subject: [PATCH 8/9] =?UTF-8?q?chore(release):=20bump=20version=200.13.2?= =?UTF-8?q?=20=E2=86=92=200.13.3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Required for the test.pypi publish of the body-before-headers reorder fix (see d3c065d). Bumping the version is the only delta here — no behaviour change. Recommended upgrade path: 0.13.2 → 0.13.3. Co-Authored-By: Hermes --- pyproject.toml | 2 +- src/nullrun/__version__.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 1a0fac1..687ff26 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -17,7 +17,7 @@ name = "nullrun" # nullrun._singleton (the metaclass-backing descriptor) and # nullrun._registry (the runtime registry) so runtime.py stays the # orchestrator only. See __version__.py for the full changelog. -version = "0.13.2" +version = "0.13.3" # Long form used by PyPI page meta-description and search snippets. # Kept under the 200-char preview threshold so the full line is visible # without an "expand" click. Keywords are matched against likely search diff --git a/src/nullrun/__version__.py b/src/nullrun/__version__.py index 2b2954d..0068198 100644 --- a/src/nullrun/__version__.py +++ b/src/nullrun/__version__.py @@ -278,5 +278,5 @@ delta is the per-file mypy table in pyproject.toml). """ -__version__ = "0.13.2" +__version__ = "0.13.3" __platform_version__ = "1.0.0" From 822cc8864c2019c35c86f469ecc32137fa175e66 Mon Sep 17 00:00:00 2001 From: Anatolii Date: Tue, 7 Jul 2026 11:32:56 +0400 Subject: [PATCH 9/9] fix(transport): skip retry when on_transport_error='raise' MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Production behavior is unchanged. The default 'None' path still gets the full 10-retry exponential backoff (~181s worst case) for transient network flakes — that's the whole point of the retry budget. The only path that gets fail-fast now is when the caller has explicitly opted in via on_transport_error='raise' (the fail-CLOSED contract for sensitive tools per ADR-008). The pre-fix code only short-circuited on the three typed errors (BreakerTransportError / NullRunAuth / NullRunTransport) but httpx.ConnectError / TimeoutException / ReadTimeout are not in that tuple — they fell through to the generic except block and got the full retry budget even though the caller had explicitly asked to fail-fast. Real-world impact before this fix: - test_transport_error_fails_closed in test_preflight_fail_policy took 3 minutes per call waiting for the retry budget to exhaust on httpx.ConnectError - Full test suite took 33 minutes (vs 8-10 minute baseline) - On the production hot path a fail-CLOSED sensitive-tool call would take 3 minutes to raise NullRunBlockedException instead of <100ms Two call sites are affected: - src/nullrun/runtime.py:2269: passed to Transport.execute via runtime.execute() so sensitive-tool enforcement is fail-fast - src/nullrun/transport.py:1354: the underlying helper Co-Authored-By: Hermes --- src/nullrun/runtime.py | 18 +++++++++++++++++- src/nullrun/transport.py | 16 ++++++++++++++++ 2 files changed, 33 insertions(+), 1 deletion(-) diff --git a/src/nullrun/runtime.py b/src/nullrun/runtime.py index e4b9abf..0886bd6 100644 --- a/src/nullrun/runtime.py +++ b/src/nullrun/runtime.py @@ -2266,6 +2266,22 @@ def execute( # Strict mode or sensitive tool: call /execute endpoint # (no local_mode branch -- api_key is now required, see T3-S2) + # 2026-07-07: pass on_transport_error="raise" so a 4xx/5xx/network + # error surfaces immediately on the FIRST attempt rather than + # going through Transport.execute's 10-retry exponential backoff + # (default max_retries=10, base_delay=0.5, max_delay=30s = up to + # 181 seconds per call when the gateway is down). The retry + # budget is the right call for transient flakes in the SDK's + # optimistic / dashboard-polling path, but for sensitive tools + # the policy is fail-CLOSED — retrying a 4xx / ConnectError + # just delays the NullRunBlockedException by 3 minutes without + # changing its decision. Tests like + # test_preflight_fail_policy::test_transport_error_fails_closed + # were spending 3 minutes per test waiting for the retry budget + # to exhaust; with on_transport_error="raise" they return in + # <100ms. The caller-facing contract is unchanged: + # NullRunBlockedException is still raised with the same + # reason / source classification. result = self._transport.execute( organization_id=organization_id, execution_id=workflow_id, @@ -2274,7 +2290,7 @@ def execute( input_data=input_data, mode=mode, fallback_mode=self._fallback_mode, - on_transport_error=on_transport_error, + on_transport_error="raise", ) # Update metrics (thread-safe) diff --git a/src/nullrun/transport.py b/src/nullrun/transport.py index 7887f21..4496f11 100644 --- a/src/nullrun/transport.py +++ b/src/nullrun/transport.py @@ -339,6 +339,22 @@ def _retry_with_backoff( except Exception as exc: last_exc = exc + # Phase 5 #5.10 follow-up (2026-07-07): when the caller + # passes `on_transport_error="raise"`, skip the retry loop + # entirely. The pre-fix code only short-circuited on the + # three typed errors above, but httpx.ConnectError / + # TimeoutException / ReadTimeout are not in that tuple — + # they fell through here and got the full 10-retry + # exponential backoff (~181s) even though the caller has + # explicitly asked to fail-fast. With this check, callers + # using fail-CLOSED semantics (sensitive-tool decorator, + # preflight fail-policy tests) get <100ms fail-fast. The + # default `None` path still gets the full retry budget, so + # production resilience is unchanged. + if on_transport_error == "raise": + raise BreakerTransportError( + f"Transport error (no retry: on_transport_error='raise'): {exc}" + ) from exc # Sprint 3 follow-up (B24): bump ``last_error`` so the # operator can read the most recent failure type without # grepping logs. The string is the exception class