fix(resume-settle): unfreeze bot during restore + centralize TYPING throttle

src/ccbot/bot/_session_create.py

@@ -86,10 +86,15 @@ async def create_and_activate_session(
     # SessionStart hook. The hook is confirmed (and the fresh session_id
     # bound) after the paint, before any pending text is forwarded.
     if resume_session_id:
-        # A near-limit transcript auto-compacts on resume (60-110s); flag the
-        # window so the held pending text (and any first message) waits for
-        # the pane to settle instead of being typed mid-compaction and lost.
-        session_manager.mark_window_resuming(created_wid)
+        # A near-limit transcript auto-compacts on resume (60-110s); flag
+        # the window so any prompt that arrives while we're still
+        # compacting buffers into _pending_sends instead of being typed
+        # mid-compaction. The background watcher drains the buffer after
+        # the pane settles AND refreshes Telegram TYPING in the meantime
+        # so the chat doesn't look frozen.
+        session_manager.mark_window_resuming(
+            created_wid, bot=context.bot, user_id=user.id
+        )
         hook_ok = await session_manager.wait_for_session_map_entry(
             created_wid, timeout=15.0
         )

src/ccbot/bot/messages.py

@@ -20,7 +20,6 @@
 from typing import Any
 
 from telegram import Bot, Update
-from telegram.constants import ChatAction
 from telegram.error import BadRequest
 from telegram.ext import ContextTypes
 
@@ -53,6 +52,7 @@
     repost_card,
     resume_card_view,
 )
+from ..handlers.typing import fire_typing
 from ..session_models import Session
 from ..handlers.inbox import save_inbox_file
 from ..markdown_v2 import convert_markdown
