feat: sync architecture upgrades from omnicloud-support-agent

Major improvements:
- SQS: Upgrade to FIFO queue for message ordering guarantee
- Docker: Upgrade Node.js to 20+ for MCP undici compatibility
- Docker: Change HOME to /tmp for MCP auth file writes
- Security: Add Telegram webhook secret token verification (HMAC)
- Handler: Add FIFO MessageGroupId/DeduplicationId support
- Handler: Add /debug command for session file download
- Handler: Add /start welcome message handler
- Handler: Refactor local commands to support handler type
- Consumer: Add continuous typing indicator (every 4s)
- Consumer: Add Markdown conversion pipeline for Telegram
- Consumer: Add message_time passthrough
- Config: Add LocalCommand dataclass for handler type support
- Dependencies: Add telegramify-markdown for Markdown conversion

Author: BukeLy

agent-sdk-client/config.py

@@ -6,6 +6,14 @@
 from pathlib import Path
 from typing import Optional
 
+
+@dataclass
+class LocalCommand:
+    """Local command configuration."""
+    type: str  # "static" or "handler"
+    response: str = ""  # for static type
+    handler: str = ""  # for handler type
+
 logger = logging.getLogger(__name__)
 DEFAULT_CONFIG_PATH = Path(__file__).with_name("config.toml")
 
@@ -28,7 +36,46 @@ def extract_command(text: Optional[str]) -> Optional[str]:
     return command
 
 
-def _load_config(config_path: Path = DEFAULT_CONFIG_PATH) -> tuple[list[str], dict[str, str], list[int | str]]:
+def _parse_local_command(name: str, value) -> tuple[str, LocalCommand | None]:
+    """Parse a single local command entry.
+
+    Args:
+        name: Command name (with or without leading slash)
+        value: Command value (string for legacy, dict for new format)
+
+    Returns:
+        Tuple of (normalized_cmd, LocalCommand) or (normalized_cmd, None) if invalid
+    """
+    cmd = f"/{name.lstrip('/')}" if not name.startswith('/') else name
+
+    # Legacy format: string value = static response
+    if isinstance(value, str):
+        return cmd, LocalCommand(type="static", response=value)
+
+    # New format: dict with type field
+    if isinstance(value, dict):
+        cmd_type = value.get('type', '')
+        if cmd_type == 'static':
+            response = value.get('response', '')
+            if not response:
+                logger.warning(f"Local command {cmd} has no response; skipping")
+                return cmd, None
+            return cmd, LocalCommand(type="static", response=response)
+        elif cmd_type == 'handler':
+            handler = value.get('handler', '')
+            if not handler:
+                logger.warning(f"Local command {cmd} has no handler; skipping")
+                return cmd, None
+            return cmd, LocalCommand(type="handler", handler=handler)
+        else:
+            logger.warning(f"Local command {cmd} has unknown type: {cmd_type}; skipping")
+            return cmd, None
+
+    logger.warning(f"Local command {cmd} has invalid value type; skipping")
+    return cmd, None
+
+
+def _load_config(config_path: Path = DEFAULT_CONFIG_PATH) -> tuple[list[str], dict[str, LocalCommand], list[int | str]]:
     """Load commands and security config from TOML config file.
 
     Returns:
@@ -48,16 +95,18 @@ def _load_config(config_path: Path = DEFAULT_CONFIG_PATH) -> tuple[list[str], di
             agent_commands = []
         agent_commands = [cmd for cmd in agent_commands if isinstance(cmd, str)]
 
-        # Load local commands
+        # Load local commands (supports both legacy string and new dict format)
         local_commands_raw = data.get('local_commands', {})
         if not isinstance(local_commands_raw, dict):
             logger.warning("Local commands config is not a table; ignoring configuration")
             local_commands_raw = {}
-        local_commands = {
-            f"/{name.lstrip('/')}" if not name.startswith('/') else name: str(value)
-            for name, value in local_commands_raw.items()
-            if isinstance(name, str) and isinstance(value, str)
-        }
+        local_commands: dict[str, LocalCommand] = {}
+        for name, value in local_commands_raw.items():
+            if not isinstance(name, str):
+                continue
+            cmd, parsed = _parse_local_command(name, value)
+            if parsed:
+                local_commands[cmd] = parsed
 
         # Load security whitelist
         security = data.get('security', {})
@@ -92,8 +141,9 @@ class Config:
     auth_token: str
     queue_url: str
     agent_commands: list[str]
-    local_commands: dict[str, str]
+    local_commands: dict[str, LocalCommand]
     user_whitelist: list[int | str]
+    telegram_webhook_secret: str = ""
 
     @classmethod
     def from_env(cls, config_path: Optional[Path] = None) -> 'Config':
@@ -107,6 +157,7 @@ def from_env(cls, config_path: Optional[Path] = None) -> 'Config':
             agent_commands=agent_cmds,
             local_commands=local_cmds,
             user_whitelist=whitelist,
+            telegram_webhook_secret=os.getenv('TELEGRAM_WEBHOOK_SECRET', ''),
         )
 
     def get_command(self, text: Optional[str]) -> Optional[str]:
@@ -118,8 +169,16 @@ def is_agent_command(self, cmd: Optional[str]) -> bool:
     def is_local_command(self, cmd: Optional[str]) -> bool:
         return bool(cmd) and cmd in self.local_commands
 
+    def get_local_command(self, cmd: str) -> LocalCommand | None:
+        """Get local command config by command name."""
+        return self.local_commands.get(cmd)
+
     def local_response(self, cmd: str) -> str:
-        return self.local_commands.get(cmd, "Unsupported command.")
+        """Get static response for a local command (legacy compatibility)."""
+        local_cmd = self.local_commands.get(cmd)
+        if local_cmd and local_cmd.type == "static":
+            return local_cmd.response
+        return "Unsupported command."
 
     def unknown_command_message(self) -> str:
         parts = []

agent-sdk-client/config.toml

@@ -7,8 +7,15 @@ commands = [
 
 [local_commands]
 # Local-only commands handled by the client
-help = "Hello World"
-newchat = "创建新对话"
+# Supports two formats:
+#   - Legacy: help = "static response text"
+#   - New: help = { type = "static", response = "text" }
+#   - Handler: newchat = { type = "handler", handler = "newchat" }
+
+help = { type = "static", response = "Hello World" }
+newchat = { type = "handler", handler = "newchat" }
+start = { type = "handler", handler = "start" }
+debug = { type = "handler", handler = "debug" }
 
 [security]
 # User IDs allowed to add bot to groups and send private messages.

agent-sdk-client/consumer.py

@@ -4,17 +4,91 @@
 """
 import asyncio
 import json
+import re
 from typing import Any
 
 import httpx
 from telegram import Bot, Update
 from telegram.constants import ParseMode, ChatAction
-from telegram.helpers import escape_markdown
+from telegramify_markdown import markdownify
 from telegram.error import BadRequest
 
 from config import Config
 
 
+def fix_heading_bold(text: str) -> str:
+    """Remove bold markers from headings: ## **Title** -> ## Title.
+
+    Only applies when heading contains **bold** markers.
+    """
+    if re.search(r'^#{1,6}\s*\*\*', text, flags=re.MULTILINE):
+        return re.sub(r'^(#{1,6})\s*\*\*(.+?)\*\*\s*$', r'\1 \2', text, flags=re.MULTILINE)
+    return text
+
+
+def fix_code_escaping(text: str) -> str:
+    """Remove escaping inside code blocks: \\| -> |, \\- -> -.
+
+    Only applies when code blocks contain escaped characters.
+    """
+    if '```' not in text and '`' not in text:
+        return text
+
+    escaped_chars = '`|-.()+!#={}[]><_*~'
+
+    def unescape(content: str) -> str:
+        for char in escaped_chars:
+            content = content.replace(f'\\{char}', char)
+        return content
+
+    # Fix fenced code blocks
+    if '```' in text:
+        text = re.sub(
+            r'```(.*?)```',
+            lambda m: f'```{unescape(m.group(1))}```',
+            text,
+            flags=re.DOTALL
+        )
+    # Fix inline code
+    if '`' in text:
+        text = re.sub(
+            r'`([^`]+)`',
+            lambda m: f'`{unescape(m.group(1))}`',
+            text
+        )
+    return text
+
+
+def fix_unescaped_chars(text: str) -> str:
+    """Escape special chars outside code blocks that markdownify missed.
+
+    Only applies when unescaped special chars exist outside code blocks.
+    """
+    # Extract code blocks to protect them
+    blocks = []
+    def save(m):
+        blocks.append(m.group(0))
+        return f'\x00{len(blocks)-1}\x00'
+
+    protected = re.sub(r'```.*?```', save, text, flags=re.DOTALL)
+    protected = re.sub(r'`[^`]+`', save, protected)
+
+    # Check if any unescaped chars exist
+    chars = r'-.!()+=|{}[]#>'
+    if not re.search(rf'(?<!\\)[{re.escape(chars)}]', protected):
+        return text
+
+    # Escape unescaped special chars
+    for char in chars:
+        protected = re.sub(rf'(?<!\\){re.escape(char)}', f'\\{char}', protected)
+
+    # Restore code blocks
+    for i, block in enumerate(blocks):
+        protected = protected.replace(f'\x00{i}\x00', block)
+
+    return protected
+
+
 def lambda_handler(event: dict, context: Any) -> dict:
     """SQS Consumer Lambda entry point."""
     for record in event['Records']:
@@ -43,6 +117,7 @@ async def process_message(message_data: dict) -> None:
     """Process single message from SQS queue."""
     import logging
     logger = logging.getLogger()
+    logger.setLevel(logging.INFO)
 
     config = Config.from_env()
     bot = Bot(config.telegram_token)
@@ -55,51 +130,6 @@ async def process_message(message_data: dict) -> None:
         logger.warning("Received update with no message or edited_message")
         return
 
-    cmd = config.get_command(message.text)
-    if cmd:
-        if config.is_local_command(cmd):
-            logger.info(
-                "Handling local command in consumer (fallback path)",
-                extra={'chat_id': message.chat_id, 'message_id': message.message_id},
-            )
-            try:
-                await bot.send_message(
-                    chat_id=message.chat_id,
-                    text=config.local_response(cmd),
-                    message_thread_id=message.message_thread_id,
-                    reply_to_message_id=message.message_id,
-                )
-            except Exception:
-                logger.warning("Failed to send local command response", exc_info=True)
-            return
-
-        if not config.is_agent_command(cmd):
-            # Defensive guard: producer should already block non-agent commands.
-            logger.info(
-                "Skipping non-agent command (consumer fallback)",
-                extra={
-                    'chat_id': message.chat_id,
-                    'message_id': message.message_id,
-                },
-            )
-            try:
-                await bot.send_message(
-                    chat_id=message.chat_id,
-                    text=config.unknown_command_message(),
-                    message_thread_id=message.message_thread_id,
-                    reply_to_message_id=message.message_id,
-                )
-            except Exception:
-                logger.warning("Failed to send local command response", exc_info=True)
-            return
-
-    # Send typing indicator
-    await bot.send_chat_action(
-        chat_id=message.chat_id,
-        action=ChatAction.TYPING,
-        message_thread_id=message.message_thread_id,
-    )
-
     # Initialize result with default error response
     # This ensures result is always defined, even if Agent Server call fails
     result = {
@@ -112,8 +142,21 @@ async def process_message(message_data: dict) -> None:
     user_message = message_data.get('text') or message.text
     thread_id = message_data.get('thread_id') or message.message_thread_id
 
-    # Call Agent Server
-    try:
+    async def keep_typing():
+        """Send typing indicator every 4 seconds (Telegram typing expires after 5s)."""
+        while True:
+            try:
+                await bot.send_chat_action(
+                    chat_id=message.chat_id,
+                    action=ChatAction.TYPING,
+                    message_thread_id=thread_id,
+                )
+            except Exception:
+                pass  # Ignore typing errors, don't interrupt main flow
+            await asyncio.sleep(4)
+
+    async def call_agent_server():
+        """Call Agent Server and return result."""
         async with httpx.AsyncClient(timeout=600.0) as client:
             response = await client.post(
                 config.agent_server_url,
@@ -125,10 +168,16 @@ async def process_message(message_data: dict) -> None:
                     'user_message': user_message,
                     'chat_id': str(message.chat_id),
                     'thread_id': str(thread_id) if thread_id else None,
+                    'message_time': message_data.get('message_time'),
                 },
             )
             response.raise_for_status()
-            result = response.json()
+            return response.json()
+
+    # Call Agent Server with continuous typing indicator
+    typing_task = asyncio.create_task(keep_typing())
+    try:
+        result = await call_agent_server()
 
     except httpx.TimeoutException:
         logger.warning(f"Agent Server timeout for chat_id={message.chat_id}")
@@ -152,6 +201,14 @@ async def process_message(message_data: dict) -> None:
             logger.error(f"Failed to send error message to Telegram: {send_error}")
         # Don't re-raise - error message already sent to user, retrying would cause duplicate messages
 
+    finally:
+        # Stop typing indicator
+        typing_task.cancel()
+        try:
+            await typing_task
+        except asyncio.CancelledError:
+            pass
+
     # Format response (result is guaranteed to be defined now)
     if result.get('is_error'):
         text = f"Agent error: {result.get('error_message', 'Unknown')}"
@@ -162,23 +219,38 @@ async def process_message(message_data: dict) -> None:
         text = text[:4000] + "\n\n... (truncated)"
 
     # Send response to Telegram
+    # Convert standard Markdown to Telegram MarkdownV2 format
+    # Pipeline: fix_heading_bold -> markdownify -> fix_code_escaping -> fix_unescaped_chars
+    telegram_text = fix_unescaped_chars(fix_code_escaping(markdownify(fix_heading_bold(text))))
+
+    # Only reply_to original message if thread_id matches (not for /newchat)
+    reply_to_id = (
+        message.message_id
+        if thread_id == message.message_thread_id
+        else None
+    )
+
     try:
         await bot.send_message(
             chat_id=message.chat_id,
-            text=text,
+            text=telegram_text,
             parse_mode=ParseMode.MARKDOWN_V2,
             message_thread_id=thread_id,
-            reply_to_message_id=message.message_id,
+            reply_to_message_id=reply_to_id,
+        )
+        logger.info(
+            "Sent response to Telegram",
+            extra={'chat_id': message.chat_id, 'thread_id': thread_id},
         )
     except BadRequest as e:
+        logger.warning(f"BadRequest sending message: {e}")
         if "parse entities" in str(e).lower():
-            safe_text = escape_markdown(text, version=2)
+            # Fallback: send as plain text without any formatting
             await bot.send_message(
                 chat_id=message.chat_id,
-                text=safe_text,
-                parse_mode=ParseMode.MARKDOWN_V2,
+                text=text,
                 message_thread_id=thread_id,
-                reply_to_message_id=message.message_id,
+                reply_to_message_id=reply_to_id,
             )
         else:
             raise