Gateway stale session routing and expiry finalization bugs

Fixes on record

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

• #54138 fix(gateway): prune stale sessions.json entries on startup

  merged to main

  files: gateway/session.py, tests/gateway/test_session_store_stale_prune.py

  Show the diff

  • gateway/session.py

    @@ -847,7 +847,58 @@ def _ensure_loaded_locked(self) -> None:
                     print(f"[gateway] Warning: Failed to load sessions: {e}")
     
             self._loaded = True
    -    
    +
    +        # Prune any sessions.json entries that point to sessions already ended
    +        # in state.db. A hard gateway crash (exit code 1) skips the graceful
    +        # shutdown path, so sessions.json is never cleared and is left pointing
    +        # at ended sessions. On the next startup those stale entries act as live
    +        # routing keys, but get_or_create_session() reuses them as long as the
    +        # time/policy reset checks pass — it never consults end_reason — so every
    +        # incoming message is silently routed into a closed session. Pruning here
    +        # (lock already held) is cheap: one lookup per routing key, once at
    +        # startup, and self-heals into a fresh session on the next message.
    +        self._prune_stale_sessions_locked()
    +
    +    def _prune_stale_sessions_locked(self) -> None:
    +        """Remove sessions.json entries whose session has ended in state.db.
    +
    +        Called once during startup (from ``_ensure_loaded_locked``, lock held).
    +        A ``session_id`` is stale when state.db reports ``end_reason IS NOT
    +        NULL`` for it. Sessions absent from the DB (never persisted / pre-SQLite
    +        legacy) are left alone, and a ``None`` DB handle (SQLite unavailable) is
    +        a no-op. DB errors are non-fatal — startup must never fail here.
    +        """
    +        db = getattr(self, "_db", None)
    +        if not db or not self._entries:
    +            return
    +
    +        stale_keys: list = []
    +        try:
    +            for key, entry in self._entries.items():
    +                row = db.get_session(entry.session_id)
    +                # row is None        -> not in DB (legacy / pre-SQLite) — keep
    +                # end_reason is None  -> session alive — keep
    +                # end_reason not None -> session ended — prune
    +                if row is not None and row.get("end_reason") is not None:
    +                    logger.warning(
    +                        "gateway.session: pruning stale sessions.json entry "
    +                        "%r -> %s (end_reason=%r); left by a crashed gateway",
    +                        key, entry.session_id, row["end_reason"],
    +                    )
    +                    stale_keys.append(key)
    +        except Exception as exc:
    +            logger.warning(
    +                "gateway.session: stale-entry pruning skipped due to DB error: %s",
    +                exc,
    +            )
    +            return
    +
    +        for key in stale_keys:
    +            del self._entries[key]
    +
    +        if stale_keys:
    +            self._save()
    +
         def _save(self) -> None:
             """Save sessions index to disk (kept for session key -> ID mapping)."""
             import tempfile

  • tests/gateway/test_session_store_stale_prune.py

    @@ -0,0 +1,181 @@
    +"""Tests for SessionStore._prune_stale_sessions_locked — crash self-healing.
    +
    +When a gateway crashes (exit code 1) the graceful shutdown path is skipped and
    +sessions.json is left pointing at sessions already ended in state.db. On the
    +next startup _ensure_loaded_locked calls _prune_stale_sessions_locked to detect
    +and remove those stale routing entries before get_or_create_session() can reuse
    +them and silently route incoming messages into a closed session (#52804).
    +"""
    +
    +import json
    +from datetime import datetime, timedelta
    +from unittest.mock import MagicMock, patch
    +
    +from gateway.config import GatewayConfig, Platform, SessionResetPolicy
    +from gateway.session import SessionEntry, SessionStore
    +
    +
    +# ---------------------------------------------------------------------------
    +# Helpers
    +# ---------------------------------------------------------------------------
    +
    +def _make_entry(key: str, session_id: str) -> SessionEntry:
    +    now = datetime.now()
    +    return SessionEntry(
    +        session_key=key,
    +        session_id=session_id,
    +        created_at=now - timedelta(hours=2),
    +        updated_at=now - timedelta(hours=1),
    +        platform=Platform.TELEGRAM,
    +        chat_type="dm",
    +    )
    +
    +
    +def _make_store_with_db(tmp_path, db_mock) -> SessionStore:
    +    """Build a SessionStore with a mock SessionDB, bypassing disk load."""
    +    config = GatewayConfig(default_reset_policy=SessionResetPolicy(mode="none"))
    +    with patch("gateway.session.SessionStore._ensure_loaded"):
    +        store = SessionStore(sessions_dir=tmp_path, config=config)
    +    store._db = db_mock
    +    store._loaded = True
    +    return store
    +
    +
    +def _db_returning(rows: dict) -> MagicMock:
    +    """SessionDB mock where get_session maps session_id -> row dict."""
    +    db = MagicMock()
    +    db.get_session.side_effect = lambda sid: rows.get(sid)
    +    return db
    +
    +
    +# ---------------------------------------------------------------------------
    +# Core behaviour
    +# ---------------------------------------------------------------------------
    +
    +class TestPruneStaleSessionsLocked:
    +    def test_prunes_ended_session(self, tmp_path):
    +        db = _db_returning({"sid_dm": {"end_reason": "agent_close", "id": "sid_dm"}})
    +        store = _make_store_with_db(tmp_path, db)
    +        store._entries["dm_key"] = _make_entry("dm_key", "sid_dm")
    +
    +        store._prune_stale_sessions_locked()
    +
    +        assert "dm_key" not in store._entries
    +
    +    def test_keeps_live_session(self, tmp_path):
    +        db = _db_returning({"sid_live": {"end_reason": None, "id": "sid_live"}})
    +        store = _make_store_with_db(tmp_path, db)
    +        store._entries["live_key"] = _make_entry("live_key", "sid_live")
    +
    +        store._prune_stale_sessions_locked()
    +
    +        assert "live_key" in store._entries
    +
    +    def test_keeps_session_absent_from_db(self, tmp_path):
    +        """Entry for a session_id not in state.db (legacy) is left alone."""
    +        db = _db_returning({})
    +        store = _make_store_with_db(tmp_path, db)
    +        store._entries["legacy_key"] = _make_entry("legacy_key", "sid_legacy")
    +
    +        store._prune_stale_sessions_locked()
    +
    +        assert "legacy_key" in store._entries
    +
    +    def test_prunes_multiple_stale_entries(self, tmp_path):
    +        db = _db_returning({
    +            "sid_a": {"end_reason": "agent_close", "id": "sid_a"},
    +            "sid_b": {"end_reason": "session_reset", "id": "sid_b"},
    +            "sid_c": {"end_reason": None, "id": "sid_c"},  # alive — keep
    +        })
    +        store = _make_store_with_db(tmp_path, db)
    +        store._entries["key_a"] = _make_entry("key_a", "sid_a")
    +        store._entries["key_b"] = _make_entry("key_b", "sid_b")
    +        store._entries["key_c"] = _make_entry("key_c", "sid_c")
    +
    +        store._prune_stale_sessions_locked()
    +
    +        assert "key_a" not in store._entries
    +        assert "key_b" not in store._entries
    +        assert "key_c" in store._entries
    +
    +    def test_noop_when_db_is_none(self, tmp_path):
    +        config = GatewayConfig(default_reset_policy=SessionResetPolicy(mode="none"))
    +        with patch("gateway.session.SessionStore._ensure_loaded"):
    +            store = SessionStore(sessions_dir=tmp_path, config=config)
    +        store._db = None
    +        store._loaded = True
    +        store._entries["key"] = _make_entry("key", "sid_x")
    +
    +        store._prune_stale_sessions_locked()  # must not raise
    +
    +        assert "key" in store._entries
    +
    +    def test_noop_when_no_entries(self, tmp_path):
    +        db = MagicMock()
    +        store = _make_store_with_db(tmp_path, db)
    +
    +        store._prune_stale_sessions_locked()
    +
    +        db.get_session.assert_not_called()
    +
    +    def test_db_error_is_non_fatal(self, tmp_path):
    +        db = MagicMock()
    +        db.get_session.side_effect = Exception("DB locked")
    +        store = _make_store_with_db(tmp_path, db)
    +        store._entries["key"] = _make_entry("key", "sid_x")
    +
    +        store._prune_stale_sessions_locked()  # must not raise
    +
    +        assert "key" in store._entries  # safe fallback — keep on error
    +
    +    def test_sessions_json_rewritten_after_pruning(self, tmp_path):
    +        db = _db_returning({"sid_stale": {"end_reason": "agent_close", "id": "sid_stale"}})
    +        store = _make_store_with_db(tmp_path, db)
    +        store._entries["stale_key"] = _make_entry("stale_key", "sid_stale")
    +
    +        with patch.object(store, "_save") as mock_save:
    +            store._prune_stale_sessions_locked()
    +            mock_save.assert_called_once()
    +
    +    def test_sessions_json_not_rewritten_when_nothing_pruned(self, tmp_path):
    +        db = _db_returning({"sid_live": {"end_reason": None, "id": "sid_live"}})
    +        store = _make_store_with_db(tmp_path, db)
    +        store._entries["live_key"] = _make_entry("live_key", "sid_live")
    +
    +        with patch.object(store, "_save") as mock_save:
    +            store._prune_stale_sessions_locked()
    +            mock_save.assert_not_called()
    +
    +
    +# ---------------------------------------------------------------------------
    +# Integration: _ensure_loaded_locked calls _prune_stale_sessions_locked
    +# ---------------------------------------------------------------------------
    +
    +class TestEnsureLoadedCallsPrune:
    +    def test_stale_entry_pruned_during_load(self, tmp_path):
    +        entry = _make_entry("dm_key", "sid_stale")
    +        (tmp_path / "sessions.json").write_text(
    +            json.dumps({"dm_key": entry.to_dict()}, indent=2), encoding="utf-8"
    +        )
    +        db = _db_returning({"sid_stale": {"end_reason": "agent_close", "id": "sid_stale"}})
    +        config = GatewayConfig(default_reset_policy=SessionResetPolicy(mode="none"))
    +        store = SessionStore(sessions_dir=tmp_path, config=config)
    +        store._db = db
    +
    +        store._ensure_loaded()
    +
    +        assert "dm_key" not in store._entries
    +
    +    def test_live_entry_survives_load(self, tmp_path):
    +        entry = _make_entry("active_key", "sid_live")
    +        (tmp_path / "sessions.json").write_text(
    +            json.dumps({"active_key": entry.to_dict()}, indent=2), encoding="utf-8"
    +        )
    +        db = _db_returning({"sid_live": {"end_reason": None, "id": "sid_live"}})
    +        config = GatewayConfig(default_reset_policy=SessionResetPolicy(mode="none"))
    +        store = SessionStore(sessions_dir=tmp_path, config=config)
    +        store._db = db
    +
    +        store._ensure_loaded()
    +
    +        assert "active_key" in store._entries

