fix(gateway): honor session_reset policy when recovering sessions
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)
This commit is contained in:
committed by
Teknium
parent
6e99531c8e
commit
784f733cf8
@@ -1903,18 +1903,37 @@ class SessionStore:
|
|||||||
) -> SessionEntry:
|
) -> SessionEntry:
|
||||||
started_at = row.get("started_at")
|
started_at = row.get("started_at")
|
||||||
try:
|
try:
|
||||||
created_at = datetime.fromtimestamp(float(started_at)) if started_at else now
|
created_at = datetime.fromtimestamp(float(started_at))
|
||||||
except (TypeError, ValueError, OSError):
|
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(
|
return SessionEntry(
|
||||||
session_key=session_key,
|
session_key=session_key,
|
||||||
session_id=str(row["id"]),
|
session_id=str(row["id"]),
|
||||||
created_at=created_at,
|
created_at=created_at,
|
||||||
updated_at=now,
|
updated_at=updated_at,
|
||||||
origin=source,
|
origin=source,
|
||||||
display_name=source.chat_name,
|
display_name=source.chat_name,
|
||||||
platform=source.platform,
|
platform=source.platform,
|
||||||
chat_type=source.chat_type,
|
chat_type=source.chat_type,
|
||||||
|
reset_had_activity=bool(had_activity),
|
||||||
)
|
)
|
||||||
|
|
||||||
def _find_gateway_session_row(
|
def _find_gateway_session_row(
|
||||||
@@ -1964,7 +1983,13 @@ class SessionStore:
|
|||||||
now: datetime,
|
now: datetime,
|
||||||
raise_on_lookup_error: bool = False,
|
raise_on_lookup_error: bool = False,
|
||||||
) -> Optional[SessionEntry]:
|
) -> 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)
|
legacy_key = self._legacy_slack_session_key(source)
|
||||||
recovered = self._find_gateway_session_row(
|
recovered = self._find_gateway_session_row(
|
||||||
session_key=session_key,
|
session_key=session_key,
|
||||||
@@ -2001,16 +2026,31 @@ class SessionStore:
|
|||||||
session_key,
|
session_key,
|
||||||
)
|
)
|
||||||
return None
|
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(
|
entry = self._create_entry_from_recovered_row(
|
||||||
row=recovered,
|
row=recovered,
|
||||||
session_key=session_key,
|
session_key=session_key,
|
||||||
source=source,
|
source=source,
|
||||||
now=now,
|
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:
|
if migrated_legacy:
|
||||||
self._record_gateway_session_peer(
|
self._record_gateway_session_peer(
|
||||||
entry.session_id,
|
entry.session_id,
|
||||||
@@ -2026,6 +2066,8 @@ class SessionStore:
|
|||||||
"""DB-only half of _recover_session_from_db (no lock needed).
|
"""DB-only half of _recover_session_from_db (no lock needed).
|
||||||
|
|
||||||
Returns a SessionEntry or None. Caller assigns _entries[key] under lock.
|
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)
|
legacy_key = self._legacy_slack_session_key(source)
|
||||||
recovered = self._find_gateway_session_row(
|
recovered = self._find_gateway_session_row(
|
||||||
@@ -2061,11 +2103,9 @@ class SessionStore:
|
|||||||
session_key,
|
session_key,
|
||||||
)
|
)
|
||||||
return None
|
return None
|
||||||
try:
|
# Reopen only after the caller evaluates reset policy against durable
|
||||||
self._db.reopen_session(str(recovered["id"]))
|
# last activity. An agent_close/ws_orphan row may need promotion to a
|
||||||
except Exception as exc:
|
# real reset boundary instead.
|
||||||
logger.debug("Gateway session DB reopen failed for %s: %s",
|
|
||||||
session_key, exc)
|
|
||||||
entry = self._create_entry_from_recovered_row(
|
entry = self._create_entry_from_recovered_row(
|
||||||
row=recovered, session_key=session_key, source=source, now=now,
|
row=recovered, session_key=session_key, source=source, now=now,
|
||||||
)
|
)
|
||||||
@@ -2638,13 +2678,29 @@ class SessionStore:
|
|||||||
session_key=session_key, source=source, now=now,
|
session_key=session_key, source=source, now=now,
|
||||||
)
|
)
|
||||||
if recovered is not None:
|
if recovered is not None:
|
||||||
with self._lock:
|
recovered_reset_reason = self._should_reset(recovered, source)
|
||||||
published = self._entries.get(session_key)
|
if recovered_reset_reason:
|
||||||
if published is None:
|
was_auto_reset = True
|
||||||
self._entries[session_key] = recovered
|
auto_reset_reason = recovered_reset_reason
|
||||||
published = recovered
|
reset_had_activity = recovered.reset_had_activity
|
||||||
entry = published
|
db_end_session_id = recovered.session_id
|
||||||
_needs_save = True
|
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:
|
if entry is None:
|
||||||
# Create a candidate outside the lock, then publish only if another
|
# Create a candidate outside the lock, then publish only if another
|
||||||
|
|||||||
@@ -165,6 +165,111 @@ class TestRuntimeStaleGuard:
|
|||||||
db.create_session.assert_called_once()
|
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:
|
class TestAdvanceCompressionSession:
|
||||||
def test_cas_advances_route_without_reopening_rows(self, tmp_path):
|
def test_cas_advances_route_without_reopening_rows(self, tmp_path):
|
||||||
db = _db_returning({})
|
db = _db_returning({})
|
||||||
|
|||||||
@@ -169,6 +169,75 @@ class TestPruneStaleSessionsLocked:
|
|||||||
assert reset_row["end_reason"] == "session_reset"
|
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
|
# Integration: _ensure_loaded_locked calls _prune_stale_sessions_locked
|
||||||
# ---------------------------------------------------------------------------
|
# ---------------------------------------------------------------------------
|
||||||
|
|||||||
Reference in New Issue
Block a user