Delete stopped session topics automatically

Author: linfeng.wang

src/telegram_codex_controller/console.py

@@ -72,6 +72,7 @@ async def start(self, application: Application) -> None:
         self._application = application
         self._prune_expired_records()
         self._restore_reply_routes()
+        self._queue_existing_stopped_session_cleanups()
         if self._stuck_task is None:
             self._stuck_task = asyncio.create_task(self._stuck_watcher())
         if self._pin_task is None:
@@ -176,7 +177,11 @@ async def update_status(self, spec: SessionStatusSpec) -> None:
         status_text = self._render_status_text(record)
         last_text = record.get("status_text")
         record["_pending_status_text"] = status_text
-        if self._session_has_pending_work(record):
+        if spec.state == "stopped":
+            self._dirty_session_routes.discard(spec.route.label)
+            self._pending_status_due_monotonic.pop(spec.route.label, None)
+            self._queue_stopped_session_cleanup(record)
+        elif self._session_has_pending_work(record):
             self._mark_session_dirty(
                 spec.route.label,
                 record,
@@ -354,6 +359,7 @@ def clear_session(self, route: ReplyRoute) -> None:
         self._topic_bump_reason_by_route.pop(route.label, None)
         self._topic_bump_queue_seen.discard(route.label)
         self._last_status_render_monotonic.pop(route.label, None)
+        self._mark_index_dirty(immediate=True, urgent=True)
         self._persist()
 
     def lookup_route_by_message(self, chat_id: int, message_id: int) -> ReplyRoute | None:
@@ -953,6 +959,10 @@ async def _flush_once(self) -> None:
             if handled:
                 return
 
+        handled = await self._flush_stopped_session_cleanup()
+        if handled:
+            return
+
         handled = await self._flush_stale_topic_cleanup()
         if handled:
             return
@@ -1397,6 +1407,100 @@ def _mark_session_dirty(
         self._pending_status_due_monotonic[route_label] = due
         self._dirty_session_routes.add(route_label)
 
+    def _queue_existing_stopped_session_cleanups(self) -> None:
+        for record in self._state.get("sessions", {}).values():
+            if record.get("state") == "stopped":
+                self._queue_stopped_session_cleanup(record)
+
+    def _queue_stopped_session_cleanup(self, record: dict[str, Any]) -> None:
+        if record.get("_pending_topic_delete"):
+            return
+        route_kind = record.get("route_kind")
+        route_target = record.get("route_target")
+        if not route_kind or not route_target:
+            return
+        record["_pending_topic_delete"] = {
+            "route_kind": route_kind,
+            "route_target": route_target,
+            "topic_id": record.get("topic_id"),
+            "status_thread_id": record.get("status_thread_id"),
+            "topic_managed": bool(record.get("topic_managed", True)),
+            "chat_id": record.get("status_chat_id") or self._console_chat_id(),
+            "status_message_id": record.get("status_message_id"),
+            "topic_bump_message_id": record.get("topic_bump_message_id"),
+            "last_event_message_id": record.get("last_event_message_id"),
+            "label": record.get("label") or route_target,
+        }
+
+    async def _flush_stopped_session_cleanup(self) -> bool:
+        if self._application is None:
+            return False
+        for route_label, record in list(self._state.get("sessions", {}).items()):
+            pending = record.get("_pending_topic_delete")
+            if not pending or record.get("state") != "stopped":
+                continue
+
+            route = ReplyRoute(kind=pending["route_kind"], target=pending["route_target"])
+            chat_id = int(pending.get("chat_id") or self._console_chat_id())
+            status_message_id = pending.get("status_message_id")
+            topic_bump_message_id = pending.get("topic_bump_message_id")
+            event_message_id = pending.get("last_event_message_id")
+            topic_ids = []
+            for candidate in [pending.get("topic_id"), pending.get("status_thread_id")]:
+                if candidate and candidate not in topic_ids:
+                    topic_ids.append(candidate)
+            topic_managed = bool(pending.get("topic_managed", True))
+            now = time.monotonic()
+
+            try:
+                for message_id in [status_message_id, topic_bump_message_id, event_message_id]:
+                    if not message_id:
+                        continue
+                    with contextlib.suppress(BadRequest):
+                        await self._application.bot.delete_message(
+                            chat_id=chat_id,
+                            message_id=int(message_id),
+                        )
+                if topic_ids and self.forum_enabled():
+                    if topic_managed:
+                        for topic_id in topic_ids:
+                            with contextlib.suppress(BadRequest):
+                                await self._application.bot.delete_forum_topic(
+                                    chat_id=chat_id,
+                                    message_thread_id=int(topic_id),
+                                )
+                    else:
+                        for topic_id in topic_ids:
+                            with contextlib.suppress(BadRequest):
+                                await self._application.bot.edit_forum_topic(
+                                    chat_id=chat_id,
+                                    message_thread_id=int(topic_id),
+                                    name=_archived_topic_title(
+                                        {
+                                            "label": pending.get("label") or pending.get("route_target") or "?",
+                                            "topic_title": record.get("topic_title") or "",
+                                        }
+                                    ),
+                                )
+                            with contextlib.suppress(BadRequest):
+                                await self._application.bot.close_forum_topic(
+                                    chat_id=chat_id,
+                                    message_thread_id=int(topic_id),
+                                )
+                self._record_write(now, state="stopped", urgent=True)
+                self.clear_session(route)
+                LOG.info("Deleted stopped session topic for %s", route_label)
+                return True
+            except RetryAfter as exc:
+                self._enter_write_backoff(float(exc.retry_after) + 1)
+                LOG.info("Stopped session cleanup backed off for %.1fs", float(exc.retry_after))
+                return False
+            except TelegramError:
+                LOG.exception("Failed to delete stopped session topic for %s", route_label)
+                self.clear_session(route)
+                return False
+        return False
+
     def _queue_stale_topic_cleanup(self, record: dict[str, Any], *, new_topic_id: int | None) -> None:
         old_topic_id = record.get("topic_id")
         if old_topic_id is None or old_topic_id == new_topic_id:

tests/test_console.py

@@ -33,6 +33,9 @@ async def edit_forum_topic(self, **kwargs: object) -> None:
     async def close_forum_topic(self, **kwargs: object) -> None:
         self.calls.append(("close_forum_topic", kwargs))
 
+    async def delete_forum_topic(self, **kwargs: object) -> None:
+        self.calls.append(("delete_forum_topic", kwargs))
+
 
 def make_settings(root: Path) -> Settings:
     sidecar = root / "sidecar" / "runner.mjs"
@@ -389,6 +392,89 @@ def test_flush_stale_topic_cleanup_archives_previous_topic(self) -> None:
                 ],
             )
 
