fix(auto-approve): escalate to manual keyboard when auto-Yes doesn't clear the prompt

src/ccbot/handlers/status_polling.py

@@ -66,6 +66,28 @@
 KB_CLEAR_CONFIRM_POLLS = 2
 _kb_clear_miss: dict[tuple[int, str], int] = {}
 
+# Auto-approve escalation. ``auto_approve=on`` sends the ``N. Yes`` digit +
+# Enter into the pane, but the number-key shortcut does NOT clear every
+# blocking prompt — a sub-agent (``Agent``/``Task``) WebFetch/WebSearch
+# approval doesn't accept the top-level hotkey, so the keystroke lands on a
+# busy REPL and the prompt survives. Without a cap, the poller re-fires the
+# same auto-Yes every cycle forever AND (because ``_maybe_auto_approve``
+# returns True each time) it permanently shadows the kb-mode / bg-status
+# surface — leaving the session wedged with no manual escape hatch.
+#
+# Verification is cross-poll: each poll compares the prompt's signature to
+# the last one we auto-approved. Same signature next poll ⇒ the previous
+# keystroke did NOT clear it. After AUTO_APPROVE_MAX_ATTEMPTS consecutive
+# no-progress attempts on an identical prompt we STOP auto-approving and
+# return False, letting the caller surface the kb-mode keyboard (active
+# session) or the ❓ bg-status badge (background) so the user can drive it
+# by hand. A genuinely different prompt (new domain / file) has a different
+# signature ⇒ the counter resets, so normal one-shot approvals are
+# untouched.
+AUTO_APPROVE_MAX_ATTEMPTS = 3
+# (user_id, window_id) -> (prompt_signature, consecutive_attempts)
+_auto_approve_attempts: dict[tuple[int, str], tuple[str, int]] = {}
+
 
 def _parse_first_yes_option(pane_text: str) -> str | None:
     """First option line whose label starts with "Yes". Returns its number."""
@@ -78,27 +100,70 @@ def _parse_first_yes_option(pane_text: str) -> str | None:
     return None
 
 
+def _auto_approve_signature(pane_text: str) -> str:
+    """Stable identity of the on-screen prompt for attempt-tracking.
+
+    Prefers the extracted prompt body + name so distinct prompts (a
+    different fetch domain, a different file) don't share a counter, while
+    redraws of the SAME (spinner-free) permission dialog collapse to one
+    signature. Falls back to the numbered option block when extraction
+    returns None.
+    """
+    content = extract_interactive_content(pane_text)
+    if content is not None:
+        return f"{content.name}\x1f{content.content}"
+    opts = [
+        f"{m.group(1)}.{m.group(2)}"
+        for line in pane_text.splitlines()
+        if (m := _OPTION_LINE_RE.match(line))
+    ]
+    return "\n".join(opts)
+
+
 async def _maybe_auto_approve(user_id: int, window_id: str, pane_text: str) -> bool:
     """Auto-Yes on the in-pane Yes/No prompt when the user opted in.
 
     Returns True iff a key was sent — caller should then skip surfacing the
-    UI to TG.
+    UI to TG. Returns False (letting the caller surface the manual keyboard)
+    when the same prompt has survived ``AUTO_APPROVE_MAX_ATTEMPTS`` auto-Yes
+    keystrokes without clearing — see ``_auto_approve_attempts``.
     """
     mode = session_manager.get_user_settings(user_id).get("auto_approve", "off")
     if mode != "on":
         return False
     digit = _parse_first_yes_option(pane_text)
     if digit is None:
         return False
+
+    key = (user_id, window_id)
+    sig = _auto_approve_signature(pane_text)
+    prev_sig, attempts = _auto_approve_attempts.get(key, ("", 0))
+    attempts = attempts + 1 if sig == prev_sig else 1
+    if attempts > AUTO_APPROVE_MAX_ATTEMPTS:
+        # Give up: the prompt isn't responding to the auto-Yes hotkey.
+        # Keep the entry so we stay escalated (return False) until the
+        # prompt clears and ``_reconcile_no_ui_state`` pops the counter.
+        _auto_approve_attempts[key] = (sig, attempts)
+        logger.warning(
+            "auto_approve giving up after %d attempts (prompt not clearing) "
+            "for user=%d window=%s — surfacing manual keyboard",
+            AUTO_APPROVE_MAX_ATTEMPTS,
+            user_id,
+            window_id,
+        )
+        return False
+
     # Number-key shortcut: typing the digit picks the option, Enter submits.
     try:
         await tmux_manager.send_keys(window_id, digit, enter=True)
     except Exception as e:
         logger.debug("auto_approve send_keys failed: %s", e)
         return False
+    _auto_approve_attempts[key] = (sig, attempts)
     logger.info(
-        "Auto-approved interactive prompt (opt=%s) for user=%d window=%s",
+        "Auto-approved interactive prompt (opt=%s, attempt=%d) for user=%d window=%s",
         digit,
+        attempts,
         user_id,
         window_id,
     )
