Coverage for core / safe_hartos_attr.py: 71.1%

1"""Singleton accessor for the canonical hart_intelligence modules. 

2 

3Worker threads must NOT do ``from hart_intelligence import X`` directly. 

4The hart_intelligence module pulls in transformers / langchain / autogen 

5on first import — a chain that holds Python's per-module import lock for 

6minutes on cold disk. When a worker thread races the canonical loader 

7(``hartos_backend_adapter._attempt_hartos_init``) on that same lock, the 

8worker stalls forever — the symptom shape is `Tier-1 

9ACTIVE` never logs, chat falls through to fallback path, daemon's 

10``is_user_recently_active()`` gate is unreachable, MMLU runaway can't be 

11stopped, etc). 

12 

13The singleton is ``sys.modules['hart_intelligence']`` — Python guarantees 

14a single dict entry per module. ONE writer (the canonical loader) does 

15the slow import. Every other reader uses this accessor. No new lock, 

16no new mechanism — we're just naming the singleton that already exists. 

17 

18Observability 

19============= 

20Every resolution is logged so silent no-ops are never invisible: 

21 

22* HIT — symbol resolved, returned to caller (DEBUG, with module name). 

23* MISS — HARTOS not yet loaded; first miss per symbol → INFO (loud once), 

24 subsequent identical misses → DEBUG (quiet). This way the 

25 logs show the *transition* without flooding when many workers 

26 hit the same not-yet-loaded symbol every tick. 

27* MISS_ATTR — modules are loaded but the requested attribute isn't 

28 bound (mid-init or symbol typo). Always INFO — distinct 

29 from "loader hasn't started" because the diagnosis is 

30 different. 

31 

32Usage 

33===== 

34:: 

35 

36 from core.safe_hartos_attr import safe_hartos_attr 

37 

38 publish_async = safe_hartos_attr('publish_async') 

39 if publish_async is not None: 

40 publish_async(topic, msg) 

41 # else: HARTOS not yet loaded — caller no-ops, same as the 

42 # `try: from hart_intelligence import …; except ImportError: pass` 

43 # pattern this replaces, but without triggering the import lock. 

44 

45Enforcement 

46=========== 

47``[tool.ruff.lint.flake8-tidy-imports.banned-api]`` in HARTOS 

48``pyproject.toml`` bans ``from hart_intelligence import …`` and 

49``from hart_intelligence_entry import …`` everywhere except the 

50canonical loader (which carries ``# noqa: TID251`` with a comment). 

51""" 

52from __future__ import annotations 

53 

54import logging 

55import sys 

56import threading 

57from typing import Any 

58 

59logger = logging.getLogger('core.safe_hartos_attr') 

60 

61_FACADE = 'hart_intelligence' 

62_ENTRY = 'hart_intelligence_entry' 

63 

64# Tracks (symbol_name, was_loaded_at_check_time) tuples we've already 

65# logged at INFO. Subsequent identical misses fall to DEBUG so a worker 

66# polling every tick doesn't flood the log. Reset implicitly on process 

67# restart; cleared explicitly by ``reset_log_dedup()`` for tests. 

68_logged_states: set[tuple[str, str]] = set() 

69_dedup_lock = threading.Lock() 

70 

71 

72def _log_outcome(name: str, outcome: str, *, mod_name: str | None = None) -> None: 

73 """Log a resolution outcome, deduplicating identical (symbol, outcome) pairs. 

74 

75 First occurrence at INFO, subsequent at DEBUG — gives a loud signal 

76 on state transitions (loader finishes, symbol becomes resolvable) 

77 while keeping steady-state ticks quiet. 

78 """ 

79 key = (name, outcome) 

80 with _dedup_lock: 

81 first = key not in _logged_states 

82 if first: 

83 _logged_states.add(key) 

84 if outcome == 'HIT': 

85 # HITs are normal once HARTOS is up — DEBUG always, single 

86 # transition log on first hit per symbol so we can see "loader 

87 # came online" in the timeline. 

88 if first: 

89 logger.info( 

90 "safe_hartos_attr HIT name=%s mod=%s (first resolution — " 

91 "HARTOS now reachable for this symbol)", 

92 name, mod_name, 

93 ) 

94 else: 

95 logger.debug("safe_hartos_attr HIT name=%s mod=%s", name, mod_name) 

