-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathruntime.py
More file actions
3547 lines (3173 loc) · 154 KB
/
Copy pathruntime.py
File metadata and controls
3547 lines (3173 loc) · 154 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
"""
NullRun Runtime - core runtime safety layer for AI agents.
This is the main entry point for the SDK. It handles:
- Authentication with NullRun cloud
- Policy fetching and caching
- Event buffering and batched flush
- Local policy enforcement (instant, no network latency)
- Pre-execution enforcement via /execute endpoint
## Pre-execution gate fail-OPEN/CLOSED contract (ADR-008)
The SDK enforces workflow safety through a set of *pre-execution gates*
that run before a protected function body executes and may raise to halt
the work. Each gate declares its own fail-OPEN/CLOSED policy -- this is
the authoritative table; deviations require an ADR amendment (Rule 5).
| Gate | Transport-error behavior | Recovery behavior | Opt-out |
|---|---|---|---|
| `check_workflow_budget` | OPEN (skip check, log warning) | silent post-hoc correction in `/track` events via `cost_correction_applied=true` | `NULLRUN_SKIP_BUDGET_CHECK=1` -- **full billing bypass**, not just check bypass (see docstring WARNING) |
| `check_control_plane` | OPEN (treat state as `Normal`) | deferred enforcement -- next WS-push or `/status` poll sees the true state | none |
| `_enforce_sensitive_tool` (default `_fallback_mode=permissive`) | CLOSED -- body MUST NOT run when `decision_source` is any `FALLBACK_*` | n/a (body did not run) | `NULLRUN_SENSITIVE_FAIL_OPEN=1` -- explicitly documented as "OPEN-when-engine-unavailable" |
| `_enforce_sensitive_tool` (`_fallback_mode=strict`) | CLOSED -- transport returns `decision=block, decision_source=FALLBACK_*` | n/a | none |
| `_emit_span_start` / `_emit_span_end` | n/a -- never blocks | n/a | n/a |
| `/track` batch path (legacy) | OPEN-on-network-error (event dropped, no retry) | n/a -- circuit breaker backoff applies | none |
**Drift fix 2026-07-04:** the SDK_README.md claim
"Fail-OPEN на инфраструктурных сбоях. Если backend недоступен, бюджет
не блокирует агента" is **partially wrong** — it conflates SDK-side
transport failure with backend-side budget-enforcement failure. The
honest split is:
* **SDK-side transport failure** (network timeout, 5xx, breaker open)
→ fail-OPEN on the *check* path so a dead backend doesn't freeze
the user's agent loop (this is what the README describes).
* **Backend-side budget-enforcement failure** (the /gate or /track
handler actually returned a wire response, just one indicating a
Redis outage or aggregate rate-limit Redis unavailable) → the
wire response is what it is, and the SDK raises the corresponding
exception. ``BUDGET_REDIS_UNAVAILABLE`` → 402 ``NullRunBudgetError``
(fail-CLOSED, the backend rejected the request because Redis was
unreachable for the budget counter — this is the authoritative
enforcement signal, not a transport blip). ``RATE_LIMIT_REDIS_UNAVAILABLE``
→ 503 ``NullRunRateLimitRedisError`` (fail-CLOSED for the same
reason). The SDK does NOT silently fall-OPEN on a wire 4xx/5xx
that names an enforcement failure.
The table above is authoritative; if any of these change, the
README claim must be updated in lockstep.
The "Opt-out" column makes it explicit that `NULLRUN_SKIP_BUDGET_CHECK=1`
is a **different category** of action than
`NULLRUN_SENSITIVE_FAIL_OPEN=1` (bypass vs. change semantics), despite
the similar naming. See `docs/adr/008-sdk-preflight-fail-policy.md`
for the full rules, including transport error classification
(`FALLBACK_NETWORK_ERROR` / `FALLBACK_GATEWAY_ERROR` / `FALLBACK_BREAKER_OPEN`).
"""
import asyncio
import logging
import os
import threading
import time
import uuid
import warnings
from collections.abc import Callable
from typing import Any, Optional
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,
)
from nullrun.context import (
generate_span_id,
generate_trace_id,
get_agent_id,
get_attempt_index,
get_span_id,
get_trace_id,
get_workflow_id,
)
from nullrun.observability import metrics
from nullrun.transport import (
HEADER_PROTOCOL,
NULLRUN_PROTOCOL_VERSION,
DecisionSource,
FallbackMode,
FlushConfig,
Transport,
TransportErrorSource,
_emit_for_transport_error,
_protocol_header_value,
)
from nullrun.uuid7 import uuid7_str # 2026-07-04 BUG #4
logger = logging.getLogger(__name__)
# Phase 0.3.1: sentinel used when a gate fires outside a
# ``with workflow(...)`` context. The double-underscore prefix
# namespacing avoids collision with a user workflow that happens
# to be named ``<unknown>`` (the previous literal was a
# collision hazard). Wire compat: still a string.
UNKNOWN_WORKFLOW_ID: str = "__nullrun_unknown__"
# 2026-07-04 (BUG #5): in-process gate cache for chain-mode
# invocations. Without this, every @protect inside `with chain(...)`
# issues a /gate HTTP roundtrip + Redis reserve. For a 100-step
# agent loop that's 100 roundtrips. The gate decision is
# deterministic for a given (workflow_id, chain_id, model) over a
# short window (chain status only changes on `chain_end`), so
# caching the LAST decision for 5s is safe.
#
# Scope: ONLY when chain_id is set. Single-shot (Hard) callers
# must NOT cache — the gate legitimately returns "allow" once and
# "block" on the next call (Hard mode binary), and a stale "allow"
# could let through a budget-exhausted call. Chain-mode callers
# share a budget envelope, so caching "allow" is consistent with
# the chain's semantics.
#
# Opt-out: NULLRUN_GATE_CACHE_DISABLE=1
_GATE_CACHE: dict[tuple[str, str | None, str | None], tuple[float, dict[str, Any]]] = {}
_GATE_CACHE_TTL_SECONDS: float = 5.0
# 2026-07-24 (Root-cause fix for the ``@sensitive`` reinit gap):
# a process-level set of tool names that the ``@sensitive``
# decorator has stamped as needing strict mode. The runtime
# singleton also tracks this via ``_sensitive_tools``, but
# that set is populated at decoration time and can be lost
# across ``init_or_die()`` calls if the user re-initializes
# the runtime (the registration landed on the OLD instance
# and the new instance starts with an empty set). The
# module-level set is decorator-driven and survives any
# runtime singleton churn, so ``is_strict_mode_forced`` is
# the second source of truth that ``runtime.execute`` consults
# before falling through to inline mode.
_STRICT_MODE_FORCED: set[str] = set()
def register_strict_mode_forced(tool_name: str) -> None:
"""Mark ``tool_name`` as needing strict mode.
Called by ``@sensitive(impact=...)`` at decoration time. The
name stays in the module-level set until process exit; it
is intentionally not cleared by ``init_or_die()`` so a
second-runtime reinit does not silently drop a tool out of
strict mode.
"""
_STRICT_MODE_FORCED.add(tool_name)
def is_strict_mode_forced(tool_name: str) -> bool:
"""Return True if ``tool_name`` was decorated with ``@sensitive``.
Complements ``runtime.is_sensitive_tool(tool_name)`` which
reads the per-runtime registry. The two are OR'd in
``runtime.execute`` so that a tool whose registration is
lost to runtime reinit still gets the strict /execute
round-trip it asked for.
"""
return tool_name in _STRICT_MODE_FORCED
# 2026-07-04 (v0.12.0 wiring fix — ):
# the maximum age (seconds) for a captured ``reservation_id``
# to be eligible for forwarding onto a /track payload. Past
# this age the underlying ``reservation:{execution_id}`` Redis
# key has expired (300s TTL per) — forwarding would
# guarantee a 503 ``RESERVATION_NOT_FOUND`` on /track. The
# 5s margin below the 300s TTL absorbs clock-skew between
# the SDK's ``time.monotonic `` and the Redis cluster's own
# TTL decay (sub-second typically, but the safety budget is
# worth the simplicity of a hard-coded threshold).
SERVER_MINTED_RESERVATION_MAX_AGE_SECONDS: float = 295.0
# Phase 0 review (2026-07-23): hard cap on server-supplied
# approval_timeout_seconds. The backend is authoritative for the
# approval window, but a misconfigured backend (or a malicious
# proxy in front of one) could advertise an absurdly long
# timeout (e.g. 1e9 seconds) and lock the calling thread
# indefinitely. We clamp the server value to this ceiling as
# the maximum time we'll ever wait for an operator click.
# The env default `NULLRUN_APPROVAL_TIMEOUT_SECONDS` is also
# clamped — see check_workflow_budget / runtime.execute for the
# exact clamp call.
MAX_APPROVAL_TIMEOUT_SECONDS: float = 3600.0
# Symmetric floor: a 0-second or negative timeout would
# deadlock the very first event.wait() call. The server is
# allowed to advertise any value in `[1, MAX]`; out-of-range
# values fall back to the env default.
MIN_APPROVAL_TIMEOUT_SECONDS: float = 1.0
def _validate_approval_timeout(value: object, log_prefix: str) -> float | None:
"""Validate and clamp a server-supplied approval_timeout_seconds.
Phase 0 review (2026-07-23): the server is authoritative for the
approval window, but it could advertise 0 (deadlock), 1e9 (lock the
thread for years), a non-numeric string, or `None`. We refuse to
forward anything outside `[MIN_APPROVAL_TIMEOUT_SECONDS,
MAX_APPROVAL_TIMEOUT_SECONDS]` and return `None` so the caller
falls back to `_approval_timeout_seconds` (the env default).
Args:
value: the raw `approval_timeout_seconds` field from the
server's wire response (may be int, float, str, None,
or anything else if a future backend drifts).
log_prefix: short caller name for the WARN log (e.g.
"check_workflow_budget" or "runtime.execute").
Returns:
A positive float in `[MIN, MAX]`, or `None` to signal
"fall back to the env default".
"""
if value is None:
return None
try:
candidate = float(value)
except (TypeError, ValueError):
logger.warning(
"%s: approval_timeout_seconds=%r is not a number; falling back to env default",
log_prefix,
value,
)
return None
if candidate < MIN_APPROVAL_TIMEOUT_SECONDS or candidate > MAX_APPROVAL_TIMEOUT_SECONDS:
logger.warning(
"%s: approval_timeout_seconds=%.1fs out of range [%.1f, %.1f]; falling back to env default",
log_prefix,
candidate,
MIN_APPROVAL_TIMEOUT_SECONDS,
MAX_APPROVAL_TIMEOUT_SECONDS,
)
return None
return candidate
# Phase 4.1: privacy boundary. Fields that MUST NOT leave the SDK on
# the wire. The transport layer (POST /api/v1/track/batch) reads
# whatever is in the event dict, so anything not allowlisted ends up
# in the user's audit log on the backend side. We strip:
#
# * ``cost_cents`` -- the SDK does not estimate cost; the backend
# recomputes it from tokens + the org's pricing policy. Sending
# a wrong number risks double-billing when the backend also
# persists its own computed cost.
# * ``_fingerprint`` -- the dedup key (sha256[:16] over the raw
# response body). Process-local; leaking it to audit logs
# would let an operator with audit-log read access fingerprint
# which prompts went through dedup, defeating the purpose.
# * ``raw_usage`` -- the vendor's full usage dict (OpenAI
# ``prompt_tokens_details``, Anthropic ``cache_*_input_tokens``
# etc.) — Phase 4.1 moved every field we care about out of
# raw_usage onto the event itself, so the original dict is now
# just an opaque blob of provider-specific data. Carrying it on
# the wire is a privacy regression: provider response payloads
# can include user-supplied metadata, organization names, or
# other PII the backend has no business logging.
#
# Anything new added here MUST also be added to the in-process
# callers that consume these fields (the dedup LRU at
# ``_seen_track_fingerprints``, any local loggers).
_WIRE_STRIP_FIELDS: frozenset[str] = frozenset({"cost_cents", "_fingerprint", "raw_usage"})
# 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.
This is a singleton that manages:
- Authentication state (organization_id)
- Cached policies from backend
- Event buffering and batched transport
- Local policy enforcement
Usage:
# Automatic (via protect )
import nullrun
nullrun.protect
# Manual
rt = NullRunRuntime.get_instance
# Note: `cost_cents` is NOT a valid event key — the SDK strips
# it before sending (see ``track_event`` / wire payload below).
# The backend computes cost from tokens + the org's pricing
# policy. Use ``tokens`` (or, for llm_call specifically
# ``input_tokens`` / ``output_tokens``) to feed cost math.
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.
_lock = threading.Lock()
def __init__(
self,
api_key: str | None = None,
secret_key: str | None = None,
api_url: str = "https://api.nullrun.io",
fallback_mode: str | None = None,
debug: bool = False,
_test_mode: bool = False,
polling: bool = True,
):
"""
Initialize NullRun Runtime.
Args:
api_key: API key from NullRun dashboard. If None, reads from
NULLRUN_API_KEY env variable. If both None, uses local mode.
secret_key: Secret key for HMAC request signing. If None, no signing.
api_url: URL of NullRun proxy server. Defaults to https:/api.nullrun.io.
debug: Enable debug logging.
_test_mode: Internal flag to skip network calls (for testing).
polling: Internal flag for tests/CI to skip the background
control-plane listener (WS or HTTP poll). Defaults True
in production. Set False when the test environment
cannot tolerate a background thread opening sockets.
Note:
- `organization_id` is set from `_authenticate ` after init; it is
NOT a public init parameter and not read from env.
- `api_key` is required as of 0.3.0 (T3-S2). The previous
`local_mode` flag was removed because it silently bypassed
every backend gate.
- `fallback_mode` is fixed at PERMISSIVE (no public override).
- `timeout`/`max_retries` are fixed at 30s / 3 (no public override).
Raises:
NullRunAuthenticationError: if neither `api_key` nor
`NULLRUN_API_KEY` is set. The public `init ` surface
performs the same check first and produces a clearer
error message; this constructor-level raise is the
direct fallback for tests and advanced callers that
build the runtime by hand.
"""
self.api_key = api_key or os.getenv("NULLRUN_API_KEY")
self.secret_key = secret_key or os.getenv("NULLRUN_SECRET_KEY")
self.api_url = api_url or os.getenv("NULLRUN_API_URL", "https://api.nullrun.io")
# T3-S2 (0.3.0): api_key is now required. The previous `local_mode`
# flag silently bypassed every backend gate (budget, policy
# control plane), which was a real safety hole in production.
# We raise NullRunAuthenticationError here instead so the
# misconfiguration is caught at startup. The public `init `
# surface raises first with a clearer message; this is the
# direct construction path used by tests and advanced callers.
if not self.api_key:
raise NullRunAuthenticationError(
"NullRunRuntime() requires an api_key. Pass api_key='nr_live_...' "
"or set NULLRUN_API_KEY. (Silent no-op fallback was removed "
"in 0.3.0 -- see CHANGELOG.)"
)
# organization_id is set by _authenticate; stays None until then.
self.organization_id: str | None = None
# Phase 139+: workflow_id is set by _authenticate from the API
# key's binding (organization_api_keys.workflow_id). Used as a
# fallback for /check, /status, and span events when the user
# hasn't entered a `with workflow(...)` context. None on legacy
# keys (pre-139 or never used) -- call sites must NOT invent one.
self.workflow_id: str | None = None
self._test_mode = _test_mode
self.polling = polling
# The string ``fallback_mode`` parameter is deprecated and
# accepted only for backward compat — the CACHED variant
# was removed in 0.7.0 because the SDK no longer maintains
# a local policy cache (see CHANGELOG D-01).
fb_upper = str(fallback_mode).upper() if fallback_mode is not None else "PERMISSIVE"
if fb_upper == "STRICT":
self._fallback_mode = FallbackMode.STRICT
else:
self._fallback_mode = FallbackMode.PERMISSIVE
self._timeout = 30
self._max_retries = 3
self._debug = debug
self._transport: Transport | None = None
# Local enforcement state
# Phase 0.3.1: the BoundedDict-based per-workflow cost /
# loop / retry counters have been removed alongside
# ``_check_local_limits``. As of 0.7.0 ALL local
# enforcement (LoopTracker / RateTracker / _local_check /
# hardcoded thresholds) has been removed -- the SDK is a
# thin client, the backend is authoritative.
self._workflow_start_time: float = time.time()
# Layer 3: ring buffer for the ``nullrun.status `` recent
# errors list. Capacity 10 — bounded so a long-lived process
# does not leak memory even if the SDK raises thousands of
# errors per minute. Fed by ``_record_error`` (called from
# ``_emit_sdk_error`` after the Layer-2 ``emit_error``).
from nullrun.observability.status import _RecentErrorRing
self._recent_errors = _RecentErrorRing(capacity=10)
# Layer 3: backend connectivity timestamps for the status
# snapshot. Set in ``_authenticate`` and updated on every
# successful / failed backend call thereafter.
self._last_backend_attempt_at: float | None = None
self._last_backend_attempt_ok: bool | None = None
# Phase D: dedup LRU. Multiple observation paths (httpx transport
# LangChain callback, OpenAI Agents tracer) can fire for the same
# LLM call. We collapse them to a single track per fingerprint.
# The fingerprint is computed at the observation point and passed
# via the `_fingerprint` event field.
from nullrun.instrumentation.auto import make_dedup_state
self._seen_track_fingerprints = make_dedup_state()
# Per ADR-008 the SDK does not track local cost. The two response
# fields below are kept in the return shape for backwards
# compatibility with 0.3.x callers but always read 0. The previous
# implementation read from `self._workflow_costs` (a BoundedDict
# removed in 0.3.1) which left `track ` raising AttributeError on
# first call.
self._local_cost_cents_estimate: int = 0
# 0.9.0: coverage counters removed. Coverage is now derived
# server-side from the llm_call span metadata (`tracked` and
# `streaming_skipped` flags set by the instrumentation layer).
# The previous per-host dicts and 60s daemon thread are gone.
# Remote control plane state (per-workflow, pushed from server via WS).
# Unified model: effective_state = max(local_state, remote_state).
# All writes and reads go through the `_remote_state_for` /
# `_set_remote_state` helpers (Phase 5 #5.1) so the WS callback
# the HTTP poll, and the gate check can run concurrently
# without a TOCTOU race. RLock because the same thread can
# re-enter via the gate's get-then-set sequence.
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.
# The HTTP poll path is preserved as a fallback when
# `NULLRUN_TRANSPORT=http` is set (env var defaults to `ws`).
self._transport_mode: str = os.getenv("NULLRUN_TRANSPORT", "ws").lower()
self._ws_thread: threading.Thread | None = None
self._ws_stop_event = threading.Event()
self._ws_connection: Any = None # WebSocketConnection; typed loosely to avoid import cycle
self._ws_loop: Any = None # asyncio loop running in the WS thread
# Legacy HTTP-poll state -- only used when transport mode is `http`.
self._poll_thread: threading.Thread | None = None
self._poll_running = False
# Action handling
self._action_handler: ActionHandler | None = None
# Initialize transport FIRST (before auth/policy) so we can reuse its client
# Transport will be started later after auth/policy succeed
self._transport = Transport(
api_url=self.api_url,
api_key=self.api_key,
secret_key=self.secret_key,
config=FlushConfig(
batch_size=50,
flush_interval=5.0,
),
)
# Note: a gRPC transport was prototyped in earlier SDK versions but the
# gRPC server at the platform is intentionally frozen until the
# activation checklist (TLS, auth, proto extensions, cost pipeline
# parity, tests) is complete. The SDK no longer attempts to construct
# a gRPC client.
# FIX 2026-06-28: was a silent no-op (logger.info) — customers who
# set NULLRUN_USE_GRPC expecting gRPC silently fell back to HTTP with
# no signal. Now we raise loudly so the misconfiguration is visible
# at startup instead of being diagnosed from a missing proto trace.
if os.getenv("NULLRUN_USE_GRPC"):
raise RuntimeError(
"NULLRUN_USE_GRPC is set but the gRPC transport is not "
"yet implemented. This option is reserved for a future "
"release. Unset the env var to use the HTTP transport. "
"See https://docs.nullrun.io/reference/sdk-api#transport"
)
# Initialize
if self._test_mode:
# Test mode: skip all network calls
self._transport.start()
else:
try:
self._authenticate()
except NullRunAuthenticationError:
raise # Re-raise auth errors immediately - don't continue in unprotected mode
except httpx.RequestError as e:
raise NullRunAuthenticationError(
f"Auth request failed: {e}. Cannot establish secure connection to NullRun. "
f"Refusing to operate in unprotected mode."
) from e
self._transport.start()
# Start remote polling unless disabled (internal `polling=False`
# for tests/CI). Production always polls.
if self.polling:
self._start_remote_polling()
# Initialize action handler
self._action_handler = ActionHandler()
# 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] = {
# Financial operations
"stripe.charge",
"stripe.refund",
"stripe.payout",
"payment.process",
# Email / communication
"send_email",
"send_sms",
"send_slack",
"send_discord",
# Database operations
"db.delete",
"db.drop",
"db.truncate",
"db.write",
# External API calls
"api.post",
"api.put",
"api.delete",
# File operations
"file.delete",
"file.write",
"s3.delete",
# Admin operations
"admin.delete",
"admin.create_user",
"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
# ``add_sensitive_tool`` without holding any lock; the
# reader in ``is_sensitive_tool`` (line 1270-ish) did
# ``tool_name in self._strict_mode_tools`` without a lock.
# Under CPython's GIL the set mutation is atomic at the
# bytecode level, but the snapshot you read can still be
# stale mid-mutation (a single-threaded read can see the
# new value fine, but a multi-threaded read can race with
# a concurrent ``add`` if both interleave on a free-threaded
# build). The lock is uncontended on the read path so the
# cost is one acquire per call.
self._tools_lock = threading.Lock()
logger.info("NullRun Runtime initialized: mode=cloud")
@classmethod
def get_instance(cls) -> "NullRunRuntime":
"""Get the singleton runtime instance.
Thread-safe: the singleton lock is held for the full read-compare-
rebuild sequence (Phase 5 #5.3). The previous version dropped the
lock between shutdown and the recursive get_instance, creating a
window where a concurrent caller could observe a half-shutdown
runtime.
"""
with cls._lock:
# Re-read env vars at every call site so credential rotation
# is observed on the next get_instance invocation.
api_key = os.getenv("NULLRUN_API_KEY")
api_url = os.getenv("NULLRUN_API_URL", "https://api.nullrun.io")
if cls._instance is None:
cls._instance = cls(api_key=api_key, api_url=api_url)
return cls._instance
existing = cls._instance
key_changed = api_key != existing.api_key
url_changed = api_url != existing.api_url
if key_changed or url_changed:
logger.info(
f"Credentials changed: api_key={'***' if key_changed else 'unchanged'}, "
f"api_url={'changed' if url_changed else 'unchanged'} - reinitializing"
)
existing.shutdown()
cls._instance = cls(api_key=api_key, api_url=api_url)
return cls._instance
return cls._instance
@classmethod
def reset_instance(cls) -> None:
"""Reset the singleton. Mainly for testing."""
with cls._lock:
if cls._instance is not None:
cls._instance.shutdown()
cls._instance = None
def status(self) -> "Any":
"""Build a Layer-3 ``NullRunStatus`` snapshot.
Synchronous, thread-safe, side-effect-free — safe to
call from the agent loop, the transport flush thread
or a debug console. The returned dataclass is frozen
so it can be cached, shared, and compared with ``==``.
State-derivation rules (see
``nullrun/observability/status.py`` for the full
rationale):
* ``misconfigured`` — no api_key, or runtime never
bound to an org.
* ``offline`` — backend not reachable AND no cached
policy. SDK is running in strict-local fallback.
* ``degraded`` — using cached policy, OR WS
disconnected, OR circuit breaker open, OR workflow
state != Normal. SDK is operating with reduced
guarantees.
* ``ok`` — everything healthy.
"""
from datetime import datetime, timezone
from nullrun.observability.status import (
STATE_DEGRADED,
STATE_MISCONFIGURED,
STATE_OFFLINE,
STATE_OK,
NullRunStatus,
RecentError,
WorkflowState,
)
# --- Auth state ---
api_key_valid: bool | None = None
api_key_prefix: str | None = self.api_key[:10] if self.api_key else None
if self.organization_id is not None:
# If we have an org, auth at least started — it
# may have failed (we'd be in misconfigured), but
# in the normal flow org binding means a 200 came
# back from /auth/verify.
api_key_valid = True
# --- Connectivity ---
backend_reachable: bool | None = None
if self._last_backend_attempt_at is not None:
# ``_last_backend_attempt_ok`` is set to True on
# a successful HTTP response, False on a transport
# error. ``None`` if no attempt since init.
backend_reachable = self._last_backend_attempt_ok
ws_connected: bool | None = None
if self._ws_connection is not None:
# ``is_open`` is the underlying websockets flag
# None when the connection has never been
# successfully established.
ws_connected = getattr(self._ws_connection, "is_open", None)
elif self._ws_stop_event.is_set():
ws_connected = False # explicit shutdown
# --- Workflow state from last WS push ---
workflow_state: WorkflowState | None = None
if self.workflow_id is not None:
cached = self._remote_state_for(self.workflow_id)
if cached:
state_str = cached.get("state", "Normal")
workflow_state = WorkflowState(
workflow_id=self.workflow_id,
state=state_str,
version=cached.get("version", 0),
reason=cached.get("reason"),
)
# --- Recent errors ---
recent_errors = self._recent_errors.snapshot()
# --- Headline state derivation ---
# Order matters: most specific first.
if self.api_key is None or (
self.organization_id is None and self._last_backend_attempt_at is not None
):
headline = STATE_MISCONFIGURED
elif (
ws_connected is False
or backend_reachable is False
or (workflow_state is not None and workflow_state.state != "Normal")
):
headline = STATE_DEGRADED
else:
headline = STATE_OK
return NullRunStatus(
state=headline,
api_key_valid=api_key_valid,
api_key_prefix=api_key_prefix,
organization_id=self.organization_id,
workflow_id=self.workflow_id,
api_url=self.api_url,
backend_reachable=backend_reachable,
ws_connected=ws_connected,
workflow_state=workflow_state,
recent_errors=recent_errors,
)
def _record_error(
self,
err: "BaseException",
stage: str,
*,
workflow_id: str | None = None,
tool_name: str | None = None,
) -> None:
"""Layer 3: append a ``RecentError`` to the runtime's
ring buffer. Called from ``_emit_sdk_error`` AFTER the
Layer-2 ``emit_error`` so both layers see the same
error. The ring buffer feeds ``NullRunStatus.recent_errors``
— the user sees the last N errors via
``nullrun.status `` without instrumenting every
call site.
"""
from datetime import datetime, timezone
from nullrun.observability.status import RecentError
# Resolve workflow_id from the contextvar when the
# caller did not pass one — same precedence as
# ``_emit_sdk_error``.
resolved_workflow_id = workflow_id
if resolved_workflow_id is None and self.workflow_id is not None:
resolved_workflow_id = self.workflow_id
self._recent_errors.push(
RecentError(
error_code=getattr(err, "error_code", "NR-0000"),
stage=stage,
workflow_id=resolved_workflow_id,
tool_name=tool_name,
timestamp=datetime.now(tz=timezone.utc),
message=str(err)[:200],
)
)
def _emit_sdk_error(
self,
err: "BaseException",
stage: str,
*,
workflow_id: str | None = None,
tool_name: str | None = None,
correlation_id: str | None = None,
extra: dict[str, Any] | None = None,
) -> None:
"""Layer 2: fire the on_error hook with the runtime's known
context fields. Called from every raise site immediately
BEFORE the ``raise`` statement so the hook sees the
fully-constructed exception while the call stack is still
live.
Best-effort: this method NEVER raises. The hook itself is
wrapped in ``emit_error`` which catches hook exceptions.
A failure inside the hook cannot break the SDK.
Layer 3: also appends to the runtime's recent-errors
ring buffer so ``nullrun.status `` surfaces the error
without the user having to register a hook. Done AFTER
the hook dispatch (so the ring buffer does not delay
the hook) and AFTER the call-stack is built (so the
ring buffer sees the resolved workflow_id).
Hot path: the no-hooks case is skipped via ``has_hooks ``
so the call cost when nobody is listening is one boolean
check + an attribute access on ``self`` (no allocation
no lock — the hook registry short-circuits inside
``emit_error``). The Layer-3 ring-buffer push is ALWAYS
done — it is the no-instrumentation path to introspection.
"""
from nullrun.observability.error_hooks import (
ErrorContext,
emit_error,
has_hooks,
)
# Layer 3 (cheap path): always push to the ring buffer
# BEFORE the hook dispatch so a failing hook cannot
# prevent the error from appearing in ``nullrun.status ``.
self._record_error(
err,
stage,
workflow_id=workflow_id,
tool_name=tool_name,
)
if not has_hooks():
return
# Lazy-resolve workflow_id: the contextvar (set by
# ``nullrun.workflow(...)`` blocks) is authoritative for
# in-loop calls, falling back to the runtime's bound
# workflow when no contextvar is active.
resolved_workflow_id = workflow_id
if resolved_workflow_id is None and self.workflow_id is not None:
resolved_workflow_id = self.workflow_id
emit_error(
err,
ErrorContext(
stage=stage,
workflow_id=resolved_workflow_id,
tool_name=tool_name,
api_key_prefix=(self.api_key[:10] if self.api_key else None),
correlation_id=correlation_id,
extra=extra or {},
),
)
def _authenticate(self) -> None:
"""Authenticate with API key and get organization_id.
Also handles key version updates for HMAC secret key rotation.
On successful auth, the server may return a new key_version indicating
a secret key rotation. The SDK stores this and uses it for signing.
"""
if not self.api_key:
from nullrun.breaker.exceptions import NullRunConfigError
err = NullRunConfigError(
"API key required for cloud mode",
error_code="NR-C001",
user_action=(
"Set NULLRUN_API_KEY env var or pass api_key='nr_live_...' "
"to nullrun.init(). The SDK cannot operate without "
"credentials — the no-op local mode was removed in 0.3.0."
),
)
self._emit_sdk_error(err, stage="auth")
raise err
logger.debug(f"Authenticating with API at {self.api_url}/auth/verify")
try:
# 2026-06-28 audit P2.3: retry transient 503/504 + network blips
# during init. Backend emits 503 + Retry-After: 5 on transient
# DB error (backend/src/proxy/handlers.rs:11346-11351). Pre-fix
# the first 503 surfaced as NR-A001 to the user as if their API
# key were bad. Three attempts, exponential backoff (0.5s → 1s
# → 2s), honor Retry-After when present. Auth-key failures (401)
# are NOT retried — the key is wrong on attempt 1 means it's
# wrong on attempt 3.
response = self._post_auth_with_retry(
f"{self.api_url}/api/v1/auth/verify",
json_body={"api_key": self.api_key},
max_attempts=3,
)
if response.status_code == 200:
data = response.json()
# STRICT MODE: organization_id is REQUIRED, no fallback
org_id = data.get("organization_id")
if not org_id:
err = NullRunAuthenticationError(
"Auth response missing organization_id - server may be outdated or compromised. "
"Refusing to operate with legacy identity.",
error_code="NR-A002",
user_action=(
"The NullRun backend returned a 200 but the response "
"is missing organization_id. This usually means the "
"backend is on an older version than the SDK expects — "
"update the backend, or downgrade the SDK to a "
"version compatible with the deployed backend."
),
)
self._emit_sdk_error(err, stage="auth")
raise err
self.organization_id = org_id
# Phase 139+: pick up the workflow this key is bound to.
# `None` on legacy keys (pre-139 or never-used) -- call
# sites that NEED a workflow (check_workflow_budget
# check_control_plane, span events) will fall through to
# the contextvar when self.workflow_id is None, exactly
# like before. New keys always have this set.
self.workflow_id = data.get("workflow_id")
# Phase 0.3.1: pre-Phase-139 API keys do not return
# workflow_id, so the SDK cannot honour the
# dashboard's KILL/PAUSE for that workflow. Emit a
# one-time WARNING so the operator knows to rotate
# the key. Without this, the kill switch silently
# no-ops (a real safety hole for legacy users).
if self.workflow_id is None:
masked_key = (
(self.api_key[:8] + "***")
if self.api_key and len(self.api_key) >= 8
else "***"
)
logger.warning(
f"API key {masked_key!s} is a legacy key with no "
f"workflow binding; remote kill/pause will not be "
f"honoured. Rotate to a Phase 139+ key in the "
f"dashboard to enable control plane enforcement."
)
# Handle key rotation: server may return new key_version and secret_key
# This allows seamless secret key rotation without downtime
new_key_version = data.get("key_version")
new_secret_key = data.get("secret_key")
if new_key_version is not None and new_secret_key is not None:
old_version = getattr(self, "_key_version", None)
if old_version != new_key_version:
logger.info(
f"Secret key rotation: version {old_version} -> {new_key_version}"
)
self._key_version = new_key_version
self.secret_key = new_secret_key
# Update transport's secret key for subsequent requests
self._transport.secret_key = new_secret_key
logger.info(f"Authenticated: organization_id={self.organization_id}")
else:
# Auth failed - raise exception instead of silent fallback
err = NullRunAuthenticationError(
f"Auth failed with status {response.status_code}. "
f"API key may be invalid or expired. Not operating in unsafe mode.",
error_code=("NR-A003" if response.status_code == 401 else "NR-A001"),
)
self._emit_sdk_error(
err,
stage="auth",
correlation_id=response.headers.get("x-correlation-id"),