Coverage for node / src / stigmem_node / routes / time_travel_gate.py: 91%

1"""Shared fail-closed gate for experimental time-travel requests.""" 

2 

3from __future__ import annotations 

4 

5from importlib import import_module 

6from typing import Any, Literal 

7 

8from fastapi import HTTPException, status 

9 

10TIME_TRAVEL_PLUGIN_NAME = "stigmem-plugin-time-travel" 

11 

12TimeTravelSurface = Literal["fact_query", "recall"] 

13 

14 

15def require_time_travel_enabled(registry: Any, *, surface: TimeTravelSurface) -> None: 

16 """Require the time-travel plugin and its explicit operator gate.""" 

17 

18 if TIME_TRAVEL_PLUGIN_NAME not in registry.registered_plugins(): 

19 raise HTTPException( 

20 status_code=status.HTTP_501_NOT_IMPLEMENTED, 

21 detail={ 

22 "code": "time_travel_plugin_not_loaded", 

23 "message": "time-travel queries require stigmem-plugin-time-travel", 

24 }, 

25 ) 

26 

27 config = _load_time_travel_config() 

28 surface_gate = { 

29 "fact_query": "allow_fact_query_as_of", 

30 "recall": "allow_recall_as_of", 

31 }[surface] 

32 if not bool(getattr(config, "enabled", False)) or not bool( 

33 getattr(config, surface_gate, False) 

34 ): 

35 raise HTTPException( 

36 status_code=status.HTTP_403_FORBIDDEN, 

37 detail={ 

38 "code": "time_travel_plugin_disabled", 

39 "message": ( 

40 "time-travel queries require explicit operator enablement " 

41 f"for {surface.replace('_', ' ')}" 

42 ), 

43 }, 

44 ) 

45 

46 

47def _load_time_travel_config() -> Any: 

48 try: 

49 config_module = import_module("stigmem_plugin_time_travel.config") 

50 return config_module.load_config_from_env() 

51 except Exception as exc: # noqa: BLE001 

52 raise HTTPException( 

53 status_code=status.HTTP_501_NOT_IMPLEMENTED, 

54 detail={ 

55 "code": "time_travel_plugin_not_loaded", 

56 "message": "time-travel plugin configuration is unavailable", 

57 }, 

58 ) from exc