test_fresh_reset_skill_injection.py
1 """Regression tests for topic/channel skill auto-injection after /new or /reset. 2 3 Covers the fix for issue #6508. 4 5 Before the fix: 6 1. User sends ``/new`` — ``reset_session`` creates a fresh SessionEntry 7 with ``created_at == updated_at``. 8 2. User sends the next message. 9 3. ``get_or_create_session`` finds the entry and bumps 10 ``entry.updated_at = now`` (microseconds after ``created_at``). 11 4. ``_handle_message_with_agent`` checks 12 ``_is_new_session = (created_at == updated_at) or was_auto_reset``. 13 Both are False → ``_is_new_session = False`` → topic/channel skills 14 are silently skipped for the first message of a manually reset session. 15 16 After the fix: 17 ``reset_session`` stamps the new entry with ``is_fresh_reset=True``. 18 ``_handle_message_with_agent`` ORs this into ``_is_new_session`` and 19 consumes the flag immediately after the check, so subsequent messages 20 are treated as continuing the session and the flag does not leak. 21 22 We use ``was_auto_reset`` for surprise resets (idle/daily/suspended) and 23 ``is_fresh_reset`` for user-initiated resets because the former also drives 24 a "Session automatically reset due to inactivity" user-facing notice and 25 a context-note prepend into the agent's prompt — both wrong for an explicit 26 /new or /reset. 27 """ 28 import pytest 29 30 from gateway.config import GatewayConfig, Platform 31 from gateway.session import SessionEntry, SessionSource, SessionStore 32 33 34 def _make_store(tmp_path): 35 return SessionStore(sessions_dir=tmp_path, config=GatewayConfig()) 36 37 38 def _make_source(chat_id="123", user_id="u1"): 39 return SessionSource( 40 platform=Platform.TELEGRAM, 41 chat_id=chat_id, 42 user_id=user_id, 43 ) 44 45 46 def _is_new_session(entry) -> bool: 47 """Mirror of the predicate in ``_handle_message_with_agent``. 48 49 Kept in-sync with the production check so this test fails loudly if the 50 upstream logic regresses. 51 """ 52 return ( 53 entry.created_at == entry.updated_at 54 or getattr(entry, "was_auto_reset", False) 55 or getattr(entry, "is_fresh_reset", False) 56 ) 57 58 59 # --------------------------------------------------------------------------- 60 # reset_session stamps is_fresh_reset=True 61 # --------------------------------------------------------------------------- 62 63 class TestResetSessionStampsFreshReset: 64 def test_reset_session_sets_is_fresh_reset_true(self, tmp_path): 65 store = _make_store(tmp_path) 66 source = _make_source() 67 store.get_or_create_session(source) 68 session_key = store._generate_session_key(source) 69 70 new_entry = store.reset_session(session_key) 71 72 assert new_entry is not None 73 assert new_entry.is_fresh_reset is True 74 75 def test_reset_session_unknown_key_returns_none(self, tmp_path): 76 store = _make_store(tmp_path) 77 assert store.reset_session("unknown:key") is None 78 79 def test_fresh_session_does_not_have_is_fresh_reset(self, tmp_path): 80 """A vanilla first-time session should not carry the flag.""" 81 store = _make_store(tmp_path) 82 entry = store.get_or_create_session(_make_source()) 83 assert entry.is_fresh_reset is False 84 85 86 # --------------------------------------------------------------------------- 87 # Core regression: _is_new_session stays True after updated_at bump 88 # --------------------------------------------------------------------------- 89 90 class TestIsNewSessionSurvivesUpdatedAtBump: 91 def test_is_new_session_true_after_reset_then_next_message(self, tmp_path): 92 """The actual bug: _is_new_session was False on message after /reset.""" 93 store = _make_store(tmp_path) 94 source = _make_source() 95 store.get_or_create_session(source) 96 session_key = store._generate_session_key(source) 97 98 # User sends /reset 99 store.reset_session(session_key) 100 101 # Next inbound message — get_or_create_session bumps updated_at 102 entry = store.get_or_create_session(source) 103 104 # Before the fix: created_at != updated_at, was_auto_reset=False → False 105 # After the fix: is_fresh_reset=True carries the signal through the bump 106 assert _is_new_session(entry) is True 107 108 def test_flag_consumed_after_first_read(self, tmp_path): 109 """After the message handler consumes is_fresh_reset, the NEXT 110 message should not be treated as a new session (skill re-injection 111 must not fire a second time). 112 """ 113 store = _make_store(tmp_path) 114 source = _make_source() 115 store.get_or_create_session(source) 116 session_key = store._generate_session_key(source) 117 store.reset_session(session_key) 118 119 # First message — handler consumes the flag 120 entry = store.get_or_create_session(source) 121 assert _is_new_session(entry) is True 122 entry.is_fresh_reset = False # what _handle_message_with_agent does 123 124 # Second message — must not be treated as new 125 entry = store.get_or_create_session(source) 126 assert _is_new_session(entry) is False 127 128 129 # --------------------------------------------------------------------------- 130 # Vanilla-session behavior is unchanged 131 # --------------------------------------------------------------------------- 132 133 class TestVanillaBehaviorUnaffected: 134 def test_ongoing_session_not_flagged_as_new(self, tmp_path): 135 store = _make_store(tmp_path) 136 source = _make_source() 137 store.get_or_create_session(source) 138 139 # Second message on the same session — updated_at bumps, 140 # is_fresh_reset was never set 141 entry = store.get_or_create_session(source) 142 assert entry.is_fresh_reset is False 143 assert _is_new_session(entry) is False 144 145 def test_idle_auto_reset_does_not_set_is_fresh_reset(self, tmp_path): 146 """Idle/daily auto-resets use was_auto_reset — confirm they do NOT 147 also set is_fresh_reset (which would double-fire the skill path and 148 not leak through the auto-reset guard). 149 """ 150 store = _make_store(tmp_path) 151 source = _make_source() 152 entry = store.get_or_create_session(source) 153 154 # Simulate the auto-reset code path: get_or_create_session's internal 155 # branch that sets was_auto_reset does NOT touch is_fresh_reset. 156 # Construct a fresh entry the same way that branch does. 157 store._entries.pop(store._generate_session_key(source)) 158 fresh = SessionEntry( 159 session_key=entry.session_key, 160 session_id="new_id", 161 created_at=entry.created_at, 162 updated_at=entry.created_at, 163 origin=source, 164 was_auto_reset=True, 165 auto_reset_reason="idle", 166 ) 167 assert fresh.is_fresh_reset is False 168 assert fresh.was_auto_reset is True 169 170 171 # --------------------------------------------------------------------------- 172 # Persistence through sessions.json round-trip 173 # --------------------------------------------------------------------------- 174 175 class TestPersistence: 176 def test_is_fresh_reset_survives_to_dict_from_dict(self, tmp_path): 177 """Protect against the gateway restarting between /reset and the 178 next message — the flag must be persisted in sessions.json. 179 """ 180 store = _make_store(tmp_path) 181 source = _make_source() 182 store.get_or_create_session(source) 183 session_key = store._generate_session_key(source) 184 new_entry = store.reset_session(session_key) 185 186 assert new_entry.is_fresh_reset is True 187 restored = SessionEntry.from_dict(new_entry.to_dict()) 188 assert restored.is_fresh_reset is True 189 190 def test_default_false_when_missing_from_dict(self, tmp_path): 191 """Older sessions.json files written before this field existed must 192 load cleanly with is_fresh_reset defaulting to False. 193 """ 194 data = { 195 "session_key": "telegram:1:123", 196 "session_id": "sess1", 197 "created_at": "2026-01-01T00:00:00", 198 "updated_at": "2026-01-01T00:00:00", 199 } 200 entry = SessionEntry.from_dict(data) 201 assert entry.is_fresh_reset is False