fix(gateway): keep idle cached agents alive until session expires

gateway/run.py

@@ -15161,6 +15161,27 @@ def _sweep_idle_cached_agents(self) -> int:
                 if last_activity is None:
                     continue
                 if (now - last_activity) > _AGENT_CACHE_IDLE_TTL_SECS:
+                    # Check whether the session has actually expired in the
+                    # session store.  If it hasn't (e.g. daily-reset mode
+                    # where the reset fires hours after the user's last
+                    # message), keep the agent in cache so the session-store
+                    # expiry watcher can still find it and call
+                    # on_session_end() with the live transcript.  Skipping
+                    # eviction here means the agent stays alive until the
+                    # session genuinely expires, at which point the watcher
+                    # (gateway/run.py _session_expiry_watcher) tears it down
+                    # properly.  (#11205 follow-up)
+                    session_entry = None
+                    try:
+                        _store = getattr(self, "session_store", None)
+                        if _store is not None:
+                            _store._ensure_loaded()
+                            session_entry = _store._entries.get(key)
+                    except Exception:
+                        pass
+                    if session_entry is not None:
+                        if not _store._is_session_expired(session_entry):
+                            continue  # keep agent — session hasn't expired
                     to_evict.append((key, agent))
             for key, _ in to_evict:
                 _cache.pop(key, None)

tests/gateway/test_agent_cache.py

@@ -604,6 +604,58 @@ def test_idle_sweep_skips_agents_without_activity_ts(self, monkeypatch):
         assert runner._sweep_idle_cached_agents() == 0
         assert "s" in runner._agent_cache
 
+    def test_idle_sweep_keeps_agent_when_session_not_expired(self, monkeypatch):
+        """Agents past idle TTL are kept if the session hasn't expired yet.
+
+        In daily-reset mode the reset can fire hours after the last
+        user message — evicting the agent early means the
+        session-expiry watcher has nothing to call on_session_end()
+        with, and memory providers miss the live transcript.
+        """
+        from gateway import run as gw_run
+
+        monkeypatch.setattr(gw_run, "_AGENT_CACHE_IDLE_TTL_SECS", 0.01)
+        runner = self._bounded_runner()
+        runner._cleanup_agent_resources = MagicMock()
+
+        import time as _t
+        stale = self._fake_agent(last_activity=_t.time() - 10.0)
+
+        # Session store says the session is still alive.
+        session_entry = MagicMock()
+        runner.session_store = MagicMock()
+        runner.session_store._entries = {"stale-session": session_entry}
+        runner.session_store._is_session_expired.return_value = False
+
+        runner._agent_cache["stale-session"] = (stale, "sig")
+
+        evicted = runner._sweep_idle_cached_agents()
+        assert evicted == 0
+        assert "stale-session" in runner._agent_cache
+
+    def test_idle_sweep_evicts_when_session_is_expired(self, monkeypatch):
+        """Agent IS evicted when past idle TTL AND session store says expired."""
+        from gateway import run as gw_run
+
+        monkeypatch.setattr(gw_run, "_AGENT_CACHE_IDLE_TTL_SECS", 0.01)
+        runner = self._bounded_runner()
+        runner._cleanup_agent_resources = MagicMock()
+
+        import time as _t
+        stale = self._fake_agent(last_activity=_t.time() - 10.0)
+
+        # Session store says the session has expired.
+        session_entry = MagicMock()
+        runner.session_store = MagicMock()
+        runner.session_store._entries = {"stale-session": session_entry}
+        runner.session_store._is_session_expired.return_value = True
+
+        runner._agent_cache["stale-session"] = (stale, "sig")
+
+        evicted = runner._sweep_idle_cached_agents()
+        assert evicted == 1
+        assert "stale-session" not in runner._agent_cache
+
     def test_plain_dict_cache_is_tolerated(self):
         """Test fixtures using plain {} don't crash _enforce_agent_cache_cap."""
         from gateway.run import GatewayRunner