fix(security): re-resolve checkpoint/sticker-cache paths per call

tools/checkpoint_manager.py's CHECKPOINT_BASE and gateway/sticker_cache.py's
CACHE_PATH are resolved once at import time via get_hermes_home(), which is
a context-local ContextVar under the multiplexed gateway (multiple profiles
sharing one process). Freezing the path at import time pins every later
checkpoint/cache read-write to whichever profile's HERMES_HOME was active
when the module was first imported -- the same bug class already fixed for
cache dirs, skills_hub, rich_sent_store, and (this session) the OAuth/auth.json/
sessions.json paths.

CheckpointManager is "owned by AIAgent" per-instance, but its methods read
the frozen module constant directly instead of taking the store root from
the instance, so a profile's CheckpointManager can read/write code-edit
checkpoints into a different profile's store.

Add a per-call resolver for each path, following the established "respect
an existing test monkeypatch of the constant, otherwise re-resolve through
get_hermes_home()" pattern so the extensive existing test seams in
tests/tools/test_checkpoint_manager.py and tests/gateway/test_sticker_cache.py
keep working unmodified.

(cherry picked from commit 03ae075d969094cb584e6ab38d2a773d15ff875c)
(cherry picked from commit b850c4b18e2ae2158a97c6cb87bd2057918b8170)
This commit is contained in:
srojk34
2026-07-01 13:57:07 +03:00
committed by Teknium
parent 819517fbac
commit e70db09f51
3 changed files with 70 additions and 11 deletions

View File

