# NousResearch/hermes-agent · Agent cache spurious invalidation fixes across sessions ## Open now - Nothing open on GitHub. Multiple bug reports and fixes addressing cases where the agent cache incorrectly invalidates or resets counters—such as usage stats—due to cross-process or cross-session-id mismatches under the same session key. The work spans tracking session IDs in cache entries, re-baselining counts, and backfilling counters from SessionDB. ## TL;DR - 4 member(s): 0 open issue(s), 1 closed issue(s), 0 open PR(s), 1 merged PR(s), 2 PR(s) closed without merging. - A fix was merged to main in PR #56210. Verify it addresses your case. - **Recommended:** read PR #56210 (and any competing PRs below) before filing. Confirm the merged change covers your symptom. ## Maintainer resolution verdict - Internal-tracker verdict over this cluster's issue members: `resolved` · deciding PR(s): `pr-56210`. This is distinct from the fix status above. Weigh it yourself. ## How this cluster was grouped - Link evidence (`STRUCTURAL_SPANNING`): every member is connected by structural evidence: shared changed files, or a closing/superseded reference. - Pairwise link density `0.50`: the share of member pairs that carry a link, counting a structural link in full and a relevance-only link at 0.36. `1.00` means every pair is linked. ## Closed / duplicate issues - #54947: \[Bug\]: Agent cache cross-process guard spuriously invalidates on session-id switch under same session_key ## Prior & competing PRs - **#56210: fix(gateway): close the agent-cache re-baseline bug class (#54947 + #46647 + #54583)**. merged to main · fix tier `verified` - maintainer direction: category `merged` - files: `gateway/run.py`, `scripts/release.py`, `tests/gateway/test_agent_cache.py`, `tests/gateway/test_first_turn_session_meta_rebaseline.py`, `tests/gateway/test_session_id_cache_coherence.py` - fix hunks: - `gateway/run.py` ```diff @@ -10679,19 +10679,12 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str, run_g _response_time, _api_calls, _resp_len, ) - # Re-baseline the cached agent's message_count snapshot now that - # this turn has completed and the agent has flushed its rows to - # the SessionDB. The cross-process coherence guard (#45966) - # snapshots the count at agent-BUILD time (before this turn's own - # writes) and never refreshes it on reuse — so without this, this - # process's own turn would grow the count and the next turn would - # see a mismatch and rebuild the agent every turn, destroying - # prompt caching. Refreshing here makes the guard fire only on a - # DIFFERENT process's writes. Uses the (possibly compaction- - # updated) live session_id. Fail-safe inside the helper. - await self._refresh_agent_cache_message_count( - session_key, session_entry.session_id - ) + # NOTE: the cross-process cache-coherence re-baseline + # (_refresh_agent_cache_message_count) is intentionally deferred + # until AFTER this turn's transcript persistence block below — it + # must include the first-turn `session_meta` marker row and the + # compression session_id swap, both of which happen later. See + # the call site after the `update_session(...)` write. # Successful turn — clear any stuck-loop counter for this session. # This ensures the counter only accumulates across CONSECUTIVE ``` ```diff @@ -11077,6 +11070,28 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str, run_g last_prompt_tokens=agent_result.get("last_prompt_tokens", 0), ) + # Re-baseline the cached agent's message_count snapshot now that + # ALL of this turn's transcript writes are done — the agent's + # flushed user/assistant/tool rows AND the first-turn `session_meta` + # marker appended above. The cross-process coherence guard (#45966) + # snapshots the count at agent-BUILD time (before this turn's own + # writes) and never refreshes it on reuse, so without this the + # process's own turn grows message_count and the next turn sees a + # mismatch and rebuilds the agent — destroying prompt caching. + # + # This MUST run after the `session_meta` append: that row also + # increments message_count, so re-baselining before it (the old + # position) left the snapshot one short and the guard mis-fired on + # turn 2 of EVERY fresh gateway conversation, rebuilding the cached + # agent and busting the prompt cache. Running here also uses the + # compaction-updated session_id (the agent_result session_id swap + # above), matching this function's documented contract. Refreshing + # here makes the guard fire only on a DIFFERENT process's writes. + # Fail-safe inside the helper. + await self._refresh_agent_cache_message_count( + session_key, session_entry.session_id + ) + # Intentional silence is a delivery decision, not a transcript # mutation. The agent's [SILENT]/NO_REPLY assistant turn above is # still persisted in session history so later turns keep normal ``` ```diff @@ -14902,6 +14917,15 @@ async def _refresh_agent_cache_message_count( only when the same agent is still cached (no rebuild/eviction raced in between). Fail-safe: any DB error leaves the snapshot as-is, which at worst costs one unnecessary rebuild on the next turn. + + When the cache entry records a ``session_id`` (4-tuple form, #54947) + that differs from the current ``session_id`` — meaning the cache + was built for a DIFFERENT conversation under the same ``session_key`` + — the snapshot is intentionally left untouched. Overwriting it with + the current session's count would corrupt the original conversation's + baseline and cause the next switch back to fire the cross-process + guard spuriously. Fail-safe: the legacy 3-tuple shape (no + ``session_id``) is still re-baselined as before. """ if self._session_db is None or not session_id: return ``` ```diff @@ -14926,8 +14950,23 @@ async def _refresh_agent_cache_message_count( and len(cached) > 2 and cached[0] is not _AGENT_PENDING_SENTINEL ): + # If the snapshot was taken for a different session_id + # (same session_key, different conversation), leave the + # snapshot alone — the current session_id's count belongs + # to a different DB row (#54947). + _snapshot_sid = cached[3] if len(cached) > 3 else None + if _snapshot_sid is not None and _snapshot_sid != session_id: + return if cached[2] != _live: - _cache[session_key] = (cached[0], cached[1], _live) + if _snapshot_sid is None: + # Legacy 3-tuple: preserve the original 3-element + # shape so existing entries stay compatible with + # callers that index ``cached[2]`` directly. + _cache[session_key] = (cached[0], cached[1], _live) + else: + _cache[session_key] = ( + cached[0], cached[1], _live, _snapshot_sid, + ) def _evict_cached_agent(self, session_key: str) -> None: """Remove a cached agent for a session (called on /new, /model, etc). ``` ```diff @@ -16634,9 +16673,25 @@ def _interim_assistant_cb(text: str, *, already_streamed: bool = False) -> None: if cached and cached[1] == _sig: # cached[2] is the message_count at cache time; # stale when a second process appended rows. + # cached[3] (when present) is the session_id the + # snapshot was taken for — used to skip the guard + # when the active session_id differs (#54947). _cached_mc = cached[2] if len(cached) > 2 else None + _cached_sid = cached[3] if len(cached) > 3 else None + # If the snapshot belongs to a different session_id + # (same session_key, different conversation), the + # message_count comparison is meaningless — the + # counts track DIFFERENT DB rows. REUSE the cached + # agent rather than rebuild and bust the prompt cache + # on every session switch (#54947). + _session_id_mismatch = ( + _cached_sid is not None + and session_id is not None + and _cached_sid != session_id + ) if ( - _cached_mc is not None + not _session_id_mismatch + and _cached_mc is not None and _current_msg_count is not None and _current_msg_count != _cached_mc ): ``` ```diff @@ -16667,7 +16722,6 @@ def _interim_assistant_cb(text: str, *, already_streamed: bool = False) -> None: _xproc_evicted_agent = _ev_agent else: agent = cached[0] - reused_cached_agent = True # Refresh LRU order so the cap enforcement evicts # truly-oldest entries, not the one we just used. if hasattr(_cache, "move_to_end"): ``` ```diff @@ -16680,6 +16734,7 @@ def _interim_assistant_cb(text: str, *, already_streamed: bool = False) -> None: # (cached agent may have been created with old config) agent.max_iterations = max_iterations logger.debug("Reusing cached agent for session %s", session_key) + reused_cached_agent = True # Lock released — now schedule cleanup of any cross-process-evicted # agent on a daemon thread so memory-provider shutdown / socket ``` ```diff @@ -16737,7 +16792,14 @@ def _interim_assistant_cb(text: str, *, already_streamed: bool = False) -> None: ) if _cache_lock and _cache is not None: with _cache_lock: - _cache[session_key] = (agent, _sig, _current_msg_count) + # Record the session_id the snapshot was taken for + # alongside the message_count, so the cross-process + # guard can skip the (meaningless) count comparison + # when the active session_id later switches under + # the same session_key (#54947). + _cache[session_key] = ( + agent, _sig, _current_msg_count, session_id, + ) self._enforce_agent_cache_cap() logger.debug("Created new agent for session %s (sig=%s)", session_key, _sig) ``` ```diff @@ -18206,6 +18268,22 @@ def _stream_confirmed_final_delivery( except Exception: pass + # Re-baseline the cached agent's message_count snapshot before + # recursing into the in-band queued (/queue) follow-up turn. + # The first turn has completed and flushed its own user + + # assistant rows to the SessionDB, so the cross-process + # coherence guard (#45966) — which this recursive _run_agent + # call re-enters — would otherwise see the grown on-disk count + # against the stale build-time snapshot and rebuild the agent + # on THIS process's OWN writes, destroying the prompt-cache + # prefix #46237 was merged to preserve. The existing + # re-baseline in _handle_message_with_agent only runs after the + # whole _run_agent chain unwinds — too late for the in-band + # follow-up. Use the same (session_key, session_id) the + # recursive call runs under so the snapshot matches exactly + # what the follow-up's guard will consult. Fail-safe in helper. + await self._refresh_agent_cache_message_count(session_key, session_id) + followup_result = await self._run_agent( message=next_message, context_prompt=context_prompt, ``` - `scripts/release.py` ```diff @@ -217,6 +217,8 @@ "157689911+itsflownium@users.noreply.github.com": "itsflownium", "dirtyren@users.noreply.github.com": "dirtyren", "13277570+justin-cyhuang@users.noreply.github.com": "justin-cyhuang", + "[redacted:email]": "Tranquil-Flow", + "jason@hermes-jc": "jcjc81", "290862769+friendshipisover@users.noreply.github.com": "friendshipisover", "51421+MattKotsenas@users.noreply.github.com": "MattKotsenas", "92324143+ypwcharles@users.noreply.github.com": "ypwcharles", ``` - `tests/gateway/test_agent_cache.py` ```diff @@ -1715,6 +1715,80 @@ def get_session(self, _sid): with runner._agent_cache_lock: assert runner._agent_cache["telegram:s1"][2] == 5 + @pytest.mark.asyncio + async def test_in_band_followup_reuses_cached_agent(self, tmp_path): + """Behavioral regression for the in-band queued (/queue) follow-up. + + #46237 re-baselines the snapshot only on the EXTERNAL-turn boundary + (in ``_handle_message_with_agent``, after the whole ``_run_agent`` + chain unwinds). The recursive in-band follow-up re-enters the cache + guard MID-CHAIN — while the cache still holds the build-time snapshot + and the first turn has already flushed its own rows — so without a + re-baseline at the follow-up boundary the guard sees the grown count + and rebuilds the agent on THIS process's own writes, re-introducing + the every-turn rebuild #46237 set out to fix, on the follow-up path. + + Pins both halves at that boundary: WITHOUT the re-baseline the in-band + follow-up would rebuild; WITH it the follow-up REUSES the warm agent. + The guard's reuse decision (``_guard_would_reuse``) mirrors the real + cache-hit guard, which reads ``get_session(session_id)`` with the same + ``session_id`` the recursive ``_run_agent`` call is given. + """ + from hermes_state import SessionDB + + db = SessionDB(db_path=tmp_path / "sessions.db") + db.create_session("s1", source="telegram") + runner = self._runner_with_db(db) + agent = object() + + # First turn: cache miss -> build. Snapshot is the pre-turn count. + _row = db.get_session("s1") + build_count = _row.get("message_count", 0) if _row else 0 + with runner._agent_cache_lock: + runner._agent_cache["telegram:s1"] = (agent, "sig", build_count) + + # First turn flushes its own user + assistant rows. + db.append_message("s1", role="user", content="u") + db.append_message("s1", role="assistant", content="a") + + # Bug reproduction: re-entering the guard at the in-band follow-up + # boundary WITHOUT the re-baseline sees the grown count and rebuilds. + assert self._guard_would_reuse(runner, "telegram:s1", "s1") is False + + # The fix: re-baseline at the follow-up boundary. + await runner._refresh_agent_cache_message_count("telegram:s1", "s1") + + # The in-band follow-up now REUSES the cached, warm-prefix agent. + assert self._guard_would_reuse(runner, "telegram:s1", "s1") is True + with runner._agent_cache_lock: + assert runner._agent_cache["telegram:s1"][0] is agent + + def test_in_band_followup_rebaseline_precedes_recursion(self): + """Pin the FIX PLACEMENT in the production source. + + The behavioral test above proves the re-baseline makes the in-band + follow-up reuse the cached agent, but it calls the helper directly — + it would still pass if the production call were deleted. This guards + the actual call site: the queued (/queue) follow-up recurses via + ``followup_result = await self._run_agent(...)`` inside + ``_run_agent_inner`` and the re-baseline MUST run BEFORE that + recursion (running it only after, like the external-turn site, is too + late for the in-band path — the follow-up would already have rebuilt). + """ + import inspect + from gateway.run import GatewayRunner + + # The recursion + pre-recursion re-baseline live in the extracted + # ``_run_agent_inner`` (older trees had them inline in ``_run_agent``). + src = inspect.getsource(GatewayRunner._run_agent_inner) + marker = "followup_result = await self._run_agent(" + assert marker in src, "in-band queued follow-up recursion not found in _run_agent_inner" + before_recursion = src[: src.index(marker)] + assert "_refresh_agent_cache_message_count" in before_recursion, ( + "the in-band queued follow-up recursion must be preceded by a " + "_refresh_agent_cache_message_count re-baseline, else the follow-up " + "rebuilds the agent on this process's own first-turn writes" + ) class TestCrossProcessInvalidationDefersCleanup: """#52197: cross-process cache invalidation must NOT run agent cleanup ``` - `tests/gateway/test_first_turn_session_meta_rebaseline.py` ```diff @@ -0,0 +1,264 @@ +"""Regression: first-turn ``session_meta`` row must be re-baselined into the +agent cache's message_count snapshot. + +Bug +--- +On a fresh gateway conversation the post-turn re-baseline +(``_refresh_agent_cache_message_count``) runs *before* the first-turn +``session_meta`` marker row is appended to the transcript: + + gateway/run.py: + ... agent run completes ... + self._refresh_agent_cache_message_count(...) # snapshot taken HERE + ... + if not history: # session_meta written LATER + append_to_transcript({"role": "session_meta", ...}) + +``append_to_transcript`` (no ``skip_db``) increments the session's +``message_count`` unconditionally (hermes_state.append_message), so the +snapshot ends up exactly +1 below the live on-disk count. + +The cross-process coherence guard (#45966) compares the live count against +that snapshot on the *next* inbound message, sees ``live != snapshot``, +mistakes this process's own ``session_meta`` write for a foreign write, and +**rebuilds the cached agent on turn 2 of every fresh conversation** — silently +busting the per-conversation prompt cache the cache exists to protect. + +The fix re-baselines AFTER all of this turn's transcript writes (including the +first-turn ``session_meta`` row), so the snapshot matches the live count and +the guard fires only on genuinely foreign writes. + +This drives the REAL ``_handle_message_with_agent`` against a REAL SessionDB +(the ``session_meta`` write actually increments ``message_count``) and asserts +the invariant: after a first turn, snapshot == live count → next turn reuses. +""" + +import sys +import threading +import types +from datetime import datetime +from unittest.mock import AsyncMock, MagicMock + +import pytest + +import gateway.run as gateway_run +from gateway.config import GatewayConfig, Platform +from gateway.platforms.base import MessageEvent +from gateway.session import SessionEntry, SessionSource + + +SESSION_KEY = "agent:main:telegram:group:-1001:12345" +SESSION_ID = "sess-first-turn" + + +def _bootstrap(monkeypatch, tmp_path, db): + """GatewayRunner wired to a REAL SessionDB for count reads, mirroring the + proven #42039 harness but with a live cache + real transcript counter.""" + fake_dotenv = types.ModuleType("dotenv") + fake_dotenv.load_dotenv = lambda *args, **kwargs: None + monkeypatch.setitem(sys.modules, "dotenv", fake_dotenv) + + config = GatewayConfig() + runner = gateway_run.GatewayRunner(config) + runner.adapters = {} + runner._running_agents = {} + runner._running_agents_ts = {} + runner._pending_messages = {} + runner._pending_approvals = {} + runner._is_user_authorized = lambda _source: True + runner._set_session_env = lambda _context: None + runner._handle_active_session_busy_message = AsyncMock(return_value=False) + # REAL SessionDB behind the async facade the gateway holds — the + # production re-baseline does ``await self._session_db.get_session(...)``, + # so it must be the AsyncSessionDB wrapper, not the raw sync DB. + from hermes_state import AsyncSessionDB + + runner._session_db = AsyncSessionDB(db) + runner._recover_telegram_topic_thread_id = lambda _source: None + runner._cache_session_source = lambda _key, _source: None + runner._is_session_run_current = lambda _key, _gen: True + runner._begin_session_run_generation = lambda _key: 1 + runner._reply_anchor_for_event = lambda _event: None + runner._get_guild_id = lambda _event: None + runner._should_send_voice_reply = lambda *_a, **_kw: False + runner.hooks = MagicMock() + runner.hooks.emit = AsyncMock() + + # Live agent cache (not a MagicMock) so the re-baseline actually rewrites + # the snapshot tuple in place. + runner._agent_cache = {} + runner._agent_cache_lock = threading.Lock() + + runner.session_store = MagicMock() + runner.session_store.get_or_create_session.return_value = SessionEntry( + session_key=SESSION_KEY, + session_id=SESSION_ID, + created_at=datetime.now(), + updated_at=datetime.now(), + platform=Platform.TELEGRAM, + chat_type="group", + ) + # Empty history → triggers the first-turn ``session_meta`` write path. + runner.session_store.load_transcript.return_value = [] + runner.session_store.has_platform_message_id.return_value = False + runner.session_store.update_session = MagicMock() + + # Mirror the real SessionStore.append_to_transcript: forward non-skip_db + # writes to the real DB so a ``session_meta`` row genuinely increments + # message_count, exactly as in production. skip_db=True writes (the agent + # already persisted them via _flush_messages_to_session_db) are no-ops here. + def _append(session_id, message, skip_db=False): + if not skip_db: + db.append_message( + session_id=session_id, + role=message.get("role", "unknown"), + content=message.get("content"), + ) + + runner.session_store.append_to_transcript = MagicMock(side_effect=_append) + + monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path) + monkeypatch.setattr( + gateway_run, "_resolve_runtime_agent_kwargs", lambda: {"api_key": "fake"} + ) + monkeypatch.setattr( + "agent.model_metadata.get_model_context_length", + lambda *_args, **_kwargs: 100_000, + ) + return runner + + +def _event(): + return MessageEvent( + text="hello world", + source=SessionSource( + platform=Platform.TELEGRAM, + chat_id="-1001", + chat_type="group", + user_id="12345", + ), + message_id="msg-1", + ) + + +def _source(): + return SessionSource( + platform=Platform.TELEGRAM, + chat_id="-1001", + chat_type="group", + user_id="12345", + ) + + +def _live_count(db, session_id): + row = db.get_session(session_id) + return (row.get("message_count", 0) if row else 0) + + [redacted:email] +async def test_first_turn_session_meta_is_captured_by_rebaseline( + monkeypatch, tmp_path +): + """After a fresh first turn, the cache snapshot must equal the live + message_count — including the first-turn ``session_meta`` row. + + WITHOUT the fix the re-baseline snapshots the count *before* the + session_meta append, leaving the snapshot one short; the cross-process + guard then rebuilds the cached agent on turn 2 (prompt-cache churn). + """ + from hermes_state import SessionDB + + db = SessionDB(db_path=tmp_path / "sessions.db") + db.create_session(SESSION_ID, source="telegram") + + runner = _bootstrap(monkeypatch, tmp_path, db) + + # Cache snapshot taken at agent-BUILD time = count before this turn's + # writes (a fresh session → 0). This is what the #45966 guard stores. + build_count = _live_count(db, SESSION_ID) + agent_obj = object() + with runner._agent_cache_lock: + runner._agent_cache[SESSION_KEY] = (agent_obj, "sig", build_count) + + # Stubbed agent run: the gateway's own user/assistant rows are persisted + # by the agent (skip_db=True downstream); only the session_meta marker is + # written by the gateway with skip_db=False. + runner._run_agent = AsyncMock( + return_value={ + "final_response": "Hi there!", + "messages": [ + {"role": "user", "content": "hello world"}, + {"role": "assistant", "content": "Hi there!"}, + ], + "tools": [{"name": "noop"}], + "history_offset": 0, + "last_prompt_tokens": 0, + } + ) + + await runner._handle_message_with_agent(_event(), _source(), SESSION_KEY, 1) + + # The first-turn session_meta row was written → live count advanced. + live = _live_count(db, SESSION_ID) + assert live == build_count + 1, ( + "first-turn session_meta should increment message_count by exactly 1" + ) + + # THE INVARIANT: the cache snapshot must now equal the live count, so the + # next turn's cross-process guard reuses the cached agent. + with runner._agent_cache_lock: + cached = runner._agent_cache[SESSION_KEY] + snapshot = cached[2] + assert snapshot == live, ( + f"cache snapshot {snapshot} != live count {live}: the first-turn " + f"session_meta write was not re-baselined, so the #45966 guard will " + f"rebuild the cached agent on turn 2 and bust the prompt cache." + ) + # And the cached agent instance must be untouched (never rebuilt). + assert cached[0] is agent_obj + + [redacted:email] +async def test_next_turn_guard_reuses_cached_agent_after_first_turn( + monkeypatch, tmp_path +): + """End-to-end consequence: with the snapshot correctly re-baselined, the + production cross-process guard's reuse condition (live == snapshot) holds + on turn 2 — no rebuild, prompt cache preserved.""" + from hermes_state import SessionDB + + db = SessionDB(db_path=tmp_path / "sessions.db") + db.create_session(SESSION_ID, source="telegram") + + runner = _bootstrap(monkeypatch, tmp_path, db) + with runner._agent_cache_lock: + runner._agent_cache[SESSION_KEY] = ( + object(), "sig", _live_count(db, SESSION_ID), + ) + + runner._run_agent = AsyncMock( + return_value={ + "final_response": "Hi there!", + "messages": [ + {"role": "user", "content": "hello world"}, + {"role": "assistant", "content": "Hi there!"}, + ], + "tools": [{"name": "noop"}], + "history_offset": 0, + "last_prompt_tokens": 0, + } + ) + + await runner._handle_message_with_agent(_event(), _source(), SESSION_KEY, 1) + + # Replicate the production cache-hit guard's reuse decision exactly: + # reuse iff live on-disk count == snapshot stored next to the agent. + live = _live_count(db, SESSION_ID) + with runner._agent_cache_lock: + snapshot = runner._agent_cache[SESSION_KEY][2] + would_reuse = (live == snapshot) + assert would_reuse, ( + "turn-2 cross-process guard would rebuild the cached agent because " + "the first-turn session_meta write was not re-baselined into the " + "snapshot — this is the prompt-cache regression under test." + ) ``` - `tests/gateway/test_session_id_cache_coherence.py` ```diff @@ -0,0 +1,318 @@ +"""Regression tests for #54947 — cross-process guard must not invalidate the +agent cache when the active ``session_id`` differs from the snapshot's +``session_id``, even when both share the same ``session_key``. + +Bug +--- +The cache key is the gateway ``session_key`` (e.g. ``agent:main:telegram:dm:USER_ID``) +which groups all DM sessions for that user. Different ``session_id``s (separate +conversation threads) can share a ``session_key``. When the user switches +between session_ids, the cached agent is shared, and the cross-process +coherence guard (``_cached_mc`` vs ``_current_msg_count``) treats different +sessions' ``message_count`` values as the same counter — invalidating the +agent on EVERY session switch and busting the per-conversation prompt cache. + +These tests pin the production guard's reuse decision across: + L1 — session-id switch must REUSE (not invalidate) the cached agent. + L2 — cache tuple records the snapshot's session_id. + L3 — re-baseline skips the cache entry when session_id differs. + L4 — same-session_id turns still re-baseline correctly (no regression + of #45966 / #46237). + L5 — legacy 2-tuples and pending sentinels are still untouched. + +All tests drive the REAL production helper (``_refresh_agent_cache_message_count``) +against a REAL ``SessionDB`` and exercise the cache-hit guard's logic with the +REAL cache lock, mirroring the structure used by +``TestAgentCacheMessageCountRebaseline``. +""" + +import threading + +import pytest + +from hermes_state import AsyncSessionDB + + +def _make_runner(): + """Create a minimal GatewayRunner with just the cache infrastructure.""" + from gateway.run import GatewayRunner + + runner = GatewayRunner.__new__(GatewayRunner) + runner._agent_cache = {} + runner._agent_cache_lock = threading.Lock() + return runner + + +def _guard_would_reuse(runner, session_key, session_id): + """Mirror the production cache-hit guard's reuse decision exactly + AFTER the fix: reuse when the session_id matches the snapshot's + session_id, OR when the entry is a legacy 2-tuple / pending sentinel. + + Reuse iff any of: + - cached session_id matches current session_id AND live count matches + snapshot count (same-process turn OR no foreign write) + - cached session_id differs from current session_id (different + conversation, snapshot is from a different DB row → meaningless + to compare, REUSE without invalidation) + - entry is a 2-tuple (legacy opt-out of guard) + - either side is None (unknown state → REUSE, fail-safe) + + Invalidate iff: + - cached session_id == current session_id AND + cached_mc is not None AND live_mc is not None AND + live_mc != cached_mc (genuine cross-process write on the SAME + session — guard fires, agent rebuilds). + """ + try: + # Mirror the production guard, which reads the sync underlying DB + # (``self._session_db._db.get_session``) off the async facade. + row = runner._session_db._db.get_session(session_id) + live = row.get("message_count", 0) if row else None + except Exception: + live = None + with runner._agent_cache_lock: + cached = runner._agent_cache.get(session_key) + + if cached is None: + return True # no entry → cache miss → fresh build (not invalidation) + # Legacy 2-tuple opts out of the guard. + if len(cached) < 3: + return True + # Pending sentinel — treat as a no-op reuse. + cached_sid = cached[3] if len(cached) > 3 else None + cached_mc = cached[2] + + # Snapshot belongs to a DIFFERENT session_id → comparison is + # meaningless; REUSE without invalidation. + if cached_sid is not None and session_id is not None and cached_sid != session_id: + return True + + # Same session_id: standard cross-process guard. + invalidate = ( + cached_mc is not None + and live is not None + and live != cached_mc + ) + return not invalidate + + +class TestSessionIdCacheCoherence: + """#54947 — guard must not invalidate the agent cache on session_id switch + under the same session_key.""" + + def test_session_id_switch_reuses_cached_agent(self, tmp_path): + """The reported bug: cache built from session A, switch to session B + under the same session_key. The guard must REUSE the cached agent + (the message_count comparison is meaningless across different + session_ids), not rebuild and bust the prompt cache. + """ + from hermes_state import SessionDB + + db = SessionDB(db_path=tmp_path / "sessions.db") + db.create_session("sA", source="telegram") + db.create_session("sB", source="telegram") + # Make counts differ to make the bug observable. + db.append_message("sA", role="user", content="hello from A") + db.append_message("sA", role="assistant", content="hi A") + db.append_message("sA", role="user", content="another from A") + # sA count = 3, sB count = 0 + runner = _make_runner() + runner._session_db = AsyncSessionDB(db) + agent = object() + + # Build cache from session A (mc=3, sid=sA). + with runner._agent_cache_lock: + runner._agent_cache["telegram:USER1"] = (agent, "sig", 3, "sA") + + # User switches to session B (mc=0, sid=sB) — same session_key. + # Guard must NOT invalidate. + assert _guard_would_reuse(runner, "telegram:USER1", "sB") is True, ( + "BUG: cache was invalidated on session_id switch — " + "the #54947 root cause is back." + ) + # The original agent must still be in the cache. + with runner._agent_cache_lock: + assert runner._agent_cache["telegram:USER1"][0] is agent + + @pytest.mark.asyncio + async def test_same_session_id_turns_still_reuse(self, tmp_path): + """#46237 / #45966 invariant: consecutive same-session turns must + REUSE the cached agent (prompt cache preserved).""" + from hermes_state import SessionDB + + db = SessionDB(db_path=tmp_path / "sessions.db") + db.create_session("s1", source="telegram") + runner = _make_runner() + runner._session_db = AsyncSessionDB(db) + agent = object() + + _row = db.get_session("s1") + build_count = _row.get("message_count", 0) if _row else 0 + with runner._agent_cache_lock: + runner._agent_cache["telegram:s1"] = (agent, "sig", build_count, "s1") + + reuses = 0 + for _ in range(1, 6): + db.append_message("s1", role="user", content="u") + db.append_message("s1", role="assistant", content="a") + # Post-turn re-baseline (the #46237 fix). + await runner._refresh_agent_cache_message_count("telegram:s1", "s1") + if _guard_would_reuse(runner, "telegram:s1", "s1"): + reuses += 1 + + assert reuses == 5 + with runner._agent_cache_lock: + assert runner._agent_cache["telegram:s1"][0] is agent + + @pytest.mark.asyncio + async def test_cross_process_write_still_invalidates(self, tmp_path): + """The original #45966 invariant must hold: a DIFFERENT process + appending to the same session in the shared DB invalidates the + cache (genuine cross-process write).""" + from hermes_state import SessionDB + + db = SessionDB(db_path=tmp_path / "sessions.db") + db.create_session("s1", source="telegram") + runner = _make_runner() + runner._session_db = AsyncSessionDB(db) + agent = object() + + with runner._agent_cache_lock: + _row = db.get_session("s1") + runner._agent_cache["telegram:s1"] = ( + agent, "sig", (_row.get("message_count", 0) if _row else 0), "s1", + ) + + # Our own turn + re-baseline → reuse next turn. + db.append_message("s1", role="user", content="u") + db.append_message("s1", role="assistant", content="a") + await runner._refresh_agent_cache_message_count("telegram:s1", "s1") + assert _guard_would_reuse(runner, "telegram:s1", "s1") is True + + # ANOTHER process (e.g. the desktop dashboard backend) appends a + # turn to the SAME session in the shared DB. + db.append_message("s1", role="user", content="external from dashboard") + + # Guard must invalidate. + assert _guard_would_reuse(runner, "telegram:s1", "s1") is False + + @pytest.mark.asyncio + async def test_refresh_skips_when_session_id_differs(self, tmp_path): + """_refresh_agent_cache_message_count must NOT refresh the cached + snapshot when the current session_id differs from the one the + snapshot belongs to. Otherwise the snapshot gets overwritten with + a different session's count, and the next switch back fires the + guard (the original bug).""" + from hermes_state import SessionDB + + db = SessionDB(db_path=tmp_path / "sessions.db") + db.create_session("sA", source="telegram") + db.create_session("sB", source="telegram") + db.append_message("sA", role="user", content="x") + runner = _make_runner() + runner._session_db = AsyncSessionDB(db) + agent = object() + + # Cache built from session A: (agent, sig, mc=1, sid=sA). + with runner._agent_cache_lock: + runner._agent_cache["telegram:USER1"] = (agent, "sig", 1, "sA") + + # Someone (the call site at line 9540) calls the re-baseline with + # the CURRENT session_id — which is sB after a switch. The + # snapshot is from sA → must NOT be touched. + await runner._refresh_agent_cache_message_count("telegram:USER1", "sB") + + with runner._agent_cache_lock: + cached = runner._agent_cache["telegram:USER1"] + assert cached[2] == 1, ( + f"BUG: snapshot was overwritten with sB's count: cached[2]={cached[2]}" + ) + assert cached[3] == "sA", ( + f"BUG: snapshot's session_id was changed: cached[3]={cached[3]}" + ) + assert cached[0] is agent + + @pytest.mark.asyncio + async def test_refresh_refreshes_when_session_id_matches(self, tmp_path): + """Sanity: when the snapshot's session_id matches the current one, + the re-baseline still runs and updates the count to the live value.""" + from hermes_state import SessionDB + + db = SessionDB(db_path=tmp_path / "sessions.db") + db.create_session("s1", source="telegram") + runner = _make_runner() + runner._session_db = AsyncSessionDB(db) + agent = object() + + with runner._agent_cache_lock: + runner._agent_cache["telegram:s1"] = (agent, "sig", 0, "s1") + + # s1's own turn flushes two rows. + db.append_message("s1", role="user", content="u") + db.append_message("s1", role="assistant", content="a") + await runner._refresh_agent_cache_message_count("telegram:s1", "s1") + + with runner._agent_cache_lock: + assert runner._agent_cache["telegram:s1"][2] == 2 + + @pytest.mark.asyncio + async def test_legacy_2tuple_and_pending_sentinel_untouched(self, tmp_path): + """Backward-compat: legacy 2-tuples and pending-sentinel 3-tuples + are not affected by the fix. The 2-tuple opts out of the guard; + the sentinel is left as-is by the re-baseline.""" + from hermes_state import SessionDB + from gateway.run import _AGENT_PENDING_SENTINEL + + db = SessionDB(db_path=tmp_path / "sessions.db") + db.create_session("s1", source="telegram") + db.append_message("s1", role="user", content="hi") + runner = _make_runner() + runner._session_db = AsyncSessionDB(db) + + # Legacy 2-tuple — untouched. + with runner._agent_cache_lock: + runner._agent_cache["telegram:s1"] = (object(), "sig") + await runner._refresh_agent_cache_message_count("telegram:s1", "s1") + with runner._agent_cache_lock: + assert len(runner._agent_cache["telegram:s1"]) == 2 + + # Pending sentinel — untouched. + with runner._agent_cache_lock: + runner._agent_cache["telegram:s1"] = (_AGENT_PENDING_SENTINEL, "sig", 0) + await runner._refresh_agent_cache_message_count("telegram:s1", "s1") + with runner._agent_cache_lock: + assert runner._agent_cache["telegram:s1"][0] is _AGENT_PENDING_SENTINEL + assert runner._agent_cache["telegram:s1"][2] == 0 + + @pytest.mark.asyncio + async def test_legacy_3tuple_session_id_unknown_still_guarded(self, tmp_path): + """An entry in the OLD 3-tuple shape (agent, sig, mc) with no + session_id — entries already in the cache from BEFORE the fix — + must STILL be guarded by the cross-process check. The fix only + ADDS a session_id-aware skip path; it does not weaken the + existing #45966 guard for entries that pre-date it. When + live count != snapshot count, the guard fires and the agent + rebuilds (same behavior as before the fix for legacy entries). + """ + from hermes_state import SessionDB + + db = SessionDB(db_path=tmp_path / "sessions.db") + db.create_session("s1", source="telegram") + db.append_message("s1", role="user", content="x") + runner = _make_runner() + runner._session_db = AsyncSessionDB(db) + + # Existing entry in old 3-tuple shape, no session_id recorded. + # Snapshot is mc=0; live is mc=1 — guard must fire (invalidate). + with runner._agent_cache_lock: + runner._agent_cache["telegram:s1"] = (object(), "sig", 0) + + # No session_id on cached entry → standard cross-process check + # still runs. live (1) != snapshot (0) → invalidates. + assert _guard_would_reuse(runner, "telegram:s1", "s1") is False + + # After the re-baseline (same session_id) snapshot matches live. + await runner._refresh_agent_cache_message_count("telegram:s1", "s1") + with runner._agent_cache_lock: + assert runner._agent_cache["telegram:s1"][2] == 1 + assert _guard_would_reuse(runner, "telegram:s1", "s1") is True ``` - **#55029: fix(gateway): scope agent-cache entries per session_id to prevent cross-session invalidation**. closed without merge - maintainer direction: category `implemented_elsewhere`; implication `superseded`; refs `#54947`, `#55013`, `#56210` - **#55234: fix: track session_id in agent cache to prevent spurious cross-process invalidation**. closed without merge - maintainer direction: category `duplicate_of`; implication `duplicate`; refs `#55013`, `#55029`, `#54947` ## Fix status - Merged to main in PR #56210 (cited above). No release data is tracked. Verify downstream. - Fix last verified (merged) `2026-07-01`. - Subsystem: `agent`