Agent cache spurious invalidation fixes across sessions

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #56210 fix(gateway): close the agent-cache re-baseline bug class (#54947 + #46647 + #54583)

  merged to main

  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

  Show the diff

  • gateway/run.py

    @@ -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
    @@ -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
    @@ -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
    @@ -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).
    @@ -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
                             ):
    @@ -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"):
    @@ -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
    @@ -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)
     
    @@ -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

    @@ -217,6 +217,8 @@
         "[email protected]": "itsflownium",
         "[email protected]": "dirtyren",
         "[email protected]": "justin-cyhuang",
    +    "[redacted:email]": "Tranquil-Flow",
    +    "jason@hermes-jc": "jcjc81",
         "[email protected]": "friendshipisover",
         "[email protected]": "MattKotsenas",
         "[email protected]": "ypwcharles",

  • tests/gateway/test_agent_cache.py

    @@ -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

    @@ -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

    @@ -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