Gateway session reset and resume_pending bug fixes

Fixes on record

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

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

  merged to main

  files: gateway/run.py, gateway/session.py, tests/gateway/test_clean_shutdown_marker.py

  Show the diff

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

  • 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