From 784f733cf81198319d6adf9eef6cecdc389cb881 Mon Sep 17 00:00:00 2001 From: Hill Chitsanupong Date: Tue, 4 Aug 2026 23:11:37 +0700 Subject: [PATCH] fix(gateway): honor session_reset policy when recovering sessions MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Both session recovery paths (the startup stale-entry repoint and the lazy in-message recovery) rebuilt the routing entry with updated_at=now and never consulted _should_reset, so an opt-in idle/daily session_reset policy was silently dead across any gateway restart: a recovered session always looked freshly active, and since every subsequent message bumps updated_at, a session recovered stale could then never age out at all. Fix in three parts: - _create_entry_from_recovered_row derives updated_at from the durable last_activity_at the finder already returns on the row (no extra DB round-trip; the original PR added SessionDB.get_last_activity for this, unnecessary post-#82633), falling back to created_at. An invalid or missing started_at now maps to epoch 0 instead of now — an invalid durable timestamp must look old, never freshly active. reset_had_activity is set from the row's durable activity/message signals so the continuity hint stays accurate. - _recover_session_from_db evaluates _should_reset on the rebuilt entry: an overdue session is durably promoted to a reset boundary (promote_to_session_reset, falling back to end_session) and the stale mapping is dropped instead of repointed. - _query_recoverable_session no longer reopens the row; the get_or_create_session recovery phase evaluates _should_reset first and either feeds the normal auto-reset create path (reset notice, prev_session_id continuity, durable promotion) or reopens and publishes the recovered entry exactly as before. Behavior is unchanged under the default session_reset mode "none": _should_reset returns None there, so recovery still resumes every recoverable row — only users who opted into idle/daily resets see the policy actually applied across restarts. Cherry-picked from #78618 and adapted to the #82633 finder. (cherry picked from commit 31c71f762961638c199287fc6ffe836115c4892b) --- gateway/session.py | 96 ++++++++++++---- .../test_session_store_runtime_stale_guard.py | 105 ++++++++++++++++++ .../gateway/test_session_store_stale_prune.py | 69 ++++++++++++ 3 files changed, 250 insertions(+), 20 deletions(-) diff --git a/gateway/session.py b/gateway/session.py index 38616c78e1..f88eb1de43 100644 --- a/gateway/session.py +++ b/gateway/session.py @@ -1903,18 +1903,37 @@ class SessionStore: ) -> SessionEntry: started_at = row.get("started_at") try: - created_at = datetime.fromtimestamp(float(started_at)) if started_at else now + created_at = datetime.fromtimestamp(float(started_at)) except (TypeError, ValueError, OSError): - created_at = now + # An invalid durable timestamp must look old, never freshly active. + created_at = datetime.fromtimestamp(0) + # The finder already returns the row's durable recency + # (last_activity_at is what it ranks candidates by), so no extra DB + # round-trip is needed: derive updated_at straight from the row. + last_activity = row.get("last_activity_at") + try: + updated_at = ( + datetime.fromtimestamp(float(last_activity)) + if last_activity is not None + else created_at + ) + except (TypeError, ValueError, OSError): + updated_at = created_at + had_activity = row.get("_has_messages") + if had_activity is None: + had_activity = bool(row.get("message_count") or 0) or ( + last_activity is not None + ) return SessionEntry( session_key=session_key, session_id=str(row["id"]), created_at=created_at, - updated_at=now, + updated_at=updated_at, origin=source, display_name=source.chat_name, platform=source.platform, chat_type=source.chat_type, + reset_had_activity=bool(had_activity), ) def _find_gateway_session_row( @@ -1964,7 +1983,13 @@ class SessionStore: now: datetime, raise_on_lookup_error: bool = False, ) -> Optional[SessionEntry]: - """Rebuild a missing session-key mapping from durable state.db data.""" + """Rebuild a missing session-key mapping from durable state.db data. + + Returns ``None`` when no row is recoverable, or when the recovered + session is already overdue under the configured reset policy — the + row is then durably promoted to a reset boundary instead of being + resurrected as freshly active. + """ legacy_key = self._legacy_slack_session_key(source) recovered = self._find_gateway_session_row( session_key=session_key, @@ -2001,16 +2026,31 @@ class SessionStore: session_key, ) return None - try: - self._db.reopen_session(str(recovered["id"])) - except Exception as exc: - logger.debug("Gateway session DB reopen failed for %s: %s", session_key, exc) entry = self._create_entry_from_recovered_row( row=recovered, session_key=session_key, source=source, now=now, ) + reset_reason = self._should_reset(entry, source) + if reset_reason: + try: + promote = getattr(self._db, "promote_to_session_reset", None) + if callable(promote): + promote(entry.session_id, reset_reason) + else: + self._db.end_session(entry.session_id, reset_reason) + except Exception as exc: + logger.debug( + "Gateway recovered-session reset promotion failed for %s: %s", + session_key, + exc, + ) + return None + try: + self._db.reopen_session(entry.session_id) + except Exception as exc: + logger.debug("Gateway session DB reopen failed for %s: %s", session_key, exc) if migrated_legacy: self._record_gateway_session_peer( entry.session_id, @@ -2026,6 +2066,8 @@ class SessionStore: """DB-only half of _recover_session_from_db (no lock needed). Returns a SessionEntry or None. Caller assigns _entries[key] under lock. + The returned entry's session row is NOT reopened here: the caller + evaluates the reset policy first and decides reset vs resume. """ legacy_key = self._legacy_slack_session_key(source) recovered = self._find_gateway_session_row( @@ -2061,11 +2103,9 @@ class SessionStore: session_key, ) return None - try: - self._db.reopen_session(str(recovered["id"])) - except Exception as exc: - logger.debug("Gateway session DB reopen failed for %s: %s", - session_key, exc) + # Reopen only after the caller evaluates reset policy against durable + # last activity. An agent_close/ws_orphan row may need promotion to a + # real reset boundary instead. entry = self._create_entry_from_recovered_row( row=recovered, session_key=session_key, source=source, now=now, ) @@ -2638,13 +2678,29 @@ class SessionStore: session_key=session_key, source=source, now=now, ) if recovered is not None: - with self._lock: - published = self._entries.get(session_key) - if published is None: - self._entries[session_key] = recovered - published = recovered - entry = published - _needs_save = True + recovered_reset_reason = self._should_reset(recovered, source) + if recovered_reset_reason: + was_auto_reset = True + auto_reset_reason = recovered_reset_reason + reset_had_activity = recovered.reset_had_activity + db_end_session_id = recovered.session_id + prev_session_id = recovered.session_id + else: + try: + self._db.reopen_session(recovered.session_id) + except Exception as exc: + logger.debug( + "Gateway session DB reopen failed for %s: %s", + session_key, + exc, + ) + with self._lock: + published = self._entries.get(session_key) + if published is None: + self._entries[session_key] = recovered + published = recovered + entry = published + _needs_save = True if entry is None: # Create a candidate outside the lock, then publish only if another diff --git a/tests/gateway/test_session_store_runtime_stale_guard.py b/tests/gateway/test_session_store_runtime_stale_guard.py index 1702c6fc7b..e248e811e1 100644 --- a/tests/gateway/test_session_store_runtime_stale_guard.py +++ b/tests/gateway/test_session_store_runtime_stale_guard.py @@ -165,6 +165,111 @@ class TestRuntimeStaleGuard: db.create_session.assert_called_once() +class TestRecoveredSessionResetPolicy: + """Recovery must not resurrect sessions as freshly active. + + ``_create_entry_from_recovered_row`` used to stamp ``updated_at=now`` on + the rebuilt entry, so an opt-in idle/daily ``session_reset`` policy could + never fire across a gateway restart: the recovered session always looked + freshly active, and every subsequent message bumped ``updated_at`` again + — a recovered stale session could never age out. The entry now carries + the durable ``last_activity_at`` the finder already returns on the row + and the recovery paths evaluate ``_should_reset`` before resuming. + """ + + def test_recovered_entry_carries_durable_last_activity(self, tmp_path): + """A recovered mapping reports the DB's last message time, not now().""" + source = _source() + started = (datetime.now() - timedelta(hours=3)).timestamp() + last_activity = (datetime.now() - timedelta(hours=2)).timestamp() + db = _db_returning({}) + db.find_latest_gateway_session_for_peer.return_value = { + "id": "sid_recovered", + "started_at": started, + "last_activity_at": last_activity, + } + store = _make_store_with_db(tmp_path, db) # default mode="none" + + result = store.get_or_create_session(source) + + assert result.session_id == "sid_recovered" + assert result.created_at == datetime.fromtimestamp(started) + assert result.updated_at == datetime.fromtimestamp(last_activity) + assert result.reset_had_activity is True + + def test_recovered_session_past_idle_policy_resets_instead_of_resuming( + self, tmp_path, + ): + """Lost mapping + overdue recoverable row → reset, not silent resume.""" + source = _source() + config = GatewayConfig( + default_reset_policy=SessionResetPolicy(mode="idle", idle_minutes=60), + ) + db = _db_returning({}) + db.find_latest_gateway_session_for_peer.return_value = { + "id": "sid_idle", + "started_at": (datetime.now() - timedelta(hours=3)).timestamp(), + "last_activity_at": ( + datetime.now() - timedelta(hours=2) + ).timestamp(), + } + with patch("gateway.session.SessionStore._ensure_loaded"): + store = SessionStore(sessions_dir=tmp_path, config=config) + store._db = db + store._loaded = True + # No in-memory entry: the mapping was lost (e.g. crash before save). + + result = store.get_or_create_session(source) + + assert result.session_id != "sid_idle" + assert result.was_auto_reset is True + assert result.auto_reset_reason == "idle" + assert result.reset_had_activity is True + assert result.prev_session_id == "sid_idle" + db.reopen_session.assert_not_called() + db.promote_to_session_reset.assert_called_once_with("sid_idle", "idle") + db.end_session.assert_not_called() + db.create_session.assert_called_once() + + def test_default_none_policy_recovery_resumes_unchanged(self, tmp_path): + """mode="none" (the default) still resumes recoverable rows as before.""" + source = _source() + db = _db_returning({}) + db.find_latest_gateway_session_for_peer.return_value = { + "id": "sid_recovered", + "started_at": (datetime.now() - timedelta(days=30)).timestamp(), + "last_activity_at": ( + datetime.now() - timedelta(days=30) + ).timestamp(), + } + store = _make_store_with_db(tmp_path, db) # default mode="none" + + result = store.get_or_create_session(source) + + assert result.session_id == "sid_recovered" + db.reopen_session.assert_called_once_with("sid_recovered") + db.promote_to_session_reset.assert_not_called() + db.end_session.assert_not_called() + db.create_session.assert_not_called() + + def test_recovery_tolerates_row_without_last_activity(self, tmp_path): + """A row lacking last_activity_at falls back to created_at.""" + source = _source() + started = (datetime.now() - timedelta(hours=3)).timestamp() + db = _db_returning({}) + db.find_latest_gateway_session_for_peer.return_value = { + "id": "sid_recovered", + "started_at": started, + } + store = _make_store_with_db(tmp_path, db) + + result = store.get_or_create_session(source) + + assert result.session_id == "sid_recovered" + assert result.updated_at == datetime.fromtimestamp(started) + assert result.reset_had_activity is False + + class TestAdvanceCompressionSession: def test_cas_advances_route_without_reopening_rows(self, tmp_path): db = _db_returning({}) diff --git a/tests/gateway/test_session_store_stale_prune.py b/tests/gateway/test_session_store_stale_prune.py index 421529d112..9d5f6cb57e 100644 --- a/tests/gateway/test_session_store_stale_prune.py +++ b/tests/gateway/test_session_store_stale_prune.py @@ -169,6 +169,75 @@ class TestPruneStaleSessionsLocked: assert reset_row["end_reason"] == "session_reset" +# --------------------------------------------------------------------------- +# Startup recovery honours the reset policy +# --------------------------------------------------------------------------- + +class TestStartupRecoveryResetPolicy: + """Startup repoint must not resurrect an overdue session as fresh. + + The startup pruner repoints a stale entry to the recovered row via + ``_recover_session_from_db``. The rebuilt entry used to be stamped + ``updated_at=now``, so an opt-in idle/daily ``session_reset`` policy was + silently skipped across every gateway restart. Recovery now evaluates + ``_should_reset`` against the durable last message timestamp and promotes + an overdue session to a durable reset boundary instead of reopening it. + """ + + def test_overdue_recovered_session_promoted_to_reset_and_pruned(self, tmp_path): + key = "agent:main:telegram:dm:5140768830" + db = _db_returning( + {"sid_parent": {"end_reason": "agent_close", "id": "sid_parent"}} + ) + db.find_latest_gateway_session_for_peer.return_value = { + "id": "sid_child", + "started_at": (datetime.now() - timedelta(hours=5)).timestamp(), + "last_activity_at": ( + datetime.now() - timedelta(hours=4) + ).timestamp(), + } + config = GatewayConfig( + default_reset_policy=SessionResetPolicy(mode="idle", idle_minutes=60), + ) + with patch("gateway.session.SessionStore._ensure_loaded"): + store = SessionStore(sessions_dir=tmp_path, config=config) + store._db = db + store._loaded = True + store._entries[key] = _make_entry_with_origin(key, "sid_parent") + + with patch.object(store, "_save"): + store._prune_stale_sessions_locked() + + assert key not in store._entries + db.promote_to_session_reset.assert_called_once_with("sid_child", "idle") + db.reopen_session.assert_not_called() + + def test_none_policy_startup_repoint_unchanged(self, tmp_path): + """mode="none" (the default) still repoints to the recovered row.""" + key = "agent:main:telegram:dm:5140768830" + db = _db_returning( + {"sid_parent": {"end_reason": "compression", "id": "sid_parent"}} + ) + last_activity = (datetime.now() - timedelta(hours=4)).timestamp() + db.find_latest_gateway_session_for_peer.return_value = { + "id": "sid_child", + "started_at": (datetime.now() - timedelta(hours=5)).timestamp(), + "last_activity_at": last_activity, + } + store = _make_store_with_db(tmp_path, db) # default mode="none" + store._entries[key] = _make_entry_with_origin(key, "sid_parent") + + with patch.object(store, "_save"): + store._prune_stale_sessions_locked() + + assert store._entries[key].session_id == "sid_child" + assert store._entries[key].updated_at == datetime.fromtimestamp( + last_activity + ) + db.reopen_session.assert_called_once_with("sid_child") + db.promote_to_session_reset.assert_not_called() + + # --------------------------------------------------------------------------- # Integration: _ensure_loaded_locked calls _prune_stale_sessions_locked # ---------------------------------------------------------------------------