• #55485 fix(gateway): self-heal stale sessions.json routing at message time (#54878)

  merged to main

  files: gateway/session.py, tests/gateway/test_session_store_runtime_stale_guard.py

  Show the diff

  • gateway/session.py

    @@ -905,11 +905,11 @@ def _ensure_loaded_locked(self) -> None:
             # in state.db. A hard gateway crash (exit code 1) skips the graceful
             # shutdown path, so sessions.json is never cleared and is left pointing
             # at ended sessions. On the next startup those stale entries act as live
    -        # routing keys, but get_or_create_session() reuses them as long as the
    -        # time/policy reset checks pass — it never consults end_reason — so every
    -        # incoming message is silently routed into a closed session. Pruning here
    -        # (lock already held) is cheap: one lookup per routing key, once at
    -        # startup, and self-heals into a fresh session on the next message.
    +        # routing keys. get_or_create_session() only consulted end_reason at
    +        # startup (here) until #54878 added a routing-time guard for the
    +        # live-gateway case; this startup prune still self-heals crash-left
    +        # entries before the first message arrives. Pruning here (lock already
    +        # held) is cheap: one lookup per routing key, once at startup.
             self._prune_stale_sessions_locked()
     
         def _prune_stale_sessions_locked(self) -> None:
    @@ -1148,6 +1148,32 @@ def _is_session_expired(self, entry: SessionEntry) -> bool:
     
             return False
     
    +    def _is_session_ended_in_db(self, session_id: str) -> bool:
    +        """Return True iff state.db has this session with a non-null end_reason.
    +
    +        Mirrors the staleness test in ``_prune_stale_sessions_locked``:
    +          - no DB handle / no session_id -> False (can't tell — keep)
    +          - row absent (legacy / not yet persisted) -> False (keep)
    +          - end_reason is None -> False (alive — keep)
    +          - end_reason not None -> True (ended — stale)
    +
    +        Used by ``get_or_create_session`` to self-heal at routing time:
    +        ``_prune_stale_sessions_locked`` only runs at startup, so a session
    +        ended in the DB while the gateway stays alive (any path that finalizes
    +        the row without clearing sessions.json) would otherwise be reused as a
    +        live routing key and silently swallow every subsequent message until
    +        the next restart (#54878 — the live-gateway variant of #52804/FM9).
    +        DB errors are non-fatal — never block routing on a failed lookup.
    +        """
    +        db = getattr(self, "_db", None)
    +        if not db or not session_id:
    +            return False
    +        try:
    +            row = db.get_session(session_id)
    +        except Exception:
    +            return False
    +        return bool(row is not None and row.get("end_reason") is not None)
    +
         def _should_reset(self, entry: SessionEntry, source: SessionSource) -> Optional[str]:
             """
             Check if a session should be reset based on policy.
    @@ -1243,39 +1269,73 @@ def get_or_create_session(
                 if session_key in self._entries and not force_new:
                     entry = self._entries[session_key]
     
    -                # Auto-reset sessions marked as suspended (e.g. after /stop
    -                # broke a stuck loop — #7536).  ``suspended`` is the hard
    -                # forced-wipe signal and always wins over ``resume_pending``,
    -                # so repeated interrupted restarts that escalate via the
    -                # existing ``.restart_failure_counts`` stuck-loop counter
    -                # still converge to a clean slate.
    -                if entry.suspended:
    -                    reset_reason = "suspended"
    -                elif entry.resume_pending:
    -                    # 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.
    -                    reset_reason = self._should_reset(entry, source)
    +                # Self-heal stale routing: if this session_key still points at
    +                # a session that has ALREADY been ended in state.db (end_reason
    +                # set), the in-memory sessions.json entry is stale.  Reusing it
    +                # would route every incoming message into a closed session and
    +                # silently drop it — with no log, no error, no response — until
    +                # the gateway restarts and _prune_stale_sessions_locked() clears
    +                # it (#54878 — the live-gateway variant of #52804/FM9, which
    +                # only the startup prune previously caught).
    +                #
    +                # Drop the stale entry and fall through to the recovery path
    +                # below.  Leaving db_end_session_id None routes us into
    +                # _recover_session_from_db, whose finder
    +                # (hermes_state.find_latest_gateway_session_for_peer) selects
    +                # rows WHERE `ended_at IS NULL OR end_reason = 'agent_close'`
    +                # — so it REOPENS gateway-cleanup-ended ('agent_close') rows and
    +                # resumes the SAME session_id (transcript preserved), but returns
    +                # None for any other end_reason (e.g. /new), which then correctly
    +                # starts a fresh session.
    +                if self._is_session_ended_in_db(entry.session_id):
    +                    logger.warning(
    +                        "gateway.session: routing key %r -> %s is ended in "
    +                        "state.db but still live in sessions.json; dropping "
    +                        "stale entry and recovering/recreating the session "
    +                        "(#54878)",
    +                        session_key, entry.session_id,
    +                    )
    +                    self._entries.pop(session_key, None)
    +                    was_auto_reset = False
    +                    auto_reset_reason = None
    +                    reset_had_activity = False
    +                    # Fall through to the recovery/create path below; the
    +                    # stale entry is gone so we must NOT consult its
    +                    # suspended/resume/reset state.
    +                else:
    +                    # Auto-reset sessions marked as suspended (e.g. after /stop
    +                    # broke a stuck loop — #7536).  ``suspended`` is the hard
    +                    # forced-wipe signal and always wins over ``resume_pending``,
    +                    # so repeated interrupted restarts that escalate via the
    +                    # existing ``.restart_failure_counts`` stuck-loop counter
    +                    # still converge to a clean slate.
    +                    if entry.suspended:
    +                        reset_reason = "suspended"
    +                    elif entry.resume_pending:
    +                        # 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.
    +                        reset_reason = self._should_reset(entry, source)
    +                        if not reset_reason:
    +                            entry.updated_at = now
    +                            self._save()
    +                            return entry
    +                    else:
    +                        reset_reason = self._should_reset(entry, source)
                         if not reset_reason:
                             entry.updated_at = now
                             self._save()
                             return entry
    -                else:
    -                    reset_reason = self._should_reset(entry, source)
    -                if not reset_reason:
    -                    entry.updated_at = now
    -                    self._save()
    -                    return entry
    -                else:
    -                    # Session is being auto-reset.
    -                    was_auto_reset = True
    -                    auto_reset_reason = reset_reason
    -                    # Track whether the expired session had any real conversation.
    -                    # total_tokens is never written (token counts migrated to
    -                    # agent-direct persistence) so it is always 0 — use
    -                    # last_prompt_tokens, which is updated on every turn.
    -                    reset_had_activity = entry.last_prompt_tokens > 0
    -                    db_end_session_id = entry.session_id
    +                    else:
    +                        # Session is being auto-reset.
    +                        was_auto_reset = True
    +                        auto_reset_reason = reset_reason
    +                        # Track whether the expired session had any real
    +                        # conversation.  total_tokens is never written (token
    +                        # counts migrated to agent-direct persistence) so it is
    +                        # always 0 — use last_prompt_tokens, updated every turn.
    +                        reset_had_activity = entry.last_prompt_tokens > 0
    +                        db_end_session_id = entry.session_id
                 else:
                     was_auto_reset = False
                     auto_reset_reason = None

  • tests/gateway/test_session_store_runtime_stale_guard.py

    @@ -0,0 +1,200 @@
    +"""Runtime self-heal for stale sessions.json routing entries (#54878).
    +
    +`_prune_stale_sessions_locked` only runs at gateway startup. A session ended
    +in state.db while the gateway stays alive (e.g. any path that finalizes the
    +row without clearing sessions.json) leaves a stale `session_key -> session_id`
    +mapping whose session has `end_reason` set. Before this fix,
    +`get_or_create_session` returned that stale entry as a live routing key (it
    +never consulted end_reason), so every subsequent message was silently routed
    +into a closed session and dropped — no log, no error, no response — until the
    +next restart pruned it.
    +
    +This is the live-gateway variant of #52804/FM9 (#52808/#54138 startup prune),
    +which required an actual gateway *crash*. Here the guard inside
    +`get_or_create_session` detects the ended row at routing time and drops the
    +stale entry, falling through to `_recover_session_from_db` (which reopens
    +`agent_close`-ended rows and resumes the SAME session_id, preserving the
    +transcript) or, failing recovery, to a fresh session.
    +"""
    +
    +from datetime import datetime, timedelta
    +from unittest.mock import MagicMock, patch
    +
    +from gateway.config import GatewayConfig, Platform, SessionResetPolicy
    +from gateway.session import SessionEntry, SessionSource, SessionStore
    +
    +
    +# ---------------------------------------------------------------------------
    +# Helpers
    +# ---------------------------------------------------------------------------
    +
    +def _make_entry(key: str, session_id: str, **kw) -> SessionEntry:
    +    now = datetime.now()
    +    return SessionEntry(
    +        session_key=key,
    +        session_id=session_id,
    +        created_at=now - timedelta(hours=2),
    +        updated_at=now - timedelta(hours=1),
    +        platform=Platform.TELEGRAM,
    +        chat_type="dm",
    +        **kw,
    +    )
    +
    +
    +def _db_returning(rows: dict) -> MagicMock:
    +    """SessionDB mock where get_session maps session_id -> row dict."""
    +    db = MagicMock()
    +    db.get_session.side_effect = lambda sid: rows.get(sid)
    +    # By default recovery finds nothing (forces a fresh session).
    +    db.find_latest_gateway_session_for_peer.return_value = None
    +    db.reopen_session.return_value = None
    +    db.create_session.return_value = None
    +    return db
    +
    +
    +def _make_store_with_db(tmp_path, db_mock) -> SessionStore:
    +    """Build a SessionStore with a mock SessionDB, bypassing disk load."""
    +    config = GatewayConfig(default_reset_policy=SessionResetPolicy(mode="none"))
    +    with patch("gateway.session.SessionStore._ensure_loaded"):
    +        store = SessionStore(sessions_dir=tmp_path, config=config)
    +    store._db = db_mock
    +    store._loaded = True
    +    return store
    +
    +
    +def _source() -> SessionSource:
    +    # session_key for this peer is deterministic; matches the entry key we seed.
    +    return SessionSource(
    +        platform=Platform.TELEGRAM,
    +        chat_id="8494508720",
    +        chat_type="dm",
    +        user_id="8494508720",
    +    )
    +
    +
    +# ---------------------------------------------------------------------------
    +# _is_session_ended_in_db helper
    +# ---------------------------------------------------------------------------
    +
    +class TestIsSessionEndedInDb:
    +    def test_ended_row_is_stale(self, tmp_path):
    +        db = _db_returning({"sid": {"end_reason": "agent_close", "id": "sid"}})
    +        store = _make_store_with_db(tmp_path, db)
    +        assert store._is_session_ended_in_db("sid") is True
    +
    +    def test_alive_row_not_stale(self, tmp_path):
    +        db = _db_returning({"sid": {"end_reason": None, "id": "sid"}})
    +        store = _make_store_with_db(tmp_path, db)
    +        assert store._is_session_ended_in_db("sid") is False
    +
    +    def test_absent_row_not_stale(self, tmp_path):
    +        # Not yet persisted / legacy — must NOT be treated as ended, else a
    +        # freshly-created in-memory session would be wrongly discarded.
    +        db = _db_returning({})
    +        store = _make_store_with_db(tmp_path, db)
    +        assert store._is_session_ended_in_db("sid_absent") is False
    +
    +    def test_no_db_not_stale(self, tmp_path):
    +        store = _make_store_with_db(tmp_path, _db_returning({}))
    +        store._db = None
    +        assert store._is_session_ended_in_db("sid") is False
    +
    +    def test_empty_session_id_not_stale(self, tmp_path):
    +        store = _make_store_with_db(tmp_path, _db_returning({}))
    +        assert store._is_session_ended_in_db("") is False
    +
    +    def test_db_error_not_stale(self, tmp_path):
    +        db = MagicMock()
    +        db.get_session.side_effect = Exception("DB locked")
    +        store = _make_store_with_db(tmp_path, db)
    +        # On error, never block routing — treat as not-stale (keep).
    +        assert store._is_session_ended_in_db("sid") is False
    +
    +
    +# ---------------------------------------------------------------------------
    +# get_or_create_session — runtime self-heal
    +# ---------------------------------------------------------------------------
    +
    +class TestRuntimeStaleGuard:
    +    def test_stale_agent_close_entry_recovered_preserving_session_id(self, tmp_path):
    +        """Stale `agent_close` entry → recovery reopens the SAME session_id."""
    +        source = _source()
    +        db = _db_returning({"sid_stale": {"end_reason": "agent_close", "id": "sid_stale"}})
    +        # Recovery finds the agent_close row and reopens it (transcript-preserving).
    +        db.find_latest_gateway_session_for_peer.return_value = {
    +            "id": "sid_stale",
    +            "started_at": (datetime.now() - timedelta(hours=2)).timestamp(),
    +        }
    +        store = _make_store_with_db(tmp_path, db)
    +        key = store._generate_session_key(source)
    +        store._entries[key] = _make_entry(key, "sid_stale")
    +
    +        result = store.get_or_create_session(source)
    +
    +        # SAME session_id (resumed), not a brand-new one, and not silently
    +        # routed into the closed entry.
    +        assert result.session_id == "sid_stale"
    +        db.reopen_session.assert_called_once_with("sid_stale")
    +        # A brand-new session row must NOT have been created.
    +        db.create_session.assert_not_called()
    +
    +    def test_stale_entry_creates_fresh_when_recovery_returns_none(self, tmp_path):
    +        """Stale entry, no recoverable row → brand-new session (no silent drop)."""
    +        source = _source()
    +        # Ended with a non-recoverable reason (e.g. /new) → finder returns None.
    +        db = _db_returning({"sid_stale": {"end_reason": "new_command", "id": "sid_stale"}})
    +        db.find_latest_gateway_session_for_peer.return_value = None
    +        store = _make_store_with_db(tmp_path, db)
    +        key = store._generate_session_key(source)
    +        store._entries[key] = _make_entry(key, "sid_stale")
    +
    +        result = store.get_or_create_session(source)
    +
    +        assert result.session_id != "sid_stale"
    +        # A fresh session row was created for the new session_id.
    +        db.create_session.assert_called_once()
    +        assert store._entries[key].session_id == result.session_id
    +
    +    def test_live_entry_returned_unchanged(self, tmp_path):
    +        """A session still alive in the DB is returned as-is (no churn)."""
    +        source = _source()
    +        db = _db_returning({"sid_live": {"end_reason": None, "id": "sid_live"}})
    +        store = _make_store_with_db(tmp_path, db)
    +        key = store._generate_session_key(source)
    +        store._entries[key] = _make_entry(key, "sid_live")
    +
    +        result = store.get_or_create_session(source)
    +
    +        assert result.session_id == "sid_live"
    +        db.find_latest_gateway_session_for_peer.assert_not_called()
    +        db.create_session.assert_not_called()
    +
    +    def test_stale_check_wins_over_suspended(self, tmp_path):
    +        """A stale entry that is ALSO suspended is still dropped via the stale
    +        path — we must not consult the dead entry's reset/suspend state."""
    +        source = _source()
    +        db = _db_returning({"sid_stale": {"end_reason": "agent_close", "id": "sid_stale"}})
    +        db.find_latest_gateway_session_for_peer.return_value = None  # → fresh
    +        store = _make_store_with_db(tmp_path, db)
    +        key = store._generate_session_key(source)
    +        store._entries[key] = _make_entry(key, "sid_stale", suspended=True)
    +
    +        result = store.get_or_create_session(source)
    +
    +        # Did not return the stale (suspended) entry; created a fresh session.
    +        assert result.session_id != "sid_stale"
    +        db.create_session.assert_called_once()
    +
    +    def test_force_new_skips_stale_check(self, tmp_path):
    +        """force_new short-circuits the whole existing-entry branch; the stale
    +        DB lookup must not even run."""
    +        source = _source()
    +        db = _db_returning({"sid_old": {"end_reason": "agent_close", "id": "sid_old"}})
    +        store = _make_store_with_db(tmp_path, db)
    +        key = store._generate_session_key(source)
    +        store._entries[key] = _make_entry(key, "sid_old")
    +
    +        result = store.get_or_create_session(source, force_new=True)
    +
    +        assert result.session_id != "sid_old"
    +        db.get_session.assert_not_called()

• #65783 fix(gateway): unify session reset boundaries vs recovery — stop resurrecting reset sessions

  merged to main

  files: docs/session-lifecycle.md, gateway/platforms/api_server.py, gateway/platforms/base.py, gateway/platforms/webhook.py, gateway/run.py, gateway/session.py, hermes_state.py, scripts/release.py, tests/gateway/test_clean_shutdown_marker.py, tests/gateway/test_restart_resume_pending.py, tests/gateway/test_session_reset_notify.py, tests/gateway/test_session_store_expiry_finalized.py, tests/gateway/test_session_store_lock_io.py, tests/gateway/test_session_store_runtime_stale_guard.py, tests/gateway/test_session_store_stale_prune.py

  Show the diff

  • docs/session-lifecycle.md

    @@ -548,7 +548,13 @@ The `_session_expiry_watcher` task runs in the gateway event loop every 300 seco
        - Clean up cached AIAgent resources (close tool resources, shut down memory provider).
        - Evict the cached agent entry.
        - Clear per-session overrides (`_session_model_overrides`, reasoning overrides, etc.).
    -   - Mark `expiry_finalized=True` and persist.
    +   - Mark `expiry_finalized=True` and persist (sessions.json + state.db).
    +   - Promote the state.db session row to `end_reason='session_reset'` via
    +     `promote_to_session_reset()` — conditional: only live rows or rows ended with a
    +     recoverable accidental reason (`agent_close`, `ws_orphan_reap`) are promoted, so
    +     explicit boundaries (`compression`, `session_switch`, …) are never overwritten. This
    +     durably records the reset so stale-route recovery cannot resurrect the expired
    +     session with its full history (#61220, #61993, #63539).
     
     2. **Sweep idle cached agents** — Calls `_sweep_idle_cached_agents()` to evict agents that
        have been idle beyond `_AGENT_CACHE_IDLE_TTL_SECS` (3600s / 1h), regardless of session

  • gateway/platforms/api_server.py

    @@ -932,6 +932,11 @@ class APIServerAdapter(BasePlatformAdapter):
         # ``async_delivery_supported()``.
         supports_async_delivery: bool = False
     
    +    # Same statelessness applies to the startup auto-resume prompt: no client
    +    # is waiting to answer "session restored — what next?", so a resumed turn
    +    # should complete the interrupted work rather than acknowledge (#57056).
    +    interactive_resume: bool = False
    +
         def __init__(self, config: PlatformConfig):
             super().__init__(config, Platform.API_SERVER)
             extra = config.extra or {}

  • gateway/platforms/base.py

    @@ -2365,6 +2365,19 @@ class BasePlatformAdapter(ABC):
         # generic seam; Slack is merely the first consumer).
         supports_inchannel_continuable: bool = False
     
    +    # Whether a human is interactively present on this platform to answer a
    +    # "session restored — what next?" prompt.  The startup auto-resume turn
    +    # (``_schedule_resume_pending_sessions`` → the ``_is_resume_pending``
    +    # branch in ``_handle_message_with_agent``) reads this to pick its
    +    # guidance: interactive platforms (Telegram, Slack, Discord DMs, …) get
    +    # "report the restore and ask what the user wants next"; non-interactive
    +    # event platforms (webhook) get "finish the interrupted work" because
    +    # nobody is there to answer, and an acknowledgement would silently
    +    # abandon the task (#57056).  Read generically via ``getattr(adapter,
    +    # "interactive_resume", True)`` — no per-platform branching at the call
    +    # site.
    +    interactive_resume: bool = True
    +
         # Back-reference to the running ``GatewayRunner``, injected by
         # ``gateway/run.py`` after the adapter is created. Adapters consume it via
         # ``getattr(self, "gateway_runner", None)`` for cross-platform delivery and

  • gateway/platforms/webhook.py

    @@ -147,6 +147,12 @@ def check_webhook_requirements() -> bool:
     class WebhookAdapter(BasePlatformAdapter):
         """Generic webhook receiver that triggers agent runs from HTTP POSTs."""
     
    +    # No human is present to answer a "session restored — what next?" prompt:
    +    # webhook runs are event-triggered.  The startup auto-resume turn must
    +    # instruct the model to FINISH the interrupted work instead of emitting an
    +    # interactive acknowledgement that abandons the task (#57056).
    +    interactive_resume: bool = False
    +
         def __init__(self, config: PlatformConfig):
             super().__init__(config, Platform.WEBHOOK)
             # ``host`` may be None (dual-stack default) or a user-pinned string.

  • gateway/run.py

    @@ -712,6 +712,75 @@ def _is_fresh_gateway_interruption(
         return current - timestamp <= window
     
     
    +def build_resume_recovery_note(
    +    reason: Optional[str],
    +    message: str = "",
    +    *,
    +    interactive: bool = True,
    +) -> str:
    +    """Build the resume-pending recovery system note for an interrupted turn.
    +
    +    ``reason`` is the session's ``resume_reason`` (``restart_timeout``,
    +    ``shutdown_timeout``, or anything else → generic interruption phrasing).
    +    ``message`` is the user's NEW message text; empty means this is the
    +    startup auto-resume turn synthesized by
    +    ``_schedule_resume_pending_sessions`` with no human message attached.
    +
    +    ``interactive`` selects the empty-message guidance: on interactive
    +    platforms a human is present, so "report the restore and ask what next"
    +    is right.  On non-interactive event platforms (webhook, API server —
    +    adapters with ``interactive_resume = False``) nobody can answer; the
    +    resumed turn must instead complete the interrupted work, or the task is
    +    silently abandoned behind a "restored" acknowledgement that goes
    +    nowhere (#57056).
    +    """
    +    reason_phrase = (
    +        "a gateway restart"
    +        if reason == "restart_timeout"
    +        else "a gateway shutdown"
    +        if reason == "shutdown_timeout"
    +        else "a gateway interruption"
    +    )
    +    if message:
    +        resume_guidance = (
    +            "Address the user's NEW message below FIRST and focus "
    +            "on what the user is asking now."
    +        )
    +        tail_guidance = (
    +            "Do NOT re-execute old tool calls — skip any "
    +            "unfinished work from the conversation history."
    +        )
    +    elif interactive:
    +        resume_guidance = (
    +            "Report to the user that the session was restored "
    +            "successfully and ask what they would like to do next."
    +        )
    +        tail_guidance = (
    +            "Do NOT re-execute old tool calls — skip any "
    +            "unfinished work from the conversation history."
    +        )
    +    else:
    +        resume_guidance = (
    +            "No user is present on this non-interactive platform, "
    +            "so do NOT emit a 'session restored' acknowledgement "
    +            "or ask questions. Review the conversation history and "
    +            "CONTINUE the interrupted task to completion."
    +        )
    +        tail_guidance = (
    +            "Do NOT re-run tool calls whose results already "
    +            "appear in the history — resume from the first step "
    +            "that has no recorded result."
    +        )
    +    return (
    +        f"[System note: The previous turn was interrupted by "
    +        f"{reason_phrase}; the gateway is now back online. "
    +        f"Any restart/shutdown command in the history has already "
    +        f"run — do NOT re-execute or verify it. {resume_guidance} "
    +        f"{tail_guidance}]"
    +        + (f"\n\n{message}" if message else "")
    +    )
    +
    +
     # Assistant-message fields that must survive transcript replay so multi-turn
     # reasoning context, prefix-cache hits, and provider-specific echo
     # requirements all behave the same on the gateway as they do in the CLI.
    @@ -6186,7 +6255,7 @@ async def _cleanup_agent_resources_off_loop(
             """
             if agent is None:
                 return
    -        if context.startswith("shutdown"):
    +        if context.startswith("shutdown") or context == "session expiry":
                 try:
                     agent._end_session_on_close = False
                 except Exception:
    @@ -11317,6 +11386,8 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str, run_g
                     context_note = "[System note: The user's previous session was stopped and suspended. This is a fresh conversation with no prior context.]"
                 elif reset_reason == "daily":
                     context_note = "[System note: The user's session was automatically reset by the daily schedule. This is a fresh conversation with no prior context.]"
    +            elif reset_reason == "resume_pending_expired":
    +                context_note = "[System note: The previous gateway session could not be recovered after a restart (API recovery timed out). This is a fresh conversation — use /resume to restore history if needed.]"
                 else:
                     context_note = "[System note: The user's previous session expired due to inactivity. This is a fresh conversation with no prior context.]"
                 context_prompt = context_note + "\n\n" + context_prompt
    @@ -11332,9 +11403,11 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str, run_g
                     )
                     platform_name = source.platform.value if source.platform else ""
                     had_activity = getattr(session_entry, 'reset_had_activity', False)
    -                # Suspended sessions always notify (they were explicitly stopped
    -                # or crashed mid-operation) — skip the policy check.
    -                should_notify = reset_reason == "suspended" or (
    +                # Suspended and restart-recovery-expired sessions always notify
    +                # regardless of policy.notify — the user had an active session
    +                # that was silently replaced, so they need to know they can
    +                # /resume it.  Idle/daily resets respect the policy flag.
    +                should_notify = reset_reason in {"suspended", "resume_pending_expired"} or (
                         policy.notify
                         and had_activity
                         and platform_name not in policy.notify_exclude_platforms
    @@ -11344,6 +11417,8 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str, run_g
                         if adapter:
                             if reset_reason == "suspended":
                                 reason_text = "previous session was stopped or interrupted"
    +                        elif reset_reason == "resume_pending_expired":
    +                            reason_text = "gateway restart recovery timed out"
                             elif reset_reason == "daily":
                                 reason_text = f"daily schedule at {policy.at_hour}:00"
                             else:
    @@ -19514,36 +19589,21 @@ def _approval_notify_sync(approval_data: dict) -> None:
     
                 if _is_resume_pending:
                     _reason = getattr(_resume_entry, "resume_reason", None) or "restart_timeout"
    -                _reason_phrase = (
    -                    "a gateway restart"
    -                    if _reason == "restart_timeout"
    -                    else "a gateway shutdown"
    -                    if _reason == "shutdown_timeout"
    -                    else "a gateway interruption"
    -                )
                     _persist_user_message_override = message
                     # The empty-message case is the auto-resume startup turn
                     # synthesized by _schedule_resume_pending_sessions — there is
    -                # no NEW user message to address, so tell the model to report
    -                # recovery instead of the (nonexistent) "new message".
    -                if message:
    -                    _resume_guidance = (
    -                        "Address the user's NEW message below FIRST and focus "
    -                        "on what the user is asking now."
    -                    )
    -                else:
    -                    _resume_guidance = (
    -                        "Report to the user that the session was restored "
    -                        "successfully and ask what they would like to do next."
    -                    )
    -                message = (
    -                    f"[System note: The previous turn was interrupted by "
    -                    f"{_reason_phrase}; the gateway is now back online. "
    -                    f"Any restart/shutdown command in the history has already "
    -                    f"run — do NOT re-execute or verify it. {_resume_guidance} "
    -                    f"Do NOT re-execute old tool calls — skip any unfinished "
    -                    f"work from the conversation history.]"
    -                    + (f"\n\n{message}" if message else "")
    +                # no NEW user message to address.  Guidance is adapter-aware:
    +                # interactive platforms report the restore and ask what next;
    +                # non-interactive event platforms (webhook, API server)
    +                # continue the interrupted work instead, because nobody is
    +                # present to answer and an acknowledgement would silently
    +                # abandon the task (#57056).
    +                _resume_adapter = self._adapter_for_source(source)
    +                _interactive_resume = bool(
    +                    getattr(_resume_adapter, "interactive_resume", True)
    +                )
    +                message = build_resume_recovery_note(
    +                    _reason, message, interactive=_interactive_resume,
                     )
                 elif _has_fresh_tool_tail:
                     _persist_user_message_override = message
    @@ -19584,22 +19644,13 @@ def _approval_notify_sync(approval_data: dict) -> None:
                     _sn_reason = (
                         getattr(_resume_entry, "resume_reason", None) or "restart_timeout"
                     )
    -                _sn_reason_phrase = (
    -                    "a gateway restart"
    -                    if _sn_reason == "restart_timeout"
    -                    else "a gateway shutdown"
    -                    if _sn_reason == "shutdown_timeout"
    -                    else "a gateway interruption"
    -                )
    -                message = (
    -                    f"[System note: The previous turn was interrupted by "
    -                    f"{_sn_reason_phrase}; the gateway is now back online. "
    -                    f"Any restart/shutdown command in the history has already "
    -                    f"run — do NOT re-execute or verify it. Report to the user "
    -                    f"that the session was restored successfully and ask what "
    -                    f"they would like to do next. Do NOT re-execute old tool "
    -                    f"calls — skip any unfinished work from the conversation "
    -                    f"history.]"
    +                _sn_adapter = self._adapter_for_source(source)
    +                message = build_resume_recovery_note(
    +                    _sn_reason,
    +                    "",
    +                    interactive=bool(
    +                        getattr(_sn_adapter, "interactive_resume", True)
    +                    ),
                     )
     
                 _approval_session_key = session_key or ""

  • gateway/session.py

    @@ -1043,6 +1043,21 @@ def __init__(self, sessions_dir: Path, config: GatewayConfig,
                 self._db = SessionDB()
             except Exception as e:
                 print(f"[gateway] Warning: SQLite session store unavailable, falling back to JSONL: {e}")
    +
    +    def _has_active_processes_safe(self, session_key: str, *, context: str) -> bool:
    +        """Return whether a session has active work, failing closed on registry errors."""
    +        if self._has_active_processes_fn is None:
    +            return False
    +        try:
    +            return bool(self._has_active_processes_fn(session_key))
    +        except Exception as exc:
    +            logger.warning(
    +                "has_active_processes_fn raised during %s for %s; keeping session alive: %s",
    +                context,
    +                session_key,
    +                exc,
    +            )
    +            return True
         
         def _ensure_loaded(self) -> None:
             """Load sessions index from disk if not already loaded."""
    @@ -1178,12 +1193,14 @@ def _prune_stale_sessions_locked(self) -> None:
                     # end_reason not None -> session ended — prune
                     if row is not None and row.get("end_reason") is not None:
                         recovered_entry = None
    +                    recovery_lookup_failed = False
                         if entry.origin is not None:
                             try:
                                 recovered_entry = self._recover_session_from_db(
                                     session_key=key,
                                     source=entry.origin,
                                     now=_now(),
    +                                raise_on_lookup_error=True,
                                 )
                             except Exception as exc:
                                 logger.debug(
    @@ -1193,6 +1210,10 @@ def _prune_stale_sessions_locked(self) -> None:
                                     entry.session_id,
                                     exc,
                                 )
    +                            recovery_lookup_failed = True
    +
    +                    if recovery_lookup_failed:
    +                        continue
     
                         # If the stale entry points at a compression-ended parent but
                         # a newer live child session exists for the exact same gateway
    @@ -1416,6 +1437,7 @@ def _recover_session_from_db(
             session_key: str,
             source: SessionSource,
             now: datetime,
    +        raise_on_lookup_error: bool = False,
         ) -> Optional[SessionEntry]:
             """Rebuild a missing session-key mapping from durable state.db data."""
             if not self._db:
    @@ -1434,6 +1456,8 @@ def _recover_session_from_db(
                 )
             except Exception as exc:
                 logger.debug("Gateway session DB recovery failed for %s: %s", session_key, exc)
    +            if raise_on_lookup_error:
    +                raise
                 return None
             if not recovered:
                 return None
    @@ -1582,6 +1606,22 @@ def set_expiry_finalized(
                             "Session DB expiry_finalized write failed for %s: %s",
                             entry.session_id, exc,
                         )
    +            try:
    +                # Expiry finalization is a real conversation boundary. Without
    +                # a durable ``session_reset`` end_reason, later agent cleanup can
    +                # close the row as ``agent_close``; stale-route recovery treats
    +                # that as resumable and resurrects the expired full history.
    +                #
    +                # promote_to_session_reset is conditional: it only promotes
    +                # live rows or rows ended with ``agent_close``.  Explicit
    +                # boundaries (compression, session_reset, new_command, etc.)
    +                # are preserved — the first writer wins.
    +                self._db.promote_to_session_reset(entry.session_id)
    +            except Exception as exc:
    +                logger.debug(
    +                    "Session DB promote_to_session_reset failed for %s: %s",
    +                    entry.session_id, exc,
    +                )
         
         def _is_session_expired(self, entry: SessionEntry) -> bool:
             """Check if a session has expired based on its reset policy.
    @@ -1590,13 +1630,12 @@ def _is_session_expired(self, entry: SessionEntry) -> bool:
             Used by the background expiry watcher to proactively flush memories.
             Sessions with active background processes are never considered expired.
             """
    -        if self._has_active_processes_fn:
    -            if self._has_active_processes_fn(entry.session_key):
    -                logger.debug(
    -                    "Session %s not expired — active background processes",
    -                    entry.session_key,
    -                )
    -                return False
    +        if self._has_active_processes_safe(entry.session_key, context="expiry"):
    +            logger.debug(
    +                "Session %s not expired — active background processes",
    +                entry.session_key,
    +            )
    +            return False
     
             policy = self.config.get_reset_policy(
                 platform=entry.platform,
    @@ -1691,14 +1730,13 @@ def _should_reset(self, entry: SessionEntry, source: SessionSource) -> Optional[
             
             Sessions with active background processes are never reset.
             """
    -        if self._has_active_processes_fn:
    -            session_key = self._generate_session_key(source)
    -            if self._has_active_processes_fn(session_key):
    -                logger.debug(
    -                    "Session reset skipped for %s — active background processes",
    -                    session_key,
    -                )
    -                return None
    +        session_key = self._generate_session_key(source)
    +        if self._has_active_processes_safe(session_key, context="reset"):
    +            logger.debug(
    +                "Session reset skipped for %s — active background processes",
    +                session_key,
    +            )
    +            return None
     
             policy = self.config.get_reset_policy(
                 platform=source.platform,
    @@ -1895,13 +1933,24 @@ def _get_or_create_session_impl(
                 elif _entry_for_checks.resume_pending:
                     _reset_reason = self._should_reset(_entry_for_checks, source)
                     if not _reset_reason:
    -                    _fw = auto_continue_freshness_window()
    -                    _ref_time = (
    -                        _entry_for_checks.last_resume_marked_at
    -                        or _entry_for_checks.updated_at
    +                    # Freshness-gate stale resume_pending zombies (#46934) —
    +                    # but honor an explicit ``session_reset.mode: none``: the
    +                    # user opted out of ALL automatic resets, so an expired
    +                    # resume marker must fall through to a normal resume of
    +                    # the preserved transcript, never a silent fresh session
    +                    # (#61052).
    +                    _policy = self.config.get_reset_policy(
    +                        platform=source.platform,
    +                        session_type=source.chat_type,
                         )
    -                    if _fw > 0 and (now - _ref_time).total_seconds() > _fw:
    -                        _reset_reason = "resume_pending_expired"
    +                    if _policy.mode != "none":
    +                        _fw = auto_continue_freshness_window()
    +                        _ref_time = (
    +                            _entry_for_checks.last_resume_marked_at
    +                            or _entry_for_checks.updated_at
    +                        )
    +                        if _fw > 0 and (now - _ref_time).total_seconds() > _fw:
    +                            _reset_reason = "resume_pending_expired"
                 else:
                     _reset_reason = self._should_reset(_entry_for_checks, source)
     
    @@ -1938,6 +1987,14 @@ def _get_or_create_session_impl(
                             session_key, entry.session_id,
                         )
                         self._entries.pop(session_key, None)
    +                    # If an expiry watcher (daily/idle reset) already finalized
    +                    # this session, honour the reset decision instead of silently
    +                    # reopening it via recovery.
    +                    if _reset_reason:
    +                        was_auto_reset = True
    +                        auto_reset_reason = _reset_reason
    +                        reset_had_activity = entry.last_prompt_tokens > 0
    +                        db_end_session_id = entry.session_id
                         entry = None
                         _needs_recover = True
                     elif entry.session_id != _stale_session_id:
    @@ -2023,8 +2080,21 @@ def _get_or_create_session_impl(
     
             # SQLite operations outside the lock (unchanged).
             if self._db and db_end_session_id:
    +            # Use the specific reset reason so state.db is auditable (e.g.
    +            # "resume_pending_expired" is distinguishable from a normal
    +            # "session_reset" caused by idle/daily expiry).
    +            _db_end_reason = auto_reset_reason if auto_reset_reason else "session_reset"
                 try:
    -                self._db.end_session(db_end_session_id, "session_reset")
    +                # promote_to_session_reset, not end_session: the row may
    +                # already be ended with a recoverable accidental reason
    +                # (agent_close / ws_orphan_reap), which first-reason-wins
    +                # end_session would preserve — leaving the reset session
    +                # resurrectable by stale-route recovery (#61220, #61993).
    +                _promote = getattr(self._db, "promote_to_session_reset", None)
    +                if callable(_promote):
    +                    _promote(db_end_session_id, _db_end_reason)
    +                else:
    +                    self._db.end_session(db_end_session_id, _db_end_reason)
                 except Exception as e:
                     logger.debug("Session DB operation failed: %s", e)
     
    @@ -2193,19 +2263,8 @@ def prune_old_entries(self, max_age_days: int) -> int:
                     # The callback is keyed by session_key (see process_registry.
                     # has_active_for_session); passing session_id here used to
                     # never match, so active sessions got pruned anyway.
    -                if self._has_active_processes_fn is not None:
    -                    try:
    -                        if self._has_active_processes_fn(entry.session_key):
    -                            continue
    -                    except Exception as exc:
    -                        logger.debug(
    -                            "has_active_processes_fn raised during prune for %s: %s",
    -                            entry.session_key, exc,
    -                        )
    -                        # Fail safe: if we can't tell whether a background
    -                        # process is attached, keep the entry rather than
    -                        # risk orphaning live work.
    -                        continue
    +                if self._has_active_processes_safe(entry.session_key, context="prune"):
    +                    continue
                     if entry.updated_at < cutoff:
                         removed_keys.append(key)
                 for key in removed_keys:
    @@ -2301,7 +2360,15 @@ def reset_session(self, session_key: str, display_name: Optional[str] = None) ->
     
             if self._db and db_end_session_id:
                 try:
    -                self._db.end_session(db_end_session_id, "session_reset")
    +                # Promote (not plain end_session): an accidental
    +                # agent_close/ws_orphan_reap end must not survive an explicit
    +                # user reset, or recovery resurrects the reset session
    +                # (#61993 — the user's /new was silently undone).
    +                _promote = getattr(self._db, "promote_to_session_reset", None)
    +                if callable(_promote):
    +                    _promote(db_end_session_id, "session_reset")
    +                else:
    +                    self._db.end_session(db_end_session_id, "session_reset")
                 except Exception as e:
                     logger.debug("Session DB operation failed: %s", e)
     
    @@ -2362,7 +2429,15 @@ def switch_session(self, session_key: str, target_session_id: str) -> Optional[S
     
             if self._db and db_end_session_id:
                 try:
    -                self._db.end_session(db_end_session_id, "session_switch")
    +                # Promote (not plain end_session): a stale agent_close /
    +                # ws_orphan_reap end on the outgoing session must be upgraded
    +                # to the explicit switch boundary, or recovery can resurrect
    +                # it over the user's /resume choice (#61220 bug class).
    +                _promote = getattr(self._db, "promote_to_session_reset", None)
    +                if callable(_promote):
    +                    _promote(db_end_session_id, "session_switch")
    +                else:
    +                    self._db.end_session(db_end_session_id, "session_switch")
                 except Exception as e:
                     logger.debug("Session DB end_session failed: %s", e)
     

  • hermes_state.py

    @@ -2286,6 +2286,55 @@ def _do(conn):
                 )
             self._execute_write(_do)
     
    +    def promote_to_session_reset(
    +        self, session_id: str, reason: str = "session_reset"
    +    ) -> bool:
    +        """Durably mark a session as ended by an intentional reset boundary.
    +
    +        Promotes *only* live rows (``ended_at IS NULL``) or rows carrying an
    +        accidental end_reason that the recovery query
    +        (``find_latest_gateway_session_for_peer``) treats as recoverable:
    +        ``agent_close`` (older gateway cleanup bug) and ``ws_orphan_reap``
    +        (mistaken TUI reaper).  Explicit conversation boundaries such as
    +        ``compression``, ``session_reset``, ``session_switch``, etc. are
    +        preserved — the first writer wins for those, and a later expiry
    +        finalization must not silently overwrite them.
    +
    +        Plain ``end_session()`` is NOT sufficient for reset boundaries: it
    +        no-ops on an already-ended row, so a row that agent cleanup already
    +        closed as ``agent_close`` would stay recoverable and stale-route
    +        recovery would resurrect the reset session with its full history
    +        (#61220, #61993, #63539).
    +
    +        Keep this promotion set in sync with the recoverable set in
    +        ``find_latest_gateway_session_for_peer`` — any reason recovery would
    +        reopen must be promotable here.
    +
    +        ``reason`` lets reset paths keep their auditable specific reasons
    +        (``idle``, ``daily``, ``suspended``, ``resume_pending_expired``).
    +
    +        Returns ``True`` when the row was promoted, ``False`` when skipped
    +        (already has a different explicit end_reason, or row not found).
    +        """
    +        if not session_id:
    +            return False
    +        now = time.time()
    +
    +        def _do(conn):
    +            cursor = conn.execute(
    +                "UPDATE sessions SET ended_at = ?, end_reason = ? "
    +                "WHERE id = ? AND (ended_at IS NULL "
    +                "OR end_reason IN ('agent_close', 'ws_orphan_reap'))",
    +                (now, reason, session_id),
    +            )
    +            return cursor.rowcount
    +
    +        try:
    +            rows = self._execute_write(_do)
    +            return bool(rows)
    +        except Exception:
    +            return False
    +
         def update_session_cwd(
             self, session_id: str, cwd: str, git_branch: str = None, git_repo_root: str = None
         ) -> None:

  • scripts/release.py

    @@ -314,6 +314,8 @@
         "[redacted:email]": "Diyoncrz18",
         "[redacted:email]": "ali-nld",
         "[redacted:email]": "necoweb3",
    +    "[redacted:email]": "DNAlec",  # PR #61743 salvage (honor reset policy in #54878 stale-heal recovery)
    +    "[redacted:email]": "joelbrilliant",  # PR #58486 salvage (session-expiry cleanup must not end row as agent_close)
         "[redacted:email]": "hydracoco7",  # PR #61382 salvage (id-less cron job freeze)
         "[redacted:email]": "AlexFucuson9",  # PR #61209 salvage (hygiene compression data loss)
         "[redacted:email]": "adambiggs",  # PR #43819 salvage (holographic shared SQLite connection)

  • tests/gateway/test_clean_shutdown_marker.py

    @@ -245,6 +245,33 @@ async def _run():
             assert agent._end_session_on_close is False
             agent.close.assert_called_once()
     
    +    def test_session_expiry_cleanup_preserves_lazy_reset_boundary(self, tmp_path, monkeypatch):
    +        """Session expiry cleanup must not turn an expired chat into an agent_close row.
    +
    +        The expiry watcher only tears down cached resources. The next inbound
    +        message owns the reset boundary, creating a fresh session with the
    +        normal auto-reset notice. If cleanup lets ``agent.close()`` end the
    +        SQLite row as ``agent_close``, stale-route recovery treats it as
    +        recoverable and resurrects the expired session instead.
    +        """
    +        monkeypatch.setattr("gateway.run._hermes_home", tmp_path)
    +        from gateway.run import GatewayRunner
    +
    +        runner = object.__new__(GatewayRunner)
    +        agent = MagicMock()
    +        agent._end_session_on_close = True
    +
    +        async def _run():
    +            await GatewayRunner._cleanup_agent_resources_off_loop(
    +                runner, agent, context="session expiry"
    +            )
    +
    +        import asyncio
    +        asyncio.get_event_loop().run_until_complete(_run())
    +
    +        assert agent._end_session_on_close is False
    +        agent.close.assert_called_once()
    +
     
     # ---------------------------------------------------------------------------
     # resume_pending freshness gate (#46934)
    @@ -285,7 +312,12 @@ def test_fresh_resume_pending_returns_same_session(self, tmp_path):
     
         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)
    +        # The freshness gate only applies when the user has opted into
    +        # automatic resets — session_reset.mode: none disables it (#61052).
    +        from gateway.config import SessionResetPolicy
    +        store = _make_store(
    +            tmp_path, policy=SessionResetPolicy(mode="idle", idle_minutes=999999)
    +        )
             source = _make_source()
             entry = self._mark_resume_pending(store, source)
     
    @@ -302,6 +334,25 @@ def test_stale_resume_pending_falls_through_to_reset(self, tmp_path, monkeypatch
             assert fresh.session_id != entry.session_id
             assert not fresh.resume_pending
     
    +    def test_reset_mode_none_disables_freshness_gate(self, tmp_path, monkeypatch):
    +        """session_reset.mode: none opts out of ALL automatic resets —
    +        including the resume_pending freshness gate (#61052)."""
    +        monkeypatch.setenv("HERMES_AUTO_CONTINUE_FRESHNESS", "3600")
    +        from gateway.config import SessionResetPolicy
    +        store = _make_store(tmp_path, policy=SessionResetPolicy(mode="none"))
    +        source = _make_source()
    +        entry = self._mark_resume_pending(store, source)
    +
    +        with store._lock:
    +            entry.last_resume_marked_at = datetime.now() - timedelta(seconds=7200)
    +            entry.updated_at = datetime.now()
    +            store._save()
    +
    +        refreshed = store.get_or_create_session(source)
    +        # Explicit opt-out honored: same session back, transcript preserved.
    +        assert refreshed.session_id == entry.session_id
    +        assert refreshed.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")

  • tests/gateway/test_restart_resume_pending.py

    @@ -41,6 +41,7 @@
         _is_fresh_gateway_interruption,
         _last_transcript_timestamp,
         _should_clear_resume_pending_after_turn,
    +    build_resume_recovery_note,
     )
     from gateway.session import SessionEntry, SessionSource, SessionStore
     from tests.gateway.restart_test_helpers import (
    @@ -152,32 +153,9 @@ def _simulate_note_injection(
     
         if is_resume_pending:
             reason = getattr(resume_entry, "resume_reason", None) or "restart_timeout"
    -        reason_phrase = (
    -            "a gateway restart"
    -            if reason == "restart_timeout"
    -            else "a gateway shutdown"
    -            if reason == "shutdown_timeout"
    -            else "a gateway interruption"
    -        )
    -        if message:
    -            resume_guidance = (
    -                "Address the user's NEW message below FIRST and focus "
    -                "on what the user is asking now."
    -            )
    -        else:
    -            resume_guidance = (
    -                "Report to the user that the session was restored "
    -                "successfully and ask what they would like to do next."
    -            )
    -        message = (
    -            f"[System note: The previous turn was interrupted by "
    -            f"{reason_phrase}; the gateway is now back online. "
    -            f"Any restart/shutdown command in the history has already "
    -            f"run — do NOT re-execute or verify it. {resume_guidance} "
    -            f"Do NOT re-execute old tool calls — skip any unfinished "
    -            f"work from the conversation history.]"
    -            + (f"\n\n{message}" if message else "")
    -        )
    +        # Real production note builder — extracted to module scope in
    +        # gateway/run.py so tests exercise the actual strings.
    +        message = build_resume_recovery_note(reason, message)
         elif has_fresh_tool_tail:
             message = (
                 "[System note: A new message has arrived. The conversation "
    @@ -196,23 +174,7 @@ def _simulate_note_injection(
             and getattr(resume_entry, "resume_pending", False)
         ):
             sn_reason = getattr(resume_entry, "resume_reason", None) or "restart_timeout"
    -        sn_reason_phrase = (
    -            "a gateway restart"
    -            if sn_reason == "restart_timeout"
    -            else "a gateway shutdown"
    -            if sn_reason == "shutdown_timeout"
    -            else "a gateway interruption"
    -        )
    -        message = (
    -            f"[System note: The previous turn was interrupted by "
    -            f"{sn_reason_phrase}; the gateway is now back online. "
    -            f"Any restart/shutdown command in the history has already "
    -            f"run — do NOT re-execute or verify it. Report to the user "
    -            f"that the session was restored successfully and ask what "
    -            f"they would like to do next. Do NOT re-execute old tool "
    -            f"calls — skip any unfinished work from the conversation "
    -            f"history.]"
    -        )
    +        message = build_resume_recovery_note(sn_reason, "")
         return message
     
     
    @@ -521,6 +483,34 @@ def test_resume_pending_shutdown_note_mentions_shutdown(self):
             )
             assert "gateway shutdown" in result
     
    +    def test_empty_message_interactive_note_asks_what_next(self):
    +        """Interactive platforms: the startup auto-resume turn reports the
    +        restore and asks the (present) human what to do next."""
    +        note = build_resume_recovery_note("restart_timeout", "", interactive=True)
    +        assert "session was restored" in note
    +        assert "ask what they would like to do next" in note
    +        assert "skip any unfinished work" in note
    +
    +    def test_empty_message_noninteractive_note_continues_task(self):
    +        """Non-interactive platforms (webhook, API server): nobody can answer
    +        'what next?', so the resumed turn must complete the interrupted work
    +        instead of acknowledging (#57056)."""
    +        note = build_resume_recovery_note("restart_timeout", "", interactive=False)
    +        assert "CONTINUE the interrupted task" in note
    +        assert "session was restored" not in note
    +        assert "ask what they would like to do next" not in note
    +        # Must not tell the model to skip the unfinished work it should finish.
    +        assert "skip any unfinished work" not in note
    +        # But still guards against re-running already-recorded tool calls.
    +        assert "already appear in the history" in note
    +
    +    def test_new_message_guidance_identical_regardless_of_interactivity(self):
    +        """A real NEW user message always wins — same guidance either way."""
    +        a = build_resume_recovery_note("restart_timeout", "do the thing", interactive=True)
    +        b = build_resume_recovery_note("restart_timeout", "do the thing", interactive=False)
    +        assert a == b
    +        assert "NEW message" in a
    +
         def test_resume_pending_fires_without_tool_tail(self):
             """Key improvement over PR #9934: the restart-resume note fires
             even when the transcript's last role is NOT ``tool``."""

  • tests/gateway/test_session_reset_notify.py

    @@ -5,10 +5,11 @@
     - SessionEntry captures auto_reset_reason
     - SessionResetPolicy.notify controls whether notifications are sent
     - notify_exclude_platforms skips notifications for excluded platforms
    +- resume_pending_expired auto-reset sets the correct reason and DB end_reason
     """
     
     from datetime import datetime, timedelta
    -
    +from unittest.mock import MagicMock, patch
     
     from gateway.config import (
         GatewayConfig,
    @@ -30,11 +31,15 @@ def _make_source(platform=Platform.TELEGRAM, chat_id="123", user_id="u1"):
         )
     
     
    -def _make_store(policy=None, tmp_path=None):
    +def _make_store(policy=None, tmp_path=None, has_active_processes_fn=None):
         config = GatewayConfig()
         if policy:
             config.default_reset_policy = policy
    -    store = SessionStore(sessions_dir=tmp_path or "/tmp/test-sessions", config=config)
    +    store = SessionStore(
    +        sessions_dir=tmp_path or "/tmp/test-sessions",
    +        config=config,
    +        has_active_processes_fn=has_active_processes_fn,
    +    )
         return store
     
     
    @@ -100,6 +105,45 @@ def test_returns_none_when_mode_is_none(self, tmp_path):
             source = _make_source()
             assert store._should_reset(entry, source) is None
     
    +    def test_returns_none_when_active_process_check_raises(self, tmp_path):
    +        def _raise(_session_key):
    +            raise RuntimeError("process registry unavailable")
    +
    +        store = _make_store(
    +            SessionResetPolicy(mode="idle", idle_minutes=30),
    +            tmp_path,
    +            has_active_processes_fn=_raise,
    +        )
    +        entry = SessionEntry(
    +            session_key="test",
    +            session_id="s1",
    +            created_at=datetime.now() - timedelta(hours=2),
    +            updated_at=datetime.now() - timedelta(hours=1),
    +        )
    +        source = _make_source()
    +
    +        assert store._should_reset(entry, source) is None
    +
    +    def test_is_session_expired_fails_closed_when_active_process_check_raises(self, tmp_path):
    +        def _raise(_session_key):
    +            raise RuntimeError("process registry unavailable")
    +
    +        store = _make_store(
    +            SessionResetPolicy(mode="idle", idle_minutes=30),
    +            tmp_path,
    +            has_active_processes_fn=_raise,
    +        )
    +        entry = SessionEntry(
    +            session_key="test",
    +            session_id="s1",
    +            platform=Platform.TELEGRAM,
    +            chat_type="dm",
    +            created_at=datetime.now() - timedelta(hours=2),
    +            updated_at=datetime.now() - timedelta(hours=1),
    +        )
    +
    +        assert store._is_session_expired(entry) is False
    +
     
     # ---------------------------------------------------------------------------
     # SessionEntry captures reason
    @@ -278,3 +322,161 @@ def test_auto_reset_reason_none_roundtrip(self, tmp_path):
             assert reloaded.was_auto_reset is False
             assert reloaded.auto_reset_reason is None
             assert reloaded.reset_had_activity is False
    +
    +
    +# ---------------------------------------------------------------------------
    +# resume_pending_expired: auto_reset_reason and DB end_reason (#58933)
    +# ---------------------------------------------------------------------------
    +
    +def _make_db_mock() -> MagicMock:
    +    """Return a SessionDB mock with safe defaults for all lookup methods."""
    +    db = MagicMock()
    +    db.get_session.return_value = None
    +    db.get_compression_tip.return_value = None  # avoids MagicMock leaking into session_id
    +    db.find_latest_gateway_session_for_peer.return_value = None
    +    db.reopen_session.return_value = None
    +    db.create_session.return_value = None
    +    return db
    +
    +
    +def _make_store_with_db(tmp_path, db_mock=None, policy=None) -> SessionStore:
    +    """Build a SessionStore with a mock SessionDB, bypassing disk load."""
    +    cfg_policy = policy or SessionResetPolicy(mode="none")
    +    config = GatewayConfig(default_reset_policy=cfg_policy)
    +    with patch("gateway.session.SessionStore._ensure_loaded"):
    +        store = SessionStore(sessions_dir=tmp_path, config=config)
    +    store._db = db_mock if db_mock is not None else _make_db_mock()
    +    store._loaded = True
    +    return store
    +
    +
    +class TestResumePendingExpiredAutoReset:
    +    """resume_pending sessions past the freshness window should fire
    +    was_auto_reset=True with auto_reset_reason='resume_pending_expired' and
    +    persist that reason to state.db (#58933)."""
    +
    +    def _seed_stale_resume_pending(self, store, source, freshness_seconds=3600):
    +        """Create a session, mark it resume_pending, then backdate the mark
    +        past the freshness window so get_or_create_session treats it as a
    +        zombie."""
    +        entry = store.get_or_create_session(source)
    +        store.mark_resume_pending(entry.session_key)
    +        with store._lock:
    +            entry = store._entries[entry.session_key]
    +            entry.last_resume_marked_at = (
    +                datetime.now() - timedelta(seconds=freshness_seconds + 60)
    +            )
    +            entry.updated_at = datetime.now()  # keep updated_at fresh
    +            store._save()
    +        return entry
    +
    +    def test_stale_resume_pending_sets_auto_reset_reason(
    +        self, tmp_path, monkeypatch
    +    ):
    +        """Stale resume_pending triggers was_auto_reset=True with reason
    +        'resume_pending_expired', NOT 'idle'."""
    +        monkeypatch.setenv("HERMES_AUTO_CONTINUE_FRESHNESS", "3600")
    +        # The freshness gate requires an opted-in reset policy — mode "none"
    +        # disables it entirely (#61052). Use a huge idle window so only the
    +        # freshness gate (not the idle policy) can fire.
    +        store = _make_store_with_db(
    +            tmp_path,
    +            policy=SessionResetPolicy(mode="idle", idle_minutes=999999),
    +        )
    +        source = _make_source()
    +
    +        old = self._seed_stale_resume_pending(store, source)
    +
    +        new = store.get_or_create_session(source)
    +
    +        assert new.session_id != old.session_id, "should have created a new session"
    +        assert new.was_auto_reset is True
    +        assert new.auto_reset_reason == "resume_pending_expired"
    +
    +    def test_stale_resume_pending_had_activity_flag(
    +        self, tmp_path, monkeypatch
    +    ):
    +        """reset_had_activity reflects whether the old session was used."""
    +        monkeypatch.setenv("HERMES_AUTO_CONTINUE_FRESHNESS", "3600")
    +        store = _make_store_with_db(
    +            tmp_path,
    +            policy=SessionResetPolicy(mode="idle", idle_minutes=999999),
    +        )
    +        source = _make_source()
    +
    +        old = self._seed_stale_resume_pending(store, source)
    +        # Simulate some conversation on the old session.
    +        with store._lock:
    +            old.last_prompt_tokens = [redacted:secret-assignment]
    +            store._save()
    +
    +        new = store.get_or_create_session(source)
    +        assert new.reset_had_activity is True
    +
    +    def test_stale_resume_pending_db_end_reason_is_specific(
    +        self, tmp_path, monkeypatch
    +    ):
    +        """state.db must record end_reason='resume_pending_expired', NOT the
    +        generic 'session_reset', so the event is auditable (#58933 fix)."""
    +        monkeypatch.setenv("HERMES_AUTO_CONTINUE_FRESHNESS", "3600")
    +        db = _make_db_mock()
    +        store = _make_store_with_db(
    +            tmp_path, db,
    +            policy=SessionResetPolicy(mode="idle", idle_minutes=999999),
    +        )
    +        source = _make_source()
    +
    +        old = self._seed_stale_resume_pending(store, source)
    +        store.get_or_create_session(source)
    +
    +        # Auto-reset now writes through promote_to_session_reset so an
    +        # accidental agent_close end can't shadow the reset boundary.
    +        db.promote_to_session_reset.assert_called_once()
    +        ended_id, ended_reason = db.promote_to_session_reset.call_args.args
    +        assert ended_id == old.session_id
    +        assert ended_reason == "resume_pending_expired", (
    +            f"expected 'resume_pending_expired', got {ended_reason!r} — "
    +            "the DB end_reason must not be the generic 'session_reset'"
    +        )
    +
    +    def test_idle_reset_db_end_reason_reflects_idle(
    +        self, tmp_path
    +    ):
    +        """Regular idle auto-reset persists 'idle' as end_reason so that all
    +        auto-reset paths are auditable (#58933 should not regress the common
    +        idle/daily path)."""
    +        db = _make_db_mock()
    +        store = _make_store_with_db(
    +            tmp_path, db, policy=SessionResetPolicy(mode="idle", idle_minutes=1)
    +        )
    +        source = _make_source()
    +
    +        entry = store.get_or_create_session(source)
    +        # Age past idle threshold.
    +        with store._lock:
    +            entry.updated_at = datetime.now() - timedelta(minutes=5)
    +            store._save()
    +
    +        store.get_or_create_session(source)
    +
    +        db.promote_to_session_reset.assert_called_once()
    +        _, ended_reason = db.promote_to_session_reset.call_args.args
    +        assert ended_reason == "idle"
    +
    +    def test_freshness_disabled_skips_resume_pending_expired(
    +        self, tmp_path, monkeypatch
    +    ):
    +        """When gateway_auto_continue_freshness=0, resume_pending is never
    +        expired — the same session is returned regardless of age."""
    +        monkeypatch.setenv("HERMES_AUTO_CONTINUE_FRESHNESS", "0")
    +        db = _make_db_mock()
    +        store = _make_store_with_db(tmp_path, db)
    +        source = _make_source()
    +
    +        old = self._seed_stale_resume_pending(store, source, freshness_seconds=999_999)
    +
    +        refreshed = store.get_or_create_session(source)
    +        # Freshness disabled → same session, no DB end_session call.
    +        assert refreshed.session_id == old.session_id
    +        db.end_session.assert_not_called()
    +        db.promote_to_session_reset.assert_not_called()

  • tests/gateway/test_session_store_expiry_finalized.py

    @@ -0,0 +1,184 @@
    +"""Session expiry finalization closes sessions as session_reset.
    +
    +Regression coverage for #61220: the expiry watcher marks a session expired,
    +then agent cleanup can close it as ``agent_close``. Stale routing recovery treats
    +``agent_close`` as recoverable, so expired sessions were reopened with full
    +history unless expiry finalization also persisted the real conversation boundary
    +as ``end_reason='session_reset'``.
    +
    +These tests use a real ``SessionDB`` (in-memory) to verify the actual recovery
    +contract in ``find_latest_gateway_session_for_peer`` — not just call counts on
    +a MagicMock.
    +"""
    +
    +from __future__ import annotations
    +
    +import time
    +from pathlib import Path
    +
    +import pytest
    +
    +from hermes_state import SessionDB
    +
    +
    [redacted:email]
    +def db(tmp_path: Path) -> SessionDB:
    +    return SessionDB(tmp_path / "state.db")
    +
    +
    +_SESSION_KEY = "agent:main:telegram:dm:8494508720"
    +_SOURCE = "telegram"
    +_USER_ID = "8494508720"
    +
    +
    +# ------------------------------------------------------------------
    +# promote_to_session_reset — unit tests on real DB
    +# ------------------------------------------------------------------
    +
    +class TestPromoteToSessionReset:
    +    """promote_to_session_reset promotes only safe rows."""
    +
    +    def test_promotes_live_row(self, db: SessionDB) -> None:
    +        """A live row (ended_at IS NULL) is promoted to session_reset."""
    +        db.create_session(
    +            "sid-live", _SOURCE,
    +            user_id=_USER_ID, session_key=_SESSION_KEY,
    +            chat_id="8494508720", chat_type="dm",
    +        )
    +        # Seed a message so recovery would match
    +        db.append_message("sid-live", "user", "hello")
    +
    +        assert db.promote_to_session_reset("sid-live") is True
    +        row = db.get_session("sid-live")
    +        assert row["end_reason"] == "session_reset"
    +        assert row["ended_at"] is not None
    +
    +    def test_promotes_agent_close_row(self, db: SessionDB) -> None:
    +        """A row ended with agent_close is promoted to session_reset."""
    +        db.create_session(
    +            "sid-ac", _SOURCE,
    +            user_id=_USER_ID, session_key=_SESSION_KEY,
    +            chat_id="8494508720", chat_type="dm",
    +        )
    +        db.append_message("sid-ac", "user", "hello")
    +        db.end_session("sid-ac", "agent_close")
    +
    +        assert db.promote_to_session_reset("sid-ac") is True
    +        row = db.get_session("sid-ac")
    +        assert row["end_reason"] == "session_reset"
    +
    +    def test_does_not_overwrite_compression(self, db: SessionDB) -> None:
    +        """An existing compression boundary must not be overwritten."""
    +        db.create_session(
    +            "sid-comp", _SOURCE,
    +            user_id=_USER_ID, session_key=_SESSION_KEY,
    +            chat_id="8494508720", chat_type="dm",
    +        )
    +        db.end_session("sid-comp", "compression")
    +
    +        assert db.promote_to_session_reset("sid-comp") is False
    +        row = db.get_session("sid-comp")
    +        assert row["end_reason"] == "compression"
    +
    +    def test_does_not_overwrite_existing_session_reset(self, db: SessionDB) -> None:
    +        """Already-promoted rows are idempotently skipped."""
    +        db.create_session(
    +            "sid-reset", _SOURCE,
    +            user_id=_USER_ID, session_key=_SESSION_KEY,
    +            chat_id="8494508720", chat_type="dm",
    +        )
    +        db.end_session("sid-reset", "session_reset")
    +
    +        # Should be a no-op (rowcount = 0)
    +        assert db.promote_to_session_reset("sid-reset") is False
    +        row = db.get_session("sid-reset")
    +        assert row["end_reason"] == "session_reset"
    +
    +    def test_does_not_overwrite_new_command(self, db: SessionDB) -> None:
    +        """A /new-command boundary is preserved."""
    +        db.create_session(
    +            "sid-new", _SOURCE,
    +            user_id=_USER_ID, session_key=_SESSION_KEY,
    +            chat_id="8494508720", chat_type="dm",
    +        )
    +        db.end_session("sid-new", "new_command")
    +
    +        assert db.promote_to_session_reset("sid-new") is False
    +        row = db.get_session("sid-new")
    +        assert row["end_reason"] == "new_command"
    +
    +    def test_noop_on_missing_session(self, db: SessionDB) -> None:
    +        """Non-existent session_id returns False without error."""
    +        assert db.promote_to_session_reset("nonexistent") is False
    +
    +    def test_noop_on_empty_session_id(self, db: SessionDB) -> None:
    +        assert db.promote_to_session_reset("") is False
    +
    +
    +# ------------------------------------------------------------------
    +# Integration: promotion blocks stale-route recovery
    +# ------------------------------------------------------------------
    +
    +class TestPromotionBlocksRecovery:
    +    """After promotion, find_latest_gateway_session_for_peer must NOT
    +    recover the session — it is now ended with session_reset, which
    +    the recovery query excludes (only live/agent_close are recoverable).
    +    """
    +
    +    def test_live_session_recoverable_before_promotion(self, db: SessionDB) -> None:
    +        db.create_session(
    +            "sid-pre", _SOURCE,
    +            user_id=_USER_ID, session_key=_SESSION_KEY,
    +            chat_id="8494508720", chat_type="dm",
    +        )
    +        db.append_message("sid-pre", "user", "hello")
    +
    +        recovered = db.find_latest_gateway_session_for_peer(
    +            source=_SOURCE, session_key=_SESSION_KEY,
    +            user_id=_USER_ID, chat_id="8494508720", chat_type="dm",
    +        )
    +        assert recovered is not None
    +        assert recovered["id"] == "sid-pre"
    +
    +    def test_promoted_session_not_recoverable(self, db: SessionDB) -> None:
    +        db.create_session(
    +            "sid-post", _SOURCE,
    +            user_id=_USER_ID, session_key=_SESSION_KEY,
    +            chat_id="8494508720", chat_type="dm",
    +        )
    +        db.append_message("sid-post", "user", "hello")
    +        db.promote_to_session_reset("sid-post")
    +
    +        recovered = db.find_latest_gateway_session_for_peer(
    +            source=_SOURCE, session_key=_SESSION_KEY,
    +            user_id=_USER_ID, chat_id="8494508720", chat_type="dm",
    +        )
    +        # session_reset rows are not in the recovery set
    +        assert recovered is None
    +
    +    def test_agent_close_session_recoverable_but_not_after_promotion(
    +        self, db: SessionDB
    +    ) -> None:
    +        db.create_session(
    +            "sid-ac-rec", _SOURCE,
    +            user_id=_USER_ID, session_key=_SESSION_KEY,
    +            chat_id="8494508720", chat_type="dm",
    +        )
    +        db.append_message("sid-ac-rec", "user", "hello")
    +        db.end_session("sid-ac-rec", "agent_close")
    +
    +        # agent_close is recoverable
    +        recovered = db.find_latest_gateway_session_for_peer(
    +            source=_SOURCE, session_key=_SESSION_KEY,
    +            user_id=_USER_ID, chat_id="8494508720", chat_type="dm",
    +        )
    +        assert recovered is not None
    +        assert recovered["id"] == "sid-ac-rec"
    +
    +        # After promotion, it is no longer recoverable
    +        db.promote_to_session_reset("sid-ac-rec")
    +        recovered2 = db.find_latest_gateway_session_for_peer(
    +            source=_SOURCE, session_key=_SESSION_KEY,
    +            user_id=_USER_ID, chat_id="8494508720", chat_type="dm",
    +        )
    +        assert recovered2 is None

  • tests/gateway/test_session_store_lock_io.py

    @@ -272,7 +272,13 @@ def test_auto_reset_does_not_recover_session_being_ended(tmp_path):
         assert entry.session_id != old.session_id
         assert entry.was_auto_reset is True
         db.reopen_session.assert_not_called()
    -    db.end_session.assert_called_once_with(old.session_id, "session_reset")
    +    # Auto-reset now writes through promote_to_session_reset (upgrades
    +    # accidental agent_close/ws_orphan_reap ends) with the specific
    +    # auditable reason — a suspended session resets as "suspended".
    +    db.promote_to_session_reset.assert_called_once_with(
    +        old.session_id, "suspended"
    +    )
    +    db.end_session.assert_not_called()
     
     
     def test_legacy_and_off_lock_saves_share_one_serialization_lock(tmp_path):

  • tests/gateway/test_session_store_runtime_stale_guard.py

    @@ -220,3 +220,50 @@ def test_force_new_skips_stale_check(self, tmp_path):
     
             assert result.session_id != "sid_old"
             db.get_session.assert_not_called()
    +
    +    def test_stale_agent_close_overdue_policy_creates_fresh_session(
    +        self, tmp_path,
    +    ):
    +        """Stale `agent_close` entry + overdue reset policy → fresh session.
    +
    +        The #54878 self-healing path popped the stale sessions.json entry and
    +        recovered the same session_id from the DB without checking whether a
    +        daily/idle reset was actually due.  This test guards the fix at
    +        gateway/session.py:1765 — when the session is overdue under the
    +        configured reset policy, we must create a fresh session (new id,
    +        auto-reset metadata set, reopen_session NOT called).
    +        """
    +        source = _source()
    +        # Idle policy: reset after 60 minutes of inactivity.
    +        config = GatewayConfig(
    +            default_reset_policy=SessionResetPolicy(mode="idle", idle_minutes=60),
    +        )
    +        db = _db_returning({"sid_stale": {"end_reason": "agent_close", "id": "sid_stale"}})
    +        # Recovery would normally reopen this row — but it shouldn't, because
    +        # the reset policy says this session is overdue.
    +        db.find_latest_gateway_session_for_peer.return_value = {
    +            "id": "sid_stale",
    +            "started_at": (datetime.now() - timedelta(hours=3)).timestamp(),
    +        }
    +
    +        with patch("gateway.session.SessionStore._ensure_loaded"):
    +            store = SessionStore(sessions_dir=tmp_path, config=config)
    +        store._db = db
    +        store._loaded = True
    +
    +        key = store._generate_session_key(source)
    +        # Entry last updated 2 hours ago → well past the 60-minute idle window.
    +        store._entries[key] = _make_entry(key, "sid_stale")
    +        store._entries[key].updated_at = datetime.now() - timedelta(hours=2)
    +
    +        result = store.get_or_create_session(source)
    +
    +        # Fresh session — NOT the stale session_id.
    +        assert result.session_id != "sid_stale"
    +        # Auto-reset metadata is set.
    +        assert result.was_auto_reset is True
    +        assert result.auto_reset_reason == "idle"
    +        # reopen_session must NOT have been called (we skipped recovery).
    +        db.reopen_session.assert_not_called()
    +        # A brand-new session row was created.
    +        db.create_session.assert_called_once()

  • tests/gateway/test_session_store_stale_prune.py

    @@ -151,6 +151,25 @@ def test_prunes_stale_entry_when_recovery_only_finds_same_ended_session(self, tm
     
             assert key not in store._entries
     
    +    def test_keeps_stale_entry_when_recovery_lookup_raises(self, tmp_path):
    +        """Indeterminate recovery must not delete the only routing handle.
    +
    +        Startup pruning sees an ended parent and tries to repoint it to the
    +        latest live gateway child.  If that recovery query raises, deleting the
    +        sessions.json entry loses the routing key entirely; keeping it lets the
    +        runtime stale guard retry recovery on the next message.
    +        """
    +        key = "agent:main:telegram:dm:5140768830"
    +        db = _db_returning({"sid_parent": {"end_reason": "compression", "id": "sid_parent"}})
    +        db.find_latest_gateway_session_for_peer.side_effect = RuntimeError("db busy")
    +        store = _make_store_with_db(tmp_path, db)
    +        store._entries[key] = _make_entry_with_origin(key, "sid_parent")
    +
    +        store._prune_stale_sessions_locked()
    +
    +        assert key in store._entries
    +        assert store._entries[key].session_id == "sid_parent"
    +
         def test_noop_when_db_is_none(self, tmp_path):
             config = GatewayConfig(default_reset_policy=SessionResetPolicy(mode="none"))
             with patch("gateway.session.SessionStore._ensure_loaded"):

• #82633 fix(gateway): make session identity durable so chat continuity survives crashes and restarts

  merged to main

  files: gateway/session.py, hermes_state.py, tests/gateway/test_session_continuity_82616.py

• #82743 fix(gateway): session recovery honors reset boundaries and real idle time (#68617 + #78618 salvage)

  merged to main

  files: contributors/emails/[redacted:email], gateway/session.py, hermes_state.py, tests/gateway/test_session_store_runtime_stale_guard.py, tests/gateway/test_session_store_stale_prune.py, tests/test_hermes_state.py