@@ -245,6 +310,7 @@ async def _surface_new_interactive_ui(
 async def _reconcile_no_ui_state(
     bot: Bot,
     user_id: int,
+    window_id: str,
     pane_text: str,
     sess: "Session | None",
     is_bg_session: bool,
@@ -254,10 +320,18 @@ async def _reconcile_no_ui_state(
     Clears a stale bg-session ❓ stash and flips the active card out of a
     dismissed slash-command kb-mode picker.
     """
+    no_ui = not is_interactive_ui(pane_text)
+
+    # The prompt is gone — reset the auto-approve escalation counter so the
+    # NEXT prompt on this window starts fresh (and a prior "gave up" state
+    # doesn't immediately re-escalate an unrelated future prompt).
+    if no_ui:
+        _auto_approve_attempts.pop((user_id, window_id), None)
+
     # No interactive UI on this pane right now. If we previously stashed
     # one for a bg session (e.g. claude dismissed the prompt without our
     # input), clear it so the ❓ badge doesn't lie.
-    if is_bg_session and sess is not None and not is_interactive_ui(pane_text):
+    if is_bg_session and sess is not None and no_ui:
         if bg_status.clear_pending_ui(user_id, sess.id):
             await refresh_panel(bot, user_id)
 
@@ -418,7 +492,9 @@ async def update_status_message(
         ):
             return
 
-    await _reconcile_no_ui_state(bot, user_id, pane_text, sess, is_bg_session)
+    await _reconcile_no_ui_state(
+        bot, user_id, window_id, pane_text, sess, is_bg_session
+    )
 
     await _drive_typing_indicator(
         bot, user_id, window_id, pane_text, sess, is_bg_session

tests/test_kb_mode_debounce.py

@@ -20,8 +20,11 @@
 import ccbot.handlers.notifications as notifications
 from ccbot.handlers import status_polling
 from ccbot.handlers.status_polling import (
+    AUTO_APPROVE_MAX_ATTEMPTS,
     KB_CLEAR_CONFIRM_POLLS,
+    _auto_approve_attempts,
     _kb_clear_miss,
+    _maybe_auto_approve,
     _reconcile_no_ui_state,
     _surface_new_interactive_ui,
 )
@@ -31,8 +34,10 @@
 @pytest.fixture(autouse=True)
 def _clear_streak():
     _kb_clear_miss.clear()
+    _auto_approve_attempts.clear()
     yield
     _kb_clear_miss.clear()
+    _auto_approve_attempts.clear()
 
 
 def _sess(sid: str = "s1") -> Session:
@@ -51,7 +56,9 @@ async def test_single_miss_does_not_clear_kb():
         patch.object(notifications, "has_pending_kb", lambda u, s: (True, True)),
         patch.object(notifications, "exit_kb_mode", exit_kb),
     ):
-        await _reconcile_no_ui_state(bot, 1, "", sess, is_bg_session=False)
+        await _reconcile_no_ui_state(
+            bot, 1, sess.window_id, "", sess, is_bg_session=False
+        )
 
     exit_kb.assert_not_called()
     assert _kb_clear_miss[(1, sess.id)] == 1
@@ -69,7 +76,9 @@ async def test_two_consecutive_misses_clear_kb():
         patch.object(notifications, "exit_kb_mode", exit_kb),
     ):
         for _ in range(KB_CLEAR_CONFIRM_POLLS):
-            await _reconcile_no_ui_state(bot, 1, "", sess, is_bg_session=False)
+            await _reconcile_no_ui_state(
+                bot, 1, sess.window_id, "", sess, is_bg_session=False
+            )
 
     exit_kb.assert_called_once_with(bot, 1, sess, clear_pending=True)
     # Streak reset after firing so a later prompt starts fresh.
@@ -89,17 +98,23 @@ async def test_pending_gone_resets_streak():
     ):
         # One miss with a pending prompt → streak 1.
         with patch.object(notifications, "has_pending_kb", lambda u, s: (True, True)):
-            await _reconcile_no_ui_state(bot, 1, "", sess, is_bg_session=False)
+            await _reconcile_no_ui_state(
+                bot, 1, sess.window_id, "", sess, is_bg_session=False
+            )
         assert _kb_clear_miss[(1, sess.id)] == 1
 
         # Prompt no longer pending → streak cleared.
         with patch.object(notifications, "has_pending_kb", lambda u, s: (False, False)):
-            await _reconcile_no_ui_state(bot, 1, "", sess, is_bg_session=False)
+            await _reconcile_no_ui_state(
+                bot, 1, sess.window_id, "", sess, is_bg_session=False
+            )
         assert (1, sess.id) not in _kb_clear_miss
 
         # A new prompt's first flicker is only miss #1, not an instant clear.
         with patch.object(notifications, "has_pending_kb", lambda u, s: (True, True)):
-            await _reconcile_no_ui_state(bot, 1, "", sess, is_bg_session=False)
+            await _reconcile_no_ui_state(
+                bot, 1, sess.window_id, "", sess, is_bg_session=False
+            )
 
     exit_kb.assert_not_called()
     assert _kb_clear_miss[(1, sess.id)] == 1
@@ -140,3 +155,99 @@ async def test_redetect_resets_miss_streak():
     assert handled is True
     enter_kb.assert_awaited_once()
     assert (1, sess.id) not in _kb_clear_miss  # streak reset on re-detect
+
+
+# --- auto-approve escalation (fix #2) -------------------------------------
+
+_YESNO_PANE = "  Do you want to proceed?\n❯ 1. Yes\n  2. No, tell Claude\n"
+
+
+def _content(name="PermissionPrompt", body="fetch example.com"):
+    obj = MagicMock()
+    obj.name = name
+    obj.content = body
+    return obj
+
+
+@pytest.mark.asyncio
+async def test_auto_approve_escalates_after_max_attempts():
+    """A prompt that survives the auto-Yes keystroke stops being re-pressed
+    after AUTO_APPROVE_MAX_ATTEMPTS and returns False so the caller can
+    surface the manual keyboard."""
+    send_keys = AsyncMock()
+    with (
+        patch.object(
+            status_polling.session_manager,
+            "get_user_settings",
+            lambda u: {"auto_approve": "on"},
+        ),
+        patch.object(status_polling.tmux_manager, "send_keys", send_keys),
+        patch.object(
+            status_polling, "extract_interactive_content", lambda p: _content()
+        ),
+    ):
+        # First MAX attempts: keystroke sent, returns True.
+        for i in range(AUTO_APPROVE_MAX_ATTEMPTS):
+            assert await _maybe_auto_approve(1, "@1", _YESNO_PANE) is True
+        assert send_keys.await_count == AUTO_APPROVE_MAX_ATTEMPTS
+        # Same prompt still up next poll → give up: no new keystroke, False.
+        assert await _maybe_auto_approve(1, "@1", _YESNO_PANE) is False
+        assert send_keys.await_count == AUTO_APPROVE_MAX_ATTEMPTS  # unchanged
+        # Stays escalated on subsequent polls.
+        assert await _maybe_auto_approve(1, "@1", _YESNO_PANE) is False
+        assert send_keys.await_count == AUTO_APPROVE_MAX_ATTEMPTS
+
+
+@pytest.mark.asyncio
+async def test_auto_approve_distinct_prompts_never_escalate():
+    """Different prompts (distinct signatures) each get a one-shot auto-Yes;
+    the counter resets per prompt so legitimate approvals aren't throttled."""
+    send_keys = AsyncMock()
+    bodies = iter([f"fetch host-{i}.com" for i in range(10)])
+    with (
+        patch.object(
+            status_polling.session_manager,
+            "get_user_settings",
+            lambda u: {"auto_approve": "on"},
+        ),
+        patch.object(status_polling.tmux_manager, "send_keys", send_keys),
+        patch.object(
+            status_polling,
+            "extract_interactive_content",
+            lambda p: _content(body=next(bodies)),
+        ),
+    ):
+        for _ in range(AUTO_APPROVE_MAX_ATTEMPTS + 3):
+            assert await _maybe_auto_approve(1, "@1", _YESNO_PANE) is True
+    assert send_keys.await_count == AUTO_APPROVE_MAX_ATTEMPTS + 3
+
+
+@pytest.mark.asyncio
+async def test_auto_approve_counter_resets_when_prompt_clears():
+    """Once the prompt clears, _reconcile_no_ui_state pops the escalation
+    counter so a later prompt starts fresh instead of instantly escalating."""
+    send_keys = AsyncMock()
+    with (
+        patch.object(
+            status_polling.session_manager,
+            "get_user_settings",
+            lambda u: {"auto_approve": "on"},
+        ),
+        patch.object(status_polling.tmux_manager, "send_keys", send_keys),
+        patch.object(
+            status_polling, "extract_interactive_content", lambda p: _content()
+        ),
+    ):
+        # Escalate.
+        for _ in range(AUTO_APPROVE_MAX_ATTEMPTS + 1):
+            await _maybe_auto_approve(1, "@1", _YESNO_PANE)
+        assert (1, "@1") in _auto_approve_attempts
+
+        # Prompt gone (empty pane → no interactive UI) → counter popped.
+        bot = AsyncMock()
+        with patch.object(notifications, "has_pending_kb", lambda u, s: (False, False)):
+            await _reconcile_no_ui_state(bot, 1, "@1", "", _sess(), is_bg_session=False)
+        assert (1, "@1") not in _auto_approve_attempts
+
+        # A fresh prompt gets auto-approved again (not instantly escalated).
+        assert await _maybe_auto_approve(1, "@1", _YESNO_PANE) is True