@@ -232,18 +232,7 @@ async def forward_command_handler(
     logger.info(
         "Forwarding command %s to window %s (user=%d)", cc_slash, display, user.id
     )
-    await update.message.chat.send_action(ChatAction.TYPING)
-    logger.info(
-        "typing_fired source=forward_command user=%d wid=%s",
-        user.id,
-        wid,
-        extra={
-            "event": "typing_fired",
-            "source": "forward_command",
-            "user_id": user.id,
-            "window_id": wid,
-        },
-    )
+    await fire_typing(context.bot, user.id, "forward_command", window_id=wid)
     if await _intercept_if_pending_ui(context.bot, user.id, wid, update.message):
         return
     sess = session_manager.find_session_by_window(wid)
@@ -380,18 +369,7 @@ async def unsupported_content_handler(
             body_parts.extend(hidden_urls)
         text_to_send = "\n".join(body_parts)
 
-        await msg.chat.send_action(ChatAction.TYPING)
-        logger.info(
-            "typing_fired source=caption_forward user=%d wid=%s",
-            user.id,
-            wid,
-            extra={
-                "event": "typing_fired",
-                "source": "caption_forward",
-                "user_id": user.id,
-                "window_id": wid,
-            },
-        )
+        await fire_typing(context.bot, user.id, "caption_forward", window_id=wid)
         if await _intercept_if_pending_ui(context.bot, user.id, wid, msg):
             return
         sess = session_manager.find_session_by_window(wid)
@@ -445,23 +423,7 @@ async def _forward_inbox_file(
     else:
         rel_path = str(file_path)
     text_to_send = f"{caption}\n\n{rel_path}" if caption.strip() else rel_path
-    try:
-        await bot.send_chat_action(chat_id=chat_id, action=ChatAction.TYPING)
-        logger.info(
-            "typing_fired source=inbox_file_forward user=%d wid=%s label=%s",
-            user_id,
-            wid,
-            label,
-            extra={
-                "event": "typing_fired",
-                "source": "inbox_file_forward",
-                "user_id": user_id,
-                "window_id": wid,
-                "label": label,
-            },
-        )
-    except Exception:
-        pass
+    await fire_typing(bot, user_id, "inbox_file_forward", window_id=wid, label=label)
     return await session_manager.send_to_window(wid, text_to_send)
 
 
@@ -638,18 +600,7 @@ async def voice_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> N
         await safe_reply(update.message, f"⚠ Transcription failed: {e}")
         return
 
-    await update.message.chat.send_action(ChatAction.TYPING)
-    logger.info(
-        "typing_fired source=voice_handler user=%d wid=%s",
-        user.id,
-        wid,
-        extra={
-            "event": "typing_fired",
-            "source": "voice_handler",
-            "user_id": user.id,
-            "window_id": wid,
-        },
-    )
+    await fire_typing(context.bot, user.id, "voice_handler", window_id=wid)
 
     if await _intercept_if_pending_ui(context.bot, user.id, wid, update.message):
         return
@@ -919,24 +870,10 @@ async def _dispatch_text_to_active(
         # its first event (long tool prelude / thinking) and
         # ``status_polling`` won't fire typing until the pane enters
         # the busy-spinner state. Without this early fire the chat
-        # looks frozen.
-        try:
-            await context.bot.send_chat_action(
-                chat_id=user_id, action=ChatAction.TYPING
-            )
-            logger.info(
-                "typing_fired source=text_handler.post_send user=%d wid=%s",
-                user_id,
-                wid,
-                extra={
-                    "event": "typing_fired",
-                    "source": "text_handler.post_send",
-                    "user_id": user_id,
-                    "window_id": wid,
-                },
-            )
-        except Exception:
-            pass
+        # looks frozen. fire_typing throttles to one call per ~4 s
+        # per user — if text_handler already fired Typing a moment
+        # ago, this is a silent no-op (the indicator is still on).
+        await fire_typing(context.bot, user_id, "text_handler.post_send", window_id=wid)
 
         sess = session_manager.find_session_by_window(wid)
         if sess is not None:
@@ -1001,18 +938,7 @@ async def text_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> No
     if wid is None:
         return
 
-    await update.message.chat.send_action(ChatAction.TYPING)
-    logger.info(
-        "typing_fired source=text_handler user=%d wid=%s",
-        user.id,
-        wid,
-        extra={
-            "event": "typing_fired",
-            "source": "text_handler",
-            "user_id": user.id,
-            "window_id": wid,
-        },
-    )
+    await fire_typing(context.bot, user.id, "text_handler", window_id=wid)
 
     # New message pushes pane content down — kill any in-flight bash capture.
     cancel_bash_capture(user.id, wid)

src/ccbot/bot/session_events.py

@@ -20,7 +20,6 @@
 from pathlib import Path
 
 from telegram import Bot
-from telegram.constants import ChatAction
 
 from ..config import config
 from ..handlers import bg_status
@@ -35,6 +34,7 @@
     refresh_panel,
     update_session_card,
 )
+from ..handlers.typing import fire_typing
 from ..session import session_manager
 from ..session_monitor import NewMessage
 from ..terminal_parser import extract_interactive_content
@@ -96,23 +96,13 @@ async def handle_new_message(msg: NewMessage, bot: Bot) -> None:
         # once events stop. Bg sessions skip — they don't surface in
         # the chat header.
         if is_active:
-            try:
-                await bot.send_chat_action(chat_id=user_id, action=ChatAction.TYPING)
-                logger.info(
-                    "typing_fired source=session_events user=%d sess=%s ctype=%s",
-                    user_id,
-                    sess.id,
-                    msg.content_type,
-                    extra={
-                        "event": "typing_fired",
-                        "source": "session_events",
-                        "user_id": user_id,
-                        "session_id": sess.id,
-                        "content_type": msg.content_type,
-                    },
-                )
-            except Exception as e:
-                logger.debug("send_chat_action TYPING failed: %s", e)
+            await fire_typing(
+                bot,
+                user_id,
+                "session_events",
+                session_id=sess.id,
+                content_type=msg.content_type,
+            )
 
         if not config.show_tool_calls and msg.content_type in (
             "tool_use",

src/ccbot/handlers/archive.py

@@ -306,7 +306,6 @@ async def restore_session(bot: Bot, user_id: int, sess: Session) -> tuple[bool,
 
     Returns (success, message).
     """
-    del bot  # unused for now; reserved for future logging
     if sess.state in ("active", "idle"):
         return False, "Session already live"
     workdir = sess.workdir or ""
@@ -321,10 +320,13 @@ async def restore_session(bot: Bot, user_id: int, sess: Session) -> tuple[bool,
         return False, message
 
     # A near-limit transcript auto-compacts on resume (60-110s) before it
-    # accepts input. Flag the window so the first message waits the pane out
-    # instead of getting typed into a busy pane and dropped.
+    # accepts input. Flag the window so any prompts that arrive while
+    # we're still compacting buffer into _pending_sends instead of being
+    # typed mid-compaction. The background watcher drains the buffer
+    # once the pane settles AND keeps Telegram TYPING refreshed so the
+    # chat doesn't look frozen during the wait.
     if sess.claude_session_id:
-        session_manager.mark_window_resuming(created_wid)
+        session_manager.mark_window_resuming(created_wid, bot=bot, user_id=user_id)
 
     hook_ok = await session_manager.wait_for_session_map_entry(
         created_wid, timeout=15.0

src/ccbot/handlers/status_polling.py

@@ -21,7 +21,6 @@
 from typing import TYPE_CHECKING
 
 from telegram import Bot
-from telegram.constants import ChatAction
 
 from ..config import config
 from ..session import session_manager
@@ -50,6 +49,7 @@
     maybe_finalize_stalled,
     refresh_panel,
 )
+from .typing import fire_typing
 
 # Match option lines like "  1. Yes" / " ❯ 2. Yes, and don't ask again".
 _OPTION_LINE_RE = re.compile(r"^[\s❯>]*?(\d+)\.\s+(.+?)\s*$")
@@ -350,30 +350,16 @@ async def _drive_typing_indicator(
         )
 
     if not is_bg_session and not in_menu and (card_busy or pane_busy):
-        try:
-            await bot.send_chat_action(chat_id=user_id, action=ChatAction.TYPING)
-            logger.info(
-                "typing_fired source=status_polling user=%d sess=%s wid=%s "
-                "card_busy=%s pane_busy=%s status=%r",
-                user_id,
-                sess.id if sess else "-",
-                window_id,
-                card_busy,
-                pane_busy,
-                status_line[:40] if status_line else "",
-                extra={
-                    "event": "typing_fired",
-                    "source": "status_polling",
-                    "user_id": user_id,
-                    "session_id": sess.id if sess else None,
-                    "window_id": window_id,
-                    "card_busy": card_busy,
-                    "pane_busy": pane_busy,
-                    "status_line": status_line[:80] if status_line else "",
-                },
-            )
-        except Exception as e:
-            logger.debug("send_chat_action TYPING failed: %s", e)
+        await fire_typing(
+            bot,
+            user_id,
+            "status_polling",
+            session_id=sess.id if sess else None,
+            window_id=window_id,
+            card_busy=card_busy,
+            pane_busy=pane_busy,
+            status_line=status_line[:80] if status_line else "",
+        )
 
 
 async def update_status_message(

src/ccbot/handlers/typing.py

@@ -0,0 +1,78 @@
+"""Per-user throttle on send_chat_action(TYPING).
+
+Telegram's typing indicator stays visible for ~5 s after one
+``send_chat_action`` call. Two of our call sites — ``status_polling``
+(every 1 s while a session is busy) and ``session_events`` (every
+inbound claude event) — used to re-fire the action many times per
+second during heavy tool sequences. Each call counts toward
+Telegram's per-chat rate budget and was a measurable contributor to
+the 429 ``Retry after 71 s`` bans seen during heavy multi-session
+usage.
+
+``fire_typing`` collapses all callers behind a single per-user
+timestamp. A call within ``TYPING_REFRESH_INTERVAL`` of the last
+successful fire for the same user is a silent no-op — the indicator
+is still on, no API call needed.
+"""
+
+from __future__ import annotations
+
+import logging
+import time
+from typing import Any
+
+from telegram import Bot
+from telegram.constants import ChatAction
+
+logger = logging.getLogger(__name__)
+
+# Telegram refreshes the indicator on every chat-action; one call
+# keeps it visible for ~5 s. We refresh at 4 s so the indicator
+# stays solid for a steadily-emitting session, but every call within
+# 4 s of the last one is dropped — that's the entire point.
+TYPING_REFRESH_INTERVAL = 4.0
+
+_last_fired: dict[int, float] = {}
+
+
+async def fire_typing(
+    bot: Bot,
+    user_id: int,
+    source: str,
+    **extra: Any,
+) -> bool:
+    """Fire ``send_chat_action(TYPING)`` for ``user_id``, throttled.
+
+    Returns ``True`` iff the chat-action was actually sent. Calls
+    within ``TYPING_REFRESH_INTERVAL`` of the last successful fire
+    for this user are dropped silently (the indicator is still
+    showing — there's nothing to log).
+
+    ``source`` and ``**extra`` are stamped onto the structured
+    ``typing_fired`` log record for parity with the prior per-site
+    logging.
+    """
+    now = time.monotonic()
+    last = _last_fired.get(user_id, 0.0)
+    if now - last < TYPING_REFRESH_INTERVAL:
+        return False
+    try:
+        await bot.send_chat_action(chat_id=user_id, action=ChatAction.TYPING)
+    except Exception as e:
+        logger.debug("send_chat_action TYPING failed: %s", e)
+        return False
+    _last_fired[user_id] = now
+    pretty = " ".join(f"{k}={v}" for k, v in extra.items())
+    logger.info(
+        "typing_fired source=%s user=%d %s",
+        source,
+        user_id,
+        pretty,
+        extra={
+            "event": "typing_fired",
+            "source": source,
+            "user_id": user_id,
+            **extra,
+        },
+    )
+    return True