Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions src/ccbot/bot.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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
Expand Down
10 changes: 9 additions & 1 deletion src/ccbot/handlers/cleanup.py
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand All @@ -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)
"""
Expand All @@ -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)

Expand Down
76 changes: 71 additions & 5 deletions src/ccbot/handlers/message_queue.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand All @@ -78,6 +86,13 @@ 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(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] = {}

Expand Down Expand Up @@ -130,7 +145,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(
Expand Down Expand Up @@ -192,6 +209,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,
)
Expand Down Expand Up @@ -386,9 +404,38 @@ 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.
#
# 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_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

Expand All @@ -404,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(
Expand All @@ -415,15 +463,25 @@ 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)

# 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 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)

# 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)


Expand Down Expand Up @@ -643,6 +701,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(
Expand All @@ -662,6 +721,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)

Expand Down Expand Up @@ -709,6 +769,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.

Expand Down
13 changes: 10 additions & 3 deletions src/ccbot/markdown_v2.py
Original file line number Diff line number Diff line change
Expand Up @@ -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.
"""
Expand All @@ -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:
Expand Down
Loading
Loading