diff --git a/gateway/platforms/base.py b/gateway/platforms/base.py index c33c2924a81c..f91a8b8f3951 100644 --- a/gateway/platforms/base.py +++ b/gateway/platforms/base.py @@ -8,8 +8,8 @@ import asyncio import logging import os -import random import re +import shutil import uuid from abc import ABC, abstractmethod @@ -27,7 +27,6 @@ from gateway.config import Platform, PlatformConfig from gateway.session import SessionSource, build_session_key from hermes_cli.config import get_hermes_home -from hermes_constants import get_hermes_dir GATEWAY_SECRET_CAPTURE_UNSUPPORTED_MESSAGE = ( @@ -45,14 +44,15 @@ # (e.g. Telegram file URLs expire after ~1 hour). # --------------------------------------------------------------------------- -# Default location: {HERMES_HOME}/cache/images/ (legacy: image_cache/) -IMAGE_CACHE_DIR = get_hermes_dir("cache/images", "image_cache") +# Default location: {HERMES_HOME}/image_cache/ +IMAGE_CACHE_DIR: Path | None = None def get_image_cache_dir() -> Path: """Return the image cache directory, creating it if it doesn't exist.""" - IMAGE_CACHE_DIR.mkdir(parents=True, exist_ok=True) - return IMAGE_CACHE_DIR + cache_dir = IMAGE_CACHE_DIR or (get_hermes_home() / "image_cache") + cache_dir.mkdir(parents=True, exist_ok=True) + return cache_dir def cache_image_from_bytes(data: bytes, ext: str = ".jpg") -> str: @@ -73,51 +73,42 @@ def cache_image_from_bytes(data: bytes, ext: str = ".jpg") -> str: return str(filepath) -async def cache_image_from_url(url: str, ext: str = ".jpg", retries: int = 2) -> str: +def cache_image_from_path(source_path: str | Path, ext: str | None = None) -> str: + """Copy an image file into the cache and return the cached absolute path.""" + source = Path(source_path) + cache_dir = get_image_cache_dir() + suffix = (ext or source.suffix or ".jpg").lower() + filepath = cache_dir / f"img_{uuid.uuid4().hex[:12]}{suffix}" + with source.open("rb") as src, filepath.open("wb") as dst: + shutil.copyfileobj(src, dst) + return str(filepath) + + +async def cache_image_from_url(url: str, ext: str = ".jpg") -> str: """ Download an image from a URL and save it to the local cache. - Retries on transient failures (timeouts, 429, 5xx) with exponential - backoff so a single slow CDN response doesn't lose the media. + Uses httpx for async download with a reasonable timeout. Args: url: The HTTP/HTTPS URL to download from. ext: File extension including the dot (e.g. ".jpg", ".png"). - retries: Number of retry attempts on transient failures. Returns: Absolute path to the cached image file as a string. """ - import asyncio import httpx - import logging as _logging - _log = _logging.getLogger(__name__) - last_exc = None async with httpx.AsyncClient(timeout=30.0, follow_redirects=True) as client: - for attempt in range(retries + 1): - try: - response = await client.get( - url, - headers={ - "User-Agent": "Mozilla/5.0 (compatible; HermesAgent/1.0)", - "Accept": "image/*,*/*;q=0.8", - }, - ) - response.raise_for_status() - return cache_image_from_bytes(response.content, ext) - except (httpx.TimeoutException, httpx.HTTPStatusError) as exc: - last_exc = exc - if isinstance(exc, httpx.HTTPStatusError) and exc.response.status_code < 429: - raise - if attempt < retries: - wait = 1.5 * (attempt + 1) - _log.debug("Media cache retry %d/%d for %s (%.1fs): %s", - attempt + 1, retries, url[:80], wait, exc) - await asyncio.sleep(wait) - continue - raise - raise last_exc + response = await client.get( + url, + headers={ + "User-Agent": "Mozilla/5.0 (compatible; HermesAgent/1.0)", + "Accept": "image/*,*/*;q=0.8", + }, + ) + response.raise_for_status() + return cache_image_from_bytes(response.content, ext) def cleanup_image_cache(max_age_hours: int = 24) -> int: @@ -148,13 +139,14 @@ def cleanup_image_cache(max_age_hours: int = 24) -> int: # here so the STT tool (OpenAI Whisper) can transcribe them from local files. # --------------------------------------------------------------------------- -AUDIO_CACHE_DIR = get_hermes_dir("cache/audio", "audio_cache") +AUDIO_CACHE_DIR: Path | None = None def get_audio_cache_dir() -> Path: """Return the audio cache directory, creating it if it doesn't exist.""" - AUDIO_CACHE_DIR.mkdir(parents=True, exist_ok=True) - return AUDIO_CACHE_DIR + cache_dir = AUDIO_CACHE_DIR or (get_hermes_home() / "audio_cache") + cache_dir.mkdir(parents=True, exist_ok=True) + return cache_dir def cache_audio_from_bytes(data: bytes, ext: str = ".ogg") -> str: @@ -175,51 +167,29 @@ def cache_audio_from_bytes(data: bytes, ext: str = ".ogg") -> str: return str(filepath) -async def cache_audio_from_url(url: str, ext: str = ".ogg", retries: int = 2) -> str: +async def cache_audio_from_url(url: str, ext: str = ".ogg") -> str: """ Download an audio file from a URL and save it to the local cache. - Retries on transient failures (timeouts, 429, 5xx) with exponential - backoff so a single slow CDN response doesn't lose the media. - Args: url: The HTTP/HTTPS URL to download from. ext: File extension including the dot (e.g. ".ogg", ".mp3"). - retries: Number of retry attempts on transient failures. Returns: Absolute path to the cached audio file as a string. """ - import asyncio import httpx - import logging as _logging - _log = _logging.getLogger(__name__) - last_exc = None async with httpx.AsyncClient(timeout=30.0, follow_redirects=True) as client: - for attempt in range(retries + 1): - try: - response = await client.get( - url, - headers={ - "User-Agent": "Mozilla/5.0 (compatible; HermesAgent/1.0)", - "Accept": "audio/*,*/*;q=0.8", - }, - ) - response.raise_for_status() - return cache_audio_from_bytes(response.content, ext) - except (httpx.TimeoutException, httpx.HTTPStatusError) as exc: - last_exc = exc - if isinstance(exc, httpx.HTTPStatusError) and exc.response.status_code < 429: - raise - if attempt < retries: - wait = 1.5 * (attempt + 1) - _log.debug("Audio cache retry %d/%d for %s (%.1fs): %s", - attempt + 1, retries, url[:80], wait, exc) - await asyncio.sleep(wait) - continue - raise - raise last_exc + response = await client.get( + url, + headers={ + "User-Agent": "Mozilla/5.0 (compatible; HermesAgent/1.0)", + "Accept": "audio/*,*/*;q=0.8", + }, + ) + response.raise_for_status() + return cache_audio_from_bytes(response.content, ext) # --------------------------------------------------------------------------- @@ -229,22 +199,31 @@ async def cache_audio_from_url(url: str, ext: str = ".ogg", retries: int = 2) -> # here so the agent can reference them by local file path. # --------------------------------------------------------------------------- -DOCUMENT_CACHE_DIR = get_hermes_dir("cache/documents", "document_cache") +DOCUMENT_CACHE_DIR: Path | None = None SUPPORTED_DOCUMENT_TYPES = { ".pdf": "application/pdf", ".md": "text/markdown", ".txt": "text/plain", + ".csv": "text/csv", + ".tsv": "text/tab-separated-values", + ".zip": "application/zip", ".docx": "application/vnd.openxmlformats-officedocument.wordprocessingml.document", ".xlsx": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", ".pptx": "application/vnd.openxmlformats-officedocument.presentationml.presentation", } +TEXT_INJECTABLE_DOCUMENT_EXTENSIONS = {".md", ".txt", ".csv", ".tsv"} +MAX_TEXT_INJECT_BYTES = 100 * 1024 +CSV_PREVIEW_LINE_LIMIT = 120 +CSV_PREVIEW_CHAR_LIMIT = 16 * 1024 + def get_document_cache_dir() -> Path: """Return the document cache directory, creating it if it doesn't exist.""" - DOCUMENT_CACHE_DIR.mkdir(parents=True, exist_ok=True) - return DOCUMENT_CACHE_DIR + cache_dir = DOCUMENT_CACHE_DIR or (get_hermes_home() / "document_cache") + cache_dir.mkdir(parents=True, exist_ok=True) + return cache_dir def cache_document_from_bytes(data: bytes, filename: str) -> str: @@ -279,6 +258,132 @@ def cache_document_from_bytes(data: bytes, filename: str) -> str: return str(filepath) +def cache_document_from_path(source_path: str | Path, filename: str | None = None) -> str: + """Copy a document file into the cache and return the cached absolute path.""" + source = Path(source_path) + cache_dir = get_document_cache_dir() + safe_name = Path(filename or source.name).name if (filename or source.name) else "document" + safe_name = safe_name.replace("\x00", "").strip() + if not safe_name or safe_name in (".", ".."): + safe_name = "document" + cached_name = f"doc_{uuid.uuid4().hex[:12]}_{safe_name}" + filepath = cache_dir / cached_name + if not filepath.resolve().is_relative_to(cache_dir.resolve()): + raise ValueError(f"Path traversal rejected: {filename!r}") + with source.open("rb") as src, filepath.open("wb") as dst: + shutil.copyfileobj(src, dst) + return str(filepath) + + +def _sanitize_attachment_display_name(display_name: str, ext: str) -> str: + return re.sub(r"[^\w.\- ]", "_", display_name or f"document{ext}") + + +def _build_csv_preview(text_content: str) -> str: + preview_lines = [] + char_count = 0 + for line in text_content.splitlines(): + if len(preview_lines) >= CSV_PREVIEW_LINE_LIMIT or char_count >= CSV_PREVIEW_CHAR_LIMIT: + break + remaining = CSV_PREVIEW_CHAR_LIMIT - char_count + if remaining <= 0: + break + line = line[:remaining] + preview_lines.append(line) + char_count += len(line) + 1 + return "\n".join(preview_lines).strip() + + +def build_text_attachment_injection(raw_bytes: bytes, display_name: str, ext: str) -> Optional[str]: + """Return injected prompt text for supported text attachments. + + Small UTF-8 text files are injected in full. Large CSV/TSV files are + previewed so the agent still sees the schema and first rows without + flooding context. + """ + ext = (ext or "").lower() + if ext not in TEXT_INJECTABLE_DOCUMENT_EXTENSIONS: + return None + + try: + text_content = raw_bytes.decode("utf-8") + except UnicodeDecodeError: + return None + + safe_name = _sanitize_attachment_display_name(display_name, ext) + should_preview_csv = ext in {".csv", ".tsv"} and len(raw_bytes) > CSV_PREVIEW_CHAR_LIMIT + if len(raw_bytes) <= MAX_TEXT_INJECT_BYTES and not should_preview_csv: + return f"[Content of {safe_name}]:\n{text_content}" + + if ext not in {".csv", ".tsv"}: + return None + + preview = _build_csv_preview(text_content) + if not preview: + return None + + return ( + f"[Preview of {safe_name}]:\n{preview}\n\n" + "[The preview is truncated because the original attachment is large.]" + ) + + +def build_text_attachment_injection_from_path( + path: str | Path, + *, + display_name: str | None = None, +) -> Optional[str]: + """Return injected prompt text for a supported text attachment on disk.""" + file_path = Path(path) + ext = file_path.suffix.lower() + if ext not in TEXT_INJECTABLE_DOCUMENT_EXTENSIONS: + return None + + safe_name = _sanitize_attachment_display_name(display_name or file_path.name, ext) + + try: + file_size = file_path.stat().st_size + except OSError: + return None + + should_preview_csv = ext in {".csv", ".tsv"} and file_size > CSV_PREVIEW_CHAR_LIMIT + + if file_size <= MAX_TEXT_INJECT_BYTES and not should_preview_csv: + try: + text_content = file_path.read_text(encoding="utf-8") + except UnicodeDecodeError: + return None + return f"[Content of {safe_name}]:\n{text_content}" + + if ext not in {".csv", ".tsv"}: + return None + + try: + with file_path.open("r", encoding="utf-8") as handle: + preview_lines = [] + char_count = 0 + for raw_line in handle: + if len(preview_lines) >= CSV_PREVIEW_LINE_LIMIT or char_count >= CSV_PREVIEW_CHAR_LIMIT: + break + remaining = CSV_PREVIEW_CHAR_LIMIT - char_count + if remaining <= 0: + break + line = raw_line.rstrip("\n")[:remaining] + preview_lines.append(line) + char_count += len(line) + 1 + except UnicodeDecodeError: + return None + + preview = "\n".join(preview_lines).strip() + if not preview: + return None + + return ( + f"[Preview of {safe_name}]:\n{preview}\n\n" + "[The preview is truncated because the original attachment is large.]" + ) + + def cleanup_document_cache(max_age_hours: int = 24) -> int: """ Delete cached documents older than *max_age_hours*. @@ -340,9 +445,6 @@ class MessageEvent: reply_to_message_id: Optional[str] = None reply_to_text: Optional[str] = None # Text of the replied-to message (for context injection) - # Auto-loaded skill for topic/channel bindings (e.g., Telegram DM Topics) - auto_skill: Optional[str] = None - # Timestamps timestamp: datetime = field(default_factory=datetime.now) @@ -356,10 +458,7 @@ def get_command(self) -> Optional[str]: return None # Split on space and get first word, strip the / parts = self.text.split(maxsplit=1) - raw = parts[0][1:].lower() if parts else None - if raw and "@" in raw: - raw = raw.split("@", 1)[0] - return raw + return parts[0][1:].lower() if parts else None def get_command_args(self) -> str: """Get the arguments after a command.""" @@ -376,24 +475,6 @@ class SendResult: message_id: Optional[str] = None error: Optional[str] = None raw_response: Any = None - retryable: bool = False # True for transient errors (network, timeout) — base will retry automatically - - -# Error substrings that indicate a transient network failure worth retrying -_RETRYABLE_ERROR_PATTERNS = ( - "connecterror", - "connectionerror", - "connectionreset", - "connectionrefused", - "timeout", - "timed out", - "network", - "broken pipe", - "remotedisconnected", - "eoferror", - "readtimeout", - "writetimeout", -) # Type for message handlers @@ -898,111 +979,6 @@ async def _keep_typing(self, chat_id: str, interval: float = 2.0, metadata=None) except Exception: pass - # ── Processing lifecycle hooks ────────────────────────────────────────── - # Subclasses override these to react to message processing events - # (e.g. Discord adds 👀/✅/❌ reactions). - - async def on_processing_start(self, event: MessageEvent) -> None: - """Hook called when background processing begins.""" - - async def on_processing_complete(self, event: MessageEvent, success: bool) -> None: - """Hook called when background processing completes.""" - - async def _run_processing_hook(self, hook_name: str, *args: Any, **kwargs: Any) -> None: - """Run a lifecycle hook without letting failures break message flow.""" - hook = getattr(self, hook_name, None) - if not callable(hook): - return - try: - await hook(*args, **kwargs) - except Exception as e: - logger.warning("[%s] %s hook failed: %s", self.name, hook_name, e) - - @staticmethod - def _is_retryable_error(error: Optional[str]) -> bool: - """Return True if the error string looks like a transient network failure.""" - if not error: - return False - lowered = error.lower() - return any(pat in lowered for pat in _RETRYABLE_ERROR_PATTERNS) - - async def _send_with_retry( - self, - chat_id: str, - content: str, - reply_to: Optional[str] = None, - metadata: Any = None, - max_retries: int = 2, - base_delay: float = 2.0, - ) -> "SendResult": - """ - Send a message with automatic retry for transient network errors. - - On permanent failures (e.g. formatting / permission errors) falls back - to a plain-text version before giving up. If all attempts fail due to - network errors, sends the user a brief delivery-failure notice so they - know to retry rather than waiting indefinitely. - """ - - result = await self.send( - chat_id=chat_id, - content=content, - reply_to=reply_to, - metadata=metadata, - ) - - if result.success: - return result - - error_str = result.error or "" - is_network = result.retryable or self._is_retryable_error(error_str) - - if is_network: - # Retry with exponential backoff for transient errors - for attempt in range(1, max_retries + 1): - delay = base_delay * (2 ** (attempt - 1)) + random.uniform(0, 1) - logger.warning( - "[%s] Send failed (attempt %d/%d, retrying in %.1fs): %s", - self.name, attempt, max_retries, delay, error_str, - ) - await asyncio.sleep(delay) - result = await self.send( - chat_id=chat_id, - content=content, - reply_to=reply_to, - metadata=metadata, - ) - if result.success: - logger.info("[%s] Send succeeded on retry %d", self.name, attempt) - return result - error_str = result.error or "" - if not (result.retryable or self._is_retryable_error(error_str)): - break # error switched to non-transient — fall through to plain-text fallback - else: - # All retries exhausted (loop completed without break) — notify user - logger.error("[%s] Failed to deliver response after %d retries: %s", self.name, max_retries, error_str) - notice = ( - "\u26a0\ufe0f Message delivery failed after multiple attempts. " - "Please try again \u2014 your request was processed but the response could not be sent." - ) - try: - await self.send(chat_id=chat_id, content=notice, reply_to=reply_to, metadata=metadata) - except Exception as notify_err: - logger.debug("[%s] Could not send delivery-failure notice: %s", self.name, notify_err) - return result - - # Non-network / post-retry formatting failure: try plain text as fallback - logger.warning("[%s] Send failed: %s — trying plain-text fallback", self.name, error_str) - fallback_result = await self.send( - chat_id=chat_id, - content=f"(Response formatting failed, plain text:)\n\n{content[:3500]}", - reply_to=reply_to, - metadata=metadata, - ) - if not fallback_result.success: - logger.error("[%s] Fallback send also failed: %s", self.name, fallback_result.error) - return fallback_result - async def handle_message(self, event: MessageEvent) -> None: """ Process an incoming message. @@ -1025,7 +1001,7 @@ async def handle_message(self, event: MessageEvent) -> None: # simultaneous messages. Queue them without interrupting the active run, # then process them immediately after the current task finishes. if event.message_type == MessageType.PHOTO: - logger.debug("[%s] Queuing photo follow-up for session %s without interrupt", self.name, session_key) + print(f"[{self.name}] 🖼️ Queuing photo follow-up for session {session_key} without interrupt") existing = self._pending_messages.get(session_key) if existing and existing.message_type == MessageType.PHOTO: existing.media_urls.extend(event.media_urls) @@ -1040,19 +1016,12 @@ async def handle_message(self, event: MessageEvent) -> None: return # Don't interrupt now - will run after current task completes # Default behavior for non-photo follow-ups: interrupt the running agent - logger.debug("[%s] New message while session %s is active — triggering interrupt", self.name, session_key) + print(f"[{self.name}] ⚡ New message while session {session_key} is active - triggering interrupt") self._pending_messages[session_key] = event # Signal the interrupt (the processing task checks this) self._active_sessions[session_key].set() return # Don't process now - will be handled after current task finishes - # Mark session as active BEFORE spawning background task to close - # the race window where a second message arriving before the task - # starts would also pass the _active_sessions check and spawn a - # duplicate task. (grammY sequentialize / aiogram EventIsolation - # pattern — set the guard synchronously, not inside the task.) - self._active_sessions[session_key] = asyncio.Event() - # Spawn background task to process this message task = asyncio.create_task(self._process_message_background(event, session_key)) try: @@ -1087,22 +1056,8 @@ def _get_human_delay() -> float: async def _process_message_background(self, event: MessageEvent, session_key: str) -> None: """Background task that actually processes the message.""" - # Track delivery outcomes for the processing-complete hook - delivery_attempted = False - delivery_succeeded = False - - def _record_delivery(result): - nonlocal delivery_attempted, delivery_succeeded - if result is None: - return - delivery_attempted = True - if getattr(result, "success", False): - delivery_succeeded = True - - # Reuse the interrupt event set by handle_message() (which marks - # the session active before spawning this task to prevent races). - # Fall back to a new Event only if the entry was removed externally. - interrupt_event = self._active_sessions.get(session_key) or asyncio.Event() + # Create interrupt event for this session + interrupt_event = asyncio.Event() self._active_sessions[session_key] = interrupt_event # Start continuous typing indicator (refreshes every 2 seconds) @@ -1110,17 +1065,12 @@ def _record_delivery(result): typing_task = asyncio.create_task(self._keep_typing(event.source.chat_id, metadata=_thread_metadata)) try: - await self._run_processing_hook("on_processing_start", event) - # Call the handler (this can take a while with tool calls) response = await self._message_handler(event) - # Send response if any. A None/empty response is normal when - # streaming already delivered the text (already_sent=True) or - # when the message was queued behind an active agent. Log at - # DEBUG to avoid noisy warnings for expected behavior. + # Send response if any if not response: - logger.debug("[%s] Handler returned empty/None response for %s", self.name, event.source.chat_id) + logger.warning("[%s] Handler returned empty/None response for %s", self.name, event.source.chat_id) if response: # Extract MEDIA: tags (from TTS tool) before other processing media_files, response = self.extract_media(response) @@ -1178,13 +1128,25 @@ def _record_delivery(result): # Send the text portion if text_content: logger.info("[%s] Sending response (%d chars) to %s", self.name, len(text_content), event.source.chat_id) - result = await self._send_with_retry( + result = await self.send( chat_id=event.source.chat_id, content=text_content, reply_to=event.message_id, metadata=_thread_metadata, ) - _record_delivery(result) + + # Log send failures (don't raise - user already saw tool progress) + if not result.success: + print(f"[{self.name}] Failed to send response: {result.error}") + # Try sending without markdown as fallback + fallback_result = await self.send( + chat_id=event.source.chat_id, + content=f"(Response formatting failed, plain text:)\n\n{text_content[:3500]}", + reply_to=event.message_id, + metadata=_thread_metadata, + ) + if not fallback_result.success: + print(f"[{self.name}] Fallback send also failed: {fallback_result.error}") # Human-like pacing delay between text and media human_delay = self._get_human_delay() @@ -1253,9 +1215,9 @@ def _record_delivery(result): ) if not media_result.success: - logger.warning("[%s] Failed to send media (%s): %s", self.name, ext, media_result.error) + print(f"[{self.name}] Failed to send media ({ext}): {media_result.error}") except Exception as media_err: - logger.warning("[%s] Error sending media: %s", self.name, media_err) + print(f"[{self.name}] Error sending media: {media_err}") # Send auto-detected local files as native attachments for file_path in local_files: @@ -1284,14 +1246,10 @@ def _record_delivery(result): except Exception as file_err: logger.error("[%s] Error sending local file %s: %s", self.name, file_path, file_err) - # Determine overall success for the processing hook - processing_ok = delivery_succeeded if delivery_attempted else not bool(response) - await self._run_processing_hook("on_processing_complete", event, processing_ok) - # Check if there's a pending message that was queued during our processing if session_key in self._pending_messages: pending_event = self._pending_messages.pop(session_key) - logger.debug("[%s] Processing queued message from interrupt", self.name) + print(f"[{self.name}] 📨 Processing queued message from interrupt") # Clean up current session before processing pending if session_key in self._active_sessions: del self._active_sessions[session_key] @@ -1304,12 +1262,10 @@ def _record_delivery(result): await self._process_message_background(pending_event, session_key) return # Already cleaned up - except asyncio.CancelledError: - await self._run_processing_hook("on_processing_complete", event, False) - raise except Exception as e: - await self._run_processing_hook("on_processing_complete", event, False) - logger.error("[%s] Error handling message: %s", self.name, e, exc_info=True) + print(f"[{self.name}] Error handling message: {e}") + import traceback + traceback.print_exc() # Send the error to the user so they aren't left with radio silence try: error_type = type(e).__name__ diff --git a/gateway/platforms/discord.py b/gateway/platforms/discord.py index 91e6710d26b4..0de72bfbc314 100644 --- a/gateway/platforms/discord.py +++ b/gateway/platforms/discord.py @@ -20,7 +20,7 @@ import time from collections import defaultdict from pathlib import Path -from typing import Callable, Dict, Optional, Any +from typing import Callable, Dict, List, Optional, Any logger = logging.getLogger(__name__) @@ -53,6 +53,7 @@ cache_image_from_url, cache_audio_from_url, cache_document_from_bytes, + build_text_attachment_injection, SUPPORTED_DOCUMENT_TYPES, ) @@ -408,7 +409,7 @@ def pcm_to_wav(pcm_data: bytes, output_path: str, class DiscordAdapter(BasePlatformAdapter): """ Discord bot adapter. - + Handles: - Receiving messages from servers and DMs - Sending responses with Discord markdown @@ -418,10 +419,10 @@ class DiscordAdapter(BasePlatformAdapter): - Auto-threading for long conversations - Reaction-based feedback """ - + # Discord message limits MAX_MESSAGE_LENGTH = 2000 - + # Auto-disconnect from voice channel after this many seconds of inactivity VOICE_TIMEOUT = 300 @@ -446,10 +447,9 @@ def __init__(self, config: PlatformConfig): # Persistent typing indicator loops per channel (DMs don't reliably # show the standard typing gateway event for bots) self._typing_tasks: Dict[str, asyncio.Task] = {} - self._bot_task: Optional[asyncio.Task] = None # Cap to prevent unbounded growth (Discord threads get archived). self._MAX_TRACKED_THREADS = 500 - + async def connect(self) -> bool: """Connect to Discord and start receiving events.""" if not DISCORD_AVAILABLE: @@ -480,23 +480,12 @@ async def connect(self) -> bool: logger.warning("Opus codec found at %s but failed to load", opus_path) if not discord.opus.is_loaded(): logger.warning("Opus codec not found — voice channel playback disabled") - + if not self.config.token: logger.error("[%s] No bot token configured", self.name) return False - + try: - # Acquire scoped lock to prevent duplicate bot token usage - from gateway.status import acquire_scoped_lock - self._token_lock_identity = self.config.token - acquired, existing = acquire_scoped_lock('discord-bot-token', self._token_lock_identity, metadata={'platform': 'discord'}) - if not acquired: - owner_pid = existing.get('pid') if isinstance(existing, dict) else None - message = f'Discord bot token already in use' + (f' (PID {owner_pid})' if owner_pid else '') + '. Stop the other gateway first.' - logger.error('[%s] %s', self.name, message) - self._set_fatal_error('discord_token_lock', message, retryable=False) - return False - # Set up intents -- members intent needed for username-to-ID resolution intents = Intents.default() intents.message_content = True @@ -504,13 +493,13 @@ async def connect(self) -> bool: intents.guild_messages = True intents.members = True intents.voice_states = True - + # Create bot self._client = commands.Bot( command_prefix="!", # Not really used, we handle raw messages intents=intents, ) - + # Parse allowed user entries (may contain usernames or IDs) allowed_env = os.getenv("DISCORD_ALLOWED_USERS", "") if allowed_env: @@ -518,17 +507,17 @@ async def connect(self) -> bool: _clean_discord_id(uid) for uid in allowed_env.split(",") if uid.strip() } - + adapter_self = self # capture for closure - + # Register event handlers @self._client.event async def on_ready(): logger.info("[%s] Connected as %s", adapter_self.name, adapter_self._client.user) - + # Resolve any usernames in the allowed list to numeric IDs await adapter_self._resolve_allowed_usernames() - + # Sync slash commands with Discord try: synced = await adapter_self._client.tree.sync() @@ -536,22 +525,18 @@ async def on_ready(): except Exception as e: # pragma: no cover - defensive logging logger.warning("[%s] Slash command sync failed: %s", adapter_self.name, e, exc_info=True) adapter_self._ready_event.set() - + @self._client.event async def on_message(message: DiscordMessage): # Always ignore our own messages if message.author == self._client.user: return - + # Ignore Discord system messages (thread renames, pins, member joins, etc.) # Allow both default and reply types — replies have a distinct MessageType. if message.type not in (discord.MessageType.default, discord.MessageType.reply): return - - # Check if the message author is in the allowed user list - if not self._is_allowed_user(str(message.author.id)): - return - + # Bot message filtering (DISCORD_ALLOW_BOTS): # "none" — ignore all other bots (default) # "mentions" — accept bot messages only when they @mention us @@ -564,23 +549,7 @@ async def on_message(message: DiscordMessage): if not self._client.user or self._client.user not in message.mentions: return # "all" falls through to handle_message - - # If the message @mentions other users but NOT the bot, the - # sender is talking to someone else — stay silent. Only - # applies in server channels; in DMs the user is always - # talking to the bot (mentions are just references). - # Controlled by DISCORD_IGNORE_NO_MENTION (default: true). - _ignore_no_mention = os.getenv( - "DISCORD_IGNORE_NO_MENTION", "true" - ).lower() in ("true", "1", "yes") - if _ignore_no_mention and message.mentions and not isinstance(message.channel, discord.DMChannel): - _bot_mentioned = ( - self._client.user is not None - and self._client.user in message.mentions - ) - if not _bot_mentioned: - return # Talking to someone else, don't interrupt - + await self._handle_message(message) @self._client.event @@ -618,23 +587,23 @@ async def on_voice_state_update(member, before, after): # Register slash commands self._register_slash_commands() - + # Start the bot in background - self._bot_task = asyncio.create_task(self._client.start(self.config.token)) - + asyncio.create_task(self._client.start(self.config.token)) + # Wait for ready await asyncio.wait_for(self._ready_event.wait(), timeout=30) - + self._running = True return True - + except asyncio.TimeoutError: logger.error("[%s] Timeout waiting for connection to Discord", self.name, exc_info=True) return False except Exception as e: # pragma: no cover - defensive logging logger.error("[%s] Failed to connect to Discord: %s", self.name, e, exc_info=True) return False - + async def disconnect(self) -> None: """Disconnect from Discord.""" # Clean up all active voice connections before closing the client @@ -653,61 +622,8 @@ async def disconnect(self) -> None: self._running = False self._client = None self._ready_event.clear() - - # Release the token lock - try: - from gateway.status import release_scoped_lock - if getattr(self, '_token_lock_identity', None): - release_scoped_lock('discord-bot-token', self._token_lock_identity) - self._token_lock_identity = None - except Exception: - pass - logger.info("[%s] Disconnected", self.name) - - async def _add_reaction(self, message: Any, emoji: str) -> bool: - """Add an emoji reaction to a Discord message.""" - if not message or not hasattr(message, "add_reaction"): - return False - try: - await message.add_reaction(emoji) - return True - except Exception as e: - logger.debug("[%s] add_reaction failed (%s): %s", self.name, emoji, e) - return False - - async def _remove_reaction(self, message: Any, emoji: str) -> bool: - """Remove the bot's own emoji reaction from a Discord message.""" - if not message or not hasattr(message, "remove_reaction") or not self._client or not self._client.user: - return False - try: - await message.remove_reaction(emoji, self._client.user) - return True - except Exception as e: - logger.debug("[%s] remove_reaction failed (%s): %s", self.name, emoji, e) - return False - - def _reactions_enabled(self) -> bool: - """Check if message reactions are enabled via config/env.""" - return os.getenv("DISCORD_REACTIONS", "true").lower() not in ("false", "0", "no") - - async def on_processing_start(self, event: MessageEvent) -> None: - """Add an in-progress reaction for normal Discord message events.""" - if not self._reactions_enabled(): - return - message = event.raw_message - if hasattr(message, "add_reaction"): - await self._add_reaction(message, "👀") - - async def on_processing_complete(self, event: MessageEvent, success: bool) -> None: - """Swap the in-progress reaction for a final success/failure reaction.""" - if not self._reactions_enabled(): - return - message = event.raw_message - if hasattr(message, "add_reaction"): - await self._remove_reaction(message, "👀") - await self._add_reaction(message, "✅" if success else "❌") - + async def send( self, chat_id: str, @@ -724,24 +640,24 @@ async def send( channel = self._client.get_channel(int(chat_id)) if not channel: channel = await self._client.fetch_channel(int(chat_id)) - + if not channel: return SendResult(success=False, error=f"Channel {chat_id} not found") - + # Format and split message if needed formatted = self.format_message(content) chunks = self.truncate_message(formatted, self.MAX_MESSAGE_LENGTH) - + message_ids = [] reference = None - + if reply_to: try: ref_msg = await channel.fetch_message(int(reply_to)) reference = ref_msg except Exception as e: logger.debug("Could not fetch reply-to message: %s", e) - + for i, chunk in enumerate(chunks): chunk_reference = reference if i == 0 else None try: @@ -768,13 +684,13 @@ async def send( else: raise message_ids.append(str(msg.id)) - + return SendResult( success=True, message_id=message_ids[0] if message_ids else None, raw_response={"message_ids": message_ids} ) - + except Exception as e: # pragma: no cover - defensive logging logger.error("[%s] Failed to send Discord message: %s", self.name, e, exc_info=True) return SendResult(success=False, error=str(e)) @@ -1246,25 +1162,25 @@ async def send_image( """Send an image natively as a Discord file attachment.""" if not self._client: return SendResult(success=False, error="Not connected") - + try: import aiohttp - + channel = self._client.get_channel(int(chat_id)) if not channel: channel = await self._client.fetch_channel(int(chat_id)) if not channel: return SendResult(success=False, error=f"Channel {chat_id} not found") - + # Download the image and send as a Discord file attachment # (Discord renders attachments inline, unlike plain URLs) async with aiohttp.ClientSession() as session: async with session.get(image_url, timeout=aiohttp.ClientTimeout(total=30)) as resp: if resp.status != 200: raise Exception(f"Failed to download image: HTTP {resp.status}") - + image_data = await resp.read() - + # Determine filename from URL or content type content_type = resp.headers.get("content-type", "image/png") ext = "png" @@ -1274,16 +1190,16 @@ async def send_image( ext = "gif" elif "webp" in content_type: ext = "webp" - + import io file = discord.File(io.BytesIO(image_data), filename=f"image.{ext}") - + msg = await channel.send( content=caption if caption else None, file=file, ) return SendResult(success=True, message_id=str(msg.id)) - + except ImportError: logger.warning( "[%s] aiohttp not installed, falling back to URL. Run: pip install aiohttp", @@ -1334,7 +1250,7 @@ async def send_document( except Exception as e: # pragma: no cover - defensive logging logger.error("[%s] Failed to send document, falling back to base adapter: %s", self.name, e, exc_info=True) return await super().send_document(chat_id, file_path, caption, file_name, reply_to, metadata=metadata) - + async def send_typing(self, chat_id: str, metadata=None) -> None: """Start a persistent typing indicator for a channel. @@ -1378,20 +1294,20 @@ async def stop_typing(self, chat_id: str) -> None: await task except (asyncio.CancelledError, Exception): pass - + async def get_chat_info(self, chat_id: str) -> Dict[str, Any]: """Get information about a Discord channel.""" if not self._client: return {"name": "Unknown", "type": "dm"} - + try: channel = self._client.get_channel(int(chat_id)) if not channel: channel = await self._client.fetch_channel(int(chat_id)) - + if not channel: return {"name": str(chat_id), "type": "dm"} - + # Determine channel type if isinstance(channel, discord.DMChannel): chat_type = "dm" @@ -1407,7 +1323,7 @@ async def get_chat_info(self, chat_id: str) -> Dict[str, Any]: else: chat_type = "channel" name = getattr(channel, "name", str(chat_id)) - + return { "name": name, "type": chat_type, @@ -1417,7 +1333,7 @@ async def get_chat_info(self, chat_id: str) -> Dict[str, Any]: except Exception as e: # pragma: no cover - defensive logging logger.error("[%s] Failed to get chat info for %s: %s", self.name, chat_id, e, exc_info=True) return {"name": str(chat_id), "type": "dm", "error": str(e)} - + async def _resolve_allowed_usernames(self) -> None: """ Resolve non-numeric entries in DISCORD_ALLOWED_USERS to Discord user IDs. @@ -1485,7 +1401,7 @@ async def _resolve_allowed_usernames(self) -> None: def format_message(self, content: str) -> str: """ Format message for Discord. - + Discord uses its own markdown variant. """ # Discord markdown is fairly standard, no special escaping needed @@ -1497,23 +1413,15 @@ async def _run_simple_slash( command_text: str, followup_msg: str | None = None, ) -> None: - """Common handler for simple slash commands that dispatch a command string. - - Defers the interaction (shows "thinking..."), dispatches the command, - then cleans up the deferred response. If *followup_msg* is provided - the "thinking..." indicator is replaced with that text; otherwise it - is deleted so the channel isn't cluttered. - """ + """Common handler for simple slash commands that dispatch a command string.""" await interaction.response.defer(ephemeral=True) event = self._build_slash_event(interaction, command_text) await self.handle_message(event) - try: - if followup_msg: - await interaction.edit_original_response(content=followup_msg) - else: - await interaction.delete_original_response() - except Exception as e: - logger.debug("Discord interaction cleanup failed: %s", e) + if followup_msg: + try: + await interaction.followup.send(followup_msg, ephemeral=True) + except Exception as e: + logger.debug("Discord followup failed: %s", e) def _register_slash_commands(self) -> None: """Register Discord slash commands on the command tree.""" @@ -1538,7 +1446,9 @@ async def slash_model(interaction: discord.Interaction, name: str = ""): @tree.command(name="reasoning", description="Show or change reasoning effort") @discord.app_commands.describe(effort="Reasoning effort: xhigh, high, medium, low, minimal, or none.") async def slash_reasoning(interaction: discord.Interaction, effort: str = ""): - await self._run_simple_slash(interaction, f"/reasoning {effort}".strip()) + await interaction.response.defer(ephemeral=True) + event = self._build_slash_event(interaction, f"/reasoning {effort}".strip()) + await self.handle_message(event) @tree.command(name="personality", description="Set a personality") @discord.app_commands.describe(name="Personality name. Leave empty to list available.") @@ -1611,22 +1521,14 @@ async def slash_reload_mcp(interaction: discord.Interaction): discord.app_commands.Choice(name="status — show current mode", value="status"), ]) async def slash_voice(interaction: discord.Interaction, mode: str = ""): - await self._run_simple_slash(interaction, f"/voice {mode}".strip()) + await interaction.response.defer(ephemeral=True) + event = self._build_slash_event(interaction, f"/voice {mode}".strip()) + await self.handle_message(event) @tree.command(name="update", description="Update Hermes Agent to the latest version") async def slash_update(interaction: discord.Interaction): await self._run_simple_slash(interaction, "/update", "Update initiated~") - @tree.command(name="approve", description="Approve a pending dangerous command") - @discord.app_commands.describe(scope="Optional: 'all', 'session', 'always', 'all session', 'all always'") - async def slash_approve(interaction: discord.Interaction, scope: str = ""): - await self._run_simple_slash(interaction, f"/approve {scope}".strip()) - - @tree.command(name="deny", description="Deny a pending dangerous command") - @discord.app_commands.describe(scope="Optional: 'all' to deny all pending commands") - async def slash_deny(interaction: discord.Interaction, scope: str = ""): - await self._run_simple_slash(interaction, f"/deny {scope}".strip()) - @tree.command(name="thread", description="Create a new thread and start a Hermes session in it") @discord.app_commands.describe( name="Thread name", @@ -1661,7 +1563,7 @@ def _build_slash_event(self, interaction: discord.Interaction, text: str) -> Mes chat_name = interaction.channel.name if hasattr(interaction.channel, "guild") and interaction.channel.guild: chat_name = f"{interaction.channel.guild.name} / #{chat_name}" - + # Get channel topic (if available) chat_topic = getattr(interaction.channel, "topic", None) @@ -1870,41 +1772,33 @@ async def _auto_create_thread(self, message: 'DiscordMessage') -> Optional[Any]: return None async def send_exec_approval( - self, chat_id: str, command: str, session_key: str, - description: str = "dangerous command", - metadata: Optional[dict] = None, + self, chat_id: str, command: str, approval_id: str ) -> SendResult: """ Send a button-based exec approval prompt for a dangerous command. - The buttons call ``resolve_gateway_approval()`` to unblock the waiting - agent thread — this replaces the text-based ``/approve`` flow on Discord. + Returns SendResult. The approval is resolved when a user clicks a button. """ if not self._client or not DISCORD_AVAILABLE: return SendResult(success=False, error="Not connected") try: - # Resolve channel — use thread_id from metadata if present - target_id = chat_id - if metadata and metadata.get("thread_id"): - target_id = metadata["thread_id"] - - channel = self._client.get_channel(int(target_id)) + channel = self._client.get_channel(int(chat_id)) if not channel: - channel = await self._client.fetch_channel(int(target_id)) + channel = await self._client.fetch_channel(int(chat_id)) # Discord embed description limit is 4096; show full command up to that max_desc = 4088 cmd_display = command if len(command) <= max_desc else command[: max_desc - 3] + "..." embed = discord.Embed( - title="⚠️ Command Approval Required", + title="Command Approval Required", description=f"```\n{cmd_display}\n```", color=discord.Color.orange(), ) - embed.add_field(name="Reason", value=description, inline=False) + embed.set_footer(text=f"Approval ID: {approval_id}") view = ExecApprovalView( - session_key=session_key, + approval_id=approval_id, allowed_user_ids=self._allowed_user_ids, ) @@ -2073,7 +1967,7 @@ async def _handle_message(self, message: DiscordMessage) -> None: if doc_ext in SUPPORTED_DOCUMENT_TYPES: msg_type = MessageType.DOCUMENT break - + # When auto-threading kicked in, route responses to the new thread effective_channel = auto_threaded_channel or message.channel @@ -2092,7 +1986,7 @@ async def _handle_message(self, message: DiscordMessage) -> None: # Get channel topic (if available - TextChannels have topics, DMs/threads don't) chat_topic = getattr(message.channel, "topic", None) - + # Build source source = self.build_source( chat_id=str(effective_channel.id), @@ -2103,7 +1997,7 @@ async def _handle_message(self, message: DiscordMessage) -> None: thread_id=thread_id, chat_topic=chat_topic, ) - + # Build media URLs -- download image attachments to local cache so the # vision tool can access them reliably (Discord CDN URLs can expire). media_urls = [] @@ -2178,35 +2072,26 @@ async def _handle_message(self, message: DiscordMessage) -> None: media_urls.append(cached_path) media_types.append(doc_mime) logger.info("[Discord] Cached user document: %s", cached_path) - # Inject text content for .txt/.md files (capped at 100 KB) - MAX_TEXT_INJECT_BYTES = 100 * 1024 - if ext in (".md", ".txt") and len(raw_bytes) <= MAX_TEXT_INJECT_BYTES: - try: - text_content = raw_bytes.decode("utf-8") - display_name = att.filename or f"document{ext}" - display_name = re.sub(r'[^\w.\- ]', '_', display_name) - injection = f"[Content of {display_name}]:\n{text_content}" - if pending_text_injection: - pending_text_injection = f"{pending_text_injection}\n\n{injection}" - else: - pending_text_injection = injection - except UnicodeDecodeError: - pass + injection = build_text_attachment_injection( + raw_bytes, + att.filename or f"document{ext}", + ext, + ) + if injection: + if pending_text_injection: + pending_text_injection = f"{pending_text_injection}\n\n{injection}" + else: + pending_text_injection = injection except Exception as e: logger.warning( "[Discord] Failed to cache document %s: %s", att.filename, e, exc_info=True, ) - + event_text = message.content if pending_text_injection: event_text = f"{pending_text_injection}\n\n{event_text}" if event_text else pending_text_injection - # Defense-in-depth: prevent empty user messages from entering session - # (can happen when user sends @mention-only with no other text) - if not event_text or not event_text.strip(): - event_text = "(The user sent a message with no text content)" - event = MessageEvent( text=event_text, message_type=msg_type, @@ -2237,15 +2122,13 @@ class ExecApprovalView(discord.ui.View): """ Interactive button view for exec approval of dangerous commands. - Shows four buttons: Allow Once, Allow Session, Always Allow, Deny. - Clicking a button calls ``resolve_gateway_approval()`` to unblock the - waiting agent thread — the same mechanism as the text ``/approve`` flow. - Only users in the allowed list can click. Times out after 5 minutes. + Shows three buttons: Allow Once (green), Always Allow (blue), Deny (red). + Only users in the allowed list can click. The view times out after 5 minutes. """ - def __init__(self, session_key: str, allowed_user_ids: set): + def __init__(self, approval_id: str, allowed_user_ids: set): super().__init__(timeout=300) # 5-minute timeout - self.session_key = session_key + self.approval_id = approval_id self.allowed_user_ids = allowed_user_ids self.resolved = False @@ -2256,10 +2139,9 @@ def _check_auth(self, interaction: discord.Interaction) -> bool: return str(interaction.user.id) in self.allowed_user_ids async def _resolve( - self, interaction: discord.Interaction, choice: str, - color: discord.Color, label: str, + self, interaction: discord.Interaction, action: str, color: discord.Color ): - """Resolve the approval via the gateway approval queue and update the embed.""" + """Resolve the approval and update the message.""" if self.resolved: await interaction.response.send_message( "This approval has already been resolved~", ephemeral=True @@ -2278,7 +2160,7 @@ async def _resolve( embed = interaction.message.embeds[0] if interaction.message.embeds else None if embed: embed.color = color - embed.set_footer(text=f"{label} by {interaction.user.display_name}") + embed.set_footer(text=f"{action} by {interaction.user.display_name}") # Disable all buttons for child in self.children: @@ -2286,40 +2168,33 @@ async def _resolve( await interaction.response.edit_message(embed=embed, view=self) - # Unblock the waiting agent thread via the gateway approval queue + # Store the approval decision try: - from tools.approval import resolve_gateway_approval - count = resolve_gateway_approval(self.session_key, choice) - logger.info( - "Discord button resolved %d approval(s) for session %s (choice=%s, user=%s)", - count, self.session_key, choice, interaction.user.display_name, - ) - except Exception as exc: - logger.error("Failed to resolve gateway approval from button: %s", exc) + from tools.approval import approve_permanent + if action == "allow_once": + pass # One-time approval handled by gateway + elif action == "allow_always": + approve_permanent(self.approval_id) + except ImportError: + pass @discord.ui.button(label="Allow Once", style=discord.ButtonStyle.green) async def allow_once( self, interaction: discord.Interaction, button: discord.ui.Button ): - await self._resolve(interaction, "once", discord.Color.green(), "Approved once") - - @discord.ui.button(label="Allow Session", style=discord.ButtonStyle.grey) - async def allow_session( - self, interaction: discord.Interaction, button: discord.ui.Button - ): - await self._resolve(interaction, "session", discord.Color.blue(), "Approved for session") + await self._resolve(interaction, "allow_once", discord.Color.green()) @discord.ui.button(label="Always Allow", style=discord.ButtonStyle.blurple) async def allow_always( self, interaction: discord.Interaction, button: discord.ui.Button ): - await self._resolve(interaction, "always", discord.Color.purple(), "Approved permanently") + await self._resolve(interaction, "allow_always", discord.Color.blue()) @discord.ui.button(label="Deny", style=discord.ButtonStyle.red) async def deny( self, interaction: discord.Interaction, button: discord.ui.Button ): - await self._resolve(interaction, "deny", discord.Color.red(), "Denied") + await self._resolve(interaction, "deny", discord.Color.red()) async def on_timeout(self): """Handle view timeout -- disable buttons and mark as expired.""" diff --git a/gateway/platforms/slack.py b/gateway/platforms/slack.py index be11803504d1..2febec28fb84 100644 --- a/gateway/platforms/slack.py +++ b/gateway/platforms/slack.py @@ -9,11 +9,10 @@ """ import asyncio -import json import logging import os import re -from typing import Dict, Optional, Any +from typing import Dict, List, Optional, Any try: from slack_bolt.async_app import AsyncApp @@ -37,7 +36,10 @@ MessageType, SendResult, SUPPORTED_DOCUMENT_TYPES, + build_text_attachment_injection, cache_document_from_bytes, + cache_image_from_url, + cache_audio_from_url, ) @@ -73,11 +75,6 @@ def __init__(self, config: PlatformConfig): self._handler: Optional[AsyncSocketModeHandler] = None self._bot_user_id: Optional[str] = None self._user_name_cache: Dict[str, str] = {} # user_id → display name - self._socket_mode_task: Optional[asyncio.Task] = None - # Multi-workspace support - self._team_clients: Dict[str, AsyncWebClient] = {} # team_id → WebClient - self._team_bot_user_ids: Dict[str, str] = {} # team_id → bot_user_id - self._channel_team: Dict[str, str] = {} # channel_id → team_id async def connect(self) -> bool: """Connect to Slack via Socket Mode.""" @@ -87,70 +84,23 @@ async def connect(self) -> bool: ) return False - raw_token = self.config.token + bot_token = self.config.token app_token = os.getenv("SLACK_APP_TOKEN") - if not raw_token: + if not bot_token: logger.error("[Slack] SLACK_BOT_TOKEN not set") return False if not app_token: logger.error("[Slack] SLACK_APP_TOKEN not set") return False - # Support comma-separated bot tokens for multi-workspace - bot_tokens = [t.strip() for t in raw_token.split(",") if t.strip()] - - # Also load tokens from OAuth token file - from hermes_constants import get_hermes_home - tokens_file = get_hermes_home() / "slack_tokens.json" - if tokens_file.exists(): - try: - saved = json.loads(tokens_file.read_text(encoding="utf-8")) - for team_id, entry in saved.items(): - tok = entry.get("token", "") if isinstance(entry, dict) else "" - if tok and tok not in bot_tokens: - bot_tokens.append(tok) - team_label = entry.get("team_name", team_id) if isinstance(entry, dict) else team_id - logger.info("[Slack] Loaded saved token for workspace %s", team_label) - except Exception as e: - logger.warning("[Slack] Failed to read %s: %s", tokens_file, e) - try: - # Acquire scoped lock to prevent duplicate app token usage - from gateway.status import acquire_scoped_lock - self._token_lock_identity = app_token - acquired, existing = acquire_scoped_lock('slack-app-token', app_token, metadata={'platform': 'slack'}) - if not acquired: - owner_pid = existing.get('pid') if isinstance(existing, dict) else None - message = f'Slack app token already in use' + (f' (PID {owner_pid})' if owner_pid else '') + '. Stop the other gateway first.' - logger.error('[%s] %s', self.name, message) - self._set_fatal_error('slack_token_lock', message, retryable=False) - return False - - # First token is the primary — used for AsyncApp / Socket Mode - primary_token = bot_tokens[0] - self._app = AsyncApp(token=primary_token) - - # Register each bot token and map team_id → client - for token in bot_tokens: - client = AsyncWebClient(token=token) - auth_response = await client.auth_test() - team_id = auth_response.get("team_id", "") - bot_user_id = auth_response.get("user_id", "") - bot_name = auth_response.get("user", "unknown") - team_name = auth_response.get("team", "unknown") - - self._team_clients[team_id] = client - self._team_bot_user_ids[team_id] = bot_user_id - - # First token sets the primary bot_user_id (backward compat) - if self._bot_user_id is None: - self._bot_user_id = bot_user_id - - logger.info( - "[Slack] Authenticated as @%s in workspace %s (team: %s)", - bot_name, team_name, team_id, - ) + self._app = AsyncApp(token=bot_token) + + # Get our own bot user ID for mention detection + auth_response = await self._app.client.auth_test() + self._bot_user_id = auth_response.get("user_id") + bot_name = auth_response.get("user", "unknown") # Register message event handler @self._app.event("message") @@ -172,13 +122,10 @@ async def handle_hermes_command(ack, command): # Start Socket Mode handler in background self._handler = AsyncSocketModeHandler(self._app, app_token) - self._socket_mode_task = asyncio.create_task(self._handler.start_async()) + asyncio.create_task(self._handler.start_async()) self._running = True - logger.info( - "[Slack] Socket Mode connected (%d workspace(s))", - len(self._team_clients), - ) + logger.info("[Slack] Connected as @%s (Socket Mode)", bot_name) return True except Exception as e: # pragma: no cover - defensive logging @@ -193,25 +140,8 @@ async def disconnect(self) -> None: except Exception as e: # pragma: no cover - defensive logging logger.warning("[Slack] Error while closing Socket Mode handler: %s", e, exc_info=True) self._running = False - - # Release the token lock (use stored identity, not re-read env) - try: - from gateway.status import release_scoped_lock - if getattr(self, '_token_lock_identity', None): - release_scoped_lock('slack-app-token', self._token_lock_identity) - self._token_lock_identity = None - except Exception: - pass - logger.info("[Slack] Disconnected") - def _get_client(self, chat_id: str) -> AsyncWebClient: - """Return the workspace-specific WebClient for a channel.""" - team_id = self._channel_team.get(chat_id) - if team_id and team_id in self._team_clients: - return self._team_clients[team_id] - return self._app.client # fallback to primary - async def send( self, chat_id: str, @@ -248,7 +178,7 @@ async def send( if broadcast and i == 0: kwargs["reply_broadcast"] = True - last_result = await self._get_client(chat_id).chat_postMessage(**kwargs) + last_result = await self._app.client.chat_postMessage(**kwargs) return SendResult( success=True, @@ -270,7 +200,7 @@ async def edit_message( if not self._app: return SendResult(success=False, error="Not connected") try: - await self._get_client(chat_id).chat_update( + await self._app.client.chat_update( channel=chat_id, ts=message_id, text=content, @@ -304,7 +234,7 @@ async def send_typing(self, chat_id: str, metadata=None) -> None: return # Can only set status in a thread context try: - await self._get_client(chat_id).assistant_threads_setStatus( + await self._app.client.assistant_threads_setStatus( channel_id=chat_id, thread_ts=thread_ts, status="is thinking...", @@ -323,18 +253,7 @@ def _resolve_thread_ts( Prefers metadata thread_id (the thread parent's ts, set by the gateway) over reply_to (which may be a child message's ts). - - When ``reply_in_thread`` is ``false`` in the platform extra config, - top-level channel messages receive direct channel replies instead of - thread replies. Messages that originate inside an existing thread are - always replied to in-thread to preserve conversation context. """ - # When reply_in_thread is disabled (default: True for backward compat), - # only thread messages that are already part of an existing thread. - if not self.config.extra.get("reply_in_thread", True): - existing_thread = (metadata or {}).get("thread_id") or (metadata or {}).get("thread_ts") - return existing_thread or None - if metadata: if metadata.get("thread_id"): return metadata["thread_id"] @@ -357,7 +276,7 @@ async def _upload_file( if not os.path.exists(file_path): raise FileNotFoundError(f"File not found: {file_path}") - result = await self._get_client(chat_id).files_upload_v2( + result = await self._app.client.files_upload_v2( channel=chat_id, file=file_path, filename=os.path.basename(file_path), @@ -459,7 +378,7 @@ async def _add_reaction( if not self._app: return False try: - await self._get_client(channel).reactions_add( + await self._app.client.reactions_add( channel=channel, timestamp=timestamp, name=emoji ) return True @@ -475,7 +394,7 @@ async def _remove_reaction( if not self._app: return False try: - await self._get_client(channel).reactions_remove( + await self._app.client.reactions_remove( channel=channel, timestamp=timestamp, name=emoji ) return True @@ -485,7 +404,7 @@ async def _remove_reaction( # ----- User identity resolution ----- - async def _resolve_user_name(self, user_id: str, chat_id: str = "") -> str: + async def _resolve_user_name(self, user_id: str) -> str: """Resolve a Slack user ID to a display name, with caching.""" if not user_id: return "" @@ -496,8 +415,7 @@ async def _resolve_user_name(self, user_id: str, chat_id: str = "") -> str: return user_id try: - client = self._get_client(chat_id) if chat_id else self._app.client - result = await client.users_info(user=user_id) + result = await self._app.client.users_info(user=user_id) user = result.get("user", {}) # Prefer display_name → real_name → user_id profile = user.get("profile", {}) @@ -561,7 +479,7 @@ async def send_image( response = await client.get(image_url) response.raise_for_status() - result = await self._get_client(chat_id).files_upload_v2( + result = await self._app.client.files_upload_v2( channel=chat_id, content=response.content, filename="image.png", @@ -621,7 +539,7 @@ async def send_video( return SendResult(success=False, error=f"Video file not found: {video_path}") try: - result = await self._get_client(chat_id).files_upload_v2( + result = await self._app.client.files_upload_v2( channel=chat_id, file=video_path, filename=os.path.basename(video_path), @@ -662,7 +580,7 @@ async def send_document( display_name = file_name or os.path.basename(file_path) try: - result = await self._get_client(chat_id).files_upload_v2( + result = await self._app.client.files_upload_v2( channel=chat_id, file=file_path, filename=display_name, @@ -690,7 +608,7 @@ async def get_chat_info(self, chat_id: str) -> Dict[str, Any]: return {"name": chat_id, "type": "unknown"} try: - result = await self._get_client(chat_id).conversations_info(channel=chat_id) + result = await self._app.client.conversations_info(channel=chat_id) channel = result.get("channel", {}) is_dm = channel.get("is_im", False) return { @@ -723,11 +641,6 @@ async def _handle_slack_message(self, event: dict) -> None: user_id = event.get("user", "") channel_id = event.get("channel", "") ts = event.get("ts", "") - team_id = event.get("team", "") - - # Track which workspace owns this channel - if team_id and channel_id: - self._channel_team[channel_id] = team_id # Determine if this is a DM or channel message channel_type = event.get("channel_type", "") @@ -744,12 +657,11 @@ async def _handle_slack_message(self, event: dict) -> None: thread_ts = event.get("thread_ts") or ts # ts fallback for channels # In channels, only respond if bot is mentioned - bot_uid = self._team_bot_user_ids.get(team_id, self._bot_user_id) - if not is_dm and bot_uid: - if f"<@{bot_uid}>" not in text: + if not is_dm and self._bot_user_id: + if f"<@{self._bot_user_id}>" not in text: return # Strip the bot mention from the text - text = text.replace(f"<@{bot_uid}>", "").strip() + text = text.replace(f"<@{self._bot_user_id}>", "").strip() # Determine message type msg_type = MessageType.TEXT @@ -769,7 +681,7 @@ async def _handle_slack_message(self, event: dict) -> None: if ext not in (".jpg", ".jpeg", ".png", ".gif", ".webp"): ext = ".jpg" # Slack private URLs require the bot token as auth header - cached = await self._download_slack_file(url, ext, team_id=team_id) + cached = await self._download_slack_file(url, ext) media_urls.append(cached) media_types.append(mimetype) msg_type = MessageType.PHOTO @@ -780,7 +692,7 @@ async def _handle_slack_message(self, event: dict) -> None: ext = "." + mimetype.split("/")[-1].split(";")[0] if ext not in (".ogg", ".mp3", ".wav", ".webm", ".m4a"): ext = ".ogg" - cached = await self._download_slack_file(url, ext, audio=True, team_id=team_id) + cached = await self._download_slack_file(url, ext, audio=True) media_urls.append(cached) media_types.append(mimetype) msg_type = MessageType.VOICE @@ -811,7 +723,7 @@ async def _handle_slack_message(self, event: dict) -> None: continue # Download and cache - raw_bytes = await self._download_slack_file_bytes(url, team_id=team_id) + raw_bytes = await self._download_slack_file_bytes(url) cached_path = cache_document_from_bytes( raw_bytes, original_filename or f"document{ext}" ) @@ -821,26 +733,22 @@ async def _handle_slack_message(self, event: dict) -> None: msg_type = MessageType.DOCUMENT logger.debug("[Slack] Cached user document: %s", cached_path) - # Inject text content for .txt/.md files (capped at 100 KB) - MAX_TEXT_INJECT_BYTES = 100 * 1024 - if ext in (".md", ".txt") and len(raw_bytes) <= MAX_TEXT_INJECT_BYTES: - try: - text_content = raw_bytes.decode("utf-8") - display_name = original_filename or f"document{ext}" - display_name = re.sub(r'[^\w.\- ]', '_', display_name) - injection = f"[Content of {display_name}]:\n{text_content}" - if text: - text = f"{injection}\n\n{text}" - else: - text = injection - except UnicodeDecodeError: - pass # Binary content, skip injection + injection = build_text_attachment_injection( + raw_bytes, + original_filename or f"document{ext}", + ext, + ) + if injection: + if text: + text = f"{injection}\n\n{text}" + else: + text = injection except Exception as e: # pragma: no cover - defensive logging logger.warning("[Slack] Failed to cache document from %s: %s", url, e, exc_info=True) # Resolve user display name (cached after first lookup) - user_name = await self._resolve_user_name(user_id, chat_id=channel_id) + user_name = await self._resolve_user_name(user_id) # Build source source = self.build_source( @@ -877,11 +785,6 @@ async def _handle_slash_command(self, command: dict) -> None: text = command.get("text", "").strip() user_id = command.get("user_id", "") channel_id = command.get("channel_id", "") - team_id = command.get("team_id", "") - - # Track which workspace owns this channel - if team_id and channel_id: - self._channel_team[channel_id] = team_id # Map subcommands to gateway commands — derived from central registry. # Also keep "compact" as a Slack-specific alias for /compress. @@ -913,66 +816,34 @@ async def _handle_slash_command(self, command: dict) -> None: await self.handle_message(event) - async def _download_slack_file(self, url: str, ext: str, audio: bool = False, team_id: str = "") -> str: - """Download a Slack file using the bot token for auth, with retry.""" - import asyncio + async def _download_slack_file(self, url: str, ext: str, audio: bool = False) -> str: + """Download a Slack file using the bot token for auth.""" import httpx - bot_token = self._team_clients[team_id].token if team_id and team_id in self._team_clients else self.config.token - last_exc = None - + bot_token = self.config.token async with httpx.AsyncClient(timeout=30.0, follow_redirects=True) as client: - for attempt in range(3): - try: - response = await client.get( - url, - headers={"Authorization": f"Bearer {bot_token}"}, - ) - response.raise_for_status() - - if audio: - from gateway.platforms.base import cache_audio_from_bytes - return cache_audio_from_bytes(response.content, ext) - else: - from gateway.platforms.base import cache_image_from_bytes - return cache_image_from_bytes(response.content, ext) - except (httpx.TimeoutException, httpx.HTTPStatusError) as exc: - last_exc = exc - if isinstance(exc, httpx.HTTPStatusError) and exc.response.status_code < 429: - raise - if attempt < 2: - logger.debug("Slack file download retry %d/2 for %s: %s", - attempt + 1, url[:80], exc) - await asyncio.sleep(1.5 * (attempt + 1)) - continue - raise - raise last_exc + response = await client.get( + url, + headers={"Authorization": f"Bearer {bot_token}"}, + ) + response.raise_for_status() - async def _download_slack_file_bytes(self, url: str, team_id: str = "") -> bytes: - """Download a Slack file and return raw bytes, with retry.""" - import asyncio - import httpx + if audio: + from gateway.platforms.base import cache_audio_from_bytes + return cache_audio_from_bytes(response.content, ext) + else: + from gateway.platforms.base import cache_image_from_bytes + return cache_image_from_bytes(response.content, ext) - bot_token = self._team_clients[team_id].token if team_id and team_id in self._team_clients else self.config.token - last_exc = None + async def _download_slack_file_bytes(self, url: str) -> bytes: + """Download a Slack file and return raw bytes.""" + import httpx + bot_token = self.config.token async with httpx.AsyncClient(timeout=30.0, follow_redirects=True) as client: - for attempt in range(3): - try: - response = await client.get( - url, - headers={"Authorization": f"Bearer {bot_token}"}, - ) - response.raise_for_status() - return response.content - except (httpx.TimeoutException, httpx.HTTPStatusError) as exc: - last_exc = exc - if isinstance(exc, httpx.HTTPStatusError) and exc.response.status_code < 429: - raise - if attempt < 2: - logger.debug("Slack file download retry %d/2 for %s: %s", - attempt + 1, url[:80], exc) - await asyncio.sleep(1.5 * (attempt + 1)) - continue - raise - raise last_exc + response = await client.get( + url, + headers={"Authorization": f"Bearer {bot_token}"}, + ) + response.raise_for_status() + return response.content diff --git a/gateway/platforms/telegram.py b/gateway/platforms/telegram.py index ad7c8f3d65a2..edf038eca293 100644 --- a/gateway/platforms/telegram.py +++ b/gateway/platforms/telegram.py @@ -8,7 +8,6 @@ """ import asyncio -import json import logging import os import re @@ -26,7 +25,6 @@ filters, ) from telegram.constants import ParseMode, ChatType - from telegram.request import HTTPXRequest TELEGRAM_AVAILABLE = True except ImportError: TELEGRAM_AVAILABLE = False @@ -36,7 +34,6 @@ Application = Any CommandHandler = Any TelegramMessageHandler = Any - HTTPXRequest = Any filters = None ParseMode = None ChatType = None @@ -60,13 +57,9 @@ class _MockContextTypes: cache_image_from_bytes, cache_audio_from_bytes, cache_document_from_bytes, + build_text_attachment_injection, SUPPORTED_DOCUMENT_TYPES, ) -from gateway.platforms.telegram_network import ( - TelegramFallbackTransport, - discover_fallback_ips, - parse_fallback_ip_env, -) def check_telegram_requirements() -> bool: @@ -123,8 +116,6 @@ def __init__(self, config: PlatformConfig): super().__init__(config, Platform.TELEGRAM) self._app: Optional[Application] = None self._bot: Optional[Bot] = None - self._webhook_mode: bool = False - self._mention_patterns = self._compile_mention_patterns() self._reply_to_mode: str = getattr(config, 'reply_to_mode', 'first') or 'first' # Buffer rapid/album photo updates so Telegram image bursts are handled # as a single MessageEvent instead of self-interrupting multiple turns. @@ -143,17 +134,6 @@ def __init__(self, config: PlatformConfig): self._polling_conflict_count: int = 0 self._polling_network_error_count: int = 0 self._polling_error_callback_ref = None - # DM Topics: map of topic_name -> message_thread_id (populated at startup) - self._dm_topics: Dict[str, int] = {} - # DM Topics config from extra.dm_topics - self._dm_topics_config: List[Dict[str, Any]] = self.config.extra.get("dm_topics", []) - - def _fallback_ips(self) -> list[str]: - """Return validated fallback IPs from config (populated by _apply_env_overrides).""" - configured = self.config.extra.get("fallback_ips", []) if getattr(self.config, "extra", None) else [] - if isinstance(configured, str): - configured = configured.split(",") - return parse_fallback_ip_env(",".join(str(v) for v in configured) if configured else None) @staticmethod def _looks_like_polling_conflict(error: Exception) -> bool: @@ -236,14 +216,7 @@ async def _handle_polling_network_error(self, error: Exception) -> None: self._polling_network_error_count = 0 except Exception as retry_err: logger.warning("[%s] Telegram polling reconnect failed: %s", self.name, retry_err) - # start_polling failed — polling is dead and no further error - # callbacks will fire, so schedule the next retry ourselves. - if not self.has_fatal_error: - task = asyncio.ensure_future( - self._handle_polling_network_error(retry_err) - ) - self._background_tasks.add(task) - task.add_done_callback(self._background_tasks.discard) + # The next network error will trigger another attempt. async def _handle_polling_conflict(self, error: Exception) -> None: if self.has_fatal_error and self.fatal_error_code == "telegram_polling_conflict": @@ -301,177 +274,8 @@ async def _handle_polling_conflict(self, error: Exception) -> None: logger.warning("[%s] Failed stopping Telegram polling after conflict: %s", self.name, stop_error, exc_info=True) await self._notify_fatal_error() - async def _create_dm_topic( - self, - chat_id: int, - name: str, - icon_color: Optional[int] = None, - icon_custom_emoji_id: Optional[str] = None, - ) -> Optional[int]: - """Create a forum topic in a private (DM) chat. - - Uses Bot API 9.4's createForumTopic which now works for 1-on-1 chats. - Returns the message_thread_id on success, None on failure. - """ - if not self._bot: - return None - try: - kwargs: Dict[str, Any] = {"chat_id": chat_id, "name": name} - if icon_color is not None: - kwargs["icon_color"] = icon_color - if icon_custom_emoji_id: - kwargs["icon_custom_emoji_id"] = icon_custom_emoji_id - - topic = await self._bot.create_forum_topic(**kwargs) - thread_id = topic.message_thread_id - logger.info( - "[%s] Created DM topic '%s' in chat %s -> thread_id=%s", - self.name, name, chat_id, thread_id, - ) - return thread_id - except Exception as e: - error_text = str(e).lower() - # If topic already exists, try to find it via getForumTopicIconStickers - # or we just log and skip — Telegram doesn't provide a "list topics" API - if "topic_name_duplicate" in error_text or "already" in error_text: - logger.info( - "[%s] DM topic '%s' already exists in chat %s (will be mapped from incoming messages)", - self.name, name, chat_id, - ) - else: - logger.warning( - "[%s] Failed to create DM topic '%s' in chat %s: %s", - self.name, name, chat_id, e, - ) - return None - - def _persist_dm_topic_thread_id(self, chat_id: int, topic_name: str, thread_id: int) -> None: - """Save a newly created thread_id back into config.yaml so it persists across restarts.""" - try: - from hermes_constants import get_hermes_home - config_path = get_hermes_home() / "config.yaml" - if not config_path.exists(): - logger.warning("[%s] Config file not found at %s, cannot persist thread_id", self.name, config_path) - return - - import yaml as _yaml - with open(config_path, "r") as f: - config = _yaml.safe_load(f) or {} - - # Navigate to platforms.telegram.extra.dm_topics - dm_topics = ( - config.get("platforms", {}) - .get("telegram", {}) - .get("extra", {}) - .get("dm_topics", []) - ) - if not dm_topics: - return - - changed = False - for chat_entry in dm_topics: - if int(chat_entry.get("chat_id", 0)) != int(chat_id): - continue - for t in chat_entry.get("topics", []): - if t.get("name") == topic_name and not t.get("thread_id"): - t["thread_id"] = thread_id - changed = True - break - - if changed: - with open(config_path, "w") as f: - _yaml.dump(config, f, default_flow_style=False, sort_keys=False) - logger.info( - "[%s] Persisted thread_id=%s for topic '%s' in config.yaml", - self.name, thread_id, topic_name, - ) - except Exception as e: - logger.warning("[%s] Failed to persist thread_id to config: %s", self.name, e, exc_info=True) - - async def _setup_dm_topics(self) -> None: - """Load or create configured DM topics for specified chats. - - Reads config.extra['dm_topics'] — a list of dicts: - [ - { - "chat_id": 123456789, - "topics": [ - {"name": "General", "icon_color": 7322096, "thread_id": 100}, - {"name": "Accessibility Auditor", "icon_color": 9367192, "skill": "accessibility-auditor"} - ] - } - ] - - If a topic already has a thread_id in the config (persisted from a previous - creation), it is loaded into the cache without calling createForumTopic. - Only topics without a thread_id are created via the API, and their thread_id - is then saved back to config.yaml for future restarts. - """ - if not self._dm_topics_config: - return - - for chat_entry in self._dm_topics_config: - chat_id = chat_entry.get("chat_id") - topics = chat_entry.get("topics", []) - if not chat_id or not topics: - continue - - logger.info( - "[%s] Setting up %d DM topic(s) for chat %s", - self.name, len(topics), chat_id, - ) - - for topic_conf in topics: - topic_name = topic_conf.get("name") - if not topic_name: - continue - - cache_key = f"{chat_id}:{topic_name}" - - # If thread_id is already persisted in config, just load into cache - existing_thread_id = topic_conf.get("thread_id") - if existing_thread_id: - self._dm_topics[cache_key] = int(existing_thread_id) - logger.info( - "[%s] DM topic loaded from config: %s -> thread_id=%s", - self.name, cache_key, existing_thread_id, - ) - continue - - # No persisted thread_id — create the topic via API - icon_color = topic_conf.get("icon_color") - icon_emoji = topic_conf.get("icon_custom_emoji_id") - - thread_id = await self._create_dm_topic( - chat_id=int(chat_id), - name=topic_name, - icon_color=icon_color, - icon_custom_emoji_id=icon_emoji, - ) - - if thread_id: - self._dm_topics[cache_key] = thread_id - logger.info( - "[%s] DM topic cached: %s -> thread_id=%s", - self.name, cache_key, thread_id, - ) - # Persist thread_id to config so we don't recreate on next restart - self._persist_dm_topic_thread_id(int(chat_id), topic_name, thread_id) - async def connect(self) -> bool: - """Connect to Telegram via polling or webhook. - - By default, uses long polling (outbound connection to Telegram). - If ``TELEGRAM_WEBHOOK_URL`` is set, starts an HTTP webhook server - instead. Webhook mode is useful for cloud deployments (Fly.io, - Railway) where inbound HTTP can wake a suspended machine. - - Env vars for webhook mode:: - - TELEGRAM_WEBHOOK_URL Public HTTPS URL (e.g. https://app.fly.dev/telegram) - TELEGRAM_WEBHOOK_PORT Local listen port (default 8443) - TELEGRAM_WEBHOOK_SECRET Secret token for update verification - """ + """Connect to Telegram and start polling for updates.""" if not TELEGRAM_AVAILABLE: logger.error( "[%s] python-telegram-bot not installed. Run: pip install python-telegram-bot", @@ -504,26 +308,7 @@ async def connect(self) -> bool: return False # Build the application - builder = Application.builder().token(self.config.token) - fallback_ips = self._fallback_ips() - if not fallback_ips: - fallback_ips = await discover_fallback_ips() - logger.info( - "[%s] Auto-discovered Telegram fallback IPs: %s", - self.name, - ", ".join(fallback_ips), - ) - if fallback_ips: - logger.warning( - "[%s] Telegram fallback IPs active: %s", - self.name, - ", ".join(fallback_ips), - ) - transport = TelegramFallbackTransport(fallback_ips) - request = HTTPXRequest(httpx_kwargs={"transport": transport}) - get_updates_request = HTTPXRequest(httpx_kwargs={"transport": transport}) - builder = builder.request(request).get_updates_request(get_updates_request) - self._app = builder.build() + self._app = Application.builder().token(self.config.token).build() self._bot = self._app.bot # Register handlers @@ -565,76 +350,37 @@ async def connect(self) -> bool: else: raise await self._app.start() + loop = asyncio.get_running_loop() - # Decide between webhook and polling mode - webhook_url = os.getenv("TELEGRAM_WEBHOOK_URL", "").strip() - - if webhook_url: - # ── Webhook mode ───────────────────────────────────── - # Telegram pushes updates to our HTTP endpoint. This - # enables cloud platforms (Fly.io, Railway) to auto-wake - # suspended machines on inbound HTTP traffic. - webhook_port = int(os.getenv("TELEGRAM_WEBHOOK_PORT", "8443")) - webhook_secret = os.getenv("TELEGRAM_WEBHOOK_SECRET", "").strip() or None - from urllib.parse import urlparse - webhook_path = urlparse(webhook_url).path or "/telegram" - - await self._app.updater.start_webhook( - listen="0.0.0.0", - port=webhook_port, - url_path=webhook_path, - webhook_url=webhook_url, - secret_token=webhook_secret, - allowed_updates=Update.ALL_TYPES, - drop_pending_updates=True, - ) - self._webhook_mode = True - logger.info( - "[%s] Webhook server listening on 0.0.0.0:%d%s", - self.name, webhook_port, webhook_path, - ) - else: - # ── Polling mode (default) ─────────────────────────── - loop = asyncio.get_running_loop() - - def _polling_error_callback(error: Exception) -> None: - if self._polling_error_task and not self._polling_error_task.done(): - return - if self._looks_like_polling_conflict(error): - self._polling_error_task = loop.create_task(self._handle_polling_conflict(error)) - elif self._looks_like_network_error(error): - logger.warning("[%s] Telegram network error, scheduling reconnect: %s", self.name, error) - self._polling_error_task = loop.create_task(self._handle_polling_network_error(error)) - else: - logger.error("[%s] Telegram polling error: %s", self.name, error, exc_info=True) + def _polling_error_callback(error: Exception) -> None: + if self._polling_error_task and not self._polling_error_task.done(): + return + if self._looks_like_polling_conflict(error): + self._polling_error_task = loop.create_task(self._handle_polling_conflict(error)) + elif self._looks_like_network_error(error): + logger.warning("[%s] Telegram network error, scheduling reconnect: %s", self.name, error) + self._polling_error_task = loop.create_task(self._handle_polling_network_error(error)) + else: + logger.error("[%s] Telegram polling error: %s", self.name, error, exc_info=True) - # Store reference for retry use in _handle_polling_conflict - self._polling_error_callback_ref = _polling_error_callback + # Store reference for retry use in _handle_polling_conflict + self._polling_error_callback_ref = _polling_error_callback - await self._app.updater.start_polling( - allowed_updates=Update.ALL_TYPES, - drop_pending_updates=True, - error_callback=_polling_error_callback, - ) + await self._app.updater.start_polling( + allowed_updates=Update.ALL_TYPES, + drop_pending_updates=True, + error_callback=_polling_error_callback, + ) # Register bot commands so Telegram shows a hint menu when users type / # List is derived from the central COMMAND_REGISTRY — adding a new # gateway command there automatically adds it to the Telegram menu. try: from telegram import BotCommand - from hermes_cli.commands import telegram_menu_commands - # Telegram allows up to 100 commands but has an undocumented - # payload size limit. Skill descriptions are truncated to 40 - # chars in telegram_menu_commands() to fit 100 commands safely. - menu_commands, hidden_count = telegram_menu_commands(max_commands=100) + from hermes_cli.commands import telegram_bot_commands await self._bot.set_my_commands([ - BotCommand(name, desc) for name, desc in menu_commands + BotCommand(name, desc) for name, desc in telegram_bot_commands() ]) - if hidden_count: - logger.info( - "[%s] Telegram menu: %d commands registered, %d hidden (over 100 limit). Use /commands for full list.", - self.name, len(menu_commands), hidden_count, - ) except Exception as e: logger.warning( "[%s] Could not register Telegram command menu: %s", @@ -644,20 +390,7 @@ def _polling_error_callback(error: Exception) -> None: ) self._mark_connected() - mode = "webhook" if self._webhook_mode else "polling" - logger.info("[%s] Connected to Telegram (%s mode)", self.name, mode) - - # Set up DM topics (Bot API 9.4 — Private Chat Topics) - # Runs after connection is established so the bot can call createForumTopic. - # Failures here are non-fatal — the bot works fine without topics. - try: - await self._setup_dm_topics() - except Exception as topics_err: - logger.warning( - "[%s] DM topics setup failed (non-fatal): %s", - self.name, topics_err, exc_info=True, - ) - + logger.info("[%s] Connected and polling for Telegram updates", self.name) return True except Exception as e: @@ -673,7 +406,7 @@ def _polling_error_callback(error: Exception) -> None: return False async def disconnect(self) -> None: - """Stop polling/webhook, cancel pending album flushes, and disconnect.""" + """Stop polling, cancel pending album flushes, and disconnect.""" pending_media_group_tasks = list(self._media_group_tasks.values()) for task in pending_media_group_tasks: task.cancel() @@ -742,10 +475,6 @@ async def send( if not self._bot: return SendResult(success=False, error="Not connected") - # Skip whitespace-only text to prevent Telegram 400 empty-text errors. - if not content or not content.strip(): - return SendResult(success=True, message_id=None) - try: # Format and split message if needed formatted = self.format_message(content) @@ -767,15 +496,9 @@ async def send( except ImportError: _NetErr = OSError # type: ignore[misc,assignment] - try: - from telegram.error import BadRequest as _BadReq - except ImportError: - _BadReq = None # type: ignore[assignment,misc] - for i, chunk in enumerate(chunks): should_thread = self._should_thread_reply(reply_to, i) reply_to_id = int(reply_to) if should_thread else None - effective_thread_id = int(thread_id) if thread_id else None msg = None for _send_attempt in range(3): @@ -787,7 +510,7 @@ async def send( text=chunk, parse_mode=ParseMode.MARKDOWN_V2, reply_to_message_id=reply_to_id, - message_thread_id=effective_thread_id, + message_thread_id=int(thread_id) if thread_id else None, ) except Exception as md_error: # Markdown parsing failed, try plain text @@ -799,40 +522,12 @@ async def send( text=plain_chunk, parse_mode=None, reply_to_message_id=reply_to_id, - message_thread_id=effective_thread_id, + message_thread_id=int(thread_id) if thread_id else None, ) else: raise break # success except _NetErr as send_err: - # BadRequest is a subclass of NetworkError in - # python-telegram-bot but represents permanent errors - # (not transient network issues). Detect and handle - # specific cases instead of blindly retrying. - if _BadReq and isinstance(send_err, _BadReq): - err_lower = str(send_err).lower() - if "thread not found" in err_lower and effective_thread_id is not None: - # Thread doesn't exist — retry without - # message_thread_id so the message still - # reaches the chat. - logger.warning( - "[%s] Thread %s not found, retrying without message_thread_id", - self.name, effective_thread_id, - ) - effective_thread_id = None - continue - if "message to be replied not found" in err_lower and reply_to_id is not None: - # Original message was deleted before we - # could reply — clear reply target and retry - # so the response is still delivered. - logger.warning( - "[%s] Reply target deleted, retrying without reply_to: %s", - self.name, send_err, - ) - reply_to_id = None - continue - # Other BadRequest errors are permanent — don't retry - raise if _send_attempt < 2: wait = 2 ** _send_attempt logger.warning("[%s] Network error on send (attempt %d/3), retrying in %ds: %s", @@ -900,9 +595,7 @@ async def edit_message( except Exception: pass # best-effort truncation return SendResult(success=True, message_id=message_id) - # Flood control / RetryAfter — short waits are retried inline, - # long waits return a failure immediately so streaming can fall back - # to a normal final send instead of leaving a truncated partial. + # Flood control / RetryAfter — back off and retry once retry_after = getattr(e, "retry_after", None) if retry_after is not None or "retry after" in err_str: wait = retry_after if retry_after else 1.0 @@ -910,8 +603,6 @@ async def edit_message( "[%s] Telegram flood control, waiting %.1fs", self.name, wait, ) - if wait > 5.0: - return SendResult(success=False, error=f"flood_control:{wait}") await asyncio.sleep(wait) try: await self._bot.edit_message_text( @@ -1388,148 +1079,6 @@ def _esc_bare(m, _seg=_seg): return text - # ── Group mention gating ────────────────────────────────────────────── - - def _telegram_require_mention(self) -> bool: - """Return whether group chats should require an explicit bot trigger.""" - configured = self.config.extra.get("require_mention") - if configured is not None: - if isinstance(configured, str): - return configured.lower() in ("true", "1", "yes", "on") - return bool(configured) - return os.getenv("TELEGRAM_REQUIRE_MENTION", "false").lower() in ("true", "1", "yes", "on") - - def _telegram_free_response_chats(self) -> set[str]: - raw = self.config.extra.get("free_response_chats") - if raw is None: - raw = os.getenv("TELEGRAM_FREE_RESPONSE_CHATS", "") - if isinstance(raw, list): - return {str(part).strip() for part in raw if str(part).strip()} - return {part.strip() for part in str(raw).split(",") if part.strip()} - - def _compile_mention_patterns(self) -> List[re.Pattern]: - """Compile optional regex wake-word patterns for group triggers.""" - patterns = self.config.extra.get("mention_patterns") - if patterns is None: - raw = os.getenv("TELEGRAM_MENTION_PATTERNS", "").strip() - if raw: - try: - loaded = json.loads(raw) - except Exception: - loaded = [part.strip() for part in raw.splitlines() if part.strip()] - if not loaded: - loaded = [part.strip() for part in raw.split(",") if part.strip()] - patterns = loaded - - if patterns is None: - return [] - if isinstance(patterns, str): - patterns = [patterns] - if not isinstance(patterns, list): - logger.warning( - "[%s] telegram mention_patterns must be a list or string; got %s", - self.name, - type(patterns).__name__, - ) - return [] - - compiled: List[re.Pattern] = [] - for pattern in patterns: - if not isinstance(pattern, str) or not pattern.strip(): - continue - try: - compiled.append(re.compile(pattern, re.IGNORECASE)) - except re.error as exc: - logger.warning("[%s] Invalid Telegram mention pattern %r: %s", self.name, pattern, exc) - if compiled: - logger.info("[%s] Loaded %d Telegram mention pattern(s)", self.name, len(compiled)) - return compiled - - def _is_group_chat(self, message: Message) -> bool: - chat = getattr(message, "chat", None) - if not chat: - return False - chat_type = str(getattr(chat, "type", "")).split(".")[-1].lower() - return chat_type in ("group", "supergroup") - - def _is_reply_to_bot(self, message: Message) -> bool: - if not self._bot or not getattr(message, "reply_to_message", None): - return False - reply_user = getattr(message.reply_to_message, "from_user", None) - return bool(reply_user and getattr(reply_user, "id", None) == getattr(self._bot, "id", None)) - - def _message_mentions_bot(self, message: Message) -> bool: - if not self._bot: - return False - - bot_username = (getattr(self._bot, "username", None) or "").lstrip("@").lower() - bot_id = getattr(self._bot, "id", None) - - def _iter_sources(): - yield getattr(message, "text", None) or "", getattr(message, "entities", None) or [] - yield getattr(message, "caption", None) or "", getattr(message, "caption_entities", None) or [] - - for source_text, entities in _iter_sources(): - if bot_username and f"@{bot_username}" in source_text.lower(): - return True - for entity in entities: - entity_type = str(getattr(entity, "type", "")).split(".")[-1].lower() - if entity_type == "mention" and bot_username: - offset = int(getattr(entity, "offset", -1)) - length = int(getattr(entity, "length", 0)) - if offset < 0 or length <= 0: - continue - if source_text[offset:offset + length].strip().lower() == f"@{bot_username}": - return True - elif entity_type == "text_mention": - user = getattr(entity, "user", None) - if user and getattr(user, "id", None) == bot_id: - return True - return False - - def _message_matches_mention_patterns(self, message: Message) -> bool: - if not self._mention_patterns: - return False - for candidate in (getattr(message, "text", None), getattr(message, "caption", None)): - if not candidate: - continue - for pattern in self._mention_patterns: - if pattern.search(candidate): - return True - return False - - def _clean_bot_trigger_text(self, text: Optional[str]) -> Optional[str]: - if not text or not self._bot or not getattr(self._bot, "username", None): - return text - username = re.escape(self._bot.username) - cleaned = re.sub(rf"(?i)@{username}\b[,:\-]*\s*", "", text).strip() - return cleaned or text - - def _should_process_message(self, message: Message, *, is_command: bool = False) -> bool: - """Apply Telegram group trigger rules. - - DMs remain unrestricted. Group/supergroup messages are accepted when: - - the chat is explicitly allowlisted in ``free_response_chats`` - - ``require_mention`` is disabled - - the message is a command - - the message replies to the bot - - the bot is @mentioned - - the text/caption matches a configured regex wake-word pattern - """ - if not self._is_group_chat(message): - return True - if str(getattr(getattr(message, "chat", None), "id", "")) in self._telegram_free_response_chats(): - return True - if not self._telegram_require_mention(): - return True - if is_command: - return True - if self._is_reply_to_bot(message): - return True - if self._message_mentions_bot(message): - return True - return self._message_matches_mention_patterns(message) - async def _handle_text_message(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """Handle incoming text messages. @@ -1539,19 +1088,14 @@ async def _handle_text_message(self, update: Update, context: ContextTypes.DEFAU """ if not update.message or not update.message.text: return - if not self._should_process_message(update.message): - return event = self._build_message_event(update.message, MessageType.TEXT) - event.text = self._clean_bot_trigger_text(event.text) self._enqueue_text_event(event) async def _handle_command(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """Handle incoming command messages.""" if not update.message or not update.message.text: return - if not self._should_process_message(update.message, is_command=True): - return event = self._build_message_event(update.message, MessageType.COMMAND) await self.handle_message(event) @@ -1560,8 +1104,6 @@ async def _handle_location_message(self, update: Update, context: ContextTypes.D """Handle incoming location/venue pin messages.""" if not update.message: return - if not self._should_process_message(update.message): - return msg = update.message venue = getattr(msg, "venue", None) @@ -1705,8 +1247,6 @@ async def _handle_media_message(self, update: Update, context: ContextTypes.DEFA """Handle incoming media messages, downloading images to local cache.""" if not update.message: return - if not self._should_process_message(update.message): - return msg = update.message @@ -1730,7 +1270,7 @@ async def _handle_media_message(self, update: Update, context: ContextTypes.DEFA # Add caption as text if msg.caption: - event.text = self._clean_bot_trigger_text(msg.caption) + event.text = msg.caption # Handle stickers: describe via vision tool with caching if msg.sticker: @@ -1840,23 +1380,16 @@ async def _handle_media_message(self, update: Update, context: ContextTypes.DEFA event.media_types = [mime_type] logger.info("[Telegram] Cached user document at %s", cached_path) - # For text files, inject content into event.text (capped at 100 KB) - MAX_TEXT_INJECT_BYTES = 100 * 1024 - if ext in (".md", ".txt") and len(raw_bytes) <= MAX_TEXT_INJECT_BYTES: - try: - text_content = raw_bytes.decode("utf-8") - display_name = original_filename or f"document{ext}" - display_name = re.sub(r'[^\w.\- ]', '_', display_name) - injection = f"[Content of {display_name}]:\n{text_content}" - if event.text: - event.text = f"{injection}\n\n{event.text}" - else: - event.text = injection - except UnicodeDecodeError: - logger.warning( - "[Telegram] Could not decode text file as UTF-8, skipping content injection", - exc_info=True, - ) + injection = build_text_attachment_injection( + raw_bytes, + original_filename or f"document{ext}", + ext, + ) + if injection: + if event.text: + event.text = f"{injection}\n\n{event.text}" + else: + event.text = injection except Exception as e: logger.warning("[Telegram] Failed to cache document: %s", e, exc_info=True) @@ -1975,100 +1508,6 @@ async def _handle_sticker(self, msg: Message, event: "MessageEvent") -> None: emoji, set_name, ) - def _reload_dm_topics_from_config(self) -> None: - """Re-read dm_topics from config.yaml and load any new thread_ids into cache. - - This allows topics created externally (e.g. by the agent via API) to be - recognized without a gateway restart. - """ - try: - from hermes_constants import get_hermes_home - config_path = get_hermes_home() / "config.yaml" - if not config_path.exists(): - return - - import yaml as _yaml - with open(config_path, "r") as f: - config = _yaml.safe_load(f) or {} - - dm_topics = ( - config.get("platforms", {}) - .get("telegram", {}) - .get("extra", {}) - .get("dm_topics", []) - ) - if not dm_topics: - return - - # Update in-memory config and cache any new thread_ids - self._dm_topics_config = dm_topics - for chat_entry in dm_topics: - cid = chat_entry.get("chat_id") - if not cid: - continue - for t in chat_entry.get("topics", []): - tid = t.get("thread_id") - name = t.get("name") - if tid and name: - cache_key = f"{cid}:{name}" - if cache_key not in self._dm_topics: - self._dm_topics[cache_key] = int(tid) - logger.info( - "[%s] Hot-loaded DM topic from config: %s -> thread_id=%s", - self.name, cache_key, tid, - ) - except Exception as e: - logger.debug("[%s] Failed to reload dm_topics from config: %s", self.name, e) - - def _get_dm_topic_info(self, chat_id: str, thread_id: Optional[str]) -> Optional[Dict[str, Any]]: - """Look up DM topic config by chat_id and thread_id. - - Returns the topic config dict (name, skill, etc.) if this thread_id - matches a known DM topic, or None. - """ - if not thread_id: - return None - - thread_id_int = int(thread_id) - - # Check cached topics first (created by us or loaded at startup) - for key, cached_tid in self._dm_topics.items(): - if cached_tid == thread_id_int and key.startswith(f"{chat_id}:"): - topic_name = key.split(":", 1)[1] - # Find the full config for this topic - for chat_entry in self._dm_topics_config: - if str(chat_entry.get("chat_id")) == chat_id: - for t in chat_entry.get("topics", []): - if t.get("name") == topic_name: - return t - return {"name": topic_name} - - # Not in cache — hot-reload config in case topics were added externally - self._reload_dm_topics_from_config() - - # Check cache again after reload - for key, cached_tid in self._dm_topics.items(): - if cached_tid == thread_id_int and key.startswith(f"{chat_id}:"): - topic_name = key.split(":", 1)[1] - for chat_entry in self._dm_topics_config: - if str(chat_entry.get("chat_id")) == chat_id: - for t in chat_entry.get("topics", []): - if t.get("name") == topic_name: - return t - return {"name": topic_name} - - return None - - def _cache_dm_topic_from_message(self, chat_id: str, thread_id: str, topic_name: str) -> None: - """Cache a thread_id -> topic_name mapping discovered from an incoming message.""" - cache_key = f"{chat_id}:{topic_name}" - if cache_key not in self._dm_topics: - self._dm_topics[cache_key] = int(thread_id) - logger.info( - "[%s] Cached DM topic from message: %s -> thread_id=%s", - self.name, cache_key, thread_id, - ) - def _build_message_event(self, message: Message, msg_type: MessageType) -> MessageEvent: """Build a MessageEvent from a Telegram message.""" chat = message.chat @@ -2080,27 +1519,7 @@ def _build_message_event(self, message: Message, msg_type: MessageType) -> Messa chat_type = "group" elif chat.type == ChatType.CHANNEL: chat_type = "channel" - - # Resolve DM topic name and skill binding - thread_id_raw = message.message_thread_id - thread_id_str = str(thread_id_raw) if thread_id_raw else None - chat_topic = None - topic_skill = None - - if chat_type == "dm" and thread_id_str: - topic_info = self._get_dm_topic_info(str(chat.id), thread_id_str) - if topic_info: - chat_topic = topic_info.get("name") - topic_skill = topic_info.get("skill") - - # Also check forum_topic_created service message for topic discovery - if hasattr(message, "forum_topic_created") and message.forum_topic_created: - created_name = message.forum_topic_created.name - if created_name: - self._cache_dm_topic_from_message(str(chat.id), thread_id_str, created_name) - if not chat_topic: - chat_topic = created_name - + # Build source source = self.build_source( chat_id=str(chat.id), @@ -2108,8 +1527,7 @@ def _build_message_event(self, message: Message, msg_type: MessageType) -> Messa chat_type=chat_type, user_id=str(user.id) if user else None, user_name=user.full_name if user else None, - thread_id=thread_id_str, - chat_topic=chat_topic, + thread_id=str(message.message_thread_id) if message.message_thread_id else None, ) # Extract reply context if this message is a reply @@ -2127,6 +1545,5 @@ def _build_message_event(self, message: Message, msg_type: MessageType) -> Messa message_id=str(message.message_id), reply_to_message_id=reply_to_id, reply_to_text=reply_to_text, - auto_skill=topic_skill, timestamp=message.date, ) diff --git a/model_tools.py b/model_tools.py index ec472ff99ea9..fb6919fa772a 100644 --- a/model_tools.py +++ b/model_tools.py @@ -139,6 +139,7 @@ def _discover_tools(): "tools.web_tools", "tools.terminal_tool", "tools.file_tools", + "tools.archive_tool", "tools.vision_tools", "tools.mixture_of_agents_tool", "tools.image_generation_tool", diff --git a/tests/gateway/test_discord_document_handling.py b/tests/gateway/test_discord_document_handling.py index b7be161cd90e..7f918d1c7383 100644 --- a/tests/gateway/test_discord_document_handling.py +++ b/tests/gateway/test_discord_document_handling.py @@ -227,16 +227,19 @@ async def test_oversized_document_skipped(self, adapter): adapter.handle_message.assert_called_once() @pytest.mark.asyncio - async def test_unsupported_type_skipped(self, adapter): - """An unsupported file type (.zip) should be skipped silently.""" + async def test_zip_document_cached(self, adapter): + """A .zip file should be cached as a supported document.""" msg = make_message([ make_attachment(filename="archive.zip", content_type="application/zip") ]) - await adapter._handle_message(msg) + + with _mock_aiohttp_download(b"PK\x03\x04test"): + await adapter._handle_message(msg) event = adapter.handle_message.call_args[0][0] - assert event.media_urls == [] - assert event.message_type == MessageType.TEXT + assert len(event.media_urls) == 1 + assert event.media_types == ["application/zip"] + assert event.message_type == MessageType.DOCUMENT @pytest.mark.asyncio async def test_download_error_handled(self, adapter): diff --git a/tests/gateway/test_document_cache.py b/tests/gateway/test_document_cache.py index 18440ed9c270..cc756cea85f7 100644 --- a/tests/gateway/test_document_cache.py +++ b/tests/gateway/test_document_cache.py @@ -151,7 +151,7 @@ def test_all_extensions_have_mime_types(self): @pytest.mark.parametrize( "ext", - [".pdf", ".md", ".txt", ".docx", ".xlsx", ".pptx"], + [".pdf", ".md", ".txt", ".zip", ".docx", ".xlsx", ".pptx"], ) def test_expected_extensions_present(self, ext): assert ext in SUPPORTED_DOCUMENT_TYPES diff --git a/tests/gateway/test_slack.py b/tests/gateway/test_slack.py index 16924b590146..81f8077ad6b3 100644 --- a/tests/gateway/test_slack.py +++ b/tests/gateway/test_slack.py @@ -408,19 +408,22 @@ async def test_large_txt_not_injected(self, adapter): assert "[Content of" not in (msg_event.text or "") @pytest.mark.asyncio - async def test_unsupported_file_type_skipped(self, adapter): - """A .zip file should be silently skipped.""" - event = self._make_event(files=[{ - "mimetype": "application/zip", - "name": "archive.zip", - "url_private_download": "https://files.slack.com/archive.zip", - "size": 1024, - }]) - await adapter._handle_slack_message(event) + async def test_zip_file_cached(self, adapter): + """A .zip file should be cached as a supported document.""" + with patch.object(adapter, "_download_slack_file_bytes", new_callable=AsyncMock) as dl: + dl.return_value = b"PK\x03\x04zip" + event = self._make_event(files=[{ + "mimetype": "application/zip", + "name": "archive.zip", + "url_private_download": "https://files.slack.com/archive.zip", + "size": 1024, + }]) + await adapter._handle_slack_message(event) msg_event = adapter.handle_message.call_args[0][0] - assert msg_event.message_type == MessageType.TEXT - assert len(msg_event.media_urls) == 0 + assert msg_event.message_type == MessageType.DOCUMENT + assert len(msg_event.media_urls) == 1 + assert msg_event.media_types == ["application/zip"] @pytest.mark.asyncio async def test_oversized_document_skipped(self, adapter): diff --git a/tests/gateway/test_telegram_documents.py b/tests/gateway/test_telegram_documents.py index 11a8df5f88c0..c6672d49e9a0 100644 --- a/tests/gateway/test_telegram_documents.py +++ b/tests/gateway/test_telegram_documents.py @@ -236,15 +236,15 @@ async def test_caption_preserved_with_injection(self, adapter): assert "Please summarize" in event.text @pytest.mark.asyncio - async def test_unsupported_type_rejected(self, adapter): + async def test_zip_document_cached(self, adapter): doc = _make_document(file_name="archive.zip", mime_type="application/zip", file_size=100) msg = _make_message(document=doc) update = _make_update(msg) await adapter._handle_media_message(update, MagicMock()) event = adapter.handle_message.call_args[0][0] - assert "Unsupported document type" in event.text - assert ".zip" in event.text + assert event.media_urls and event.media_urls[0].endswith("archive.zip") + assert event.media_types == ["application/zip"] @pytest.mark.asyncio async def test_oversized_file_rejected(self, adapter): diff --git a/tests/tools/test_archive_tool.py b/tests/tools/test_archive_tool.py new file mode 100644 index 000000000000..4496082891fa --- /dev/null +++ b/tests/tools/test_archive_tool.py @@ -0,0 +1,98 @@ +import io +import json +import os +import tarfile +import zipfile +from pathlib import Path + +from tools.archive_tool import extract_archive_tool + + +def _make_zip(path: Path, members: dict[str, bytes]) -> None: + with zipfile.ZipFile(path, "w") as zf: + for name, data in members.items(): + zf.writestr(name, data) + + +def _make_zip_with_symlink(path: Path, link_name: str, target: str) -> None: + info = zipfile.ZipInfo(link_name) + info.create_system = 3 + info.external_attr = 0o120777 << 16 + with zipfile.ZipFile(path, "w") as zf: + zf.writestr(info, target) + + +def _make_tar_gz(path: Path, members: dict[str, bytes]) -> None: + with tarfile.open(path, "w:gz") as tf: + for name, data in members.items(): + info = tarfile.TarInfo(name) + info.size = len(data) + tf.addfile(info, io.BytesIO(data)) + + +def test_extract_archive_tool_extracts_zip(tmp_path): + archive = tmp_path / "sample.zip" + _make_zip(archive, {"nested/hello.txt": b"hi"}) + + result = json.loads(extract_archive_tool(str(archive))) + + assert result["success"] is True + output_dir = Path(result["output_dir"]) + assert (output_dir / "nested" / "hello.txt").read_text() == "hi" + assert "nested/hello.txt" in result["extracted_files"] + + +def test_extract_archive_tool_extracts_tar_gz(tmp_path): + archive = tmp_path / "sample.tar.gz" + _make_tar_gz(archive, {"hello.txt": b"hi"}) + + result = json.loads(extract_archive_tool(str(archive))) + + assert result["success"] is True + output_dir = Path(result["output_dir"]) + assert (output_dir / "hello.txt").read_text() == "hi" + + +def test_extract_archive_tool_blocks_zip_slip(tmp_path): + archive = tmp_path / "escape.zip" + _make_zip(archive, {"../../escape.txt": b"pwnd"}) + + result = json.loads(extract_archive_tool(str(archive))) + + assert result["success"] is False + assert "unsafe archive member path" in result["error"].lower() + assert not (tmp_path / "escape.txt").exists() + + +def test_extract_archive_tool_rejects_zip_symlink(tmp_path): + archive = tmp_path / "symlink.zip" + _make_zip_with_symlink(archive, "link", "target.txt") + + result = json.loads(extract_archive_tool(str(archive))) + + assert result["success"] is False + assert "unsupported archive member type" in result["error"].lower() + + +def test_extract_archive_tool_rejects_symlinked_destination(tmp_path): + archive = tmp_path / "sample.zip" + _make_zip(archive, {"nested/hello.txt": b"hi"}) + real_dir = tmp_path / "real" + real_dir.mkdir() + symlink_dir = tmp_path / "linked-out" + os.symlink(real_dir, symlink_dir) + + result = json.loads(extract_archive_tool(str(archive), output_dir=str(symlink_dir))) + + assert result["success"] is False + assert "symlinked destination" in result["error"].lower() + + +def test_extract_archive_tool_rejects_unsupported_extension(tmp_path): + archive = tmp_path / "sample.bin" + archive.write_bytes(b"not an archive") + + result = json.loads(extract_archive_tool(str(archive))) + + assert result["success"] is False + assert "unsupported archive type" in result["error"].lower() diff --git a/tests/tools/test_modal_sandbox_fixes.py b/tests/tools/test_modal_sandbox_fixes.py index 7e3feb5cf4c6..1fe8b4a81ea2 100644 --- a/tests/tools/test_modal_sandbox_fixes.py +++ b/tests/tools/test_modal_sandbox_fixes.py @@ -36,15 +36,15 @@ class TestToolResolution: """Verify get_tool_definitions returns all expected tools for eval.""" def test_terminal_and_file_toolsets_resolve_all_tools(self): - """enabled_toolsets=['terminal', 'file'] should produce 6 tools.""" + """enabled_toolsets=['terminal', 'file'] should include terminal and file tools.""" from model_tools import get_tool_definitions tools = get_tool_definitions( enabled_toolsets=["terminal", "file"], quiet_mode=True, ) names = {t["function"]["name"] for t in tools} - expected = {"terminal", "process", "read_file", "write_file", "search_files", "patch"} - assert expected == names, f"Expected {expected}, got {names}" + expected_subset = {"terminal", "process", "read_file", "write_file", "search_files", "patch", "extract_archive"} + assert expected_subset.issubset(names), f"Expected at least {expected_subset}, got {names}" def test_terminal_tool_present(self): """The terminal tool must be present (not silently dropped).""" diff --git a/tools/archive_tool.py b/tools/archive_tool.py new file mode 100644 index 000000000000..a6b722070c82 --- /dev/null +++ b/tools/archive_tool.py @@ -0,0 +1,276 @@ +#!/usr/bin/env python3 +"""Archive extraction tool for ZIP and tar archives.""" + +from __future__ import annotations + +import json +import logging +import shutil +import tarfile +import zipfile +from pathlib import Path, PurePosixPath, PureWindowsPath +from tempfile import mkdtemp +from typing import Iterable + +from tools.registry import registry + +logger = logging.getLogger(__name__) + +_SUPPORTED_ARCHIVE_SUFFIXES = ( + ".zip", + ".tar", + ".tar.gz", + ".tgz", + ".tar.bz2", + ".tbz2", + ".tar.xz", + ".txz", +) + +_MAX_EXTRACTED_FILE_LIST = 200 +_MAX_ARCHIVE_BYTES = 100 * 1024 * 1024 +_MAX_EXTRACTED_BYTES = 250 * 1024 * 1024 +_MAX_ARCHIVE_ENTRIES = 10_000 + + +def _normalize_archive_member_parts(member_name: str) -> list[str]: + """Return safe path parts for an archive member.""" + normalized_name = (member_name or "").replace("\\", "/") + posix_path = PurePosixPath(normalized_name) + windows_path = PureWindowsPath(member_name or "") + + if ( + not normalized_name + or posix_path.is_absolute() + or windows_path.is_absolute() + or windows_path.drive + ): + raise ValueError(f"Unsafe archive member path: {member_name}") + + parts = [part for part in posix_path.parts if part not in ("", ".")] + if not parts or any(part == ".." for part in parts): + raise ValueError(f"Unsafe archive member path: {member_name}") + return parts + + +def _resolve_output_dir(archive_path: Path, output_dir: str | None) -> Path: + if output_dir: + raw_target = Path(output_dir).expanduser() + target = (Path.cwd() / raw_target) if not raw_target.is_absolute() else raw_target + if target.exists() and target.is_symlink(): + raise ValueError(f"Refusing to extract into symlinked destination: {target}") + target = target.resolve() + target.mkdir(parents=True, exist_ok=True) + return target + + base_name = archive_path.name + for suffix in (".tar.gz", ".tar.bz2", ".tar.xz", ".tgz", ".tbz2", ".txz", ".zip", ".tar"): + if base_name.lower().endswith(suffix): + base_name = base_name[: -len(suffix)] + break + safe_stem = base_name or "archive" + return Path(mkdtemp(prefix=f"extract_{safe_stem}_", dir=str(archive_path.parent))) + + +def _record_path(collected: list[str], relative_path: Path) -> None: + if len(collected) < _MAX_EXTRACTED_FILE_LIST: + collected.append(relative_path.as_posix()) + + +def _validate_archive_size(archive_path: Path) -> None: + archive_size = archive_path.stat().st_size + if archive_size > _MAX_ARCHIVE_BYTES: + raise ValueError( + f"Archive exceeds size limit ({_MAX_ARCHIVE_BYTES // (1024 * 1024)}MB): {archive_path}" + ) + + +def _ensure_no_symlink_components(destination: Path, target: Path) -> None: + current = destination + for part in target.relative_to(destination).parts: + current = current / part + if current.exists() and current.is_symlink(): + raise ValueError(f"Refusing to extract through symlinked path: {current}") + + +def _is_zip_symlink(member: zipfile.ZipInfo) -> bool: + mode = (member.external_attr >> 16) & 0xFFFF + return (mode & 0o170000) == 0o120000 + + +def _extract_zip(archive_path: Path, destination: Path) -> list[str]: + extracted_files: list[str] = [] + extracted_bytes = 0 + with zipfile.ZipFile(archive_path, "r") as zf: + infos = zf.infolist() + if len(infos) > _MAX_ARCHIVE_ENTRIES: + raise ValueError(f"Archive has too many entries ({len(infos)} > {_MAX_ARCHIVE_ENTRIES})") + for member in infos: + if _is_zip_symlink(member): + raise ValueError(f"Unsupported archive member type: {member.filename}") + parts = _normalize_archive_member_parts(member.filename) + target = destination.joinpath(*parts) + _ensure_no_symlink_components(destination, target.parent if not member.is_dir() else target) + + if member.is_dir(): + target.mkdir(parents=True, exist_ok=True) + continue + + extracted_bytes += member.file_size + if extracted_bytes > _MAX_EXTRACTED_BYTES: + raise ValueError( + f"Archive extracted content exceeds size limit ({_MAX_EXTRACTED_BYTES // (1024 * 1024)}MB)" + ) + + target.parent.mkdir(parents=True, exist_ok=True) + with zf.open(member, "r") as src, open(target, "wb") as dst: + shutil.copyfileobj(src, dst) + _record_path(extracted_files, target.relative_to(destination)) + return extracted_files + + +def _iter_tar_members(tf: tarfile.TarFile) -> Iterable[tarfile.TarInfo]: + for member in tf.getmembers(): + if member.issym() or member.islnk(): + raise ValueError(f"Unsupported archive member type: {member.name}") + yield member + + +def _extract_tar(archive_path: Path, destination: Path) -> list[str]: + extracted_files: list[str] = [] + extracted_bytes = 0 + with tarfile.open(archive_path, "r:*") as tf: + members = list(_iter_tar_members(tf)) + if len(members) > _MAX_ARCHIVE_ENTRIES: + raise ValueError(f"Archive has too many entries ({len(members)} > {_MAX_ARCHIVE_ENTRIES})") + for member in members: + parts = _normalize_archive_member_parts(member.name) + target = destination.joinpath(*parts) + _ensure_no_symlink_components(destination, target.parent if not member.isdir() else target) + + if member.isdir(): + target.mkdir(parents=True, exist_ok=True) + continue + if not member.isfile(): + raise ValueError(f"Unsupported archive member type: {member.name}") + + extracted_bytes += max(0, member.size) + if extracted_bytes > _MAX_EXTRACTED_BYTES: + raise ValueError( + f"Archive extracted content exceeds size limit ({_MAX_EXTRACTED_BYTES // (1024 * 1024)}MB)" + ) + + target.parent.mkdir(parents=True, exist_ok=True) + extracted = tf.extractfile(member) + if extracted is None: + raise ValueError(f"Cannot read archive member: {member.name}") + with extracted, open(target, "wb") as dst: + shutil.copyfileobj(extracted, dst) + _record_path(extracted_files, target.relative_to(destination)) + + try: + target.chmod(member.mode & 0o777) + except OSError: + pass + return extracted_files + + +def _detect_archive_type(archive_path: Path) -> str | None: + lower_name = archive_path.name.lower() + for suffix in _SUPPORTED_ARCHIVE_SUFFIXES: + if lower_name.endswith(suffix): + return suffix + return None + + +def extract_archive_tool(archive_path: str, output_dir: str | None = None, task_id: str | None = None) -> str: + del task_id # reserved for parity with other tool handlers + + try: + path = Path(archive_path).expanduser() + if not path.is_absolute(): + path = (Path.cwd() / path).resolve() + else: + path = path.resolve() + + if not path.exists(): + return json.dumps({"success": False, "error": f"Archive not found: {path}"}) + if not path.is_file(): + return json.dumps({"success": False, "error": f"Not a file: {path}"}) + + _validate_archive_size(path) + + archive_type = _detect_archive_type(path) + if archive_type is None: + return json.dumps( + { + "success": False, + "error": ( + "Unsupported archive type. Supported: " + + ", ".join(_SUPPORTED_ARCHIVE_SUFFIXES) + ), + } + ) + + destination = _resolve_output_dir(path, output_dir) + if archive_type == ".zip": + extracted_files = _extract_zip(path, destination) + else: + extracted_files = _extract_tar(path, destination) + + return json.dumps( + { + "success": True, + "archive_path": str(path), + "archive_type": archive_type, + "output_dir": str(destination), + "extracted_count": len(extracted_files), + "extracted_files": extracted_files, + "truncated": len(extracted_files) >= _MAX_EXTRACTED_FILE_LIST, + } + ) + except (ValueError, tarfile.TarError, zipfile.BadZipFile, OSError) as exc: + logger.warning("Archive extraction failed for %s: %s", archive_path, exc) + return json.dumps({"success": False, "error": str(exc)}) + + +def check_archive_tool_requirements() -> bool: + return True + + +EXTRACT_ARCHIVE_SCHEMA = { + "name": "extract_archive", + "description": ( + "Safely extract a ZIP or tar archive that already exists on disk. " + "Use this for uploaded project bundles or compressed datasets when you " + "need Hermes to unpack them on the host machine instead of relying on shell unzip commands." + ), + "parameters": { + "type": "object", + "properties": { + "archive_path": { + "type": "string", + "description": "Path to a .zip, .tar, .tar.gz, .tgz, .tar.bz2, .tbz2, .tar.xz, or .txz archive.", + }, + "output_dir": { + "type": "string", + "description": "Optional destination directory. Omit to extract into a new sibling directory next to the archive.", + }, + }, + "required": ["archive_path"], + }, +} + + +registry.register( + name="extract_archive", + toolset="file", + schema=EXTRACT_ARCHIVE_SCHEMA, + handler=lambda args, **kw: extract_archive_tool( + archive_path=args.get("archive_path", ""), + output_dir=args.get("output_dir"), + task_id=kw.get("task_id"), + ), + check_fn=check_archive_tool_requirements, + emoji="🗜️", +) diff --git a/toolsets.py b/toolsets.py index 84c19637f940..51f1ec6de170 100644 --- a/toolsets.py +++ b/toolsets.py @@ -34,7 +34,7 @@ # Terminal + process management "terminal", "process", # File manipulation - "read_file", "write_file", "patch", "search_files", + "read_file", "write_file", "patch", "search_files", "extract_archive", # Vision + image generation "vision_analyze", "image_generate", # MoA @@ -148,7 +148,7 @@ "file": { "description": "File manipulation tools: read, write, patch (with fuzzy matching), and search (content + files)", - "tools": ["read_file", "write_file", "patch", "search_files"], + "tools": ["read_file", "write_file", "patch", "search_files", "extract_archive"], "includes": [] },