test(gateway): add overdue-policy guard for stale-agent-close recovery path

When the #54878 self-healing path drops a stale sessions.json entry, the
fix at gateway/session.py:1765 now checks _should_reset() before falling
through to DB recovery. This test covers the case where the stale entry's
session is overdue under an idle/daily reset policy — it must create a
fresh session, set auto-reset metadata, and NOT call reopen_session().
This commit is contained in:
Alec 2026-07-13 09:03:03 +08:00 • committed by Teknium
parent 4b12b7a359
commit 039f6b2f1b

View file

@ -220,3 +220,50 @@ class TestRuntimeStaleGuard:
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()