diff --git a/gateway/run_heartbeat_restore.py b/gateway/run_heartbeat_restore.py index c4022fb66d..6a3595a371 100644 --- a/gateway/run_heartbeat_restore.py +++ b/gateway/run_heartbeat_restore.py @@ -19,6 +19,13 @@ def _watched_homes(runner, default_home) -> list: return list(dict.fromkeys(Path(home) for home in homes)) +def _scope_key(runner, origin) -> str: + """Profile home ``_profile_scope_for_source`` would bind for ``origin`` (one key when not multiplexed).""" + if getattr(getattr(runner, "config", None), "multiplex_profiles", False): + return str(runner._resolve_profile_home_for_source(origin)) + return "" + + async def restore_heartbeat_watches(runner) -> None: """Retryable startup/poll scan; failed reads never prune existing watches. @@ -43,18 +50,30 @@ async def restore_heartbeat_watches(runner) -> None: if not any(profile_has_active_heartbeat(h) for h in _watched_homes(runner, home)): return restored with _profile_runtime_scope(home): - entries = store.list_sessions() - for entry in entries: + # Enter each profile's scope once per scan, not once per routed session: a scope entry + # hydrates the secret scope and terminal policy, so N sessions cost N parses otherwise. + # Sources are read through _restored_source so the persisted receiving bot is re-pinned + # before the scope key is derived; the same source object is what gets registered. + by_scope: dict = {} + for entry in store.list_sessions(): if entry.origin is None or not entry.session_id or entry.suspended: continue try: source = runner._restored_source(entry) - with runner._profile_scope_for_source(source): - manager = HeartbeatManager(entry.session_id) - if manager.is_active(): - restored.append((entry.session_key, source, entry.session_id)) + by_scope.setdefault(_scope_key(runner, source), []).append((entry, source)) except Exception: logger.debug("heartbeat restore for %s failed", entry.session_key, exc_info=True) + for group in by_scope.values(): + try: + with runner._profile_scope_for_source(group[0][1]): + for entry, source in group: + try: + if HeartbeatManager(entry.session_id).is_active(): + restored.append((entry.session_key, source, entry.session_id)) + except Exception: + logger.debug("heartbeat restore for %s failed", entry.session_key, exc_info=True) + except Exception: + logger.debug("heartbeat restore scope for %s failed", group[0][0].session_key, exc_info=True) return restored try: diff --git a/hermes_cli/env_loader.py b/hermes_cli/env_loader.py index 741aa573c0..0026f14819 100644 --- a/hermes_cli/env_loader.py +++ b/hermes_cli/env_loader.py @@ -15,7 +15,7 @@ from pathlib import Path # wiped (#57828) so early recovery provably runs before third-party imports (test_early_recovery). # The parser internals are imported lazily below because gateway tests stub ``sys.modules["dotenv"]``. import dotenv # noqa: F401 -from utils import atomic_replace, fast_safe_load +from utils import atomic_replace, fast_safe_load, load_yaml_file_readonly logger = logging.getLogger(__name__) @@ -632,9 +632,9 @@ def _load_secrets_config(home_path: Path) -> dict: return data.get("secrets") or {} except Exception: pass + # Routed profiles re-enter their scope on every poll/turn; only re-parse after the file changed. try: - with open(config_path, "r", encoding="utf-8") as f: - data = fast_safe_load(f) or {} + data = load_yaml_file_readonly(config_path) or {} except Exception: # noqa: BLE001 return {} return data.get("secrets") or {} diff --git a/tests/gateway/test_heartbeat_watch_restore.py b/tests/gateway/test_heartbeat_watch_restore.py index 88e6dbff56..6f10f7f0aa 100644 --- a/tests/gateway/test_heartbeat_watch_restore.py +++ b/tests/gateway/test_heartbeat_watch_restore.py @@ -173,3 +173,56 @@ async def test_startup_arms_retry_poller_even_without_any_watches(monkeypatch): if task: task.cancel() await asyncio.gather(task, return_exceptions=True) + + +@pytest.mark.asyncio +async def test_restore_enters_each_profile_scope_once_per_scan(tmp_path, monkeypatch): + """N routed sessions in one profile cost one scope entry, not N (scope entry re-parses config).""" + from gateway import run_heartbeat_restore + from gateway.run_heartbeat_restore import restore_heartbeat_watches + + home = tmp_path / '.hermes' + named = home / 'profiles' / 'work' + named.mkdir(parents=True) + (named / 'config.yaml').write_text('{}') + monkeypatch.setattr(Path, 'home', lambda: tmp_path) + monkeypatch.setenv('HERMES_HOME', str(home)) + dbs = {str(p): SessionDB(db_path=p / 'state.db') for p in (home, named)} + monkeypatch.setattr(goals, '_DB_CACHE', dbs) + config = GatewayConfig(multiplex_profiles=True) + store = SessionStore(home / 'sessions', config) + try: + expected = {} + for profile in (None, 'work', 'work', None, 'work'): + source = SessionSource(platform=Platform.TELEGRAM, chat_id='chat', + thread_id=str(len(expected)), profile=profile, scope_id='workspace') + with _profile_runtime_scope(named if profile else home): + entry = store.get_or_create_session(source) + HeartbeatManager(entry.session_id).set('check', 60) + expected[entry.session_key] = (entry.origin, entry.session_id) + store.close_all_db_handles() + runner = GatewayRunner.__new__(GatewayRunner) + runner.config = config + runner.session_store = SessionStore(home / 'sessions', config) + runner._heartbeat_watch = {} + runner._start_heartbeat_poller = lambda: None + runner._profile_name_for_source = lambda source: source.profile + runner._adapter_for_source = lambda source: object() + runner._run_in_executor_with_context = asyncio.to_thread + entered = [] + real = GatewayRunner._profile_scope_for_source + + def counting(self, source): + entered.append(source.profile) + return real(self, source) + + monkeypatch.setattr(GatewayRunner, '_profile_scope_for_source', counting) + await restore_heartbeat_watches(runner) + assert runner._heartbeat_watch == expected + assert sorted(entered, key=str) == [None, 'work'] + finally: + store.close_all_db_handles() + if 'runner' in locals(): + runner.session_store.close_all_db_handles() + for db in dbs.values(): + db.close() diff --git a/tests/test_load_yaml_file_readonly.py b/tests/test_load_yaml_file_readonly.py new file mode 100644 index 0000000000..a64a5843fa --- /dev/null +++ b/tests/test_load_yaml_file_readonly.py @@ -0,0 +1,35 @@ +"""``load_yaml_file_readonly`` re-parses only when the file signature changes.""" +import os + +import pytest + +from utils import load_yaml_file_readonly + + +def test_cache_hit_returns_same_object_and_invalidates_on_rewrite(tmp_path): + path = tmp_path / "config.yaml" + path.write_text("secrets: {a: 1}\n") + first = load_yaml_file_readonly(path) + assert first == {"secrets": {"a": 1}} + assert load_yaml_file_readonly(path) is first + + path.write_text("secrets: {a: 2}\n") + os.utime(path, ns=(os.stat(path).st_mtime_ns + 1_000_000,) * 2) + second = load_yaml_file_readonly(path) + assert second == {"secrets": {"a": 2}} + assert second is not first + + +def test_parse_error_propagates_and_is_not_cached(tmp_path): + path = tmp_path / "config.yaml" + path.write_text("secrets: [unclosed\n") + with pytest.raises(Exception): + load_yaml_file_readonly(path) + path.write_text("secrets: {}\n") + os.utime(path, ns=(os.stat(path).st_mtime_ns + 1_000_000,) * 2) + assert load_yaml_file_readonly(path) == {"secrets": {}} + + +def test_missing_file_raises(tmp_path): + with pytest.raises(FileNotFoundError): + load_yaml_file_readonly(tmp_path / "nope.yaml") diff --git a/tools/terminal_scope.py b/tools/terminal_scope.py index 1883ffb123..a7cf9ba644 100644 --- a/tools/terminal_scope.py +++ b/tools/terminal_scope.py @@ -143,11 +143,11 @@ def build_profile_terminal_scope( except Exception as exc: raise TerminalPolicyUnavailable(f"cannot resolve terminal config in {home}: {exc}") from exc if config_exists: - from hermes_cli.config import fast_safe_load + from utils import load_yaml_file_readonly try: - with open(config_path, encoding="utf-8") as f: - raw = fast_safe_load(f) + # Signature-cached: a scope is rebuilt per routed turn/poll, the file rarely changes. + raw = load_yaml_file_readonly(config_path) except Exception as exc: raise TerminalPolicyUnavailable(f"cannot parse {config_path}: {exc}") from exc raw_terminal = raw.get("terminal") if isinstance(raw, dict) else None diff --git a/utils.py b/utils.py index 8add707c0c..cb4db2f4ad 100644 --- a/utils.py +++ b/utils.py @@ -7,6 +7,7 @@ import os import shutil import stat import tempfile +import threading import time from contextlib import suppress from pathlib import Path @@ -605,6 +606,29 @@ def fast_safe_load(stream: Any) -> Any: return yaml.load(stream, Loader=_fast_yaml_loader) +_YAML_FILE_CACHE: dict = {} +_YAML_FILE_CACHE_LOCK = threading.Lock() + + +def load_yaml_file_readonly(path: Union[str, Path]) -> Any: + """``fast_safe_load`` of a file, re-parsed only when its :func:`file_signature` changes. + + Returns the cached object itself — callers must never mutate it. Parse errors propagate and + are not cached; a missing file raises ``FileNotFoundError`` like ``open`` does.""" + path = Path(path) + sig = file_signature(path.stat()) + key = str(path) + with _YAML_FILE_CACHE_LOCK: + cached = _YAML_FILE_CACHE.get(key) + if cached is not None and cached[0] == sig: + return cached[1] + with open(path, encoding="utf-8") as f: + data = fast_safe_load(f) + with _YAML_FILE_CACHE_LOCK: + _YAML_FILE_CACHE[key] = (sig, data) + return data + + def _env_number(key: str, default, cast): raw = os.getenv(key, "").strip() try: