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 a0c970db..dbfd8372 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 @@ -483,6 +491,16 @@ def __init__( collections.OrderedDict() ) self._message_cache_max = 200 + # 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. 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 # Rate limiter for replies to unauthorized users: user_id -> monotonic ts self._unauth_reply_times: dict[int, float] = {} @@ -825,6 +843,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 +985,91 @@ 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 _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 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 or not self._reply_routing_enabled(): + return + 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) + + def _lookup_reply_route( + self, message_id: int, chat_id: int, + ) -> str | None: + """Session that produced ``message_id`` in ``chat_id``, or None. + + 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. + """ + return self._reply_routes.get((chat_id, message_id)) + + async def _resolve_reply( + self, reply_message_id: int | None, chat_id: int, + ) -> 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 ("active", None) + session_id = self._lookup_reply_route(reply_message_id, chat_id) + if not session_id: + return ("reject", None) + try: + row = await self.router.get_session(session_id) + except Exception: + row = None + if not row or row.get("status") == "archived": + 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 # # ------------------------------------------------------------------ # @@ -1624,11 +1728,42 @@ async def _handle_message(self, update: Update, context: Any) -> None: if images: metadata["images"] = images + # 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: 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", - channel_key=f"telegram:{chat_id}", + channel_key=channel_key, sender_id=str(chat_id), text=text, + session_id=target_session, metadata=metadata, ) @@ -1790,11 +1925,42 @@ async def _process_media_group(self, group_id: str) -> None: if images: metadata["images"] = images + # 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: 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", - channel_key=f"telegram:{chat_id}", + channel_key=channel_key, sender_id=str(chat_id), text=text, + session_id=target_session, metadata=metadata, ) 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/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..dccfc665 --- /dev/null +++ b/tests/test_telegram_reply_routing.py @@ -0,0 +1,222 @@ +"""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 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 + +import pytest + +from nerve.channels.base import OutboundMessage +from nerve.channels.telegram import TelegramChannel + + +class _FakeRouter: + """Minimal router exposing the methods the channel calls here.""" + + 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, 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()) + + +# --------------------------------------------------------------------------- # +# 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_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 + 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 — three-way: active / route / reject # +# --------------------------------------------------------------------------- # + +@pytest.mark.asyncio +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(None, 42) == ("active", None) + + +@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(100, 42) == ("route", "sessA") + + +@pytest.mark.asyncio +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(100, 42) == ("reject", None) + + +@pytest.mark.asyncio +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(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_rejects_on_chat_mismatch(): + ch = _make_channel(_FakeRouter({"sessA": {"status": "active"}})) + ch._record_reply_route(100, 42, "sessA") # recorded for chat 42 + assert await ch._resolve_reply(100, 43) == ("reject", None) # replied in chat 43 + + +# --------------------------------------------------------------------------- # +# 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 + + +# --------------------------------------------------------------------------- # +# 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") + + +# --------------------------------------------------------------------------- # +# 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