From e07627896fbb911d63cc695175e77fa13dbcfdee Mon Sep 17 00:00:00 2001 From: Andrey <3605840+Diaspar4u@users.noreply.github.com> Date: Mon, 31 Aug 2026 20:14:23 -0400 Subject: [PATCH 1/2] fix(gateway): honor per-platform session isolation overrides Co-authored-by: dhruv kejriwal <96516827+dhruvkej9@users.noreply.github.com> --- gateway/config_loader.py | 13 +- gateway/run.py | 17 ++- gateway/run_inbound.py | 8 +- gateway/session.py | 62 +++++++++- gateway/session_recovery.py | 55 ++++++--- gateway/slash_commands_session.py | 22 +++- tests/gateway/test_config.py | 25 ++++ tests/gateway/test_session.py | 189 ++++++++++++++++++++++++++++++ 8 files changed, 356 insertions(+), 35 deletions(-) diff --git a/gateway/config_loader.py b/gateway/config_loader.py index 46fe00fc66c1..81340dfdd751 100644 --- a/gateway/config_loader.py +++ b/gateway/config_loader.py @@ -279,13 +279,24 @@ def bridge_platform_shared_keys( if isinstance(ov_data, dict) } enabled_was_explicit = cfg_toplevel and "enabled" in platform_cfg - if not bridged and not enabled_was_explicit and not has_channel_overrides: + nested_extra = platform_cfg.get("extra") if cfg_toplevel else None + has_nested_extra = isinstance(nested_extra, dict) + if ( + not bridged + and not enabled_was_explicit + and not has_channel_overrides + and not has_nested_extra + ): continue plat_data = _dict_slot(platforms_data, plat.value) extra = _dict_slot(plat_data, "extra") if enabled_was_explicit: plat_data["enabled"] = platform_cfg["enabled"] extra["_enabled_explicit"] = True + # Preserve a platform's own nested ``extra`` values. Top-level bridged + # keys are applied afterwards and retain their documented precedence. + if has_nested_extra: + extra.update(nested_extra) extra.update(bridged) diff --git a/gateway/run.py b/gateway/run.py index 62f69cf98d0a..59d18651352c 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -2168,7 +2168,7 @@ def _load_bridge_config(config_path: Path) -> dict: ChannelOverride, Platform, GatewayConfig, PlatformConfig, _getenv, load_gateway_config) from gateway.session import ( AsyncSessionStore, SessionStore, SessionSource, SessionContext, build_session_key, - profile_from_session_key_namespace) + profile_from_session_key_namespace, resolve_runner_session_isolation) # Telegram topic routing (#22773, regression fixed #52060): a # ``telegram::`` cron target is ambiguous — a forum-style topic in a # private chat and a genuine Bot API channel Direct-Messages topic share the same shape and need OPPOSITE @@ -3917,6 +3917,12 @@ def _warn_if_docker_media_delivery_is_risky(self) -> None: exit_reason = property(lambda self: self._exit_reason) exit_code = property(lambda self: self._exit_code) + def _resolve_session_isolation_for_source( + self, source: SessionSource + ) -> tuple[bool, bool]: + """Resolve the store's effective scope, with a bare-runner config fallback.""" + return resolve_runner_session_isolation(self, source) + def _session_key_for_source(self, source: SessionSource) -> str: """Resolve the current session key for a source, honoring gateway config when available.""" if hasattr(self, "session_store") and self.session_store is not None: @@ -3944,10 +3950,13 @@ def _session_key_for_source(self, source: SessionSource) -> str: _profile = get_active_profile_name() or "default" except Exception: _profile = None + group_per_user, thread_per_user = self._resolve_session_isolation_for_source(source) return build_session_key( - source, group_sessions_per_user=getattr(config, "group_sessions_per_user", True), - thread_sessions_per_user=getattr(config, "thread_sessions_per_user", False), - profile=_profile) + source, + group_sessions_per_user=group_per_user, + thread_sessions_per_user=thread_per_user, + profile=_profile, + ) # Telegram General topic in forum-enabled private chats: clients omit message_thread_id or send "1"; both = root. _TELEGRAM_GENERAL_TOPIC_IDS = frozenset({"", "1"}) diff --git a/gateway/run_inbound.py b/gateway/run_inbound.py index 614434de50bc..fa225f7d2cff 100644 --- a/gateway/run_inbound.py +++ b/gateway/run_inbound.py @@ -27,7 +27,7 @@ ) from gateway.session import ( SessionSource, build_session_context, is_shared_multi_user_session, - neutralize_untrusted_inline_text, + neutralize_untrusted_inline_text, resolve_runner_session_isolation, ) from gateway.turn_lease import TurnLeaseTimeoutError from typing import Any, Dict, List, Optional, Tuple @@ -1408,9 +1408,11 @@ def _restore_pending_one_turn_model_override(self, session_key: str, run_generat def _prefix_inbound_sender_context(self, event: MessageEvent, source: SessionSource, message_text: str) -> str: """Attribute the sender in shared multi-user sessions and prepend history-backfill channel context.""" + group_per_user, thread_per_user = resolve_runner_session_isolation(self, source) _is_shared_multi_user = is_shared_multi_user_session( - source, group_sessions_per_user=getattr(self.config, "group_sessions_per_user", True), - thread_sessions_per_user=getattr(self.config, "thread_sessions_per_user", False), + source, + group_sessions_per_user=group_per_user, + thread_sessions_per_user=thread_per_user, ) if _is_shared_multi_user and source.user_name: # Display names are attacker-influenceable: neutralize newlines/control chars or a diff --git a/gateway/session.py b/gateway/session.py index a2be18cbbcdd..e28f4c650aa8 100644 --- a/gateway/session.py +++ b/gateway/session.py @@ -621,6 +621,52 @@ def build_channel_continuity_note(entry: "SessionEntry", source: SessionSource) ) +def effective_session_thread_id(source: SessionSource) -> Optional[str]: + """Return the thread identity used by session routing.""" + if source.thread_id: + return source.thread_id + if source.platform == Platform.DISCORD and source.chat_type != "dm": + return source.prospective_thread_id + return None + + +def effective_session_chat_type(source: SessionSource) -> str: + """Return the chat type represented by the routed session key.""" + if effective_session_thread_id(source) and not source.thread_id: + return "thread" + return source.chat_type + + +def resolve_session_isolation( + config: Any, source: SessionSource +) -> tuple[bool, bool]: + """Resolve global session-isolation defaults plus platform overrides.""" + group_per_user = getattr(config, "group_sessions_per_user", True) + thread_per_user = getattr(config, "thread_sessions_per_user", False) + platform_cfg = getattr(config, "platforms", {}).get(source.platform) + extra = getattr(platform_cfg, "extra", None) if platform_cfg else None + if isinstance(extra, dict): + group_per_user = extra.get("group_sessions_per_user", group_per_user) + thread_per_user = extra.get("thread_sessions_per_user", thread_per_user) + return group_per_user, thread_per_user + + +def resolve_runner_session_isolation( + runner: Any, source: SessionSource +) -> tuple[bool, bool]: + """Resolve through a runner's real store, with a config fallback for bare callers.""" + store = getattr(runner, "session_store", None) + resolver = getattr(store, "_resolve_session_isolation", None) + if callable(resolver): + try: + resolved = resolver(source) + if isinstance(resolved, tuple) and len(resolved) == 2: + return bool(resolved[0]), bool(resolved[1]) + except Exception: + pass + return resolve_session_isolation(getattr(runner, "config", None), source) + + def is_shared_multi_user_session( source: SessionSource, *, group_sessions_per_user: bool = True, thread_sessions_per_user: bool = False, @@ -629,7 +675,11 @@ def is_shared_multi_user_session( isolation rules in :func:`build_session_key`).""" if source.chat_type == "dm": return False - return not (thread_sessions_per_user if source.thread_id else group_sessions_per_user) + return not ( + thread_sessions_per_user + if effective_session_thread_id(source) + else group_sessions_per_user + ) def _session_key_namespace(profile: Optional[str]) -> str: @@ -680,8 +730,8 @@ def build_session_key( # Discord auto-thread continuity: key a channel-initiating message on the thread it WILL be # delivered into (prospective_thread_id), and normalize the chat_type slot to "thread" so # in-thread follow-ups byte-match. A real thread_id always wins. DMs use thread_id only. - thread_id = source.thread_id or (None if is_dm else source.prospective_thread_id) - chat_type_slot = "thread" if thread_id and not source.thread_id else source.chat_type + thread_id = effective_session_thread_id(source) + chat_type_slot = effective_session_chat_type(source) if is_dm: # No chat_id: fall back to the sender id before the bare per-platform sink, or every # chat_id-less DM shares one agent. @@ -1268,9 +1318,11 @@ def build_session_context( ) -> SessionContext: """Build a full session context (for system prompt injection).""" connected = config.get_connected_platforms() + group_per_user, thread_per_user = resolve_session_isolation(config, source) shared = is_shared_multi_user_session( - source, group_sessions_per_user=getattr(config, "group_sessions_per_user", True), - thread_sessions_per_user=getattr(config, "thread_sessions_per_user", False), + source, + group_sessions_per_user=group_per_user, + thread_sessions_per_user=thread_per_user, ) context = SessionContext( source=source, connected_platforms=connected, shared_multi_user_session=shared, diff --git a/gateway/session_recovery.py b/gateway/session_recovery.py index a466af5a25de..c674f2447b46 100644 --- a/gateway/session_recovery.py +++ b/gateway/session_recovery.py @@ -73,35 +73,49 @@ def _active_profile_name() -> str: def _recovered_row_allowed_for_active_profile( self, *, requested_session_key: str, recovered: Dict[str, Any] ) -> bool: - """Prevent a gateway from reviving another profile's row. Single-profile: the row's - namespace must match the ACTIVE profile. Multiplexed: it must match the requested key's - namespace (the active profile is meaningless there). Keyless rows stay adoptable. - - Multiplexed: several profiles serve traffic at once, so the active profile is meaningless — the - requested key carries the profile the turn was routed to, and the recovered row must sit in the same - ``agent::`` namespace (#74285). Rows with no key namespace stay adoptable in both modes - (legacy/keyless data owned by this store). + """Prevent a gateway from reviving another profile's row. + + Single-profile recovery follows the durable owner first, then the row's key namespace. + Multiplexed recovery follows the requested key's profile namespace because no one process-active + profile owns all routed turns. Rows with neither owner nor key remain adoptable legacy data. """ + multiplex_profiles = getattr(self.config, "multiplex_profiles", False) + active_profile = self._active_profile_name() + requested_profile = self._profile_from_session_key(requested_session_key) + + durable_profile = str(recovered.get("profile_name") or "").strip() + if durable_profile: + if multiplex_profiles: + return requested_profile is None or durable_profile == requested_profile + return durable_profile == active_profile + recovered_key = str(recovered.get("session_key") or "") if not recovered_key or recovered_key == requested_session_key: return True recovered_profile = self._profile_from_session_key(recovered_key) if recovered_profile is None: return True - if getattr(self.config, "multiplex_profiles", False): - requested_profile = self._profile_from_session_key(requested_session_key) + if multiplex_profiles: return requested_profile is None or recovered_profile == requested_profile - return recovered_profile == self._active_profile_name() + return recovered_profile == active_profile def _generate_session_key(self, source: SessionSource, key_source: Optional[SessionSource] = None) -> str: """Session key for *source* (profile from *source*; key from *key_source* if given).""" from gateway.session import build_session_key + group_per_user, thread_per_user = self._resolve_session_isolation( + key_source if key_source is not None else source + ) return build_session_key( key_source if key_source is not None else source, - group_sessions_per_user=getattr(self.config, "group_sessions_per_user", True), - thread_sessions_per_user=getattr(self.config, "thread_sessions_per_user", False), + group_sessions_per_user=group_per_user, + thread_sessions_per_user=thread_per_user, profile=self._resolve_profile_for_key(source)) + def _resolve_session_isolation(self, source: SessionSource) -> tuple[bool, bool]: + """Resolve group/thread isolation from platform config, then global defaults.""" + from gateway.session import resolve_session_isolation + return resolve_session_isolation(self.config, source) + def _legacy_slack_session_key(self, source: SessionSource) -> Optional[str]: """Pre-workspace Slack key for an explicitly scoped source. Deliberately Slack-only: an unscoped Slack session may be claimed by only one workspace (old key cannot tell teams).""" @@ -169,12 +183,14 @@ def _find_gateway_session_row( """Query one durable gateway session row. Scoped Slack lookups disable SessionDB's platform/chat/user fallback: that tuple has no workspace id and could revive another team's session; the caller performs one explicit exact lookup of the old unscoped key instead.""" + from gateway.session import effective_session_thread_id + return self._peer_row( self._db_for_key(session_key), source=source.platform.value, session_key=session_key, user_id=source.user_id, chat_id=source.chat_id if allow_peer_fallback else None, chat_type=source.chat_type if allow_peer_fallback else None, - thread_id=source.thread_id, raise_on_lookup_error=raise_on_lookup_error) + thread_id=effective_session_thread_id(source), raise_on_lookup_error=raise_on_lookup_error) @staticmethod def _peer_row(db, *, source: str, session_key: str, raise_on_lookup_error: bool = False, @@ -333,13 +349,16 @@ def _record_gateway_session_peer( db = self._db_for_key(session_key) if not db or not source: return + from gateway.session import effective_session_chat_type, effective_session_thread_id + recorder = getattr(db, "record_gateway_session_peer", None) if not callable(recorder): return from gateway.session_identity import transport_profile_of peer = dict( 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) + chat_id=source.chat_id, chat_type=effective_session_chat_type(source), + thread_id=effective_session_thread_id(source)) try: recorder( session_id, **peer, display_name=display_name or source.chat_name, @@ -416,15 +435,17 @@ def _session_create_kwargs( """kwargs for ``SessionDB.create_session``. Identity (origin_json) and lineage (parent/_reset_from) land atomically in the INSERT so a crash right after cannot strand the row unroutable.""" + from gateway.session import effective_session_chat_type, effective_session_thread_id from gateway.session_identity import transport_profile_of + return { "session_id": session_id, "source": source_value, "user_id": origin.user_id if origin else None, "session_key": session_key, "chat_id": origin.chat_id if origin else None, - "chat_type": origin.chat_type if origin else None, - "thread_id": origin.thread_id if origin else None, + "chat_type": effective_session_chat_type(origin) if origin else None, + "thread_id": effective_session_thread_id(origin) if origin else None, "profile_name": origin.profile if origin else None, "transport_profile": transport_profile_of(origin), "origin_json": _origin_json(origin), diff --git a/gateway/slash_commands_session.py b/gateway/slash_commands_session.py index c35965b1483b..fd9ceeb59a7a 100644 --- a/gateway/slash_commands_session.py +++ b/gateway/slash_commands_session.py @@ -19,7 +19,13 @@ from gateway.config import Platform from gateway.platforms.base import EphemeralReply from gateway.platforms.event import MessageEvent, MessageType -from gateway.session import SessionSource, build_session_key, is_shared_multi_user_session +from gateway.session import ( + SessionSource, + build_session_key, + effective_session_thread_id, + is_shared_multi_user_session, + resolve_runner_session_isolation, +) from gateway.session_transcript import TranscriptReadError from gateway.slash_commands_branch_thread import ( BRANCH_THREAD_PLATFORMS, branch_dest_source, branch_thread_parent, format_thread_ref, parse_branch_args, @@ -274,7 +280,9 @@ def _same_origin_chat(self, current: SessionSource, origin: Optional[SessionSour if origin.platform != current.platform or origin.chat_id != current.chat_id: return False # thread_id is part of every session key: threads of one chat are DIFFERENT sessions. - if _sattr(current, "thread_id") != _sattr(origin, "thread_id"): + if str(effective_session_thread_id(current) or "") != str( + effective_session_thread_id(origin) or "" + ): return False if _sattr(current, "chat_type").lower() in _DM_CHAT_TYPES: # An equal non-empty chat_id IS the DM key. build_session_key falls back to the @@ -296,9 +304,12 @@ def _same_origin_chat(self, current: SessionSource, origin: Optional[SessionSour def _is_shared_session_source(self, source: SessionSource) -> bool: """Whether *source*'s session key is shared by every participant (not per-user); mirrors build_session_key's isolation rules so the guards stay in lock-step with the key.""" + group_per_user, thread_per_user = resolve_runner_session_isolation(self, source) return is_shared_multi_user_session( - source, group_sessions_per_user=getattr(self.config, "group_sessions_per_user", True), - thread_sessions_per_user=getattr(self.config, "thread_sessions_per_user", False)) + source, + group_sessions_per_user=group_per_user, + thread_sessions_per_user=thread_per_user, + ) def _resume_caller_is_admin(self, source: SessionSource) -> bool: """Whether *source* is an EXPLICITLY-configured admin (cross-origin /resume, /sessions). @@ -323,8 +334,9 @@ def _persisted_row_proves_owner(self, source: SessionSource, row: dict) -> bool: if not caller_uid: return False row_thread = str(row.get("thread_id") or "") + caller_thread = str(effective_session_thread_id(source) or "") if not (row_src and caller_src and str(row_src) == str(caller_src) - and row_thread == _sattr(source, "thread_id")): + and row_thread == caller_thread): return False # blank/legacy source cannot prove the platform; other thread = other session row_uid = str(row.get("user_id") or "") row_chat = str(row.get("chat_id") or "") diff --git a/tests/gateway/test_config.py b/tests/gateway/test_config.py index e0d8af4fcb46..7aaedc244978 100644 --- a/tests/gateway/test_config.py +++ b/tests/gateway/test_config.py @@ -245,6 +245,31 @@ def test_roundtrip_preserves_unauthorized_dm_behavior(self): assert restored.unauthorized_dm_behavior == "ignore" assert restored.platforms[Platform.WHATSAPP].extra["unauthorized_dm_behavior"] == "pair" + @pytest.mark.parametrize("with_bridged_key", [False, True]) + def test_top_level_platform_nested_extra_survives_loading( + self, tmp_path, monkeypatch, with_bridged_key + ): + hermes_home = tmp_path / ".hermes" + hermes_home.mkdir() + policy = " group_policy: open\n" if with_bridged_key else "" + (hermes_home / "config.yaml").write_text( + "whatsapp:\n" + " extra:\n" + " group_sessions_per_user: false\n" + " bridge_port: 3000\n" + " group_policy: allowlist\n" + f"{policy}", + encoding="utf-8", + ) + monkeypatch.setenv("HERMES_HOME", str(hermes_home)) + + extra = load_gateway_config().platforms[Platform.WHATSAPP].extra + + assert extra["group_sessions_per_user"] is False + assert extra["bridge_port"] == 3000 + if with_bridged_key: + assert extra["group_policy"] == "open" + def test_email_defaults_to_ignore_for_unauthorized_dm_behavior(self): config = GatewayConfig( platforms={Platform.EMAIL: PlatformConfig(enabled=True)}, diff --git a/tests/gateway/test_session.py b/tests/gateway/test_session.py index 824a830eb448..ac5368c34b3b 100644 --- a/tests/gateway/test_session.py +++ b/tests/gateway/test_session.py @@ -18,6 +18,7 @@ build_session_context_prompt, build_session_key, canonical_whatsapp_identifier, + is_shared_multi_user_session, neutralize_untrusted_inline_text, ) @@ -302,6 +303,194 @@ def test_prompt_quotes_untrusted_metadata_labels(self): assert "\n**Platform notes:** hacked" not in prompt +class TestPerPlatformSessionIsolation: + @staticmethod + def _whatsapp_source(user_id, user_name): + return SessionSource( + platform=Platform.WHATSAPP, + chat_id="120363000000000000@g.us", + chat_type="group", + user_id=user_id, + user_name=user_name, + ) + + def test_override_drives_key_context_sender_and_authorization( + self, tmp_path, monkeypatch + ): + import hermes_state + from gateway.run import GatewayRunner + + monkeypatch.setattr(hermes_state, "DEFAULT_DB_PATH", tmp_path / "state.db") + config = GatewayConfig( + group_sessions_per_user=True, + platforms={ + Platform.WHATSAPP: PlatformConfig( + enabled=True, + extra={"group_sessions_per_user": False}, + ) + }, + ) + store = SessionStore(sessions_dir=tmp_path, config=config) + db = store._db + assert isinstance(db, SessionDB) + runner = GatewayRunner.__new__(GatewayRunner) + runner.config = config + runner.session_store = store + alice = self._whatsapp_source("alice@lid", "Alice") + bob = self._whatsapp_source("bob@lid", "Bob") + + alice_entry = store.get_or_create_session(alice) + bob_entry = store.get_or_create_session(bob) + context = build_session_context(alice, config) + + assert alice_entry.session_key == bob_entry.session_key + assert alice_entry.session_id == bob_entry.session_id + assert runner._session_key_for_source(alice) == alice_entry.session_key + bare_runner = GatewayRunner.__new__(GatewayRunner) + bare_runner.config = config + assert bare_runner._session_key_for_source(alice) == alice_entry.session_key + assert context.shared_multi_user_session is True + assert runner._prefix_inbound_sender_context( + MessageEvent(text="hello", source=alice), alice, "hello" + ) == "[Alice] hello" + assert runner._same_origin_chat(alice, bob) is True + assert runner._persisted_row_proves_owner( + alice, + { + "source": "whatsapp", + "user_id": "bob@lid", + "chat_id": alice.chat_id, + "thread_id": None, + }, + ) is True + + config.platforms[Platform.WHATSAPP].extra["group_sessions_per_user"] = True + assert store._generate_session_key(alice) != store._generate_session_key(bob) + assert runner._same_origin_chat(alice, bob) is False + assert runner._persisted_row_proves_owner( + alice, + { + "source": "whatsapp", + "user_id": "bob@lid", + "chat_id": alice.chat_id, + "thread_id": None, + }, + ) is False + db.close() + + def test_prospective_thread_is_discord_only_and_persists_one_identity( + self, tmp_path, monkeypatch + ): + import hermes_state + + monkeypatch.setattr(hermes_state, "DEFAULT_DB_PATH", tmp_path / "state.db") + config = GatewayConfig( + group_sessions_per_user=True, + platforms={ + Platform.DISCORD: PlatformConfig( + enabled=True, + extra={"thread_sessions_per_user": False}, + ) + }, + ) + store = SessionStore(sessions_dir=tmp_path, config=config) + db = store._db + assert isinstance(db, SessionDB) + discord = SessionSource( + platform=Platform.DISCORD, + chat_id="channel-1", + chat_type="group", + user_id="alice", + prospective_thread_id="future-thread", + ) + telegram = replace(discord, platform=Platform.TELEGRAM) + discord_dm = replace(discord, chat_type="dm", chat_id="dm-1") + + entry = store.get_or_create_session(discord) + row = db.get_session(entry.session_id) + assert row is not None + + assert entry.session_key == "agent:main:discord:thread:channel-1:future-thread" + assert row["chat_type"] == "thread" + assert row["thread_id"] == "future-thread" + assert build_session_key(telegram) == "agent:main:telegram:group:channel-1:alice" + assert is_shared_multi_user_session(telegram) is False + assert "future-thread" not in build_session_key(discord_dm) + db.close() + + def test_durable_profile_owner_controls_shared_key_recovery( + self, tmp_path, monkeypatch + ): + import hermes_state + + monkeypatch.setattr(hermes_state, "DEFAULT_DB_PATH", tmp_path / "state.db") + monkeypatch.setattr( + "hermes_cli.profiles.get_active_profile_name", lambda: "dev" + ) + source = self._whatsapp_source("member@lid", "Member") + isolated = SessionStore(sessions_dir=tmp_path, config=GatewayConfig()) + isolated_db = isolated._db + assert isinstance(isolated_db, SessionDB) + legacy = isolated.get_or_create_session(source) + isolated_db.create_session( + legacy.session_id, + source="whatsapp", + profile_name="dev", + ) + isolated.append_to_transcript( + legacy.session_id, + {"role": "user", "content": "legacy group context"}, + ) + isolated_db.close() + (tmp_path / "sessions.json").unlink() + + shared = SessionStore( + sessions_dir=tmp_path, + config=GatewayConfig( + platforms={ + Platform.WHATSAPP: PlatformConfig( + enabled=True, + extra={"group_sessions_per_user": False}, + ) + } + ), + ) + shared_db = shared._db + assert isinstance(shared_db, SessionDB) + recovered = shared.get_or_create_session(source) + + assert recovered.session_id == legacy.session_id + assert recovered.session_key == ( + "agent:main:whatsapp:group:120363000000000000@g.us" + ) + assert [ + message["content"] + for message in shared_db.get_messages_as_conversation( + recovered.session_id + ) + ] == ["legacy group context"] + + bare = object.__new__(SessionStore) + bare.config = GatewayConfig() + assert not bare._recovered_row_allowed_for_active_profile( + requested_session_key=recovered.session_key, + recovered={ + "session_key": recovered.session_key, + "profile_name": "other", + }, + ) + bare.config = GatewayConfig(multiplex_profiles=True) + for durable_profile, allowed in (("eva", True), ("other", False)): + assert bare._recovered_row_allowed_for_active_profile( + requested_session_key="agent:eva:whatsapp:group:family", + recovered={ + "session_key": "agent:main:whatsapp:group:family:member", + "profile_name": durable_profile, + }, + ) is allowed + shared_db.close() + + class TestSenderPrefixWithBackfill: """Regression: sender prefix must not wrap the backfill context block. From c3f33518acc5d110ea1d0872a12272970c9a1bc1 Mon Sep 17 00:00:00 2001 From: Andrey <3605840+Diaspar4u@users.noreply.github.com> Date: Sat, 19 Sep 2026 03:18:14 -0400 Subject: [PATCH 2/2] test(gateway): read stalled-session spools as UTF-8 --- tests/gateway/test_session.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/gateway/test_session.py b/tests/gateway/test_session.py index ac5368c34b3b..1c21cb6a7e45 100644 --- a/tests/gateway/test_session.py +++ b/tests/gateway/test_session.py @@ -1852,7 +1852,7 @@ def test_no_usable_db_counts_failures_and_spools_backlog_before_cap( store.append_to_transcript("s-dead", {"role": "user", "content": f"m{i}"}) assert store._transcript_append_failures["s-dead"] == threshold assert [r.levelno for r in caplog.records if "transcript append failed" in r.getMessage()][-1] == logging.ERROR - spooled = sorted(json.loads(p.read_text())["data"]["message"]["content"] + spooled = sorted(json.loads(p.read_text(encoding="utf-8"))["data"]["message"]["content"] for p in (tmp_path / "pending_messages").glob("pending-*.json")) assert spooled == [f"m{i}" for i in range(threshold)] # durable before the cap assert "s-dead" not in store._dirty_transcripts