diff --git a/acp_adapter/events.py b/acp_adapter/events.py index 5d10309d56a8..08da40a685e8 100644 --- a/acp_adapter/events.py +++ b/acp_adapter/events.py @@ -54,14 +54,18 @@ def make_tool_progress_cb( Signature expected by AIAgent:: - tool_progress_callback(name: str, preview: str, args: dict) + tool_progress_callback(event_type: str, name: str, preview: str, args: dict, **kwargs) - Emits ``ToolCallStart`` for each tool invocation and tracks IDs in a FIFO + Emits ``ToolCallStart`` for ``tool.started`` events and tracks IDs in a FIFO queue per tool name so duplicate/parallel same-name calls still complete - against the correct ACP tool call. + against the correct ACP tool call. Other event types (``tool.completed``, + ``reasoning.available``) are silently ignored. """ - def _tool_progress(name: str, preview: str, args: Any = None) -> None: + def _tool_progress(event_type: str, name: str = None, preview: str = None, args: Any = None, **kwargs) -> None: + # Only emit ACP ToolCallStart for tool.started; ignore other event types + if event_type != "tool.started": + return if isinstance(args, str): try: args = json.loads(args) diff --git a/acp_adapter/server.py b/acp_adapter/server.py index c5c29c5ad392..11064a1e4e3e 100644 --- a/acp_adapter/server.py +++ b/acp_adapter/server.py @@ -12,7 +12,8 @@ import acp from acp.schema import ( AgentCapabilities, AuthenticateResponse, - AuthMethod, + AvailableCommand, + AvailableCommandsUpdate, ClientCapabilities, EmbeddedResourceContentBlock, ForkSessionResponse, @@ -37,9 +38,16 @@ from acp.schema import ( SessionListCapabilities, SessionInfo, TextContentBlock, + UnstructuredCommandInput, Usage, ) +# AuthMethodAgent was renamed from AuthMethod in agent-client-protocol 0.9.0 +try: + from acp.schema import AuthMethodAgent +except ImportError: + from acp.schema import AuthMethod as AuthMethodAgent # type: ignore[attr-defined] + from acp_adapter.auth import detect_provider, has_provider from acp_adapter.events import ( make_message_cb, @@ -84,6 +92,48 @@ def _extract_text( class HermesACPAgent(acp.Agent): """ACP Agent implementation wrapping Hermes AIAgent.""" + _SLASH_COMMANDS = { + "help": "Show available commands", + "model": "Show or change current model", + "tools": "List available tools", + "context": "Show conversation context info", + "reset": "Clear conversation history", + "compact": "Compress conversation context", + "version": "Show Hermes version", + } + + _ADVERTISED_COMMANDS = ( + { + "name": "help", + "description": "List available commands", + }, + { + "name": "model", + "description": "Show current model and provider, or switch models", + "input_hint": "model name to switch to", + }, + { + "name": "tools", + "description": "List available tools with descriptions", + }, + { + "name": "context", + "description": "Show conversation message counts by role", + }, + { + "name": "reset", + "description": "Clear conversation history", + }, + { + "name": "compact", + "description": "Compress conversation context", + }, + { + "name": "version", + "description": "Show Hermes version", + }, + ) + def __init__(self, session_manager: SessionManager | None = None): super().__init__() self.session_manager = session_manager or SessionManager() @@ -177,7 +227,7 @@ class HermesACPAgent(acp.Agent): auth_methods = None if provider: auth_methods = [ - AuthMethod( + AuthMethodAgent( id=provider, name=f"{provider} runtime credentials", description=f"Authenticate Hermes using the currently configured {provider} runtime credentials.", @@ -219,6 +269,7 @@ class HermesACPAgent(acp.Agent): state = self.session_manager.create_session(cwd=cwd) await self._register_session_mcp_servers(state, mcp_servers) logger.info("New session %s (cwd=%s)", state.session_id, cwd) + self._schedule_available_commands_update(state.session_id) return NewSessionResponse(session_id=state.session_id) async def load_session( @@ -234,6 +285,7 @@ class HermesACPAgent(acp.Agent): return None await self._register_session_mcp_servers(state, mcp_servers) logger.info("Loaded session %s", session_id) + self._schedule_available_commands_update(session_id) return LoadSessionResponse() async def resume_session( @@ -249,6 +301,7 @@ class HermesACPAgent(acp.Agent): state = self.session_manager.create_session(cwd=cwd) await self._register_session_mcp_servers(state, mcp_servers) logger.info("Resumed session %s", state.session_id) + self._schedule_available_commands_update(state.session_id) return ResumeSessionResponse() async def cancel(self, session_id: str, **kwargs: Any) -> None: @@ -274,6 +327,8 @@ class HermesACPAgent(acp.Agent): if state is not None: await self._register_session_mcp_servers(state, mcp_servers) logger.info("Forked session %s -> %s", session_id, new_id) + if new_id: + self._schedule_available_commands_update(new_id) return ForkSessionResponse(session_id=new_id) async def list_sessions( @@ -411,15 +466,50 @@ class HermesACPAgent(acp.Agent): # ---- Slash commands (headless) ------------------------------------------- - _SLASH_COMMANDS = { - "help": "Show available commands", - "model": "Show or change current model", - "tools": "List available tools", - "context": "Show conversation context info", - "reset": "Clear conversation history", - "compact": "Compress conversation context", - "version": "Show Hermes version", - } + @classmethod + def _available_commands(cls) -> list[AvailableCommand]: + commands: list[AvailableCommand] = [] + for spec in cls._ADVERTISED_COMMANDS: + input_hint = spec.get("input_hint") + commands.append( + AvailableCommand( + name=spec["name"], + description=spec["description"], + input=UnstructuredCommandInput(hint=input_hint) + if input_hint + else None, + ) + ) + return commands + + async def _send_available_commands_update(self, session_id: str) -> None: + """Advertise supported slash commands to the connected ACP client.""" + if not self._conn: + return + + try: + await self._conn.session_update( + session_id=session_id, + update=AvailableCommandsUpdate( + sessionUpdate="available_commands_update", + availableCommands=self._available_commands(), + ), + ) + except Exception: + logger.warning( + "Failed to advertise ACP slash commands for session %s", + session_id, + exc_info=True, + ) + + def _schedule_available_commands_update(self, session_id: str) -> None: + """Send the command advertisement after the session response is queued.""" + if not self._conn: + return + loop = asyncio.get_running_loop() + loop.call_soon( + asyncio.create_task, self._send_available_commands_update(session_id) + ) def _handle_slash_command(self, text: str, state: SessionState) -> str | None: """Dispatch a slash command and return the response text. @@ -539,11 +629,39 @@ class HermesACPAgent(acp.Agent): return "Nothing to compress — conversation is empty." try: agent = state.agent - if hasattr(agent, "compress_context"): - agent.compress_context(state.history) - self.session_manager.save_session(state.session_id) - return f"Context compressed. Messages: {len(state.history)}" - return "Context compression not available for this agent." + if not getattr(agent, "compression_enabled", True): + return "Context compression is disabled for this agent." + if not hasattr(agent, "_compress_context"): + return "Context compression not available for this agent." + + from agent.model_metadata import estimate_messages_tokens_rough + + original_count = len(state.history) + approx_tokens = estimate_messages_tokens_rough(state.history) + original_session_db = getattr(agent, "_session_db", None) + + try: + # ACP sessions must keep a stable session id, so avoid the + # SQLite session-splitting side effect inside _compress_context. + agent._session_db = None + compressed, _ = agent._compress_context( + state.history, + getattr(agent, "_cached_system_prompt", "") or "", + approx_tokens=approx_tokens, + task_id=state.session_id, + ) + finally: + agent._session_db = original_session_db + + state.history = compressed + self.session_manager.save_session(state.session_id) + + new_count = len(state.history) + new_tokens = estimate_messages_tokens_rough(state.history) + return ( + f"Context compressed: {original_count} -> {new_count} messages\n" + f"~{approx_tokens:,} -> ~{new_tokens:,} tokens" + ) except Exception as e: return f"Compression failed: {e}" diff --git a/acp_adapter/session.py b/acp_adapter/session.py index f36f8f64d1d6..b489c39841e3 100644 --- a/acp_adapter/session.py +++ b/acp_adapter/session.py @@ -13,6 +13,7 @@ from hermes_constants import get_hermes_home import copy import json import logging +import sys import uuid from dataclasses import dataclass, field from threading import Lock @@ -21,6 +22,17 @@ from typing import Any, Dict, List, Optional logger = logging.getLogger(__name__) +def _acp_stderr_print(*args, **kwargs) -> None: + """Best-effort human-readable output sink for ACP stdio sessions. + + ACP reserves stdout for JSON-RPC frames, so any incidental CLI/status output + from AIAgent must be redirected away from stdout. Route it to stderr instead. + """ + kwargs = dict(kwargs) + kwargs.setdefault("file", sys.stderr) + print(*args, **kwargs) + + def _register_task_cwd(task_id: str, cwd: str) -> None: """Bind a task/session id to the editor's working directory for tools.""" if not task_id: @@ -458,4 +470,8 @@ class SessionManager: logger.debug("ACP session falling back to default provider resolution", exc_info=True) _register_task_cwd(session_id, cwd) - return AIAgent(**kwargs) + agent = AIAgent(**kwargs) + # ACP stdio transport requires stdout to remain protocol-only JSON-RPC. + # Route any incidental human-readable agent output to stderr instead. + agent._print_fn = _acp_stderr_print + return agent diff --git a/agent/credential_pool.py b/agent/credential_pool.py index 2cf9efe56206..311abea98f7f 100644 --- a/agent/credential_pool.py +++ b/agent/credential_pool.py @@ -8,7 +8,9 @@ import threading import time import uuid import os +import re from dataclasses import dataclass, fields, replace +from datetime import datetime, timezone from typing import Any, Dict, List, Optional, Set, Tuple from hermes_constants import OPENROUTER_BASE_URL @@ -95,6 +97,9 @@ class PooledCredential: last_status: Optional[str] = None last_status_at: Optional[float] = None last_error_code: Optional[int] = None + last_error_reason: Optional[str] = None + last_error_message: Optional[str] = None + last_error_reset_at: Optional[float] = None base_url: Optional[str] = None expires_at: Optional[str] = None expires_at_ms: Optional[int] = None @@ -129,7 +134,14 @@ class PooledCredential: return cls(provider=provider, **data) def to_dict(self) -> Dict[str, Any]: - _ALWAYS_EMIT = {"last_status", "last_status_at", "last_error_code"} + _ALWAYS_EMIT = { + "last_status", + "last_status_at", + "last_error_code", + "last_error_reason", + "last_error_message", + "last_error_reset_at", + } result: Dict[str, Any] = {} for field_def in fields(self): if field_def.name in ("provider", "extra"): @@ -180,6 +192,85 @@ def _exhausted_ttl(error_code: Optional[int]) -> int: return EXHAUSTED_TTL_DEFAULT_SECONDS +def _parse_absolute_timestamp(value: Any) -> Optional[float]: + """Best-effort parse for provider reset timestamps. + + Accepts epoch seconds, epoch milliseconds, and ISO-8601 strings. + Returns seconds since epoch. + """ + if value is None or value == "": + return None + if isinstance(value, (int, float)): + numeric = float(value) + if numeric <= 0: + return None + return numeric / 1000.0 if numeric > 1_000_000_000_000 else numeric + if isinstance(value, str): + raw = value.strip() + if not raw: + return None + try: + numeric = float(raw) + except ValueError: + numeric = None + if numeric is not None: + return numeric / 1000.0 if numeric > 1_000_000_000_000 else numeric + try: + return datetime.fromisoformat(raw.replace("Z", "+00:00")).timestamp() + except ValueError: + return None + return None + + +def _extract_retry_delay_seconds(message: str) -> Optional[float]: + if not message: + return None + delay_match = re.search(r"quotaResetDelay[:\s\"]+(\d+(?:\.\d+)?)(ms|s)", message, re.IGNORECASE) + if delay_match: + value = float(delay_match.group(1)) + return value / 1000.0 if delay_match.group(2).lower() == "ms" else value + sec_match = re.search(r"retry\s+(?:after\s+)?(\d+(?:\.\d+)?)\s*(?:sec|secs|seconds|s\b)", message, re.IGNORECASE) + if sec_match: + return float(sec_match.group(1)) + return None + + +def _normalize_error_context(error_context: Optional[Dict[str, Any]]) -> Dict[str, Any]: + if not isinstance(error_context, dict): + return {} + normalized: Dict[str, Any] = {} + reason = error_context.get("reason") + if isinstance(reason, str) and reason.strip(): + normalized["reason"] = reason.strip() + message = error_context.get("message") + if isinstance(message, str) and message.strip(): + normalized["message"] = message.strip() + reset_at = ( + error_context.get("reset_at") + or error_context.get("resets_at") + or error_context.get("retry_until") + ) + parsed_reset_at = _parse_absolute_timestamp(reset_at) + if parsed_reset_at is None and isinstance(message, str): + retry_delay_seconds = _extract_retry_delay_seconds(message) + if retry_delay_seconds is not None: + parsed_reset_at = time.time() + retry_delay_seconds + if parsed_reset_at is not None: + normalized["reset_at"] = parsed_reset_at + return normalized + + +def _exhausted_until(entry: PooledCredential) -> Optional[float]: + if entry.last_status != STATUS_EXHAUSTED: + return None + reset_at = _parse_absolute_timestamp(getattr(entry, "last_error_reset_at", None)) + if reset_at is not None: + return reset_at + if entry.last_status_at: + return entry.last_status_at + _exhausted_ttl(entry.last_error_code) + return None + + def _normalize_custom_pool_name(name: str) -> str: """Normalize a custom provider name for use as a pool key suffix.""" return name.strip().lower().replace(" ", "-") @@ -292,12 +383,21 @@ class CredentialPool: [entry.to_dict() for entry in self._entries], ) - def _mark_exhausted(self, entry: PooledCredential, status_code: Optional[int]) -> PooledCredential: + def _mark_exhausted( + self, + entry: PooledCredential, + status_code: Optional[int], + error_context: Optional[Dict[str, Any]] = None, + ) -> PooledCredential: + normalized_error = _normalize_error_context(error_context) updated = replace( entry, last_status=STATUS_EXHAUSTED, last_status_at=time.time(), last_error_code=status_code, + last_error_reason=normalized_error.get("reason"), + last_error_message=normalized_error.get("message"), + last_error_reset_at=normalized_error.get("reset_at"), ) self._replace_entry(entry, updated) self._persist() @@ -462,7 +562,15 @@ class CredentialPool: self._mark_exhausted(entry, None) return None - updated = replace(updated, last_status=STATUS_OK, last_status_at=None, last_error_code=None) + updated = replace( + updated, + last_status=STATUS_OK, + last_status_at=None, + last_error_code=None, + last_error_reason=None, + last_error_message=None, + last_error_reset_at=None, + ) self._replace_entry(entry, updated) self._persist() return updated @@ -522,11 +630,19 @@ class CredentialPool: entry = synced cleared_any = True if entry.last_status == STATUS_EXHAUSTED: - ttl = _exhausted_ttl(entry.last_error_code) - if entry.last_status_at and now - entry.last_status_at < ttl: + exhausted_until = _exhausted_until(entry) + if exhausted_until is not None and now < exhausted_until: continue if clear_expired: - cleared = replace(entry, last_status=STATUS_OK, last_status_at=None, last_error_code=None) + cleared = replace( + entry, + last_status=STATUS_OK, + last_status_at=None, + last_error_code=None, + last_error_reason=None, + last_error_message=None, + last_error_reset_at=None, + ) self._replace_entry(entry, cleared) entry = cleared cleared_any = True @@ -576,12 +692,17 @@ class CredentialPool: available = self._available_entries() return available[0] if available else None - def mark_exhausted_and_rotate(self, *, status_code: Optional[int]) -> Optional[PooledCredential]: + def mark_exhausted_and_rotate( + self, + *, + status_code: Optional[int], + error_context: Optional[Dict[str, Any]] = None, + ) -> Optional[PooledCredential]: with self._lock: entry = self.current() or self._select_unlocked() if entry is None: return None - self._mark_exhausted(entry, status_code) + self._mark_exhausted(entry, status_code, error_context) self._current_id = None return self._select_unlocked() @@ -603,7 +724,17 @@ class CredentialPool: new_entries = [] for entry in self._entries: if entry.last_status or entry.last_status_at or entry.last_error_code: - new_entries.append(replace(entry, last_status=None, last_status_at=None, last_error_code=None)) + new_entries.append( + replace( + entry, + last_status=None, + last_status_at=None, + last_error_code=None, + last_error_reason=None, + last_error_message=None, + last_error_reset_at=None, + ) + ) count += 1 else: new_entries.append(entry) @@ -625,6 +756,31 @@ class CredentialPool: self._current_id = None return removed + def resolve_target(self, target: Any) -> Tuple[Optional[int], Optional[PooledCredential], Optional[str]]: + raw = str(target or "").strip() + if not raw: + return None, None, "No credential target provided." + + for idx, entry in enumerate(self._entries, start=1): + if entry.id == raw: + return idx, entry, None + + label_matches = [ + (idx, entry) + for idx, entry in enumerate(self._entries, start=1) + if entry.label.strip().lower() == raw.lower() + ] + if len(label_matches) == 1: + return label_matches[0][0], label_matches[0][1], None + if len(label_matches) > 1: + return None, None, f'Ambiguous credential label "{raw}". Use the numeric index or entry id instead.' + if raw.isdigit(): + index = int(raw) + if 1 <= index <= len(self._entries): + return index, self._entries[index - 1], None + return None, None, f"No credential #{index}." + return None, None, f'No credential matching "{raw}".' + def add_entry(self, entry: PooledCredential) -> PooledCredential: entry = replace(entry, priority=_next_priority(self._entries)) self._entries.append(entry) diff --git a/agent/models_dev.py b/agent/models_dev.py index b4b699558442..61483b6a1091 100644 --- a/agent/models_dev.py +++ b/agent/models_dev.py @@ -1,19 +1,31 @@ -"""Models.dev registry integration for provider-aware context length detection. +"""Models.dev registry integration — primary database for providers and models. -Fetches model metadata from https://models.dev/api.json — a community-maintained -database of 3800+ models across 100+ providers, including per-provider context -windows, pricing, and capabilities. +Fetches from https://models.dev/api.json — a community-maintained database +of 4000+ models across 109+ providers. Provides: -Data is cached in memory (1hr TTL) and on disk (~/.hermes/models_dev_cache.json) -to avoid cold-start network latency. +- **Provider metadata**: name, base URL, env vars, documentation link +- **Model metadata**: context window, max output, cost/M tokens, capabilities + (reasoning, tools, vision, PDF, audio), modalities, knowledge cutoff, + open-weights flag, family grouping, deprecation status + +Data resolution order (like TypeScript OpenCode): + 1. Bundled snapshot (ships with the package — offline-first) + 2. Disk cache (~/.hermes/models_dev_cache.json) + 3. Network fetch (https://models.dev/api.json) + 4. Background refresh every 60 minutes + +Other modules should import the dataclasses and query functions from here +rather than parsing the raw JSON themselves. """ +import difflib import json import logging import os import time +from dataclasses import dataclass, field from pathlib import Path -from typing import Any, Dict, Optional +from typing import Any, Dict, List, Optional, Tuple, Union from utils import atomic_json_write @@ -28,7 +40,110 @@ _MODELS_DEV_CACHE_TTL = 3600 # 1 hour in-memory _models_dev_cache: Dict[str, Any] = {} _models_dev_cache_time: float = 0 -# Provider ID mapping: Hermes provider names → models.dev provider IDs + +# --------------------------------------------------------------------------- +# Dataclasses — rich metadata for providers and models +# --------------------------------------------------------------------------- + +@dataclass +class ModelInfo: + """Full metadata for a single model from models.dev.""" + + id: str + name: str + family: str + provider_id: str # models.dev provider ID (e.g. "anthropic") + + # Capabilities + reasoning: bool = False + tool_call: bool = False + attachment: bool = False # supports image/file attachments (vision) + temperature: bool = False + structured_output: bool = False + open_weights: bool = False + + # Modalities + input_modalities: Tuple[str, ...] = () # ("text", "image", "pdf", ...) + output_modalities: Tuple[str, ...] = () + + # Limits + context_window: int = 0 + max_output: int = 0 + max_input: Optional[int] = None + + # Cost (per million tokens, USD) + cost_input: float = 0.0 + cost_output: float = 0.0 + cost_cache_read: Optional[float] = None + cost_cache_write: Optional[float] = None + + # Metadata + knowledge_cutoff: str = "" + release_date: str = "" + status: str = "" # "alpha", "beta", "deprecated", or "" + interleaved: Any = False # True or {"field": "reasoning_content"} + + def has_cost_data(self) -> bool: + return self.cost_input > 0 or self.cost_output > 0 + + def supports_vision(self) -> bool: + return self.attachment or "image" in self.input_modalities + + def supports_pdf(self) -> bool: + return "pdf" in self.input_modalities + + def supports_audio_input(self) -> bool: + return "audio" in self.input_modalities + + def format_cost(self) -> str: + """Human-readable cost string, e.g. '$3.00/M in, $15.00/M out'.""" + if not self.has_cost_data(): + return "unknown" + parts = [f"${self.cost_input:.2f}/M in", f"${self.cost_output:.2f}/M out"] + if self.cost_cache_read is not None: + parts.append(f"cache read ${self.cost_cache_read:.2f}/M") + return ", ".join(parts) + + def format_capabilities(self) -> str: + """Human-readable capabilities, e.g. 'reasoning, tools, vision, PDF'.""" + caps = [] + if self.reasoning: + caps.append("reasoning") + if self.tool_call: + caps.append("tools") + if self.supports_vision(): + caps.append("vision") + if self.supports_pdf(): + caps.append("PDF") + if self.supports_audio_input(): + caps.append("audio") + if self.structured_output: + caps.append("structured output") + if self.open_weights: + caps.append("open weights") + return ", ".join(caps) if caps else "basic" + + +@dataclass +class ProviderInfo: + """Full metadata for a provider from models.dev.""" + + id: str # models.dev provider ID + name: str # display name + env: Tuple[str, ...] # env var names for API key + api: str # base URL + doc: str = "" # documentation URL + model_count: int = 0 + + def has_api_url(self) -> bool: + return bool(self.api) + + +# --------------------------------------------------------------------------- +# Provider ID mapping: Hermes ↔ models.dev +# --------------------------------------------------------------------------- + +# Hermes provider names → models.dev provider IDs PROVIDER_TO_MODELS_DEV: Dict[str, str] = { "openrouter": "openrouter", "anthropic": "anthropic", @@ -44,8 +159,28 @@ PROVIDER_TO_MODELS_DEV: Dict[str, str] = { "opencode-go": "opencode-go", "kilocode": "kilo", "fireworks": "fireworks-ai", + "huggingface": "huggingface", + "google": "google", + "xai": "xai", + "nvidia": "nvidia", + "groq": "groq", + "mistral": "mistral", + "togetherai": "togetherai", + "perplexity": "perplexity", + "cohere": "cohere", } +# Reverse mapping: models.dev → Hermes (built lazily) +_MODELS_DEV_TO_PROVIDER: Optional[Dict[str, str]] = None + + +def _get_reverse_mapping() -> Dict[str, str]: + """Return models.dev ID → Hermes provider ID mapping.""" + global _MODELS_DEV_TO_PROVIDER + if _MODELS_DEV_TO_PROVIDER is None: + _MODELS_DEV_TO_PROVIDER = {v: k for k, v in PROVIDER_TO_MODELS_DEV.items()} + return _MODELS_DEV_TO_PROVIDER + def _get_cache_path() -> Path: """Return path to disk cache file.""" @@ -170,3 +305,443 @@ def _extract_context(entry: Dict[str, Any]) -> Optional[int]: if isinstance(ctx, (int, float)) and ctx > 0: return int(ctx) return None + + +# --------------------------------------------------------------------------- +# Model capability metadata +# --------------------------------------------------------------------------- + + +@dataclass +class ModelCapabilities: + """Structured capability metadata for a model from models.dev.""" + + supports_tools: bool = True + supports_vision: bool = False + supports_reasoning: bool = False + context_window: int = 200000 + max_output_tokens: int = 8192 + model_family: str = "" + + +def _get_provider_models(provider: str) -> Optional[Dict[str, Any]]: + """Resolve a Hermes provider ID to its models dict from models.dev. + + Returns the models dict or None if the provider is unknown or has no data. + """ + mdev_provider_id = PROVIDER_TO_MODELS_DEV.get(provider) + if not mdev_provider_id: + return None + + data = fetch_models_dev() + provider_data = data.get(mdev_provider_id) + if not isinstance(provider_data, dict): + return None + + models = provider_data.get("models", {}) + if not isinstance(models, dict): + return None + + return models + + +def _find_model_entry(models: Dict[str, Any], model: str) -> Optional[Dict[str, Any]]: + """Find a model entry by exact match, then case-insensitive fallback.""" + # Exact match + entry = models.get(model) + if isinstance(entry, dict): + return entry + + # Case-insensitive match + model_lower = model.lower() + for mid, mdata in models.items(): + if mid.lower() == model_lower and isinstance(mdata, dict): + return mdata + + return None + + +def get_model_capabilities(provider: str, model: str) -> Optional[ModelCapabilities]: + """Look up full capability metadata from models.dev cache. + + Uses the existing fetch_models_dev() and PROVIDER_TO_MODELS_DEV mapping. + Returns None if model not found. + + Extracts from model entry fields: + - reasoning (bool) → supports_reasoning + - tool_call (bool) → supports_tools + - attachment (bool) → supports_vision + - limit.context (int) → context_window + - limit.output (int) → max_output_tokens + - family (str) → model_family + """ + models = _get_provider_models(provider) + if models is None: + return None + + entry = _find_model_entry(models, model) + if entry is None: + return None + + # Extract capability flags (default to False if missing) + supports_tools = bool(entry.get("tool_call", False)) + supports_vision = bool(entry.get("attachment", False)) + supports_reasoning = bool(entry.get("reasoning", False)) + + # Extract limits + limit = entry.get("limit", {}) + if not isinstance(limit, dict): + limit = {} + + ctx = limit.get("context") + context_window = int(ctx) if isinstance(ctx, (int, float)) and ctx > 0 else 200000 + + out = limit.get("output") + max_output_tokens = int(out) if isinstance(out, (int, float)) and out > 0 else 8192 + + model_family = entry.get("family", "") or "" + + return ModelCapabilities( + supports_tools=supports_tools, + supports_vision=supports_vision, + supports_reasoning=supports_reasoning, + context_window=context_window, + max_output_tokens=max_output_tokens, + model_family=model_family, + ) + + +def list_provider_models(provider: str) -> List[str]: + """Return all model IDs for a provider from models.dev. + + Returns an empty list if the provider is unknown or has no data. + """ + models = _get_provider_models(provider) + if models is None: + return [] + return list(models.keys()) + + +def search_models_dev( + query: str, provider: str = None, limit: int = 5 +) -> List[Dict[str, Any]]: + """Fuzzy search across models.dev catalog. Returns matching model entries. + + Args: + query: Search string to match against model IDs. + provider: Optional Hermes provider ID to restrict search scope. + If None, searches across all providers in PROVIDER_TO_MODELS_DEV. + limit: Maximum number of results to return. + + Returns: + List of dicts, each containing 'provider', 'model_id', and the full + model 'entry' from models.dev. + """ + data = fetch_models_dev() + if not data: + return [] + + # Build list of (provider_id, model_id, entry) candidates + candidates: List[tuple] = [] + + if provider is not None: + # Search only the specified provider + mdev_provider_id = PROVIDER_TO_MODELS_DEV.get(provider) + if not mdev_provider_id: + return [] + provider_data = data.get(mdev_provider_id, {}) + if isinstance(provider_data, dict): + models = provider_data.get("models", {}) + if isinstance(models, dict): + for mid, mdata in models.items(): + candidates.append((provider, mid, mdata)) + else: + # Search across all mapped providers + for hermes_prov, mdev_prov in PROVIDER_TO_MODELS_DEV.items(): + provider_data = data.get(mdev_prov, {}) + if isinstance(provider_data, dict): + models = provider_data.get("models", {}) + if isinstance(models, dict): + for mid, mdata in models.items(): + candidates.append((hermes_prov, mid, mdata)) + + if not candidates: + return [] + + # Use difflib for fuzzy matching — case-insensitive comparison + model_ids_lower = [c[1].lower() for c in candidates] + query_lower = query.lower() + + # First try exact substring matches (more intuitive than pure edit-distance) + substring_matches = [] + for prov, mid, mdata in candidates: + if query_lower in mid.lower(): + substring_matches.append({"provider": prov, "model_id": mid, "entry": mdata}) + + # Then add difflib fuzzy matches for any remaining slots + fuzzy_ids = difflib.get_close_matches( + query_lower, model_ids_lower, n=limit * 2, cutoff=0.4 + ) + + seen_ids: set = set() + results: List[Dict[str, Any]] = [] + + # Prioritize substring matches + for match in substring_matches: + key = (match["provider"], match["model_id"]) + if key not in seen_ids: + seen_ids.add(key) + results.append(match) + if len(results) >= limit: + return results + + # Add fuzzy matches + for fid in fuzzy_ids: + # Find original-case candidates matching this lowered ID + for prov, mid, mdata in candidates: + if mid.lower() == fid: + key = (prov, mid) + if key not in seen_ids: + seen_ids.add(key) + results.append({"provider": prov, "model_id": mid, "entry": mdata}) + if len(results) >= limit: + return results + + return results + + +# --------------------------------------------------------------------------- +# Rich dataclass constructors — parse raw models.dev JSON into dataclasses +# --------------------------------------------------------------------------- + +def _parse_model_info(model_id: str, raw: Dict[str, Any], provider_id: str) -> ModelInfo: + """Convert a raw models.dev model entry dict into a ModelInfo dataclass.""" + limit = raw.get("limit") or {} + if not isinstance(limit, dict): + limit = {} + + cost = raw.get("cost") or {} + if not isinstance(cost, dict): + cost = {} + + modalities = raw.get("modalities") or {} + if not isinstance(modalities, dict): + modalities = {} + + input_mods = modalities.get("input") or [] + output_mods = modalities.get("output") or [] + + ctx = limit.get("context") + ctx_int = int(ctx) if isinstance(ctx, (int, float)) and ctx > 0 else 0 + out = limit.get("output") + out_int = int(out) if isinstance(out, (int, float)) and out > 0 else 0 + inp = limit.get("input") + inp_int = int(inp) if isinstance(inp, (int, float)) and inp > 0 else None + + return ModelInfo( + id=model_id, + name=raw.get("name", "") or model_id, + family=raw.get("family", "") or "", + provider_id=provider_id, + reasoning=bool(raw.get("reasoning", False)), + tool_call=bool(raw.get("tool_call", False)), + attachment=bool(raw.get("attachment", False)), + temperature=bool(raw.get("temperature", False)), + structured_output=bool(raw.get("structured_output", False)), + open_weights=bool(raw.get("open_weights", False)), + input_modalities=tuple(input_mods) if isinstance(input_mods, list) else (), + output_modalities=tuple(output_mods) if isinstance(output_mods, list) else (), + context_window=ctx_int, + max_output=out_int, + max_input=inp_int, + cost_input=float(cost.get("input", 0) or 0), + cost_output=float(cost.get("output", 0) or 0), + cost_cache_read=float(cost["cache_read"]) if "cache_read" in cost and cost["cache_read"] is not None else None, + cost_cache_write=float(cost["cache_write"]) if "cache_write" in cost and cost["cache_write"] is not None else None, + knowledge_cutoff=raw.get("knowledge", "") or "", + release_date=raw.get("release_date", "") or "", + status=raw.get("status", "") or "", + interleaved=raw.get("interleaved", False), + ) + + +def _parse_provider_info(provider_id: str, raw: Dict[str, Any]) -> ProviderInfo: + """Convert a raw models.dev provider entry dict into a ProviderInfo.""" + env = raw.get("env") or [] + models = raw.get("models") or {} + return ProviderInfo( + id=provider_id, + name=raw.get("name", "") or provider_id, + env=tuple(env) if isinstance(env, list) else (), + api=raw.get("api", "") or "", + doc=raw.get("doc", "") or "", + model_count=len(models) if isinstance(models, dict) else 0, + ) + + +# --------------------------------------------------------------------------- +# Provider-level queries +# --------------------------------------------------------------------------- + +def get_provider_info(provider_id: str) -> Optional[ProviderInfo]: + """Get full provider metadata from models.dev. + + Accepts either a Hermes provider ID (e.g. "kilocode") or a models.dev + ID (e.g. "kilo"). Returns None if the provider is not in the catalog. + """ + # Resolve Hermes ID → models.dev ID + mdev_id = PROVIDER_TO_MODELS_DEV.get(provider_id, provider_id) + + data = fetch_models_dev() + raw = data.get(mdev_id) + if not isinstance(raw, dict): + return None + + return _parse_provider_info(mdev_id, raw) + + +def list_all_providers() -> Dict[str, ProviderInfo]: + """Return all providers from models.dev as {provider_id: ProviderInfo}. + + Returns the full catalog — 109+ providers. For providers that have + a Hermes alias, both the models.dev ID and the Hermes ID are included. + """ + data = fetch_models_dev() + result: Dict[str, ProviderInfo] = {} + + for pid, pdata in data.items(): + if isinstance(pdata, dict): + info = _parse_provider_info(pid, pdata) + result[pid] = info + + return result + + +def get_providers_for_env_var(env_var: str) -> List[str]: + """Reverse lookup: find all providers that use a given env var. + + Useful for auto-detection: "user has ANTHROPIC_API_KEY set, which + providers does that enable?" + + Returns list of models.dev provider IDs. + """ + data = fetch_models_dev() + matches: List[str] = [] + + for pid, pdata in data.items(): + if isinstance(pdata, dict): + env = pdata.get("env", []) + if isinstance(env, list) and env_var in env: + matches.append(pid) + + return matches + + +# --------------------------------------------------------------------------- +# Model-level queries (rich ModelInfo) +# --------------------------------------------------------------------------- + +def get_model_info( + provider_id: str, model_id: str +) -> Optional[ModelInfo]: + """Get full model metadata from models.dev. + + Accepts Hermes or models.dev provider ID. Tries exact match then + case-insensitive fallback. Returns None if not found. + """ + mdev_id = PROVIDER_TO_MODELS_DEV.get(provider_id, provider_id) + + data = fetch_models_dev() + pdata = data.get(mdev_id) + if not isinstance(pdata, dict): + return None + + models = pdata.get("models", {}) + if not isinstance(models, dict): + return None + + # Exact match + raw = models.get(model_id) + if isinstance(raw, dict): + return _parse_model_info(model_id, raw, mdev_id) + + # Case-insensitive fallback + model_lower = model_id.lower() + for mid, mdata in models.items(): + if mid.lower() == model_lower and isinstance(mdata, dict): + return _parse_model_info(mid, mdata, mdev_id) + + return None + + +def get_model_info_any_provider(model_id: str) -> Optional[ModelInfo]: + """Search all providers for a model by ID. + + Useful when you have a full slug like "anthropic/claude-sonnet-4.6" or + a bare name and want to find it anywhere. Checks Hermes-mapped providers + first, then falls back to all models.dev providers. + """ + data = fetch_models_dev() + + # Try Hermes-mapped providers first (more likely what the user wants) + for hermes_id, mdev_id in PROVIDER_TO_MODELS_DEV.items(): + pdata = data.get(mdev_id) + if not isinstance(pdata, dict): + continue + models = pdata.get("models", {}) + if not isinstance(models, dict): + continue + + raw = models.get(model_id) + if isinstance(raw, dict): + return _parse_model_info(model_id, raw, mdev_id) + + # Case-insensitive + model_lower = model_id.lower() + for mid, mdata in models.items(): + if mid.lower() == model_lower and isinstance(mdata, dict): + return _parse_model_info(mid, mdata, mdev_id) + + # Fall back to ALL providers + for pid, pdata in data.items(): + if pid in _get_reverse_mapping(): + continue # already checked + if not isinstance(pdata, dict): + continue + models = pdata.get("models", {}) + if not isinstance(models, dict): + continue + + raw = models.get(model_id) + if isinstance(raw, dict): + return _parse_model_info(model_id, raw, pid) + + return None + + +def list_provider_model_infos(provider_id: str) -> List[ModelInfo]: + """Return all models for a provider as ModelInfo objects. + + Filters out deprecated models by default. + """ + mdev_id = PROVIDER_TO_MODELS_DEV.get(provider_id, provider_id) + + data = fetch_models_dev() + pdata = data.get(mdev_id) + if not isinstance(pdata, dict): + return [] + + models = pdata.get("models", {}) + if not isinstance(models, dict): + return [] + + result: List[ModelInfo] = [] + for mid, mdata in models.items(): + if not isinstance(mdata, dict): + continue + status = mdata.get("status", "") + if status == "deprecated": + continue + result.append(_parse_model_info(mid, mdata, mdev_id)) + + return result diff --git a/agent/skill_commands.py b/agent/skill_commands.py index 8a434ea7990c..d40572d55bbf 100644 --- a/agent/skill_commands.py +++ b/agent/skill_commands.py @@ -217,6 +217,25 @@ def get_skill_commands() -> Dict[str, Dict[str, Any]]: return _skill_commands +def resolve_skill_command_key(command: str) -> Optional[str]: + """Resolve a user-typed /command to its canonical skill_cmds key. + + Skills are always stored with hyphens — ``scan_skill_commands`` normalizes + spaces and underscores to hyphens when building the key. Hyphens and + underscores are treated interchangeably in user input: this matches + ``_check_unavailable_skill`` and accommodates Telegram bot-command names + (which disallow hyphens, so ``/claude-code`` is registered as + ``/claude_code`` and comes back in the underscored form). + + Returns the matching ``/slug`` key from ``get_skill_commands()`` or + ``None`` if no match. + """ + if not command: + return None + cmd_key = f"/{command.replace('_', '-')}" + return cmd_key if cmd_key in get_skill_commands() else None + + def build_skill_invocation_message( cmd_key: str, user_instruction: str = "", diff --git a/agent/subdirectory_hints.py b/agent/subdirectory_hints.py new file mode 100644 index 000000000000..a6ca2adc5109 --- /dev/null +++ b/agent/subdirectory_hints.py @@ -0,0 +1,219 @@ +"""Progressive subdirectory hint discovery. + +As the agent navigates into subdirectories via tool calls (read_file, terminal, +search_files, etc.), this module discovers and loads project context files +(AGENTS.md, CLAUDE.md, .cursorrules) from those directories. Discovered hints +are appended to the tool result so the model gets relevant context at the moment +it starts working in a new area of the codebase. + +This complements the startup context loading in ``prompt_builder.py`` which only +loads from the CWD. Subdirectory hints are discovered lazily and injected into +the conversation without modifying the system prompt (preserving prompt caching). + +Inspired by Block/goose's SubdirectoryHintTracker. +""" + +import logging +import os +import re +import shlex +from pathlib import Path +from typing import Dict, Any, Optional, Set + +from agent.prompt_builder import _scan_context_content + +logger = logging.getLogger(__name__) + +# Context files to look for in subdirectories, in priority order. +# Same filenames as prompt_builder.py but we load ALL found (not first-wins) +# since different subdirectories may use different conventions. +_HINT_FILENAMES = [ + "AGENTS.md", "agents.md", + "CLAUDE.md", "claude.md", + ".cursorrules", +] + +# Maximum chars per hint file to prevent context bloat +_MAX_HINT_CHARS = 8_000 + +# Tool argument keys that typically contain file paths +_PATH_ARG_KEYS = {"path", "file_path", "workdir"} + +# Tools that take shell commands where we should extract paths +_COMMAND_TOOLS = {"terminal"} + +# How many parent directories to walk up when looking for hints. +# Prevents scanning all the way to / for deeply nested paths. +_MAX_ANCESTOR_WALK = 5 + +class SubdirectoryHintTracker: + """Track which directories the agent visits and load hints on first access. + + Usage:: + + tracker = SubdirectoryHintTracker(working_dir="/path/to/project") + + # After each tool call: + hints = tracker.check_tool_call("read_file", {"path": "backend/src/main.py"}) + if hints: + tool_result += hints # append to the tool result string + """ + + def __init__(self, working_dir: Optional[str] = None): + self.working_dir = Path(working_dir or os.getcwd()).resolve() + self._loaded_dirs: Set[Path] = set() + # Pre-mark the working dir as loaded (startup context handles it) + self._loaded_dirs.add(self.working_dir) + + def check_tool_call( + self, + tool_name: str, + tool_args: Dict[str, Any], + ) -> Optional[str]: + """Check tool call arguments for new directories and load any hint files. + + Returns formatted hint text to append to the tool result, or None. + """ + dirs = self._extract_directories(tool_name, tool_args) + if not dirs: + return None + + all_hints = [] + for d in dirs: + hints = self._load_hints_for_directory(d) + if hints: + all_hints.append(hints) + + if not all_hints: + return None + + return "\n\n" + "\n\n".join(all_hints) + + def _extract_directories( + self, tool_name: str, args: Dict[str, Any] + ) -> list: + """Extract directory paths from tool call arguments.""" + candidates: Set[Path] = set() + + # Direct path arguments + for key in _PATH_ARG_KEYS: + val = args.get(key) + if isinstance(val, str) and val.strip(): + self._add_path_candidate(val, candidates) + + # Shell commands — extract path-like tokens + if tool_name in _COMMAND_TOOLS: + cmd = args.get("command", "") + if isinstance(cmd, str): + self._extract_paths_from_command(cmd, candidates) + + return list(candidates) + + def _add_path_candidate(self, raw_path: str, candidates: Set[Path]): + """Resolve a raw path and add its directory + ancestors to candidates. + + Walks up from the resolved directory toward the filesystem root, + stopping at the first directory already in ``_loaded_dirs`` (or after + ``_MAX_ANCESTOR_WALK`` levels). This ensures that reading + ``project/src/main.py`` discovers ``project/AGENTS.md`` even when + ``project/src/`` has no hint files of its own. + """ + try: + p = Path(raw_path).expanduser() + if not p.is_absolute(): + p = self.working_dir / p + p = p.resolve() + # Use parent if it's a file path (has extension or doesn't exist as dir) + if p.suffix or (p.exists() and p.is_file()): + p = p.parent + # Walk up ancestors — stop at already-loaded or root + for _ in range(_MAX_ANCESTOR_WALK): + if p in self._loaded_dirs: + break + if self._is_valid_subdir(p): + candidates.add(p) + parent = p.parent + if parent == p: + break # filesystem root + p = parent + except (OSError, ValueError): + pass + + def _extract_paths_from_command(self, cmd: str, candidates: Set[Path]): + """Extract path-like tokens from a shell command string.""" + try: + tokens = shlex.split(cmd) + except ValueError: + tokens = cmd.split() + + for token in tokens: + # Skip flags + if token.startswith("-"): + continue + # Must look like a path (contains / or .) + if "/" not in token and "." not in token: + continue + # Skip URLs + if token.startswith(("http://", "https://", "git@")): + continue + self._add_path_candidate(token, candidates) + + def _is_valid_subdir(self, path: Path) -> bool: + """Check if path is a valid directory to scan for hints.""" + if not path.is_dir(): + return False + if path in self._loaded_dirs: + return False + return True + + def _load_hints_for_directory(self, directory: Path) -> Optional[str]: + """Load hint files from a directory. Returns formatted text or None.""" + self._loaded_dirs.add(directory) + + found_hints = [] + for filename in _HINT_FILENAMES: + hint_path = directory / filename + if not hint_path.is_file(): + continue + try: + content = hint_path.read_text(encoding="utf-8").strip() + if not content: + continue + # Same security scan as startup context loading + content = _scan_context_content(content, filename) + if len(content) > _MAX_HINT_CHARS: + content = ( + content[:_MAX_HINT_CHARS] + + f"\n\n[...truncated {filename}: {len(content):,} chars total]" + ) + # Best-effort relative path for display + rel_path = str(hint_path) + try: + rel_path = str(hint_path.relative_to(self.working_dir)) + except ValueError: + try: + rel_path = str(hint_path.relative_to(Path.home())) + rel_path = "~/" + rel_path + except ValueError: + pass # keep absolute + found_hints.append((rel_path, content)) + # First match wins per directory (like startup loading) + break + except Exception as exc: + logger.debug("Could not read %s: %s", hint_path, exc) + + if not found_hints: + return None + + sections = [] + for rel_path, content in found_hints: + sections.append( + f"[Subdirectory context discovered: {rel_path}]\n{content}" + ) + + logger.debug( + "Loaded subdirectory hints from %s: %s", + directory, + [h[0] for h in found_hints], + ) + return "\n\n".join(sections) diff --git a/cli-config.yaml.example b/cli-config.yaml.example index f43b90838d23..6b1809273f33 100644 --- a/cli-config.yaml.example +++ b/cli-config.yaml.example @@ -34,6 +34,12 @@ model: # base_url: "http://localhost:1234/v1" # No API key needed — local servers typically ignore auth. # + # For Ollama Cloud (https://ollama.com/pricing): + # provider: "custom" + # base_url: "https://ollama.com/v1" + # Set OLLAMA_API_KEY in .env — automatically picked up when base_url + # points to ollama.com. + # # Can also be overridden with --provider flag or HERMES_INFERENCE_PROVIDER env var. provider: "auto" @@ -789,6 +795,27 @@ display: # skin: default +# ============================================================================= +# Model Aliases — short names for /model command +# ============================================================================= +# Map short aliases to exact (model, provider, base_url) tuples. +# Used by /model tab completion and resolve_alias(). +# Aliases are checked BEFORE the models.dev catalog, so they can route +# to endpoints not in the catalog (e.g. Ollama Cloud, local servers). +# +# model_aliases: +# opus: +# model: claude-opus-4-6 +# provider: anthropic +# qwen: +# model: "qwen3.5:397b" +# provider: custom +# base_url: "https://ollama.com/v1" +# glm: +# model: glm-4.7 +# provider: custom +# base_url: "https://ollama.com/v1" + # ============================================================================= # Privacy # ============================================================================= diff --git a/cli.py b/cli.py index de21d81e502f..ad7127e7cce0 100644 --- a/cli.py +++ b/cli.py @@ -1257,8 +1257,11 @@ class HermesCLI: # Parse and validate toolsets self.enabled_toolsets = toolsets if toolsets and "all" not in toolsets and "*" not in toolsets: - # Validate each toolset - invalid = [t for t in toolsets if not validate_toolset(t)] + # Validate each toolset — MCP server names are added by + # _get_platform_tools() but aren't registered in TOOLSETS yet + # (that happens later in _sync_mcp_toolsets), so exclude them. + mcp_names = set((CLI_CONFIG.get("mcp_servers") or {}).keys()) + invalid = [t for t in toolsets if not validate_toolset(t) and t not in mcp_names] if invalid: self.console.print(f"[bold red]Warning: Unknown toolsets: {', '.join(invalid)}[/]") @@ -3519,6 +3522,181 @@ class HermesCLI: remaining = len(self.conversation_history) print(f" {remaining} message(s) remaining in history.") + def _handle_model_switch(self, cmd_original: str): + """Handle /model command — switch model for this session. + + Supports: + /model — show current model + usage hints + /model — switch for this session only + /model --global — switch and persist to config.yaml + /model --provider — switch provider + model + /model --provider — switch to provider, auto-detect model + """ + from hermes_cli.model_switch import switch_model, parse_model_flags, list_authenticated_providers + from hermes_cli.providers import get_label + + # Parse args from the original command + parts = cmd_original.split(None, 1) # split off '/model' + raw_args = parts[1].strip() if len(parts) > 1 else "" + + # Parse --provider and --global flags + model_input, explicit_provider, persist_global = parse_model_flags(raw_args) + + # No args at all: show available providers + models + if not model_input and not explicit_provider: + model_display = self.model or "unknown" + provider_display = get_label(self.provider) if self.provider else "unknown" + _cprint(f" Current: {model_display} on {provider_display}") + _cprint("") + + # Show authenticated providers with top models + try: + # Load user providers from config + user_provs = None + try: + from hermes_cli.config import load_config + cfg = load_config() + user_provs = cfg.get("providers") + except Exception: + pass + + providers = list_authenticated_providers( + current_provider=self.provider or "", + user_providers=user_provs, + max_models=6, + ) + if providers: + for p in providers: + tag = " (current)" if p["is_current"] else "" + _cprint(f" {p['name']} [--provider {p['slug']}]{tag}:") + if p["models"]: + model_strs = ", ".join(p["models"]) + extra = f" (+{p['total_models'] - len(p['models'])} more)" if p["total_models"] > len(p["models"]) else "" + _cprint(f" {model_strs}{extra}") + elif p.get("api_url"): + _cprint(f" {p['api_url']} (use /model --provider {p['slug']})") + else: + _cprint(f" (no models listed)") + _cprint("") + else: + _cprint(" No authenticated providers found.") + _cprint("") + except Exception: + pass + + # Aliases + from hermes_cli.model_switch import MODEL_ALIASES + alias_list = ", ".join(sorted(MODEL_ALIASES.keys())) + _cprint(f" Aliases: {alias_list}") + _cprint("") + _cprint(" /model switch model") + _cprint(" /model --provider switch provider") + _cprint(" /model --global persist to config") + return + + # Perform the switch + result = switch_model( + raw_input=model_input, + current_provider=self.provider or "", + current_model=self.model or "", + current_base_url=self.base_url or "", + current_api_key=self.api_key or "", + is_global=persist_global, + explicit_provider=explicit_provider, + ) + + if not result.success: + _cprint(f" ✗ {result.error_message}") + return + + # Apply to CLI state. + # Update requested_provider so _ensure_runtime_credentials() doesn't + # overwrite the switch on the next turn (it re-resolves from this). + old_model = self.model + self.model = result.new_model + self.provider = result.target_provider + self.requested_provider = result.target_provider + if result.api_key: + self.api_key = result.api_key + self._explicit_api_key = result.api_key + if result.base_url: + self.base_url = result.base_url + self._explicit_base_url = result.base_url + if result.api_mode: + self.api_mode = result.api_mode + + # Apply to running agent (in-place swap) + if self.agent is not None: + try: + self.agent.switch_model( + new_model=result.new_model, + new_provider=result.target_provider, + api_key=result.api_key, + base_url=result.base_url, + api_mode=result.api_mode, + ) + except Exception as exc: + _cprint(f" ⚠ Agent swap failed ({exc}); change applied to next session.") + + # Store a note to prepend to the next user message so the model + # knows a switch occurred (avoids injecting system messages mid-history + # which breaks providers and prompt caching). + self._pending_model_switch_note = ( + f"[Note: model was just switched from {old_model} to {result.new_model} " + f"via {result.provider_label or result.target_provider}. " + f"Adjust your self-identification accordingly.]" + ) + + # Display confirmation with full metadata + provider_label = result.provider_label or result.target_provider + _cprint(f" ✓ Model switched: {result.new_model}") + _cprint(f" Provider: {provider_label}") + + # Rich metadata from models.dev + mi = result.model_info + if mi: + if mi.context_window: + _cprint(f" Context: {mi.context_window:,} tokens") + if mi.max_output: + _cprint(f" Max output: {mi.max_output:,} tokens") + if mi.has_cost_data(): + _cprint(f" Cost: {mi.format_cost()}") + _cprint(f" Capabilities: {mi.format_capabilities()}") + else: + # Fallback to old context length lookup + try: + from agent.model_metadata import get_model_context_length + ctx = get_model_context_length( + result.new_model, + base_url=result.base_url or self.base_url, + api_key=result.api_key or self.api_key, + provider=result.target_provider, + ) + _cprint(f" Context: {ctx:,} tokens") + except Exception: + pass + + # Cache notice + cache_enabled = ( + ("openrouter" in (result.base_url or "").lower() and "claude" in result.new_model.lower()) + or result.api_mode == "anthropic_messages" + ) + if cache_enabled: + _cprint(" Prompt caching: enabled") + + # Warning from validation + if result.warning_message: + _cprint(f" ⚠ {result.warning_message}") + + # Persistence + if persist_global: + save_config_value("model.name", result.new_model) + if result.provider_changed: + save_config_value("model.provider", result.target_provider) + _cprint(" Saved to config.yaml (--global)") + else: + _cprint(" (session only — add --global to persist)") + def _show_model_and_providers(self): """Show current model + provider and list all authenticated providers. @@ -4134,6 +4312,8 @@ class HermesCLI: self.new_session() elif canonical == "resume": self._handle_resume_command(cmd_original) + elif canonical == "model": + self._handle_model_switch(cmd_original) elif canonical == "provider": self._show_model_and_providers() elif canonical == "prompt": @@ -5277,14 +5457,17 @@ class HermesCLI: # Tool progress callback (audio cues for voice mode) # ==================================================================== - def _on_tool_progress(self, function_name: str, preview: str, function_args: dict): - """Called when a tool starts executing. + def _on_tool_progress(self, event_type: str, function_name: str = None, preview: str = None, function_args: dict = None, **kwargs): + """Called on tool lifecycle events (tool.started, tool.completed, reasoning.available, etc.). Updates the TUI spinner widget so the user can see what the agent is doing during tool execution (fills the gap between thinking spinner and next response). Also plays audio cue in voice mode. """ - if not function_name.startswith("_"): + # Only act on tool.started; ignore tool.completed, reasoning.available, etc. + if event_type != "tool.started": + return + if function_name and not function_name.startswith("_"): from agent.display import get_tool_emoji emoji = get_tool_emoji(function_name) label = preview or function_name @@ -5297,7 +5480,7 @@ class HermesCLI: if not self._voice_mode: return - if function_name.startswith("_"): + if not function_name or function_name.startswith("_"): return try: from tools.voice_mode import play_beep @@ -6184,6 +6367,11 @@ class HermesCLI: def run_agent(): nonlocal result agent_message = _voice_prefix + message if _voice_prefix else message + # Prepend pending model switch note so the model knows about the switch + _msn = getattr(self, '_pending_model_switch_note', None) + if _msn: + agent_message = _msn + "\n\n" + agent_message + self._pending_model_switch_note = None try: result = self.agent.run_conversation( user_message=agent_message, diff --git a/cron/scheduler.py b/cron/scheduler.py index b014799837e3..860980e0e771 100644 --- a/cron/scheduler.py +++ b/cron/scheduler.py @@ -15,7 +15,6 @@ import logging import os import subprocess import sys -import traceback # fcntl is Unix-only; on Windows use msvcrt for file locking try: @@ -35,6 +34,14 @@ from hermes_time import now as _hermes_now logger = logging.getLogger(__name__) +# Valid delivery platforms — used to validate user-supplied platform names +# in cron delivery targets, preventing env var enumeration via crafted names. +_KNOWN_DELIVERY_PLATFORMS = frozenset({ + "telegram", "discord", "slack", "whatsapp", "signal", + "matrix", "mattermost", "homeassistant", "dingtalk", "feishu", + "wecom", "sms", "email", "webhook", +}) + # Add parent directory to path for imports sys.path.insert(0, str(Path(__file__).parent.parent)) @@ -74,34 +81,51 @@ def _resolve_delivery_target(job: dict) -> Optional[dict]: return None if deliver == "origin": - if not origin: - return None - return { - "platform": origin["platform"], - "chat_id": str(origin["chat_id"]), - "thread_id": origin.get("thread_id"), - } + if origin: + return { + "platform": origin["platform"], + "chat_id": str(origin["chat_id"]), + "thread_id": origin.get("thread_id"), + } + # Origin missing (e.g. job created via API/script) — try each + # platform's home channel as a fallback instead of silently dropping. + for platform_name in ("matrix", "telegram", "discord", "slack"): + chat_id = os.getenv(f"{platform_name.upper()}_HOME_CHANNEL", "") + if chat_id: + logger.info( + "Job '%s' has deliver=origin but no origin; falling back to %s home channel", + job.get("name", job.get("id", "?")), + platform_name, + ) + return { + "platform": platform_name, + "chat_id": chat_id, + "thread_id": None, + } + return None if ":" in deliver: platform_name, rest = deliver.split(":", 1) - # Check for thread_id suffix (e.g. "telegram:-1003724596514:17") - if ":" in rest: - chat_id, thread_id = rest.split(":", 1) + platform_key = platform_name.lower() + + from tools.send_message_tool import _parse_target_ref + + parsed_chat_id, parsed_thread_id, is_explicit = _parse_target_ref(platform_key, rest) + if is_explicit: + chat_id, thread_id = parsed_chat_id, parsed_thread_id else: chat_id, thread_id = rest, None # Resolve human-friendly labels like "Alice (dm)" to real IDs. - # send_message(action="list") shows labels with display suffixes - # that aren't valid platform IDs (e.g. WhatsApp JIDs). try: from gateway.channel_directory import resolve_channel_name - target = chat_id - # Strip display suffix like " (dm)" or " (group)" - if target.endswith(")") and " (" in target: - target = target.rsplit(" (", 1)[0].strip() - resolved = resolve_channel_name(platform_name.lower(), target) + resolved = resolve_channel_name(platform_key, chat_id) if resolved: - chat_id = resolved + parsed_chat_id, parsed_thread_id, resolved_is_explicit = _parse_target_ref(platform_key, resolved) + if resolved_is_explicit: + chat_id, thread_id = parsed_chat_id, parsed_thread_id + else: + chat_id = resolved except Exception: pass @@ -119,6 +143,8 @@ def _resolve_delivery_target(job: dict) -> Optional[dict]: "thread_id": origin.get("thread_id"), } + if platform_name.lower() not in _KNOWN_DELIVERY_PLATFORMS: + return None chat_id = os.getenv(f"{platform_name.upper()}_HOME_CHANNEL", "") if not chat_id: return None @@ -130,12 +156,14 @@ def _resolve_delivery_target(job: dict) -> Optional[dict]: } -def _deliver_result(job: dict, content: str) -> None: +def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> None: """ Deliver job output to the configured target (origin chat, specific platform, etc.). - Uses the standalone platform send functions from send_message_tool so delivery - works whether or not the gateway is running. + When ``adapters`` and ``loop`` are provided (gateway is running), tries to + use the live adapter first — this supports E2EE rooms (e.g. Matrix) where + the standalone HTTP path cannot encrypt. Falls back to standalone send if + the adapter path fails or is unavailable. """ target = _resolve_delivery_target(job) if not target: @@ -206,7 +234,33 @@ def _deliver_result(job: dict, content: str) -> None: else: delivery_content = content - # Run the async send in a fresh event loop (safe from any thread) + # Prefer the live adapter when the gateway is running — this supports E2EE + # rooms (e.g. Matrix) where the standalone HTTP path cannot encrypt. + runtime_adapter = (adapters or {}).get(platform) + if runtime_adapter is not None and loop is not None and getattr(loop, "is_running", lambda: False)(): + send_metadata = {"thread_id": thread_id} if thread_id else None + try: + future = asyncio.run_coroutine_threadsafe( + runtime_adapter.send(chat_id, delivery_content, metadata=send_metadata), + loop, + ) + send_result = future.result(timeout=60) + if send_result and not getattr(send_result, "success", True): + err = getattr(send_result, "error", "unknown") + logger.warning( + "Job '%s': live adapter send to %s:%s failed (%s), falling back to standalone", + job["id"], platform_name, chat_id, err, + ) + else: + logger.info("Job '%s': delivered to %s:%s via live adapter", job["id"], platform_name, chat_id) + return + except Exception as e: + logger.warning( + "Job '%s': live adapter delivery to %s:%s failed (%s), falling back to standalone", + job["id"], platform_name, chat_id, e, + ) + + # Standalone path: run the async send in a fresh event loop (safe from any thread) coro = _send_to_platform(platform, pconfig, chat_id, delivery_content, thread_id=thread_id) try: result = asyncio.run(coro) @@ -248,7 +302,13 @@ def _run_job_script(script_path: str) -> tuple[bool, str]: path = Path(script_path).expanduser() if not path.is_absolute(): # Resolve relative paths against HERMES_HOME/scripts/ - path = get_hermes_home() / "scripts" / path + scripts_dir = get_hermes_home() / "scripts" + path = (scripts_dir / path).resolve() + # Guard against path traversal (e.g. "../../etc/passwd") + try: + path.relative_to(scripts_dir.resolve()) + except ValueError: + return False, f"Script path escapes the scripts directory: {script_path!r}" if not path.exists(): return False, f"Script not found: {path}" @@ -274,6 +334,13 @@ def _run_job_script(script_path: str) -> tuple[bool, str]: parts.append(f"stdout:\n{stdout}") return False, "\n".join(parts) + # Redact any secrets that may appear in script output before + # they are injected into the LLM prompt context. + try: + from agent.redact import redact_sensitive_text + stdout = redact_sensitive_text(stdout) + except Exception: + pass return True, stdout except subprocess.TimeoutExpired: @@ -572,7 +639,7 @@ def run_job(job: dict) -> tuple[bool, str, str, Optional[str]]: except Exception as e: error_msg = f"{type(e).__name__}: {str(e)}" - logger.error("Job '%s' failed: %s", job_name, error_msg) + logger.exception("Job '%s' failed: %s", job_name, error_msg) output = f"""# Cron Job: {job_name} (FAILED) @@ -588,8 +655,6 @@ def run_job(job: dict) -> tuple[bool, str, str, Optional[str]]: ``` {error_msg} - -{traceback.format_exc()} ``` """ return False, output, "", error_msg @@ -616,7 +681,7 @@ def run_job(job: dict) -> tuple[bool, str, str, Optional[str]]: logger.debug("Job '%s': failed to close SQLite session store: %s", job_id, e) -def tick(verbose: bool = True) -> int: +def tick(verbose: bool = True, adapters=None, loop=None) -> int: """ Check and run all due jobs. @@ -625,6 +690,8 @@ def tick(verbose: bool = True) -> int: Args: verbose: Whether to print status messages + adapters: Optional dict mapping Platform → live adapter (from gateway) + loop: Optional asyncio event loop (from gateway) for live adapter sends Returns: Number of jobs executed (0 if another tick is already running) @@ -681,7 +748,7 @@ def tick(verbose: bool = True) -> int: if should_deliver: try: - _deliver_result(job, deliver_content) + _deliver_result(job, deliver_content, adapters=adapters, loop=loop) except Exception as de: logger.error("Delivery failed for job %s: %s", job["id"], de) diff --git a/gateway/channel_directory.py b/gateway/channel_directory.py index 235f11f59fde..cdd2ff9a2417 100644 --- a/gateway/channel_directory.py +++ b/gateway/channel_directory.py @@ -18,6 +18,20 @@ logger = logging.getLogger(__name__) DIRECTORY_PATH = get_hermes_home() / "channel_directory.json" +def _normalize_channel_query(value: str) -> str: + return value.lstrip("#").strip().lower() + + +def _channel_target_name(platform_name: str, channel: Dict[str, Any]) -> str: + """Return the human-facing target label shown to users for a channel entry.""" + name = channel["name"] + if platform_name == "discord" and channel.get("guild"): + return f"#{name}" + if platform_name != "discord" and channel.get("type"): + return f"{name} ({channel['type']})" + return name + + def _session_entry_id(origin: Dict[str, Any]) -> Optional[str]: chat_id = origin.get("chat_id") if not chat_id: @@ -188,23 +202,25 @@ def resolve_channel_name(platform_name: str, name: str) -> Optional[str]: if not channels: return None - query = name.lstrip("#").lower() + query = _normalize_channel_query(name) - # 1. Exact name match + # 1. Exact name match, including the display labels shown by send_message(action="list") for ch in channels: - if ch["name"].lower() == query: + if _normalize_channel_query(ch["name"]) == query: + return ch["id"] + if _normalize_channel_query(_channel_target_name(platform_name, ch)) == query: return ch["id"] # 2. Guild-qualified match for Discord ("GuildName/channel") if "/" in query: guild_part, ch_part = query.rsplit("/", 1) for ch in channels: - guild = ch.get("guild", "").lower() - if guild == guild_part and ch["name"].lower() == ch_part: + guild = ch.get("guild", "").strip().lower() + if guild == guild_part and _normalize_channel_query(ch["name"]) == ch_part: return ch["id"] # 3. Partial prefix match (only if unambiguous) - matches = [ch for ch in channels if ch["name"].lower().startswith(query)] + matches = [ch for ch in channels if _normalize_channel_query(ch["name"]).startswith(query)] if len(matches) == 1: return matches[0]["id"] @@ -239,17 +255,16 @@ def format_directory_for_display() -> str: for guild_name, guild_channels in sorted(guilds.items()): lines.append(f"Discord ({guild_name}):") for ch in sorted(guild_channels, key=lambda c: c["name"]): - lines.append(f" discord:#{ch['name']}") + lines.append(f" discord:{_channel_target_name(plat_name, ch)}") if dms: lines.append("Discord (DMs):") for ch in dms: - lines.append(f" discord:{ch['name']}") + lines.append(f" discord:{_channel_target_name(plat_name, ch)}") lines.append("") else: lines.append(f"{plat_name.title()}:") for ch in channels: - type_label = f" ({ch['type']})" if ch.get("type") else "" - lines.append(f" {plat_name}:{ch['name']}{type_label}") + lines.append(f" {plat_name}:{_channel_target_name(plat_name, ch)}") lines.append("") lines.append('Use these as the "target" parameter when sending.') diff --git a/gateway/platforms/api_server.py b/gateway/platforms/api_server.py index 86af84307d65..7ced55c1eb8f 100644 --- a/gateway/platforms/api_server.py +++ b/gateway/platforms/api_server.py @@ -7,6 +7,8 @@ Exposes an HTTP server with endpoints: - GET /v1/responses/{response_id} — Retrieve a stored response - DELETE /v1/responses/{response_id} — Delete a stored response - GET /v1/models — lists hermes-agent as an available model +- POST /v1/runs — start a run, returns run_id immediately (202) +- GET /v1/runs/{run_id}/events — SSE stream of structured lifecycle events - GET /health — health check Any OpenAI-compatible frontend (Open WebUI, LobeChat, LibreChat, @@ -300,6 +302,10 @@ class APIServerAdapter(BasePlatformAdapter): self._runner: Optional["web.AppRunner"] = None self._site: Optional["web.TCPSite"] = None self._response_store = ResponseStore() + # Active run streams: run_id -> asyncio.Queue of SSE event dicts + self._run_streams: Dict[str, "asyncio.Queue[Optional[Dict]]"] = {} + # Creation timestamps for orphaned-run TTL sweep + self._run_streams_created: Dict[str, float] = {} self._session_db: Optional[Any] = None # Lazy-init SessionDB for session continuity @staticmethod @@ -421,6 +427,11 @@ class APIServerAdapter(BasePlatformAdapter): max_iterations = int(os.getenv("HERMES_MAX_ITERATIONS", "90")) + # Load fallback provider chain so the API server platform has the + # same fallback behaviour as Telegram/Discord/Slack (fixes #4954). + from gateway.run import GatewayRunner + fallback_model = GatewayRunner._load_fallback_model() + agent = AIAgent( model=model, **runtime_kwargs, @@ -434,6 +445,7 @@ class APIServerAdapter(BasePlatformAdapter): stream_delta_callback=stream_delta_callback, tool_progress_callback=tool_progress_callback, session_db=self._ensure_session_db(), + fallback_model=fallback_model, ) return agent @@ -962,6 +974,18 @@ class APIServerAdapter(BasePlatformAdapter): resume_job as _cron_resume, trigger_job as _cron_trigger, ) + # Wrap as staticmethod to prevent descriptor binding — these are plain + # module functions, not instance methods. Without this, self._cron_*() + # injects ``self`` as the first positional argument and every call + # raises TypeError. + _cron_list = staticmethod(_cron_list) + _cron_get = staticmethod(_cron_get) + _cron_create = staticmethod(_cron_create) + _cron_update = staticmethod(_cron_update) + _cron_remove = staticmethod(_cron_remove) + _cron_pause = staticmethod(_cron_pause) + _cron_resume = staticmethod(_cron_resume) + _cron_trigger = staticmethod(_cron_trigger) _CRON_AVAILABLE = True except ImportError: pass @@ -1281,6 +1305,236 @@ class APIServerAdapter(BasePlatformAdapter): return await loop.run_in_executor(None, _run) + # ------------------------------------------------------------------ + # /v1/runs — structured event streaming + # ------------------------------------------------------------------ + + _MAX_CONCURRENT_RUNS = 10 # Prevent unbounded resource allocation + _RUN_STREAM_TTL = 300 # seconds before orphaned runs are swept + + def _make_run_event_callback(self, run_id: str, loop: "asyncio.AbstractEventLoop"): + """Return a tool_progress_callback that pushes structured events to the run's SSE queue.""" + def _push(event: Dict[str, Any]) -> None: + q = self._run_streams.get(run_id) + if q is None: + return + try: + loop.call_soon_threadsafe(q.put_nowait, event) + except Exception: + pass + + def _callback(event_type: str, tool_name: str = None, preview: str = None, args=None, **kwargs): + ts = time.time() + if event_type == "tool.started": + _push({ + "event": "tool.started", + "run_id": run_id, + "timestamp": ts, + "tool": tool_name, + "preview": preview, + }) + elif event_type == "tool.completed": + _push({ + "event": "tool.completed", + "run_id": run_id, + "timestamp": ts, + "tool": tool_name, + "duration": round(kwargs.get("duration", 0), 3), + "error": kwargs.get("is_error", False), + }) + elif event_type == "reasoning.available": + _push({ + "event": "reasoning.available", + "run_id": run_id, + "timestamp": ts, + "text": preview or "", + }) + # _thinking and subagent_progress are intentionally not forwarded + + return _callback + + async def _handle_runs(self, request: "web.Request") -> "web.Response": + """POST /v1/runs — start an agent run, return run_id immediately.""" + auth_err = self._check_auth(request) + if auth_err: + return auth_err + + # Enforce concurrency limit + if len(self._run_streams) >= self._MAX_CONCURRENT_RUNS: + return web.json_response( + _openai_error(f"Too many concurrent runs (max {self._MAX_CONCURRENT_RUNS})", code="rate_limit_exceeded"), + status=429, + ) + + try: + body = await request.json() + except Exception: + return web.json_response(_openai_error("Invalid JSON"), status=400) + + raw_input = body.get("input") + if not raw_input: + return web.json_response(_openai_error("Missing 'input' field"), status=400) + + user_message = raw_input if isinstance(raw_input, str) else (raw_input[-1].get("content", "") if isinstance(raw_input, list) else "") + if not user_message: + return web.json_response(_openai_error("No user message found in input"), status=400) + + run_id = f"run_{uuid.uuid4().hex}" + loop = asyncio.get_running_loop() + q: "asyncio.Queue[Optional[Dict]]" = asyncio.Queue() + self._run_streams[run_id] = q + self._run_streams_created[run_id] = time.time() + + event_cb = self._make_run_event_callback(run_id, loop) + + # Also wire stream_delta_callback so message.delta events flow through + def _text_cb(delta: Optional[str]) -> None: + if delta is None: + return + try: + loop.call_soon_threadsafe(q.put_nowait, { + "event": "message.delta", + "run_id": run_id, + "timestamp": time.time(), + "delta": delta, + }) + except Exception: + pass + + instructions = body.get("instructions") + previous_response_id = body.get("previous_response_id") + conversation_history: List[Dict[str, str]] = [] + if previous_response_id: + stored = self._response_store.get(previous_response_id) + if stored: + conversation_history = list(stored.get("conversation_history", [])) + if instructions is None: + instructions = stored.get("instructions") + + session_id = body.get("session_id") or run_id + ephemeral_system_prompt = instructions + + async def _run_and_close(): + try: + agent = self._create_agent( + ephemeral_system_prompt=ephemeral_system_prompt, + session_id=session_id, + stream_delta_callback=_text_cb, + tool_progress_callback=event_cb, + ) + def _run_sync(): + r = agent.run_conversation( + user_message=user_message, + conversation_history=conversation_history, + ) + u = { + "input_tokens": getattr(agent, "session_prompt_tokens", 0) or 0, + "output_tokens": getattr(agent, "session_completion_tokens", 0) or 0, + "total_tokens": getattr(agent, "session_total_tokens", 0) or 0, + } + return r, u + + result, usage = await asyncio.get_running_loop().run_in_executor(None, _run_sync) + final_response = result.get("final_response", "") if isinstance(result, dict) else "" + q.put_nowait({ + "event": "run.completed", + "run_id": run_id, + "timestamp": time.time(), + "output": final_response, + "usage": usage, + }) + except Exception as exc: + logger.exception("[api_server] run %s failed", run_id) + try: + q.put_nowait({ + "event": "run.failed", + "run_id": run_id, + "timestamp": time.time(), + "error": str(exc), + }) + except Exception: + pass + finally: + # Sentinel: signal SSE stream to close + try: + q.put_nowait(None) + except Exception: + pass + + task = asyncio.create_task(_run_and_close()) + try: + self._background_tasks.add(task) + except TypeError: + pass + if hasattr(task, "add_done_callback"): + task.add_done_callback(self._background_tasks.discard) + + return web.json_response({"run_id": run_id, "status": "started"}, status=202) + + async def _handle_run_events(self, request: "web.Request") -> "web.StreamResponse": + """GET /v1/runs/{run_id}/events — SSE stream of structured agent lifecycle events.""" + auth_err = self._check_auth(request) + if auth_err: + return auth_err + + run_id = request.match_info["run_id"] + + # Allow subscribing slightly before the run is registered (race condition window) + for _ in range(20): + if run_id in self._run_streams: + break + await asyncio.sleep(0.05) + else: + return web.json_response(_openai_error(f"Run not found: {run_id}", code="run_not_found"), status=404) + + q = self._run_streams[run_id] + + response = web.StreamResponse( + status=200, + headers={ + "Content-Type": "text/event-stream", + "Cache-Control": "no-cache", + "X-Accel-Buffering": "no", + }, + ) + await response.prepare(request) + + try: + while True: + try: + event = await asyncio.wait_for(q.get(), timeout=30.0) + except asyncio.TimeoutError: + await response.write(b": keepalive\n\n") + continue + if event is None: + # Run finished — send final SSE comment and close + await response.write(b": stream closed\n\n") + break + payload = f"data: {json.dumps(event)}\n\n" + await response.write(payload.encode()) + except Exception as exc: + logger.debug("[api_server] SSE stream error for run %s: %s", run_id, exc) + finally: + self._run_streams.pop(run_id, None) + self._run_streams_created.pop(run_id, None) + + return response + + async def _sweep_orphaned_runs(self) -> None: + """Periodically clean up run streams that were never consumed.""" + while True: + await asyncio.sleep(60) + now = time.time() + stale = [ + run_id + for run_id, created_at in list(self._run_streams_created.items()) + if now - created_at > self._RUN_STREAM_TTL + ] + for run_id in stale: + logger.debug("[api_server] sweeping orphaned run %s", run_id) + self._run_streams.pop(run_id, None) + self._run_streams_created.pop(run_id, None) + # ------------------------------------------------------------------ # BasePlatformAdapter interface # ------------------------------------------------------------------ @@ -1311,6 +1565,17 @@ class APIServerAdapter(BasePlatformAdapter): self._app.router.add_post("/api/jobs/{job_id}/pause", self._handle_pause_job) self._app.router.add_post("/api/jobs/{job_id}/resume", self._handle_resume_job) self._app.router.add_post("/api/jobs/{job_id}/run", self._handle_run_job) + # Structured event streaming + self._app.router.add_post("/v1/runs", self._handle_runs) + self._app.router.add_get("/v1/runs/{run_id}/events", self._handle_run_events) + # Start background sweep to clean up orphaned (unconsumed) run streams + sweep_task = asyncio.create_task(self._sweep_orphaned_runs()) + try: + self._background_tasks.add(sweep_task) + except TypeError: + pass + if hasattr(sweep_task, "add_done_callback"): + sweep_task.add_done_callback(self._background_tasks.discard) # Port conflict detection — fail fast if port is already in use import socket as _socket diff --git a/gateway/platforms/base.py b/gateway/platforms/base.py index 51a50c8cd22d..98ea4a6b63ee 100644 --- a/gateway/platforms/base.py +++ b/gateway/platforms/base.py @@ -377,23 +377,26 @@ 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 + retryable: bool = False # True for transient connection errors — base will retry automatically -# Error substrings that indicate a transient network failure worth retrying +# Error substrings that indicate a transient *connection* failure worth retrying. +# "timeout" / "timed out" / "readtimeout" / "writetimeout" are intentionally +# excluded: a read/write timeout on a non-idempotent call (e.g. send_message) +# means the request may have reached the server — retrying risks duplicate +# delivery. "connecttimeout" is safe because the connection was never +# established. Platforms that know a timeout is safe to retry should set +# SendResult.retryable = True explicitly. _RETRYABLE_ERROR_PATTERNS = ( "connecterror", "connectionerror", "connectionreset", "connectionrefused", - "timeout", - "timed out", + "connecttimeout", "network", "broken pipe", "remotedisconnected", "eoferror", - "readtimeout", - "writetimeout", ) @@ -927,6 +930,18 @@ class BasePlatformAdapter(ABC): lowered = error.lower() return any(pat in lowered for pat in _RETRYABLE_ERROR_PATTERNS) + @staticmethod + def _is_timeout_error(error: Optional[str]) -> bool: + """Return True if the error string indicates a read/write timeout. + + Timeout errors are NOT retryable and should NOT trigger plain-text + fallback — the request may have already been delivered. + """ + if not error: + return False + lowered = error.lower() + return "timed out" in lowered or "readtimeout" in lowered or "writetimeout" in lowered + async def _send_with_retry( self, chat_id: str, @@ -958,6 +973,11 @@ class BasePlatformAdapter(ABC): error_str = result.error or "" is_network = result.retryable or self._is_retryable_error(error_str) + # Timeout errors are not safe to retry (message may have been + # delivered) and not formatting errors — return the failure as-is. + if not is_network and self._is_timeout_error(error_str): + return result + if is_network: # Retry with exponential backoff for transient errors for attempt in range(1, max_retries + 1): @@ -1048,6 +1068,28 @@ class BasePlatformAdapter(ABC): logger.error("[%s] Approval dispatch failed: %s", self.name, e, exc_info=True) return + # /status must also bypass the active-session guard so it always + # returns a system-generated response instead of being queued as + # user text and passed to the agent (#5046). + if cmd == "status": + logger.debug( + "[%s] Status command bypassing active-session guard for %s", + self.name, session_key, + ) + try: + _thread_meta = {"thread_id": event.source.thread_id} if event.source.thread_id else None + response = await self._message_handler(event) + if response: + await self._send_with_retry( + chat_id=event.source.chat_id, + content=response, + reply_to=event.message_id, + metadata=_thread_meta, + ) + except Exception as e: + logger.error("[%s] Status dispatch failed: %s", self.name, e, exc_info=True) + return + # Special case: photo bursts/albums frequently arrive as multiple near- # simultaneous messages. Queue them without interrupting the active run, # then process them immediately after the current task finishes. diff --git a/gateway/platforms/discord.py b/gateway/platforms/discord.py index 21fa69b6eb72..847c2bb9de5f 100644 --- a/gateway/platforms/discord.py +++ b/gateway/platforms/discord.py @@ -502,19 +502,6 @@ class DiscordAdapter(BasePlatformAdapter): 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 - intents.dm_messages = True - 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", "") @@ -524,6 +511,25 @@ class DiscordAdapter(BasePlatformAdapter): if uid.strip() } + # Set up intents. + # Message Content is required for normal text replies. + # Server Members is only needed when the allowlist contains usernames + # that must be resolved to numeric IDs. Requesting privileged intents + # that aren't enabled in the Discord Developer Portal can prevent the + # bot from coming online at all, so avoid requesting members intent + # unless it is actually necessary. + intents = Intents.default() + intents.message_content = True + intents.dm_messages = True + intents.guild_messages = True + intents.members = any(not entry.isdigit() for entry in self._allowed_user_ids) + intents.voice_states = True + + # Create bot + self._client = commands.Bot( + command_prefix="!", # Not really used, we handle raw messages + intents=intents, + ) adapter_self = self # capture for closure # Register event handlers @@ -648,9 +654,23 @@ class DiscordAdapter(BasePlatformAdapter): except asyncio.TimeoutError: logger.error("[%s] Timeout waiting for connection to Discord", self.name, exc_info=True) + 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 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) + 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 return False async def disconnect(self) -> None: @@ -1932,6 +1952,37 @@ class DiscordAdapter(BasePlatformAdapter): except Exception as e: return SendResult(success=False, error=str(e)) + async def send_update_prompt( + self, chat_id: str, prompt: str, default: str = "", + session_key: str = "", + ) -> SendResult: + """Send an interactive button-based update prompt (Yes / No). + + Used by the gateway ``/update`` watcher when ``hermes update --gateway`` + needs user input (stash restore, config migration). + """ + if not self._client or not DISCORD_AVAILABLE: + return SendResult(success=False, error="Not connected") + try: + channel = self._client.get_channel(int(chat_id)) + if not channel: + channel = await self._client.fetch_channel(int(chat_id)) + + default_hint = f" (default: {default})" if default else "" + embed = discord.Embed( + title="⚕ Update Needs Your Input", + description=f"{prompt}{default_hint}", + color=discord.Color.gold(), + ) + view = UpdatePromptView( + session_key=session_key, + allowed_user_ids=self._allowed_user_ids, + ) + msg = await channel.send(embed=embed, view=view) + return SendResult(success=True, message_id=str(msg.id)) + except Exception as e: + return SendResult(success=False, error=str(e)) + def _get_parent_channel_id(self, channel: Any) -> Optional[str]: """Return the parent channel ID for a Discord thread-like channel, if present.""" parent = getattr(channel, "parent", None) @@ -2344,3 +2395,82 @@ if DISCORD_AVAILABLE: self.resolved = True for child in self.children: child.disabled = True + + class UpdatePromptView(discord.ui.View): + """Interactive Yes/No buttons for ``hermes update`` prompts. + + Clicking a button writes the answer to ``.update_response`` so the + detached update process can pick it up. Only authorized users can + click. Times out after 5 minutes (the update process also has a + 5-minute timeout on its side). + """ + + def __init__(self, session_key: str, allowed_user_ids: set): + super().__init__(timeout=300) + self.session_key = session_key + self.allowed_user_ids = allowed_user_ids + self.resolved = False + + def _check_auth(self, interaction: discord.Interaction) -> bool: + if not self.allowed_user_ids: + return True + return str(interaction.user.id) in self.allowed_user_ids + + async def _respond( + self, interaction: discord.Interaction, answer: str, + color: discord.Color, label: str, + ): + if self.resolved: + await interaction.response.send_message( + "Already answered~", ephemeral=True + ) + return + if not self._check_auth(interaction): + await interaction.response.send_message( + "You're not authorized~", ephemeral=True + ) + return + + self.resolved = True + + # Update embed + 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}") + + for child in self.children: + child.disabled = True + await interaction.response.edit_message(embed=embed, view=self) + + # Write response file + try: + from hermes_constants import get_hermes_home + home = get_hermes_home() + response_path = home / ".update_response" + tmp = response_path.with_suffix(".tmp") + tmp.write_text(answer) + tmp.replace(response_path) + logger.info( + "Discord update prompt answered '%s' by %s", + answer, interaction.user.display_name, + ) + except Exception as exc: + logger.error("Failed to write update response: %s", exc) + + @discord.ui.button(label="Yes", style=discord.ButtonStyle.green, emoji="✓") + async def yes_btn( + self, interaction: discord.Interaction, button: discord.ui.Button + ): + await self._respond(interaction, "y", discord.Color.green(), "Yes") + + @discord.ui.button(label="No", style=discord.ButtonStyle.red, emoji="✗") + async def no_btn( + self, interaction: discord.Interaction, button: discord.ui.Button + ): + await self._respond(interaction, "n", discord.Color.red(), "No") + + async def on_timeout(self): + self.resolved = True + for child in self.children: + child.disabled = True diff --git a/gateway/platforms/matrix.py b/gateway/platforms/matrix.py index 4f77e920a87f..35cf72ad414a 100644 --- a/gateway/platforms/matrix.py +++ b/gateway/platforms/matrix.py @@ -10,8 +10,10 @@ Environment variables: MATRIX_USER_ID Full user ID (@bot:server) — required for password login MATRIX_PASSWORD Password (alternative to access token) MATRIX_ENCRYPTION Set "true" to enable E2EE - MATRIX_ALLOWED_USERS Comma-separated Matrix user IDs (@user:server) - MATRIX_HOME_ROOM Room ID for cron/notification delivery + MATRIX_ALLOWED_USERS Comma-separated Matrix user IDs (@user:server) + MATRIX_HOME_ROOM Room ID for cron/notification delivery + MATRIX_REACTIONS Set "false" to disable processing lifecycle reactions + (eyes/checkmark/cross). Default: true MATRIX_REQUIRE_MENTION Require @mention in rooms (default: true) MATRIX_FREE_RESPONSE_ROOMS Comma-separated room IDs exempt from mention requirement MATRIX_AUTO_THREAD Auto-create threads for room messages (default: true) @@ -30,6 +32,8 @@ import time from pathlib import Path from typing import Any, Dict, Optional, Set +from html import escape as _html_escape + from gateway.config import Platform, PlatformConfig from gateway.platforms.base import ( BasePlatformAdapter, @@ -130,6 +134,11 @@ class MatrixAdapter(BasePlatformAdapter): self._bot_participated_threads: set = self._load_participated_threads() self._MAX_TRACKED_THREADS = 500 + # Reactions: configurable via MATRIX_REACTIONS (default: true). + self._reactions_enabled: bool = os.getenv( + "MATRIX_REACTIONS", "true" + ).lower() not in ("false", "0", "no") + def _is_duplicate_event(self, event_id) -> bool: """Return True if this event was already processed. Tracks the ID otherwise.""" if not event_id: @@ -273,8 +282,23 @@ class MatrixAdapter(BasePlatformAdapter): client.add_event_callback(self._on_room_message_media, nio.RoomMessageAudio) client.add_event_callback(self._on_room_message_media, nio.RoomMessageVideo) client.add_event_callback(self._on_room_message_media, nio.RoomMessageFile) + for encrypted_media_cls in ( + getattr(nio, "RoomEncryptedImage", None), + getattr(nio, "RoomEncryptedAudio", None), + getattr(nio, "RoomEncryptedVideo", None), + getattr(nio, "RoomEncryptedFile", None), + ): + if encrypted_media_cls is not None: + client.add_event_callback(self._on_room_message_media, encrypted_media_cls) client.add_event_callback(self._on_invite, nio.InviteMemberEvent) + # Reaction events (m.reaction). + if hasattr(nio, "ReactionEvent"): + client.add_event_callback(self._on_reaction, nio.ReactionEvent) + else: + # Older matrix-nio versions: use UnknownEvent fallback. + client.add_event_callback(self._on_unknown_event, nio.UnknownEvent) + # If E2EE: handle encrypted events. if self._encryption and hasattr(client, "olm"): client.add_event_callback( @@ -604,6 +628,7 @@ class MatrixAdapter(BasePlatformAdapter): io.BytesIO(data), content_type=content_type, filename=filename, + filesize=len(data), ) if not isinstance(resp, nio.UploadResponse): err = getattr(resp, "message", str(resp)) @@ -683,6 +708,13 @@ class MatrixAdapter(BasePlatformAdapter): if isinstance(resp, nio.SyncError): if self._closing: return + err_msg = str(getattr(resp, "message", resp)).lower() + if "m_unknown_token" in err_msg or "m_forbidden" in err_msg or "401" in err_msg: + logger.error( + "Matrix: permanent auth error from sync: %s — stopping sync", + getattr(resp, "message", resp), + ) + return logger.warning( "Matrix: sync returned %s: %s — retrying in 5s", type(resp).__name__, @@ -697,6 +729,12 @@ class MatrixAdapter(BasePlatformAdapter): except Exception as exc: if self._closing: return + # Detect permanent auth/permission failures that will never + # succeed on retry — stop syncing instead of looping forever. + err_str = str(exc).lower() + if "401" in err_str or "403" in err_str or "unauthorized" in err_str or "forbidden" in err_str: + logger.error("Matrix: permanent auth error: %s — stopping sync", exc) + return logger.warning("Matrix: sync error: %s — retrying in 5s", exc) await asyncio.sleep(5) @@ -980,6 +1018,9 @@ class MatrixAdapter(BasePlatformAdapter): if thread_id: self._track_thread(thread_id) + # Acknowledge receipt so the room shows as read (fire-and-forget). + self._background_read_receipt(room.room_id, event.event_id) + await self.handle_message(msg_event) async def _on_room_message_media(self, room: Any, event: Any) -> None: @@ -1011,47 +1052,132 @@ class MatrixAdapter(BasePlatformAdapter): # Use the MIME type from the event's content info when available, # falling back to category-level MIME types for downstream matching # (gateway/run.py checks startswith("image/"), startswith("audio/"), etc.) - content_info = getattr(event, "content", {}) if isinstance(getattr(event, "content", None), dict) else {} - event_mimetype = (content_info.get("info") or {}).get("mimetype", "") + source_content = getattr(event, "source", {}).get("content", {}) + if not isinstance(source_content, dict): + source_content = {} + event_content = getattr(event, "content", {}) + if not isinstance(event_content, dict): + event_content = {} + content_info = event_content.get("info") if isinstance(event_content, dict) else {} + if not isinstance(content_info, dict) or not content_info: + content_info = source_content.get("info", {}) if isinstance(source_content, dict) else {} + event_mimetype = ( + (content_info.get("mimetype") if isinstance(content_info, dict) else None) + or getattr(event, "mimetype", "") + or "" + ) + # For encrypted media, the URL may be in file.url instead of event.url. + file_content = source_content.get("file", {}) if isinstance(source_content, dict) else {} + if not url and isinstance(file_content, dict): + url = file_content.get("url", "") or "" + if url and url.startswith("mxc://"): + http_url = self._mxc_to_http(url) + media_type = "application/octet-stream" msg_type = MessageType.DOCUMENT + + # Safely resolve encrypted media classes — they may not exist on older + # nio versions, and in test environments nio may be mocked (MagicMock + # auto-attributes are not valid types for isinstance). + def _safe_isinstance(obj, cls_name): + cls = getattr(nio, cls_name, None) + if cls is None or not isinstance(cls, type): + return False + return isinstance(obj, cls) + + is_encrypted_image = _safe_isinstance(event, "RoomEncryptedImage") + is_encrypted_audio = _safe_isinstance(event, "RoomEncryptedAudio") + is_encrypted_video = _safe_isinstance(event, "RoomEncryptedVideo") + is_encrypted_file = _safe_isinstance(event, "RoomEncryptedFile") + is_encrypted_media = any((is_encrypted_image, is_encrypted_audio, is_encrypted_video, is_encrypted_file)) is_voice_message = False - - if isinstance(event, nio.RoomMessageImage): + + if isinstance(event, nio.RoomMessageImage) or is_encrypted_image: msg_type = MessageType.PHOTO media_type = event_mimetype or "image/png" - elif isinstance(event, nio.RoomMessageAudio): - # Check for MSC3245 voice flag: org.matrix.msc3245.voice: {} - source_content = getattr(event, "source", {}).get("content", {}) + elif isinstance(event, nio.RoomMessageAudio) or is_encrypted_audio: if source_content.get("org.matrix.msc3245.voice") is not None: is_voice_message = True msg_type = MessageType.VOICE else: msg_type = MessageType.AUDIO media_type = event_mimetype or "audio/ogg" - elif isinstance(event, nio.RoomMessageVideo): + elif isinstance(event, nio.RoomMessageVideo) or is_encrypted_video: msg_type = MessageType.VIDEO media_type = event_mimetype or "video/mp4" elif event_mimetype: media_type = event_mimetype - # For images, download and cache locally so vision tools can access them. - # Matrix MXC URLs require authentication, so direct URL access fails. + # Cache media locally when downstream tools need a real file path: + # - photos (vision tools can't access MXC URLs) + # - voice messages (transcription tools need local files) + # - any encrypted media (HTTP fallback would point at ciphertext) cached_path = None - if msg_type == MessageType.PHOTO and url: + should_cache_locally = ( + msg_type == MessageType.PHOTO or is_voice_message or is_encrypted_media + ) + if should_cache_locally and url: try: - ext_map = { - "image/jpeg": ".jpg", "image/png": ".png", - "image/gif": ".gif", "image/webp": ".webp", - } - ext = ext_map.get(event_mimetype, ".jpg") - download_resp = await self._client.download(url) - if isinstance(download_resp, nio.DownloadResponse): - from gateway.platforms.base import cache_image_from_bytes - cached_path = cache_image_from_bytes(download_resp.body, ext=ext) - logger.info("[Matrix] Cached user image at %s", cached_path) + if is_voice_message: + download_resp = await self._client.download(mxc=url) + else: + download_resp = await self._client.download(url) + file_bytes = getattr(download_resp, "body", None) + if file_bytes is not None: + if is_encrypted_media: + from nio.crypto.attachments import decrypt_attachment + + hashes_value = getattr(event, "hashes", None) + if hashes_value is None and isinstance(file_content, dict): + hashes_value = file_content.get("hashes") + hash_value = hashes_value.get("sha256") if isinstance(hashes_value, dict) else None + + key_value = getattr(event, "key", None) + if key_value is None and isinstance(file_content, dict): + key_value = file_content.get("key") + if isinstance(key_value, dict): + key_value = key_value.get("k") + + iv_value = getattr(event, "iv", None) + if iv_value is None and isinstance(file_content, dict): + iv_value = file_content.get("iv") + + if key_value and hash_value and iv_value: + file_bytes = decrypt_attachment(file_bytes, key_value, hash_value, iv_value) + else: + logger.warning( + "[Matrix] Encrypted media event missing decryption metadata for %s", + event.event_id, + ) + file_bytes = None + + if file_bytes is not None: + from gateway.platforms.base import ( + cache_audio_from_bytes, + cache_document_from_bytes, + cache_image_from_bytes, + ) + + if msg_type == MessageType.PHOTO: + ext_map = { + "image/jpeg": ".jpg", + "image/png": ".png", + "image/gif": ".gif", + "image/webp": ".webp", + } + ext = ext_map.get(media_type, ".jpg") + cached_path = cache_image_from_bytes(file_bytes, ext=ext) + logger.info("[Matrix] Cached user image at %s", cached_path) + elif msg_type in (MessageType.AUDIO, MessageType.VOICE): + ext = Path(body or ("voice.ogg" if is_voice_message else "audio.ogg")).suffix or ".ogg" + cached_path = cache_audio_from_bytes(file_bytes, ext=ext) + else: + filename = body or ( + "video.mp4" if msg_type == MessageType.VIDEO else "document" + ) + cached_path = cache_document_from_bytes(file_bytes, filename) except Exception as e: - logger.warning("[Matrix] Failed to cache image: %s", e) + logger.warning("[Matrix] Failed to cache media: %s", e) is_dm = self._dm_rooms.get(room.room_id, False) if not is_dm and room.member_count == 2: @@ -1059,7 +1185,6 @@ class MatrixAdapter(BasePlatformAdapter): chat_type = "dm" if is_dm else "group" # Thread/reply detection. - source_content = getattr(event, "source", {}).get("content", {}) relates_to = source_content.get("m.relates_to", {}) thread_id = None if relates_to.get("rel_type") == "m.thread": @@ -1089,31 +1214,6 @@ class MatrixAdapter(BasePlatformAdapter): thread_id = event.event_id self._track_thread(thread_id) - # For voice messages, cache audio locally for transcription tools. - # Use the authenticated nio client to download (Matrix requires auth for media). - media_urls = [http_url] if http_url else None - media_types = [media_type] if http_url else None - - if is_voice_message and url and url.startswith("mxc://"): - try: - import nio - from gateway.platforms.base import cache_audio_from_bytes - - resp = await self._client.download(mxc=url) - if isinstance(resp, nio.MemoryDownloadResponse): - # Extract extension from mimetype or default to .ogg - ext = ".ogg" - if media_type and "/" in media_type: - subtype = media_type.split("/")[1] - ext = f".{subtype}" if subtype else ".ogg" - local_path = cache_audio_from_bytes(resp.body, ext) - media_urls = [local_path] - logger.debug("Matrix: cached voice message to %s", local_path) - else: - logger.warning("Matrix: failed to download voice: %s", getattr(resp, "message", resp)) - except Exception as e: - logger.warning("Matrix: failed to cache voice message, using HTTP URL: %s", e) - source = self.build_source( chat_id=room.room_id, chat_type=chat_type, @@ -1122,9 +1222,8 @@ class MatrixAdapter(BasePlatformAdapter): thread_id=thread_id, ) - # Use cached local path for images (voice messages already handled above). - if cached_path: - media_urls = [cached_path] + allow_http_fallback = bool(http_url) and not is_encrypted_media + media_urls = [cached_path] if cached_path else ([http_url] if allow_http_fallback else None) media_types = [media_type] if media_urls else None msg_event = MessageEvent( @@ -1140,6 +1239,9 @@ class MatrixAdapter(BasePlatformAdapter): if thread_id: self._track_thread(thread_id) + # Acknowledge receipt so the room shows as read (fire-and-forget). + self._background_read_receipt(room.room_id, event.event_id) + await self.handle_message(msg_event) async def _on_invite(self, room: Any, event: Any) -> None: @@ -1175,6 +1277,369 @@ class MatrixAdapter(BasePlatformAdapter): except Exception as exc: logger.warning("Matrix: error joining %s: %s", room.room_id, exc) + # ------------------------------------------------------------------ + # Reactions (send, receive, processing lifecycle) + # ------------------------------------------------------------------ + + async def _send_reaction( + self, room_id: str, event_id: str, emoji: str, + ) -> bool: + """Send an emoji reaction to a message in a room.""" + import nio + + if not self._client: + return False + content = { + "m.relates_to": { + "rel_type": "m.annotation", + "event_id": event_id, + "key": emoji, + } + } + try: + resp = await self._client.room_send( + room_id, "m.reaction", content, + ignore_unverified_devices=True, + ) + if isinstance(resp, nio.RoomSendResponse): + logger.debug("Matrix: sent reaction %s to %s", emoji, event_id) + return True + logger.debug("Matrix: reaction send failed: %s", resp) + return False + except Exception as exc: + logger.debug("Matrix: reaction send error: %s", exc) + return False + + async def _redact_reaction( + self, room_id: str, reaction_event_id: str, reason: str = "", + ) -> bool: + """Remove a reaction by redacting its event.""" + return await self.redact_message(room_id, reaction_event_id, reason) + + async def on_processing_start(self, event: MessageEvent) -> None: + """Add eyes reaction when the agent starts processing a message.""" + if not self._reactions_enabled: + return + msg_id = event.message_id + room_id = event.source.chat_id + if msg_id and room_id: + await self._send_reaction(room_id, msg_id, "\U0001f440") + + async def on_processing_complete( + self, event: MessageEvent, success: bool, + ) -> None: + """Replace eyes with checkmark (success) or cross (failure).""" + if not self._reactions_enabled: + return + msg_id = event.message_id + room_id = event.source.chat_id + if not msg_id or not room_id: + return + # Note: Matrix doesn't support removing a specific reaction easily + # without tracking the reaction event_id. We send the new reaction; + # the eyes stays (acceptable UX — both are visible). + await self._send_reaction( + room_id, msg_id, "\u2705" if success else "\u274c", + ) + + async def _on_reaction(self, room: Any, event: Any) -> None: + """Handle incoming reaction events.""" + if event.sender == self._user_id: + return + if self._is_duplicate_event(getattr(event, "event_id", None)): + return + # Log for now; future: trigger agent actions based on emoji. + reacts_to = getattr(event, "reacts_to", "") + key = getattr(event, "key", "") + logger.info( + "Matrix: reaction %s from %s on %s in %s", + key, event.sender, reacts_to, room.room_id, + ) + + async def _on_unknown_event(self, room: Any, event: Any) -> None: + """Fallback handler for events not natively parsed by matrix-nio. + + Catches m.reaction on older nio versions that lack ReactionEvent. + """ + source = getattr(event, "source", {}) + if source.get("type") != "m.reaction": + return + content = source.get("content", {}) + relates_to = content.get("m.relates_to", {}) + if relates_to.get("rel_type") != "m.annotation": + return + if source.get("sender") == self._user_id: + return + logger.info( + "Matrix: reaction %s from %s on %s in %s", + relates_to.get("key", "?"), + source.get("sender", "?"), + relates_to.get("event_id", "?"), + room.room_id, + ) + + # ------------------------------------------------------------------ + # Read receipts + # ------------------------------------------------------------------ + + def _background_read_receipt(self, room_id: str, event_id: str) -> None: + """Fire-and-forget read receipt with error logging.""" + async def _send() -> None: + try: + await self.send_read_receipt(room_id, event_id) + except Exception as exc: # pragma: no cover — defensive + logger.debug("Matrix: background read receipt failed: %s", exc) + asyncio.ensure_future(_send()) + + async def send_read_receipt(self, room_id: str, event_id: str) -> bool: + """Send a read receipt (m.read) for an event. + + Also sets the fully-read marker so the room is marked as read + in all clients. + """ + if not self._client: + return False + try: + if hasattr(self._client, "room_read_markers"): + await self._client.room_read_markers( + room_id, + fully_read_event=event_id, + read_event=event_id, + ) + else: + # Fallback for older matrix-nio. + await self._client.room_send( + room_id, "m.receipt", {"event_id": event_id}, + ) + logger.debug("Matrix: sent read receipt for %s in %s", event_id, room_id) + return True + except Exception as exc: + logger.debug("Matrix: read receipt failed: %s", exc) + return False + + # ------------------------------------------------------------------ + # Message redaction + # ------------------------------------------------------------------ + + async def redact_message( + self, room_id: str, event_id: str, reason: str = "", + ) -> bool: + """Redact (delete) a message or event from a room.""" + import nio + + if not self._client: + return False + try: + resp = await self._client.room_redact( + room_id, event_id, reason=reason, + ) + if isinstance(resp, nio.RoomRedactResponse): + logger.info("Matrix: redacted %s in %s", event_id, room_id) + return True + logger.warning("Matrix: redact failed: %s", resp) + return False + except Exception as exc: + logger.warning("Matrix: redact error: %s", exc) + return False + + # ------------------------------------------------------------------ + # Room history + # ------------------------------------------------------------------ + + async def fetch_room_history( + self, + room_id: str, + limit: int = 50, + start: str = "", + ) -> list: + """Fetch recent messages from a room. + + Returns a list of dicts with keys: event_id, sender, body, + timestamp, type. Uses the ``room_messages()`` API. + """ + import nio + + if not self._client: + return [] + try: + resp = await self._client.room_messages( + room_id, + start=start or "", + limit=limit, + direction=nio.Api.MessageDirection.back + if hasattr(nio.Api, "MessageDirection") + else "b", + ) + except Exception as exc: + logger.warning("Matrix: room_messages failed for %s: %s", room_id, exc) + return [] + + if not isinstance(resp, nio.RoomMessagesResponse): + logger.warning("Matrix: room_messages returned %s", type(resp).__name__) + return [] + + messages = [] + for event in reversed(resp.chunk): + body = getattr(event, "body", "") or "" + messages.append({ + "event_id": getattr(event, "event_id", ""), + "sender": getattr(event, "sender", ""), + "body": body, + "timestamp": getattr(event, "server_timestamp", 0), + "type": type(event).__name__, + }) + return messages + + # ------------------------------------------------------------------ + # Room creation & management + # ------------------------------------------------------------------ + + async def create_room( + self, + name: str = "", + topic: str = "", + invite: Optional[list] = None, + is_direct: bool = False, + preset: str = "private_chat", + ) -> Optional[str]: + """Create a new Matrix room. + + Args: + name: Human-readable room name. + topic: Room topic. + invite: List of user IDs to invite. + is_direct: Mark as a DM room. + preset: One of private_chat, public_chat, trusted_private_chat. + + Returns the room_id on success, None on failure. + """ + import nio + + if not self._client: + return None + try: + resp = await self._client.room_create( + name=name or None, + topic=topic or None, + invite=invite or [], + is_direct=is_direct, + preset=getattr( + nio.Api.RoomPreset if hasattr(nio.Api, "RoomPreset") else type("", (), {}), + preset, None, + ) or preset, + ) + if isinstance(resp, nio.RoomCreateResponse): + room_id = resp.room_id + self._joined_rooms.add(room_id) + logger.info("Matrix: created room %s (%s)", room_id, name or "unnamed") + return room_id + logger.warning("Matrix: room_create failed: %s", resp) + return None + except Exception as exc: + logger.warning("Matrix: room_create error: %s", exc) + return None + + async def invite_user(self, room_id: str, user_id: str) -> bool: + """Invite a user to a room.""" + import nio + + if not self._client: + return False + try: + resp = await self._client.room_invite(room_id, user_id) + if isinstance(resp, nio.RoomInviteResponse): + logger.info("Matrix: invited %s to %s", user_id, room_id) + return True + logger.warning("Matrix: invite failed: %s", resp) + return False + except Exception as exc: + logger.warning("Matrix: invite error: %s", exc) + return False + + # ------------------------------------------------------------------ + # Presence + # ------------------------------------------------------------------ + + _VALID_PRESENCE_STATES = frozenset(("online", "offline", "unavailable")) + + async def set_presence(self, state: str = "online", status_msg: str = "") -> bool: + """Set the bot's presence status.""" + if not self._client: + return False + if state not in self._VALID_PRESENCE_STATES: + logger.warning("Matrix: invalid presence state %r", state) + return False + try: + if hasattr(self._client, "set_presence"): + await self._client.set_presence(state, status_msg=status_msg or None) + logger.debug("Matrix: presence set to %s", state) + return True + except Exception as exc: + logger.debug("Matrix: set_presence failed: %s", exc) + return False + + # ------------------------------------------------------------------ + # Emote & notice message types + # ------------------------------------------------------------------ + + async def send_emote( + self, chat_id: str, text: str, metadata: Optional[Dict[str, Any]] = None, + ) -> SendResult: + """Send an emote message (/me style action).""" + import nio + + if not self._client or not text: + return SendResult(success=False, error="No client or empty text") + + msg_content: Dict[str, Any] = { + "msgtype": "m.emote", + "body": text, + } + html = self._markdown_to_html(text) + if html and html != text: + msg_content["format"] = "org.matrix.custom.html" + msg_content["formatted_body"] = html + + try: + resp = await self._client.room_send( + chat_id, "m.room.message", msg_content, + ignore_unverified_devices=True, + ) + if isinstance(resp, nio.RoomSendResponse): + return SendResult(success=True, message_id=resp.event_id) + return SendResult(success=False, error=str(resp)) + except Exception as exc: + return SendResult(success=False, error=str(exc)) + + async def send_notice( + self, chat_id: str, text: str, metadata: Optional[Dict[str, Any]] = None, + ) -> SendResult: + """Send a notice message (bot-appropriate, non-alerting).""" + import nio + + if not self._client or not text: + return SendResult(success=False, error="No client or empty text") + + msg_content: Dict[str, Any] = { + "msgtype": "m.notice", + "body": text, + } + html = self._markdown_to_html(text) + if html and html != text: + msg_content["format"] = "org.matrix.custom.html" + msg_content["formatted_body"] = html + + try: + resp = await self._client.room_send( + chat_id, "m.room.message", msg_content, + ignore_unverified_devices=True, + ) + if isinstance(resp, nio.RoomSendResponse): + return SendResult(success=True, message_id=resp.event_id) + return SendResult(success=False, error=str(resp)) + except Exception as exc: + return SendResult(success=False, error=str(exc)) + # ------------------------------------------------------------------ # Helpers # ------------------------------------------------------------------ @@ -1326,29 +1791,196 @@ class MatrixAdapter(BasePlatformAdapter): return f"{self._homeserver}/_matrix/client/v1/media/download/{parts}" def _markdown_to_html(self, text: str) -> str: - """Convert Markdown to Matrix-compatible HTML. + """Convert Markdown to Matrix-compatible HTML (org.matrix.custom.html). - Uses a simple conversion for common patterns. For full fidelity - a markdown-it style library could be used, but this covers the - common cases without an extra dependency. + Uses the ``markdown`` library when available (installed with the + ``matrix`` extra). Falls back to a comprehensive regex converter + that handles fenced code blocks, inline code, headers, bold, + italic, strikethrough, links, blockquotes, lists, and horizontal + rules — everything the Matrix HTML spec allows. """ try: - import markdown - html = markdown.markdown( - text, - extensions=["fenced_code", "tables", "nl2br"], + import markdown as _md + + md = _md.Markdown( + extensions=["fenced_code", "tables", "nl2br", "sane_lists"], ) - # Strip wrapping

tags for single-paragraph messages. + # Remove the raw HTML preprocessor so ") + assert "") + assert "") + assert "") + assert "*") + assert "\n```') + assert "<script>" in result + assert "