diff --git a/gateway/session.py b/gateway/session.py index 02fdd0043aeb2..e4f71a298f957 100644 --- a/gateway/session.py +++ b/gateway/session.py @@ -890,6 +890,35 @@ def _session_key_namespace(profile: Optional[str]) -> str: return f"agent:{profile}" +# The DM separator in a BlueBubbles chat GUID (``;-;``). +# Groups use ``;+;`` and an opaque chat id, so they are never unwrapped. +_BLUEBUBBLES_DM_SEPARATOR = ";-;" + + +def canonical_bluebubbles_identifier(chat_id: Optional[str]) -> Optional[str]: + """Collapse BlueBubbles DM chat GUID variants to the bare handle. + + BlueBubbles surfaces one 1:1 conversation under several chat ids: a + service-prefixed GUID (``iMessage;-;+1555…``, ``any;-;+1555…``, and the + prefix flips when the thread moves between iMessage and SMS) when the + webhook carries one, and the bare handle (``+1555…``) when it does not + and the adapter falls back to ``chatIdentifier``. Keyed raw, each form + becomes a separate session, so a message arriving on a long-untouched + variant trips the idle reset and clears an active conversation. + + Only the DM separator ``;-;`` is unwrapped. Group GUIDs use ``;+;`` and + carry an opaque chat id that is not a participant handle, so they are + returned untouched. + """ + if not chat_id: + return chat_id + value = chat_id.strip() + service, sep, handle = value.partition(_BLUEBUBBLES_DM_SEPARATOR) + if sep and handle and ";" not in service: + return handle.strip() + return value + + def build_session_key( source: SessionSource, group_sessions_per_user: bool = True, @@ -930,6 +959,8 @@ def build_session_key( dm_chat_id = source.chat_id if source.platform == Platform.WHATSAPP: dm_chat_id = canonical_whatsapp_identifier(source.chat_id) + elif source.platform == Platform.BLUEBUBBLES: + dm_chat_id = canonical_bluebubbles_identifier(source.chat_id) if dm_chat_id: if source.thread_id: @@ -1081,6 +1112,86 @@ def _routing_scope(self) -> str: except Exception: return str(self.sessions_dir) + def _migrate_loaded_bluebubbles_keys_locked( + self, + ) -> tuple[bool, List[tuple[str, SessionEntry]], List[str]]: + """Canonicalize legacy BlueBubbles DM routes before startup resumes. + + Older routing indexes may contain several raw GUID keys for one DM. + Collapse them to the current key and keep the most recently active + entry so stale siblings cannot be scheduled or reset independently. + Returns whether the index changed, the winners whose state.db peer row + must be rewritten to the canonical key, and the retired sibling + sessions that must be ended. Dropping a sibling from the index is not + enough: it stays live in state.db under the canonical key, and because + a stray sibling is usually *started* later than the real conversation, + a later ``ORDER BY started_at DESC`` peer lookup would reopen the stray + and orphan the live transcript. + """ + # Only a key still carrying a raw DM GUID can move, so once an install + # has migrated this pass is inert. Keep its steady-state cost to one + # substring test per key instead of regrouping the whole index forever. + if not any(_BLUEBUBBLES_DM_SEPARATOR in key for key in self._entries): + return False, [], [] + + group_per_user = getattr(self.config, "group_sessions_per_user", True) + thread_per_user = getattr(self.config, "thread_sessions_per_user", False) + grouped: Dict[str, List[tuple[str, SessionEntry]]] = {} + for stored_key, entry in self._entries.items(): + target_key = stored_key + origin = entry.origin + if ( + origin is not None + and origin.platform == Platform.BLUEBUBBLES + and origin.chat_type == "dm" + ): + profile = self._profile_from_session_key(stored_key) + if profile is not None: + target_key = build_session_key( + origin, + group_sessions_per_user=group_per_user, + thread_sessions_per_user=thread_per_user, + profile=profile, + ) + grouped.setdefault(target_key, []).append((stored_key, entry)) + + retired = 0 + migrated_entries: Dict[str, SessionEntry] = {} + peer_updates: List[tuple[str, SessionEntry]] = [] + retired_sessions: List[str] = [] + for target_key, candidates in grouped.items(): + # Most recently active wins; on an exact tie prefer whichever entry + # already owns the canonical key. + winner_key, winner = max( + candidates, + key=lambda item: (item[1].updated_at, item[0] == target_key), + ) + # Routing keys are unique, so at most one candidate per group can + # already be canonical — every other one is an alias being retired. + retired += sum( + stored_key != target_key for stored_key, _entry in candidates + ) + for _stored_key, entry in candidates: + if entry is winner or not entry.session_id: + continue + if entry.session_id != winner.session_id: + retired_sessions.append(entry.session_id) + winner.session_key = target_key + migrated_entries[target_key] = winner + if winner_key != target_key: + peer_updates.append((target_key, winner)) + + changed = retired > 0 + if changed: + self._entries = migrated_entries + logger.info( + "gateway.session: canonicalized BlueBubbles routing index " + "(%d legacy alias%s retired)", + retired, + "" if retired == 1 else "es", + ) + return changed, peer_updates, retired_sessions + def _ensure_loaded_locked(self) -> None: """Load the routing index. Must be called with self._lock held. @@ -1174,6 +1285,45 @@ def _ensure_loaded_locked(self) -> None: # held) is cheap: one lookup per routing key, once at startup. self._prune_stale_sessions_locked() + # Compute alias winners only after stale pruning. Pruning may repoint an + # ended canonical entry to a still-live legacy alias; calculating the + # retirement list first would then end the session pruning just rescued. + routing_migrated, peer_updates, retired_sessions = ( + self._migrate_loaded_bluebubbles_keys_locked() + ) + if routing_migrated: + # Retire the siblings before rewriting the winner's peer row, so no + # window exists where two live rows share the canonical key. + for retired_session_id in retired_sessions: + self._end_retired_alias_session(retired_session_id) + for canonical_key, entry in peer_updates: + if self._entries.get(canonical_key) is not entry: + continue + self._record_gateway_session_peer( + entry.session_id, + canonical_key, + entry.origin, + display_name=entry.display_name, + ) + self._save() + + def _end_retired_alias_session(self, session_id: str) -> None: + """End a sibling retired by the alias migration. + + ``session_key_migration`` is deliberately not one of the reasons + ``find_latest_gateway_session_for_peer`` treats as recoverable, so the + retired row can never be reopened, while its transcript stays readable + via ``/resume``. + """ + if not self._db: + return + try: + self._db.end_session(session_id, "session_key_migration") + except Exception: + logger.debug( + "Could not retire aliased session %s", session_id, exc_info=True + ) + def _prune_stale_sessions_locked(self) -> None: """Remove sessions.json entries whose session has ended in state.db. @@ -1435,6 +1585,26 @@ def _create_entry_from_recovered_row( chat_type=source.chat_type, ) + @staticmethod + def _gateway_peer_lookup_kwargs( + *, session_key: str, source: SessionSource + ) -> Dict[str, Any]: + kwargs: Dict[str, Any] = { + "source": source.platform.value, + "user_id": source.user_id, + "session_key": session_key, + "chat_id": source.chat_id, + "chat_type": source.chat_type, + "thread_id": source.thread_id, + } + if ( + source.platform == Platform.BLUEBUBBLES + and source.chat_type == "dm" + and str(source.user_id or "").strip() + ): + kwargs["match_by_participant_identity"] = True + return kwargs + def _recover_session_from_db( self, *, @@ -1451,12 +1621,9 @@ def _recover_session_from_db( return None try: recovered = finder( - source=source.platform.value, - user_id=source.user_id, - session_key=session_key, - chat_id=source.chat_id, - chat_type=source.chat_type, - thread_id=source.thread_id, + **self._gateway_peer_lookup_kwargs( + session_key=session_key, source=source + ) ) except Exception as exc: logger.debug("Gateway session DB recovery failed for %s: %s", session_key, exc) @@ -1500,12 +1667,9 @@ def _query_recoverable_session(self, *, session_key, source, now): return None try: recovered = finder( - source=source.platform.value, - user_id=source.user_id, - session_key=session_key, - chat_id=source.chat_id, - chat_type=source.chat_type, - thread_id=source.thread_id, + **self._gateway_peer_lookup_kwargs( + session_key=session_key, source=source + ) ) except Exception as exc: logger.debug("Gateway session DB recovery failed for %s: %s", @@ -2036,6 +2200,13 @@ def _get_or_create_session_impl( published = recovered entry = published _needs_save = True + if published is recovered: + self._record_gateway_session_peer( + recovered.session_id, + session_key, + source, + display_name=recovered.display_name, + ) if entry is None: # Create a candidate outside the lock, then publish only if another diff --git a/hermes_state.py b/hermes_state.py index ff27d5b13e39b..714cd372bb431 100644 --- a/hermes_state.py +++ b/hermes_state.py @@ -2287,6 +2287,7 @@ def find_latest_gateway_session_for_peer( chat_id: Optional[str] = None, chat_type: Optional[str] = None, thread_id: Optional[str] = None, + match_by_participant_identity: bool = False, ) -> Optional[Dict[str, Any]]: """Find the latest recoverable gateway session for a routing peer. @@ -2298,6 +2299,12 @@ def find_latest_gateway_session_for_peer( (dashboard viewer disconnect before #60609) are treated as recoverable; explicit conversation boundaries such as /new, /resume switches, and compression splits are not. + + ``match_by_participant_identity`` permits a final fallback that matches the same + routing namespace, source, non-empty user_id, chat type, and thread + while ignoring chat_id. This is reserved for DM transports whose + session-key contract intentionally aliases multiple transport chat IDs + for one stable participant identity. """ if not session_key: return None @@ -2341,6 +2348,40 @@ def find_latest_gateway_session_for_peer( """, (source, user_id, chat_id, chat_type, thread_id), ).fetchone() + if row is not None or not match_by_participant_identity: + return dict(row) if row else None + + stable_user_id = str(user_id or "").strip() + key_parts = str(session_key).split(":", 4) + if not stable_user_id or len(key_parts) < 4: + return None + routing_prefix = ":".join(key_parts[:4]) + ":" + row = self._conn.execute( + """ + SELECT * FROM sessions + WHERE source = ? + -- stable_user_id is non-empty, so match the column directly: + -- COALESCE() here would defeat idx_sessions_gateway_peer. + AND user_id = ? + AND COALESCE(chat_type, '') = COALESCE(?, '') + AND COALESCE(thread_id, '') = COALESCE(?, '') + AND substr(COALESCE(session_key, ''), 1, ?) = ? + AND (ended_at IS NULL OR end_reason IN ('agent_close', 'ws_orphan_reap')) + AND (COALESCE(message_count, 0) > 0 OR EXISTS ( + SELECT 1 FROM messages WHERE messages.session_id = sessions.id LIMIT 1 + )) + ORDER BY started_at DESC + LIMIT 1 + """, + ( + source, + stable_user_id, + chat_type, + thread_id, + len(routing_prefix), + routing_prefix, + ), + ).fetchone() return dict(row) if row else None def end_session(self, session_id: str, end_reason: str) -> None: diff --git a/tests/gateway/test_bluebubbles_session_key.py b/tests/gateway/test_bluebubbles_session_key.py new file mode 100644 index 0000000000000..527516e957d20 --- /dev/null +++ b/tests/gateway/test_bluebubbles_session_key.py @@ -0,0 +1,314 @@ +"""BlueBubbles DM session-key stability across chatGuid variants. + +BlueBubbles delivers the same 1:1 conversation under several chat_id forms: +the service-prefixed GUID (``iMessage;-;+1555…``, ``any;-;+1555…``) when the +webhook carries one, and the bare handle (``+1555…``) when it does not (the +``chat_identifier`` fallback in the adapter). Keying sessions on the raw +chat_id splits one conversation across several SessionEntries; a message +landing on a long-untouched variant then trips the idle reset and wipes an +actively-used conversation. +""" + +from datetime import datetime, timedelta + +import pytest + +from gateway.config import GatewayConfig, Platform +from gateway.session import ( + SessionEntry, + SessionSource, + SessionStore, + build_session_key, + canonical_bluebubbles_identifier, +) + + +HANDLE = "+15551234567" + + +class TestCanonicalBlueBubblesIdentifier: + @pytest.mark.parametrize( + "raw", + [ + f"iMessage;-;{HANDLE}", + f"any;-;{HANDLE}", + f"SMS;-;{HANDLE}", + HANDLE, + ], + ) + def test_dm_guid_variants_collapse_to_bare_handle(self, raw): + assert canonical_bluebubbles_identifier(raw) == HANDLE + + def test_email_handle_is_preserved(self): + assert ( + canonical_bluebubbles_identifier("iMessage;-;user@example.com") + == "user@example.com" + ) + + def test_group_guid_is_left_alone(self): + """Group GUIDs use ``;+;`` and are opaque — never rewrite them.""" + guid = "iMessage;+;chat9876543210" + assert canonical_bluebubbles_identifier(guid) == guid + + @pytest.mark.parametrize("value", ["", None]) + def test_empty_values_pass_through(self, value): + assert canonical_bluebubbles_identifier(value) == value + + +class TestBlueBubblesDMSessionKey: + def _key(self, chat_id): + return build_session_key( + SessionSource( + platform=Platform.BLUEBUBBLES, + chat_id=chat_id, + chat_type="dm", + user_id=HANDLE, + ) + ) + + def test_all_dm_variants_share_one_session_key(self): + """The regression: three forms of one conversation, one key.""" + keys = { + self._key(f"iMessage;-;{HANDLE}"), + self._key(f"any;-;{HANDLE}"), + self._key(HANDLE), + } + assert keys == {f"agent:main:bluebubbles:dm:{HANDLE}"} + + def test_distinct_contacts_stay_isolated(self): + assert self._key(f"any;-;{HANDLE}") != self._key("any;-;+15559999999") + + def test_group_chats_keep_their_raw_guid(self): + """Groups must be untouched: their GUID is not a participant handle.""" + source = SessionSource( + platform=Platform.BLUEBUBBLES, + chat_id="iMessage;+;chat9876543210", + chat_type="group", + user_id=HANDLE, + ) + key = build_session_key(source) + assert "iMessage;+;chat9876543210" in key + + def test_threaded_dm_variants_share_one_key(self): + def keyed(chat_id): + return build_session_key( + SessionSource( + platform=Platform.BLUEBUBBLES, + chat_id=chat_id, + chat_type="dm", + user_id=HANDLE, + thread_id="t1", + ) + ) + + assert keyed(f"any;-;{HANDLE}") == keyed(HANDLE) + assert keyed(HANDLE) == f"agent:main:bluebubbles:dm:{HANDLE}:t1" + + def test_other_platforms_unaffected(self): + """Only BlueBubbles is canonicalized; a lookalike id elsewhere is raw.""" + source = SessionSource( + platform=Platform.TELEGRAM, + chat_id=f"iMessage;-;{HANDLE}", + chat_type="dm", + user_id="u1", + ) + assert build_session_key(source) == ( + f"agent:main:telegram:dm:iMessage;-;{HANDLE}" + ) + + +class TestBlueBubblesRoutingMigration: + @pytest.fixture(autouse=True) + def _isolated_db(self, tmp_path, monkeypatch): + # Each test gets its own state.db — DEFAULT_DB_PATH is module-level and + # would otherwise be shared by every SessionDB() in this file's + # subprocess, leaking sessions between tests. + import hermes_state + + monkeypatch.setattr(hermes_state, "DEFAULT_DB_PATH", tmp_path / "state.db") + + @staticmethod + def _seed(store, *, chat_id, session_id, updated_at, created_at=None, handle=HANDLE): + """Seed one legacy route: index entry + durable session + a message.""" + source = SessionSource( + platform=Platform.BLUEBUBBLES, + chat_id=chat_id, + chat_type="dm", + user_id=handle, + ) + session_key = f"agent:main:bluebubbles:dm:{chat_id}" + store._entries[session_key] = SessionEntry( + session_key=session_key, + session_id=session_id, + created_at=created_at or updated_at, + updated_at=updated_at, + origin=source, + platform=Platform.BLUEBUBBLES, + ) + store._db.create_session( + session_id, + "bluebubbles", + user_id=handle, + session_key=session_key, + chat_id=chat_id, + chat_type="dm", + ) + store._db.append_message(session_id, "user", "hello") + return source + + def test_loaded_legacy_alias_is_replaced_by_canonical_key(self, tmp_path): + """An upgrade leaves one durable route and one restart candidate.""" + sessions_dir = tmp_path / "sessions" + config = GatewayConfig() + now = datetime.now() + + initial = SessionStore(sessions_dir=sessions_dir, config=config) + initial._loaded = True + source = self._seed( + initial, + chat_id=f"any;-;{HANDLE}", + session_id="legacy-bb-session", + updated_at=now, + ) + initial._save_entries() + + restarted = SessionStore(sessions_dir=sessions_dir, config=config) + recovered = restarted.get_or_create_session(source) + canonical_key = build_session_key(source) + + assert recovered.session_id == "legacy-bb-session" + aliases = { + key + for key, entry in restarted._entries.items() + if entry.session_id == "legacy-bb-session" + } + assert aliases == {canonical_key} + assert restarted.suspend_recently_active(max_age_seconds=120) == 1 + assert set( + restarted._db.load_gateway_routing_entries( + scope=restarted._routing_scope() + ) + ) == {canonical_key} + assert restarted._db.get_session("legacy-bb-session")["session_key"] == ( + canonical_key + ) + + def test_loaded_alias_collision_keeps_most_recent_session(self, tmp_path): + """Multiple old GUID variants collapse without reviving a stale one.""" + sessions_dir = tmp_path / "sessions" + config = GatewayConfig() + now = datetime.now() + + initial = SessionStore(sessions_dir=sessions_dir, config=config) + initial._loaded = True + self._seed( + initial, + chat_id=f"iMessage;-;{HANDLE}", + session_id="older-bb-session", + updated_at=now - timedelta(minutes=5), + ) + source = self._seed( + initial, + chat_id=f"any;-;{HANDLE}", + session_id="newer-bb-session", + updated_at=now, + ) + initial._save_entries() + + restarted = SessionStore(sessions_dir=sessions_dir, config=config) + recovered = restarted.get_or_create_session(source) + canonical_key = build_session_key(source) + + assert recovered.session_id == "newer-bb-session" + assert set(restarted._entries) == {canonical_key} + + def test_retired_alias_sibling_cannot_strand_the_live_transcript(self, tmp_path): + """A retired sibling must be ended, not merely dropped from the index. + + The stray bare-handle session is typically created *later* than the real + conversation (a GUID-less webhook lands mid-thread), so a peer lookup + ordered by ``started_at`` prefers it. If the migration only drops it + from the routing index, it stays live in state.db under the canonical + key and a later recovery reopens the near-empty stray, orphaning the + real transcript. + """ + sessions_dir = tmp_path / "sessions" + config = GatewayConfig() + now = datetime.now() + + store = SessionStore(sessions_dir=sessions_dir, config=config) + store._loaded = True + real_src = self._seed( + store, + chat_id=f"any;-;{HANDLE}", + session_id="real-session", + created_at=now - timedelta(hours=2), + updated_at=now, + ) + self._seed( + store, + chat_id=HANDLE, + session_id="stray-session", + created_at=now, + updated_at=now - timedelta(hours=1), + ) + # Pin the started_at ordering the bug depends on: the stray is newer. + store._db._conn.execute( + "UPDATE sessions SET started_at = 1000 WHERE id = 'real-session'" + ) + store._db._conn.execute( + "UPDATE sessions SET started_at = 2000 WHERE id = 'stray-session'" + ) + store._db._conn.commit() + store._save_entries() + + restarted = SessionStore(sessions_dir=sessions_dir, config=config) + resumed = restarted.get_or_create_session(real_src) + assert resumed.session_id == "real-session" + + stray_row = restarted._db.get_session("stray-session") + assert stray_row["ended_at"] is not None, "retired sibling left live" + + recovered = restarted._db.find_latest_gateway_session_for_peer( + source="bluebubbles", + user_id=HANDLE, + session_key=build_session_key(real_src), + chat_id=f"any;-;{HANDLE}", + chat_type="dm", + match_by_participant_identity=True, + ) + assert recovered is not None + assert recovered["id"] == "real-session", ( + "recovery reopened the retired stray and orphaned the transcript" + ) + + def test_stale_canonical_winner_does_not_retire_live_alias(self, tmp_path): + """Stale pruning may promote an alias that migration planned to retire.""" + sessions_dir = tmp_path / "sessions" + config = GatewayConfig() + now = datetime.now() + + initial = SessionStore(sessions_dir=sessions_dir, config=config) + initial._loaded = True + live_source = self._seed( + initial, + chat_id=f"any;-;{HANDLE}", + session_id="live-alias", + updated_at=now - timedelta(minutes=5), + ) + self._seed( + initial, + chat_id=HANDLE, + session_id="ended-canonical", + updated_at=now, + ) + initial._db.end_session("ended-canonical", "user_reset") + initial._save_entries() + + restarted = SessionStore(sessions_dir=sessions_dir, config=config) + recovered = restarted.get_or_create_session(live_source) + + assert recovered.session_id == "live-alias" + assert restarted._db.get_session("live-alias")["ended_at"] is None + canonical_key = build_session_key(live_source) + assert restarted._entries[canonical_key].session_id == "live-alias" diff --git a/tests/test_hermes_state.py b/tests/test_hermes_state.py index c12b17a7f805d..071c32a346926 100644 --- a/tests/test_hermes_state.py +++ b/tests/test_hermes_state.py @@ -5708,6 +5708,118 @@ def test_gateway_session_peer_round_trip_and_recovery(db): assert recovered["id"] == "gw-session" +def test_bluebubbles_legacy_guid_session_recovers_under_canonical_key(db): + """Canonicalizing the BlueBubbles DM key must not orphan existing sessions. + + Sessions created before ``canonical_bluebubbles_identifier`` are keyed on + the raw chat GUID (``iMessage;-;+1555…``). After the upgrade the routing + key is the bare handle, and BlueBubbles may also report the same DM using a + different service prefix. The stable participant identity and routing + namespace must still recover the existing transcript. + + Drives ``build_session_key`` rather than hardcoding the new key, so this + fails if canonicalization is dropped or alias recovery is not enabled. + """ + from gateway.config import Platform + from gateway.session import SessionSource, build_session_key + + legacy_chat_id = "iMessage;-;+15551234567" + current_chat_id = "any;-;+15551234567" + legacy_key = "agent:main:bluebubbles:dm:" + legacy_chat_id + db.create_session( + "legacy-bb-session", + "bluebubbles", + user_id="+15551234567", + session_key=legacy_key, + chat_id=legacy_chat_id, + chat_type="dm", + thread_id=None, + ) + db.append_message("legacy-bb-session", "user", "hello") + + # The server changed the service prefix before the first post-upgrade event. + source = SessionSource( + platform=Platform.BLUEBUBBLES, + chat_id=current_chat_id, + chat_type="dm", + user_id="+15551234567", + ) + new_key = build_session_key(source) + assert new_key != legacy_key, "canonicalization is not in effect" + + recovered = db.find_latest_gateway_session_for_peer( + source="bluebubbles", + user_id=source.user_id, + session_key=new_key, + # Keep the current transport GUID; alias recovery may ignore this only + # because the non-empty participant identity and routing scope match. + chat_id=source.chat_id, + chat_type=source.chat_type, + match_by_participant_identity=True, + ) + assert recovered is not None, "legacy GUID-keyed session was orphaned" + assert recovered["id"] == "legacy-bb-session" + + +def test_gateway_session_chat_alias_recovery_is_opt_in(db): + """chat_id aliasing must never apply unless the caller asks for it. + + Only transports whose session key deliberately aliases several transport + chat IDs onto one participant identity may ignore chat_id. Every other + caller must keep the strict peer tuple, so a mismatched chat_id finds + nothing. Replaces the pre-alias guard that pinned "source.chat_id is never + canonicalized": that invariant is now enforced by opt-in, not by the + fallback silently depending on a raw chat_id. + """ + db.create_session( + "legacy-optin-session", + "bluebubbles", + user_id="+15550001111", + session_key="agent:main:bluebubbles:dm:iMessage;-;+15550001111", + chat_id="iMessage;-;+15550001111", + chat_type="dm", + ) + db.append_message("legacy-optin-session", "user", "hello") + + lookup = dict( + source="bluebubbles", + user_id="+15550001111", + session_key="agent:main:bluebubbles:dm:+15550001111", + chat_id="any;-;+15550001111", + chat_type="dm", + ) + assert db.find_latest_gateway_session_for_peer(**lookup) is None + opted_in = db.find_latest_gateway_session_for_peer( + **lookup, match_by_participant_identity=True + ) + assert opted_in is not None + assert opted_in["id"] == "legacy-optin-session" + + +def test_gateway_session_chat_alias_recovery_requires_user_id(db): + """Ignoring chat_id is safe only with a stable participant identity.""" + db.create_session( + "legacy-bb-session", + "bluebubbles", + user_id="+15551234567", + session_key="agent:main:bluebubbles:dm:iMessage;-;+15551234567", + chat_id="iMessage;-;+15551234567", + chat_type="dm", + ) + db.append_message("legacy-bb-session", "user", "hello") + + recovered = db.find_latest_gateway_session_for_peer( + source="bluebubbles", + user_id=None, + session_key="agent:main:bluebubbles:dm:+15551234567", + chat_id="any;-;+15551234567", + chat_type="dm", + match_by_participant_identity=True, + ) + + assert recovered is None + + def test_gateway_session_recovery_reopens_ws_orphan_reap_rows(db): """Rows wrongly ended by the TUI ws-orphan reaper must be recoverable (#63207).""" db.create_session(