fix(gateway): freshness-gate resume_pending against per-message zombies

A crash-interrupted session marked resume_pending is returned by
get_or_create_session so its transcript reloads intact. The idle/daily
reset policy (#54442) keys on updated_at, which is bumped to now on every
message — so a zombie session that keeps receiving messages never trips
it and resumes stale context forever (context bleed reported on Telegram
and Feishu).

Gate the resume_pending branch on last_resume_marked_at (set once at
resume-mark, never bumped per-message) against the auto-continue freshness
window. If resume has been pending past the window, fall through to
auto-reset with reason "resume_pending_expired". A window <= 0 disables
the gate (opt-out for the pre-fix always-fresh behaviour).

Also hoist auto_continue_freshness_window() into gateway/session.py as the
single source of truth; gateway/run._auto_continue_freshness_window() now
delegates to it (keeps the existing import/patch surface).

Fixes #46934

Co-authored-by: Hermes Agent <noreply@nousresearch.com>

gateway/run.py

@@ -618,20 +618,19 @@ def _coerce_gateway_timestamp(value: Any) -> Optional[float]:
 def _auto_continue_freshness_window() -> float:
     """Return the configured auto-continue freshness window in seconds.
 
-    Reads ``HERMES_AUTO_CONTINUE_FRESHNESS`` (bridged from
-    ``config.yaml`` ``agent.gateway_auto_continue_freshness`` at gateway
-    startup, same pattern as ``HERMES_AGENT_TIMEOUT``).  Falls back to the
-    module default when unset or malformed.  Non-positive values disable
-    the freshness gate (restores the pre-fix "always fresh" behaviour for
-    users who want to opt out).
+    Thin wrapper that delegates to the canonical implementation in
+    ``gateway.session`` (the single source of truth shared with the
+    routing-time zombie gate in ``get_or_create_session``).  Reads
+    ``HERMES_AUTO_CONTINUE_FRESHNESS`` (bridged from ``config.yaml``
+    ``agent.gateway_auto_continue_freshness`` at gateway startup, same
+    pattern as ``HERMES_AGENT_TIMEOUT``).  Falls back to the module default
+    when unset or malformed.  Non-positive values disable the freshness gate
+    (restores the pre-fix "always fresh" behaviour for users who want to opt
+    out).  Kept here so existing call sites and test patches importing it
+    from ``gateway.run`` continue to work.
     """
-    raw = os.environ.get("HERMES_AUTO_CONTINUE_FRESHNESS")
-    if raw is None or raw == "":
-        return float(_AUTO_CONTINUE_FRESHNESS_SECS_DEFAULT)
-    try:
-        return float(raw)
-    except (TypeError, ValueError):
-        return float(_AUTO_CONTINUE_FRESHNESS_SECS_DEFAULT)
+    from gateway.session import auto_continue_freshness_window
+    return auto_continue_freshness_window()
 
 
 def _float_env(name: str, default: float) -> float:

gateway/session.py

@@ -27,6 +27,35 @@ def _now() -> datetime:
     return datetime.now()
 
 
+# Default auto-continue freshness window in seconds (1 hour).  A session
+# interrupted by a restart is only auto-resumed — and only returned by
+# ``get_or_create_session`` — while it stays within this window of when
+# ``resume_pending`` was marked.  ``gateway/run.py`` bridges
+# ``config.yaml`` ``agent.gateway_auto_continue_freshness`` into
+# ``HERMES_AUTO_CONTINUE_FRESHNESS`` at startup.
+_AUTO_CONTINUE_FRESHNESS_SECS_DEFAULT = 60 * 60
+
+
+def auto_continue_freshness_window() -> float:
+    """Return the configured auto-continue freshness window in seconds.
+
+    Single source of truth for both the resume scheduler (``gateway/run.py``)
+    and the routing-time zombie gate in ``get_or_create_session``.  Reads
+    ``HERMES_AUTO_CONTINUE_FRESHNESS`` (bridged from ``config.yaml``
+    ``agent.gateway_auto_continue_freshness`` at gateway startup) and falls
+    back to the module default when unset or malformed.  A non-positive value
+    disables the freshness gate (restores the pre-fix "always fresh" behaviour
+    for users who want to opt out).
+    """
+    raw = os.environ.get("HERMES_AUTO_CONTINUE_FRESHNESS")
+    if raw is None or raw == "":
+        return float(_AUTO_CONTINUE_FRESHNESS_SECS_DEFAULT)
+    try:
+        return float(raw)
+    except (TypeError, ValueError):
+        return float(_AUTO_CONTINUE_FRESHNESS_SECS_DEFAULT)
+
+
 # ---------------------------------------------------------------------------
 # PII redaction helpers
 # ---------------------------------------------------------------------------
@@ -1325,11 +1354,27 @@ class SessionStore:
                         # Restart-interrupted session: preserve the session_id
                         # and return the existing entry so the transcript reloads
                         # intact, but still honour normal daily/idle reset policy.
+                        #
+                        # Freshness gate (#46934): the idle/daily policy checks
+                        # ``updated_at``, which is bumped to ``now`` on every
+                        # message — so a zombie session that keeps receiving
+                        # messages never trips it and would resume stale context
+                        # forever.  ``last_resume_marked_at`` is set once when
+                        # resume was marked and never bumped per-message, so it
+                        # correctly measures how long resume has been pending.
+                        # If that exceeds the auto-continue freshness window, the
+                        # recovery turn either never ran or failed — treat the
+                        # session as a zombie and fall through to auto-reset.
                         reset_reason = self._should_reset(entry, source)
                         if not reset_reason:
-                            entry.updated_at = now
-                            self._save()
-                            return entry
+                            _fw = auto_continue_freshness_window()
+                            _ref_time = entry.last_resume_marked_at or entry.updated_at
+                            if _fw > 0 and (now - _ref_time).total_seconds() > _fw:
+                                reset_reason = "resume_pending_expired"
+                            else:
+                                entry.updated_at = now
+                                self._save()
+                                return entry
                     else:
                         reset_reason = self._should_reset(entry, source)
                     if not reset_reason:

tests/gateway/test_clean_shutdown_marker.py

@@ -244,3 +244,76 @@ class TestCleanShutdownMarker:
 
         assert agent._end_session_on_close is False
         agent.close.assert_called_once()
+
+
+# ---------------------------------------------------------------------------
+# resume_pending freshness gate (#46934)
+# ---------------------------------------------------------------------------
+
+class TestResumePendingFreshnessGate:
+    """A resume_pending session is only returned while it is still fresh.
+
+    ``get_or_create_session`` returns a ``resume_pending`` session so its
+    transcript reloads intact after a restart.  But the idle/daily reset
+    policy keys on ``updated_at``, which is bumped to ``now`` on every
+    message — so a zombie session that keeps receiving messages never trips
+    it and would resume stale context forever.  The freshness gate keys on
+    ``last_resume_marked_at`` (set once at resume-mark, never bumped) so it
+    catches that case.
+    """
+
+    def _mark_resume_pending(self, store, source):
+        """Put the session into resume_pending and return the entry."""
+        store.get_or_create_session(source)
+        count = store.suspend_recently_active()
+        assert count == 1
+        with store._lock:
+            entry = store._entries[store._generate_session_key(source)]
+        assert entry.resume_pending
+        assert entry.last_resume_marked_at is not None
+        return entry
+
+    def test_fresh_resume_pending_returns_same_session(self, tmp_path):
+        store = _make_store(tmp_path)
+        source = _make_source()
+        entry = self._mark_resume_pending(store, source)
+
+        # Within the freshness window (marked just now) → same session back.
+        refreshed = store.get_or_create_session(source)
+        assert refreshed.session_id == entry.session_id
+        assert refreshed.resume_pending
+
+    def test_stale_resume_pending_falls_through_to_reset(self, tmp_path, monkeypatch):
+        monkeypatch.setenv("HERMES_AUTO_CONTINUE_FRESHNESS", "3600")
+        store = _make_store(tmp_path)
+        source = _make_source()
+        entry = self._mark_resume_pending(store, source)
+
+        # Backdate the resume mark past the freshness window. Keep updated_at
+        # fresh (as a per-message zombie would have) so the idle/daily policy
+        # would NOT fire — only the freshness gate should catch this.
+        with store._lock:
+            entry.last_resume_marked_at = datetime.now() - timedelta(seconds=7200)
+            entry.updated_at = datetime.now()
+            store._save()
+
+        fresh = store.get_or_create_session(source)
+        # Zombie detected → brand-new session, not the stale transcript.
+        assert fresh.session_id != entry.session_id
+        assert not fresh.resume_pending
+
+    def test_freshness_gate_disabled_returns_stale_session(self, tmp_path, monkeypatch):
+        # Opt-out: window <= 0 restores the pre-fix "always fresh" behaviour.
+        monkeypatch.setenv("HERMES_AUTO_CONTINUE_FRESHNESS", "0")
+        store = _make_store(tmp_path)
+        source = _make_source()
+        entry = self._mark_resume_pending(store, source)
+
+        with store._lock:
+            entry.last_resume_marked_at = datetime.now() - timedelta(seconds=999999)
+            entry.updated_at = datetime.now()
+            store._save()
+
+        refreshed = store.get_or_create_session(source)
+        assert refreshed.session_id == entry.session_id
+        assert refreshed.resume_pending