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:

- SessionDB.get_last_activity(session_id) returns the last stored
  message timestamp, and _create_entry_from_recovered_row derives
  updated_at from it (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 durable transcript 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. Recovery stays lock-free on
the message path (TestRecoverOutsideLock), and the pre-existing session
recovery suites pass unmodified.
This commit is contained in:
Hill Chitsanupong 2026-08-04 23:11:37 +07:00
parent e6977f41bc
commit 31c71f7629
5 changed files with 281 additions and 20 deletions

View File

@ -1809,18 +1809,39 @@ 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)
last_activity = None
getter = getattr(self._db, "get_last_activity", None)
if callable(getter):
try:
last_activity = getter(str(row["id"]))
except Exception:
logger.debug(
"Gateway session last-activity lookup failed for %s",
row.get("id"),
exc_info=True,
)
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
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=last_activity is not None,
)
def _find_gateway_session_row(
@ -1870,7 +1891,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,
@ -1907,16 +1934,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,
@ -1932,6 +1974,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(
@ -1967,11 +2011,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,
)
@ -2544,13 +2586,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

View File

@ -5174,6 +5174,19 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
row = cursor.fetchone()
return self._session_row_dict(row) if row else None
def get_last_activity(self, session_id: str) -> Optional[float]:
"""Return the last stored message timestamp for one session."""
if not session_id:
return None
with self._read_ctx() as conn:
row = conn.execute(
"SELECT MAX(timestamp) AS last_activity "
"FROM messages WHERE session_id = ?",
(session_id,),
).fetchone()
value = row["last_activity"] if row is not None else None
return float(value) if value is not None else None
def resolve_session_id(self, session_id_or_prefix: str) -> Optional[str]:
"""Resolve an exact or uniquely prefixed session ID to the full ID.

View File

@ -165,6 +165,113 @@ 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 message timestamp (``SessionDB.get_last_activity``) 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,
}
db.get_last_activity.return_value = 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
db.get_last_activity.assert_called_once_with("sid_recovered")
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(),
}
db.get_last_activity.return_value = (
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(),
}
db.get_last_activity.return_value = (
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_db_without_last_activity(self, tmp_path):
"""Older SessionDB without get_last_activity 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,
}
db.get_last_activity = None # attribute exists but is not callable
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({})

View File

@ -127,6 +127,75 @@ class TestPruneStaleSessionsLocked:
mock_save.assert_called_once()
# ---------------------------------------------------------------------------
# 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(),
}
db.get_last_activity.return_value = (
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"}}
)
db.find_latest_gateway_session_for_peer.return_value = {
"id": "sid_child",
"started_at": (datetime.now() - timedelta(hours=5)).timestamp(),
}
last_activity = (datetime.now() - timedelta(hours=4)).timestamp()
db.get_last_activity.return_value = 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
# ---------------------------------------------------------------------------

View File

@ -394,6 +394,20 @@ class TestMessageStorage:
assert messages[0]["content"] == "Hello"
assert messages[1]["role"] == "assistant"
def test_get_last_activity(self, db):
db.create_session(session_id="s1", source="cli")
assert db.get_last_activity("s1") is None
db.append_message("s1", role="user", content="Hello")
db.append_message("s1", role="assistant", content="Hi there!")
last_activity = db.get_last_activity("s1")
messages = db.get_messages("s1")
assert last_activity == max(m["timestamp"] for m in messages)
assert db.get_last_activity("") is None
assert db.get_last_activity("missing") is None
def test_startup_heals_null_active_rows(self, tmp_path):