"""Profile-isolation regression tests for single-process multi-profile runtimes. In runtimes that serve every profile from one OS process (the desktop ``tui_gateway``), the profile boundary is the context-local ``_HERMES_HOME_OVERRIDE`` ContextVar, not the process environment. State that escapes the request call stack — import-time-frozen path constants, direct ``os.environ`` reads, or worker threads that don't inherit the request context — silently reverts to the launch/default profile and leaks one profile's data into another. These tests drive each previously-leaking site under override A then override B with real temp HERMES_HOME directories (no mocks) and assert the *active* profile's path is used. They are the productionized form of the manual smoke probes used to confirm the bug class. """ import threading from pathlib import Path import pytest from hermes_constants import ( get_hermes_home, reset_hermes_home_override, set_hermes_home_override, ) @pytest.fixture def two_profiles(tmp_path): """Two distinct profile HERMES_HOME dirs with the dir skeleton created.""" prof_a = tmp_path / "profA" prof_b = tmp_path / "profB" for p in (prof_a, prof_b): (p / "skills").mkdir(parents=True, exist_ok=True) (p / "state").mkdir(parents=True, exist_ok=True) (p / "cache").mkdir(parents=True, exist_ok=True) return prof_a, prof_b def _under_override(home: Path, fn): """Run ``fn`` with the profile override set to ``home`` and reset after.""" token = set_hermes_home_override(str(home)) try: return fn() finally: reset_hermes_home_override(token) # --------------------------------------------------------------------------- # M1 — import-time path globals / direct os.environ reads # --------------------------------------------------------------------------- class TestSkillsHubPathResolution: """tools/skills_hub.py path constants must reflect the active profile.""" def test_skills_dir_follows_override(self, two_profiles): prof_a, prof_b = two_profiles import tools.skills_hub as sh # Importing/touching under A must NOT pin the path for B. a_seen = _under_override(prof_a, lambda: Path(sh.SKILLS_DIR)) b_seen = _under_override(prof_b, lambda: Path(sh.SKILLS_DIR)) assert a_seen == prof_a / "skills" assert b_seen == prof_b / "skills" assert a_seen != b_seen def test_hub_derived_paths_follow_override(self, two_profiles): prof_a, prof_b = two_profiles import tools.skills_hub as sh b_lock = _under_override(prof_b, lambda: Path(sh.LOCK_FILE)) b_audit = _under_override(prof_b, lambda: Path(sh.AUDIT_LOG)) b_index = _under_override(prof_b, lambda: Path(sh.INDEX_CACHE_DIR)) assert b_lock == prof_b / "skills" / ".hub" / "lock.json" assert b_audit == prof_b / "skills" / ".hub" / "audit.log" assert b_index == prof_b / "skills" / ".hub" / "index-cache" class TestGatewayCacheDirResolution: """gateway/platforms/base.py cache getters must follow the active profile.""" def test_image_cache_dir_follows_override(self, two_profiles): prof_a, prof_b = two_profiles import gateway.platforms.base as gb a_seen = _under_override(prof_a, lambda: gb.get_image_cache_dir()) b_seen = _under_override(prof_b, lambda: gb.get_image_cache_dir()) assert str(a_seen).startswith(str(prof_a)) assert str(b_seen).startswith(str(prof_b)) assert a_seen != b_seen class TestRichSentStorePathResolution: """gateway/rich_sent_store.py must honor the override, not read os.environ.""" def test_store_path_follows_override(self, two_profiles, monkeypatch): prof_a, prof_b = two_profiles # Ensure no ambient HERMES_HOME env masks the test. monkeypatch.delenv("HERMES_HOME", raising=False) import gateway.rich_sent_store as rss b_seen = _under_override(prof_b, lambda: rss._store_path()) assert b_seen.startswith(str(prof_b)) assert b_seen.endswith("state/rich_sent_index.json") class TestGatewayHooksDirResolution: """gateway/hooks.py's HookRegistry must discover hooks from the active profile's directory — otherwise one profile's HookRegistry loads and executes a DIFFERENT profile's hook handlers (arbitrary Python) against its own live event context under the multiplexed gateway.""" def test_hooks_dir_follows_override(self, two_profiles): prof_a, prof_b = two_profiles import gateway.hooks as gh a_seen = _under_override(prof_a, lambda: gh._resolve_hooks_dir()) b_seen = _under_override(prof_b, lambda: gh._resolve_hooks_dir()) assert a_seen == prof_a / "hooks" assert b_seen == prof_b / "hooks" assert a_seen != b_seen def test_discover_and_load_uses_active_profile_hooks_dir(self, two_profiles): """End-to-end: a hook that only exists under profile B's hooks dir must not be discovered when profile A's override is active, and vice versa — proving the registry doesn't fall back to a frozen profile.""" prof_a, prof_b = two_profiles import gateway.hooks as gh b_hook_dir = prof_b / "hooks" / "only-in-b" b_hook_dir.mkdir(parents=True) (b_hook_dir / "HOOK.yaml").write_text( "name: only-in-b\nevents: [\"agent:start\"]\n", encoding="utf-8", ) (b_hook_dir / "handler.py").write_text( "async def handle(event_type, context):\n pass\n", encoding="utf-8", ) def _load_and_names(): reg = gh.HookRegistry() reg.discover_and_load() return [h["name"] for h in reg.loaded_hooks] a_hooks = _under_override(prof_a, _load_and_names) b_hooks = _under_override(prof_b, _load_and_names) assert "only-in-b" not in a_hooks assert "only-in-b" in b_hooks class TestCheckpointManagerPathResolution: """tools/checkpoint_manager.py's checkpoint store root must honor the active profile — otherwise one profile's CheckpointManager instance can read/write code-edit checkpoints into a different profile's store under the multiplexed gateway.""" def test_checkpoint_base_follows_override(self, two_profiles): prof_a, prof_b = two_profiles import tools.checkpoint_manager as cm a_seen = _under_override(prof_a, lambda: cm._resolve_checkpoint_base()) b_seen = _under_override(prof_b, lambda: cm._resolve_checkpoint_base()) assert a_seen == prof_a / "checkpoints" assert b_seen == prof_b / "checkpoints" assert a_seen != b_seen def test_store_path_follows_override(self, two_profiles): prof_a, prof_b = two_profiles import tools.checkpoint_manager as cm b_seen = _under_override(prof_b, lambda: cm._store_path()) assert b_seen == prof_b / "checkpoints" / "store" class TestStickerCachePathResolution: """gateway/sticker_cache.py's cache file must honor the active profile — otherwise one profile's Telegram sticker-description cache leaks into a different profile's under the multiplexed gateway.""" def test_cache_path_follows_override(self, two_profiles): prof_a, prof_b = two_profiles import gateway.sticker_cache as sc a_seen = _under_override(prof_a, lambda: sc._resolve_cache_path()) b_seen = _under_override(prof_b, lambda: sc._resolve_cache_path()) assert a_seen == prof_a / "sticker_cache.json" assert b_seen == prof_b / "sticker_cache.json" assert a_seen != b_seen # --------------------------------------------------------------------------- # M2 — thread / executor context propagation # --------------------------------------------------------------------------- class TestThreadContextPropagation: """Worker threads must inherit the spawning turn's profile override.""" def test_raw_thread_loses_override(self, two_profiles): """Document the underlying hazard: a bare thread does NOT inherit it.""" _prof_a, prof_b = two_profiles seen = {} def worker(): seen["home"] = str(get_hermes_home()) def run(): t = threading.Thread(target=worker) t.start() t.join() _under_override(prof_b, run) # A bare thread falls back to the process default — this is WHY the fix # primitive is needed. (Asserted as the hazard, not the desired state.) assert seen["home"] != str(prof_b) def test_run_async_worker_preserves_override(self, two_profiles): """model_tools._run_async's worker-thread branch must keep the override. This is the generic sync->async bridge for every async tool; if it leaks, every async tool that resolves get_hermes_home() leaks. """ import asyncio _prof_a, prof_b = two_profiles import model_tools async def reads_home(): return str(get_hermes_home()) async def driver(): # Inside a running loop, _run_async spawns a worker thread + loop. return model_tools._run_async(reads_home()) seen = _under_override(prof_b, lambda: asyncio.run(driver())) assert seen == str(prof_b)