From 6e94ed035162d1a08050d3c6ef5d718937942265 Mon Sep 17 00:00:00 2001
From: sunchan-park <sunchan.dev@gmail.com>
Date: Sat, 7 Mar 2026 20:58:46 +0900
Subject: [PATCH] fix: adaptive throttle for timer-based status updates to
 avoid Telegram rate limits
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

When Claude Code waits on a long-running task (e.g. agent sub-tasks),
the status line timer increments every second ("Thinking… 1s", "2s", …).
The status poller detects each tick as a text change and enqueues a
Telegram edit_message_text call — roughly 60/min.  This quickly exceeds
Telegram's rate limit, triggering 429 responses and a flood-control ban
that silences the bot for extended periods.

Add _should_send_status() in status_polling.py that detects timer
suffixes via regex and applies an adaptive send interval based on how
long the same status has been active:

  0–10 s  → every 1 s  (real-time, useful for short tasks)
  10–60 s → every 5 s
  60 s+   → every 30 s

Key design decisions:
- Poll interval stays at 1 s — interactive UI detection is unaffected
- Non-timer status changes always send immediately (no delay)
- Timer detection uses a trailing regex matching "5s", "1m 30s", etc.
- Intervals are configurable via STATUS_THROTTLE_INTERVALS env var
  (comma-separated, e.g. "1,5,30"; set "1,1,1" to disable)

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
---
 src/ccbot/config.py                         |  20 ++
 src/ccbot/handlers/status_polling.py        |  92 ++++++-
 tests/ccbot/handlers/test_timer_throttle.py | 258 ++++++++++++++++++++
 tests/ccbot/test_config.py                  |  33 +++
 4 files changed, 401 insertions(+), 2 deletions(-)
 create mode 100644 tests/ccbot/handlers/test_timer_throttle.py

diff --git a/src/ccbot/config.py b/src/ccbot/config.py
index ca3d6744..1e77d865 100644
--- a/src/ccbot/config.py
+++ b/src/ccbot/config.py
@@ -84,6 +84,26 @@ def __init__(self) -> None:
 
         self.monitor_poll_interval = float(os.getenv("MONITOR_POLL_INTERVAL", "2.0"))
 
+        # Adaptive throttle for timer-based status lines.
+        # Three comma-separated intervals (seconds) for elapsed-time tiers:
+        #   tier 1 (0–10 s), tier 2 (10–60 s), tier 3 (60 s+)
+        # Default "1,5,30" means: real-time for first 10 s, then every 5 s,
+        # then every 30 s.  Set to "1,1,1" to disable throttling.
+        self.status_throttle_intervals: tuple[float, float, float] = (1.0, 5.0, 30.0)
+        raw = os.getenv("STATUS_THROTTLE_INTERVALS", "")
+        if raw:
+            try:
+                parts = [float(x.strip()) for x in raw.split(",")]
+                if len(parts) != 3 or any(p < 0 for p in parts):
+                    raise ValueError("need exactly 3 non-negative numbers")
+                self.status_throttle_intervals = (parts[0], parts[1], parts[2])
+            except ValueError as e:
+                logger.warning(
+                    "Invalid STATUS_THROTTLE_INTERVALS=%r (%s), using defaults",
+                    raw,
+                    e,
+                )
+
         # Display user messages in history and real-time notifications
         # When True, user messages are shown with a 👤 prefix
         self.show_user_messages = True
diff --git a/src/ccbot/handlers/status_polling.py b/src/ccbot/handlers/status_polling.py
index c4de1c6e..602631a2 100644
--- a/src/ccbot/handlers/status_polling.py
+++ b/src/ccbot/handlers/status_polling.py
@@ -8,21 +8,25 @@
   - Periodically probes topic existence via unpin_all_forum_topic_messages
     (silent no-op when no pins); cleans up deleted topics (kills tmux window
     + unbinds thread)
+  - Adaptive throttle for timer-based status lines to avoid Telegram rate limits
 
 Key components:
   - STATUS_POLL_INTERVAL: Polling frequency (1 second)
   - TOPIC_CHECK_INTERVAL: Topic existence probe frequency (60 seconds)
   - status_poll_loop: Background polling task
   - update_status_message: Poll and enqueue status updates
