diff --git a/gateway/run_busy.py b/gateway/run_busy.py index b2e5b091ae..2cad3b9871 100644 --- a/gateway/run_busy.py +++ b/gateway/run_busy.py @@ -20,7 +20,8 @@ from gateway.config import Platform from gateway.platforms.base import EphemeralReply from gateway.platforms.event import MessageEvent, MessageType from gateway.session import SessionSource -from typing import Any, Dict, Optional, Tuple, Union +from gateway.whatsapp_identity import canonical_whatsapp_identifier +from typing import Any, Dict, List, Optional, Tuple, Union if TYPE_CHECKING: # string annotations only; never imported at runtime (cycle) from gateway.run import GatewayRunner # noqa: F401 @@ -30,40 +31,44 @@ if TYPE_CHECKING: # string annotations only; never imported at runtime (cycle) logger = logging.getLogger("gateway.run") -def _key_namespace(key: str) -> str: - """``agent:`` prefix of a session key (``build_session_key``'s first two slots).""" - return ":".join(key.split(":")[:2]) +def _strip_slot(text: str, slot: str) -> Optional[str]: + """Remainder after ``slot`` when ``text`` starts with it as a WHOLE slot, else None. + + The ``:``-delimited slot layout is ``build_session_key``'s: an id that merely starts with another + must never match, so a whole-slot comparison is what every caller here uses. A text that IS the + slot yields ``""``.""" + if text == slot: + return "" + if text.startswith(slot + ":"): + return text[len(slot) + 1:] + return None def _tail_has_slot(tail: str, slot: str) -> bool: """True when ``tail``'s FIRST slot is ``slot`` (``tail`` is ``""`` when the key ends at the - chat id). Mirrors ``build_session_key``'s ``key == prefix or key.startswith(prefix + ":")`` - discipline, so an id that merely starts with another never matches.""" - return tail == slot or tail.startswith(slot + ":") + chat id).""" + return _strip_slot(tail, slot) is not None def _same_chat_key_slots( - key: str, *, namespace: str, platform: str, chat_id: str, scope_id: Optional[str], + key: str, *, prefix: str, chat_id: str, scope_id: Optional[str], ) -> Optional[Tuple[str, str]]: - """``(chat_type, tail)`` when ``key`` names the SAME chat, else None. + """``(chat_type, tail)`` when ``key`` names the SAME chat as ``prefix`` + ``chat_id``, else None. - Key layout: ``agent:::[:][:][:][:]``. - Only that fixed-shape head is split into slots (the namespace is always the first two); the chat - id and everything after it are matched as TEXT, because ids may themselves contain ``:`` (Matrix - ``!room:example.org``). ``scope_id`` is Slack's workspace slot — ``build_session_key`` emits it - there alone — and it may be absent from either side, since a key without it still names the same - chat; a key carrying a DIFFERENT known scope is another workspace's chat. ``tail`` is ``""`` when - the key ends at the chat id. + ``prefix`` is the key's fixed-shape head, ``agent:::``. Everything after it is + matched as TEXT, because ids may themselves contain ``:`` (Matrix ``!room:example.org``). + ``scope_id`` is Slack's workspace slot — ``build_session_key`` emits it there alone — and a key + without it still names the same chat; a key carrying a DIFFERENT known scope is another + workspace's chat. ``tail`` is ``""`` when the key ends at the chat id. """ - head = key.split(":", 3) - if len(head) < 4 or ":".join(head[:2]) != namespace or head[2] != platform: + if not key.startswith(prefix): return None - chat_type, _, rem = head[3].partition(":") - for prefix in ((f"{scope_id}:{chat_id}", chat_id) if scope_id else (chat_id,)): - if rem == prefix: - return chat_type, "" - if rem.startswith(prefix + ":"): - return chat_type, rem[len(prefix) + 1:] + chat_type, _, rem = key[len(prefix):].partition(":") + candidates = (f"{scope_id}:{chat_id}", chat_id) if scope_id else (chat_id,) + for candidate in candidates: + tail = _strip_slot(rem, candidate) + if tail is not None: + return chat_type, tail return None @@ -1068,65 +1073,69 @@ class GatewayBusySessionMixin: ) return f"⛔ /{canonical_cmd} is admin-only here. {suffix}" - def _same_chat_runs(self, source: SessionSource, own_key: str) -> list: + def _same_chat_runs(self, source: SessionSource, own_key: str) -> List[Tuple[str, str, str]]: """``(key, chat_type, tail)`` for every OTHER running turn in the caller's chat (``tail`` is the key text after the chat id, ``""`` when the key ends there). The namespace comes from ``own_key`` — the session store's own answer, so a named-profile stop matches that profile's runs and never a literal. ``_snapshot_running_agents`` already drops the pending sentinel (a session still being set up has no agent). Callers gate on - authorization; ``own_key`` is excluded. + authorization; ``own_key`` is excluded. Both tiers share one call. """ chat_id = str(getattr(source, "chat_id", None) or "") if not chat_id: return [] - namespace = _key_namespace(own_key) - platform = source.platform.value - scope_id = getattr(source, "scope_id", None) + if source.chat_type == "dm" and source.platform == Platform.WHATSAPP: + # Match the same text build_session_key keyed: WhatsApp DM chat ids are canonicalised + # there, so a raw JID/LID alias would never line up with the stored key. + chat_id = canonical_whatsapp_identifier(chat_id) or chat_id + namespace = ":".join(own_key.split(":", 2)[:2]) + prefix = f"{namespace}:{source.platform.value}:" + scope_id = str(getattr(source, "scope_id", None) or "") or None runs = [] for key in self._snapshot_running_agents(): if key == own_key: continue - parsed = _same_chat_key_slots( - key, namespace=namespace, platform=platform, chat_id=chat_id, scope_id=scope_id, - ) + parsed = _same_chat_key_slots(key, prefix=prefix, chat_id=chat_id, scope_id=scope_id) if parsed is not None: runs.append((key, parsed[0], parsed[1])) return runs - def _sibling_thread_run_keys(self, source: SessionSource, own_key: str) -> list: - """Running-agent keys of OTHER participants in the caller's own thread (per-user thread mode - keys are ``...:{thread_id}:{user_id}``, so another user's run is invisible to the caller's - own ``/stop``). Callers still gate on authz.""" + def _sibling_thread_run_keys( + self, source: SessionSource, runs: List[Tuple[str, str, str]], + ) -> List[str]: + """Keys from ``runs`` belonging to OTHER participants in the caller's own thread (per-user + thread mode keys are ``...:{thread_id}:{user_id}``, so another user's run is invisible to the + caller's own ``/stop``). Callers still gate on authz.""" thread_id = str(getattr(source, "thread_id", None) or "") chat_type = getattr(source, "chat_type", None) or "" if not thread_id or not chat_type: return [] return [ key - for key, key_chat_type, tail in self._same_chat_runs(source, own_key) + for key, key_chat_type, tail in runs if key_chat_type == chat_type and _tail_has_slot(tail, thread_id) ] - def _chat_scoped_run_keys(self, source: SessionSource, own_key: str) -> list: - """Running-agent keys for ANY session of the same chat, regardless of the chat_type/thread/ + def _chat_scoped_run_keys( + self, source: SessionSource, runs: List[Tuple[str, str, str]], + ) -> List[str]: + """Keys from ``runs`` for ANY session of the same chat, whatever the chat_type/thread/ participant slots. Two supported shapes make a /stop key miss a run in the same chat (found via Slack's native stop button, gateway-gateway#286): a top-level channel turn keys ``channel`` while an in-thread /stop normalizes to ``thread``, and rolling-DM configs key without the thread slot the stop carries. "/stop" means "stop what's running in THIS chat", - so the handler falls back chat-wide on an exact + thread-sibling miss — which is also what - lets a human stop a peer's per-sender group run (see ``_same_chat_runs``). + which is also what lets a human stop a peer's per-sender group run (see ``_same_chat_runs``). A stop sent from INSIDE a thread only reaches runs whose own thread slot is that thread (or - that carry no thread slot at all — a top-level channel turn's relay-stamped slot is the - reply thread it belongs to, the rolling-DM shape is slotless). Anything else in the channel - is a different conversation: another reply thread, or a peer's top-level run. Callers gate - on authz. + that carry no thread slot at all — the rolling-DM shape). Anything else in the channel is a + different conversation: another reply thread, or a peer's top-level run. Callers gate on + authz. """ thread_id = str(getattr(source, "thread_id", None) or "") return [ key - for key, _key_chat_type, tail in self._same_chat_runs(source, own_key) + for key, _key_chat_type, tail in runs if not thread_id or not tail or _tail_has_slot(tail, thread_id) ] diff --git a/gateway/slash_commands.py b/gateway/slash_commands.py index 23d02e6feb..9a7955b821 100644 --- a/gateway/slash_commands.py +++ b/gateway/slash_commands.py @@ -436,14 +436,21 @@ class GatewaySlashCommandsMixin( return EphemeralReply(t("gateway.stop.stopped")) # No run under the caller's own key: a live turn in THIS chat may still carry a differently - # shaped key. Narrowest tier first (another participant's run in the caller's own thread), - # then any run in the chat; both are authorization-gated. See `_chat_scoped_run_keys` for the - # shapes it covers and its isolation bounds. - fallback_keys = self._sibling_thread_run_keys(source, session_key) - reason = "stop_command_thread_sibling" - if not fallback_keys: - fallback_keys = self._chat_scoped_run_keys(source, session_key) - reason = "stop_command_chat_scope" + # shaped key. One scan feeds both tiers; the chat tier is a superset of the thread-sibling + # tier (a sibling needs the caller's own thread slot, which satisfies the chat predicate), so + # it is the set to act on — acting on the sibling subset alone would reply "Stopped" while a + # same-thread run under a differently shaped key kept going. See `_chat_scoped_run_keys` for + # the shapes and isolation bounds; both tiers are authorization-gated. + runs = self._same_chat_runs(source, session_key) + sibling_keys = self._sibling_thread_run_keys(source, runs) + fallback_keys = self._chat_scoped_run_keys(source, runs) + # Reason is per-stop, not per-key: a stop that only ever had thread siblings keeps its own + # label for hook consumers, anything wider is a chat-scope stop. + reason = ( + "stop_command_thread_sibling" + if fallback_keys == sibling_keys + else "stop_command_chat_scope" + ) if fallback_keys and self._is_user_authorized_for_source(source): for fallback_key in fallback_keys: await _stop(fallback_key, reason) diff --git a/tests/gateway/test_stop_chat_scope.py b/tests/gateway/test_stop_chat_scope.py index 444f50779a..83e0277e6f 100644 --- a/tests/gateway/test_stop_chat_scope.py +++ b/tests/gateway/test_stop_chat_scope.py @@ -116,35 +116,94 @@ async def test_stop_reaches_peer_run_in_per_sender_group(): @pytest.mark.asyncio async def test_stop_does_not_reach_a_different_thread_of_the_same_channel(): - # Another thread in the same channel is another conversation: exact key, thread sibling - # and chat-scope fallback must all leave it running. + # Another thread in the same channel is another conversation, while the caller's own thread + # stays reachable — proven in the same call so a matcher that returns nothing cannot pass. + same_thread = build_session_key(_slack_source("channel", "C9", thread_id="170.100")) other_thread = build_session_key(_slack_source("thread", "C9", thread_id="170.200")) stop_source = _slack_source("thread", "C9", thread_id="170.100") - interrupted, result = await _stop(stop_source, other_thread) + interrupted, result = await _stop(stop_source, [same_thread, other_thread]) - assert interrupted == [] - assert result == t("gateway.stop.no_active") + assert interrupted == [(same_thread, "stop_command_chat_scope")] + assert result == t("gateway.stop.stopped") @pytest.mark.asyncio async def test_stop_does_not_reach_another_reply_thread_of_a_channel_keyed_run(): # A top-level channel turn keeps chat_type "channel" with the relay-stamped reply-thread ts, - # so the SAME boundary must apply to it: a stop inside thread .100 must not reach the run - # whose reply thread is .200. + # so the SAME boundary must apply to it: a stop inside thread .100 interrupts the run whose + # reply thread is .100 and must not touch the one belonging to .200. + same_reply_thread = build_session_key(_slack_source("channel", "C9", thread_id="170.100")) other_reply_thread = build_session_key(_slack_source("channel", "C9", thread_id="170.200")) stop_source = _slack_source("thread", "C9", thread_id="170.100") - interrupted, result = await _stop(stop_source, other_reply_thread) + interrupted, result = await _stop(stop_source, [same_reply_thread, other_reply_thread]) - assert interrupted == [] - assert result == t("gateway.stop.no_active") + assert interrupted == [(same_reply_thread, "stop_command_chat_scope")] + assert result == t("gateway.stop.stopped") + + +@pytest.mark.asyncio +async def test_stop_interrupts_every_run_of_the_callers_thread(): + # A per-user thread sibling AND the same-thread channel-keyed run (the #286 shape this fix + # exists for) are both live: the reply must not claim "Stopped" while one of them keeps going. + sibling = build_session_key( + _slack_source("thread", "C9", thread_id="170.100", user_id="U-bob"), + thread_sessions_per_user=True, + ) + same_thread_channel_run = build_session_key(_slack_source("channel", "C9", thread_id="170.100")) + stop_source = _slack_source("thread", "C9", thread_id="170.100") + + interrupted, result = await _stop(stop_source, [sibling, same_thread_channel_run]) + + assert sorted(key for key, _ in interrupted) == sorted([sibling, same_thread_channel_run]) + assert result == t("gateway.stop.stopped") + + +@pytest.mark.asyncio +async def test_a_lone_thread_sibling_keeps_its_own_invalidation_reason(): + # With nothing else live in the chat, the stop IS a thread-sibling stop: hook consumers must + # still see that label rather than the wider chat-scope one. + sibling = build_session_key( + _slack_source("thread", "C9", thread_id="170.100", user_id="U-bob"), + thread_sessions_per_user=True, + ) + stop_source = _slack_source("thread", "C9", thread_id="170.100") + + interrupted, result = await _stop(stop_source, sibling) + + assert interrupted == [(sibling, "stop_command_thread_sibling")] + assert result == t("gateway.stop.stopped") + + +@pytest.mark.asyncio +async def test_stop_matches_a_canonicalised_whatsapp_dm_chat_id(): + # build_session_key canonicalises a WhatsApp DM chat id, so the fallback must match the same + # text: a stop whose source carries the raw JID has to reach the canonicalised run. + running_key = build_session_key( + SessionSource(platform=Platform.WHATSAPP, chat_type="dm", chat_id="1234567890@s.whatsapp.net", + user_id="1234567890@s.whatsapp.net") + ) + stop_source = SessionSource(platform=Platform.WHATSAPP, chat_type="dm", + chat_id="1234567890:7@s.whatsapp.net", + user_id="1234567890:7@s.whatsapp.net", thread_id="t1") + assert build_session_key(stop_source) != running_key + + interrupted, result = await _stop(stop_source, running_key) + + assert interrupted == [(running_key, "stop_command_chat_scope")] + assert result == t("gateway.stop.stopped") @pytest.mark.asyncio async def test_stop_in_a_dm_does_not_reach_a_group_run_that_ends_in_the_same_user_id(): # Non-Slack DMs key chat_id as the USER id (Telegram), and a per-sender group key ends with - # that same user id — the group run is a different chat and must stay untouched. + # that same user id — the group run is a different chat and must stay untouched, while the + # DM's own threaded run in the same chat is still reached. + same_chat = build_session_key( + SessionSource(platform=Platform.TELEGRAM, chat_type="dm", chat_id="777", + thread_id="42", user_id="777") + ) group_run = build_session_key( SessionSource(platform=Platform.TELEGRAM, chat_type="group", chat_id="-100123", user_id="777") @@ -152,10 +211,10 @@ async def test_stop_in_a_dm_does_not_reach_a_group_run_that_ends_in_the_same_use dm_stop = SessionSource(platform=Platform.TELEGRAM, chat_type="dm", chat_id="777", user_id="777") - interrupted, result = await _stop(dm_stop, group_run) + interrupted, result = await _stop(dm_stop, [same_chat, group_run]) - assert interrupted == [] - assert result == t("gateway.stop.no_active") + assert interrupted == [(same_chat, "stop_command_chat_scope")] + assert result == t("gateway.stop.stopped") @pytest.mark.asyncio diff --git a/tests/gateway/test_stop_thread_sibling.py b/tests/gateway/test_stop_thread_sibling.py index 234931393f..f2e2ed3649 100644 --- a/tests/gateway/test_stop_thread_sibling.py +++ b/tests/gateway/test_stop_thread_sibling.py @@ -43,7 +43,9 @@ def _per_user_key(uid, thread_id="thr1", chat_id="chan1"): def test_sibling_returns_empty_for_non_thread_source(): - # Non-thread group/channel must NOT trigger the cross-user fallback. + # The sibling TIER is thread-scoped: a non-thread group/channel source has no thread slot to + # match. The room-wide fallback for that shape is the chat tier's job — see + # tests/gateway/test_stop_chat_scope.py::test_stop_reaches_peer_run_in_per_sender_group. runner = object.__new__(GatewayRunner) nonthread = SessionSource( platform=Platform.DISCORD, chat_type="group", chat_id="chan1", user_id="userA" @@ -54,12 +56,15 @@ def test_sibling_returns_empty_for_non_thread_source(): ) ) runner._running_agents = {grp_b: _FakeAgent()} - assert runner._sibling_thread_run_keys(nonthread, "agent:main:discord:group:chan1:userA") == [] + own_key = "agent:main:discord:group:chan1:userA" + assert runner._sibling_thread_run_keys( + nonthread, runner._same_chat_runs(nonthread, own_key) + ) == [] def test_sibling_matches_named_profile_runs(): - # Under multiplexing the sibling prefix must follow the source's profile namespace, - # so /stop finds another participant's run under the SAME named profile... + # A named-profile stop must match another participant's run under the SAME profile, taken from + # the caller's own key (the session store's answer)... runner = object.__new__(GatewayRunner) source = _thread_source("userA") source.profile = "work" @@ -67,9 +72,8 @@ def test_sibling_matches_named_profile_runs(): _thread_source("userB"), thread_sessions_per_user=True, profile="work" ) runner._running_agents = {key_b: _FakeAgent()} - assert runner._sibling_thread_run_keys( - source, "agent:work:discord:forum:chan1:thr1:userA" - ) == [key_b] + own_key = "agent:work:discord:forum:chan1:thr1:userA" + assert runner._sibling_thread_run_keys(source, runner._same_chat_runs(source, own_key)) == [key_b] def test_sibling_does_not_cross_profiles(): @@ -79,12 +83,8 @@ def test_sibling_does_not_cross_profiles(): source.profile = "work" main_key = build_session_key(_thread_source("userB"), thread_sessions_per_user=True) runner._running_agents = {main_key: _FakeAgent()} - assert ( - runner._sibling_thread_run_keys( - source, "agent:work:discord:forum:chan1:thr1:userA" - ) - == [] - ) + own_key = "agent:work:discord:forum:chan1:thr1:userA" + assert runner._sibling_thread_run_keys(source, runner._same_chat_runs(source, own_key)) == [] # --------------------------------------------------------------------------- diff --git a/website/docs/user-guide/features/hooks.md b/website/docs/user-guide/features/hooks.md index d78c18ea68..f11f292b98 100644 --- a/website/docs/user-guide/features/hooks.md +++ b/website/docs/user-guide/features/hooks.md @@ -1066,7 +1066,7 @@ def my_callback(session_key: str, platform: str, reason: str, invalidation_reaso | `reason` | `str` | Why the agent was interrupted (e.g. `"user_stop"`, the reset/new reason). | | `invalidation_reason` | `str` | Why queued session state was invalidated (e.g. `"stop_command"`, `"stop_command_thread_sibling"`, `"stop_command_chat_scope"`, `"reset_command"`). | -**Fires:** In `gateway/run.py::_interrupt_and_clear_session`, immediately after `request_hard_interrupt()` interrupts the running agent. Only when a real agent was running — the pending-sentinel `/stop` path (no agent loop yet started) does **not** fire this hook, since there is no in-flight work to drop. On the slow `/new` reset path, `on_session_finalize` fires later in `_handle_reset_command` instead. +**Fires:** In `gateway/run_agent_cache.py::_interrupt_and_clear_session`, immediately after `request_hard_interrupt()` interrupts the running agent. Only when a real agent was running — the pending-sentinel `/stop` path (no agent loop yet started) does **not** fire this hook, since there is no in-flight work to drop. On the slow `/new` reset path, `on_session_finalize` fires later in `_handle_reset_command` instead. **Return value:** Ignored. diff --git a/website/docs/user-guide/sessions.md b/website/docs/user-guide/sessions.md index d5f2c6228d..f64e552966 100644 --- a/website/docs/user-guide/sessions.md +++ b/website/docs/user-guide/sessions.md @@ -796,15 +796,15 @@ By default, Hermes uses `group_sessions_per_user: true` in `config.yaml`. That m - Alice and Bob can both talk to Hermes in the same Discord channel without sharing transcript history - one user's long tool-heavy task does not pollute another user's context window -- a running turn is interrupted under its own key, so `/stop` in the room finds a turn that - started under a different sender's key (a peer's or a bot's message) — see below +- a running turn is keyed to the sender that started it, but `/stop` still reaches it — see below `/stop` means "stop what is running in this chat": it first tries the caller's own session key, -then other participants' runs in the caller's own thread, then any live turn in the same room — +then any live turn in this chat — other participants' runs in the caller's own thread included — authorization-gated, and never another room, workspace or profile. So an idle Alice's `/stop` can end a turn Bob (or a bot) started in the room she is in. A `/stop` sent from *inside* a thread is -narrower: it only reaches runs belonging to that thread (or to no thread at all), never a turn in -another thread of the same channel. +narrower: it reaches runs belonging to that thread and a room-wide run that carries no thread slot +(the rolling-DM shape), but never another thread of the same channel and never a peer's per-sender +top-level run. If you want one shared "room brain" instead, set: