Refactor messaging around explicit ports (#878)
Some checks failed
CI / pytest (push) Has been cancelled
CI / ruff-check (push) Has been cancelled
CI / ruff-format (push) Has been cancelled
CI / ty (push) Has been cancelled
CI / Ban type ignore suppressions (push) Has been cancelled

This commit is contained in:
Ali Khokhar 2026-06-20 09:35:38 -07:00 committed by GitHub
parent acf5885b16
commit d281d52ced
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
40 changed files with 1390 additions and 1420 deletions

View file

@ -486,17 +486,31 @@ selects Codex for Discord or Telegram.
## Messaging Architecture
Messaging is optional. [api/runtime.py](api/runtime.py) calls
`create_messaging_platform()` from
`create_messaging_components()` from
[messaging/platforms/factory.py](messaging/platforms/factory.py) during startup.
If `MESSAGING_PLATFORM` is `none`, or if the selected platform token is missing,
the messaging bridge is skipped.
Platform adapters in [messaging/platforms/telegram.py](messaging/platforms/telegram.py)
and [messaging/platforms/discord.py](messaging/platforms/discord.py) are thin SDK
shells: they own client lifecycle, event extraction, SDK retries, attachment
download, and raw send/edit/delete calls. Shared delivery policy lives in
[messaging/platforms/outbox.py](messaging/platforms/outbox.py), which owns queued
send/edit/delete, dedup keys, limiter delegation, and fire-and-forget behavior.
The platform factory returns a `MessagingPlatformComponents` bundle from
[messaging/platforms/ports.py](messaging/platforms/ports.py): a
`MessagingRuntime` for lifecycle and inbound callbacks, an `OutboundMessenger`
for queued sends/edits/deletes, and an optional `VoiceCancellation` port for
reply-scoped `/clear` during voice transcription. Workflow code depends on
these ports, not on Telegram or Discord SDK objects.
Runtime adapters in
[messaging/platforms/telegram.py](messaging/platforms/telegram.py) and
[messaging/platforms/discord.py](messaging/platforms/discord.py) own SDK client
lifecycle, event subscription, inbound handoff, and voice-note handoff. Inbound
normalization lives in
[messaging/platforms/telegram_inbound.py](messaging/platforms/telegram_inbound.py)
and [messaging/platforms/discord_inbound.py](messaging/platforms/discord_inbound.py).
Outbound SDK calls live in
[messaging/platforms/telegram_io.py](messaging/platforms/telegram_io.py) and
[messaging/platforms/discord_io.py](messaging/platforms/discord_io.py). Shared
delivery policy lives in [messaging/platforms/outbox.py](messaging/platforms/outbox.py),
which owns queued send/edit/delete, dedup keys, limiter delegation, and
fire-and-forget behavior.
Shared voice-note orchestration lives in
[messaging/platforms/voice_flow.py](messaging/platforms/voice_flow.py), which owns
pending voice registration, temp-file cleanup, transcription, cancellation, error
@ -517,7 +531,7 @@ and session cleanup.
[messaging/command_context.py](messaging/command_context.py) defines the typed
dependency surface for `/stop`, `/clear`, and `/stats`; commands should not
depend on the concrete workflow object.
depend on the concrete workflow object or on platform SDK runtimes.
[messaging/trees/manager.py](messaging/trees/manager.py) preserves
per-conversation ordering with tree-aware queues. Replies become child nodes, and
@ -534,7 +548,8 @@ uses debounced atomic writes and flushes pending saves on shutdown.
```mermaid
sequenceDiagram
participant Platform as DiscordOrTelegram
participant Runtime as DiscordOrTelegramRuntime
participant Outbound as OutboundMessenger
participant Workflow as MessagingWorkflow
participant Intake as MessagingTurnIntake
participant Queue as TreeQueueManager
@ -543,7 +558,7 @@ sequenceDiagram
participant CLI as ClaudeCode
participant Proxy as LocalProxy
Platform->>Workflow: IncomingMessage
Runtime->>Workflow: IncomingMessage
Workflow->>Intake: handle inbound turn
Intake->>Queue: create or extend message tree
Queue->>Runner: process node in order
@ -551,7 +566,7 @@ sequenceDiagram
Manager->>CLI: launch JSON stream task
CLI->>Proxy: provider-backed API calls
CLI-->>Runner: parsed stdout events
Runner-->>Platform: status and transcript updates
Runner-->>Outbound: status and transcript updates
```
## Observability, Diagnostics, And Safety
@ -639,15 +654,20 @@ when maintainers want branch-level assurance.
### Add A Messaging Platform
1. Implement `MessagingPlatform` from
[messaging/platforms/base.py](messaging/platforms/base.py).
2. Add construction logic to
1. Implement a `MessagingRuntime`, `OutboundMessenger`, and inbound normalizer
under [messaging/platforms/](messaging/platforms/).
2. Reuse [messaging/platforms/outbox.py](messaging/platforms/outbox.py) for
queued outbound delivery and
[messaging/platforms/voice_flow.py](messaging/platforms/voice_flow.py) for
voice-note handoff when the platform supports audio.
3. Add construction logic to
[messaging/platforms/factory.py](messaging/platforms/factory.py).
3. Add settings and admin fields for tokens, allowlists, and platform-specific
4. Add settings and admin fields for tokens, allowlists, and platform-specific
runtime options.
4. Add rendering profile support in
5. Add rendering profile support in
[messaging/rendering/profiles.py](messaging/rendering/profiles.py) if needed.
5. Add fake-platform deterministic tests and optional live smoke targets.
6. Add deterministic runtime/outbound/workflow tests and optional live smoke
targets.
### Add Protocol Behavior

View file

@ -578,7 +578,7 @@ free-claude-code/
├── core/ # Shared Anthropic protocol helpers, SSE, OpenAI Responses
│ └── openai_responses/ # Responses ↔ Anthropic conversion and SSE mapping
├── providers/ # Provider transports, registry, rate limiting
├── messaging/ # Discord/Telegram adapters, sessions, voice
├── messaging/ # Discord/Telegram runtimes, outbound ports, voice
├── cli/ # Package entry points and client CLI process management
├── config/ # Settings, provider catalog, logging
└── tests/ # Unit and contract tests
@ -640,7 +640,7 @@ CI also enforces a ban on `# type: ignore` / `# ty: ignore` suppressions; `scrip
- Add Anthropic Messages providers by extending `AnthropicMessagesTransport`.
- Extend OpenAI Responses conversion in `core/openai_responses/` when Codex adds new request or stream shapes.
- Register provider metadata in `config.provider_catalog` and factory wiring in `providers.registry`.
- Add messaging platforms by implementing the `MessagingPlatform` interface in `messaging/`.
- Add messaging platforms by wiring runtime, outbound, and inbound-normalizer ports in `messaging/platforms/`.
## Contributing

View file

@ -18,7 +18,7 @@ from providers.registry import ProviderRegistry
if TYPE_CHECKING:
from cli.managed import ManagedClaudeSessionManager
from messaging.platforms.base import MessagingPlatform
from messaging.platforms.ports import MessagingPlatformComponents, MessagingRuntime
from messaging.session import SessionStore
from messaging.workflow import MessagingWorkflow
@ -88,7 +88,7 @@ class AppRuntime:
app: FastAPI
settings: Settings
_provider_registry: ProviderRegistry | None = field(default=None, init=False)
messaging_platform: MessagingPlatform | None = None
messaging_runtime: MessagingRuntime | None = None
messaging_workflow: MessagingWorkflow | None = None
cli_manager: ManagedClaudeSessionManager | None = None
@ -153,10 +153,10 @@ class AppRuntime:
)
logger.info("Shutdown requested, cleaning up...")
if self.messaging_platform:
if self.messaging_runtime:
await best_effort(
"messaging_platform.stop",
self.messaging_platform.stop(),
"messaging_runtime.stop",
self.messaging_runtime.stop(),
log_verbose_errors=verbose,
)
if self.cli_manager:
@ -178,10 +178,10 @@ class AppRuntime:
try:
from messaging.platforms.factory import (
MessagingPlatformOptions,
create_messaging_platform,
create_messaging_components,
)
self.messaging_platform = create_messaging_platform(
components = create_messaging_components(
self.settings.messaging_platform,
MessagingPlatformOptions(
telegram_bot_token=self.settings.telegram_bot_token,
@ -200,8 +200,8 @@ class AppRuntime:
),
)
if self.messaging_platform:
await self._start_messaging_workflow()
if components:
await self._start_messaging_workflow(components)
except ImportError as e:
if self.settings.log_api_error_tracebacks:
@ -223,7 +223,9 @@ class AppRuntime:
type(e).__name__,
)
async def _start_messaging_workflow(self) -> None:
async def _start_messaging_workflow(
self, components: MessagingPlatformComponents
) -> None:
from cli.managed import ManagedClaudeSessionManager
from messaging.session import SessionStore
from messaging.workflow import MessagingWorkflow
@ -259,10 +261,11 @@ class AppRuntime:
storage_path=os.path.join(data_path, "sessions.json"),
message_log_cap=self.settings.max_message_log_entries_per_chat,
)
platform = self.messaging_platform
assert platform is not None
self.messaging_runtime = components.runtime
self.messaging_workflow = MessagingWorkflow(
platform=platform,
platform_name=components.name,
outbound=components.outbound,
voice_cancellation=components.voice_cancellation,
cli_manager=self.cli_manager,
session_store=session_store,
debug_platform_edits=self.settings.debug_platform_edits,
@ -273,9 +276,9 @@ class AppRuntime:
)
self._restore_tree_state(session_store)
platform.on_message(self.messaging_workflow.handle_message)
await platform.start()
logger.info(f"{platform.name} platform started with messaging workflow")
components.runtime.on_message(self.messaging_workflow.handle_message)
await components.runtime.start()
logger.info("{} platform started with messaging workflow", components.name)
def _restore_tree_state(self, session_store: SessionStore) -> None:
saved_trees = session_store.get_all_trees()
@ -304,7 +307,7 @@ class AppRuntime:
)
def _publish_state(self) -> None:
self.app.state.messaging_platform = self.messaging_platform
self.app.state.messaging_runtime = self.messaging_runtime
self.app.state.messaging_workflow = self.messaging_workflow
self.app.state.cli_manager = self.cli_manager

View file

@ -1,12 +1,12 @@
"""Platform-agnostic messaging layer."""
from .event_parser import parse_cli_event
from .models import IncomingMessage
from .platforms.base import (
from .managed_protocols import (
ManagedClaudeSessionManagerProtocol,
ManagedClaudeSessionProtocol,
MessagingPlatform,
)
from .models import IncomingMessage
from .platforms.ports import OutboundMessenger
from .session import SessionStore
from .trees import MessageNode, MessageState, MessageTree, TreeQueueManager
from .workflow import MessagingWorkflow
@ -18,8 +18,8 @@ __all__ = [
"MessageNode",
"MessageState",
"MessageTree",
"MessagingPlatform",
"MessagingWorkflow",
"OutboundMessenger",
"SessionStore",
"TreeQueueManager",
"parse_cli_event",

View file

@ -4,7 +4,8 @@ from __future__ import annotations
from typing import Protocol
from .platforms.base import ManagedClaudeSessionManagerProtocol, MessagingPlatform
from .managed_protocols import ManagedClaudeSessionManagerProtocol
from .platforms.ports import OutboundMessenger, VoiceCancellation
from .session import SessionStore
from .transcript import RenderCtx
from .trees import MessageNode, MessageTree, TreeQueueManager
@ -13,7 +14,8 @@ from .trees import MessageNode, MessageTree, TreeQueueManager
class MessagingCommandContext(Protocol):
"""Operations commands need from the messaging workflow."""
platform: MessagingPlatform
outbound: OutboundMessenger
voice_cancellation: VoiceCancellation | None
cli_manager: ManagedClaudeSessionManagerProtocol
session_store: SessionStore

View file

@ -26,7 +26,7 @@ async def handle_stop_command(
node_id = handler.tree_queue.resolve_parent_node_id(reply_id) if tree else None
if not node_id:
msg_id = await handler.platform.queue_send_message(
msg_id = await handler.outbound.queue_send_message(
incoming.chat_id,
handler.format_status(
"", "Stopped.", "Nothing to stop for that message."
@ -41,7 +41,7 @@ async def handle_stop_command(
count = await handler.stop_task(node_id)
noun = "request" if count == 1 else "requests"
msg_id = await handler.platform.queue_send_message(
msg_id = await handler.outbound.queue_send_message(
incoming.chat_id,
handler.format_status("", "Stopped.", f"Cancelled {count} {noun}."),
fire_and_forget=False,
@ -54,7 +54,7 @@ async def handle_stop_command(
# Global stop: legacy behavior (stop everything)
count = await handler.stop_all_tasks()
msg_id = await handler.platform.queue_send_message(
msg_id = await handler.outbound.queue_send_message(
incoming.chat_id,
handler.format_status(
"", "Stopped.", f"Cancelled {count} pending or active requests."
@ -74,7 +74,7 @@ async def handle_stats_command(
stats = handler.cli_manager.get_stats()
tree_count = handler.tree_queue.get_tree_count()
ctx = handler.get_render_ctx()
msg_id = await handler.platform.queue_send_message(
msg_id = await handler.outbound.queue_send_message(
incoming.chat_id,
"📊 "
+ ctx.bold("Stats")
@ -118,7 +118,7 @@ async def _delete_message_ids(
CHUNK = 100
for i in range(0, len(ordered), CHUNK):
chunk = ordered[i : i + CHUNK]
await handler.platform.queue_delete_messages(
await handler.outbound.queue_delete_messages(
chat_id, chunk, fire_and_forget=False
)
except Exception as e:
@ -195,16 +195,17 @@ async def handle_clear_command(
handler.tree_queue.resolve_parent_node_id(reply_id) if tree else None
)
if not branch_root_id:
cancel_fn = getattr(handler.platform, "cancel_pending_voice", None)
if cancel_fn is not None:
cancelled = await cancel_fn(incoming.chat_id, reply_id)
if handler.voice_cancellation is not None:
cancelled = await handler.voice_cancellation.cancel_pending_voice(
incoming.chat_id, reply_id
)
if cancelled is not None:
voice_msg_id, status_msg_id = cancelled
msg_ids_to_del: set[str] = {voice_msg_id, status_msg_id}
if incoming.message_id is not None:
msg_ids_to_del.add(str(incoming.message_id))
await _delete_message_ids(handler, incoming.chat_id, msg_ids_to_del)
msg_id = await handler.platform.queue_send_message(
msg_id = await handler.outbound.queue_send_message(
incoming.chat_id,
handler.format_status("🗑", "Cleared.", "Voice note cancelled."),
fire_and_forget=False,
@ -214,7 +215,7 @@ async def handle_clear_command(
incoming.platform, incoming.chat_id, msg_id, "command"
)
return
msg_id = await handler.platform.queue_send_message(
msg_id = await handler.outbound.queue_send_message(
incoming.chat_id,
handler.format_status(
"🗑", "Cleared.", "Nothing to clear for that message."

View file

@ -0,0 +1,37 @@
"""Protocols for messaging-owned managed Claude sessions."""
from __future__ import annotations
from collections.abc import AsyncGenerator
from typing import Any, Protocol, runtime_checkable
@runtime_checkable
class ManagedClaudeSessionProtocol(Protocol):
"""Protocol for managed Claude sessions used by messaging."""
def start_task(
self, prompt: str, session_id: str | None = None, fork_session: bool = False
) -> AsyncGenerator[dict, Any]: ...
@property
def is_busy(self) -> bool: ...
@runtime_checkable
class ManagedClaudeSessionManagerProtocol(Protocol):
"""Protocol for the managed Claude session pool used by messaging."""
async def get_or_create_session(
self, session_id: str | None = None
) -> tuple[ManagedClaudeSessionProtocol, str, bool]: ...
async def register_real_session_id(
self, temp_id: str, real_session_id: str
) -> bool: ...
async def stop_all(self) -> None: ...
async def remove_session(self, session_id: str) -> bool: ...
def get_stats(self) -> dict: ...

View file

@ -10,7 +10,7 @@ from loguru import logger
from core.trace import trace_event
from .cli_event_constants import TRANSCRIPT_EVENT_TYPES, get_status_for_event
from .platforms.base import ManagedClaudeSessionManagerProtocol
from .managed_protocols import ManagedClaudeSessionManagerProtocol
from .safe_diagnostics import text_len_hint
from .session import SessionStore
from .transcript import TranscriptBuffer

View file

@ -11,8 +11,9 @@ from core.anthropic import format_user_error_preview, get_user_facing_error_mess
from core.trace import trace_event
from .event_parser import parse_cli_event
from .managed_protocols import ManagedClaudeSessionManagerProtocol
from .node_event_pipeline import handle_session_info_event, process_parsed_cli_event
from .platforms.base import ManagedClaudeSessionManagerProtocol, MessagingPlatform
from .platforms.ports import OutboundMessenger
from .safe_diagnostics import format_exception_for_log
from .session import SessionStore
from .transcript import RenderCtx, TranscriptBuffer
@ -26,7 +27,8 @@ class MessagingNodeRunner:
def __init__(
self,
*,
platform: MessagingPlatform,
platform_name: str,
outbound: OutboundMessenger,
cli_manager: ManagedClaudeSessionManagerProtocol,
session_store: SessionStore,
get_tree_queue: Callable[[], TreeQueueManager],
@ -39,7 +41,8 @@ class MessagingNodeRunner:
log_raw_cli_diagnostics: bool = False,
log_messaging_error_details: bool = False,
) -> None:
self.platform = platform
self.platform_name = platform_name
self.outbound = outbound
self.cli_manager = cli_manager
self.session_store = session_store
self._get_tree_queue = get_tree_queue
@ -103,7 +106,7 @@ class MessagingNodeRunner:
last_status: str | None = None
parent_session_id = None
platform_nm = getattr(self.platform, "name", "messaging")
platform_nm = self.platform_name
if tree and node.parent_id:
parent_session_id = tree.get_parent_session_id(node_id)
if parent_session_id:
@ -117,7 +120,7 @@ class MessagingNodeRunner:
)
editor = ThrottledTranscriptEditor(
platform=self.platform,
outbound=self.outbound,
parse_mode=self._get_parse_mode(),
get_limit_chars=self._get_limit_chars,
transcript=transcript,
@ -329,8 +332,8 @@ class MessagingNodeRunner:
if affected:
self._save_tree(tree_queue.get_tree_for_node(node_id))
for child in affected[1:]:
self.platform.fire_and_forget(
self.platform.queue_edit_message(
self.outbound.fire_and_forget(
self.outbound.queue_edit_message(
child.incoming.chat_id,
child.status_message_id,
self._format_status("", "Cancelled:", child_status_text),

View file

@ -1,15 +1,18 @@
"""Messaging platform adapters (Telegram, Discord, etc.)."""
"""Messaging platform runtimes and ports."""
from .base import (
ManagedClaudeSessionManagerProtocol,
ManagedClaudeSessionProtocol,
MessagingPlatform,
from .factory import MessagingPlatformOptions, create_messaging_components
from .ports import (
MessagingPlatformComponents,
MessagingRuntime,
OutboundMessenger,
VoiceCancellation,
)
from .factory import create_messaging_platform
__all__ = [
"ManagedClaudeSessionManagerProtocol",
"ManagedClaudeSessionProtocol",
"MessagingPlatform",
"create_messaging_platform",
"MessagingPlatformComponents",
"MessagingPlatformOptions",
"MessagingRuntime",
"OutboundMessenger",
"VoiceCancellation",
"create_messaging_components",
]

View file

@ -1,228 +0,0 @@
"""Abstract base class for messaging platforms."""
from abc import ABC, abstractmethod
from collections.abc import AsyncGenerator, Awaitable, Callable
from typing import (
Any,
Protocol,
runtime_checkable,
)
from ..models import IncomingMessage
@runtime_checkable
class ManagedClaudeSessionProtocol(Protocol):
"""Protocol for managed Claude sessions - avoid circular imports."""
def start_task(
self, prompt: str, session_id: str | None = None, fork_session: bool = False
) -> AsyncGenerator[dict, Any]: ...
@property
def is_busy(self) -> bool: ...
@runtime_checkable
class ManagedClaudeSessionManagerProtocol(Protocol):
"""
Protocol for managed Claude session managers to avoid tight coupling.
"""
async def get_or_create_session(
self, session_id: str | None = None
) -> tuple[ManagedClaudeSessionProtocol, str, bool]:
"""
Get an existing session or create a new one.
Returns: Tuple of (session, session_id, is_new_session)
"""
...
async def register_real_session_id(
self, temp_id: str, real_session_id: str
) -> bool:
"""Register the real session ID from CLI output."""
...
async def stop_all(self) -> None:
"""Stop all sessions."""
...
async def remove_session(self, session_id: str) -> bool:
"""Remove a session from the manager."""
...
def get_stats(self) -> dict:
"""Get session statistics."""
...
class MessagingPlatform(ABC):
"""
Base class for all messaging platform adapters.
Implement this to add support for Telegram, Discord, Slack, etc.
"""
name: str = "base"
@abstractmethod
async def start(self) -> None:
"""Initialize and connect to the messaging platform."""
pass
@abstractmethod
async def stop(self) -> None:
"""Disconnect and cleanup resources."""
pass
@abstractmethod
async def send_message(
self,
chat_id: str,
text: str,
reply_to: str | None = None,
parse_mode: str | None = None,
message_thread_id: str | None = None,
) -> str:
"""
Send a message to a chat.
Args:
chat_id: The chat/channel ID to send to
text: Message content
reply_to: Optional message ID to reply to
parse_mode: Optional formatting mode ("markdown", "html")
message_thread_id: Optional thread or topic id for threaded channels
(e.g. forum topics); unused on platforms that do not support it.
Returns:
The message ID of the sent message
"""
pass
@abstractmethod
async def edit_message(
self,
chat_id: str,
message_id: str,
text: str,
parse_mode: str | None = None,
) -> None:
"""
Edit an existing message.
Args:
chat_id: The chat/channel ID
message_id: The message ID to edit
text: New message content
parse_mode: Optional formatting mode
"""
pass
@abstractmethod
async def delete_message(
self,
chat_id: str,
message_id: str,
) -> None:
"""
Delete a message from a chat.
Args:
chat_id: The chat/channel ID
message_id: The message ID to delete
"""
pass
@abstractmethod
async def queue_send_message(
self,
chat_id: str,
text: str,
reply_to: str | None = None,
parse_mode: str | None = None,
fire_and_forget: bool = True,
message_thread_id: str | None = None,
) -> str | None:
"""
Enqueue a message to be sent.
If fire_and_forget is True, returns None immediately.
Otherwise, waits for the rate limiter and returns message ID.
"""
pass
@abstractmethod
async def queue_edit_message(
self,
chat_id: str,
message_id: str,
text: str,
parse_mode: str | None = None,
fire_and_forget: bool = True,
) -> None:
"""
Enqueue a message edit.
If fire_and_forget is True, returns immediately.
Otherwise, waits for the rate limiter.
"""
pass
@abstractmethod
async def queue_delete_message(
self,
chat_id: str,
message_id: str,
fire_and_forget: bool = True,
) -> None:
"""
Enqueue a message deletion.
If fire_and_forget is True, returns immediately.
Otherwise, waits for the rate limiter.
"""
pass
async def queue_delete_messages(
self,
chat_id: str,
message_ids: list[str],
*,
fire_and_forget: bool = True,
) -> None:
"""Delete many messages; default loops :meth:`queue_delete_message`.
Adapters with native bulk delete should override.
"""
for mid in message_ids:
await self.queue_delete_message(
chat_id, mid, fire_and_forget=fire_and_forget
)
@abstractmethod
def on_message(
self,
handler: Callable[[IncomingMessage], Awaitable[None]],
) -> None:
"""
Register a message handler callback.
The handler will be called for each incoming message.
Args:
handler: Async function that processes incoming messages
"""
pass
@abstractmethod
def fire_and_forget(self, task: Awaitable[Any]) -> None:
"""Execute a coroutine without awaiting it."""
pass
@property
def is_connected(self) -> bool:
"""Check if the platform is connected."""
return False

View file

@ -1,13 +1,11 @@
"""
Discord Platform Adapter
"""Discord messaging runtime."""
Implements MessagingPlatform for Discord using discord.py.
"""
from __future__ import annotations
import asyncio
import contextlib
from collections.abc import Awaitable, Callable
from typing import Any, cast
from typing import Any
from loguru import logger
@ -15,14 +13,15 @@ from core.anthropic import format_user_error_preview
from ..models import IncomingMessage
from ..rendering.discord_markdown import format_status_discord
from .base import MessagingPlatform
from .outbox import PlatformOutbox
from .voice_flow import (
VoiceNoteFlow,
VoiceNoteRequest,
audio_suffix_from_metadata,
is_audio_metadata,
from .discord_inbound import (
discord_text_message_from_event,
discord_voice_request_from_event,
get_audio_attachment,
parse_allowed_channels,
)
from .discord_io import DiscordMessenger
from .ports import InboundMessageHandler
from .voice_flow import VoiceNoteFlow
_discord_module: Any = None
try:
@ -33,11 +32,9 @@ try:
except ImportError:
DISCORD_AVAILABLE = False
DISCORD_MESSAGE_LIMIT = 2000
def _get_discord() -> Any:
"""Return the discord module. Raises if not available."""
"""Return the discord module or raise a setup error."""
if not DISCORD_AVAILABLE or _discord_module is None:
raise ImportError(
"discord.py is required. Install with: pip install discord.py"
@ -45,46 +42,32 @@ def _get_discord() -> Any:
return _discord_module
def _parse_allowed_channels(raw: str | None) -> set[str]:
"""Parse comma-separated channel IDs into a set of strings."""
if not raw or not raw.strip():
return set()
return {s.strip() for s in raw.split(",") if s.strip()}
if DISCORD_AVAILABLE and _discord_module is not None:
_discord = _discord_module
class _DiscordClient(_discord.Client):
"""Internal Discord client that forwards events to DiscordPlatform."""
"""Internal Discord client that forwards events to the runtime."""
def __init__(
self,
platform: DiscordPlatform,
runtime: DiscordRuntime,
intents: _discord.Intents,
) -> None:
super().__init__(intents=intents)
self._platform = platform
self._runtime = runtime
async def on_ready(self) -> None:
"""Called when the bot is ready."""
self._platform._connected = True
self._runtime._connected = True
logger.info("Discord platform connected")
async def on_message(self, message: Any) -> None:
"""Handle incoming Discord messages."""
await self._platform._handle_client_message(message)
await self._runtime._handle_client_message(message)
else:
_DiscordClient = None
class DiscordPlatform(MessagingPlatform):
"""
Discord messaging platform adapter.
Uses discord.py for Discord access.
Requires a Bot Token from Discord Developer Portal and message_content intent.
"""
class DiscordRuntime:
"""Owns Discord SDK lifecycle and inbound event handoff."""
name = "discord"
@ -102,15 +85,14 @@ class DiscordPlatform(MessagingPlatform):
messaging_rate_window: float = 1.0,
log_raw_messaging_content: bool = False,
log_api_error_tracebacks: bool = False,
):
) -> None:
if not DISCORD_AVAILABLE:
raise ImportError(
"discord.py is required. Install with: pip install discord.py"
)
self.bot_token = bot_token
self.allowed_channel_ids = _parse_allowed_channels(allowed_channel_ids)
self.allowed_channel_ids = parse_allowed_channels(allowed_channel_ids)
if not self.bot_token:
logger.warning("DISCORD_BOT_TOKEN not set")
@ -120,51 +102,14 @@ class DiscordPlatform(MessagingPlatform):
assert _DiscordClient is not None
self._client = _DiscordClient(self, intents)
self._message_handler: Callable[[IncomingMessage], Awaitable[None]] | None = (
None
)
self._message_handler: InboundMessageHandler | None = None
self._connected = False
self._limiter: Any | None = None
self._start_task: asyncio.Task | None = None
async def send_operation(
chat_id: str,
text: str,
reply_to: str | None,
parse_mode: str | None,
message_thread_id: str | None,
) -> str:
return await self.send_message(
chat_id,
text,
reply_to,
parse_mode,
message_thread_id,
)
async def edit_operation(
chat_id: str,
message_id: str,
text: str,
parse_mode: str | None,
) -> None:
await self.edit_message(chat_id, message_id, text, parse_mode)
async def delete_operation(chat_id: str, message_id: str) -> None:
await self.delete_message(chat_id, message_id)
async def delete_many_operation(
chat_id: str,
message_ids: list[str],
) -> None:
await self.delete_messages(chat_id, message_ids)
self._outbox = PlatformOutbox(
self.outbound = DiscordMessenger(
get_client=lambda: self._client,
get_discord=_get_discord,
get_limiter=lambda: self._limiter,
send=send_operation,
edit=edit_operation,
delete=delete_operation,
delete_many=delete_many_operation,
)
self._voice_flow = VoiceNoteFlow(
voice_note_enabled=voice_note_enabled,
@ -181,76 +126,24 @@ class DiscordPlatform(MessagingPlatform):
self._log_api_error_tracebacks = log_api_error_tracebacks
async def _handle_client_message(self, message: Any) -> None:
"""Adapter entry point used by the internal discord client."""
"""Adapter entry point used by the internal Discord client."""
await self._on_discord_message(message)
async def _register_pending_voice(
self, chat_id: str, voice_msg_id: str, status_msg_id: str
) -> None:
"""Register a voice note as pending transcription."""
await self._voice_flow.register_pending_voice(
chat_id,
voice_msg_id,
status_msg_id,
)
async def cancel_pending_voice(
self, chat_id: str, reply_id: str
) -> tuple[str, str] | None:
"""Cancel a pending voice transcription. Returns (voice_msg_id, status_msg_id) if found."""
"""Cancel a pending voice transcription."""
return await self._voice_flow.cancel_pending_voice(chat_id, reply_id)
async def _is_voice_still_pending(self, chat_id: str, voice_msg_id: str) -> bool:
"""Check if a voice note is still pending (not cancelled)."""
return await self._voice_flow.is_voice_still_pending(chat_id, voice_msg_id)
def _get_audio_attachment(self, message: Any) -> Any | None:
"""Return first audio attachment, or None."""
for att in message.attachments:
if is_audio_metadata(att.filename, att.content_type):
return att
return None
async def _handle_voice_note(
self, message: Any, attachment: Any, channel_id: str
) -> bool:
"""Handle voice/audio attachment. Returns True if handled."""
user_id = str(message.author.id)
message_id = str(message.id)
reply_to = (
str(message.reference.message_id)
if message.reference and message.reference.message_id
else None
)
ct = attachment.content_type or "audio/ogg"
async def _download_to(tmp_path) -> None:
await attachment.save(str(tmp_path))
async def _reply_text(text: str) -> None:
await message.reply(text)
"""Handle a Discord audio attachment."""
return await self._voice_flow.handle(
VoiceNoteRequest(
platform="discord",
chat_id=channel_id,
user_id=user_id,
message_id=message_id,
raw_event=message,
content_type=ct,
temp_suffix=audio_suffix_from_metadata(
filename=attachment.filename,
content_type=attachment.content_type,
),
status_text=format_status_discord("Transcribing voice note..."),
reply_to_message_id=reply_to,
username=message.author.display_name,
download_to=_download_to,
reply_text=_reply_text,
),
discord_voice_request_from_event(message, attachment, channel_id),
message_handler=self._message_handler,
queue_send_message=self.queue_send_message,
queue_delete_message=self.queue_delete_message,
queue_send_message=self.outbound.queue_send_message,
queue_delete_message=self.outbound.queue_delete_message,
)
async def _on_discord_message(self, message: Any) -> None:
@ -259,61 +152,22 @@ class DiscordPlatform(MessagingPlatform):
return
channel_id = str(message.channel.id)
if not self.allowed_channel_ids or channel_id not in self.allowed_channel_ids:
return
# Handle voice/audio attachments when message has no text content
if not message.content:
audio_att = self._get_audio_attachment(message)
audio_att = get_audio_attachment(message)
if audio_att:
await self._handle_voice_note(message, audio_att, channel_id)
return
return
user_id = str(message.author.id)
message_id = str(message.id)
reply_to = (
str(message.reference.message_id)
if message.reference and message.reference.message_id
else None
incoming = discord_text_message_from_event(
message,
log_raw_messaging_content=self._log_raw_messaging_content,
)
raw_content = message.content or ""
if self._log_raw_messaging_content:
text_preview = raw_content[:80]
if len(raw_content) > 80:
text_preview += "..."
logger.info(
"DISCORD_MSG: chat_id={} message_id={} reply_to={} text_preview={!r}",
channel_id,
message_id,
reply_to,
text_preview,
)
else:
logger.info(
"DISCORD_MSG: chat_id={} message_id={} reply_to={} text_len={}",
channel_id,
message_id,
reply_to,
len(raw_content),
)
if not self._message_handler:
if self._message_handler is None:
return
incoming = IncomingMessage(
text=message.content,
chat_id=channel_id,
user_id=user_id,
message_id=message_id,
platform="discord",
reply_to_message_id=reply_to,
username=message.author.display_name,
raw_event=message,
)
try:
await self._message_handler(incoming)
except Exception as e:
@ -322,18 +176,12 @@ class DiscordPlatform(MessagingPlatform):
else:
logger.error("Error handling message: exc_type={}", type(e).__name__)
with contextlib.suppress(Exception):
await self.send_message(
await self.outbound.send_message(
channel_id,
format_status_discord("Error:", format_user_error_preview(e)),
reply_to=message_id,
reply_to=str(message.id),
)
def _truncate(self, text: str, limit: int = DISCORD_MESSAGE_LIMIT) -> str:
"""Truncate text to Discord's message limit."""
if len(text) <= limit:
return text
return text[: limit - 3] + "..."
async def start(self) -> None:
"""Initialize and connect to Discord."""
if not self.bot_token:
@ -352,7 +200,7 @@ class DiscordPlatform(MessagingPlatform):
)
max_wait = 30
waited = 0
waited = 0.0
while not self._connected and waited < max_wait:
await asyncio.sleep(0.5)
waited += 0.5
@ -363,7 +211,7 @@ class DiscordPlatform(MessagingPlatform):
logger.info("Discord platform started")
async def stop(self) -> None:
"""Stop the bot."""
"""Stop Discord SDK resources."""
if self._client.is_closed():
self._connected = False
return
@ -380,188 +228,11 @@ class DiscordPlatform(MessagingPlatform):
self._connected = False
logger.info("Discord platform stopped")
async def _send_message_raw(
self,
chat_id: str,
text: str,
reply_to: str | None = None,
parse_mode: str | None = None,
message_thread_id: str | None = None,
) -> str:
"""Send a message to a channel."""
channel = self._client.get_channel(int(chat_id))
if not channel or not hasattr(channel, "send"):
raise RuntimeError(f"Channel {chat_id} not found")
text = self._truncate(text)
channel = cast(Any, channel)
discord = _get_discord()
if reply_to:
ref = discord.MessageReference(
message_id=int(reply_to),
channel_id=int(chat_id),
)
msg = await channel.send(content=text, reference=ref)
else:
msg = await channel.send(content=text)
return str(msg.id)
async def send_message(
self,
chat_id: str,
text: str,
reply_to: str | None = None,
parse_mode: str | None = None,
message_thread_id: str | None = None,
) -> str:
"""Send a message to a channel."""
return await self._send_message_raw(
chat_id,
text,
reply_to,
parse_mode,
message_thread_id,
)
async def _edit_message_raw(
self,
chat_id: str,
message_id: str,
text: str,
parse_mode: str | None = None,
) -> None:
"""Edit an existing message."""
channel = self._client.get_channel(int(chat_id))
if not channel or not hasattr(channel, "fetch_message"):
raise RuntimeError(f"Channel {chat_id} not found")
discord = _get_discord()
channel = cast(Any, channel)
try:
msg = await channel.fetch_message(int(message_id))
except discord.NotFound:
return
text = self._truncate(text)
await msg.edit(content=text)
async def edit_message(
self,
chat_id: str,
message_id: str,
text: str,
parse_mode: str | None = None,
) -> None:
"""Edit an existing message."""
await self._edit_message_raw(chat_id, message_id, text, parse_mode)
async def _delete_message_raw(
self,
chat_id: str,
message_id: str,
) -> None:
"""Delete a message from a channel."""
channel = self._client.get_channel(int(chat_id))
if not channel or not hasattr(channel, "fetch_message"):
return
discord = _get_discord()
channel = cast(Any, channel)
try:
msg = await channel.fetch_message(int(message_id))
await msg.delete()
except discord.NotFound, discord.Forbidden:
pass
async def delete_message(
self,
chat_id: str,
message_id: str,
) -> None:
"""Delete a message from a channel."""
await self._delete_message_raw(chat_id, message_id)
async def _delete_messages_raw(self, chat_id: str, message_ids: list[str]) -> None:
"""Delete multiple messages (best-effort)."""
for mid in message_ids:
await self._delete_message_raw(chat_id, mid)
async def delete_messages(self, chat_id: str, message_ids: list[str]) -> None:
"""Delete multiple messages (best-effort)."""
await self._delete_messages_raw(chat_id, message_ids)
async def queue_send_message(
self,
chat_id: str,
text: str,
reply_to: str | None = None,
parse_mode: str | None = None,
fire_and_forget: bool = True,
message_thread_id: str | None = None,
) -> str | None:
"""Enqueue a message to be sent."""
return await self._outbox.queue_send_message(
chat_id,
text,
reply_to,
parse_mode,
fire_and_forget,
message_thread_id,
)
async def queue_edit_message(
self,
chat_id: str,
message_id: str,
text: str,
parse_mode: str | None = None,
fire_and_forget: bool = True,
) -> None:
"""Enqueue a message edit."""
await self._outbox.queue_edit_message(
chat_id,
message_id,
text,
parse_mode,
fire_and_forget,
)
async def queue_delete_message(
self,
chat_id: str,
message_id: str,
fire_and_forget: bool = True,
) -> None:
"""Enqueue a message delete."""
await self._outbox.queue_delete_message(chat_id, message_id, fire_and_forget)
async def queue_delete_messages(
self,
chat_id: str,
message_ids: list[str],
fire_and_forget: bool = True,
) -> None:
"""Enqueue a bulk delete."""
await self._outbox.queue_delete_messages(
chat_id,
message_ids,
fire_and_forget,
)
def fire_and_forget(self, task: Awaitable[Any]) -> None:
"""Execute a coroutine without awaiting it."""
self._outbox.fire_and_forget(task)
def on_message(
self,
handler: Callable[[IncomingMessage], Awaitable[None]],
) -> None:
"""Register a message handler callback."""
def on_message(self, handler: Callable[[IncomingMessage], Awaitable[None]]) -> None:
"""Register the workflow callback for inbound messages."""
self._message_handler = handler
@property
def is_connected(self) -> bool:
"""Check if connected."""
"""Return whether Discord startup completed."""
return self._connected

View file

@ -0,0 +1,114 @@
"""Discord inbound event normalization."""
from __future__ import annotations
from typing import Any
from loguru import logger
from ..models import IncomingMessage
from ..rendering.discord_markdown import format_status_discord
from .voice_flow import (
VoiceNoteRequest,
audio_suffix_from_metadata,
is_audio_metadata,
)
def parse_allowed_channels(raw: str | None) -> set[str]:
"""Parse comma-separated Discord channel IDs."""
if not raw or not raw.strip():
return set()
return {s.strip() for s in raw.split(",") if s.strip()}
def get_audio_attachment(message: Any) -> Any | None:
"""Return the first audio attachment from a Discord message."""
for att in message.attachments:
if is_audio_metadata(att.filename, att.content_type):
return att
return None
def discord_text_message_from_event(
message: Any,
*,
log_raw_messaging_content: bool,
) -> IncomingMessage:
"""Normalize a Discord message into an incoming text message."""
channel_id = str(message.channel.id)
message_id = str(message.id)
reply_to = (
str(message.reference.message_id)
if message.reference and message.reference.message_id
else None
)
raw_content = message.content or ""
if log_raw_messaging_content:
text_preview = raw_content[:80]
if len(raw_content) > 80:
text_preview += "..."
logger.info(
"DISCORD_MSG: chat_id={} message_id={} reply_to={} text_preview={!r}",
channel_id,
message_id,
reply_to,
text_preview,
)
else:
logger.info(
"DISCORD_MSG: chat_id={} message_id={} reply_to={} text_len={}",
channel_id,
message_id,
reply_to,
len(raw_content),
)
return IncomingMessage(
text=message.content,
chat_id=channel_id,
user_id=str(message.author.id),
message_id=message_id,
platform="discord",
reply_to_message_id=reply_to,
username=message.author.display_name,
raw_event=message,
)
def discord_voice_request_from_event(
message: Any,
attachment: Any,
channel_id: str,
) -> VoiceNoteRequest:
"""Normalize a Discord voice/audio attachment into a voice-note request."""
message_id = str(message.id)
reply_to = (
str(message.reference.message_id)
if message.reference and message.reference.message_id
else None
)
async def _download_to(tmp_path) -> None:
await attachment.save(str(tmp_path))
async def _reply_text(text: str) -> None:
await message.reply(text)
return VoiceNoteRequest(
platform="discord",
chat_id=channel_id,
user_id=str(message.author.id),
message_id=message_id,
raw_event=message,
content_type=attachment.content_type or "audio/ogg",
temp_suffix=audio_suffix_from_metadata(
filename=attachment.filename,
content_type=attachment.content_type,
),
status_text=format_status_discord("Transcribing voice note..."),
reply_to_message_id=reply_to,
username=message.author.display_name,
download_to=_download_to,
reply_text=_reply_text,
)

View file

@ -0,0 +1,175 @@
"""Discord outbound delivery."""
from __future__ import annotations
from collections.abc import Awaitable, Callable
from typing import Any, cast
from .outbox import PlatformOutbox
DISCORD_MESSAGE_LIMIT = 2000
ClientGetter = Callable[[], Any]
DiscordGetter = Callable[[], Any]
LimiterGetter = Callable[[], Any | None]
def truncate_discord_message(text: str, limit: int = DISCORD_MESSAGE_LIMIT) -> str:
"""Return text that fits Discord's message limit."""
if len(text) <= limit:
return text
return text[: limit - 3] + "..."
class DiscordMessenger:
"""Owns Discord sends, edits, deletes, and queued delivery."""
def __init__(
self,
*,
get_client: ClientGetter,
get_discord: DiscordGetter,
get_limiter: LimiterGetter,
) -> None:
self._get_client = get_client
self._get_discord = get_discord
self._outbox = PlatformOutbox(
get_limiter=get_limiter,
send=self.send_message,
edit=self.edit_message,
delete=self.delete_message,
delete_many=self.delete_messages,
)
async def send_message(
self,
chat_id: str,
text: str,
reply_to: str | None = None,
parse_mode: str | None = None,
message_thread_id: str | None = None,
) -> str:
"""Send a Discord message immediately."""
client = self._get_client()
channel = client.get_channel(int(chat_id))
if not channel or not hasattr(channel, "send"):
raise RuntimeError(f"Channel {chat_id} not found")
text = truncate_discord_message(text)
channel = cast(Any, channel)
if reply_to:
discord = self._get_discord()
ref = discord.MessageReference(
message_id=int(reply_to),
channel_id=int(chat_id),
)
msg = await channel.send(content=text, reference=ref)
else:
msg = await channel.send(content=text)
return str(msg.id)
async def edit_message(
self,
chat_id: str,
message_id: str,
text: str,
parse_mode: str | None = None,
) -> None:
"""Edit a Discord message immediately."""
client = self._get_client()
channel = client.get_channel(int(chat_id))
if not channel or not hasattr(channel, "fetch_message"):
raise RuntimeError(f"Channel {chat_id} not found")
discord = self._get_discord()
channel = cast(Any, channel)
try:
msg = await channel.fetch_message(int(message_id))
except discord.NotFound:
return
await msg.edit(content=truncate_discord_message(text))
async def delete_message(self, chat_id: str, message_id: str) -> None:
"""Delete a Discord message immediately."""
client = self._get_client()
channel = client.get_channel(int(chat_id))
if not channel or not hasattr(channel, "fetch_message"):
return
discord = self._get_discord()
channel = cast(Any, channel)
try:
msg = await channel.fetch_message(int(message_id))
await msg.delete()
except discord.NotFound, discord.Forbidden:
pass
async def delete_messages(self, chat_id: str, message_ids: list[str]) -> None:
"""Delete multiple Discord messages best-effort."""
for mid in message_ids:
await self.delete_message(chat_id, mid)
async def queue_send_message(
self,
chat_id: str,
text: str,
reply_to: str | None = None,
parse_mode: str | None = None,
fire_and_forget: bool = True,
message_thread_id: str | None = None,
) -> str | None:
"""Queue a Discord send."""
return await self._outbox.queue_send_message(
chat_id,
text,
reply_to,
parse_mode,
fire_and_forget,
message_thread_id,
)
async def queue_edit_message(
self,
chat_id: str,
message_id: str,
text: str,
parse_mode: str | None = None,
fire_and_forget: bool = True,
) -> None:
"""Queue a Discord edit."""
await self._outbox.queue_edit_message(
chat_id,
message_id,
text,
parse_mode,
fire_and_forget,
)
async def queue_delete_message(
self,
chat_id: str,
message_id: str,
fire_and_forget: bool = True,
) -> None:
"""Queue a Discord delete."""
await self._outbox.queue_delete_message(chat_id, message_id, fire_and_forget)
async def queue_delete_messages(
self,
chat_id: str,
message_ids: list[str],
fire_and_forget: bool = True,
) -> None:
"""Queue a Discord bulk delete."""
await self._outbox.queue_delete_messages(
chat_id,
message_ids,
fire_and_forget,
)
def fire_and_forget(self, task: Awaitable[Any]) -> None:
"""Execute a coroutine without awaiting it."""
self._outbox.fire_and_forget(task)

View file

@ -1,10 +1,4 @@
"""Messaging platform factory.
Creates the appropriate messaging platform adapter based on configuration.
To add a new platform (e.g. Discord, Slack):
1. Create a new class implementing MessagingPlatform in messaging/platforms/
2. Add a case to create_messaging_platform() below
"""
"""Messaging platform component factory."""
from __future__ import annotations
@ -12,12 +6,12 @@ from dataclasses import dataclass
from loguru import logger
from .base import MessagingPlatform
from .ports import MessagingPlatformComponents
@dataclass(frozen=True, slots=True)
class MessagingPlatformOptions:
"""Typed wiring from :class:`~api.runtime.AppRuntime` into platform adapters."""
"""Typed wiring from app settings into messaging platform runtimes."""
telegram_bot_token: str | None = None
allowed_telegram_user_id: str | None = None
@ -34,19 +28,11 @@ class MessagingPlatformOptions:
log_api_error_tracebacks: bool = False
def create_messaging_platform(
def create_messaging_components(
platform_type: str,
options: MessagingPlatformOptions | None = None,
) -> MessagingPlatform | None:
"""Create a messaging platform instance based on type.
Args:
platform_type: Platform identifier (``telegram``, ``discord``, ``none``).
options: Token, allowlist, and voice / transcription settings.
Returns:
Configured :class:`MessagingPlatform` instance, or None if not configured.
"""
) -> MessagingPlatformComponents | None:
"""Create runtime/outbound components for the configured messaging platform."""
opts = options or MessagingPlatformOptions()
if platform_type == "none":
logger.info("Messaging platform disabled by configuration")
@ -58,9 +44,9 @@ def create_messaging_platform(
logger.info("No Telegram bot token configured, skipping platform setup")
return None
from .telegram import TelegramPlatform
from .telegram import TelegramRuntime
return TelegramPlatform(
runtime = TelegramRuntime(
bot_token=bot_token,
allowed_user_id=opts.allowed_telegram_user_id,
voice_note_enabled=opts.voice_note_enabled,
@ -73,6 +59,12 @@ def create_messaging_platform(
log_raw_messaging_content=opts.log_raw_messaging_content,
log_api_error_tracebacks=opts.log_api_error_tracebacks,
)
return MessagingPlatformComponents(
name=runtime.name,
runtime=runtime,
outbound=runtime.outbound,
voice_cancellation=runtime,
)
if platform_type == "discord":
bot_token = opts.discord_bot_token
@ -80,9 +72,9 @@ def create_messaging_platform(
logger.info("No Discord bot token configured, skipping platform setup")
return None
from .discord import DiscordPlatform
from .discord import DiscordRuntime
return DiscordPlatform(
runtime = DiscordRuntime(
bot_token=bot_token,
allowed_channel_ids=opts.allowed_discord_channels,
voice_note_enabled=opts.voice_note_enabled,
@ -95,9 +87,15 @@ def create_messaging_platform(
log_raw_messaging_content=opts.log_raw_messaging_content,
log_api_error_tracebacks=opts.log_api_error_tracebacks,
)
return MessagingPlatformComponents(
name=runtime.name,
runtime=runtime,
outbound=runtime.outbound,
voice_cancellation=runtime,
)
logger.warning(
f"Unknown messaging platform: '{platform_type}'. "
"Supported: 'none', 'telegram', 'discord'"
"Unknown messaging platform: '{}'. Supported: 'none', 'telegram', 'discord'",
platform_type,
)
return None

View file

@ -0,0 +1,87 @@
"""Messaging platform ports used by the customer-facing workflow."""
from __future__ import annotations
from collections.abc import Awaitable, Callable
from dataclasses import dataclass
from typing import Any, Protocol, runtime_checkable
from ..models import IncomingMessage
InboundMessageHandler = Callable[[IncomingMessage], Awaitable[None]]
@runtime_checkable
class MessagingRuntime(Protocol):
"""Owns inbound SDK lifecycle for one messaging platform."""
@property
def name(self) -> str: ...
async def start(self) -> None: ...
async def stop(self) -> None: ...
def on_message(self, handler: InboundMessageHandler) -> None: ...
@property
def is_connected(self) -> bool: ...
@runtime_checkable
class OutboundMessenger(Protocol):
"""Owns queued outbound platform delivery."""
async def queue_send_message(
self,
chat_id: str,
text: str,
reply_to: str | None = None,
parse_mode: str | None = None,
fire_and_forget: bool = True,
message_thread_id: str | None = None,
) -> str | None: ...
async def queue_edit_message(
self,
chat_id: str,
message_id: str,
text: str,
parse_mode: str | None = None,
fire_and_forget: bool = True,
) -> None: ...
async def queue_delete_message(
self,
chat_id: str,
message_id: str,
fire_and_forget: bool = True,
) -> None: ...
async def queue_delete_messages(
self,
chat_id: str,
message_ids: list[str],
fire_and_forget: bool = True,
) -> None: ...
def fire_and_forget(self, task: Awaitable[Any]) -> None: ...
@runtime_checkable
class VoiceCancellation(Protocol):
"""Optional voice-note cancellation port used by /clear replies."""
async def cancel_pending_voice(
self, chat_id: str, reply_id: str
) -> tuple[str, str] | None: ...
@dataclass(frozen=True, slots=True)
class MessagingPlatformComponents:
"""Runtime/outbound bundle for one configured messaging platform."""
name: str
runtime: MessagingRuntime
outbound: OutboundMessenger
voice_cancellation: VoiceCancellation | None = None

View file

@ -1,38 +1,35 @@
"""
Telegram Platform Adapter
"""Telegram messaging runtime."""
Implements MessagingPlatform for Telegram using python-telegram-bot.
"""
from __future__ import annotations
import asyncio
import contextlib
import os
# Opt-in to future behavior for python-telegram-bot (retry_after as timedelta)
# This must be set BEFORE importing telegram.error
os.environ["PTB_TIMEDELTA"] = "1"
from collections.abc import Awaitable, Callable
from typing import TYPE_CHECKING, Any
# Opt-in to future behavior for python-telegram-bot (retry_after as timedelta).
os.environ["PTB_TIMEDELTA"] = "1"
from loguru import logger
from core.anthropic import format_user_error_preview
from ..models import IncomingMessage
from ..rendering.telegram_markdown import escape_md_v2
from .ports import InboundMessageHandler
from .telegram_inbound import (
telegram_text_message_from_update,
telegram_voice_request_from_update,
)
from .telegram_io import TelegramMessenger
from .voice_flow import VoiceNoteFlow
if TYPE_CHECKING:
from telegram import Update
from telegram.ext import ContextTypes
from ..models import IncomingMessage
from ..rendering.telegram_markdown import escape_md_v2, format_status
from .base import MessagingPlatform
from .outbox import PlatformOutbox
from .voice_flow import VoiceNoteFlow, VoiceNoteRequest, audio_suffix_from_metadata
# Optional import - python-telegram-bot may not be installed
try:
from telegram import Update
from telegram.error import NetworkError, RetryAfter, TelegramError
from telegram.ext import (
Application,
CommandHandler,
@ -47,13 +44,8 @@ except ImportError:
TELEGRAM_AVAILABLE = False
class TelegramPlatform(MessagingPlatform):
"""
Telegram messaging platform adapter.
Uses python-telegram-bot (BoT API) for Telegram access.
Requires a Bot Token from @BotFather.
"""
class TelegramRuntime:
"""Owns Telegram SDK lifecycle and inbound event handoff."""
name = "telegram"
@ -71,7 +63,7 @@ class TelegramPlatform(MessagingPlatform):
messaging_rate_window: float = 1.0,
log_raw_messaging_content: bool = False,
log_api_error_tracebacks: bool = False,
):
) -> None:
if not TELEGRAM_AVAILABLE:
raise ImportError(
"python-telegram-bot is required. Install with: pip install python-telegram-bot"
@ -79,57 +71,16 @@ class TelegramPlatform(MessagingPlatform):
self.bot_token = bot_token
self.allowed_user_id = allowed_user_id
if not self.bot_token:
# We don't raise here to allow instantiation for testing/conditional logic,
# but start() will fail.
logger.warning("TELEGRAM_BOT_TOKEN not set")
self._application: Application | None = None
self._message_handler: Callable[[IncomingMessage], Awaitable[None]] | None = (
None
)
self._message_handler: InboundMessageHandler | None = None
self._connected = False
self._limiter: Any | None = None # Will be MessagingRateLimiter
async def send_operation(
chat_id: str,
text: str,
reply_to: str | None,
parse_mode: str | None,
message_thread_id: str | None,
) -> str:
return await self.send_message(
chat_id,
text,
reply_to,
parse_mode,
message_thread_id,
)
async def edit_operation(
chat_id: str,
message_id: str,
text: str,
parse_mode: str | None,
) -> None:
await self.edit_message(chat_id, message_id, text, parse_mode)
async def delete_operation(chat_id: str, message_id: str) -> None:
await self.delete_message(chat_id, message_id)
async def delete_many_operation(
chat_id: str,
message_ids: list[str],
) -> None:
await self.delete_messages(chat_id, message_ids)
self._outbox = PlatformOutbox(
self._limiter: Any | None = None
self.outbound = TelegramMessenger(
get_application=lambda: self._application,
get_limiter=lambda: self._limiter,
send=send_operation,
edit=edit_operation,
delete=delete_operation,
delete_many=delete_many_operation,
)
self._voice_flow = VoiceNoteFlow(
voice_note_enabled=voice_note_enabled,
@ -145,82 +96,60 @@ class TelegramPlatform(MessagingPlatform):
self._log_raw_messaging_content = log_raw_messaging_content
self._log_api_error_tracebacks = log_api_error_tracebacks
async def _register_pending_voice(
self, chat_id: str, voice_msg_id: str, status_msg_id: str
) -> None:
"""Register a voice note as pending transcription (for /clear reply during transcription)."""
await self._voice_flow.register_pending_voice(
chat_id,
voice_msg_id,
status_msg_id,
)
async def cancel_pending_voice(
self, chat_id: str, reply_id: str
) -> tuple[str, str] | None:
"""Cancel a pending voice transcription. Returns (voice_msg_id, status_msg_id) if found."""
"""Cancel a pending voice transcription."""
return await self._voice_flow.cancel_pending_voice(chat_id, reply_id)
async def _is_voice_still_pending(self, chat_id: str, voice_msg_id: str) -> bool:
"""Check if a voice note is still pending (not cancelled)."""
return await self._voice_flow.is_voice_still_pending(chat_id, voice_msg_id)
async def start(self) -> None:
"""Initialize and connect to Telegram."""
if not self.bot_token:
raise ValueError("TELEGRAM_BOT_TOKEN is required")
# Configure request with longer timeouts
request = HTTPXRequest(
connection_pool_size=8, connect_timeout=30.0, read_timeout=30.0
)
# Build Application
builder = Application.builder().token(self.bot_token).request(request)
self._application = builder.build()
# Register Internal Handlers
# We catch ALL text messages and commands to forward them
self._application.add_handler(
MessageHandler(filters.TEXT & (~filters.COMMAND), self._on_telegram_message)
)
self._application.add_handler(CommandHandler("start", self._on_start_command))
# Catch-all for other commands if needed, or let them fall through
self._application.add_handler(
MessageHandler(filters.COMMAND, self._on_telegram_message)
)
# Voice note handler
self._application.add_handler(
MessageHandler(filters.VOICE, self._on_telegram_voice)
)
# Initialize internal components with retry logic
max_retries = 3
for attempt in range(max_retries):
try:
await self._application.initialize()
await self._application.start()
# Start polling (non-blocking way for integration)
if self._application.updater:
await self._application.updater.start_polling(
drop_pending_updates=False
)
self._connected = True
break
except (NetworkError, Exception) as e:
except Exception as e:
if attempt < max_retries - 1:
wait_time = 2 * (attempt + 1)
logger.warning(
f"Connection failed (attempt {attempt + 1}/{max_retries}): {e}. Retrying in {wait_time}s..."
"Connection failed (attempt {}/{}): {}. Retrying in {}s...",
attempt + 1,
max_retries,
e,
wait_time,
)
await asyncio.sleep(wait_time)
else:
logger.error(f"Failed to connect after {max_retries} attempts")
logger.error("Failed to connect after {} attempts", max_retries)
raise
# Initialize rate limiter
from ..limiter import MessagingRateLimiter
self._limiter = await MessagingRateLimiter.get_instance(
@ -228,7 +157,6 @@ class TelegramPlatform(MessagingPlatform):
rate_window=self._messaging_rate_window,
)
# Send startup notification
try:
target = self.allowed_user_id
if target:
@ -236,10 +164,7 @@ class TelegramPlatform(MessagingPlatform):
f"🚀 *{escape_md_v2('Claude Code Proxy is online!')}* "
f"{escape_md_v2('(Bot API)')}"
)
await self.send_message(
target,
startup_text,
)
await self.outbound.send_message(target, startup_text)
except Exception as e:
if self._log_api_error_tracebacks:
logger.warning("Could not send startup message: {}", e)
@ -252,7 +177,7 @@ class TelegramPlatform(MessagingPlatform):
logger.info("Telegram platform started (Bot API)")
async def stop(self) -> None:
"""Stop the bot."""
"""Stop Telegram polling and SDK resources."""
if self._application and self._application.updater:
await self._application.updater.stop()
await self._application.stop()
@ -261,350 +186,33 @@ class TelegramPlatform(MessagingPlatform):
self._connected = False
logger.info("Telegram platform stopped")
async def _with_retry(
self, func: Callable[..., Awaitable[Any]], *args, **kwargs
) -> Any:
"""Helper to execute a function with exponential backoff on network errors."""
max_retries = 3
for attempt in range(max_retries):
try:
return await func(*args, **kwargs)
except (TimeoutError, NetworkError) as e:
if "Message is not modified" in str(e):
return None
if attempt < max_retries - 1:
wait_time = 2**attempt # 1s, 2s, 4s
logger.warning(
f"Telegram API network error (attempt {attempt + 1}/{max_retries}): {e}. Retrying in {wait_time}s..."
)
await asyncio.sleep(wait_time)
else:
logger.error(
f"Telegram API failed after {max_retries} attempts: {e}"
)
raise
except RetryAfter as e:
# Telegram explicitly tells us to wait (PTB_TIMEDELTA: retry_after is timedelta)
from datetime import timedelta
retry_after = e.retry_after
if isinstance(retry_after, timedelta):
wait_secs = retry_after.total_seconds()
else:
wait_secs = float(retry_after)
logger.warning(f"Rate limited by Telegram, waiting {wait_secs}s...")
await asyncio.sleep(wait_secs)
# We don't increment attempt here, as this is a specific instruction
return await func(*args, **kwargs)
except TelegramError as e:
# Non-network Telegram errors
err_lower = str(e).lower()
if "message is not modified" in err_lower:
return None
# Best-effort no-op cases (common during chat cleanup / /clear).
if any(
x in err_lower
for x in [
"message to edit not found",
"message to delete not found",
"message can't be deleted",
"message can't be edited",
"not enough rights to delete",
]
):
return None
if "Can't parse entities" in str(e) and kwargs.get("parse_mode"):
logger.warning("Markdown failed, retrying without parse_mode")
kwargs["parse_mode"] = None
return await func(*args, **kwargs)
raise
async def _send_message_raw(
self,
chat_id: str,
text: str,
reply_to: str | None = None,
parse_mode: str | None = "MarkdownV2",
message_thread_id: str | None = None,
) -> str:
"""Send a message to a chat."""
app = self._application
if not app or not app.bot:
raise RuntimeError("Telegram application or bot not initialized")
async def _do_send(parse_mode=parse_mode):
bot = app.bot
kwargs: dict[str, Any] = {
"chat_id": chat_id,
"text": text,
"reply_to_message_id": int(reply_to) if reply_to else None,
"parse_mode": parse_mode,
}
if message_thread_id is not None:
kwargs["message_thread_id"] = int(message_thread_id)
msg = await bot.send_message(**kwargs)
return str(msg.message_id)
return await self._with_retry(_do_send, parse_mode=parse_mode)
async def send_message(
self,
chat_id: str,
text: str,
reply_to: str | None = None,
parse_mode: str | None = "MarkdownV2",
message_thread_id: str | None = None,
) -> str:
"""Send a message to a chat."""
return await self._send_message_raw(
chat_id,
text,
reply_to,
parse_mode,
message_thread_id,
)
async def _edit_message_raw(
self,
chat_id: str,
message_id: str,
text: str,
parse_mode: str | None = "MarkdownV2",
) -> None:
"""Edit an existing message."""
app = self._application
if not app or not app.bot:
raise RuntimeError("Telegram application or bot not initialized")
async def _do_edit(parse_mode=parse_mode):
bot = app.bot
await bot.edit_message_text(
chat_id=chat_id,
message_id=int(message_id),
text=text,
parse_mode=parse_mode,
)
await self._with_retry(_do_edit, parse_mode=parse_mode)
async def edit_message(
self,
chat_id: str,
message_id: str,
text: str,
parse_mode: str | None = "MarkdownV2",
) -> None:
"""Edit an existing message."""
await self._edit_message_raw(chat_id, message_id, text, parse_mode)
async def _delete_message_raw(
self,
chat_id: str,
message_id: str,
) -> None:
"""Delete a message from a chat."""
app = self._application
if not app or not app.bot:
raise RuntimeError("Telegram application or bot not initialized")
async def _do_delete():
bot = app.bot
await bot.delete_message(chat_id=chat_id, message_id=int(message_id))
await self._with_retry(_do_delete)
async def delete_message(
self,
chat_id: str,
message_id: str,
) -> None:
"""Delete a message from a chat."""
await self._delete_message_raw(chat_id, message_id)
async def _delete_messages_raw(self, chat_id: str, message_ids: list[str]) -> None:
"""Delete multiple messages (best-effort)."""
if not message_ids:
return
app = self._application
if not app or not app.bot:
raise RuntimeError("Telegram application or bot not initialized")
# PTB supports bulk deletion via delete_messages; fall back to per-message.
bot = app.bot
if hasattr(bot, "delete_messages"):
async def _do_bulk():
mids = []
for mid in message_ids:
try:
mids.append(int(mid))
except Exception:
continue
if not mids:
return None
# delete_messages accepts a sequence of ints (up to 100).
await bot.delete_messages(chat_id=chat_id, message_ids=mids)
await self._with_retry(_do_bulk)
return
for mid in message_ids:
await self._delete_message_raw(chat_id, mid)
async def delete_messages(self, chat_id: str, message_ids: list[str]) -> None:
"""Delete multiple messages (best-effort)."""
await self._delete_messages_raw(chat_id, message_ids)
async def queue_send_message(
self,
chat_id: str,
text: str,
reply_to: str | None = None,
parse_mode: str | None = "MarkdownV2",
fire_and_forget: bool = True,
message_thread_id: str | None = None,
) -> str | None:
"""Enqueue a message to be sent (using limiter)."""
return await self._outbox.queue_send_message(
chat_id,
text,
reply_to,
parse_mode,
fire_and_forget,
message_thread_id,
)
async def queue_edit_message(
self,
chat_id: str,
message_id: str,
text: str,
parse_mode: str | None = "MarkdownV2",
fire_and_forget: bool = True,
) -> None:
"""Enqueue a message edit."""
await self._outbox.queue_edit_message(
chat_id,
message_id,
text,
parse_mode,
fire_and_forget,
)
async def queue_delete_message(
self,
chat_id: str,
message_id: str,
fire_and_forget: bool = True,
) -> None:
"""Enqueue a message delete."""
await self._outbox.queue_delete_message(chat_id, message_id, fire_and_forget)
async def queue_delete_messages(
self,
chat_id: str,
message_ids: list[str],
fire_and_forget: bool = True,
) -> None:
"""Enqueue a bulk delete (if supported) or a sequence of deletes."""
await self._outbox.queue_delete_messages(
chat_id,
message_ids,
fire_and_forget,
)
def fire_and_forget(self, task: Awaitable[Any]) -> None:
"""Execute a coroutine without awaiting it."""
self._outbox.fire_and_forget(task)
def on_message(
self,
handler: Callable[[IncomingMessage], Awaitable[None]],
) -> None:
"""Register a message handler callback."""
def on_message(self, handler: Callable[[IncomingMessage], Awaitable[None]]) -> None:
"""Register the workflow callback for inbound messages."""
self._message_handler = handler
@property
def is_connected(self) -> bool:
"""Check if connected."""
"""Return whether Telegram startup completed."""
return self._connected
async def _on_start_command(
self, update: Update, context: ContextTypes.DEFAULT_TYPE
) -> None:
"""Handle /start command."""
if update.message:
await update.message.reply_text("👋 Hello! I am the Claude Code Proxy Bot.")
# We can also treat this as a message if we want it to trigger something
await self._on_telegram_message(update, context)
async def _on_telegram_message(
self, update: Update, context: ContextTypes.DEFAULT_TYPE
) -> None:
"""Handle incoming updates."""
if (
not update.message
or not update.message.text
or not update.effective_user
or not update.effective_chat
):
return
user_id = str(update.effective_user.id)
chat_id = str(update.effective_chat.id)
# Security check
if self.allowed_user_id and user_id != str(self.allowed_user_id).strip():
logger.warning(f"Unauthorized access attempt from {user_id}")
return
message_id = str(update.message.message_id)
reply_to = (
str(update.message.reply_to_message.message_id)
if update.message.reply_to_message
else None
incoming = telegram_text_message_from_update(
update,
allowed_user_id=self.allowed_user_id,
log_raw_messaging_content=self._log_raw_messaging_content,
)
thread_id = (
str(update.message.message_thread_id)
if getattr(update.message, "message_thread_id", None) is not None
else None
)
raw_text = update.message.text or ""
if self._log_raw_messaging_content:
text_preview = raw_text[:80]
if len(raw_text) > 80:
text_preview += "..."
logger.info(
"TELEGRAM_MSG: chat_id={} message_id={} reply_to={} text_preview={!r}",
chat_id,
message_id,
reply_to,
text_preview,
)
else:
logger.info(
"TELEGRAM_MSG: chat_id={} message_id={} reply_to={} text_len={}",
chat_id,
message_id,
reply_to,
len(raw_text),
)
if not self._message_handler:
if incoming is None or self._message_handler is None:
return
incoming = IncomingMessage(
text=update.message.text,
chat_id=chat_id,
user_id=user_id,
message_id=message_id,
platform="telegram",
reply_to_message_id=reply_to,
message_thread_id=thread_id,
raw_event=update,
)
try:
await self._message_handler(incoming)
except Exception as e:
@ -613,76 +221,37 @@ class TelegramPlatform(MessagingPlatform):
else:
logger.error("Error handling message: exc_type={}", type(e).__name__)
with contextlib.suppress(Exception):
await self.send_message(
chat_id,
await self.outbound.send_message(
incoming.chat_id,
f"❌ *{escape_md_v2('Error:')}* {escape_md_v2(format_user_error_preview(e))}",
reply_to=incoming.message_id,
message_thread_id=thread_id,
message_thread_id=incoming.message_thread_id,
parse_mode="MarkdownV2",
)
async def _on_telegram_voice(
self, update: Update, context: ContextTypes.DEFAULT_TYPE
) -> None:
"""Handle incoming voice messages."""
message = update.message
effective_user = update.effective_user
effective_chat = update.effective_chat
if (
message is None
or message.voice is None
or effective_user is None
or effective_chat is None
):
return
voice = message.voice
async def _reply_text(text: str) -> None:
await message.reply_text(text)
if message is not None:
await message.reply_text(text)
if await self._voice_flow.reply_if_disabled(_reply_text):
return
user_id = str(effective_user.id)
chat_id = str(effective_chat.id)
if self.allowed_user_id and user_id != str(self.allowed_user_id).strip():
logger.warning(f"Unauthorized voice access attempt from {user_id}")
request = telegram_voice_request_from_update(
update,
context,
allowed_user_id=self.allowed_user_id,
)
if request is None:
return
thread_id = (
str(message.message_thread_id)
if getattr(message, "message_thread_id", None) is not None
else None
)
message_id = str(message.message_id)
reply_to = (
str(message.reply_to_message.message_id)
if message.reply_to_message
else None
)
async def _download_to(tmp_path) -> None:
tg_file = await context.bot.get_file(voice.file_id)
await tg_file.download_to_drive(custom_path=str(tmp_path))
await self._voice_flow.handle(
VoiceNoteRequest(
platform="telegram",
chat_id=chat_id,
user_id=user_id,
message_id=message_id,
raw_event=update,
content_type=voice.mime_type or "audio/ogg",
temp_suffix=audio_suffix_from_metadata(content_type=voice.mime_type),
status_text=format_status("", "Transcribing voice note..."),
status_parse_mode="MarkdownV2",
message_thread_id=thread_id,
reply_to_message_id=reply_to,
download_to=_download_to,
reply_text=_reply_text,
),
request,
message_handler=self._message_handler,
queue_send_message=self.queue_send_message,
queue_delete_message=self.queue_delete_message,
queue_send_message=self.outbound.queue_send_message,
queue_delete_message=self.outbound.queue_delete_message,
)

View file

@ -0,0 +1,139 @@
"""Telegram inbound event normalization."""
from __future__ import annotations
from typing import TYPE_CHECKING
from loguru import logger
from ..models import IncomingMessage
from ..rendering.telegram_markdown import format_status
from .voice_flow import VoiceNoteRequest, audio_suffix_from_metadata
if TYPE_CHECKING:
from telegram import Update
from telegram.ext import ContextTypes
def telegram_text_message_from_update(
update: Update,
*,
allowed_user_id: str | None,
log_raw_messaging_content: bool,
) -> IncomingMessage | None:
"""Normalize a Telegram text update into an incoming message."""
if (
not update.message
or not update.message.text
or not update.effective_user
or not update.effective_chat
):
return None
user_id = str(update.effective_user.id)
chat_id = str(update.effective_chat.id)
if allowed_user_id and user_id != str(allowed_user_id).strip():
logger.warning("Unauthorized access attempt from {}", user_id)
return None
message = update.message
message_id = str(message.message_id)
reply_to = (
str(message.reply_to_message.message_id) if message.reply_to_message else None
)
thread_id = (
str(message.message_thread_id)
if getattr(message, "message_thread_id", None) is not None
else None
)
raw_text = message.text or ""
if log_raw_messaging_content:
text_preview = raw_text[:80]
if len(raw_text) > 80:
text_preview += "..."
logger.info(
"TELEGRAM_MSG: chat_id={} message_id={} reply_to={} text_preview={!r}",
chat_id,
message_id,
reply_to,
text_preview,
)
else:
logger.info(
"TELEGRAM_MSG: chat_id={} message_id={} reply_to={} text_len={}",
chat_id,
message_id,
reply_to,
len(raw_text),
)
return IncomingMessage(
text=raw_text,
chat_id=chat_id,
user_id=user_id,
message_id=message_id,
platform="telegram",
reply_to_message_id=reply_to,
message_thread_id=thread_id,
raw_event=update,
)
def telegram_voice_request_from_update(
update: Update,
context: ContextTypes.DEFAULT_TYPE,
*,
allowed_user_id: str | None,
) -> VoiceNoteRequest | None:
"""Normalize a Telegram voice update into a voice-note request."""
message = update.message
effective_user = update.effective_user
effective_chat = update.effective_chat
if (
message is None
or message.voice is None
or effective_user is None
or effective_chat is None
):
return None
user_id = str(effective_user.id)
if allowed_user_id and user_id != str(allowed_user_id).strip():
logger.warning("Unauthorized voice access attempt from {}", user_id)
return None
voice = message.voice
chat_id = str(effective_chat.id)
message_id = str(message.message_id)
thread_id = (
str(message.message_thread_id)
if getattr(message, "message_thread_id", None) is not None
else None
)
reply_to = (
str(message.reply_to_message.message_id) if message.reply_to_message else None
)
async def _download_to(tmp_path) -> None:
tg_file = await context.bot.get_file(voice.file_id)
await tg_file.download_to_drive(custom_path=str(tmp_path))
async def _reply_text(text: str) -> None:
await message.reply_text(text)
return VoiceNoteRequest(
platform="telegram",
chat_id=chat_id,
user_id=user_id,
message_id=message_id,
raw_event=update,
content_type=voice.mime_type or "audio/ogg",
temp_suffix=audio_suffix_from_metadata(content_type=voice.mime_type),
status_text=format_status("", "Transcribing voice note..."),
status_parse_mode="MarkdownV2",
message_thread_id=thread_id,
reply_to_message_id=reply_to,
username=None,
download_to=_download_to,
reply_text=_reply_text,
)

View file

@ -0,0 +1,266 @@
"""Telegram outbound delivery."""
from __future__ import annotations
import asyncio
from collections.abc import Awaitable, Callable
from datetime import timedelta
from typing import Any
from loguru import logger
from .outbox import PlatformOutbox
TelegramNetworkError: type[BaseException]
TelegramRetryAfter: type[BaseException]
TelegramBaseError: type[BaseException]
try:
from telegram.error import (
NetworkError as _TelegramNetworkError,
)
from telegram.error import (
RetryAfter as _TelegramRetryAfter,
)
from telegram.error import (
TelegramError as _TelegramBaseError,
)
TelegramNetworkError = _TelegramNetworkError
TelegramRetryAfter = _TelegramRetryAfter
TelegramBaseError = _TelegramBaseError
except ImportError:
TelegramNetworkError = TimeoutError
TelegramRetryAfter = TimeoutError
TelegramBaseError = Exception
ApplicationGetter = Callable[[], Any | None]
LimiterGetter = Callable[[], Any | None]
class TelegramMessenger:
"""Owns Telegram sends, edits, deletes, and queued delivery."""
def __init__(
self,
*,
get_application: ApplicationGetter,
get_limiter: LimiterGetter,
) -> None:
self._get_application = get_application
self._outbox = PlatformOutbox(
get_limiter=get_limiter,
send=self.send_message,
edit=self.edit_message,
delete=self.delete_message,
delete_many=self.delete_messages,
)
async def _with_retry(
self, func: Callable[..., Awaitable[Any]], *args: Any, **kwargs: Any
) -> Any:
"""Execute a Telegram API call with the platform retry policy."""
max_retries = 3
for attempt in range(max_retries):
try:
return await func(*args, **kwargs)
except (TimeoutError, TelegramNetworkError) as e:
if "Message is not modified" in str(e):
return None
if attempt < max_retries - 1:
wait_time = 2**attempt
logger.warning(
"Telegram API network error (attempt {}/{}): {}. "
"Retrying in {}s...",
attempt + 1,
max_retries,
e,
wait_time,
)
await asyncio.sleep(wait_time)
else:
logger.error(
"Telegram API failed after {} attempts: {}",
max_retries,
e,
)
raise
except TelegramRetryAfter as e:
retry_after = getattr(e, "retry_after", 0)
wait_secs = (
retry_after.total_seconds()
if isinstance(retry_after, timedelta)
else float(retry_after)
)
logger.warning("Rate limited by Telegram, waiting {}s...", wait_secs)
await asyncio.sleep(wait_secs)
return await func(*args, **kwargs)
except TelegramBaseError as e:
err_lower = str(e).lower()
if "message is not modified" in err_lower:
return None
if any(
x in err_lower
for x in [
"message to edit not found",
"message to delete not found",
"message can't be deleted",
"message can't be edited",
"not enough rights to delete",
]
):
return None
if "Can't parse entities" in str(e) and kwargs.get("parse_mode"):
logger.warning("Markdown failed, retrying without parse_mode")
kwargs["parse_mode"] = None
return await func(*args, **kwargs)
raise
return None
async def send_message(
self,
chat_id: str,
text: str,
reply_to: str | None = None,
parse_mode: str | None = "MarkdownV2",
message_thread_id: str | None = None,
) -> str:
"""Send a Telegram message immediately."""
app = self._get_application()
if not app or not app.bot:
raise RuntimeError("Telegram application or bot not initialized")
async def _do_send(parse_mode: str | None = parse_mode) -> str:
kwargs: dict[str, Any] = {
"chat_id": chat_id,
"text": text,
"reply_to_message_id": int(reply_to) if reply_to else None,
"parse_mode": parse_mode,
}
if message_thread_id is not None:
kwargs["message_thread_id"] = int(message_thread_id)
msg = await app.bot.send_message(**kwargs)
return str(msg.message_id)
return await self._with_retry(_do_send, parse_mode=parse_mode)
async def edit_message(
self,
chat_id: str,
message_id: str,
text: str,
parse_mode: str | None = "MarkdownV2",
) -> None:
"""Edit a Telegram message immediately."""
app = self._get_application()
if not app or not app.bot:
raise RuntimeError("Telegram application or bot not initialized")
async def _do_edit(parse_mode: str | None = parse_mode) -> None:
await app.bot.edit_message_text(
chat_id=chat_id,
message_id=int(message_id),
text=text,
parse_mode=parse_mode,
)
await self._with_retry(_do_edit, parse_mode=parse_mode)
async def delete_message(self, chat_id: str, message_id: str) -> None:
"""Delete a Telegram message immediately."""
app = self._get_application()
if not app or not app.bot:
raise RuntimeError("Telegram application or bot not initialized")
async def _do_delete() -> None:
await app.bot.delete_message(chat_id=chat_id, message_id=int(message_id))
await self._with_retry(_do_delete)
async def delete_messages(self, chat_id: str, message_ids: list[str]) -> None:
"""Delete multiple Telegram messages best-effort."""
if not message_ids:
return
app = self._get_application()
if not app or not app.bot:
raise RuntimeError("Telegram application or bot not initialized")
bot = app.bot
if hasattr(bot, "delete_messages"):
async def _do_bulk() -> None:
mids: list[int] = []
for mid in message_ids:
try:
mids.append(int(mid))
except Exception:
continue
if mids:
await bot.delete_messages(chat_id=chat_id, message_ids=mids)
await self._with_retry(_do_bulk)
return
for mid in message_ids:
await self.delete_message(chat_id, mid)
async def queue_send_message(
self,
chat_id: str,
text: str,
reply_to: str | None = None,
parse_mode: str | None = "MarkdownV2",
fire_and_forget: bool = True,
message_thread_id: str | None = None,
) -> str | None:
"""Queue a Telegram send."""
return await self._outbox.queue_send_message(
chat_id,
text,
reply_to,
parse_mode,
fire_and_forget,
message_thread_id,
)
async def queue_edit_message(
self,
chat_id: str,
message_id: str,
text: str,
parse_mode: str | None = "MarkdownV2",
fire_and_forget: bool = True,
) -> None:
"""Queue a Telegram edit."""
await self._outbox.queue_edit_message(
chat_id,
message_id,
text,
parse_mode,
fire_and_forget,
)
async def queue_delete_message(
self,
chat_id: str,
message_id: str,
fire_and_forget: bool = True,
) -> None:
"""Queue a Telegram delete."""
await self._outbox.queue_delete_message(chat_id, message_id, fire_and_forget)
async def queue_delete_messages(
self,
chat_id: str,
message_ids: list[str],
fire_and_forget: bool = True,
) -> None:
"""Queue a Telegram bulk delete."""
await self._outbox.queue_delete_messages(
chat_id,
message_ids,
fire_and_forget,
)
def fire_and_forget(self, task: Awaitable[Any]) -> None:
"""Execute a coroutine without awaiting it."""
self._outbox.fire_and_forget(task)

View file

@ -16,7 +16,7 @@ from .command_dispatcher import (
parse_command_base,
)
from .models import IncomingMessage
from .platforms.base import MessagingPlatform
from .platforms.ports import OutboundMessenger
from .safe_diagnostics import format_exception_for_log
from .session import SessionStore
from .trees import MessageNode, MessageState, MessageTree, TreeQueueManager
@ -28,7 +28,8 @@ class MessagingTurnIntake:
def __init__(
self,
*,
platform: MessagingPlatform,
platform_name: str,
outbound: OutboundMessenger,
session_store: SessionStore,
command_context: MessagingCommandContext,
get_tree_queue: Callable[[], TreeQueueManager],
@ -38,7 +39,8 @@ class MessagingTurnIntake:
record_outgoing_message: Callable[[str, str, str | None, str], None],
log_messaging_error_details: bool = False,
) -> None:
self.platform = platform
self.platform_name = platform_name
self.outbound = outbound
self.session_store = session_store
self._command_context = command_context
self._get_tree_queue = get_tree_queue
@ -99,7 +101,7 @@ class MessagingTurnIntake:
status_text = self._get_initial_status(tree, parent_node_id)
if incoming.status_message_id:
status_msg_id = incoming.status_message_id
await self.platform.queue_edit_message(
await self.outbound.queue_edit_message(
incoming.chat_id,
status_msg_id,
status_text,
@ -107,7 +109,7 @@ class MessagingTurnIntake:
fire_and_forget=False,
)
else:
status_msg_id = await self.platform.queue_send_message(
status_msg_id = await self.outbound.queue_send_message(
incoming.chat_id,
status_text,
reply_to=incoming.message_id,
@ -152,13 +154,13 @@ class MessagingTurnIntake:
trace_event(
stage="routing",
event="turn.queued",
source=getattr(self.platform, "name", "messaging"),
source=self.platform_name,
chat_id=incoming.chat_id,
platform_message_id=node_id,
status_message_id=status_msg_id,
queue_size=queue_size,
)
await self.platform.queue_edit_message(
await self.outbound.queue_edit_message(
incoming.chat_id,
status_msg_id,
self._format_status(
@ -189,8 +191,8 @@ class MessagingTurnIntake:
if not node or node.state != MessageState.PENDING:
continue
position += 1
self.platform.fire_and_forget(
self.platform.queue_edit_message(
self.outbound.fire_and_forget(
self.outbound.queue_edit_message(
node.incoming.chat_id,
node.status_message_id,
self._format_status(
@ -205,8 +207,8 @@ class MessagingTurnIntake:
node = tree.get_node(node_id)
if not node or node.state == MessageState.ERROR:
return
self.platform.fire_and_forget(
self.platform.queue_edit_message(
self.outbound.fire_and_forget(
self.outbound.queue_edit_message(
node.incoming.chat_id,
node.status_message_id,
self._format_status("🔄", "Processing...", None),

View file

@ -7,7 +7,7 @@ from collections.abc import Callable
from loguru import logger
from .platforms.base import MessagingPlatform
from .platforms.ports import OutboundMessenger
from .safe_diagnostics import format_exception_for_log
from .transcript import RenderCtx, TranscriptBuffer
@ -18,7 +18,7 @@ class ThrottledTranscriptEditor:
def __init__(
self,
*,
platform: MessagingPlatform,
outbound: OutboundMessenger,
parse_mode: str | None,
get_limit_chars: Callable[[], int],
transcript: TranscriptBuffer,
@ -29,7 +29,7 @@ class ThrottledTranscriptEditor:
debug_platform_edits: bool,
log_messaging_error_details: bool = False,
) -> None:
self._platform = platform
self._outbound = outbound
self._parse_mode = parse_mode
self._get_limit_chars = get_limit_chars
self._transcript = transcript
@ -85,7 +85,7 @@ class ThrottledTranscriptEditor:
logger.debug("PLATFORM_EDIT_TEXT:\n{}", display)
self._last_displayed_text = display
try:
await self._platform.queue_edit_message(
await self._outbound.queue_edit_message(
self._chat_id,
self._status_msg_id,
display,

View file

@ -6,9 +6,10 @@ from loguru import logger
from core.trace import trace_event
from .managed_protocols import ManagedClaudeSessionManagerProtocol
from .models import IncomingMessage
from .node_runner import MessagingNodeRunner
from .platforms.base import ManagedClaudeSessionManagerProtocol, MessagingPlatform
from .platforms.ports import OutboundMessenger, VoiceCancellation
from .rendering.profiles import build_rendering_profile
from .safe_diagnostics import format_exception_for_log
from .session import SessionStore
@ -27,25 +28,30 @@ class MessagingWorkflow:
def __init__(
self,
platform: MessagingPlatform,
outbound: OutboundMessenger,
cli_manager: ManagedClaudeSessionManagerProtocol,
session_store: SessionStore,
*,
platform_name: str | None = None,
voice_cancellation: VoiceCancellation | None = None,
debug_platform_edits: bool = False,
debug_subagent_stack: bool = False,
log_raw_messaging_content: bool = False,
log_raw_cli_diagnostics: bool = False,
log_messaging_error_details: bool = False,
):
self.platform = platform
self.platform_name = platform_name or "messaging"
self.outbound = outbound
self.voice_cancellation = voice_cancellation
self.cli_manager = cli_manager
self.session_store = session_store
self._log_messaging_error_details = log_messaging_error_details
self._tree_queue = TreeQueueManager()
self._rendering_profile = build_rendering_profile(platform.name)
self._rendering_profile = build_rendering_profile(self.platform_name)
self.node_runner = MessagingNodeRunner(
platform=platform,
platform_name=self.platform_name,
outbound=outbound,
cli_manager=cli_manager,
session_store=session_store,
get_tree_queue=lambda: self._tree_queue,
@ -59,7 +65,8 @@ class MessagingWorkflow:
log_messaging_error_details=log_messaging_error_details,
)
self.turn_intake = MessagingTurnIntake(
platform=platform,
platform_name=self.platform_name,
outbound=outbound,
session_store=session_store,
command_context=self,
get_tree_queue=lambda: self._tree_queue,
@ -105,11 +112,10 @@ class MessagingWorkflow:
"""
Main entry point for handling an incoming platform message.
"""
platform_name = getattr(self.platform, "name", "messaging")
trace_event(
stage="ingress",
event="turn.received",
source=platform_name,
source=self.platform_name,
chat_id=incoming.chat_id,
platform_message_id=incoming.message_id,
reply_to_message_id=incoming.reply_to_message_id,
@ -182,8 +188,8 @@ class MessagingWorkflow:
"""Update status messages and persist tree state for cancelled nodes."""
trees_to_save: dict[str, MessageTree] = {}
for node in nodes:
self.platform.fire_and_forget(
self.platform.queue_edit_message(
self.outbound.fire_and_forget(
self.outbound.queue_edit_message(
node.incoming.chat_id,
node.status_message_id,
self.format_status("", "Stopped."),

View file

@ -4,7 +4,7 @@ build-backend = "hatchling.build"
[project]
name = "free-claude-code"
version = "2.3.13"
version = "2.3.14"
description = "Middleware between Claude Code CLI (Anthropic API) and NVIDIA NIM"
readme = "README.md"
requires-python = ">=3.14.0"

View file

@ -456,7 +456,7 @@ CAPABILITY_CONTRACTS: tuple[CapabilityContract, ...] = (
"extensible_provider_platform_abcs",
"providers.registry and messaging.platforms.factory",
"new provider/platform implementations",
"registered BaseProvider or MessagingPlatform",
"registered BaseProvider or messaging component bundle",
"unknown platform returns None; unknown provider errors",
(
"tests/contracts/test_feature_manifest.py",

View file

@ -9,7 +9,7 @@ import subprocess
import time
import uuid
import wave
from collections.abc import AsyncGenerator, Awaitable, Callable, Iterator, Sequence
from collections.abc import AsyncGenerator, Awaitable, Callable, Iterator
from contextlib import contextmanager
from dataclasses import dataclass, field
from pathlib import Path
@ -28,7 +28,6 @@ from core.anthropic.stream_contracts import (
text_content,
)
from messaging.models import IncomingMessage
from messaging.platforms.base import MessagingPlatform
from messaging.session import SessionStore
from messaging.workflow import MessagingWorkflow
from smoke.lib.config import ProviderModel, SmokeConfig, auth_headers
@ -306,7 +305,7 @@ class ClientProtocolDriver:
)
class FakePlatform(MessagingPlatform):
class FakePlatform:
"""In-memory platform that exercises the real message handler."""
def __init__(self, name: str) -> None:
@ -423,7 +422,7 @@ class FakePlatform(MessagingPlatform):
async def queue_delete_messages(
self,
chat_id: str,
message_ids: Sequence[str],
message_ids: list[str],
fire_and_forget: bool = True,
) -> None:
for message_id in message_ids:
@ -438,9 +437,9 @@ class FakePlatform(MessagingPlatform):
)
async def cancel_pending_voice(
self, chat_id: str, voice_message_id: str
self, chat_id: str, reply_id: str
) -> tuple[str, str] | None:
return self._pending_voice.pop((chat_id, voice_message_id), None)
return self._pending_voice.pop((chat_id, reply_id), None)
class FakeCLISession:
@ -515,7 +514,11 @@ class FakePlatformDriver:
storage_path=str(self.tmp_path / f"{self.platform_name}-sessions.json")
)
self.workflow = MessagingWorkflow(
self.platform, self.cli_manager, self.session_store
self.platform,
self.cli_manager,
self.session_store,
platform_name=self.platform_name,
voice_cancellation=self.platform,
)
self.platform.on_message(self.workflow.handle_message)

View file

@ -6,7 +6,7 @@ import subprocess
import pytest
from config.settings import Settings
from messaging.platforms.factory import create_messaging_platform
from messaging.platforms.factory import create_messaging_components
from providers.registry import PROVIDER_DESCRIPTORS, build_provider_config
from smoke.lib.child_process import cmd_free_claude_code_serve, cmd_python_c
from smoke.lib.config import SmokeConfig
@ -161,9 +161,9 @@ def _settings_init_key(field_name: str) -> str:
@pytest.mark.smoke_target("extensibility")
def test_platform_factory_e2e() -> None:
assert create_messaging_platform("not-a-platform") is None
assert create_messaging_platform("telegram") is None
assert create_messaging_platform("discord") is None
assert create_messaging_components("not-a-platform") is None
assert create_messaging_components("telegram") is None
assert create_messaging_components("discord") is None
@pytest.mark.smoke_target("cli")

View file

@ -39,6 +39,26 @@ def _app_settings(**kwargs):
return SimpleNamespace(**data)
def _fake_messaging_components(runtime: MagicMock | None = None) -> SimpleNamespace:
runtime = runtime or MagicMock()
runtime.name = getattr(runtime, "name", "fake")
runtime.on_message = getattr(runtime, "on_message", MagicMock())
runtime.start = getattr(runtime, "start", AsyncMock())
runtime.stop = getattr(runtime, "stop", AsyncMock())
outbound = MagicMock()
outbound.queue_send_message = AsyncMock(return_value="msg")
outbound.queue_edit_message = AsyncMock()
outbound.queue_delete_message = AsyncMock()
outbound.queue_delete_messages = AsyncMock()
outbound.fire_and_forget = MagicMock()
return SimpleNamespace(
name=runtime.name,
runtime=runtime,
outbound=outbound,
voice_cancellation=None,
)
def test_warn_if_process_auth_token_logs_warning():
api_runtime_mod = importlib.import_module("api.runtime")
settings = cast(
@ -94,7 +114,7 @@ async def test_runtime_startup_logs_admin_url_without_printed_server_banner(tmp_
patch.object(ProviderRegistry, "start_model_list_refresh"),
patch.object(ProviderRegistry, "cleanup", new=AsyncMock()),
patch(
"messaging.platforms.factory.create_messaging_platform",
"messaging.platforms.factory.create_messaging_components",
return_value=None,
),
):
@ -286,6 +306,7 @@ def test_app_lifespan_sets_state_and_cleans_up(tmp_path, messaging_enabled):
fake_platform.on_message = MagicMock()
fake_platform.start = AsyncMock()
fake_platform.stop = AsyncMock()
fake_components = _fake_messaging_components(fake_platform)
session_store = MagicMock()
session_store.get_all_trees.return_value = [{"t": 1}] if messaging_enabled else []
@ -309,9 +330,9 @@ def test_app_lifespan_sets_state_and_cleans_up(tmp_path, messaging_enabled):
patch.object(api_app_mod, "get_settings", return_value=settings),
patch.object(ProviderRegistry, "cleanup", new=registry_cleanup),
patch(
"messaging.platforms.factory.create_messaging_platform",
return_value=fake_platform if messaging_enabled else None,
) as create_platform,
"messaging.platforms.factory.create_messaging_components",
return_value=fake_components if messaging_enabled else None,
) as create_components,
patch("messaging.session.SessionStore", return_value=session_store),
patch("cli.managed.ManagedClaudeSessionManager", return_value=cli_manager),
patch(
@ -323,7 +344,7 @@ def test_app_lifespan_sets_state_and_cleans_up(tmp_path, messaging_enabled):
pass
if messaging_enabled:
create_platform.assert_called_once()
create_components.assert_called_once()
fake_platform.on_message.assert_called_once()
fake_platform.start.assert_awaited_once()
fake_platform.stop.assert_awaited_once()
@ -337,7 +358,7 @@ def test_app_lifespan_sets_state_and_cleans_up(tmp_path, messaging_enabled):
fake_platform.start.assert_not_awaited()
fake_platform.stop.assert_not_awaited()
cli_manager.stop_all.assert_not_awaited()
assert getattr(app.state, "messaging_platform", "missing") is None
assert getattr(app.state, "messaging_runtime", "missing") is None
registry_cleanup.assert_awaited_once()
@ -364,6 +385,7 @@ def test_app_lifespan_cleanup_continues_if_platform_stop_raises(tmp_path):
fake_platform.on_message = MagicMock()
fake_platform.start = AsyncMock()
fake_platform.stop = AsyncMock(side_effect=RuntimeError("stop failed"))
fake_components = _fake_messaging_components(fake_platform)
session_store = MagicMock()
session_store.get_all_trees.return_value = []
@ -379,8 +401,8 @@ def test_app_lifespan_cleanup_continues_if_platform_stop_raises(tmp_path):
patch.object(api_app_mod, "get_settings", return_value=settings),
patch.object(ProviderRegistry, "cleanup", new=registry_cleanup),
patch(
"messaging.platforms.factory.create_messaging_platform",
return_value=fake_platform,
"messaging.platforms.factory.create_messaging_components",
return_value=fake_components,
),
patch("messaging.session.SessionStore", return_value=session_store),
patch("cli.managed.ManagedClaudeSessionManager", return_value=cli_manager),
@ -421,16 +443,16 @@ async def test_runtime_startup_validation_failure_does_not_block_server(tmp_path
patch.object(ProviderRegistry, "cleanup", new=cleanup),
patch.object(api_runtime_mod.logger, "warning") as log_warning,
patch(
"messaging.platforms.factory.create_messaging_platform",
"messaging.platforms.factory.create_messaging_components",
return_value=None,
) as create_platform,
) as create_components,
):
await runtime.startup()
await runtime.shutdown()
validation.assert_awaited_once_with(settings)
cleanup.assert_awaited_once()
create_platform.assert_called_once()
create_components.assert_called_once()
logged = " ".join(
str(arg) for call in log_warning.call_args_list for arg in call.args
)
@ -507,14 +529,14 @@ def test_app_lifespan_messaging_import_error_no_crash(tmp_path, caplog):
patch.object(api_app_mod, "get_settings", return_value=settings),
patch.object(ProviderRegistry, "cleanup", new=registry_cleanup),
patch(
"messaging.platforms.factory.create_messaging_platform",
"messaging.platforms.factory.create_messaging_components",
side_effect=ImportError("discord not installed"),
),
TestClient(app),
):
pass
assert getattr(app.state, "messaging_platform", None) is None
assert getattr(app.state, "messaging_runtime", None) is None
registry_cleanup.assert_awaited_once()
@ -541,6 +563,7 @@ def test_app_lifespan_platform_start_exception_cleanup_still_runs(tmp_path):
fake_platform.on_message = MagicMock()
fake_platform.start = AsyncMock(side_effect=RuntimeError("start failed"))
fake_platform.stop = AsyncMock()
fake_components = _fake_messaging_components(fake_platform)
session_store = MagicMock()
session_store.get_all_trees.return_value = []
@ -556,8 +579,8 @@ def test_app_lifespan_platform_start_exception_cleanup_still_runs(tmp_path):
patch.object(api_app_mod, "get_settings", return_value=settings),
patch.object(ProviderRegistry, "cleanup", new=registry_cleanup),
patch(
"messaging.platforms.factory.create_messaging_platform",
return_value=fake_platform,
"messaging.platforms.factory.create_messaging_components",
return_value=fake_components,
),
patch("messaging.session.SessionStore", return_value=session_store),
patch("cli.managed.ManagedClaudeSessionManager", return_value=cli_manager),
@ -591,6 +614,7 @@ def test_app_lifespan_flush_pending_save_exception_warning_only(tmp_path):
fake_platform.on_message = MagicMock()
fake_platform.start = AsyncMock()
fake_platform.stop = AsyncMock()
fake_components = _fake_messaging_components(fake_platform)
session_store = MagicMock()
session_store.get_all_trees.return_value = []
@ -607,8 +631,8 @@ def test_app_lifespan_flush_pending_save_exception_warning_only(tmp_path):
patch.object(api_app_mod, "get_settings", return_value=settings),
patch.object(ProviderRegistry, "cleanup", new=registry_cleanup),
patch(
"messaging.platforms.factory.create_messaging_platform",
return_value=fake_platform,
"messaging.platforms.factory.create_messaging_components",
return_value=fake_components,
),
patch("messaging.session.SessionStore", return_value=session_store),
patch("cli.managed.ManagedClaudeSessionManager", return_value=cli_manager),

View file

@ -28,7 +28,7 @@ async def test_messaging_start_failure_default_logs_exclude_traceback(caplog):
with (
patch(
"messaging.platforms.factory.create_messaging_platform",
"messaging.platforms.factory.create_messaging_components",
side_effect=RuntimeError("SECRET_RUNTIME_DETAIL"),
),
caplog.at_level(logging.ERROR),

View file

@ -85,7 +85,7 @@ def llamacpp_provider(provider_config):
@pytest.fixture
def mock_cli_session():
from messaging.platforms.base import ManagedClaudeSessionProtocol
from messaging.managed_protocols import ManagedClaudeSessionProtocol
session = MagicMock(spec=ManagedClaudeSessionProtocol)
session.start_task = MagicMock() # This will return an async generator
@ -95,7 +95,7 @@ def mock_cli_session():
@pytest.fixture
def mock_cli_manager():
from messaging.platforms.base import ManagedClaudeSessionManagerProtocol
from messaging.managed_protocols import ManagedClaudeSessionManagerProtocol
manager = MagicMock(spec=ManagedClaudeSessionManagerProtocol)
manager.get_or_create_session = AsyncMock()
@ -108,9 +108,9 @@ def mock_cli_manager():
@pytest.fixture
def mock_platform():
from messaging.platforms.base import MessagingPlatform
from messaging.platforms.ports import OutboundMessenger
platform = MagicMock(spec=MessagingPlatform)
platform = MagicMock(spec=OutboundMessenger)
platform.send_message = AsyncMock(return_value="msg_123")
platform.edit_message = AsyncMock()
platform.delete_message = AsyncMock()
@ -126,6 +126,7 @@ def mock_platform():
await qdm(chat_id, mid, fire_and_forget=fire_and_forget)
platform.queue_delete_messages = AsyncMock(side_effect=_queue_delete_messages)
platform.cancel_pending_voice = AsyncMock(return_value=None)
def _fire_and_forget(task):
if asyncio.iscoroutine(task):

View file

@ -3,7 +3,7 @@ from __future__ import annotations
import re
from pathlib import Path
from messaging.platforms.factory import create_messaging_platform
from messaging.platforms.factory import create_messaging_components
from providers.base import BaseProvider
from providers.cerebras import CerebrasProvider
from providers.codestral import CodestralProvider
@ -98,7 +98,7 @@ def test_provider_and_platform_registries_include_advertised_builtins() -> None:
for provider_class in provider_classes.values():
assert issubclass(provider_class, BaseProvider)
assert create_messaging_platform("not-a-platform") is None
assert create_messaging_components("not-a-platform") is None
def _collect_test_names(root: Path) -> set[str]:

View file

@ -244,19 +244,28 @@ def test_messaging_platforms_use_shared_outbox_and_voice_flow() -> None:
repo_root = Path(__file__).resolve().parents[2]
platforms_root = repo_root / "messaging" / "platforms"
assert not (platforms_root / "base.py").exists()
assert (platforms_root / "ports.py").exists()
assert (platforms_root / "outbox.py").exists()
assert (platforms_root / "voice_flow.py").exists()
for adapter in {
for runtime in {
platforms_root / "telegram.py",
platforms_root / "discord.py",
}:
text = adapter.read_text(encoding="utf-8")
assert "PlatformOutbox" in text
text = runtime.read_text(encoding="utf-8")
assert "PlatformOutbox" not in text
assert "VoiceNoteFlow" in text
assert "from ..voice" not in text
assert "NamedTemporaryFile" not in text
for messenger in {
platforms_root / "telegram_io.py",
platforms_root / "discord_io.py",
}:
text = messenger.read_text(encoding="utf-8")
assert "PlatformOutbox" in text
def test_cli_surfaces_are_explicit_launchers_and_managed_claude() -> None:
repo_root = Path(__file__).resolve().parents[2]
@ -297,13 +306,16 @@ def test_cli_surfaces_are_explicit_launchers_and_managed_claude() -> None:
assert '"190000"' not in text
assert '"fcc-no-auth"' not in text
messaging_base_text = (repo_root / "messaging" / "platforms" / "base.py").read_text(
encoding="utf-8"
messaging_protocols_text = (
repo_root / "messaging" / "managed_protocols.py"
).read_text(encoding="utf-8")
assert "class ManagedClaudeSessionProtocol(Protocol)" in messaging_protocols_text
assert "class ManagedClaudeSession(Protocol)" not in messaging_protocols_text
assert (
"class ManagedClaudeSessionManagerProtocol(Protocol)"
in messaging_protocols_text
)
assert "class ManagedClaudeSessionProtocol(Protocol)" in messaging_base_text
assert "class ManagedClaudeSession(Protocol)" not in messaging_base_text
assert "class ManagedClaudeSessionManagerProtocol(Protocol)" in messaging_base_text
assert "class SessionManagerInterface(Protocol)" not in messaging_base_text
assert "class SessionManagerInterface(Protocol)" not in messaging_protocols_text
for path in {
repo_root / "messaging" / "__init__.py",
repo_root / "messaging" / "platforms" / "__init__.py",

View file

@ -7,10 +7,14 @@ import pytest
from messaging.platforms.discord import (
DISCORD_AVAILABLE,
DiscordPlatform,
DiscordRuntime,
_get_discord,
_parse_allowed_channels,
)
from messaging.platforms.discord_inbound import (
discord_text_message_from_event,
parse_allowed_channels,
)
from messaging.platforms.discord_io import truncate_discord_message
class TestGetDiscord:
@ -31,31 +35,56 @@ class TestParseAllowedChannels:
"""Tests for _parse_allowed_channels helper."""
def test_empty_string_returns_empty_set(self):
assert _parse_allowed_channels("") == set()
assert _parse_allowed_channels(None) == set()
assert parse_allowed_channels("") == set()
assert parse_allowed_channels(None) == set()
def test_whitespace_only_returns_empty_set(self):
assert _parse_allowed_channels(" ") == set()
assert parse_allowed_channels(" ") == set()
def test_single_channel(self):
assert _parse_allowed_channels("123456789") == {"123456789"}
assert parse_allowed_channels("123456789") == {"123456789"}
def test_comma_separated(self):
assert _parse_allowed_channels("111,222,333") == {"111", "222", "333"}
assert parse_allowed_channels("111,222,333") == {"111", "222", "333"}
def test_strips_whitespace(self):
assert _parse_allowed_channels(" 111 , 222 ") == {"111", "222"}
assert parse_allowed_channels(" 111 , 222 ") == {"111", "222"}
def test_empty_parts_ignored(self):
assert _parse_allowed_channels("111,,222,") == {"111", "222"}
assert parse_allowed_channels("111,,222,") == {"111", "222"}
class TestDiscordInbound:
def test_text_message_normalizes_accepted_event(self):
msg = MagicMock()
msg.author.id = 456
msg.author.display_name = "User"
msg.content = "hello"
msg.channel.id = 123
msg.id = 789
msg.reference.message_id = 555
incoming = discord_text_message_from_event(
msg,
log_raw_messaging_content=False,
)
assert incoming.text == "hello"
assert incoming.chat_id == "123"
assert incoming.user_id == "456"
assert incoming.message_id == "789"
assert incoming.platform == "discord"
assert incoming.reply_to_message_id == "555"
assert incoming.username == "User"
assert incoming.raw_event is msg
@pytest.mark.skipif(not DISCORD_AVAILABLE, reason="discord.py not installed")
class TestDiscordPlatform:
"""Tests for DiscordPlatform (requires discord.py)."""
class TestDiscordRuntime:
"""Tests for Discord runtime and messenger behavior."""
def test_init_with_token(self):
platform = DiscordPlatform(
platform = DiscordRuntime(
bot_token="test_token",
allowed_channel_ids="123,456",
)
@ -64,47 +93,42 @@ class TestDiscordPlatform:
def test_init_without_allowed_channels(self):
with patch.dict("os.environ", {"ALLOWED_DISCORD_CHANNELS": ""}, clear=False):
platform = DiscordPlatform(bot_token="token", allowed_channel_ids="")
platform = DiscordRuntime(bot_token="token", allowed_channel_ids="")
assert platform.allowed_channel_ids == set()
def test_empty_allowed_channels_rejects_all_messages(self):
"""When allowed_channel_ids is empty, no channels are allowed (secure default)."""
with patch.dict("os.environ", {"ALLOWED_DISCORD_CHANNELS": ""}, clear=False):
platform = DiscordPlatform(bot_token="token", allowed_channel_ids="")
platform = DiscordRuntime(bot_token="token", allowed_channel_ids="")
assert platform.allowed_channel_ids == set()
# Empty set means: not self.allowed_channel_ids is True -> reject
def test_truncate_long_message(self):
platform = DiscordPlatform(bot_token="token")
long_text = "x" * 2500
truncated = platform._truncate(long_text)
truncated = truncate_discord_message(long_text)
assert len(truncated) == 2000
assert truncated.endswith("...")
def test_truncate_short_message_unchanged(self):
platform = DiscordPlatform(bot_token="token")
short = "hello"
assert platform._truncate(short) == short
assert truncate_discord_message(short) == short
def test_truncate_exactly_at_limit_unchanged(self):
platform = DiscordPlatform(bot_token="token")
exact = "x" * 2000
assert platform._truncate(exact) == exact
assert truncate_discord_message(exact) == exact
def test_truncate_one_over_limit_truncates(self):
platform = DiscordPlatform(bot_token="token")
over = "x" * 2001
result = platform._truncate(over)
result = truncate_discord_message(over)
assert len(result) == 2000
assert result.endswith("...")
def test_truncate_empty_string(self):
platform = DiscordPlatform(bot_token="token")
assert platform._truncate("") == ""
assert truncate_discord_message("") == ""
@pytest.mark.asyncio
async def test_send_message_returns_message_id(self):
platform = DiscordPlatform(bot_token="token")
platform = DiscordRuntime(bot_token="token")
mock_msg = MagicMock()
mock_msg.id = 999
mock_channel = AsyncMock()
@ -113,12 +137,12 @@ class TestDiscordPlatform:
with patch.object(
platform._client, "get_channel", MagicMock(return_value=mock_channel)
):
msg_id = await platform.send_message("123", "Hello")
msg_id = await platform.outbound.send_message("123", "Hello")
assert msg_id == "999"
@pytest.mark.asyncio
async def test_edit_message(self):
platform = DiscordPlatform(bot_token="token")
platform = DiscordRuntime(bot_token="token")
mock_msg = AsyncMock()
mock_channel = AsyncMock()
mock_channel.fetch_message = AsyncMock(return_value=mock_msg)
@ -126,22 +150,22 @@ class TestDiscordPlatform:
with patch.object(
platform._client, "get_channel", MagicMock(return_value=mock_channel)
):
await platform.edit_message("123", "456", "Updated text")
await platform.outbound.edit_message("123", "456", "Updated text")
mock_msg.edit.assert_called_once_with(content="Updated text")
@pytest.mark.asyncio
async def test_send_message_channel_not_found_raises(self):
platform = DiscordPlatform(bot_token="token")
platform = DiscordRuntime(bot_token="token")
platform._connected = True
with (
patch.object(platform._client, "get_channel", MagicMock(return_value=None)),
pytest.raises(RuntimeError, match="Channel"),
):
await platform.send_message("123", "Hello")
await platform.outbound.send_message("123", "Hello")
@pytest.mark.asyncio
async def test_send_message_channel_no_send_raises(self):
platform = DiscordPlatform(bot_token="token")
platform = DiscordRuntime(bot_token="token")
platform._connected = True
mock_channel = MagicMock(spec=[]) # No send attr
with (
@ -150,11 +174,11 @@ class TestDiscordPlatform:
),
pytest.raises(RuntimeError, match="Channel"),
):
await platform.send_message("123", "Hello")
await platform.outbound.send_message("123", "Hello")
@pytest.mark.asyncio
async def test_queue_send_message_without_limiter_calls_send_message(self):
platform = DiscordPlatform(bot_token="token")
platform = DiscordRuntime(bot_token="token")
platform._limiter = None
platform._connected = True
mock_channel = AsyncMock()
@ -164,13 +188,13 @@ class TestDiscordPlatform:
with patch.object(
platform._client, "get_channel", MagicMock(return_value=mock_channel)
):
result = await platform.queue_send_message("123", "hi")
result = await platform.outbound.queue_send_message("123", "hi")
assert result == "42"
mock_channel.send.assert_awaited_once()
@pytest.mark.asyncio
async def test_queue_edit_message_without_limiter_calls_edit_message(self):
platform = DiscordPlatform(bot_token="token")
platform = DiscordRuntime(bot_token="token")
platform._limiter = None
platform._connected = True
mock_msg = AsyncMock()
@ -179,12 +203,12 @@ class TestDiscordPlatform:
with patch.object(
platform._client, "get_channel", MagicMock(return_value=mock_channel)
):
await platform.queue_edit_message("123", "456", "Updated")
await platform.outbound.queue_edit_message("123", "456", "Updated")
mock_msg.edit.assert_called_once_with(content="Updated")
@pytest.mark.asyncio
async def test_on_discord_message_bot_ignored(self):
platform = DiscordPlatform(bot_token="token", allowed_channel_ids="123")
platform = DiscordRuntime(bot_token="token", allowed_channel_ids="123")
handler = AsyncMock()
platform.on_message(handler)
msg = MagicMock()
@ -196,7 +220,7 @@ class TestDiscordPlatform:
@pytest.mark.asyncio
async def test_on_discord_message_empty_content_ignored(self):
platform = DiscordPlatform(bot_token="token", allowed_channel_ids="123")
platform = DiscordRuntime(bot_token="token", allowed_channel_ids="123")
handler = AsyncMock()
platform.on_message(handler)
msg = MagicMock()
@ -208,7 +232,7 @@ class TestDiscordPlatform:
@pytest.mark.asyncio
async def test_on_discord_message_channel_not_allowed_ignored(self):
platform = DiscordPlatform(bot_token="token", allowed_channel_ids="123")
platform = DiscordRuntime(bot_token="token", allowed_channel_ids="123")
handler = AsyncMock()
platform.on_message(handler)
msg = MagicMock()
@ -220,7 +244,7 @@ class TestDiscordPlatform:
@pytest.mark.asyncio
async def test_on_discord_message_valid_calls_handler(self):
platform = DiscordPlatform(bot_token="token", allowed_channel_ids="123")
platform = DiscordRuntime(bot_token="token", allowed_channel_ids="123")
handler = AsyncMock()
platform.on_message(handler)
msg = MagicMock()
@ -242,7 +266,7 @@ class TestDiscordPlatform:
@pytest.mark.asyncio
async def test_send_message_with_reply_to(self):
platform = DiscordPlatform(bot_token="token")
platform = DiscordRuntime(bot_token="token")
mock_msg = MagicMock()
mock_msg.id = 999
mock_channel = AsyncMock()
@ -256,7 +280,10 @@ class TestDiscordPlatform:
):
mock_discord = MagicMock()
mock_get.return_value = mock_discord
msg_id = await platform.send_message("123", "Hello", reply_to="456")
platform.outbound._get_discord = mock_get
msg_id = await platform.outbound.send_message(
"123", "Hello", reply_to="456"
)
assert msg_id == "999"
mock_channel.send.assert_awaited_once()
call_kw = mock_channel.send.call_args[1]
@ -266,7 +293,7 @@ class TestDiscordPlatform:
async def test_edit_message_not_found_returns_gracefully(self):
import discord as discord_pkg
platform = DiscordPlatform(bot_token="token")
platform = DiscordRuntime(bot_token="token")
mock_channel = AsyncMock()
mock_resp = MagicMock()
mock_resp.status = 404
@ -277,12 +304,12 @@ class TestDiscordPlatform:
with patch.object(
platform._client, "get_channel", MagicMock(return_value=mock_channel)
):
await platform.edit_message("123", "456", "Updated")
await platform.outbound.edit_message("123", "456", "Updated")
# Should not raise - NotFound is caught and we return
@pytest.mark.asyncio
async def test_delete_message(self):
platform = DiscordPlatform(bot_token="token")
platform = DiscordRuntime(bot_token="token")
mock_msg = AsyncMock()
mock_channel = AsyncMock()
mock_channel.fetch_message = AsyncMock(return_value=mock_msg)
@ -294,12 +321,13 @@ class TestDiscordPlatform:
patch("messaging.platforms.discord._get_discord") as mock_get,
):
mock_get.return_value = MagicMock()
await platform.delete_message("123", "456")
platform.outbound._get_discord = mock_get
await platform.outbound.delete_message("123", "456")
mock_msg.delete.assert_awaited_once()
@pytest.mark.asyncio
async def test_fire_and_forget_with_coroutine(self):
platform = DiscordPlatform(bot_token="token")
platform = DiscordRuntime(bot_token="token")
async def _task():
pass
@ -311,12 +339,12 @@ class TestDiscordPlatform:
return asyncio.ensure_future(c)
mock_create.side_effect = _run
platform.fire_and_forget(coro)
platform.outbound.fire_and_forget(coro)
mock_create.assert_called_once()
await asyncio.sleep(0)
def test_on_message_registers_handler(self):
platform = DiscordPlatform(bot_token="token")
platform = DiscordRuntime(bot_token="token")
handler = AsyncMock()
platform.on_message(handler)
assert platform._message_handler is handler
@ -324,13 +352,13 @@ class TestDiscordPlatform:
@pytest.mark.asyncio
async def test_start_requires_token(self):
with patch.dict("os.environ", {"DISCORD_BOT_TOKEN": ""}, clear=False):
platform = DiscordPlatform(bot_token="")
platform = DiscordRuntime(bot_token="")
with pytest.raises(ValueError, match="DISCORD_BOT_TOKEN"):
await platform.start()
@pytest.mark.asyncio
async def test_start_connects(self):
platform = DiscordPlatform(bot_token="token")
platform = DiscordRuntime(bot_token="token")
async def _fake_start(_token):
platform._connected = True
@ -352,7 +380,7 @@ class TestDiscordPlatform:
@pytest.mark.asyncio
async def test_stop_when_already_closed(self):
platform = DiscordPlatform(bot_token="token")
platform = DiscordRuntime(bot_token="token")
platform._connected = True
with patch.object(
platform._client, "is_closed", new_callable=MagicMock, return_value=True
@ -362,7 +390,7 @@ class TestDiscordPlatform:
@pytest.mark.asyncio
async def test_stop_closes_client(self):
platform = DiscordPlatform(bot_token="token")
platform = DiscordRuntime(bot_token="token")
platform._connected = True
mock_close = AsyncMock()
with (

View file

@ -10,7 +10,12 @@ from messaging.workflow import MessagingWorkflow
@pytest.fixture
def handler(mock_platform, mock_cli_manager, mock_session_store):
return MessagingWorkflow(mock_platform, mock_cli_manager, mock_session_store)
return MessagingWorkflow(
mock_platform,
mock_cli_manager,
mock_session_store,
voice_cancellation=mock_platform,
)
@pytest.mark.asyncio

View file

@ -1,7 +1,7 @@
"""Tests for messaging/ module."""
import json
from unittest.mock import patch
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
@ -43,15 +43,32 @@ class TestMessagingModels:
assert msg.reply_to_message_id == "100"
class TestMessagingBase:
"""Test MessagingPlatform ABC."""
class TestMessagingPorts:
"""Test explicit messaging platform component ports."""
def test_platform_is_abstract(self):
"""Verify MessagingPlatform cannot be instantiated."""
from messaging.platforms.base import MessagingPlatform
def test_components_bundle_runtime_and_outbound(self):
"""Verify the factory handoff shape is explicit."""
from messaging.platforms.ports import MessagingPlatformComponents
with pytest.raises(TypeError):
MessagingPlatform()
runtime = MagicMock()
runtime.name = "telegram"
runtime.start = AsyncMock()
runtime.stop = AsyncMock()
runtime.on_message = MagicMock()
outbound = MagicMock()
outbound.queue_send_message = AsyncMock()
outbound.queue_edit_message = AsyncMock()
outbound.queue_delete_message = AsyncMock()
outbound.queue_delete_messages = AsyncMock()
outbound.fire_and_forget = MagicMock()
components = MessagingPlatformComponents(
name="telegram",
runtime=runtime,
outbound=outbound,
voice_cancellation=None,
)
assert components.runtime is runtime
assert components.outbound is outbound
class TestSessionStore:

View file

@ -4,24 +4,26 @@ from unittest.mock import MagicMock, patch
from messaging.platforms.factory import (
MessagingPlatformOptions,
create_messaging_platform,
create_messaging_components,
)
class TestCreateMessagingPlatform:
"""Tests for create_messaging_platform factory function."""
class TestCreateMessagingComponents:
"""Tests for create_messaging_components factory function."""
def test_telegram_with_token(self):
"""Create Telegram platform when bot_token is provided."""
mock_platform = MagicMock()
mock_runtime = MagicMock()
mock_runtime.name = "telegram"
mock_runtime.outbound = MagicMock()
with (
patch("messaging.platforms.telegram.TELEGRAM_AVAILABLE", True),
patch(
"messaging.platforms.telegram.TelegramPlatform",
return_value=mock_platform,
) as platform_cls,
"messaging.platforms.telegram.TelegramRuntime",
return_value=mock_runtime,
) as runtime_cls,
):
result = create_messaging_platform(
result = create_messaging_components(
"telegram",
MessagingPlatformOptions(
telegram_bot_token="test_token",
@ -32,8 +34,11 @@ class TestCreateMessagingPlatform:
),
)
assert result is mock_platform
platform_cls.assert_called_once_with(
assert result is not None
assert result.runtime is mock_runtime
assert result.outbound is mock_runtime.outbound
assert result.voice_cancellation is mock_runtime
runtime_cls.assert_called_once_with(
bot_token="test_token",
allowed_user_id="12345",
voice_note_enabled=False,
@ -49,27 +54,29 @@ class TestCreateMessagingPlatform:
def test_telegram_without_token(self):
"""Return None when no bot_token for Telegram."""
result = create_messaging_platform("telegram")
result = create_messaging_components("telegram")
assert result is None
def test_telegram_empty_token(self):
"""Return None when bot_token is empty string."""
result = create_messaging_platform(
result = create_messaging_components(
"telegram", MessagingPlatformOptions(telegram_bot_token="")
)
assert result is None
def test_discord_with_token(self):
"""Create Discord platform when discord_bot_token is provided."""
mock_platform = MagicMock()
mock_runtime = MagicMock()
mock_runtime.name = "discord"
mock_runtime.outbound = MagicMock()
with (
patch("messaging.platforms.discord.DISCORD_AVAILABLE", True),
patch(
"messaging.platforms.discord.DiscordPlatform",
return_value=mock_platform,
) as platform_cls,
"messaging.platforms.discord.DiscordRuntime",
return_value=mock_runtime,
) as runtime_cls,
):
result = create_messaging_platform(
result = create_messaging_components(
"discord",
MessagingPlatformOptions(
discord_bot_token="test_token",
@ -80,8 +87,11 @@ class TestCreateMessagingPlatform:
),
)
assert result is mock_platform
platform_cls.assert_called_once_with(
assert result is not None
assert result.runtime is mock_runtime
assert result.outbound is mock_runtime.outbound
assert result.voice_cancellation is mock_runtime
runtime_cls.assert_called_once_with(
bot_token="test_token",
allowed_channel_ids="123,456",
voice_note_enabled=False,
@ -97,12 +107,12 @@ class TestCreateMessagingPlatform:
def test_discord_without_token(self):
"""Return None when no discord_bot_token for Discord."""
result = create_messaging_platform("discord")
result = create_messaging_components("discord")
assert result is None
def test_discord_empty_token(self):
"""Return None when discord_bot_token is empty string."""
result = create_messaging_platform(
result = create_messaging_components(
"discord",
MessagingPlatformOptions(
discord_bot_token="",
@ -113,12 +123,12 @@ class TestCreateMessagingPlatform:
def test_unknown_platform(self):
"""Return None for unknown platform types."""
result = create_messaging_platform("slack")
result = create_messaging_components("slack")
assert result is None
def test_unknown_platform_with_kwargs(self):
"""Return None for unknown platform even with kwargs."""
result = create_messaging_platform(
result = create_messaging_components(
"slack", MessagingPlatformOptions(telegram_bot_token="token")
)
assert result is None

View file

@ -3,13 +3,13 @@ from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from telegram.error import NetworkError, RetryAfter, TelegramError
from messaging.platforms.telegram import TelegramPlatform
from messaging.platforms.telegram import TelegramRuntime
@pytest.fixture
def telegram_platform():
with patch("messaging.platforms.telegram.TELEGRAM_AVAILABLE", True):
platform = TelegramPlatform(bot_token="test_token", allowed_user_id="12345")
platform = TelegramRuntime(bot_token="test_token", allowed_user_id="12345")
return platform
@ -31,7 +31,7 @@ async def test_telegram_retry_on_network_error(telegram_platform):
# We need to patch asyncio.sleep to speed up the test
with patch("asyncio.sleep", AsyncMock()) as mock_sleep:
msg_id = await telegram_platform.send_message("chat_1", "hello")
msg_id = await telegram_platform.outbound.send_message("chat_1", "hello")
assert msg_id == "999"
assert mock_bot.send_message.call_count == 3
@ -51,7 +51,7 @@ async def test_telegram_retry_on_retry_after(telegram_platform):
telegram_platform._application.bot = mock_bot
with patch("asyncio.sleep", AsyncMock()) as mock_sleep:
msg_id = await telegram_platform.send_message("chat_1", "hello")
msg_id = await telegram_platform.outbound.send_message("chat_1", "hello")
assert msg_id == "1000"
assert mock_bot.send_message.call_count == 2
@ -69,7 +69,7 @@ async def test_telegram_no_retry_on_bad_request(telegram_platform):
telegram_platform._application.bot = mock_bot
with pytest.raises(TelegramError):
await telegram_platform.send_message("chat_1", "hello")
await telegram_platform.outbound.send_message("chat_1", "hello")
assert mock_bot.send_message.call_count == 1

View file

@ -2,19 +2,19 @@ from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from messaging.platforms.telegram import TelegramPlatform
from messaging.platforms.telegram import TelegramRuntime
@pytest.fixture
def telegram_platform():
with patch("messaging.platforms.telegram.TELEGRAM_AVAILABLE", True):
platform = TelegramPlatform(bot_token="test_token", allowed_user_id="12345")
platform = TelegramRuntime(bot_token="test_token", allowed_user_id="12345")
return platform
def test_telegram_platform_init_no_token():
with patch.dict("os.environ", {}, clear=True):
platform = TelegramPlatform(bot_token=None)
platform = TelegramRuntime(bot_token=None)
assert platform.bot_token is None
@ -47,7 +47,7 @@ async def test_telegram_platform_send_message_success(telegram_platform):
telegram_platform._application = MagicMock()
telegram_platform._application.bot = mock_bot
msg_id = await telegram_platform.send_message("chat_1", "hello")
msg_id = await telegram_platform.outbound.send_message("chat_1", "hello")
assert msg_id == "999"
mock_bot.send_message.assert_called_once_with(
@ -64,7 +64,7 @@ async def test_telegram_platform_edit_message_success(telegram_platform):
telegram_platform._application = MagicMock()
telegram_platform._application.bot = mock_bot
await telegram_platform.edit_message("chat_1", "999", "new text")
await telegram_platform.outbound.edit_message("chat_1", "999", "new text")
mock_bot.edit_message_text.assert_called_once_with(
chat_id="chat_1", message_id=999, text="new text", parse_mode="MarkdownV2"
@ -76,7 +76,9 @@ async def test_telegram_platform_queue_send_message(telegram_platform):
mock_limiter = AsyncMock()
telegram_platform._limiter = mock_limiter
await telegram_platform.queue_send_message("chat_1", "hello", fire_and_forget=False)
await telegram_platform.outbound.queue_send_message(
"chat_1", "hello", fire_and_forget=False
)
mock_limiter.enqueue.assert_called_once()

View file

@ -8,10 +8,10 @@ from telegram.error import NetworkError, RetryAfter, TelegramError
def test_telegram_platform_init_raises_when_dependency_missing():
with patch("messaging.platforms.telegram.TELEGRAM_AVAILABLE", False):
from messaging.platforms.telegram import TelegramPlatform
from messaging.platforms.telegram import TelegramRuntime
with pytest.raises(ImportError):
TelegramPlatform(bot_token="x")
TelegramRuntime(bot_token="x")
@pytest.mark.asyncio
@ -20,9 +20,9 @@ async def test_telegram_platform_start_requires_token():
patch.dict("os.environ", {}, clear=True),
patch("messaging.platforms.telegram.TELEGRAM_AVAILABLE", True),
):
from messaging.platforms.telegram import TelegramPlatform
from messaging.platforms.telegram import TelegramRuntime
platform = TelegramPlatform(bot_token=None)
platform = TelegramRuntime(bot_token=None)
with pytest.raises(ValueError):
await platform.start()
@ -30,9 +30,9 @@ async def test_telegram_platform_start_requires_token():
@pytest.mark.asyncio
async def test_telegram_platform_stop_no_application_is_noop():
with patch("messaging.platforms.telegram.TELEGRAM_AVAILABLE", True):
from messaging.platforms.telegram import TelegramPlatform
from messaging.platforms.telegram import TelegramRuntime
platform = TelegramPlatform(bot_token="t")
platform = TelegramRuntime(bot_token="t")
platform._application = None
platform._connected = True
await platform.stop()
@ -42,22 +42,22 @@ async def test_telegram_platform_stop_no_application_is_noop():
@pytest.mark.asyncio
async def test_with_retry_returns_none_when_message_not_modified_network_error():
with patch("messaging.platforms.telegram.TELEGRAM_AVAILABLE", True):
from messaging.platforms.telegram import TelegramPlatform
from messaging.platforms.telegram import TelegramRuntime
platform = TelegramPlatform(bot_token="t")
platform = TelegramRuntime(bot_token="t")
async def _f():
raise NetworkError("Message is not modified")
assert await platform._with_retry(_f) is None
assert await platform.outbound._with_retry(_f) is None
@pytest.mark.asyncio
async def test_with_retry_retries_network_error_then_succeeds(monkeypatch):
with patch("messaging.platforms.telegram.TELEGRAM_AVAILABLE", True):
from messaging.platforms.telegram import TelegramPlatform
from messaging.platforms.telegram import TelegramRuntime
platform = TelegramPlatform(bot_token="t")
platform = TelegramRuntime(bot_token="t")
monkeypatch.setattr(asyncio, "sleep", AsyncMock())
@ -69,16 +69,16 @@ async def test_with_retry_retries_network_error_then_succeeds(monkeypatch):
raise NetworkError("temporary")
return "ok"
assert await platform._with_retry(_f) == "ok"
assert await platform.outbound._with_retry(_f) == "ok"
assert calls["n"] == 2
@pytest.mark.asyncio
async def test_with_retry_honors_retry_after_timedelta(monkeypatch):
with patch("messaging.platforms.telegram.TELEGRAM_AVAILABLE", True):
from messaging.platforms.telegram import TelegramPlatform
from messaging.platforms.telegram import TelegramRuntime
platform = TelegramPlatform(bot_token="t")
platform = TelegramRuntime(bot_token="t")
monkeypatch.setattr(asyncio, "sleep", AsyncMock())
@ -90,16 +90,16 @@ async def test_with_retry_honors_retry_after_timedelta(monkeypatch):
raise RetryAfter(retry_after=timedelta(seconds=0.01))
return "ok"
assert await platform._with_retry(_f) == "ok"
assert await platform.outbound._with_retry(_f) == "ok"
assert calls["n"] == 2
@pytest.mark.asyncio
async def test_with_retry_drops_parse_mode_on_markdown_entity_error():
with patch("messaging.platforms.telegram.TELEGRAM_AVAILABLE", True):
from messaging.platforms.telegram import TelegramPlatform
from messaging.platforms.telegram import TelegramRuntime
platform = TelegramPlatform(bot_token="t")
platform = TelegramRuntime(bot_token="t")
calls = []
@ -109,58 +109,61 @@ async def test_with_retry_drops_parse_mode_on_markdown_entity_error():
raise TelegramError("Can't parse entities: bad markdown")
return "ok"
assert await platform._with_retry(_f, parse_mode="MarkdownV2") == "ok"
assert await platform.outbound._with_retry(_f, parse_mode="MarkdownV2") == "ok"
assert calls == ["MarkdownV2", None]
@pytest.mark.asyncio
async def test_queue_send_message_without_limiter_calls_send_message():
with patch("messaging.platforms.telegram.TELEGRAM_AVAILABLE", True):
from messaging.platforms.telegram import TelegramPlatform
from messaging.platforms.telegram import TelegramRuntime
platform = TelegramPlatform(bot_token="t")
platform = TelegramRuntime(bot_token="t")
platform._limiter = None
with patch.object(
platform, "send_message", new_callable=AsyncMock
) as mock_send:
mock_send.return_value = "1"
assert await platform.queue_send_message("c", "t") == "1"
mock_send.assert_awaited_once()
platform._application = MagicMock()
mock_msg = MagicMock()
mock_msg.message_id = 1
platform._application.bot = AsyncMock()
platform._application.bot.send_message = AsyncMock(return_value=mock_msg)
assert await platform.outbound.queue_send_message("c", "t") == "1"
platform._application.bot.send_message.assert_awaited_once()
@pytest.mark.asyncio
async def test_queue_edit_message_without_limiter_calls_edit_message():
with patch("messaging.platforms.telegram.TELEGRAM_AVAILABLE", True):
from messaging.platforms.telegram import TelegramPlatform
from messaging.platforms.telegram import TelegramRuntime
platform = TelegramPlatform(bot_token="t")
platform = TelegramRuntime(bot_token="t")
platform._limiter = None
with patch.object(
platform, "edit_message", new_callable=AsyncMock
) as mock_edit:
await platform.queue_edit_message("c", "1", "t")
mock_edit.assert_awaited_once()
platform._application = MagicMock()
platform._application.bot = AsyncMock()
platform._application.bot.edit_message_text = AsyncMock()
await platform.outbound.queue_edit_message("c", "1", "t")
platform._application.bot.edit_message_text.assert_awaited_once()
def test_fire_and_forget_non_coroutine_uses_ensure_future(monkeypatch):
with patch("messaging.platforms.telegram.TELEGRAM_AVAILABLE", True):
from messaging.platforms.telegram import TelegramPlatform
from messaging.platforms.telegram import TelegramRuntime
platform = TelegramPlatform(bot_token="t")
platform = TelegramRuntime(bot_token="t")
ef = MagicMock()
monkeypatch.setattr(asyncio, "ensure_future", ef)
platform.fire_and_forget(MagicMock())
platform.outbound.fire_and_forget(MagicMock())
ef.assert_called_once()
@pytest.mark.asyncio
async def test_on_start_command_replies_and_forwards():
with patch("messaging.platforms.telegram.TELEGRAM_AVAILABLE", True):
from messaging.platforms.telegram import TelegramPlatform
from messaging.platforms.telegram import TelegramRuntime
platform = TelegramPlatform(bot_token="t")
platform = TelegramRuntime(bot_token="t")
with patch.object(
platform, "_on_telegram_message", new_callable=AsyncMock
) as mock_msg:
@ -175,11 +178,11 @@ async def test_on_start_command_replies_and_forwards():
@pytest.mark.asyncio
async def test_on_telegram_message_handler_error_sends_error_message():
with patch("messaging.platforms.telegram.TELEGRAM_AVAILABLE", True):
from messaging.platforms.telegram import TelegramPlatform
from messaging.platforms.telegram import TelegramRuntime
platform = TelegramPlatform(bot_token="t", allowed_user_id="123")
platform = TelegramRuntime(bot_token="t", allowed_user_id="123")
with patch.object(
platform, "send_message", new_callable=AsyncMock
platform.outbound, "send_message", new_callable=AsyncMock
) as mock_send:
async def _boom(_incoming):
@ -201,9 +204,9 @@ async def test_on_telegram_message_handler_error_sends_error_message():
@pytest.mark.asyncio
async def test_telegram_start_retries_on_network_error(monkeypatch):
with patch("messaging.platforms.telegram.TELEGRAM_AVAILABLE", True):
from messaging.platforms.telegram import TelegramPlatform
from messaging.platforms.telegram import TelegramRuntime
platform = TelegramPlatform(bot_token="token", allowed_user_id=None)
platform = TelegramRuntime(bot_token="token", allowed_user_id=None)
monkeypatch.setattr(asyncio, "sleep", AsyncMock())
@ -226,9 +229,9 @@ async def test_telegram_start_retries_on_network_error(monkeypatch):
async def test_edit_message_with_text_exceeding_4096_raises():
"""edit_message with text > 4096 raises TelegramError (BadRequest)."""
with patch("messaging.platforms.telegram.TELEGRAM_AVAILABLE", True):
from messaging.platforms.telegram import TelegramPlatform
from messaging.platforms.telegram import TelegramRuntime
platform = TelegramPlatform(bot_token="t")
platform = TelegramRuntime(bot_token="t")
platform._application = MagicMock()
platform._application.bot = AsyncMock()
platform._application.bot.edit_message_text = AsyncMock(
@ -236,21 +239,21 @@ async def test_edit_message_with_text_exceeding_4096_raises():
)
with pytest.raises(TelegramError):
await platform.edit_message("c", "1", "x" * 5000)
await platform.outbound.edit_message("c", "1", "x" * 5000)
@pytest.mark.asyncio
async def test_edit_message_empty_string():
"""edit_message with empty string - Telegram accepts (no-op edit)."""
with patch("messaging.platforms.telegram.TELEGRAM_AVAILABLE", True):
from messaging.platforms.telegram import TelegramPlatform
from messaging.platforms.telegram import TelegramRuntime
platform = TelegramPlatform(bot_token="t")
platform = TelegramRuntime(bot_token="t")
platform._application = MagicMock()
platform._application.bot = AsyncMock()
platform._application.bot.edit_message_text = AsyncMock()
await platform.edit_message("c", "1", "")
await platform.outbound.edit_message("c", "1", "")
platform._application.bot.edit_message_text.assert_awaited_once_with(
chat_id="c", message_id=1, text="", parse_mode="MarkdownV2"
)
@ -260,16 +263,16 @@ async def test_edit_message_empty_string():
async def test_send_message_empty_string():
"""send_message with empty string - Telegram may reject; we pass through."""
with patch("messaging.platforms.telegram.TELEGRAM_AVAILABLE", True):
from messaging.platforms.telegram import TelegramPlatform
from messaging.platforms.telegram import TelegramRuntime
platform = TelegramPlatform(bot_token="t")
platform = TelegramRuntime(bot_token="t")
platform._application = MagicMock()
mock_msg = MagicMock()
mock_msg.message_id = 1
platform._application.bot = AsyncMock()
platform._application.bot.send_message = AsyncMock(return_value=mock_msg)
msg_id = await platform.send_message("c", "")
msg_id = await platform.outbound.send_message("c", "")
assert msg_id == "1"
platform._application.bot.send_message.assert_awaited_once()
@ -278,9 +281,9 @@ async def test_send_message_empty_string():
async def test_on_telegram_message_non_text_update_ignored():
"""Update with message.photo but no text returns early without calling handler."""
with patch("messaging.platforms.telegram.TELEGRAM_AVAILABLE", True):
from messaging.platforms.telegram import TelegramPlatform
from messaging.platforms.telegram import TelegramRuntime
platform = TelegramPlatform(bot_token="t", allowed_user_id="123")
platform = TelegramRuntime(bot_token="t", allowed_user_id="123")
handler = AsyncMock()
platform.on_message(handler)
@ -300,12 +303,12 @@ async def test_on_telegram_message_non_text_update_ignored():
async def test_with_retry_message_not_found_returns_none():
"""'message to edit not found' returns None without retry."""
with patch("messaging.platforms.telegram.TELEGRAM_AVAILABLE", True):
from messaging.platforms.telegram import TelegramPlatform
from messaging.platforms.telegram import TelegramRuntime
platform = TelegramPlatform(bot_token="t")
platform = TelegramRuntime(bot_token="t")
async def _f():
raise TelegramError("message to edit not found")
result = await platform._with_retry(_f)
result = await platform.outbound._with_retry(_f)
assert result is None

View file

@ -6,21 +6,22 @@ from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from messaging.platforms.discord import DISCORD_AVAILABLE, DiscordPlatform
from messaging.platforms.telegram import TelegramPlatform
from messaging.platforms.discord import DISCORD_AVAILABLE, DiscordRuntime
from messaging.platforms.discord_inbound import get_audio_attachment
from messaging.platforms.telegram import TelegramRuntime
@pytest.fixture
def telegram_platform():
with patch("messaging.platforms.telegram.TELEGRAM_AVAILABLE", True):
return TelegramPlatform(bot_token="test_token", allowed_user_id="12345")
return TelegramRuntime(bot_token="test_token", allowed_user_id="12345")
@pytest.mark.asyncio
async def test_telegram_voice_disabled_sends_reply():
"""When voice_note_enabled is False, reply with disabled message."""
with patch("messaging.platforms.telegram.TELEGRAM_AVAILABLE", True):
telegram_platform = TelegramPlatform(
telegram_platform = TelegramRuntime(
bot_token="test_token",
allowed_user_id="12345",
voice_note_enabled=False,
@ -87,7 +88,7 @@ async def test_telegram_voice_success_invokes_handler(telegram_platform):
return_value="Hello from voice",
),
patch.object(
telegram_platform,
telegram_platform.outbound,
"queue_send_message",
mock_queue_send,
),
@ -116,44 +117,40 @@ class TestDiscordGetAudioAttachment:
"""Tests for _get_audio_attachment helper."""
def test_returns_none_when_no_attachments(self):
platform = DiscordPlatform(bot_token="token")
msg = MagicMock()
msg.attachments = []
assert platform._get_audio_attachment(msg) is None
assert get_audio_attachment(msg) is None
def test_returns_none_when_no_audio_attachments(self):
platform = DiscordPlatform(bot_token="token")
msg = MagicMock()
att = MagicMock()
att.content_type = "image/png"
att.filename = "pic.png"
msg.attachments = [att]
assert platform._get_audio_attachment(msg) is None
assert get_audio_attachment(msg) is None
def test_returns_attachment_by_content_type(self):
platform = DiscordPlatform(bot_token="token")
msg = MagicMock()
att = MagicMock()
att.content_type = "audio/ogg"
att.filename = "voice.ogg"
msg.attachments = [att]
assert platform._get_audio_attachment(msg) is att
assert get_audio_attachment(msg) is att
def test_returns_attachment_by_extension(self):
platform = DiscordPlatform(bot_token="token")
msg = MagicMock()
att = MagicMock()
att.content_type = "application/octet-stream"
att.filename = "voice.ogg"
msg.attachments = [att]
assert platform._get_audio_attachment(msg) is att
assert get_audio_attachment(msg) is att
@pytest.mark.skipif(not DISCORD_AVAILABLE, reason="discord.py not installed")
@pytest.mark.asyncio
async def test_discord_voice_disabled_sends_reply():
"""When voice_note_enabled is False, reply with disabled message."""
platform = DiscordPlatform(
platform = DiscordRuntime(
bot_token="token",
allowed_channel_ids="123",
voice_note_enabled=False,

2
uv.lock generated
View file

@ -561,7 +561,7 @@ wheels = [
[[package]]
name = "free-claude-code"
version = "2.3.13"
version = "2.3.14"
source = { editable = "." }
dependencies = [
{ name = "aiohttp" },