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

gateway/run.py

@@ -147,6 +147,7 @@ def _gateway_surface_passes_raw_text(platform: Any) -> bool:
 _GATEWAY_SECRET_PATTERNS = (
     re.compile(r"\bsk-[A-Za-z0-9][A-Za-z0-9_\-]{12,}\b"),
     re.compile(r"\bgh[pousr]_[A-Za-z0-9_]{20,}\b"),
+    re.compile(r"\bxapp-\d+-[A-Za-z0-9\-]{20,}\b"),
     re.compile(r"\bxox[baprs]-[A-Za-z0-9\-]{20,}\b"),
     re.compile(r"\bhf_[A-Za-z0-9]{20,}\b"),
     re.compile(r"\bglpat-[A-Za-z0-9_\-]{20,}\b"),
@@ -617,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 @@ def get_or_create_session(
                         # 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:
@@ -1789,17 +1834,27 @@ def has_platform_message_id(
             logger.debug("has_platform_message_id lookup failed", exc_info=True)
             return False
 
-    def rewrite_transcript(self, session_id: str, messages: List[Dict[str, Any]]) -> None:
+    def rewrite_transcript(self, session_id: str, messages: List[Dict[str, Any]]) -> bool:
         """Replace the entire transcript for a session with new messages.
 
         Used by /retry, /undo, and /compress to persist modified conversation
         history. state.db is the canonical store.
+
+        Returns ``True`` when the write lands (or there is no DB to write to)
+        and ``False`` when the canonical write fails. Most callers can ignore
+        the result, but callers that would otherwise commit a destructive state
+        change on top of a failed write — e.g. /compress repointing the live
+        session onto a fresh session_id — must check it so they can surface an
+        error instead of silently dropping the conversation.
         """
-        if self._db:
-            try:
-                self._db.replace_messages(session_id, messages)
-            except Exception as e:
-                logger.debug("Failed to rewrite transcript in DB: %s", e)
+        if not self._db:
+            return True
+        try:
+            self._db.replace_messages(session_id, messages)
+            return True
+        except Exception as e:
+            logger.debug("Failed to rewrite transcript in DB: %s", e)
+            return False
 
     def load_transcript(self, session_id: str) -> List[Dict[str, Any]]:
         """Load all messages from a session's transcript.

tests/gateway/test_clean_shutdown_marker.py

@@ -244,3 +244,76 @@ async def _run():
 
         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