Telegram: route a reply to the session that sent the replied-to message

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
 

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

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,
+            ),
         )
 
 

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