From 1c741b3bfbee59794c247f1854ef92608eda8157 Mon Sep 17 00:00:00 2001 From: serxa Date: Thu, 27 Aug 2026 16:57:21 +0000 Subject: [PATCH 1/5] Telegram: route a reply to the session that sent the replied-to message MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replying (Telegram reply) to a message the bot sent now routes the reply to the session that produced that message — and makes it the active session — instead of the chat's active session. This makes it easy to answer messages from cron/background sessions and to talk to several sessions without an explicit /session switch. The channel records message_id -> (chat_id, session_id) for every outbound message (interactive send() and notification delivery) in a bounded LRU, and resolves it on inbound replies. It falls back to the active session when the mapping is unknown or the target session is gone/archived, and requires the chat id to match so a reply can never cross into a session bound to a different chat. Co-Authored-By: Claude Opus 4.8 --- nerve/channels/telegram.py | 85 +++++++++++++++ nerve/notifications/service.py | 5 +- tests/test_telegram_reply_routing.py | 152 +++++++++++++++++++++++++++ 3 files changed, 241 insertions(+), 1 deletion(-) create mode 100644 tests/test_telegram_reply_routing.py diff --git a/nerve/channels/telegram.py b/nerve/channels/telegram.py index a0c970db..7a71acce 100644 --- a/nerve/channels/telegram.py +++ b/nerve/channels/telegram.py @@ -483,6 +483,14 @@ def __init__( collections.OrderedDict() ) self._message_cache_max = 200 + # Reply routing: telegram message_id -> (chat_id, session_id) for + # messages the bot has sent, so a user REPLY targets the session that + # produced the replied-to message (e.g. a cron/background session) + # instead of the chat's active session. Bounded LRU. + self._reply_routes: collections.OrderedDict[int, tuple[int, str]] = ( + collections.OrderedDict() + ) + self._reply_routes_max = 2000 # Rate limiter for replies to unauthorized users: user_id -> monotonic ts self._unauth_reply_times: dict[int, float] = {} @@ -825,6 +833,7 @@ async def send(self, message: OutboundMessage) -> None: text=chunk, ) self._cache_message(sent.message_id, chat_id, chunk) + self._record_reply_route(sent.message_id, chat_id, message.session_id) def format_response(self, text: str) -> str: """Return text unchanged. @@ -966,6 +975,64 @@ def _cache_message(self, message_id: int, chat_id: int, text: str) -> None: while len(self._message_cache) > self._message_cache_max: self._message_cache.popitem(last=False) + # ------------------------------------------------------------------ # + # Reply routing (reply → originating session) # + # ------------------------------------------------------------------ # + + def _record_reply_route( + self, message_id: int, chat_id: int, session_id: str, + ) -> None: + """Remember which session produced an outbound Telegram message. + + Lets a later user REPLY to that message route back to the + originating session (see ``_resolve_reply_session``). No-op when the + session is unknown. Bounded LRU keyed by ``message_id``. + """ + if not session_id: + return + self._reply_routes[message_id] = (chat_id, session_id) + self._reply_routes.move_to_end(message_id) + while len(self._reply_routes) > self._reply_routes_max: + self._reply_routes.popitem(last=False) + + def _lookup_reply_route( + self, message_id: int, chat_id: int, + ) -> str | None: + """Session that produced ``message_id`` in ``chat_id``, or None. + + The chat must match, so a ``message_id`` from one chat can never + route a reply into a session bound to a different chat. + """ + entry = self._reply_routes.get(message_id) + if not entry: + return None + cached_chat_id, session_id = entry + if cached_chat_id != chat_id: + return None + return session_id + + async def _resolve_reply_session( + self, reply_message_id: int | None, chat_id: int, + ) -> str | None: + """Target session for a reply, or None to use the active session. + + Returns a session id only when the user replied to a message this + bot produced and that session still exists and is not archived; + otherwise None so the caller falls back to the chat's active session. + """ + if reply_message_id is None: + return None + session_id = self._lookup_reply_route(reply_message_id, chat_id) + if not session_id: + return None + try: + row = await self.router.get_session(session_id) + except Exception: + return None + if not row or row.get("status") == "archived": + return None + return session_id + # ------------------------------------------------------------------ # # Auth # # ------------------------------------------------------------------ # @@ -1624,11 +1691,20 @@ async def _handle_message(self, update: Update, context: Any) -> None: if images: metadata["images"] = images + # A reply to a message the bot sent routes to the session that + # produced it (and makes it the active session); None → active session. + routed_session = await self._resolve_reply_session( + reply_msg.message_id if reply_msg else None, chat_id, + ) + if routed_session: + logger.info("Telegram reply routed to session %s", routed_session) + msg = InboundMessage( channel_name="telegram", channel_key=f"telegram:{chat_id}", sender_id=str(chat_id), text=text, + session_id=routed_session, metadata=metadata, ) @@ -1790,11 +1866,20 @@ async def _process_media_group(self, group_id: str) -> None: if images: metadata["images"] = images + # A reply to a message the bot sent routes to the session that + # produced it (and makes it the active session); None → active session. + routed_session = await self._resolve_reply_session( + reply_msg.message_id if reply_msg else None, chat_id, + ) + if routed_session: + logger.info("Telegram reply routed to session %s", routed_session) + msg = InboundMessage( channel_name="telegram", channel_key=f"telegram:{chat_id}", sender_id=str(chat_id), text=text, + session_id=routed_session, metadata=metadata, ) diff --git a/nerve/notifications/service.py b/nerve/notifications/service.py index 0cd9ee0d..eb39b489 100644 --- a/nerve/notifications/service.py +++ b/nerve/notifications/service.py @@ -969,11 +969,14 @@ async def _deliver_telegram( msg = await self._send_telegram_html(bot, chat_id, text, silent=silent) msg_id = str(msg.message_id) - # Cache for reaction context lookups + # Cache for reaction context lookups, and record the reply route so a + # user replying to this notification reaches the session that raised + # it (e.g. a cron/background session). if msg_id: channel = self._get_telegram_channel() if channel: channel._cache_message(int(msg_id), chat_id, text) + channel._record_reply_route(int(msg_id), chat_id, session_id) return msg_id diff --git a/tests/test_telegram_reply_routing.py b/tests/test_telegram_reply_routing.py new file mode 100644 index 00000000..2df8d7d5 --- /dev/null +++ b/tests/test_telegram_reply_routing.py @@ -0,0 +1,152 @@ +"""Tests for reply → originating-session routing (nerve.channels.telegram). + +When a user replies (Telegram reply) to a message the bot sent, the reply is +routed to the session that produced that message instead of the chat's active +session. The channel records ``message_id -> (chat_id, session_id)`` on every +outbound send and resolves it back on inbound replies. +""" + +from types import SimpleNamespace + +import pytest + +from nerve.channels.base import OutboundMessage +from nerve.channels.telegram import TelegramChannel + + +class _FakeRouter: + """Minimal router exposing the one method the channel calls here.""" + + def __init__(self, sessions: dict | None = None): + self._sessions = sessions or {} + + async def get_session(self, session_id: str): + return self._sessions.get(session_id) + + +def _make_channel(router: _FakeRouter | None = None) -> TelegramChannel: + cfg = SimpleNamespace(telegram=SimpleNamespace(allowed_users=[])) + return TelegramChannel(lambda: cfg, router or _FakeRouter()) + + +# --------------------------------------------------------------------------- # +# Record / lookup # +# --------------------------------------------------------------------------- # + +def test_record_and_lookup_hit(): + ch = _make_channel() + ch._record_reply_route(100, 42, "sessA") + assert ch._lookup_reply_route(100, 42) == "sessA" + + +def test_lookup_unknown_message_is_none(): + ch = _make_channel() + assert ch._lookup_reply_route(999, 42) is None + + +def test_lookup_requires_matching_chat(): + # A message_id recorded for chat 42 must never route a reply seen in + # another chat (message ids are only unique per chat). + ch = _make_channel() + ch._record_reply_route(100, 42, "sessA") + assert ch._lookup_reply_route(100, 43) is None + + +def test_empty_session_is_not_recorded(): + ch = _make_channel() + ch._record_reply_route(100, 42, "") + assert ch._lookup_reply_route(100, 42) is None + + +def test_reply_routes_are_lru_bounded(): + ch = _make_channel() + ch._reply_routes_max = 3 + for mid in range(1, 6): # 1..5, cap 3 → 1 and 2 evicted + ch._record_reply_route(mid, 42, f"s{mid}") + assert ch._lookup_reply_route(1, 42) is None + assert ch._lookup_reply_route(2, 42) is None + assert ch._lookup_reply_route(3, 42) == "s3" + assert ch._lookup_reply_route(5, 42) == "s5" + assert len(ch._reply_routes) == 3 + + +def test_re_record_refreshes_lru_recency(): + ch = _make_channel() + ch._reply_routes_max = 2 + ch._record_reply_route(1, 42, "s1") + ch._record_reply_route(2, 42, "s2") + ch._record_reply_route(1, 42, "s1") # touch 1 → 2 is now oldest + ch._record_reply_route(3, 42, "s3") # evicts 2 + assert ch._lookup_reply_route(2, 42) is None + assert ch._lookup_reply_route(1, 42) == "s1" + assert ch._lookup_reply_route(3, 42) == "s3" + + +# --------------------------------------------------------------------------- # +# Resolve (with the live-session guard) # +# --------------------------------------------------------------------------- # + +@pytest.mark.asyncio +async def test_resolve_routes_to_live_session(): + ch = _make_channel(_FakeRouter({"sessA": {"status": "active"}})) + ch._record_reply_route(100, 42, "sessA") + assert await ch._resolve_reply_session(100, 42) == "sessA" + + +@pytest.mark.asyncio +async def test_resolve_none_when_not_a_reply(): + ch = _make_channel(_FakeRouter({"sessA": {"status": "active"}})) + ch._record_reply_route(100, 42, "sessA") + assert await ch._resolve_reply_session(None, 42) is None + + +@pytest.mark.asyncio +async def test_resolve_falls_back_when_session_gone(): + # Mapping exists but the session no longer exists → fall back (None). + ch = _make_channel(_FakeRouter({})) + ch._record_reply_route(100, 42, "sessA") + assert await ch._resolve_reply_session(100, 42) is None + + +@pytest.mark.asyncio +async def test_resolve_falls_back_when_session_archived(): + ch = _make_channel(_FakeRouter({"sessA": {"status": "archived"}})) + ch._record_reply_route(100, 42, "sessA") + assert await ch._resolve_reply_session(100, 42) is None + + +@pytest.mark.asyncio +async def test_resolve_none_for_unmapped_reply(): + ch = _make_channel(_FakeRouter({"sessA": {"status": "active"}})) + # Replying to a message the bot never recorded (e.g. one the user sent). + assert await ch._resolve_reply_session(555, 42) is None + + +# --------------------------------------------------------------------------- # +# Outbound send() records the route # +# --------------------------------------------------------------------------- # + +@pytest.mark.asyncio +async def test_send_records_reply_route(): + ch = _make_channel() + + class _Bot: + async def send_message(self, **kwargs): + return SimpleNamespace(message_id=777) + + ch._app = SimpleNamespace(bot=_Bot()) + await ch.send(OutboundMessage(target="42", text="hello", session_id="sX")) + assert ch._lookup_reply_route(777, 42) == "sX" + + +@pytest.mark.asyncio +async def test_send_without_session_id_records_nothing(): + ch = _make_channel() + + class _Bot: + async def send_message(self, **kwargs): + return SimpleNamespace(message_id=778) + + ch._app = SimpleNamespace(bot=_Bot()) + await ch.send(OutboundMessage(target="42", text="hello")) # session_id="" + assert ch._lookup_reply_route(778, 42) is None From d0bf1f5046b8c0c7361d3117601cb45b8c416059 Mon Sep 17 00:00:00 2001 From: serxa Date: Thu, 27 Aug 2026 19:05:09 +0000 Subject: [PATCH 2/5] Telegram reply routing: reject unresolvable replies instead of mis-routing MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A reply we can't tie to a live session — the replied-to message was never recorded / was evicted from the LRU, or its session is gone/archived — is now rejected with a static, no-LLM message asking the user to pick a session with /sessions and resend, rather than silently delivered to the chat's active session (which would be a mis-route to the wrong session). _resolve_reply() is now three-way (active / route / reject); the inbound handlers send the nudge and drop a rejected reply without starting a turn. Co-Authored-By: Claude Opus 4.8 --- nerve/channels/telegram.py | 74 +++++++++++++++++++++------- tests/test_telegram_reply_routing.py | 33 ++++++++----- 2 files changed, 76 insertions(+), 31 deletions(-) diff --git a/nerve/channels/telegram.py b/nerve/channels/telegram.py index 7a71acce..93c19034 100644 --- a/nerve/channels/telegram.py +++ b/nerve/channels/telegram.py @@ -44,6 +44,14 @@ # Telegram message length limit MAX_MSG_LEN = 4096 +# Shown (no LLM) when a user replies to a message we can't tie to a live session, +# rather than silently mis-routing the reply into the chat's active session. +REPLY_UNRESOLVED_MSG = ( + "⚠️ I couldn't find the session that produced the message you replied to " + "(it may be too old, or its session was archived). Nothing was sent, to " + "avoid routing it to the wrong session. Use /sessions to pick the session " + "you want, then send your message again." +) # Minimum interval between message edits (seconds) to avoid rate limits EDIT_INTERVAL = 1.5 # Watchdog: check every 30s, log heartbeat every ~5 min @@ -1011,27 +1019,35 @@ def _lookup_reply_route( return None return session_id - async def _resolve_reply_session( + async def _resolve_reply( self, reply_message_id: int | None, chat_id: int, - ) -> str | None: - """Target session for a reply, or None to use the active session. - - Returns a session id only when the user replied to a message this - bot produced and that session still exists and is not archived; - otherwise None so the caller falls back to the chat's active session. + ) -> tuple[str, str | None]: + """Decide how to route an inbound message given its reply target. + + Returns ``(action, session_id)``: + + - ``("active", None)`` — not a reply; use the chat's active session + (the default; plain messages are unchanged). + - ``("route", sid)`` — a reply to a message this bot produced whose + session is still live; route there and make it active. + - ``("reject", None)`` — a reply we cannot tie to a live session (the + message was never recorded / evicted from the LRU, or its session is + gone/archived). The caller nudges the user and drops the message + WITHOUT starting a turn — an explicit reply is never silently + delivered to the active session, which would be a mis-route. """ if reply_message_id is None: - return None + return ("active", None) session_id = self._lookup_reply_route(reply_message_id, chat_id) if not session_id: - return None + return ("reject", None) try: row = await self.router.get_session(session_id) except Exception: - return None + row = None if not row or row.get("status") == "archived": - return None - return session_id + return ("reject", None) + return ("route", session_id) # ------------------------------------------------------------------ # # Auth # @@ -1691,11 +1707,22 @@ async def _handle_message(self, update: Update, context: Any) -> None: if images: metadata["images"] = images - # A reply to a message the bot sent routes to the session that - # produced it (and makes it the active session); None → active session. - routed_session = await self._resolve_reply_session( + # Route a reply to the session that produced the replied-to message + # (and make it active). An unresolvable reply is rejected with a no-LLM + # nudge rather than mis-routed into the chat's active session. + action, routed_session = await self._resolve_reply( reply_msg.message_id if reply_msg else None, chat_id, ) + if action == "reject": + logger.info( + "Telegram reply from chat %s not tied to a live session; nudging", + chat_id, + ) + try: + await update.message.reply_text(REPLY_UNRESOLVED_MSG) + except Exception: + logger.error("Failed to send reply-unresolved nudge to chat %s", chat_id) + return if routed_session: logger.info("Telegram reply routed to session %s", routed_session) @@ -1866,11 +1893,22 @@ async def _process_media_group(self, group_id: str) -> None: if images: metadata["images"] = images - # A reply to a message the bot sent routes to the session that - # produced it (and makes it the active session); None → active session. - routed_session = await self._resolve_reply_session( + # Route a reply to the session that produced the replied-to message + # (and make it active). An unresolvable reply is rejected with a no-LLM + # nudge rather than mis-routed into the chat's active session. + action, routed_session = await self._resolve_reply( reply_msg.message_id if reply_msg else None, chat_id, ) + if action == "reject": + logger.info( + "Telegram reply from chat %s not tied to a live session; nudging", + chat_id, + ) + try: + await updates[0].message.reply_text(REPLY_UNRESOLVED_MSG) + except Exception: + logger.error("Failed to send reply-unresolved nudge to chat %s", chat_id) + return if routed_session: logger.info("Telegram reply routed to session %s", routed_session) diff --git a/tests/test_telegram_reply_routing.py b/tests/test_telegram_reply_routing.py index 2df8d7d5..612f7338 100644 --- a/tests/test_telegram_reply_routing.py +++ b/tests/test_telegram_reply_routing.py @@ -83,43 +83,50 @@ def test_re_record_refreshes_lru_recency(): # --------------------------------------------------------------------------- # -# Resolve (with the live-session guard) # +# Resolve — three-way: active / route / reject # # --------------------------------------------------------------------------- # @pytest.mark.asyncio -async def test_resolve_routes_to_live_session(): +async def test_resolve_active_when_not_a_reply(): ch = _make_channel(_FakeRouter({"sessA": {"status": "active"}})) ch._record_reply_route(100, 42, "sessA") - assert await ch._resolve_reply_session(100, 42) == "sessA" + assert await ch._resolve_reply(None, 42) == ("active", None) @pytest.mark.asyncio -async def test_resolve_none_when_not_a_reply(): +async def test_resolve_routes_to_live_session(): ch = _make_channel(_FakeRouter({"sessA": {"status": "active"}})) ch._record_reply_route(100, 42, "sessA") - assert await ch._resolve_reply_session(None, 42) is None + assert await ch._resolve_reply(100, 42) == ("route", "sessA") @pytest.mark.asyncio -async def test_resolve_falls_back_when_session_gone(): - # Mapping exists but the session no longer exists → fall back (None). +async def test_resolve_rejects_when_session_gone(): + # Mapping exists but the session no longer exists → reject, never mis-route. ch = _make_channel(_FakeRouter({})) ch._record_reply_route(100, 42, "sessA") - assert await ch._resolve_reply_session(100, 42) is None + assert await ch._resolve_reply(100, 42) == ("reject", None) @pytest.mark.asyncio -async def test_resolve_falls_back_when_session_archived(): +async def test_resolve_rejects_when_session_archived(): ch = _make_channel(_FakeRouter({"sessA": {"status": "archived"}})) ch._record_reply_route(100, 42, "sessA") - assert await ch._resolve_reply_session(100, 42) is None + assert await ch._resolve_reply(100, 42) == ("reject", None) + + +@pytest.mark.asyncio +async def test_resolve_rejects_unmapped_reply(): + ch = _make_channel(_FakeRouter({"sessA": {"status": "active"}})) + # Replying to a message the bot never recorded (user's own / LRU-evicted). + assert await ch._resolve_reply(555, 42) == ("reject", None) @pytest.mark.asyncio -async def test_resolve_none_for_unmapped_reply(): +async def test_resolve_rejects_on_chat_mismatch(): ch = _make_channel(_FakeRouter({"sessA": {"status": "active"}})) - # Replying to a message the bot never recorded (e.g. one the user sent). - assert await ch._resolve_reply_session(555, 42) is None + ch._record_reply_route(100, 42, "sessA") # recorded for chat 42 + assert await ch._resolve_reply(100, 43) == ("reject", None) # replied in chat 43 # --------------------------------------------------------------------------- # From b3091f189973d33cd06e4bd80c34f73a162f4c49 Mon Sep 17 00:00:00 2001 From: serxa Date: Fri, 28 Aug 2026 06:09:20 +0000 Subject: [PATCH 3/5] Telegram reply routing: also record inbound messages' landing session MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Record each inbound message against the session it is delivered to (the resolved reply target, else the active session), not just outbound bot messages. A user can then reply to their OWN earlier message to route the reply to that message's session — the same first-class targeting as replying to a bot message, and a clean way to re-target without /session. Adds _delivery_session(); both inbound handlers record the incoming message id -> target session before dispatch, and dispatch with that session explicitly. Co-Authored-By: Claude Opus 4.8 --- nerve/channels/telegram.py | 44 +++++++++++++++++++++++----- tests/test_telegram_reply_routing.py | 37 +++++++++++++++++++++-- 2 files changed, 70 insertions(+), 11 deletions(-) diff --git a/nerve/channels/telegram.py b/nerve/channels/telegram.py index 93c19034..908e911d 100644 --- a/nerve/channels/telegram.py +++ b/nerve/channels/telegram.py @@ -491,10 +491,11 @@ def __init__( collections.OrderedDict() ) self._message_cache_max = 200 - # Reply routing: telegram message_id -> (chat_id, session_id) for - # messages the bot has sent, so a user REPLY targets the session that - # produced the replied-to message (e.g. a cron/background session) - # instead of the chat's active session. Bounded LRU. + # Reply routing: telegram message_id -> (chat_id, session_id). Records + # both messages the bot sent and the session each inbound message landed + # in, so a user REPLY targets that session — a cron/background session's + # message, or the user's own earlier message — instead of the chat's + # active session. Bounded LRU. self._reply_routes: collections.OrderedDict[int, tuple[int, str]] = ( collections.OrderedDict() ) @@ -1049,6 +1050,19 @@ async def _resolve_reply( return ("reject", None) return ("route", session_id) + async def _delivery_session( + self, routed_session: str | None, channel_key: str, + ) -> str: + """The session an inbound message will actually be delivered to: + the explicit reply target if one resolved, else the chat's active + session. Recorded against the incoming message id so a later reply to + it — including the user replying to their OWN message — routes back to + the same session. + """ + if routed_session: + return routed_session + return await self.router.get_active_session(channel_key, source="telegram") + # ------------------------------------------------------------------ # # Auth # # ------------------------------------------------------------------ # @@ -1726,12 +1740,19 @@ async def _handle_message(self, update: Update, context: Any) -> None: if routed_session: logger.info("Telegram reply routed to session %s", routed_session) + channel_key = f"telegram:{chat_id}" + target_session = await self._delivery_session(routed_session, channel_key) + # Record this inbound message against the session it lands in, so a later + # reply to it — including the user replying to their own message — routes + # back to the same session. + self._record_reply_route(update.message.message_id, chat_id, target_session) + msg = InboundMessage( channel_name="telegram", - channel_key=f"telegram:{chat_id}", + channel_key=channel_key, sender_id=str(chat_id), text=text, - session_id=routed_session, + session_id=target_session, metadata=metadata, ) @@ -1912,12 +1933,19 @@ async def _process_media_group(self, group_id: str) -> None: if routed_session: logger.info("Telegram reply routed to session %s", routed_session) + channel_key = f"telegram:{chat_id}" + target_session = await self._delivery_session(routed_session, channel_key) + # Record this inbound message against the session it lands in, so a later + # reply to it — including the user replying to their own message — routes + # back to the same session. + self._record_reply_route(updates[0].message.message_id, chat_id, target_session) + msg = InboundMessage( channel_name="telegram", - channel_key=f"telegram:{chat_id}", + channel_key=channel_key, sender_id=str(chat_id), text=text, - session_id=routed_session, + session_id=target_session, metadata=metadata, ) diff --git a/tests/test_telegram_reply_routing.py b/tests/test_telegram_reply_routing.py index 612f7338..7cdd3b5b 100644 --- a/tests/test_telegram_reply_routing.py +++ b/tests/test_telegram_reply_routing.py @@ -3,7 +3,8 @@ When a user replies (Telegram reply) to a message the bot sent, the reply is routed to the session that produced that message instead of the chat's active session. The channel records ``message_id -> (chat_id, session_id)`` on every -outbound send and resolves it back on inbound replies. +outbound send and for each inbound message's landing session, then resolves it +back on inbound replies — including the user replying to their own message. """ from types import SimpleNamespace @@ -15,14 +16,18 @@ class _FakeRouter: - """Minimal router exposing the one method the channel calls here.""" + """Minimal router exposing the methods the channel calls here.""" - def __init__(self, sessions: dict | None = None): + def __init__(self, sessions: dict | None = None, active: str | None = None): self._sessions = sessions or {} + self._active = active async def get_session(self, session_id: str): return self._sessions.get(session_id) + async def get_active_session(self, channel_key: str, source: str): + return self._active + def _make_channel(router: _FakeRouter | None = None) -> TelegramChannel: cfg = SimpleNamespace(telegram=SimpleNamespace(allowed_users=[])) @@ -157,3 +162,29 @@ async def send_message(self, **kwargs): ch._app = SimpleNamespace(bot=_Bot()) await ch.send(OutboundMessage(target="42", text="hello")) # session_id="" assert ch._lookup_reply_route(778, 42) is None + + +# --------------------------------------------------------------------------- # +# Delivery session + recording the user's own messages # +# --------------------------------------------------------------------------- # + +@pytest.mark.asyncio +async def test_delivery_session_uses_routed_target(): + ch = _make_channel(_FakeRouter(active="active1")) + assert await ch._delivery_session("sessB", "telegram:42") == "sessB" + + +@pytest.mark.asyncio +async def test_delivery_session_falls_back_to_active(): + ch = _make_channel(_FakeRouter(active="active1")) + assert await ch._delivery_session(None, "telegram:42") == "active1" + + +@pytest.mark.asyncio +async def test_reply_to_own_recorded_message_routes_to_where_it_landed(): + # Recording the user's own inbound message against its landing session means + # a later reply to that same message routes back there (serxa's request). + ch = _make_channel(_FakeRouter({"active1": {"status": "active"}}, active="active1")) + landing = await ch._delivery_session(None, "telegram:42") # → active1 + ch._record_reply_route(500, 42, landing) # user's own msg id 500 + assert await ch._resolve_reply(500, 42) == ("route", "active1") From cccaac4bd2a459e62238ca78875ef18e2d87ec24 Mon Sep 17 00:00:00 2001 From: serxa Date: Tue, 1 Sep 2026 13:44:01 +0000 Subject: [PATCH 4/5] Telegram reply routing: key the map by (chat_id, message_id) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Telegram message ids are unique only within a chat, so keying the reply-route map on message_id alone let one chat's message clobber another chat's identically-numbered one — the clobbered chat's later reply would then miss and be wrongly rejected. Key on (chat_id, message_id) instead. Addresses review feedback. Co-Authored-By: Claude Opus 4.8 --- nerve/channels/telegram.py | 32 +++++++++++++--------------- tests/test_telegram_reply_routing.py | 10 +++++++++ 2 files changed, 25 insertions(+), 17 deletions(-) diff --git a/nerve/channels/telegram.py b/nerve/channels/telegram.py index 908e911d..33bca77e 100644 --- a/nerve/channels/telegram.py +++ b/nerve/channels/telegram.py @@ -491,12 +491,13 @@ def __init__( collections.OrderedDict() ) self._message_cache_max = 200 - # Reply routing: telegram message_id -> (chat_id, session_id). Records - # both messages the bot sent and the session each inbound message landed - # in, so a user REPLY targets that session — a cron/background session's + # Reply routing: (chat_id, message_id) -> session_id. Records both + # messages the bot sent and the session each inbound message landed in, + # so a user REPLY targets that session — a cron/background session's # message, or the user's own earlier message — instead of the chat's - # active session. Bounded LRU. - self._reply_routes: collections.OrderedDict[int, tuple[int, str]] = ( + # active session. Keyed by (chat_id, message_id) because Telegram + # message ids are unique only within a chat, not across chats. Bounded LRU. + self._reply_routes: collections.OrderedDict[tuple[int, int], str] = ( collections.OrderedDict() ) self._reply_routes_max = 2000 @@ -994,13 +995,16 @@ def _record_reply_route( """Remember which session produced an outbound Telegram message. Lets a later user REPLY to that message route back to the - originating session (see ``_resolve_reply_session``). No-op when the - session is unknown. Bounded LRU keyed by ``message_id``. + originating session (see ``_resolve_reply``). No-op when the session + is unknown. Bounded LRU keyed by ``(chat_id, message_id)`` — message + ids repeat across chats, so keying on the id alone would let one chat + clobber another chat's identically-numbered message. """ if not session_id: return - self._reply_routes[message_id] = (chat_id, session_id) - self._reply_routes.move_to_end(message_id) + key = (chat_id, message_id) + self._reply_routes[key] = session_id + self._reply_routes.move_to_end(key) while len(self._reply_routes) > self._reply_routes_max: self._reply_routes.popitem(last=False) @@ -1009,16 +1013,10 @@ def _lookup_reply_route( ) -> str | None: """Session that produced ``message_id`` in ``chat_id``, or None. - The chat must match, so a ``message_id`` from one chat can never + The key is per-chat, so a ``message_id`` from one chat can never route a reply into a session bound to a different chat. """ - entry = self._reply_routes.get(message_id) - if not entry: - return None - cached_chat_id, session_id = entry - if cached_chat_id != chat_id: - return None - return session_id + return self._reply_routes.get((chat_id, message_id)) async def _resolve_reply( self, reply_message_id: int | None, chat_id: int, diff --git a/tests/test_telegram_reply_routing.py b/tests/test_telegram_reply_routing.py index 7cdd3b5b..5a2b297a 100644 --- a/tests/test_telegram_reply_routing.py +++ b/tests/test_telegram_reply_routing.py @@ -63,6 +63,16 @@ def test_empty_session_is_not_recorded(): assert ch._lookup_reply_route(100, 42) is None +def test_same_message_id_in_two_chats_do_not_collide(): + # Telegram message ids repeat across chats; keying on the id alone would + # let chat B's message clobber chat A's identically-numbered one. + ch = _make_channel() + ch._record_reply_route(100, 42, "sessA") # chat 42, msg 100 + ch._record_reply_route(100, 77, "sessB") # chat 77, SAME msg id 100 + assert ch._lookup_reply_route(100, 42) == "sessA" + assert ch._lookup_reply_route(100, 77) == "sessB" + + def test_reply_routes_are_lru_bounded(): ch = _make_channel() ch._reply_routes_max = 3 From 18227b0fe59ea4cdf5c4b58ebd3db9fdc13a5902 Mon Sep 17 00:00:00 2001 From: serxa Date: Tue, 1 Sep 2026 15:09:45 +0000 Subject: [PATCH 5/5] Telegram reply routing: gate behind a setting, default off MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Put the new reply→origin-session routing behind an opt-in setting telegram.reply_routes_to_origin_session (default false), so the existing context-only reply behavior is unchanged unless enabled — addresses review feedback that the current behavior is intended. When off, no record / resolve / reject runs and a reply lands in the active session like a plain message. The flag is read live per message (no restart to toggle). Adds the setting to TelegramConfig + docs (settings + hot-reload tables); gates the recorder and both inbound handlers; adds on/off gate tests. Co-Authored-By: Claude Opus 4.8 --- docs/config.md | 3 +- nerve/channels/telegram.py | 125 +++++++++++++++------------ nerve/config.py | 9 ++ tests/test_telegram_reply_routing.py | 26 +++++- 4 files changed, 106 insertions(+), 57 deletions(-) diff --git a/docs/config.md b/docs/config.md index af122d04..fcd1287a 100644 --- a/docs/config.md +++ b/docs/config.md @@ -244,7 +244,7 @@ A reload is always explicit. Two things cause one: | `retention.*`, `backup.*`, and the `sessions.*` the background loops read | ✅ from the next cycle of that loop | | `external_agents.targets` (including each target's `enabled`), `.sync_interval_minutes`, `.conflict_policy` | ✅ from the next sweep, provided at least one target existed at startup (see the restart table) | | `sessions.sticky_period_minutes` | ✅ | -| `telegram.dm_policy`, `.stream_mode` | ✅ read per update. Tightening `open` to `pairing` takes effect on the next message; `allowed_users` does not follow it (see the restart table) | +| `telegram.dm_policy`, `.stream_mode`, `.reply_routes_to_origin_session` | ✅ read per update. Tightening `open` to `pairing` takes effect on the next message; toggling `reply_routes_to_origin_session` takes effect on the next message; `allowed_users` does not follow it (see the restart table) | | `workflows.*` and `workflows.review_loop.*` — budget caps, concurrency, the warning fraction, iteration and criteria caps, leg engines/models, the verifier sandbox | ✅ read per use, by loops and runs already in flight as well as new ones. The two `enabled` flags and the two loop cadences are the exceptions; see the restart table | | `provider.*` and the API keys it selects (`aws_region`, `aws_profile`, `aws_access_key_id`, and the effective Anthropic key) | ✅ for sessions started **after** the reload. Each client's environment is built from the live reference when the session is created, by the same seam as `agent.*` below | | **`agent.*` and `codex.*`**: backend choice and models (`agent.backend`, `agent.cron_model`, `agent.model`, `codex.model`, `codex.cron_model`), `max_turns`, `agent.effort`/`cron_effort` and `codex.effort_map`, `agent.thinking`, `agent.context_1m*`, `agent.background_agent_permissions`, `agent.agent_teams`, idle timeouts, cache TTL, `codex.sandbox`, `.approval_policy`, `.web_search`, `.extra_config`, `.tool_timeout_sec`, `.bin_path`, `.auth`/`.api_key`/`.api_key_env`, `.pricing`, `.min_version`/`.max_version`, `.ultracode.*` | ✅ for sessions and turns **started after** the reload. The engine and both backends resolve these through one live reference, so a key cannot be hot in one and frozen in the other | @@ -1052,6 +1052,7 @@ carry text. A `.png` or `.ico` has to be committed by a human. | `telegram.dm_policy` | string | `pairing` | `pairing` (allowlist + one-time pairing codes) or `open` (anyone — dangerous) | | `telegram.allowed_users` | list[int] | `[]` | Telegram user IDs allowed to DM the bot | | `telegram.stream_mode` | string | `partial` | `partial` (edit msgs) or `full` | +| `telegram.reply_routes_to_origin_session` | bool | `false` | When on, a reply routes to the session that produced the replied-to message (and makes it active); an unresolvable reply is refused. Off = a reply is context-only and lands in the active session | ### Pairing diff --git a/nerve/channels/telegram.py b/nerve/channels/telegram.py index 33bca77e..dbfd8372 100644 --- a/nerve/channels/telegram.py +++ b/nerve/channels/telegram.py @@ -989,18 +989,27 @@ def _cache_message(self, message_id: int, chat_id: int, text: str) -> None: # Reply routing (reply → originating session) # # ------------------------------------------------------------------ # + def _reply_routing_enabled(self) -> bool: + """Whether reply→origin-session routing is on (a telegram config flag). + + Off by default: a reply is then context-only and lands in the active + session, and none of the record / resolve / reject machinery runs. + Read live per message so a config reload can flip it without a restart. + """ + return bool(self.config.telegram.reply_routes_to_origin_session) + def _record_reply_route( self, message_id: int, chat_id: int, session_id: str, ) -> None: """Remember which session produced an outbound Telegram message. - Lets a later user REPLY to that message route back to the - originating session (see ``_resolve_reply``). No-op when the session - is unknown. Bounded LRU keyed by ``(chat_id, message_id)`` — message - ids repeat across chats, so keying on the id alone would let one chat - clobber another chat's identically-numbered message. + Lets a later user REPLY to that message route back to the originating + session (see ``_resolve_reply``). No-op when routing is disabled or the + session is unknown. Bounded LRU keyed by ``(chat_id, message_id)`` — + message ids repeat across chats, so keying on the id alone would let one + chat clobber another chat's identically-numbered message. """ - if not session_id: + if not session_id or not self._reply_routing_enabled(): return key = (chat_id, message_id) self._reply_routes[key] = session_id @@ -1719,31 +1728,35 @@ async def _handle_message(self, update: Update, context: Any) -> None: if images: metadata["images"] = images - # Route a reply to the session that produced the replied-to message - # (and make it active). An unresolvable reply is rejected with a no-LLM - # nudge rather than mis-routed into the chat's active session. - action, routed_session = await self._resolve_reply( - reply_msg.message_id if reply_msg else None, chat_id, - ) - if action == "reject": - logger.info( - "Telegram reply from chat %s not tied to a live session; nudging", - chat_id, - ) - try: - await update.message.reply_text(REPLY_UNRESOLVED_MSG) - except Exception: - logger.error("Failed to send reply-unresolved nudge to chat %s", chat_id) - return - if routed_session: - logger.info("Telegram reply routed to session %s", routed_session) - + # Reply routing (telegram.reply_routes_to_origin_session, default off). + # When off, a reply is context-only and lands in the active session + # (session_id stays None), exactly like a plain message. channel_key = f"telegram:{chat_id}" - target_session = await self._delivery_session(routed_session, channel_key) - # Record this inbound message against the session it lands in, so a later - # reply to it — including the user replying to their own message — routes - # back to the same session. - self._record_reply_route(update.message.message_id, chat_id, target_session) + target_session: str | None = None + if self._reply_routing_enabled(): + # Route a reply to the session that produced the replied-to message + # (and make it active). An unresolvable reply is rejected with a + # no-LLM nudge rather than mis-routed into the active session. + action, routed_session = await self._resolve_reply( + reply_msg.message_id if reply_msg else None, chat_id, + ) + if action == "reject": + logger.info( + "Telegram reply from chat %s not tied to a live session; nudging", + chat_id, + ) + try: + await update.message.reply_text(REPLY_UNRESOLVED_MSG) + except Exception: + logger.error("Failed to send reply-unresolved nudge to chat %s", chat_id) + return + if routed_session: + logger.info("Telegram reply routed to session %s", routed_session) + target_session = await self._delivery_session(routed_session, channel_key) + # Record this inbound message against the session it lands in, so a + # later reply to it — including the user replying to their own + # message — routes back to the same session. + self._record_reply_route(update.message.message_id, chat_id, target_session) msg = InboundMessage( channel_name="telegram", @@ -1912,31 +1925,35 @@ async def _process_media_group(self, group_id: str) -> None: if images: metadata["images"] = images - # Route a reply to the session that produced the replied-to message - # (and make it active). An unresolvable reply is rejected with a no-LLM - # nudge rather than mis-routed into the chat's active session. - action, routed_session = await self._resolve_reply( - reply_msg.message_id if reply_msg else None, chat_id, - ) - if action == "reject": - logger.info( - "Telegram reply from chat %s not tied to a live session; nudging", - chat_id, - ) - try: - await updates[0].message.reply_text(REPLY_UNRESOLVED_MSG) - except Exception: - logger.error("Failed to send reply-unresolved nudge to chat %s", chat_id) - return - if routed_session: - logger.info("Telegram reply routed to session %s", routed_session) - + # Reply routing (telegram.reply_routes_to_origin_session, default off). + # When off, a reply is context-only and lands in the active session + # (session_id stays None), exactly like a plain message. channel_key = f"telegram:{chat_id}" - target_session = await self._delivery_session(routed_session, channel_key) - # Record this inbound message against the session it lands in, so a later - # reply to it — including the user replying to their own message — routes - # back to the same session. - self._record_reply_route(updates[0].message.message_id, chat_id, target_session) + target_session: str | None = None + if self._reply_routing_enabled(): + # Route a reply to the session that produced the replied-to message + # (and make it active). An unresolvable reply is rejected with a + # no-LLM nudge rather than mis-routed into the active session. + action, routed_session = await self._resolve_reply( + reply_msg.message_id if reply_msg else None, chat_id, + ) + if action == "reject": + logger.info( + "Telegram reply from chat %s not tied to a live session; nudging", + chat_id, + ) + try: + await updates[0].message.reply_text(REPLY_UNRESOLVED_MSG) + except Exception: + logger.error("Failed to send reply-unresolved nudge to chat %s", chat_id) + return + if routed_session: + logger.info("Telegram reply routed to session %s", routed_session) + target_session = await self._delivery_session(routed_session, channel_key) + # Record this inbound message against the session it lands in, so a + # later reply to it — including the user replying to their own + # message — routes back to the same session. + self._record_reply_route(updates[0].message.message_id, chat_id, target_session) msg = InboundMessage( channel_name="telegram", diff --git a/nerve/config.py b/nerve/config.py index dc192688..e304e89c 100644 --- a/nerve/config.py +++ b/nerve/config.py @@ -953,6 +953,12 @@ class TelegramConfig: # agent access for any Telegram user. A warning # is logged at startup. dm_policy: str = "pairing" + # When True, a Telegram *reply* is routed to the session that produced the + # replied-to message (and that session becomes active); a reply that can't + # be tied to a live session is refused rather than delivered to the active + # session. When False (default) a reply is context-only — it quotes the + # replied-to message and lands in the active session, like a plain message. + reply_routes_to_origin_session: bool = False @classmethod @_coerced @@ -996,6 +1002,9 @@ def from_dict(cls, d: dict, locked: bool = False) -> TelegramConfig: allowed_users=d.get("allowed_users") or [], stream_mode=d.get("stream_mode", "partial"), dm_policy=dm_policy, + reply_routes_to_origin_session=d.get( + "reply_routes_to_origin_session", False, + ), ) diff --git a/tests/test_telegram_reply_routing.py b/tests/test_telegram_reply_routing.py index 5a2b297a..dccfc665 100644 --- a/tests/test_telegram_reply_routing.py +++ b/tests/test_telegram_reply_routing.py @@ -29,8 +29,13 @@ async def get_active_session(self, channel_key: str, source: str): return self._active -def _make_channel(router: _FakeRouter | None = None) -> TelegramChannel: - cfg = SimpleNamespace(telegram=SimpleNamespace(allowed_users=[])) +def _make_channel( + router: _FakeRouter | None = None, routing_enabled: bool = True, +) -> TelegramChannel: + cfg = SimpleNamespace(telegram=SimpleNamespace( + allowed_users=[], + reply_routes_to_origin_session=routing_enabled, + )) return TelegramChannel(lambda: cfg, router or _FakeRouter()) @@ -198,3 +203,20 @@ async def test_reply_to_own_recorded_message_routes_to_where_it_landed(): landing = await ch._delivery_session(None, "telegram:42") # → active1 ch._record_reply_route(500, 42, landing) # user's own msg id 500 assert await ch._resolve_reply(500, 42) == ("route", "active1") + + +# --------------------------------------------------------------------------- # +# Feature flag: telegram.reply_routes_to_origin_session (default off) # +# --------------------------------------------------------------------------- # + +def test_routing_enabled_reflects_config(): + assert _make_channel(routing_enabled=True)._reply_routing_enabled() is True + assert _make_channel(routing_enabled=False)._reply_routing_enabled() is False + + +def test_recording_is_skipped_when_routing_disabled(): + # With the flag off, nothing is recorded — so no message is ever a route + # target and replies fall through to the active session (legacy behavior). + ch = _make_channel(routing_enabled=False) + ch._record_reply_route(100, 42, "sessA") + assert ch._lookup_reply_route(100, 42) is None