+    def test_stopped_transition_queues_topic_delete_cleanup(self) -> None:
+        with tempfile.TemporaryDirectory() as temp_dir:
+            settings = make_settings(Path(temp_dir))
+            manager = TelegramConsoleManager(settings)
+            manager._application = FakeApplication(bot=FakeBot())
+            manager.set_forum_enabled(True)
+            route = ReplyRoute("mirror", "ttys039")
+            manager._state["sessions"][route.label] = {
+                "route_kind": "mirror",
+                "route_target": "ttys039",
+                "kind": "mirror",
+                "label": "ttys039",
+                "title": "Terminal",
+                "state": "running",
+                "step": "mirroring",
+                "summary": "still working",
+                "updated_at": "2026-03-15T21:00:00+00:00",
+                "status_message_id": 11,
+                "status_chat_id": -1001,
+                "topic_id": 42,
+                "topic_title": "🟢 ttys039",
+                "topic_bump_message_id": 12,
+            }
+
+            asyncio.run(
+                manager.update_status(
+                    SessionStatusSpec(
+                        route=route,
+                        kind="mirror",
+                        label="ttys039",
+                        title="Terminal",
+                        state="stopped",
+                        step="session ended",
+                        summary="Mirror source disappeared.",
+                    )
+                )
+            )
+
+            pending = manager._state["sessions"][route.label].get("_pending_topic_delete")
+            self.assertIsNotNone(pending)
+            self.assertEqual(pending["topic_id"], 42)
+            self.assertEqual(pending["status_message_id"], 11)
+
+    def test_flush_stopped_session_cleanup_deletes_managed_topic_and_clears_session(self) -> None:
+        with tempfile.TemporaryDirectory() as temp_dir:
+            settings = make_settings(Path(temp_dir))
+            bot = FakeBot()
+            manager = TelegramConsoleManager(settings)
+            manager._application = FakeApplication(bot=bot)
+            manager.set_forum_enabled(True)
+            manager._state["sessions"]["mirror:ttys039"] = {
+                "route_kind": "mirror",
+                "route_target": "ttys039",
+                "label": "ttys039",
+                "kind": "mirror",
+                "state": "stopped",
+                "_pending_topic_delete": {
+                    "route_kind": "mirror",
+                    "route_target": "ttys039",
+                    "topic_id": 42,
+                    "topic_managed": True,
+                    "chat_id": -1001,
+                    "status_message_id": 11,
+                    "topic_bump_message_id": 12,
+                    "last_event_message_id": 13,
+                    "label": "ttys039",
+                },
+            }
+
+            handled = asyncio.run(manager._flush_stopped_session_cleanup())
+
+            self.assertTrue(handled)
+            self.assertNotIn("mirror:ttys039", manager._state["sessions"])
+            self.assertEqual(
+                bot.calls,
+                [
+                    ("delete_message", {"chat_id": -1001, "message_id": 11}),
+                    ("delete_message", {"chat_id": -1001, "message_id": 12}),
+                    ("delete_message", {"chat_id": -1001, "message_id": 13}),
+                    ("delete_forum_topic", {"chat_id": -1001, "message_thread_id": 42}),
+                ],
+            )
+
 
 if __name__ == "__main__":
     unittest.main()