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
10 changes: 10 additions & 0 deletions .env.example
Original file line number Diff line number Diff line change
Expand Up @@ -12,3 +12,13 @@ CLAUDE_COMMAND=claude

# Monitor polling interval in seconds (optional, defaults to 2.0)
MONITOR_POLL_INTERVAL=2.0

# Topic (thread) IDs this instance handles (optional, comma-separated).
# Empty/unset means all topics are allowed. Lets several ccbot instances
# share one Telegram group, each owning a subset of topics.
# CCBOT_TOPIC_ALLOWLIST=299,305

# Topic (thread) IDs where Permission Prompt / Bash-approval UIs are
# auto-confirmed instead of shown to the user (optional, comma-separated).
# Does not apply to AskUserQuestion, ExitPlanMode, RestoreCheckpoint, or Settings.
# CCBOT_TOPIC_AUTO_CONFIRM=299
2 changes: 2 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -97,6 +97,8 @@ ALLOWED_USERS=your_telegram_user_id
| `CCBOT_SHOW_HIDDEN_DIRS` | `false` | Show hidden (dot) directories in directory browser |
| `OPENAI_API_KEY` | _(none)_ | OpenAI API key for voice message transcription |
| `OPENAI_BASE_URL` | `https://api.openai.com/v1` | OpenAI API base URL (for proxies or compatible APIs) |
| `CCBOT_TOPIC_ALLOWLIST` | _(none, all allowed)_ | Comma-separated topic (thread) IDs this instance handles; other topics are ignored entirely. Lets several ccbot instances (e.g. on different servers) share one Telegram group, each owning a subset of topics |
| `CCBOT_TOPIC_AUTO_CONFIRM` | _(none)_ | Comma-separated topic (thread) IDs where Permission Prompt / Bash-approval UIs are auto-confirmed (Enter sent automatically) instead of shown to the user. Does not apply to AskUserQuestion, ExitPlanMode, RestoreCheckpoint, or Settings |

Message formatting is always HTML via `chatgpt-md-converter` (`chatgpt_md_converter` package).
There is no runtime formatter switch to MarkdownV2.
Expand Down
27 changes: 27 additions & 0 deletions src/ccbot/bot.py
Original file line number Diff line number Diff line change
Expand Up @@ -50,10 +50,12 @@
from telegram.ext import (
AIORateLimiter,
Application,
ApplicationHandlerStop,
CallbackQueryHandler,
CommandHandler,
ContextTypes,
MessageHandler,
TypeHandler,
filters,
)

Expand Down Expand Up @@ -174,6 +176,20 @@ def _get_thread_id(update: Update) -> int | None:
return tid


async def _topic_gate(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""Drop updates for topics not in CCBOT_TOPIC_ALLOWLIST before any other
handler sees them.

Registered in an earlier handler group (see create_bot) so it always runs
first; raising ApplicationHandlerStop stops PTB from dispatching this
update any further. No-op (allows everything) when the allowlist is
empty — the default, unrestricted behavior.
"""
thread_id = _get_thread_id(update)
if thread_id is not None and not config.is_topic_allowed(thread_id):
raise ApplicationHandlerStop


# --- Command handlers ---


Expand Down Expand Up @@ -1802,6 +1818,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 +1834,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 Expand Up @@ -1910,6 +1933,10 @@ def create_bot() -> Application:
.build()
)

# Topic allowlist gate — runs before everything else (earlier group), so
# a disallowed topic never reaches any other handler below.
application.add_handler(TypeHandler(Update, _topic_gate), group=-1)

application.add_handler(CommandHandler("start", start_command))
application.add_handler(CommandHandler("history", history_command))
application.add_handler(CommandHandler("screenshot", screenshot_command))
Expand Down
40 changes: 40 additions & 0 deletions src/ccbot/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -101,6 +101,38 @@ def __init__(self) -> None:
os.getenv("CCBOT_SHOW_HIDDEN_DIRS", "").lower() == "true"
)

# Topic allowlist: if non-empty, only these thread_ids are processed
# (everything else is dropped before any handler runs). Empty/unset
# means no restriction — lets several ccbot instances (e.g. on
# different servers) share one Telegram group, each owning a subset
# of topics, without fighting over topics that belong to another one.
topic_allowlist_str = os.getenv("CCBOT_TOPIC_ALLOWLIST", "")
try:
self.topic_allowlist: set[int] = {
int(t.strip()) for t in topic_allowlist_str.split(",") if t.strip()
}
except ValueError as e:
raise ValueError(
f"CCBOT_TOPIC_ALLOWLIST contains non-numeric value: {e}. "
"Expected comma-separated topic (thread) IDs."
) from e

# Topics where Permission Prompt / Bash-approval UIs are auto-confirmed
# (Enter sent automatically, as if the default "Yes" were tapped)
# instead of being posted to the user. Scoped to just these two UI
# types — AskUserQuestion/ExitPlanMode/RestoreCheckpoint/Settings have
# no safe universal default and are always shown normally.
topic_auto_confirm_str = os.getenv("CCBOT_TOPIC_AUTO_CONFIRM", "")
try:
self.topic_auto_confirm: set[int] = {
int(t.strip()) for t in topic_auto_confirm_str.split(",") if t.strip()
}
except ValueError as e:
raise ValueError(
f"CCBOT_TOPIC_AUTO_CONFIRM contains non-numeric value: {e}. "
"Expected comma-separated topic (thread) IDs."
) from e

# OpenAI API for voice message transcription (optional)
self.openai_api_key: str = os.getenv("OPENAI_API_KEY", "")
self.openai_base_url: str = os.getenv(
Expand All @@ -126,5 +158,13 @@ def is_user_allowed(self, user_id: int) -> bool:
"""Check if a user is in the allowed list."""
return user_id in self.allowed_users

def is_topic_allowed(self, thread_id: int) -> bool:
"""Check if a topic is allowed. An empty allowlist allows all topics."""
return not self.topic_allowlist or thread_id in self.topic_allowlist

def is_topic_auto_confirm(self, thread_id: int) -> bool:
"""Check if a topic has Permission Prompt / Bash-approval auto-confirm enabled."""
return thread_id in self.topic_auto_confirm


config = Config()
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
Loading
Loading