From 5d3bc34e3b7dccb673961764c844d8cc4cbe1a9e Mon Sep 17 00:00:00 2001 From: Stelatix Date: Sun, 9 Aug 2026 16:04:00 +0200 Subject: [PATCH 1/4] fix: require "**" prefix for Telegram expandable blockquotes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit _render_expandable_quote() built raw MarkdownV2 blockquotes as ">line1\n>line2||" — missing the "**" that Telegram's grammar requires to open an *expandable* quote. Without it, the trailing "||" is an unmatched spoiler entity: Telegram rejects the whole sendMessage call with "Can't parse entities", and the caller's fallback resends the message as unformatted plain text with no collapsing at all. This path is hit by most non-trivial notifications: thinking blocks, Bash/Grep/Glob/Task/WebFetch/WebSearch tool_result output, multi-line tool errors, and Edit diffs all go through _format_expandable_quote(). So instead of a collapsed "N lines, tap to expand" block, the full raw content (including repeated tracebacks in log-heavy output) was landing directly in the chat. Fix: promote the first ">" of the rendered block to "**>", matching telegramify_markdown's own render_quote() implementation. Existing tests asserted the old (broken) shape; updated them and added coverage for multi-line and truncated quotes. Co-Authored-By: Claude Sonnet 5 --- src/ccbot/markdown_v2.py | 13 ++++++++++--- tests/ccbot/test_markdown_v2.py | 25 +++++++++++++++++++++++-- 2 files changed, 33 insertions(+), 5 deletions(-) diff --git a/src/ccbot/markdown_v2.py b/src/ccbot/markdown_v2.py index d01bbdf86..ff3c7e7a8 100644 --- a/src/ccbot/markdown_v2.py +++ b/src/ccbot/markdown_v2.py @@ -124,6 +124,12 @@ def _escape_mdv2(text: str) -> str: def _render_expandable_quote(m: re.Match[str]) -> str: """Render an expandable blockquote block in raw MarkdownV2. + Telegram's expandable-quote grammar requires the first quoted line to + start with "**>" (not plain ">"); otherwise the trailing "||" is an + unmatched spoiler entity and the whole sendMessage call is rejected + with "Can't parse entities", silently degrading the message to + unformatted plain text via the caller's fallback path. + Truncates the rendered output to _EXPQUOTE_MAX_RENDERED chars to ensure the final message fits within Telegram's 4096 limit. """ @@ -148,9 +154,10 @@ def _render_expandable_quote(m: re.Match[str]) -> str: break built.append(f">{line}") total_len += line_cost - if truncated: - return "\n".join(built) + suffix - return "\n".join(built) + "||" + body = "\n".join(built) + (suffix if truncated else "||") + # Mark the block as expandable by promoting the first ">" to "**>". + idx = body.index(">") + return body[:idx] + "**" + body[idx:] def _markdownify(text: str) -> str: diff --git a/tests/ccbot/test_markdown_v2.py b/tests/ccbot/test_markdown_v2.py index 3628b185f..ceb77a3a5 100644 --- a/tests/ccbot/test_markdown_v2.py +++ b/tests/ccbot/test_markdown_v2.py @@ -46,13 +46,34 @@ def test_expandable_quote_sentinels(self) -> None: result = convert_markdown(text) assert EXP_START not in result assert EXP_END not in result - assert ">quoted content||" in result + # Telegram requires "**>" (not plain ">") to open an *expandable* + # blockquote — otherwise the trailing "||" is an unmatched spoiler + # entity and Telegram rejects the whole message. + assert "**>quoted content||" in result def test_mixed_text_and_expandable_quote(self) -> None: text = f"before {EXP_START}inside quote{EXP_END} after" result = convert_markdown(text) assert EXP_START not in result assert EXP_END not in result - assert ">inside quote||" in result + assert "**>inside quote||" in result assert "before" in result assert "after" in result + + def test_multiline_expandable_quote_only_first_line_marked(self) -> None: + text = f"{EXP_START}line one\nline two\nline three{EXP_END}" + result = convert_markdown(text) + assert "**>line one" in result + assert "\n>line two" in result + assert "\n>line three||" in result + # Only the first line carries the "**" expandability marker. + assert result.count("**>") == 1 + + def test_expandable_quote_truncation_keeps_expandable_marker(self) -> None: + text = f"{EXP_START}{'x' * 10000}{EXP_END}" + result = convert_markdown(text) + assert "**>" in result + assert result.index("**>") < 10 + assert "\\(truncated\\)||" in result + # Rendered quote block must still fit Telegram's message limit. + assert len(result) < 4096 From e0089f1abb3f23f52ea1e223afaad045ed7b96c0 Mon Sep 17 00:00:00 2001 From: Stelatix Date: Sun, 9 Aug 2026 16:04:16 +0200 Subject: [PATCH 2/4] feat: delete secondary messages once the next one arrives MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Live notifications for a topic used to accumulate indefinitely: every thinking block, tool_use/tool_result, local-command echo, and even the bot's own echo of the user's outgoing message stayed in the chat forever, alongside the actual assistant answer. For tool-call-heavy sessions (e.g. a monitoring loop polling journalctl) this buried the real answer under a growing wall of technical messages. Add per-topic tracking of the last "secondary" message (_secondary_msg_info, keyed like the existing _status_msg_info / _tool_msg_ids). Secondary means everything except the final assistant text answer: thinking, tool_use, tool_result, local_command, and the role=user echo of a forwarded message. Whenever a new message arrives for a topic, the previously tracked secondary message is deleted; the new message is tracked in its place if it's itself secondary, or left untouched if it's the final answer. Whatever happens to be last simply never gets deleted, so a completed turn naturally leaves only the answer behind — no separate "is this the final message" detection needed. Final text answers are never deleted or tracked, per prior discussion (multiple text blocks in one turn should all persist). tool_use -> tool_result keeps its existing edit-in-place behavior (same message_id, so the secondary tracking stays valid across the edit) rather than being folded into delete+resend. _can_merge_tasks() now also refuses to merge a secondary task with a non-secondary one, since the merged task can only carry one is_secondary value. clear_topic_state() clears the new tracking on topic close. Co-Authored-By: Claude Sonnet 5 --- src/ccbot/bot.py | 7 + src/ccbot/handlers/cleanup.py | 10 +- src/ccbot/handlers/message_queue.py | 51 ++++++- tests/ccbot/handlers/test_message_queue.py | 158 +++++++++++++++++++++ 4 files changed, 220 insertions(+), 6 deletions(-) create mode 100644 tests/ccbot/handlers/test_message_queue.py diff --git a/src/ccbot/bot.py b/src/ccbot/bot.py index 809efcc9d..5172ec7bd 100644 --- a/src/ccbot/bot.py +++ b/src/ccbot/bot.py @@ -1802,6 +1802,12 @@ async def handle_new_message(msg: NewMessage, bot: Bot) -> None: # Enqueue content message task # Note: tool_result editing is handled inside _process_content_task # to ensure sequential processing with tool_use message sending + # + # Everything except the final assistant text answer (thinking, + # tool_use, tool_result, local_command, user-message echo) is + # "secondary" — it gets deleted once the next message for this + # topic arrives, so only the actual answer is left behind. + is_secondary = not (msg.role == "assistant" and msg.content_type == "text") await enqueue_content_message( bot=bot, user_id=user_id, @@ -1812,6 +1818,7 @@ async def handle_new_message(msg: NewMessage, bot: Bot) -> None: text=msg.text, thread_id=thread_id, image_data=msg.image_data, + is_secondary=is_secondary, ) # Update user's read offset to current file position diff --git a/src/ccbot/handlers/cleanup.py b/src/ccbot/handlers/cleanup.py index cd7e75805..4751841be 100644 --- a/src/ccbot/handlers/cleanup.py +++ b/src/ccbot/handlers/cleanup.py @@ -12,7 +12,11 @@ from telegram import Bot from .interactive_ui import clear_interactive_msg -from .message_queue import clear_status_msg_info, clear_tool_msg_ids_for_topic +from .message_queue import ( + clear_secondary_msg_info, + clear_status_msg_info, + clear_tool_msg_ids_for_topic, +) async def clear_topic_state( @@ -30,6 +34,7 @@ async def clear_topic_state( Cleans up: - _status_msg_info (status message tracking) - _tool_msg_ids (tool_use → message_id mapping) + - _secondary_msg_info (last deletable secondary message per topic) - _interactive_msgs and _interactive_mode (interactive UI state) - user_data pending state (_pending_thread_id, _pending_thread_text) """ @@ -39,6 +44,9 @@ async def clear_topic_state( # Clear tool message ID tracking clear_tool_msg_ids_for_topic(user_id, thread_id) + # Clear secondary-message tracking + clear_secondary_msg_info(user_id, thread_id) + # Clear interactive UI state (also deletes message from chat) await clear_interactive_msg(user_id, bot, thread_id) diff --git a/src/ccbot/handlers/message_queue.py b/src/ccbot/handlers/message_queue.py index 536c21fee..22f6d586d 100644 --- a/src/ccbot/handlers/message_queue.py +++ b/src/ccbot/handlers/message_queue.py @@ -15,6 +15,10 @@ - Message queue worker: Background task processing user's queue - Content task processing with tool_use/tool_result handling - Status message tracking and conversion (keyed by (user_id, thread_id)) + - Secondary message tracking: the last non-final (thinking/tool_use/ + tool_result/local_command/user-echo) message per topic is deleted as + soon as the next message arrives, so only the final assistant answer + is left behind once a turn completes. """ import asyncio @@ -64,6 +68,10 @@ class MessageTask: content_type: str = "text" thread_id: int | None = None # Telegram topic thread_id for targeted send image_data: list[tuple[str, bytes]] | None = None # From tool_result images + # True for everything except the final assistant text answer (thinking, + # tool_use, tool_result, local_command, user-message echo). Secondary + # messages are deleted as soon as the next message for the topic arrives. + is_secondary: bool = False # Per-user message queues and worker tasks @@ -78,6 +86,10 @@ class MessageTask: # Status message tracking: (user_id, thread_id_or_0) -> (message_id, window_id, last_text) _status_msg_info: dict[tuple[int, int], tuple[int, str, str]] = {} +# Last secondary (non-final) content message per topic, deleted once the +# next message arrives: (user_id, thread_id_or_0) -> (message_id, window_id) +_secondary_msg_info: dict[tuple[int, int], tuple[int, str]] = {} + # Flood control: user_id -> monotonic time when ban expires _flood_until: dict[int, float] = {} @@ -130,7 +142,9 @@ def _can_merge_tasks(base: MessageTask, candidate: MessageTask) -> bool: return False if candidate.content_type in ("tool_use", "tool_result"): return False - return True + # Never fold a secondary (deletable) message together with the final + # answer — the merged task would inherit only one is_secondary value. + return base.is_secondary == candidate.is_secondary async def _merge_content_tasks( @@ -192,6 +206,7 @@ async def _merge_content_tasks( tool_use_id=first.tool_use_id, content_type=first.content_type, thread_id=first.thread_id, + is_secondary=first.is_secondary, ), merge_count, ) @@ -386,7 +401,20 @@ async def _process_content_task(bot: Bot, user_id: int, task: MessageTask) -> No logger.debug(f"Failed to edit tool msg {edit_msg_id}, sending new") # Fall through to send as new message - # 2. Send content messages, converting status message to first content part + # 2. A new message is arriving for this topic — delete whatever secondary + # (non-final) message was left over from the previous one. Applies + # whether this new task is itself secondary or the final answer: either + # way the old one is now superseded and safe to remove. + skey = (user_id, tid) + old_secondary = _secondary_msg_info.pop(skey, None) + if old_secondary: + old_msg_id, _old_wid = old_secondary + try: + await bot.delete_message(chat_id=chat_id, message_id=old_msg_id) + except Exception: + pass + + # 3. Send content messages, converting status message to first content part first_part = True last_msg_id: int | None = None for part in task.parts: @@ -416,14 +444,19 @@ async def _process_content_task(bot: Bot, user_id: int, task: MessageTask) -> No if sent: last_msg_id = sent.message_id - # 3. Record tool_use message ID for later editing + # 4. Record tool_use message ID for later editing if last_msg_id and task.tool_use_id and task.content_type == "tool_use": _tool_msg_ids[(task.tool_use_id, user_id, tid)] = last_msg_id - # 4. Send images if present (from tool_result with base64 image blocks) + # 5. Track this message as the topic's current secondary message, so it + # gets deleted once the next message (of any kind) arrives. + if last_msg_id and task.is_secondary: + _secondary_msg_info[skey] = (last_msg_id, wid) + + # 6. Send images if present (from tool_result with base64 image blocks) await _send_task_images(bot, chat_id, task) - # 5. After content, check and send status + # 7. After content, check and send status await _check_and_send_status(bot, user_id, wid, task.thread_id) @@ -643,6 +676,7 @@ async def enqueue_content_message( text: str | None = None, thread_id: int | None = None, image_data: list[tuple[str, bytes]] | None = None, + is_secondary: bool = False, ) -> None: """Enqueue a content message task.""" logger.debug( @@ -662,6 +696,7 @@ async def enqueue_content_message( content_type=content_type, thread_id=thread_id, image_data=image_data, + is_secondary=is_secondary, ) queue.put_nowait(task) @@ -709,6 +744,12 @@ def clear_status_msg_info(user_id: int, thread_id: int | None = None) -> None: _status_msg_info.pop(skey, None) +def clear_secondary_msg_info(user_id: int, thread_id: int | None = None) -> None: + """Clear secondary-message tracking for a user (and optionally a specific thread).""" + skey = (user_id, thread_id or 0) + _secondary_msg_info.pop(skey, None) + + def clear_tool_msg_ids_for_topic(user_id: int, thread_id: int | None = None) -> None: """Clear tool message ID tracking for a specific topic. diff --git a/tests/ccbot/handlers/test_message_queue.py b/tests/ccbot/handlers/test_message_queue.py new file mode 100644 index 000000000..9e1f3f79f --- /dev/null +++ b/tests/ccbot/handlers/test_message_queue.py @@ -0,0 +1,158 @@ +"""Tests for message_queue — secondary-message rolling delete-on-next behavior. + +Covers: _can_merge_tasks refusing to fold secondary and final-answer tasks +together, _process_content_task deleting the previous secondary message +when the next one arrives, the tool_use/tool_result edit-in-place path +being left untouched by that deletion, and clear_secondary_msg_info. +""" + +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from ccbot.handlers.message_queue import ( + MessageTask, + _can_merge_tasks, + _process_content_task, + _secondary_msg_info, + _tool_msg_ids, + clear_secondary_msg_info, +) + +USER_ID = 1 +THREAD_ID = 42 +CHAT_ID = 555 + + +@pytest.fixture +def mock_bot(): + bot = AsyncMock() + sent = MagicMock() + sent.message_id = 100 + bot.send_message.return_value = sent + return bot + + +@pytest.fixture(autouse=True) +def _clear_queue_state(): + _secondary_msg_info.clear() + _tool_msg_ids.clear() + yield + _secondary_msg_info.clear() + _tool_msg_ids.clear() + + +def _task( + content_type: str, + is_secondary: bool, + parts: list[str] | None = None, + tool_use_id: str | None = None, +) -> MessageTask: + return MessageTask( + task_type="content", + window_id="@1", + parts=parts or ["hello"], + content_type=content_type, + thread_id=THREAD_ID, + is_secondary=is_secondary, + tool_use_id=tool_use_id, + ) + + +class TestCanMergeSecondary: + def test_secondary_and_final_do_not_merge(self) -> None: + base = _task("thinking", is_secondary=True) + candidate = _task("text", is_secondary=False) + assert _can_merge_tasks(base, candidate) is False + + def test_same_secondary_merges(self) -> None: + base = _task("thinking", is_secondary=True) + candidate = _task("thinking", is_secondary=True) + assert _can_merge_tasks(base, candidate) is True + + +@pytest.mark.asyncio +class TestSecondaryMessageRollingDelete: + async def test_secondary_message_is_tracked(self, mock_bot: AsyncMock) -> None: + with ( + patch("ccbot.handlers.message_queue.session_manager") as mock_sm, + patch("ccbot.handlers.message_queue.tmux_manager") as mock_tmux, + ): + mock_sm.resolve_chat_id.return_value = CHAT_ID + mock_tmux.find_window_by_id = AsyncMock(return_value=None) + + await _process_content_task(mock_bot, USER_ID, _task("thinking", True)) + + assert _secondary_msg_info[(USER_ID, THREAD_ID)] == (100, "@1") + mock_bot.delete_message.assert_not_called() + + async def test_next_secondary_deletes_previous(self, mock_bot: AsyncMock) -> None: + with ( + patch("ccbot.handlers.message_queue.session_manager") as mock_sm, + patch("ccbot.handlers.message_queue.tmux_manager") as mock_tmux, + ): + mock_sm.resolve_chat_id.return_value = CHAT_ID + mock_tmux.find_window_by_id = AsyncMock(return_value=None) + + await _process_content_task(mock_bot, USER_ID, _task("thinking", True)) + + second_sent = MagicMock() + second_sent.message_id = 101 + mock_bot.send_message.return_value = second_sent + await _process_content_task(mock_bot, USER_ID, _task("tool_result", True)) + + mock_bot.delete_message.assert_called_once_with(chat_id=CHAT_ID, message_id=100) + assert _secondary_msg_info[(USER_ID, THREAD_ID)] == (101, "@1") + + async def test_final_answer_deletes_previous_but_is_not_tracked( + self, mock_bot: AsyncMock + ) -> None: + with ( + patch("ccbot.handlers.message_queue.session_manager") as mock_sm, + patch("ccbot.handlers.message_queue.tmux_manager") as mock_tmux, + ): + mock_sm.resolve_chat_id.return_value = CHAT_ID + mock_tmux.find_window_by_id = AsyncMock(return_value=None) + + await _process_content_task(mock_bot, USER_ID, _task("thinking", True)) + + final_sent = MagicMock() + final_sent.message_id = 102 + mock_bot.send_message.return_value = final_sent + await _process_content_task( + mock_bot, USER_ID, _task("text", False, parts=["Fixed it"]) + ) + + mock_bot.delete_message.assert_called_once_with(chat_id=CHAT_ID, message_id=100) + assert (USER_ID, THREAD_ID) not in _secondary_msg_info + + async def test_tool_result_edit_in_place_does_not_delete( + self, mock_bot: AsyncMock + ) -> None: + with ( + patch("ccbot.handlers.message_queue.session_manager") as mock_sm, + patch("ccbot.handlers.message_queue.tmux_manager") as mock_tmux, + ): + mock_sm.resolve_chat_id.return_value = CHAT_ID + mock_tmux.find_window_by_id = AsyncMock(return_value=None) + + await _process_content_task( + mock_bot, USER_ID, _task("tool_use", True, tool_use_id="tu1") + ) + assert _secondary_msg_info[(USER_ID, THREAD_ID)] == (100, "@1") + + await _process_content_task( + mock_bot, USER_ID, _task("tool_result", True, tool_use_id="tu1") + ) + + mock_bot.delete_message.assert_not_called() + mock_bot.edit_message_text.assert_called_once() + # Same message_id throughout — edited in place, tracking stays valid. + assert _secondary_msg_info[(USER_ID, THREAD_ID)] == (100, "@1") + + +class TestClearSecondaryMsgInfo: + def test_clears_tracked_entry(self) -> None: + _secondary_msg_info[(USER_ID, THREAD_ID)] = (100, "@1") + clear_secondary_msg_info(USER_ID, THREAD_ID) + assert (USER_ID, THREAD_ID) not in _secondary_msg_info From b381ccca386150a469d7b639c68762d1dbbc34cb Mon Sep 17 00:00:00 2001 From: Stelatix Date: Sun, 9 Aug 2026 16:32:02 +0200 Subject: [PATCH 3/4] fix: don't orphan secondary messages on flood-control delete failures MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit _process_content_task's rolling delete-on-next logic popped the tracked secondary message out of _secondary_msg_info before attempting bot.delete_message, then swallowed all errors with a bare `except Exception: pass` — which also caught RetryAfter. Under a burst of several messages to the same topic within a couple seconds (thinking + tool_use/tool_result + task-notification text + interactive-UI polling, all hitting the same chat), Telegram's flood control could reject the delete; the exception was silently dropped and the message was already forgotten, so nothing ever cleaned it up. Keep the entry in _secondary_msg_info until the delete actually succeeds (or is given up on for a non-retryable reason), and re-raise RetryAfter so the existing _process_content_with_retry wrapper retries the whole task — a retry now finds the same tracked message and can still delete it, instead of leaking it forever. Co-Authored-By: Claude Sonnet 5 --- src/ccbot/handlers/message_queue.py | 16 +++++-- tests/ccbot/handlers/test_message_queue.py | 50 ++++++++++++++++++++++ 2 files changed, 63 insertions(+), 3 deletions(-) diff --git a/src/ccbot/handlers/message_queue.py b/src/ccbot/handlers/message_queue.py index 22f6d586d..eb2829ed1 100644 --- a/src/ccbot/handlers/message_queue.py +++ b/src/ccbot/handlers/message_queue.py @@ -405,14 +405,24 @@ async def _process_content_task(bot: Bot, user_id: int, task: MessageTask) -> No # (non-final) message was left over from the previous one. Applies # whether this new task is itself secondary or the final answer: either # way the old one is now superseded and safe to remove. + # + # Left in place (not popped) until the delete actually succeeds or is + # given up on: on RetryAfter we re-raise so _process_content_with_retry + # retries this whole task, and the retry must still find the same + # tracked message to delete — popping eagerly here would silently + # orphan it (never deleted, never retried) under flood control. skey = (user_id, tid) - old_secondary = _secondary_msg_info.pop(skey, None) + old_secondary = _secondary_msg_info.get(skey) if old_secondary: old_msg_id, _old_wid = old_secondary try: await bot.delete_message(chat_id=chat_id, message_id=old_msg_id) - except Exception: - pass + _secondary_msg_info.pop(skey, None) + except RetryAfter: + raise + except Exception as e: + logger.debug("Failed to delete secondary message %d: %s", old_msg_id, e) + _secondary_msg_info.pop(skey, None) # 3. Send content messages, converting status message to first content part first_part = True diff --git a/tests/ccbot/handlers/test_message_queue.py b/tests/ccbot/handlers/test_message_queue.py index 9e1f3f79f..04e478cf8 100644 --- a/tests/ccbot/handlers/test_message_queue.py +++ b/tests/ccbot/handlers/test_message_queue.py @@ -9,6 +9,7 @@ from unittest.mock import AsyncMock, MagicMock, patch import pytest +from telegram.error import RetryAfter from ccbot.handlers.message_queue import ( MessageTask, @@ -150,6 +151,55 @@ async def test_tool_result_edit_in_place_does_not_delete( # Same message_id throughout — edited in place, tracking stays valid. assert _secondary_msg_info[(USER_ID, THREAD_ID)] == (100, "@1") + async def test_retry_after_on_delete_keeps_tracking_for_retry( + self, mock_bot: AsyncMock + ) -> None: + """A flood-controlled delete must not orphan the tracked message. + + Regression test: popping _secondary_msg_info before the delete call + succeeded meant a RetryAfter (bare `except Exception: pass`) silently + dropped the delete and forgot the message — it was never cleaned up + by a later retry. The entry must survive so a retried call can still + find and delete it. + """ + with ( + patch("ccbot.handlers.message_queue.session_manager") as mock_sm, + patch("ccbot.handlers.message_queue.tmux_manager") as mock_tmux, + ): + mock_sm.resolve_chat_id.return_value = CHAT_ID + mock_tmux.find_window_by_id = AsyncMock(return_value=None) + + await _process_content_task(mock_bot, USER_ID, _task("thinking", True)) + assert _secondary_msg_info[(USER_ID, THREAD_ID)] == (100, "@1") + + mock_bot.delete_message.side_effect = RetryAfter(5) + with pytest.raises(RetryAfter): + await _process_content_task(mock_bot, USER_ID, _task("tool_result", True)) + + # Still tracked (not silently dropped) so a retry can delete it. + assert _secondary_msg_info[(USER_ID, THREAD_ID)] == (100, "@1") + + async def test_non_retryable_delete_failure_still_clears_tracking( + self, mock_bot: AsyncMock + ) -> None: + with ( + patch("ccbot.handlers.message_queue.session_manager") as mock_sm, + patch("ccbot.handlers.message_queue.tmux_manager") as mock_tmux, + ): + mock_sm.resolve_chat_id.return_value = CHAT_ID + mock_tmux.find_window_by_id = AsyncMock(return_value=None) + + await _process_content_task(mock_bot, USER_ID, _task("thinking", True)) + + mock_bot.delete_message.side_effect = Exception("message to delete not found") + second_sent = MagicMock() + second_sent.message_id = 101 + mock_bot.send_message.return_value = second_sent + await _process_content_task(mock_bot, USER_ID, _task("tool_result", True)) + + # Gives up cleanly rather than retrying forever, and still tracks the new message. + assert _secondary_msg_info[(USER_ID, THREAD_ID)] == (101, "@1") + class TestClearSecondaryMsgInfo: def test_clears_tracked_entry(self) -> None: From 00b58ed47adb19d58cc5f4c87cd7d71b19bcbec8 Mon Sep 17 00:00:00 2001 From: Stelatix Date: Sun, 9 Aug 2026 16:53:57 +0200 Subject: [PATCH 4/4] fix: track every message from a merged secondary task, not just the last MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit _merge_content_tasks folds several secondary entries that arrive close together (task-notification bursts observed at 3-4 messages within tens of milliseconds) into one MessageTask with multiple `parts`. Each part is sent as its own Telegram message, but _secondary_msg_info only recorded the id of the last one — the earlier N-1 messages were sent, immediately forgotten, and never deleted by the rolling cleanup, no matter how many further messages arrived afterward. Same issue applies to a single (unmerged) secondary task whose content gets paginated into multiple messages by split_message. Change _secondary_msg_info's value from a single message_id to a list of ids, collected from every part actually sent (including one converted from a status message). Deletion now walks the whole list; on RetryAfter mid-list, already-deleted ids are dropped and the remainder is written back before re-raising, so a retry of the task resumes exactly where it left off instead of re-deleting or losing track of entries. Co-Authored-By: Claude Sonnet 5 --- src/ccbot/handlers/message_queue.py | 57 ++++++++----- tests/ccbot/handlers/test_message_queue.py | 97 +++++++++++++++++++--- 2 files changed, 122 insertions(+), 32 deletions(-) diff --git a/src/ccbot/handlers/message_queue.py b/src/ccbot/handlers/message_queue.py index eb2829ed1..db7cd8521 100644 --- a/src/ccbot/handlers/message_queue.py +++ b/src/ccbot/handlers/message_queue.py @@ -86,9 +86,12 @@ class MessageTask: # Status message tracking: (user_id, thread_id_or_0) -> (message_id, window_id, last_text) _status_msg_info: dict[tuple[int, int], tuple[int, str, str]] = {} -# Last secondary (non-final) content message per topic, deleted once the -# next message arrives: (user_id, thread_id_or_0) -> (message_id, window_id) -_secondary_msg_info: dict[tuple[int, int], tuple[int, str]] = {} +# Last secondary (non-final) content message(s) per topic, deleted once the +# next message arrives: (user_id, thread_id_or_0) -> (message_ids, window_id). +# A list because merged tasks (several secondary entries queued together get +# folded into one MessageTask by _merge_content_tasks) or a paginated +# local_command can produce more than one Telegram message per task. +_secondary_msg_info: dict[tuple[int, int], tuple[list[int], str]] = {} # Flood control: user_id -> monotonic time when ban expires _flood_until: dict[int, float] = {} @@ -406,27 +409,33 @@ async def _process_content_task(bot: Bot, user_id: int, task: MessageTask) -> No # whether this new task is itself secondary or the final answer: either # way the old one is now superseded and safe to remove. # - # Left in place (not popped) until the delete actually succeeds or is - # given up on: on RetryAfter we re-raise so _process_content_with_retry - # retries this whole task, and the retry must still find the same - # tracked message to delete — popping eagerly here would silently - # orphan it (never deleted, never retried) under flood control. + # Remaining (not-yet-deleted) ids are written back before re-raising + # RetryAfter, so _process_content_with_retry's retry of this whole + # task resumes deleting exactly where it left off instead of losing + # track of ids already popped — same reasoning as for a single id, + # just extended to a list (see the flood-control regression test). skey = (user_id, tid) old_secondary = _secondary_msg_info.get(skey) if old_secondary: - old_msg_id, _old_wid = old_secondary - try: - await bot.delete_message(chat_id=chat_id, message_id=old_msg_id) - _secondary_msg_info.pop(skey, None) - except RetryAfter: - raise - except Exception as e: - logger.debug("Failed to delete secondary message %d: %s", old_msg_id, e) - _secondary_msg_info.pop(skey, None) + old_msg_ids, old_wid = old_secondary + remaining_ids = list(old_msg_ids) + while remaining_ids: + old_msg_id = remaining_ids[0] + try: + await bot.delete_message(chat_id=chat_id, message_id=old_msg_id) + remaining_ids.pop(0) + except RetryAfter: + _secondary_msg_info[skey] = (remaining_ids, old_wid) + raise + except Exception as e: + logger.debug("Failed to delete secondary message %d: %s", old_msg_id, e) + remaining_ids.pop(0) + _secondary_msg_info.pop(skey, None) # 3. Send content messages, converting status message to first content part first_part = True last_msg_id: int | None = None + sent_msg_ids: list[int] = [] for part in task.parts: sent = None @@ -442,6 +451,7 @@ async def _process_content_task(bot: Bot, user_id: int, task: MessageTask) -> No ) if converted_msg_id is not None: last_msg_id = converted_msg_id + sent_msg_ids.append(converted_msg_id) continue sent = await send_with_fallback( @@ -453,15 +463,20 @@ async def _process_content_task(bot: Bot, user_id: int, task: MessageTask) -> No if sent: last_msg_id = sent.message_id + sent_msg_ids.append(sent.message_id) # 4. Record tool_use message ID for later editing if last_msg_id and task.tool_use_id and task.content_type == "tool_use": _tool_msg_ids[(task.tool_use_id, user_id, tid)] = last_msg_id - # 5. Track this message as the topic's current secondary message, so it - # gets deleted once the next message (of any kind) arrives. - if last_msg_id and task.is_secondary: - _secondary_msg_info[skey] = (last_msg_id, wid) + # 5. Track every message sent for this task as the topic's current + # secondary message(s), so all of them get deleted once the next + # message (of any kind) arrives — not just the last one. A merged + # task (several secondary entries queued together, folded into one + # MessageTask by _merge_content_tasks) or a paginated local_command + # can produce more than one Telegram message here. + if sent_msg_ids and task.is_secondary: + _secondary_msg_info[skey] = (sent_msg_ids, wid) # 6. Send images if present (from tool_result with base64 image blocks) await _send_task_images(bot, chat_id, task) diff --git a/tests/ccbot/handlers/test_message_queue.py b/tests/ccbot/handlers/test_message_queue.py index 04e478cf8..81a4a1b58 100644 --- a/tests/ccbot/handlers/test_message_queue.py +++ b/tests/ccbot/handlers/test_message_queue.py @@ -1,9 +1,11 @@ """Tests for message_queue — secondary-message rolling delete-on-next behavior. Covers: _can_merge_tasks refusing to fold secondary and final-answer tasks -together, _process_content_task deleting the previous secondary message -when the next one arrives, the tool_use/tool_result edit-in-place path -being left untouched by that deletion, and clear_secondary_msg_info. +together, _process_content_task deleting the previous secondary message(s) +when the next one arrives (including multi-part tasks from merging or +pagination), the tool_use/tool_result edit-in-place path being left +untouched by that deletion, flood-control resilience, and +clear_secondary_msg_info. """ from unittest.mock import AsyncMock, MagicMock, patch @@ -84,7 +86,7 @@ async def test_secondary_message_is_tracked(self, mock_bot: AsyncMock) -> None: await _process_content_task(mock_bot, USER_ID, _task("thinking", True)) - assert _secondary_msg_info[(USER_ID, THREAD_ID)] == (100, "@1") + assert _secondary_msg_info[(USER_ID, THREAD_ID)] == ([100], "@1") mock_bot.delete_message.assert_not_called() async def test_next_secondary_deletes_previous(self, mock_bot: AsyncMock) -> None: @@ -103,7 +105,7 @@ async def test_next_secondary_deletes_previous(self, mock_bot: AsyncMock) -> Non await _process_content_task(mock_bot, USER_ID, _task("tool_result", True)) mock_bot.delete_message.assert_called_once_with(chat_id=CHAT_ID, message_id=100) - assert _secondary_msg_info[(USER_ID, THREAD_ID)] == (101, "@1") + assert _secondary_msg_info[(USER_ID, THREAD_ID)] == ([101], "@1") async def test_final_answer_deletes_previous_but_is_not_tracked( self, mock_bot: AsyncMock @@ -140,7 +142,7 @@ async def test_tool_result_edit_in_place_does_not_delete( await _process_content_task( mock_bot, USER_ID, _task("tool_use", True, tool_use_id="tu1") ) - assert _secondary_msg_info[(USER_ID, THREAD_ID)] == (100, "@1") + assert _secondary_msg_info[(USER_ID, THREAD_ID)] == ([100], "@1") await _process_content_task( mock_bot, USER_ID, _task("tool_result", True, tool_use_id="tu1") @@ -149,7 +151,58 @@ async def test_tool_result_edit_in_place_does_not_delete( mock_bot.delete_message.assert_not_called() mock_bot.edit_message_text.assert_called_once() # Same message_id throughout — edited in place, tracking stays valid. - assert _secondary_msg_info[(USER_ID, THREAD_ID)] == (100, "@1") + assert _secondary_msg_info[(USER_ID, THREAD_ID)] == ([100], "@1") + + async def test_multi_part_secondary_task_tracks_every_message( + self, mock_bot: AsyncMock + ) -> None: + """Regression test for merged secondary tasks losing all but the last id. + + _merge_content_tasks folds several secondary entries that arrived + close together (e.g. task-notification bursts) into one MessageTask + with several `parts`. Each part is sent as its own Telegram message; + all of them — not just the last — must be tracked so the next + message deletes every one of them instead of leaking all but one. + """ + with ( + patch("ccbot.handlers.message_queue.session_manager") as mock_sm, + patch("ccbot.handlers.message_queue.tmux_manager") as mock_tmux, + ): + mock_sm.resolve_chat_id.return_value = CHAT_ID + mock_tmux.find_window_by_id = AsyncMock(return_value=None) + + sent_ids = iter([201, 202, 203]) + + def _send(*_a, **_kw): + m = MagicMock() + m.message_id = next(sent_ids) + return m + + mock_bot.send_message.side_effect = _send + + merged = _task("text", True, parts=["notif 1", "notif 2", "notif 3"]) + await _process_content_task(mock_bot, USER_ID, merged) + + assert _secondary_msg_info[(USER_ID, THREAD_ID)] == ([201, 202, 203], "@1") + assert mock_bot.send_message.call_count == 3 + + # Next message must delete all three, not just the last. + with ( + patch("ccbot.handlers.message_queue.session_manager") as mock_sm, + patch("ccbot.handlers.message_queue.tmux_manager") as mock_tmux, + ): + mock_sm.resolve_chat_id.return_value = CHAT_ID + mock_tmux.find_window_by_id = AsyncMock(return_value=None) + + next_sent = MagicMock() + next_sent.message_id = 204 + mock_bot.send_message.side_effect = None + mock_bot.send_message.return_value = next_sent + await _process_content_task(mock_bot, USER_ID, _task("thinking", True)) + + deleted_ids = {c.kwargs["message_id"] for c in mock_bot.delete_message.call_args_list} + assert deleted_ids == {201, 202, 203} + assert _secondary_msg_info[(USER_ID, THREAD_ID)] == ([204], "@1") async def test_retry_after_on_delete_keeps_tracking_for_retry( self, mock_bot: AsyncMock @@ -170,14 +223,36 @@ async def test_retry_after_on_delete_keeps_tracking_for_retry( mock_tmux.find_window_by_id = AsyncMock(return_value=None) await _process_content_task(mock_bot, USER_ID, _task("thinking", True)) - assert _secondary_msg_info[(USER_ID, THREAD_ID)] == (100, "@1") + assert _secondary_msg_info[(USER_ID, THREAD_ID)] == ([100], "@1") mock_bot.delete_message.side_effect = RetryAfter(5) with pytest.raises(RetryAfter): await _process_content_task(mock_bot, USER_ID, _task("tool_result", True)) # Still tracked (not silently dropped) so a retry can delete it. - assert _secondary_msg_info[(USER_ID, THREAD_ID)] == (100, "@1") + assert _secondary_msg_info[(USER_ID, THREAD_ID)] == ([100], "@1") + + async def test_retry_after_partway_through_a_multi_id_delete_keeps_remainder( + self, mock_bot: AsyncMock + ) -> None: + """RetryAfter partway through deleting several tracked ids must not + re-attempt (or lose track of) the ones already handled.""" + _secondary_msg_info[(USER_ID, THREAD_ID)] = ([301, 302, 303], "@1") + with ( + patch("ccbot.handlers.message_queue.session_manager") as mock_sm, + patch("ccbot.handlers.message_queue.tmux_manager") as mock_tmux, + ): + mock_sm.resolve_chat_id.return_value = CHAT_ID + mock_tmux.find_window_by_id = AsyncMock(return_value=None) + + # First delete succeeds, second is flood-controlled. + mock_bot.delete_message.side_effect = [None, RetryAfter(5)] + with pytest.raises(RetryAfter): + await _process_content_task(mock_bot, USER_ID, _task("thinking", True)) + + assert mock_bot.delete_message.call_count == 2 + # 301 already deleted — must not be retried; 302/303 remain for the retry. + assert _secondary_msg_info[(USER_ID, THREAD_ID)] == ([302, 303], "@1") async def test_non_retryable_delete_failure_still_clears_tracking( self, mock_bot: AsyncMock @@ -198,11 +273,11 @@ async def test_non_retryable_delete_failure_still_clears_tracking( await _process_content_task(mock_bot, USER_ID, _task("tool_result", True)) # Gives up cleanly rather than retrying forever, and still tracks the new message. - assert _secondary_msg_info[(USER_ID, THREAD_ID)] == (101, "@1") + assert _secondary_msg_info[(USER_ID, THREAD_ID)] == ([101], "@1") class TestClearSecondaryMsgInfo: def test_clears_tracked_entry(self) -> None: - _secondary_msg_info[(USER_ID, THREAD_ID)] = (100, "@1") + _secondary_msg_info[(USER_ID, THREAD_ID)] = ([100], "@1") clear_secondary_msg_info(USER_ID, THREAD_ID) assert (USER_ID, THREAD_ID) not in _secondary_msg_info