+  - _should_send_status: Adaptive throttle for timer status updates
 """
 
 import asyncio
 import logging
+import re
 import time
 
 from telegram import Bot
 from telegram.error import BadRequest
 
+from ..config import config
 from ..session import session_manager
 from ..terminal_parser import is_interactive_ui, parse_status_line
 from ..tmux_manager import tmux_manager
@@ -36,12 +40,94 @@
 
 logger = logging.getLogger(__name__)
 
-# Status polling interval
-STATUS_POLL_INTERVAL = 1.0  # seconds - faster response (rate limiting at send layer)
+# Status polling interval — kept at 1s for fast interactive UI detection.
+# Timer-based status updates are throttled adaptively by _should_send_status().
+STATUS_POLL_INTERVAL = 1.0  # seconds
 
 # Topic existence probe interval
 TOPIC_CHECK_INTERVAL = 60.0  # seconds
 
+# ── Adaptive throttle for timer status lines ─────────────────────────────
+#
+# Claude Code shows a running timer in the status line (e.g. "Thinking… 5s",
+# "Bash echo hello 1m 30s").  Without throttling, every tick produces a
+# Telegram edit_message_text call (~60/min), which quickly hits Telegram's
+# rate limit and causes the bot to go silent for extended periods.
+#
+# Strategy: detect timer suffixes, then increase the update interval the
+# longer the same status persists.  Default tiers (configurable via
+# STATUS_THROTTLE_INTERVALS env var, comma-separated):
+#   0–10 s  →  every 1 s   (real-time, meaningful for short tasks)
+#   10–60 s →  every 5 s
+#   60 s+   →  every 30 s
+#
+# Non-timer status changes (e.g. "Reading file" → "Writing file") always
+# send immediately.  The poll interval itself stays at 1 s so interactive
+# UI detection is never delayed.
+
+# Matches a timer in the status line.  Claude Code uses two formats:
+#   1. Bare suffix:        "Thinking… 5s"  or  "Bash echo hello 1m 30s"
+#   2. Inside parentheses: "Drizzling… (54s · ↓ 776 tokens)"
+#                          "Coalescing… (25m 8s · ↓ 5.8k tokens · thought for 6s)"
+# Both may include optional trailing text (parenthetical or · metadata).
+_TIMER_RE = re.compile(
+    r"\s+(?:"
+    r"\((?:\d+m\s*)?\d+s\b[^)]*\)"   # (54s · …) or (1m 30s · …)
+    r"|"
+    r"(?:(?:\d+m\s*)?\d+s|\d+m)"      # bare 5s / 1m 30s / 2m
+    r"(?:\s+\(.*\))?"                  # optional trailing (Esc to interrupt)
+    r")\s*$"
+)
+
+# (user_id, thread_id_or_0) → (base_text, first_seen, last_sent)
+_timer_throttle: dict[tuple[int, int], tuple[str, float, float]] = {}
+
+
+def _should_send_status(
+    user_id: int, thread_id: int | None, status_text: str
+) -> bool:
+    """Decide whether a status update should be enqueued.
+
+    For non-timer status lines, always returns True.
+    For timer status lines, applies adaptive interval based on elapsed time.
+    """
+    key = (user_id, thread_id or 0)
+    now = time.monotonic()
+
+    m = _TIMER_RE.search(status_text)
+    if not m:
+        # Not a timer — always send, clear any tracked state
+        _timer_throttle.pop(key, None)
+        return True
+
+    # Extract base text (everything before the timer suffix)
+    base = status_text[: m.start()].rstrip()
+
+    prev = _timer_throttle.get(key)
+    if prev is None or prev[0] != base:
+        # New status or base text changed — reset and send immediately
+        _timer_throttle[key] = (base, now, now)
+        return True
+
+    _, first_seen, last_sent = prev
+    elapsed = now - first_seen
+    since_sent = now - last_sent
+
+    # Adaptive interval: widen as the timer runs longer
+    t1, t2, t3 = config.status_throttle_intervals
+    if elapsed <= 10:
+        min_interval = t1    # real-time for the first 10 seconds
+    elif elapsed <= 60:
+        min_interval = t2    # every few seconds up to 1 minute
+    else:
+        min_interval = t3    # reduced frequency for long-running tasks
+
+    if since_sent >= min_interval:
+        _timer_throttle[key] = (base, first_seen, now)
+        return True
+
+    return False
+
 
 async def update_status_message(
     bot: Bot,
@@ -109,6 +195,8 @@ async def update_status_message(
     status_line = parse_status_line(pane_text)
 
     if status_line:
+        if not _should_send_status(user_id, thread_id, status_line):
+            return
         await enqueue_status_update(
             bot,
             user_id,
diff --git a/tests/ccbot/handlers/test_timer_throttle.py b/tests/ccbot/handlers/test_timer_throttle.py
new file mode 100644
index 00000000..b3d46a62
--- /dev/null
+++ b/tests/ccbot/handlers/test_timer_throttle.py
@@ -0,0 +1,258 @@
+"""Tests for adaptive timer throttle in status_polling.
+
+Verifies that _should_send_status() correctly:
+  - Always passes non-timer status lines
+  - Sends timer updates every poll for the first 10 s
+  - Throttles to ~5 s intervals between 10–60 s
+  - Throttles to ~30 s intervals after 60 s
+  - Resets when base text changes (different task)
+  - Resets when status switches from timer to non-timer
+"""
+
+from unittest.mock import patch
+
+import pytest
+
+from ccbot.handlers.status_polling import (
+    _TIMER_RE,
+    _should_send_status,
+    _timer_throttle,
+)
+
+
+@pytest.fixture(autouse=True)
+def _clear_throttle_state():
+    """Ensure throttle state is clean for each test."""
+    _timer_throttle.clear()
+    yield
+    _timer_throttle.clear()
+
+
+# ── Regex tests ──────────────────────────────────────────────────────────
+
+
+class TestTimerRegex:
+    """Verify _TIMER_RE matches expected timer suffixes."""
+
+    @pytest.mark.parametrize(
+        "text",
+        [
+            "Thinking… 5s",
+            "Reading file.py 12s",
+            "Bash echo hello 1m 30s",
+            "Bash echo hello 1m30s",
+            "Working 2m",
+            "Thinking… 5s (Esc to interrupt)",
+            "Bash ls 1m 30s (Esc to interrupt)",
+            "Drizzling… (54s · ↓ 776 tokens)",
+            "Coalescing… (25m 8s · ↓ 5.8k tokens · thought for 6s)",
+            "Thinking… (3s)",
+            "Thinking… (1m 5s · ↓ 12k tokens · thought for 3s)",
+        ],
+    )
+    def test_timer_detected(self, text: str):
+        assert _TIMER_RE.search(text) is not None
+
+    @pytest.mark.parametrize(
+        "text",
+        [
+            "Reading file.py",
+            "Writing to output",
+            "Bash echo hello",
+            "file2s",           # no preceding space
+            "Processing items",
+        ],
+    )
+    def test_non_timer_not_matched(self, text: str):
+        assert _TIMER_RE.search(text) is None
+
+    def test_base_text_extraction(self):
+        text = "Thinking… 45s (Esc to interrupt)"
+        m = _TIMER_RE.search(text)
+        assert m is not None
+        base = text[: m.start()].rstrip()
+        assert base == "Thinking…"
+
+    def test_base_text_bash(self):
+        text = "Bash echo hello 1m 30s"
+        m = _TIMER_RE.search(text)
+        assert m is not None
+        base = text[: m.start()].rstrip()
+        assert base == "Bash echo hello"
+
+    def test_base_text_parenthesized_timer(self):
+        text = "Coalescing… (25m 8s · ↓ 5.8k tokens · thought for 6s)"
+        m = _TIMER_RE.search(text)
+        assert m is not None
+        base = text[: m.start()].rstrip()
+        assert base == "Coalescing…"
+
+    def test_base_text_short_parenthesized(self):
+        text = "Drizzling… (54s · ↓ 776 tokens)"
+        m = _TIMER_RE.search(text)
+        assert m is not None
+        base = text[: m.start()].rstrip()
+        assert base == "Drizzling…"
+
+
+# ── Throttle logic tests ────────────────────────────────────────────────
+
+
+class TestShouldSendStatus:
+    """Test adaptive throttle intervals."""
+
+    def test_non_timer_always_passes(self):
+        assert _should_send_status(1, 42, "Reading file.py") is True
+        assert _should_send_status(1, 42, "Reading file.py") is True
+        assert _should_send_status(1, 42, "Writing output") is True
+
+    def test_first_timer_always_passes(self):
+        assert _should_send_status(1, 42, "Thinking… 1s") is True
+
+    def test_realtime_within_10s(self):
+        """Timer updates within first 10s should all pass (1s interval)."""
+        t0 = 1000.0
+        with patch("ccbot.handlers.status_polling.time") as mock_time:
+            mock_time.monotonic.return_value = t0
+            assert _should_send_status(1, 42, "Thinking… 1s") is True
+
+            # 2 seconds later — still within 10s window
+            mock_time.monotonic.return_value = t0 + 2
+            assert _should_send_status(1, 42, "Thinking… 3s") is True
+
+            # 1 second later — also passes
+            mock_time.monotonic.return_value = t0 + 3
+            assert _should_send_status(1, 42, "Thinking… 4s") is True
+
+    def test_throttled_after_10s(self):
+        """After 10s elapsed, interval becomes 5s."""
+        t0 = 1000.0
+        with patch("ccbot.handlers.status_polling.time") as mock_time:
+            # First call
+            mock_time.monotonic.return_value = t0
+            assert _should_send_status(1, 42, "Thinking… 1s") is True
+
+            # At 11s — first_seen=t0, elapsed=11s, since_sent=11s → passes (>= 5s)
+            mock_time.monotonic.return_value = t0 + 11
+            assert _should_send_status(1, 42, "Thinking… 12s") is True
+
+            # At 13s — since_sent=2s → blocked (< 5s)
+            mock_time.monotonic.return_value = t0 + 13
+            assert _should_send_status(1, 42, "Thinking… 14s") is False
+
+            # At 16s — since_sent=5s → passes (>= 5s)
+            mock_time.monotonic.return_value = t0 + 16
+            assert _should_send_status(1, 42, "Thinking… 17s") is True
+
+    def test_throttled_after_60s(self):
+        """After 60s elapsed, interval becomes 30s."""
+        t0 = 1000.0
+        with patch("ccbot.handlers.status_polling.time") as mock_time:
+            mock_time.monotonic.return_value = t0
+            assert _should_send_status(1, 42, "Thinking… 1s") is True
+
+            # At 61s — passes (since_sent=61s >= 30s)
+            mock_time.monotonic.return_value = t0 + 61
+            assert _should_send_status(1, 42, "Thinking… 1m 2s") is True
+
+            # At 70s — blocked (since_sent=9s < 30s)
+            mock_time.monotonic.return_value = t0 + 70
+            assert _should_send_status(1, 42, "Thinking… 1m 11s") is False
+
+            # At 91s — passes (since_sent=30s >= 30s)
+            mock_time.monotonic.return_value = t0 + 91
+            assert _should_send_status(1, 42, "Thinking… 1m 32s") is True
+
+    def test_base_text_change_resets(self):
+        """Changing base text (different task) resets the throttle."""
+        t0 = 1000.0
+        with patch("ccbot.handlers.status_polling.time") as mock_time:
+            mock_time.monotonic.return_value = t0
+            assert _should_send_status(1, 42, "Thinking… 1s") is True
+
+            # Jump to 65s (would be in 30s throttle zone)
+            mock_time.monotonic.return_value = t0 + 65
+            # Same base text — just record it as sent
+            assert _should_send_status(1, 42, "Thinking… 1m 6s") is True
+
+            # 2s later, different base text — should pass immediately
+            mock_time.monotonic.return_value = t0 + 67
+            assert _should_send_status(1, 42, "Bash echo hello 2s") is True
+
+    def test_non_timer_clears_state(self):
+        """Switching to non-timer status clears throttle state."""
+        t0 = 1000.0
+        with patch("ccbot.handlers.status_polling.time") as mock_time:
+            mock_time.monotonic.return_value = t0
+            assert _should_send_status(1, 42, "Thinking… 1s") is True
+            assert (1, 42) in _timer_throttle
+
+            # Non-timer clears state
+            mock_time.monotonic.return_value = t0 + 5
+            assert _should_send_status(1, 42, "Reading file.py") is True
+            assert (1, 42) not in _timer_throttle
+
+    def test_independent_per_user_thread(self):
+        """Throttle state is independent per (user_id, thread_id)."""
+        t0 = 1000.0
+        with patch("ccbot.handlers.status_polling.time") as mock_time:
+            mock_time.monotonic.return_value = t0
+            assert _should_send_status(1, 42, "Thinking… 1s") is True
+            assert _should_send_status(1, 99, "Thinking… 1s") is True
+            assert _should_send_status(2, 42, "Thinking… 1s") is True
+
+            # Jump to 15s zone (5s interval)
+            mock_time.monotonic.return_value = t0 + 15
+            assert _should_send_status(1, 42, "Thinking… 16s") is True
+
+            # 2s later — user 1/thread 42 blocked, but user 2/thread 42 passes
+            mock_time.monotonic.return_value = t0 + 17
+            assert _should_send_status(1, 42, "Thinking… 18s") is False
+            assert _should_send_status(2, 42, "Thinking… 18s") is True
+
+    def test_none_thread_id(self):
+        """thread_id=None is treated as 0."""
+        assert _should_send_status(1, None, "Thinking… 1s") is True
+        assert (1, 0) in _timer_throttle
+
+    def test_custom_intervals_from_config(self):
+        """Intervals are read from config.status_throttle_intervals."""
+        t0 = 1000.0
+        # Set aggressive throttle: 2s / 10s / 60s
+        with (
+            patch("ccbot.handlers.status_polling.time") as mock_time,
+            patch(
+                "ccbot.handlers.status_polling.config.status_throttle_intervals",
+                (2.0, 10.0, 60.0),
+            ),
+        ):
+            mock_time.monotonic.return_value = t0
+            assert _should_send_status(1, 42, "Thinking… 1s") is True
+
+            # At t0+1 — within tier 1 (0-10s), but interval=2s → blocked
+            mock_time.monotonic.return_value = t0 + 1
+            assert _should_send_status(1, 42, "Thinking… 2s") is False
+
+            # At t0+2 — since_sent=2s >= 2s → passes
+            mock_time.monotonic.return_value = t0 + 2
+            assert _should_send_status(1, 42, "Thinking… 3s") is True
+
+    def test_disabled_throttle(self):
+        """Setting all intervals to 1 effectively disables throttling."""
+        t0 = 1000.0
+        with (
+            patch("ccbot.handlers.status_polling.time") as mock_time,
+            patch(
+                "ccbot.handlers.status_polling.config.status_throttle_intervals",
+                (1.0, 1.0, 1.0),
+            ),
+        ):
+            mock_time.monotonic.return_value = t0
+            assert _should_send_status(1, 42, "Thinking… 1s") is True
+
+            # At 65s elapsed, interval is still 1s → passes after 1s
+            mock_time.monotonic.return_value = t0 + 65
+            assert _should_send_status(1, 42, "Thinking… 1m 6s") is True
+
+            mock_time.monotonic.return_value = t0 + 66
+            assert _should_send_status(1, 42, "Thinking… 1m 7s") is True
diff --git a/tests/ccbot/test_config.py b/tests/ccbot/test_config.py
index 95cf35f9..ad5089e1 100644
--- a/tests/ccbot/test_config.py
+++ b/tests/ccbot/test_config.py
@@ -117,3 +117,36 @@ def test_openai_api_key_scrubbed_from_env(self, monkeypatch):
         monkeypatch.setenv("OPENAI_API_KEY", "sk-secret")
         Config()
         assert os.environ.get("OPENAI_API_KEY") is None
+
+
+@pytest.mark.usefixtures("_base_env")
+class TestConfigStatusThrottle:
+    def test_default_intervals(self, monkeypatch):
+        monkeypatch.delenv("STATUS_THROTTLE_INTERVALS", raising=False)
+        cfg = Config()
+        assert cfg.status_throttle_intervals == (1.0, 5.0, 30.0)
+
+    def test_custom_intervals(self, monkeypatch):
+        monkeypatch.setenv("STATUS_THROTTLE_INTERVALS", "2,10,60")
+        cfg = Config()
+        assert cfg.status_throttle_intervals == (2.0, 10.0, 60.0)
+
+    def test_disable_throttle(self, monkeypatch):
+        monkeypatch.setenv("STATUS_THROTTLE_INTERVALS", "1,1,1")
+        cfg = Config()
+        assert cfg.status_throttle_intervals == (1.0, 1.0, 1.0)
+
+    def test_invalid_format_uses_defaults(self, monkeypatch):
+        monkeypatch.setenv("STATUS_THROTTLE_INTERVALS", "bad,input")
+        cfg = Config()
+        assert cfg.status_throttle_intervals == (1.0, 5.0, 30.0)
+
+    def test_wrong_count_uses_defaults(self, monkeypatch):
+        monkeypatch.setenv("STATUS_THROTTLE_INTERVALS", "1,5")
+        cfg = Config()
+        assert cfg.status_throttle_intervals == (1.0, 5.0, 30.0)
+
+    def test_negative_value_uses_defaults(self, monkeypatch):
+        monkeypatch.setenv("STATUS_THROTTLE_INTERVALS", "1,-5,30")
+        cfg = Config()
+        assert cfg.status_throttle_intervals == (1.0, 5.0, 30.0)