Coverage for node / src / stigmem_node / lifecycle / tombstone_gate.py: 100%

1"""Runtime gate for experimental RTBF tombstone behavior.""" 

2 

3from __future__ import annotations 

4 

5import os 

6from collections.abc import Mapping 

7 

8TOMBSTONE_PLUGIN_NAME = "stigmem-plugin-tombstones" 

9_ENV_PREFIX = "STIGMEM_TOMBSTONES_" 

10 

11 

12def tombstone_plugin_registered() -> bool: 

13 """Return True when the RTBF tombstone plugin is active in the registry.""" 

14 from ..plugins import get_registry 

15 

16 return TOMBSTONE_PLUGIN_NAME in get_registry().registered_plugins() 

17 

18 

19def tombstone_filter_enabled(environ: Mapping[str, str] | None = None) -> bool: 

20 """Return True only when tombstone filtering is explicitly enabled.""" 

21 

22 if not tombstone_plugin_registered(): 

23 return False 

24 env = environ if environ is not None else os.environ 

25 return _env_bool(env, f"{_ENV_PREFIX}ENABLED") and _env_bool( 

26 env, f"{_ENV_PREFIX}ALLOW_RECALL_FILTER" 

27 ) 

28 

29 

30def _env_bool(env: Mapping[str, str], name: str, *, default: bool = False) -> bool: 

31 raw = env.get(name) 

32 if raw is None: 

33 return default 

34 return raw.strip().lower() in {"1", "true", "yes", "on"}