@@ -6,12 +6,21 @@ Stickers are described via the vision tool once and cached by file_unique_id
import json
import time
from pathlib import Path
from typing import Optional
from hermes_cli.config import get_hermes_home
from utils import atomic_json_write
CACHE_PATH = get_hermes_home() / "sticker_cache.json"
_CACHE_PATH_AT_IMPORT = CACHE_PATH
def _resolve_cache_path() -> Path:
"""Active profile's cache file at call time: the patched ``CACHE_PATH`` when a test changed
it, else live profile-scoped HERMES_HOME — under the multiplexed gateway one process serves
every profile, so the import-time constant would pin every profile to the launch home."""
return CACHE_PATH if CACHE_PATH != _CACHE_PATH_AT_IMPORT else get_hermes_home() / "sticker_cache.json"
# Kept concise to save tokens.
STICKER_VISION_PROMPT = (
@@ -22,13 +31,13 @@ STICKER_VISION_PROMPT = (
def _load_cache() -> dict:
try:
return json.loads(CACHE_PATH.read_text(encoding="utf-8"))
return json.loads(_resolve_cache_path().read_text(encoding="utf-8"))
except (FileNotFoundError, json.JSONDecodeError, OSError):
return {}
def _save_cache(cache: dict) -> None:
atomic_json_write(CACHE_PATH, cache)
atomic_json_write(_resolve_cache_path(), cache)
def get_cached_description(file_unique_id: str) -> Optional[dict]:

View File

@@ -156,6 +156,47 @@ class TestGatewayHooksDirResolution:
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

View File

@@ -31,6 +31,15 @@ from utils import env_int
logger = logging.getLogger(__name__)
CHECKPOINT_BASE = get_hermes_home() / "checkpoints"
_CHECKPOINT_BASE_AT_IMPORT = CHECKPOINT_BASE
def _resolve_checkpoint_base() -> Path:
"""Active profile's checkpoint root at call time: the patched ``CHECKPOINT_BASE`` when a test
changed it, else live profile-scoped HERMES_HOME — under the multiplexed gateway one process
serves every profile, so the import-time constant would write every profile's code-edit
checkpoints into the launch profile's store."""
return CHECKPOINT_BASE if CHECKPOINT_BASE != _CHECKPOINT_BASE_AT_IMPORT else get_hermes_home() / "checkpoints"
_STORE_DIRNAME, _INDEXES_DIRNAME, _PROJECTS_DIRNAME, _LEDGERS_DIRNAME = "store", "indexes", "projects", "ledgers"
_REFS_PREFIX, _LEGACY_PREFIX, _PRUNE_MARKER_NAME = "refs/hermes", "legacy-", ".last_prune"
@@ -104,7 +113,7 @@ def _project_hash(working_dir: str) -> str:
def _store_path(base: Optional[Path] = None) -> Path:
return (base or CHECKPOINT_BASE) / _STORE_DIRNAME
return (base or _resolve_checkpoint_base()) / _STORE_DIRNAME
def _store_has_head(store: Path) -> bool:
@@ -516,7 +525,7 @@ class _ProjectRefs(NamedTuple):
def _project_refs(working_dir: str) -> _ProjectRefs:
abs_dir = str(_normalize_path(working_dir))
store, dir_hash = _store_path(CHECKPOINT_BASE), _project_hash(abs_dir)
store, dir_hash = _store_path(), _project_hash(abs_dir)
return _ProjectRefs(abs_dir, store, dir_hash, _index_path(store, dir_hash), _ref_name(dir_hash))
@@ -591,7 +600,7 @@ class CheckpointManager:
digest = _hash_file(path)
if digest is None:
return
store, dir_hash = _store_path(CHECKPOINT_BASE), _project_hash(self.get_working_dir_for_path(str(path)))
store, dir_hash = _store_path(), _project_hash(self.get_working_dir_for_path(str(path)))
_save_ledger(store, dir_hash, {**_load_ledger(store, dir_hash), str(path): {"sha256": digest, "ts": time.time()}})
except Exception as exc:
logger.debug("record_agent_write failed for %s: %s", file_path, exc)
@@ -674,7 +683,7 @@ class CheckpointManager:
Each entry carries the extra ``workdir`` key so callers can label which project a checkpoint belongs
to.
"""
store = _store_path(CHECKPOINT_BASE)
store = _store_path()
if not _store_has_head(store):
return []
results = [{**entry, "workdir": workdir}
@@ -1062,7 +1071,7 @@ def prune_checkpoints(retention_days: int = 7, delete_orphans: bool = True, chec
``str``) binds orphan deletion to exactly what a ``store_status()`` preview showed — a project
orphaned after the preview is skipped; ``None`` deletes every current orphan (``--force``,
unattended). ``max_total_size_mb > 0`` drops the oldest commit per project until the store fits."""
base = checkpoint_base or CHECKPOINT_BASE
base = checkpoint_base or _resolve_checkpoint_base()
result = _empty_prune_result()
if not base.exists():
return result
@@ -1087,7 +1096,7 @@ def maybe_auto_prune_checkpoints(retention_days: int = 7, min_interval_hours: in
"""Idempotent wrapper around ``prune_checkpoints`` for startup hooks: writes
``CHECKPOINT_BASE/.last_prune`` so calls within ``min_interval_hours`` short-circuit.
Returns ``{"skipped": bool, "result": prune dict, "error": optional str}``."""
base = checkpoint_base or CHECKPOINT_BASE
base = checkpoint_base or _resolve_checkpoint_base()
out: Dict[str, object] = {"skipped": False}
try:
if not base.exists():
@@ -1125,7 +1134,7 @@ def store_status(checkpoint_base: Optional[Path] = None) -> Dict:
``pre_v2_projects`` are repos still on the pre-v2 layout, distinct from the migrated
``legacy_archives``; an orphan-deletion preview must include both ``projects`` and
``pre_v2_projects`` since ``prune_checkpoints`` sweeps both."""
base = checkpoint_base or CHECKPOINT_BASE
base = checkpoint_base or _resolve_checkpoint_base()
out: Dict = {"base": str(base), "store_size_bytes": 0, "legacy_size_bytes": 0, "total_size_bytes": 0,
"project_count": 0, "projects": [], "pre_v2_projects": [], "legacy_archives": []}
if not base.exists():
@@ -1155,7 +1164,7 @@ def store_status(checkpoint_base: Optional[Path] = None) -> Dict:
def clear_all(checkpoint_base: Optional[Path] = None) -> Dict[str, int]:
"""Nuke the entire checkpoint base (store + legacy). Irreversible.
Returns ``{"bytes_freed": N, "deleted": bool}``."""
base = checkpoint_base or CHECKPOINT_BASE
base = checkpoint_base or _resolve_checkpoint_base()
out = {"bytes_freed": 0, "deleted": False}
if not base.exists():
return out
@@ -1170,7 +1179,7 @@ def clear_all(checkpoint_base: Optional[Path] = None) -> Dict[str, int]:
def clear_legacy(checkpoint_base: Optional[Path] = None) -> Dict[str, int]:
"""Delete all ``legacy-*`` archive directories. Returns ``{"bytes_freed": N, "deleted": count}``."""
base = checkpoint_base or CHECKPOINT_BASE
base = checkpoint_base or _resolve_checkpoint_base()
out = {"bytes_freed": 0, "deleted": 0}
if not base.exists():
return out