96 elif outcome == 'MISS': 

97 if first: 

98 logger.info( 

99 "safe_hartos_attr MISS name=%s — HARTOS not yet loaded " 

100 "(canonical loader hartos_backend_adapter._attempt_hartos_init " 

101 "hasn't published sys.modules['hart_intelligence']). " 

102 "Caller will no-op gracefully; subsequent identical misses " 

103 "logged at DEBUG.", 

104 name, 

105 ) 

106 else: 

107 logger.debug("safe_hartos_attr MISS name=%s", name) 

108 elif outcome == 'MISS_ATTR': 

109 # Different bug shape from MISS: modules are loaded but symbol 

110 # not bound. Always INFO — caller likely typo'd or symbol moved. 

111 logger.info( 

112 "safe_hartos_attr MISS_ATTR name=%s mod=%s — module loaded " 

113 "but attribute not bound (typo? renamed? mid-init?)", 

114 name, mod_name, 

115 ) 

116 

117 

118def safe_hartos_attr(name: str, default: Any = None) -> Any: 

119 """Read a symbol from the loaded hart_intelligence singleton. 

120 

121 Resolution order: 

122 1. ``sys.modules['hart_intelligence']`` (the canonical facade) 

123 2. ``sys.modules['hart_intelligence_entry']`` (the implementation) 

124 3. ``default`` if neither is loaded yet 

125 

126 The facade re-exports everything from the entry module via 

127 ``from hart_intelligence_entry import *``, so once the loader has 

128 finished, both names point to the same symbol set — but during early 

129 boot only the entry module may be partially loaded. Checking both 

130 keeps callers working at every loader stage without a single false 

131 ImportError on the cold path. 

132 

133 NEVER triggers an import. If neither module is in ``sys.modules``, 

134 returns ``default``. Caller is expected to handle ``None``/``default`` 

135 by no-op'ing — exactly the shape ``idle_detection.get_idle_opted_in_agents`` 

136 and ``world_model_bridge._init_in_process`` already use. 

137 

138 Every call is logged (deduplicated) so silent fall-throughs are 

139 never invisible — see module docstring for log levels. 

140 

141 Args: 

142 name: attribute name to look up (e.g. ``'publish_async'``). 

143 default: value to return if HARTOS isn't loaded or the attribute 

144 doesn't exist. Defaults to ``None``. 

145 

146 Returns: 

147 The attribute value, or ``default``. 

148 """ 

149 facade = sys.modules.get(_FACADE) 

150 entry = sys.modules.get(_ENTRY) 

151 mod = facade or entry 

152 if mod is None: 

153 _log_outcome(name, 'MISS') 

154 return default 

155 mod_name = _FACADE if facade is not None else _ENTRY 

156 val = getattr(mod, name, None) 

157 if val is None: 

158 _log_outcome(name, 'MISS_ATTR', mod_name=mod_name) 

159 return default 

160 _log_outcome(name, 'HIT', mod_name=mod_name) 

161 return val 

162 

163 

164def hartos_loaded() -> bool: 

165 """True if the canonical loader has finished and the facade is ready. 

166 

167 Cheaper than ``safe_hartos_attr`` for "is it ready?" checks where the 

168 caller doesn't need a specific symbol — no ``getattr`` traversal, 

169 just a dict lookup. Use this for short-circuit gates:: 

170 

171 if not hartos_loaded(): 

172 return # nothing for us to do yet 

173 

174 A ``True`` result means at least one of the two module names is in 

175 ``sys.modules``; it does NOT promise every symbol the loader will 

176 eventually publish is bound yet (mid-init reads can still see 

177 ``None`` from ``safe_hartos_attr``). Use ``safe_hartos_attr`` with 

178 a defensive ``if x is not None`` for the actual call. 

179 """ 

180 return _FACADE in sys.modules or _ENTRY in sys.modules 

181 

182 

183def reset_log_dedup() -> None: 

184 """Clear the (symbol, outcome) dedup memory. 

185 

186 Tests use this to assert on first-time INFO emission without process 

187 restart noise. Production code should not call this — repeated INFO 

188 logs of the same state transition are pure noise on the chat hot path. 

189 """ 

190 with _dedup_lock: 

191 _logged_states.clear()