hermes-admin/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 28 Packages Projects Releases Wiki Activity

fix(gateway): a /stop in a thread stops every run of that thread

Browse Source 
The thread-sibling tier short-circuited the chat-scope tier, so when a per-user
thread sibling was live the handler interrupted ONLY that run and replied
"Stopped" while a same-thread run under a differently shaped key — the
channel-keyed #286 shape this fallback exists for — kept going. The chat tier is
a superset of the sibling tier (a sibling needs the caller's own thread slot,
which satisfies the chat predicate), so it is the set acted on; the reason still
resolves to `stop_command_thread_sibling` when the chat tier found nothing beyond
siblings, so hook consumers keep that label.

Both tiers also share ONE scan now (they each called `_same_chat_runs`, building
the running-agent snapshot twice), and the matcher is a single text pass:
`_same_chat_key_slots` takes the caller's key prefix instead of re-deriving the
namespace per key, and the whole-slot rule lives once in `_strip_slot` (the tail
check uses it too). Concrete annotations on the three methods (`List[...]` was
missing from the typing import), and the prefix construction reads as a named
namespace rather than a nested f-string.

WhatsApp DMs: `build_session_key` canonicalises the DM chat id, so the fallback
matched nothing there — it now canonicalises the same way, which makes the
chat-scope fallback reach a run keyed from any JID/LID alias.

Docs: `sessions.md` described the tiers sequentially and over-claimed the
widening's reach — an in-thread stop reaches that thread plus a thread-less
room-wide run, never another thread or a peer's per-sender top-level run.
`hooks.md` pointed at `gateway/run.py` for `_interrupt_and_clear_session`, which
lives in `gateway/run_agent_cache.py`.

Guards: the partial-stop case (per-user sibling + same-thread channel run, both
must be interrupted), the lone-sibling reason label, and the WhatsApp canonical
id. Each is mutation-checked individually: restoring the short-circuit, forcing
the reason to chat_scope, or dropping the canonicalisation each fails exactly
its own test. 21 passed in the two /stop test files.
This commit is contained in:
kshitijk4poor
2026-09-18 14:28:54 +05:30
committed by kshitij
parent 045eb44363
commit 0fb56906fc
6 changed files with 162 additions and 87 deletions
Download Patch File Download Diff File Expand all files Collapse all files

101
gateway/run_busy.py
View File

@@ -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:<profile>`` 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:<profile>:<platform>:<chat_type>[:<scope_id>][:<chat_id>][:<thread_id>][:<user>]``.
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:<profile>:<platform>:``. 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)
]

23
gateway/slash_commands.py
View File

@@ -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)

87
tests/gateway/test_stop_chat_scope.py
View File

@@ -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

26
tests/gateway/test_stop_thread_sibling.py
View File

@@ -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)) == []
# ---------------------------------------------------------------------------

2
website/docs/user-guide/features/hooks.md
View File

@@ -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.

10
website/docs/user-guide/sessions.md
View File

@@ -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: