diff --git a/gateway/platforms/tlon.py b/gateway/platforms/tlon.py
index 049e604a977..2c373b5f72f 100644
--- a/gateway/platforms/tlon.py
+++ b/gateway/platforms/tlon.py
@@ -2299,6 +2299,11 @@ async def _dispatch_pending_message(self, approval: PendingApproval) -> None:
             user_id=approval.requesting_ship,
             user_name=approval.requesting_ship,
             thread_id=str(raw.get("thread_id")) if raw.get("thread_id") else None,
+            parent_chat_id=(
+                str(raw.get("parent_chat_id"))
+                if raw.get("parent_chat_id")
+                else None
+            ),
         )
         event_obj = MessageEvent(
             text=str(raw.get("text") or ""),
@@ -2684,6 +2689,15 @@ async def _handle_channel_event(self, event: Any) -> None:
                 return
 
             # Check user authorization
+            group_id = self._channel_to_group.get(nest)
+            if not group_id and self._sse:
+                try:
+                    await self._discover_channels()
+                    group_id = self._channel_to_group.get(nest)
+                except Exception as exc:
+                    logger.debug("[tlon] Group lookup refresh failed for %s: %s", nest, exc)
+            group_name = self._group_names.get(group_id or "")
+
             if not self._is_channel_allowed(sender, nest):
                 logger.info("[tlon] Unauthorized user %s in %s", sender, nest)
                 if self.owner_ship:
@@ -2697,6 +2711,7 @@ async def _handle_channel_event(self, event: Any) -> None:
                             "chat_id": nest,
                             "chat_name": (_parse_channel_nest(nest) or {}).get("name", nest),
                             "chat_type": "group",
+                            "parent_chat_id": group_id,
                             "text": self._strip_bot_mention(text) if mentioned else text,
                             "message_id": str(effective_id),
                             "reply_to_message_id": str(parent_id) if parent_id else None,
@@ -2716,13 +2731,16 @@ async def _handle_channel_event(self, event: Any) -> None:
 
             # Build message event
             parsed = _parse_channel_nest(nest)
+            channel_name = parsed["name"] if parsed else nest
+            chat_name = f"{group_name} / {channel_name}" if group_name else channel_name
             source = self.build_source(
                 chat_id=nest,
-                chat_name=parsed["name"] if parsed else nest,
+                chat_name=chat_name,
                 chat_type="group",
                 user_id=sender,
                 user_name=sender,
                 thread_id=str(parent_id) if parent_id else None,
+                parent_chat_id=group_id,
             )
 
             event_obj = MessageEvent(
diff --git a/gateway/run.py b/gateway/run.py
index 7d4f7f3c8be..9d19d386dbe 100644
--- a/gateway/run.py
+++ b/gateway/run.py
@@ -2732,6 +2732,33 @@ def _interrupt_running_agents(self, reason: str) -> None:
             except Exception as e:
                 logger.debug("Failed interrupting agent during shutdown: %s", e)
 
+    def _status_notification_target(
+        self,
+        platform: Platform,
+        adapter: BasePlatformAdapter,
+        chat_id: Any,
+        thread_id: Optional[Any] = None,
+    ) -> Optional[tuple[str, Optional[str]]]:
+        """Return the delivery target for gateway lifecycle/status notices.
+
+        Tlon group channels should not receive gateway lifecycle noise.  Route
+        those notices to the configured owner DM instead so the operator still
+        gets shutdown/restart/update visibility without polluting groups.
+        """
+        if platform == Platform.TLON:
+            owner_ship = str(
+                getattr(adapter, "owner_ship", "")
+                or os.getenv("TLON_OWNER_SHIP", "")
+            ).strip()
+            if not owner_ship:
+                logger.info("Skipping Tlon status notification: no owner ship configured")
+                return None
+            if not owner_ship.startswith("~"):
+                owner_ship = f"~{owner_ship}"
+            return owner_ship, None
+
+        return str(chat_id), str(thread_id) if thread_id else None
+
     async def _notify_active_sessions_of_shutdown(self) -> None:
         """Send shutdown/restart notifications to active chats and home channels.
 
@@ -2782,13 +2809,6 @@ async def _notify_active_sessions_of_shutdown(self) -> None:
                 chat_id = _parsed["chat_id"]
                 thread_id = _parsed.get("thread_id")
 
-            # Deduplicate only identical delivery targets. Thread/topic-aware
-            # platforms can share a parent chat while still routing to distinct
-            # destinations via metadata.
-            dedup_key = (platform_str, chat_id, str(thread_id) if thread_id else None)
-            if dedup_key in notified:
-                continue
-
             try:
                 platform = Platform(platform_str)
                 adapter = self.adapters.get(platform)
@@ -2803,16 +2823,28 @@ async def _notify_active_sessions_of_shutdown(self) -> None:
                     )
                     continue
 
+                target = self._status_notification_target(platform, adapter, chat_id, thread_id)
+                if target is None:
+                    continue
+                target_chat_id, target_thread_id = target
+
+                # Deduplicate only identical delivery targets. Thread/topic-aware
+                # platforms can share a parent chat while still routing to distinct
+                # destinations via metadata.
+                dedup_key = (platform_str, target_chat_id, target_thread_id)
+                if dedup_key in notified:
+                    continue
+
                 # Include thread_id if present so the message lands in the
                 # correct forum topic / thread.
-                metadata = {"thread_id": thread_id} if thread_id else None
+                metadata = {"thread_id": target_thread_id} if target_thread_id else None
 
-                result = await adapter.send(chat_id, msg, metadata=metadata)
+                result = await adapter.send(target_chat_id, msg, metadata=metadata)
                 if result is not None and getattr(result, "success", True) is False:
                     logger.debug(
                         "Failed to send shutdown notification to %s:%s: %s",
                         platform_str,
-                        chat_id,
+                        target_chat_id,
                         getattr(result, "error", "send returned success=False"),
                     )
                     continue
@@ -2820,7 +2852,7 @@ async def _notify_active_sessions_of_shutdown(self) -> None:
                 notified.add(dedup_key)
                 logger.info(
                     "Sent shutdown notification to active chat %s:%s",
-                    platform_str, chat_id,
+                    platform_str, target_chat_id,
                 )
             except Exception as e:
                 logger.debug(
@@ -2846,21 +2878,26 @@ async def _notify_active_sessions_of_shutdown(self) -> None:
                 )
                 continue
 
-            dedup_key = (platform.value, str(home.chat_id), str(home.thread_id) if home.thread_id else None)
+            target = self._status_notification_target(platform, adapter, home.chat_id, home.thread_id)
+            if target is None:
+                continue
+            target_chat_id, target_thread_id = target
+
+            dedup_key = (platform.value, target_chat_id, target_thread_id)
             if dedup_key in notified:
                 continue
 
             try:
-                metadata = {"thread_id": home.thread_id} if home.thread_id else None
+                metadata = {"thread_id": target_thread_id} if target_thread_id else None
                 if metadata:
-                    result = await adapter.send(str(home.chat_id), msg, metadata=metadata)
+                    result = await adapter.send(target_chat_id, msg, metadata=metadata)
                 else:
-                    result = await adapter.send(str(home.chat_id), msg)
+                    result = await adapter.send(target_chat_id, msg)
                 if result is not None and getattr(result, "success", True) is False:
                     logger.debug(
                         "Failed to send shutdown notification to home channel %s:%s: %s",
                         platform.value,
-                        home.chat_id,
+                        target_chat_id,
                         getattr(result, "error", "send returned success=False"),
                     )
                     continue
@@ -2869,7 +2906,7 @@ async def _notify_active_sessions_of_shutdown(self) -> None:
                 logger.info(
                     "Sent shutdown notification to home channel %s:%s",
                     platform.value,
-                    home.chat_id,
+                    target_chat_id,
                 )
             except Exception as e:
                 logger.debug(
@@ -12513,7 +12550,7 @@ def _collect_and_upload():
 
         return await loop.run_in_executor(None, _collect_and_upload)
 
-    async def _handle_update_command(self, event: MessageEvent) -> str:
+    async def _handle_update_command(self, event: MessageEvent) -> Optional[str]:
         """Handle /update command — update Hermes Agent to the latest version.
 
         Spawns ``hermes update`` in a detached session (via ``setsid``) so it
@@ -12657,7 +12694,29 @@ async def _handle_update_command(self, event: MessageEvent) -> str:
             return t("gateway.update.start_failed", error=e)
 
         self._schedule_update_notification_watch()
-        return t("gateway.update.starting")
+        start_message = t("gateway.update.starting")
+        if event.source.platform == Platform.TLON:
+            adapter = self.adapters.get(Platform.TLON)
+            if not adapter:
+                return None
+            target = self._status_notification_target(
+                Platform.TLON,
+                adapter,
+                event.source.chat_id,
+                event.source.thread_id,
+            )
+            if target is None:
+                return None
+            target_chat_id, target_thread_id = target
+            source_thread_id = str(event.source.thread_id) if event.source.thread_id else None
+            if target_chat_id != str(event.source.chat_id) or target_thread_id != source_thread_id:
+                metadata = {"thread_id": target_thread_id} if target_thread_id else None
+                try:
+                    await adapter.send(target_chat_id, start_message, metadata=metadata)
+                except Exception as exc:
+                    logger.warning("Tlon update start notification failed: %s", exc)
+                return None
+        return start_message
 
     def _schedule_update_notification_watch(self) -> None:
         """Ensure a background task is watching for update completion."""
@@ -12700,12 +12759,14 @@ async def _watch_update_progress(
         chat_id = None
         session_key = None
         metadata = None
+        prompt_session_keys: set[str] = set()
         for path in (claimed_path, pending_path):
             if path.exists():
                 try:
                     pending = json.loads(path.read_text())
                     platform_str = pending.get("platform")
                     chat_id = pending.get("chat_id")
+                    user_id = pending.get("user_id")
                     session_key = pending.get("session_key")
                     thread_id = pending.get("thread_id")
                     metadata = {"thread_id": thread_id} if thread_id else None
@@ -12715,6 +12776,38 @@ async def _watch_update_progress(
                         # Fallback session key if not stored (old pending files)
                         if not session_key:
                             session_key = f"{platform_str}:{chat_id}"
+                        prompt_session_keys = {session_key}
+                        if adapter:
+                            target = self._status_notification_target(
+                                platform,
+                                adapter,
+                                chat_id,
+                                thread_id,
+                            )
+                            if target is None:
+                                adapter = None
+                                chat_id = None
+                                break
+                            target_chat_id, target_thread_id = target
+                            source_thread_id = str(thread_id) if thread_id else None
+                            if target_chat_id != str(chat_id) or target_thread_id != source_thread_id:
+                                try:
+                                    target_source = SessionSource(
+                                        platform=platform,
+                                        chat_id=target_chat_id,
+                                        chat_type="dm",
+                                        user_id=str(user_id or target_chat_id),
+                                    )
+                                    prompt_session_keys.add(
+                                        self._session_key_for_source(target_source)
+                                    )
+                                except Exception:
+                                    prompt_session_keys.add(f"{platform_str}:{target_chat_id}")
+                            chat_id = target_chat_id
+                            metadata = (
+                                {"thread_id": target_thread_id}
+                                if target_thread_id else None
+                            )
                     break
                 except Exception:
                     pass
@@ -12795,7 +12888,8 @@ async def _flush_buffer() -> None:
                           exit_code_path, prompt_path):
                     p.unlink(missing_ok=True)
                 (_hermes_home / ".update_response").unlink(missing_ok=True)
-                self._update_prompt_pending.pop(session_key, None)
+                for key in prompt_session_keys:
+                    self._update_prompt_pending.pop(key, None)
                 return
 
             # Check for new output
@@ -12817,7 +12911,10 @@ async def _flush_buffer() -> None:
             # watcher would re-read the same .update_prompt.json every poll
             # cycle and spam the user with duplicate prompt messages.
             if (prompt_path.exists() and session_key
-                    and not self._update_prompt_pending.get(session_key)):
+                    and not any(
+                        self._update_prompt_pending.get(key)
+                        for key in prompt_session_keys
+                    )):
                 try:
                     prompt_data = json.loads(prompt_path.read_text())
                     prompt_text = prompt_data.get("prompt", "")
@@ -12855,7 +12952,8 @@ async def _flush_buffer() -> None:
                         # next watcher can recover by re-forwarding it from
                         # disk. Duplicate sends in the same process are
                         # still suppressed by _update_prompt_pending.
-                        self._update_prompt_pending[session_key] = True
+                        for key in prompt_session_keys:
+                            self._update_prompt_pending[key] = True
                         # .update_response to continue — it doesn't re-check
                         logger.info("Forwarded update prompt to %s: %s", session_key, prompt_text[:80])
                 except (json.JSONDecodeError, OSError) as e:
@@ -12880,7 +12978,8 @@ async def _flush_buffer() -> None:
                       exit_code_path, prompt_path):
                 p.unlink(missing_ok=True)
             (_hermes_home / ".update_response").unlink(missing_ok=True)
-            self._update_prompt_pending.pop(session_key, None)
+            for key in prompt_session_keys:
+                self._update_prompt_pending.pop(key, None)
 
     async def _send_update_notification(self) -> bool:
         """If an update finished, notify the user.
@@ -12937,7 +13036,11 @@ async def _send_update_notification(self) -> bool:
             adapter = self.adapters.get(platform)
 
             if adapter and chat_id:
-                metadata = {"thread_id": thread_id} if thread_id else None
+                target = self._status_notification_target(platform, adapter, chat_id, thread_id)
+                if target is None:
+                    return True
+                target_chat_id, target_thread_id = target
+                metadata = {"thread_id": target_thread_id} if target_thread_id else None
                 # Strip ANSI escape codes for clean display
                 output = re.sub(r'\x1b\[[0-9;]*m', '', output).strip()
                 if output:
@@ -12951,11 +13054,11 @@ async def _send_update_notification(self) -> bool:
                     msg = "✅ Hermes update finished successfully."
                 else:
                     msg = "❌ Hermes update failed. Check the gateway logs or run `hermes update` manually for details."
-                await adapter.send(chat_id, msg, metadata=metadata)
+                await adapter.send(target_chat_id, msg, metadata=metadata)
                 logger.info(
                     "Sent post-update notification to %s:%s (exit=%s)",
                     platform_str,
-                    chat_id,
+                    target_chat_id,
                     exit_code,
                 )
         except Exception as e:
@@ -13001,9 +13104,14 @@ async def _send_restart_notification(self) -> Optional[tuple[str, str, Optional[
                 )
                 return None
 
-            metadata = {"thread_id": thread_id} if thread_id else None
+            target = self._status_notification_target(platform, adapter, chat_id, thread_id)
+            if target is None:
+                return None
+            target_chat_id, target_thread_id = target
+
+            metadata = {"thread_id": target_thread_id} if target_thread_id else None
             result = await adapter.send(
-                str(chat_id),
+                target_chat_id,
                 "♻ Gateway restarted successfully. Your session continues.",
                 metadata=metadata,
             )
@@ -13015,7 +13123,7 @@ async def _send_restart_notification(self) -> Optional[tuple[str, str, Optional[
                 logger.warning(
                     "Restart notification to %s:%s was not delivered: %s",
                     platform_str,
-                    chat_id,
+                    target_chat_id,
                     getattr(result, "error", "send returned success=False"),
                 )
                 return None
@@ -13023,9 +13131,9 @@ async def _send_restart_notification(self) -> Optional[tuple[str, str, Optional[
             logger.info(
                 "Sent restart notification to %s:%s",
                 platform_str,
-                chat_id,
+                target_chat_id,
             )
-            return str(platform_str), str(chat_id), str(thread_id) if thread_id else None
+            return str(platform_str), target_chat_id, target_thread_id
         except Exception as e:
             logger.warning("Restart notification failed: %s", e)
             return None
@@ -13060,21 +13168,26 @@ async def _send_home_channel_startup_notifications(
                 )
                 continue
 
-            target = (platform.value, str(home.chat_id), str(home.thread_id) if home.thread_id else None)
+            status_target = self._status_notification_target(platform, adapter, home.chat_id, home.thread_id)
+            if status_target is None:
+                continue
+            target_chat_id, target_thread_id = status_target
+
+            target = (platform.value, target_chat_id, target_thread_id)
             if target in skipped or target in delivered:
                 continue
 
             try:
-                metadata = {"thread_id": home.thread_id} if home.thread_id else None
+                metadata = {"thread_id": target_thread_id} if target_thread_id else None
                 if metadata:
-                    result = await adapter.send(str(home.chat_id), message, metadata=metadata)
+                    result = await adapter.send(target_chat_id, message, metadata=metadata)
                 else:
-                    result = await adapter.send(str(home.chat_id), message)
+                    result = await adapter.send(target_chat_id, message)
                 if result is not None and getattr(result, "success", True) is False:
                     logger.warning(
                         "Home-channel startup notification failed for %s:%s: %s",
                         platform.value,
-                        home.chat_id,
+                        target_chat_id,
                         getattr(result, "error", "send returned success=False"),
                     )
                     continue
@@ -13083,7 +13196,7 @@ async def _send_home_channel_startup_notifications(
                 logger.info(
                     "Sent home-channel startup notification to %s:%s",
                     platform.value,
-                    home.chat_id,
+                    target_chat_id,
                 )
             except Exception as exc:
                 logger.warning(
diff --git a/gateway/session.py b/gateway/session.py
index dfa2ca9651d..073455e3d4f 100644
--- a/gateway/session.py
+++ b/gateway/session.py
@@ -373,6 +373,19 @@ def build_session_context_prompt(
             "Use target='yuanbao:direct:<account_id>' for DM "
             "and target='yuanbao:group:<group_code>' for group chat."
         )
+    elif context.source.platform == Platform.TLON:
+        src = context.source
+        if src.chat_type == "group":
+            lines.append("")
+            lines.append("**Tlon IDs:**")
+            if src.parent_chat_id:
+                lines.append(f"  - Group: `{src.parent_chat_id}` (use as `group_id` for the `tlon` tool)")
+            lines.append(f"  - Channel: `{src.chat_id}` (use as `channel_id` for messages/channels)")
+            if src.thread_id:
+                lines.append(f"  - Thread/root post: `{src.thread_id}`")
+            lines.append(
+                "When the user refers to this group or channel, use these exact IDs."
+            )
 
     # Connected platforms
     platforms_list = ["local (files on this machine)"]
diff --git a/tests/gateway/test_restart_drain.py b/tests/gateway/test_restart_drain.py
index 844af427308..b3e1cd602af 100644
--- a/tests/gateway/test_restart_drain.py
+++ b/tests/gateway/test_restart_drain.py
@@ -14,6 +14,16 @@
 from tests.gateway.restart_test_helpers import make_restart_runner, make_restart_source
 
 
+def make_tlon_status_runner(owner_ship="~malmur-halmex"):
+    runner, adapter = make_restart_runner()
+    adapter.owner_ship = owner_ship
+    runner.config.platforms = {
+        gateway_run.Platform.TLON: gateway_run.PlatformConfig(enabled=True),
+    }
+    runner.adapters = {gateway_run.Platform.TLON: adapter}
+    return runner, adapter
+
+
 @pytest.mark.asyncio
 async def test_restart_command_while_busy_requests_drain_without_interrupt(monkeypatch):
     # Ensure INVOCATION_ID is NOT set — systemd sets this in service mode,
@@ -318,3 +328,55 @@ async def test_shutdown_notification_uses_persisted_origin_for_colon_ids():
 
     assert adapter.send.await_count == 1
     assert adapter.send.await_args.args[0] == "!room123:example.org"
+
+
+@pytest.mark.asyncio
+async def test_tlon_shutdown_notification_routes_group_session_to_owner_dm():
+    """Tlon lifecycle status must go to the owner DM, never the group channel."""
+    runner, adapter = make_tlon_status_runner()
+    source = make_restart_source(
+        chat_id="chat/~ramlud-bintun/v1fsl36d",
+        chat_type="group",
+    )
+    source.platform = gateway_run.Platform.TLON
+    session_key = build_session_key(source)
+    runner._running_agents[session_key] = MagicMock()
+    runner.session_store._entries = {
+        session_key: SessionEntry(
+            session_key=session_key,
+            session_id="sess-tlon",
+            created_at=datetime.now(),
+            updated_at=datetime.now(),
+            origin=source,
+            platform=source.platform,
+            chat_type=source.chat_type,
+        )
+    }
+
+    await runner._notify_active_sessions_of_shutdown()
+
+    assert len(adapter.sent_calls) == 1
+    chat_id, content, metadata = adapter.sent_calls[0]
+    assert chat_id == "~malmur-halmex"
+    assert "Gateway shutting down" in content
+    assert metadata is None
+
+
+@pytest.mark.asyncio
+async def test_tlon_shutdown_notification_skips_without_owner_ship(monkeypatch):
+    monkeypatch.delenv("TLON_OWNER_SHIP", raising=False)
+    runner, adapter = make_tlon_status_runner(owner_ship="")
+    source = make_restart_source(
+        chat_id="chat/~ramlud-bintun/v1fsl36d",
+        chat_type="group",
+    )
+    source.platform = gateway_run.Platform.TLON
+    session_key = build_session_key(source)
+    runner._running_agents[session_key] = MagicMock()
+    runner.session_store._entries = {
+        session_key: MagicMock(origin=source),
+    }
+
+    await runner._notify_active_sessions_of_shutdown()
+
+    assert adapter.sent_calls == []
diff --git a/tests/gateway/test_restart_notification.py b/tests/gateway/test_restart_notification.py
index 3d5d5ee9557..545e709591c 100644
--- a/tests/gateway/test_restart_notification.py
+++ b/tests/gateway/test_restart_notification.py
@@ -17,6 +17,16 @@
 )
 
 
+def make_tlon_status_runner(owner_ship="~malmur-halmex"):
+    runner, adapter = make_restart_runner()
+    adapter.owner_ship = owner_ship
+    runner.config.platforms = {
+        Platform.TLON: gateway_run.PlatformConfig(enabled=True),
+    }
+    runner.adapters = {Platform.TLON: adapter}
+    return runner, adapter
+
+
 # ── restart marker helpers ───────────────────────────────────────────────
 
 
@@ -388,6 +398,48 @@ async def test_send_restart_notification_with_thread(tmp_path, monkeypatch):
     assert not notify_path.exists()
 
 
+@pytest.mark.asyncio
+async def test_tlon_restart_notification_routes_group_to_owner_dm(tmp_path, monkeypatch):
+    monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path)
+
+    notify_path = tmp_path / ".restart_notify.json"
+    notify_path.write_text(json.dumps({
+        "platform": "tlon",
+        "chat_id": "chat/~ramlud-bintun/v1fsl36d",
+        "thread_id": "170.141",
+    }))
+
+    runner, adapter = make_tlon_status_runner()
+
+    delivered_target = await runner._send_restart_notification()
+
+    assert delivered_target == ("tlon", "~malmur-halmex", None)
+    chat_id, content, metadata = adapter.sent_calls[0]
+    assert chat_id == "~malmur-halmex"
+    assert "restarted" in content.lower()
+    assert metadata is None
+    assert not notify_path.exists()
+
+
+@pytest.mark.asyncio
+async def test_tlon_home_startup_notification_routes_home_group_to_owner_dm():
+    runner, adapter = make_tlon_status_runner()
+    runner.config.platforms[Platform.TLON].home_channel = HomeChannel(
+        platform=Platform.TLON,
+        chat_id="chat/~ramlud-bintun/v1fsl36d",
+        name="Hermes Group",
+        thread_id="170.141",
+    )
+
+    delivered = await runner._send_home_channel_startup_notifications()
+
+    assert delivered == {("tlon", "~malmur-halmex", None)}
+    chat_id, content, metadata = adapter.sent_calls[0]
+    assert chat_id == "~malmur-halmex"
+    assert "Gateway online" in content
+    assert metadata is None
+
+
 @pytest.mark.asyncio
 async def test_send_restart_notification_noop_when_no_file(tmp_path, monkeypatch):
     """Nothing happens if there's no pending restart notification."""
diff --git a/tests/gateway/test_session.py b/tests/gateway/test_session.py
index b8fd45558cd..13db98b7fde 100644
--- a/tests/gateway/test_session.py
+++ b/tests/gateway/test_session.py
@@ -214,6 +214,29 @@ def test_bluebubbles_prompt_mentions_short_conversational_i_message_format(self)
         assert "short and conversational" in prompt
         assert "blank line" in prompt
 
+    def test_tlon_group_prompt_includes_group_and_channel_ids(self):
+        config = GatewayConfig(
+            platforms={
+                Platform.TLON: PlatformConfig(enabled=True),
+            },
+        )
+        source = SessionSource(
+            platform=Platform.TLON,
+            chat_id="chat/~ramlud-bintun/v1fsl36d",
+            chat_name="Hermes Group / general",
+            chat_type="group",
+            user_id="~malmur-halmex",
+            user_name="~malmur-halmex",
+            parent_chat_id="~ramlud-bintun/v1l3qcoq",
+        )
+        ctx = build_session_context(source, config)
+        prompt = build_session_context_prompt(ctx)
+
+        assert "Tlon IDs" in prompt
+        assert "Group: `~ramlud-bintun/v1l3qcoq`" in prompt
+        assert "Channel: `chat/~ramlud-bintun/v1fsl36d`" in prompt
+        assert "use these exact IDs" in prompt
+
     def test_discord_prompt(self):
         config = GatewayConfig(
             platforms={
diff --git a/tests/gateway/test_tlon_adapter.py b/tests/gateway/test_tlon_adapter.py
index 4d15da57890..5847950e6cd 100644
--- a/tests/gateway/test_tlon_adapter.py
+++ b/tests/gateway/test_tlon_adapter.py
@@ -416,8 +416,11 @@ async def test_foreigns_event_accepts_owner_invite_even_without_auto_accept(monk
 @pytest.mark.asyncio
 async def test_channel_event_routes_top_level_mentions(monkeypatch):
     monkeypatch.setenv("TLON_SHIP_NAME", "~bot-palnet")
+    monkeypatch.setenv("TLON_ALLOW_ALL_USERS", "true")
     adapter = TlonAdapter(PlatformConfig())
     adapter.monitored_channels = {"chat/~host/test"}
+    adapter._channel_to_group["chat/~host/test"] = "~host/group"
+    adapter._group_names["~host/group"] = "Test Group"
     adapter.handle_message = AsyncMock()
 
     await adapter._handle_channel_event({
@@ -447,6 +450,8 @@ async def test_channel_event_routes_top_level_mentions(monkeypatch):
     assert event.message_id == "170141184507864167403996323545639550976"
     assert event.reply_to_message_id is None
     assert event.source.chat_id == "chat/~host/test"
+    assert event.source.parent_chat_id == "~host/group"
+    assert event.source.chat_name == "Test Group / test"
     assert event.source.user_id == "~zod"
     assert isinstance(event.timestamp, datetime)
 
@@ -661,6 +666,7 @@ async def test_channel_event_ignores_owner_when_owner_listen_disabled_for_channe
 @pytest.mark.asyncio
 async def test_channel_event_routes_thread_reply_to_parent(monkeypatch):
     monkeypatch.setenv("TLON_SHIP_NAME", "~bot-palnet")
+    monkeypatch.setenv("TLON_ALLOW_ALL_USERS", "true")
     adapter = TlonAdapter(PlatformConfig())
     adapter.monitored_channels = {"chat/~host/test"}
     adapter.handle_message = AsyncMock()
@@ -741,6 +747,50 @@ async def test_channel_event_routes_openclaw_thread_reply_essay(monkeypatch):
     assert event.source.thread_id == "parent-post"
 
 
+@pytest.mark.asyncio
+async def test_channel_event_refreshes_group_mapping_for_context(monkeypatch):
+    monkeypatch.setenv("TLON_SHIP_NAME", "~bot-palnet")
+    monkeypatch.setenv("TLON_ALLOW_ALL_USERS", "true")
+    monkeypatch.setenv("TLON_AUTO_DISCOVER", "true")
+    adapter = TlonAdapter(PlatformConfig())
+    adapter.monitored_channels = {"chat/~host/test"}
+    adapter._sse = AsyncMock()
+    adapter._sse.scry.return_value = {
+        "groups": {
+            "~host/group": {
+                "meta": {"title": "Test Group"},
+                "channels": {"chat/~host/test": {}},
+            }
+        }
+    }
+    adapter.handle_message = AsyncMock()
+
+    await adapter._handle_channel_event({
+        "nest": "chat/~host/test",
+        "response": {
+            "post": {
+                "id": "post-id",
+                "r-post": {
+                    "set": {
+                        "seal": {"id": "post-id"},
+                        "essay": {
+                            "author": "~zod",
+                            "sent": 1_700_000_000_000,
+                            "content": [
+                                {"inline": [{"ship": "~bot-palnet"}, " hello"]}
+                            ],
+                        },
+                    }
+                },
+            }
+        },
+    })
+
+    event = adapter.handle_message.await_args.args[0]
+    assert event.source.parent_chat_id == "~host/group"
+    assert event.source.chat_name == "Test Group / test"
+
+
 @pytest.mark.asyncio
 async def test_channel_event_routes_blob_only_owner_message(monkeypatch):
     monkeypatch.setenv("TLON_SHIP_NAME", "~bot-palnet")
diff --git a/tests/gateway/test_update_command.py b/tests/gateway/test_update_command.py
index aa6240aa5b5..ac51dcbd9eb 100644
--- a/tests/gateway/test_update_command.py
+++ b/tests/gateway/test_update_command.py
@@ -486,6 +486,36 @@ async def test_sends_notification_with_thread_metadata(self, tmp_path):
 
         assert mock_adapter.send.call_args.kwargs["metadata"] == {"thread_id": "777"}
 
+    @pytest.mark.asyncio
+    async def test_tlon_update_notification_routes_group_to_owner_dm(self, tmp_path):
+        """Tlon update status goes to owner DM instead of the source group."""
+        runner = _make_runner()
+        hermes_home = tmp_path / "hermes"
+        hermes_home.mkdir()
+
+        pending = {
+            "platform": "tlon",
+            "chat_id": "chat/~ramlud-bintun/v1fsl36d",
+            "thread_id": "170.141",
+            "user_id": "~malmur-halmex",
+        }
+        (hermes_home / ".update_pending.json").write_text(json.dumps(pending))
+        (hermes_home / ".update_output.txt").write_text("done")
+        (hermes_home / ".update_exit_code").write_text("0")
+
+        mock_adapter = AsyncMock()
+        mock_adapter.owner_ship = "~malmur-halmex"
+        mock_adapter.send = AsyncMock()
+        runner.adapters = {Platform.TLON: mock_adapter}
+
+        with patch("gateway.run._hermes_home", hermes_home):
+            await runner._send_update_notification()
+
+        mock_adapter.send.assert_called_once()
+        call_args = mock_adapter.send.call_args
+        assert call_args.args[0] == "~malmur-halmex"
+        assert call_args.kwargs["metadata"] is None
+
     @pytest.mark.asyncio
     async def test_strips_ansi_codes(self, tmp_path):
         """ANSI escape codes are removed from output."""
diff --git a/tests/gateway/test_update_streaming.py b/tests/gateway/test_update_streaming.py
index 932bd1b0579..678ca030a8f 100644
--- a/tests/gateway/test_update_streaming.py
+++ b/tests/gateway/test_update_streaming.py
@@ -437,6 +437,40 @@ async def test_failure_exit_code(self, tmp_path):
         all_sent = " ".join(str(c) for c in mock_adapter.send.call_args_list)
         assert "failed" in all_sent.lower()
 
+    @pytest.mark.asyncio
+    async def test_tlon_streaming_update_routes_to_owner_dm(self, tmp_path):
+        """Tlon update stream/final status must not post in the source group."""
+        runner = _make_runner()
+        hermes_home = tmp_path / "hermes"
+        hermes_home.mkdir()
+
+        pending = {
+            "platform": "tlon",
+            "chat_id": "chat/~ramlud-bintun/v1fsl36d",
+            "thread_id": "170.141",
+            "user_id": "~malmur-halmex",
+            "session_key": "agent:main:tlon:group:chat/~ramlud-bintun/v1fsl36d:170.141",
+        }
+        (hermes_home / ".update_pending.json").write_text(json.dumps(pending))
+        (hermes_home / ".update_output.txt").write_text("done\n")
+        (hermes_home / ".update_exit_code").write_text("0")
+
+        mock_adapter = AsyncMock()
+        mock_adapter.owner_ship = "~malmur-halmex"
+        runner.adapters = {Platform.TLON: mock_adapter}
+
+        with patch("gateway.run._hermes_home", hermes_home):
+            await runner._watch_update_progress(
+                poll_interval=0.1,
+                stream_interval=0.2,
+                timeout=5.0,
+            )
+
+        assert mock_adapter.send.call_count >= 1
+        sent_targets = [call.args[0] for call in mock_adapter.send.call_args_list]
+        assert set(sent_targets) == {"~malmur-halmex"}
+        assert all(call.kwargs.get("metadata") is None for call in mock_adapter.send.call_args_list)
+
     @pytest.mark.asyncio
     async def test_falls_back_when_adapter_unavailable(self, tmp_path):
         """Falls back to legacy notification when adapter can't be resolved."""
diff --git a/tests/tools/test_tlon_tool.py b/tests/tools/test_tlon_tool.py
index 083e1f2b9b3..f64adb2fc4a 100644
--- a/tests/tools/test_tlon_tool.py
+++ b/tests/tools/test_tlon_tool.py
@@ -189,6 +189,7 @@ async def test_group_create_owned_creates_group_and_assigns_admin():
     assert result["admin_assigned"] is True
     assert result["admin_assignment"]["promoted"] == ["~malmur-halmex"]
 
+
 @pytest.mark.asyncio
 async def test_group_create_with_admins_force_adds_admin_seats():
     client = FakeTlonClient()
diff --git a/tools/tlon_tool.py b/tools/tlon_tool.py
index 3c1283bdf11..a5b1227875f 100644
--- a/tools/tlon_tool.py
+++ b/tools/tlon_tool.py
@@ -847,11 +847,36 @@ async def handle(self, action: str, args: Dict[str, Any]) -> Dict[str, Any]:
                 return _ok(action, mode=mode, channels=_filter_init_channels(init, mode))
             raise TlonToolError("channels_list mode must be all, groups, dms, or group_dms")
         if action == "channel_info":
-            channel_id = _required(args, "channel_id")
-            group_id = args.get("group_id")
+            channel_id = _normalize_channel_ref(_required(args, "channel_id"))
+            group_id = _normalize_group_ref(str(args.get("group_id") or ""))
             if group_id:
-                group = await self.client.scry("groups", f"/v2/ui/groups/{group_id}")
-                return _ok(action, channel_id=channel_id, group_id=group_id, channel=_find_channel(group, channel_id), group=group)
+                try:
+                    group = await self.client.scry("groups", f"/v2/ui/groups/{group_id}")
+                    return _ok(action, channel_id=channel_id, group_id=group_id, channel=_find_channel(group, channel_id), group=group)
+                except TlonHttpError as exc:
+                    if exc.status != 404:
+                        raise
+                    groups = await _groups_index(self.client)
+                    match = _find_group_in_groups(groups, channel_id)
+                    if match:
+                        resolved_group_id, group = match
+                        return _ok(
+                            action,
+                            channel_id=channel_id,
+                            group_id=resolved_group_id,
+                            requested_group_id=group_id,
+                            channel=_find_channel(group, channel_id),
+                            group=group,
+                            resolved_from="channel_id",
+                        )
+                    return _ok(
+                        action,
+                        found=False,
+                        channel_id=channel_id,
+                        requested_group_id=group_id,
+                        candidates=_candidate_groups(groups, group_id, channel_id),
+                        hint="Group not found for channel. Use one of the candidate group_id values.",
+                    )
             groups = await self.client.scry("groups", "/v2/groups")
             return _ok(action, channel_id=channel_id, match=_find_channel_in_groups(groups, channel_id))
         if action == "channel_create":
@@ -1906,6 +1931,8 @@ def _candidate_groups(groups: Any, requested_group_id: str = "", channel_id: str
     for group_id, group in _iter_group_items(groups):
         channels = _group_channel_ids(group)
         if channel_id and channel_id not in channels:
+            # Keep host matches as useful context even when the specific
+            # channel was not present in the local groups index.
             if requested_host and not group_id.startswith(f"{requested_host}/"):
                 continue
         elif requested_host and not group_id.startswith(f"{requested_host}/"):