"""

Gateway runner - entry point for messaging platform integrations.

This module provides:

- start_gateway(): Start all configured platform adapters

- GatewayRunner: Main class managing the gateway lifecycle

Usage:

# Start the gateway

python -m gateway.run

# Or from CLI

python cli.py --gateway

"""

import asyncio

import logging

import os

import re

import sys

import signal

import threading

from logging.handlers import RotatingFileHandler

from pathlib import Path

from datetime import datetime

from typing import Dict, Optional, Any, List

# Add parent directory to path

sys.path.insert(0, str(Path(__file__).parent.parent))

# Resolve Hermes home directory (respects HERMES_HOME override)

_hermes_home = Path(os.getenv("HERMES_HOME", Path.home() / ".hermes"))

# Load environment variables from ~/.hermes/.env first

from dotenv import load_dotenv

_env_path = _hermes_home / '.env'

if _env_path.exists():

try:

load_dotenv(_env_path, encoding="utf-8")

except UnicodeDecodeError:

load_dotenv(_env_path, encoding="latin-1")

# Also try project .env as fallback

load_dotenv()

# Bridge config.yaml values into the environment so os.getenv() picks them up.

# config.yaml is authoritative for terminal settings — overrides .env.

_config_path = _hermes_home / 'config.yaml'

if _config_path.exists():

try:

import yaml as _yaml

with open(_config_path) as _f:

_cfg = _yaml.safe_load(_f) or {}

# Top-level simple values (fallback only — don't override .env)

for _key, _val in _cfg.items():

if isinstance(_val, (str, int, float, bool)) and _key not in os.environ:

os.environ[_key] = str(_val)

# Terminal config is nested — bridge to TERMINAL_* env vars.

# config.yaml overrides .env for these since it's the documented config path.

_terminal_cfg = _cfg.get("terminal", {})

if _terminal_cfg and isinstance(_terminal_cfg, dict):

_terminal_env_map = {

"backend": "TERMINAL_ENV",

"cwd": "TERMINAL_CWD",

"timeout": "TERMINAL_TIMEOUT",

"lifetime_seconds": "TERMINAL_LIFETIME_SECONDS",

"docker_image": "TERMINAL_DOCKER_IMAGE",

"singularity_image": "TERMINAL_SINGULARITY_IMAGE",

"modal_image": "TERMINAL_MODAL_IMAGE",

"ssh_host": "TERMINAL_SSH_HOST",

"ssh_user": "TERMINAL_SSH_USER",

"ssh_port": "TERMINAL_SSH_PORT",

"ssh_key": "TERMINAL_SSH_KEY",

"container_cpu": "TERMINAL_CONTAINER_CPU",

"container_memory": "TERMINAL_CONTAINER_MEMORY",

"container_disk": "TERMINAL_CONTAINER_DISK",

"container_persistent": "TERMINAL_CONTAINER_PERSISTENT",

}

for _cfg_key, _env_var in _terminal_env_map.items():

if _cfg_key in _terminal_cfg:

os.environ[_env_var] = str(_terminal_cfg[_cfg_key])

_compression_cfg = _cfg.get("compression", {})

if _compression_cfg and isinstance(_compression_cfg, dict):

_compression_env_map = {

"enabled": "CONTEXT_COMPRESSION_ENABLED",

"threshold": "CONTEXT_COMPRESSION_THRESHOLD",

"summary_model": "CONTEXT_COMPRESSION_MODEL",

}

for _cfg_key, _env_var in _compression_env_map.items():

if _cfg_key in _compression_cfg:

os.environ[_env_var] = str(_compression_cfg[_cfg_key])

_agent_cfg = _cfg.get("agent", {})

if _agent_cfg and isinstance(_agent_cfg, dict):

if "max_turns" in _agent_cfg:

os.environ["HERMES_MAX_ITERATIONS"] = str(_agent_cfg["max_turns"])

except Exception:

pass # Non-fatal; gateway can still run with .env values

# Gateway runs in quiet mode - suppress debug output and use cwd directly (no temp dirs)

os.environ["HERMES_QUIET"] = "1"

# Enable interactive exec approval for dangerous commands on messaging platforms

os.environ["HERMES_EXEC_ASK"] = "1"

# Set terminal working directory for messaging platforms

# Uses MESSAGING_CWD if set, otherwise defaults to home directory

# This is separate from CLI which uses the directory where `hermes` is run

messaging_cwd = os.getenv("MESSAGING_CWD") or str(Path.home())

os.environ["TERMINAL_CWD"] = messaging_cwd

from gateway.config import (

Platform,

GatewayConfig,

load_gateway_config,

)

from gateway.session import (

SessionStore,

SessionSource,

SessionContext,

build_session_context,

build_session_context_prompt,

)

from gateway.delivery import DeliveryRouter, DeliveryTarget

from gateway.platforms.base import BasePlatformAdapter, MessageEvent, MessageType

logger = logging.getLogger(__name__)

def _resolve_runtime_agent_kwargs() -> dict:

"""Resolve provider credentials for gateway-created AIAgent instances."""

from hermes_cli.runtime_provider import (

resolve_runtime_provider,

format_runtime_provider_error,

)

try:

runtime = resolve_runtime_provider(

requested=os.getenv("HERMES_INFERENCE_PROVIDER"),

)

except Exception as exc:

raise RuntimeError(format_runtime_provider_error(exc)) from exc

return {

"api_key": runtime.get("api_key"),

"base_url": runtime.get("base_url"),

"provider": runtime.get("provider"),

"api_mode": runtime.get("api_mode"),

}

class GatewayRunner:

"""

Main gateway controller.

Manages the lifecycle of all platform adapters and routes

messages to/from the agent.

"""

def __init__(self, config: Optional[GatewayConfig] = None):

self.config = config or load_gateway_config()

self.adapters: Dict[Platform, BasePlatformAdapter] = {}

# Load ephemeral config from config.yaml / env vars.

# Both are injected at API-call time only and never persisted.

self._prefill_messages = self._load_prefill_messages()

self._ephemeral_system_prompt = self._load_ephemeral_system_prompt()

self._reasoning_config = self._load_reasoning_config()

self._provider_routing = self._load_provider_routing()

# Wire process registry into session store for reset protection

from tools.process_registry import process_registry

self.session_store = SessionStore(

self.config.sessions_dir, self.config,

has_active_processes_fn=lambda key: process_registry.has_active_for_session(key),

on_auto_reset=self._flush_memories_before_reset,

)

self.delivery_router = DeliveryRouter(self.config)

self._running = False

self._shutdown_event = asyncio.Event()

# Track running agents per session for interrupt support

# Key: session_key, Value: AIAgent instance

self._running_agents: Dict[str, Any] = {}

self._pending_messages: Dict[str, str] = {} # Queued messages during interrupt

# Track pending exec approvals per session

# Key: session_key, Value: {"command": str, "pattern_key": str}

self._pending_approvals: Dict[str, Dict[str, str]] = {}

# Initialize session database for session_search tool support

self._session_db = None

try:

from hermes_state import SessionDB

self._session_db = SessionDB()

except Exception as e:

logger.debug("SQLite session store not available: %s", e)

# DM pairing store for code-based user authorization

from gateway.pairing import PairingStore

self.pairing_store = PairingStore()

# Event hook system

from gateway.hooks import HookRegistry

self.hooks = HookRegistry()

def _flush_memories_before_reset(self, old_entry):

"""Prompt the agent to save memories/skills before an auto-reset.

Called synchronously by SessionStore before destroying an expired session.

Loads the transcript, gives the agent a real turn with memory + skills

tools, and explicitly asks it to preserve anything worth keeping.

"""

try:

history = self.session_store.load_transcript(old_entry.session_id)

if not history or len(history) < 4:

return

from run_agent import AIAgent

runtime_kwargs = _resolve_runtime_agent_kwargs()

if not runtime_kwargs.get("api_key"):

return

tmp_agent = AIAgent(

**runtime_kwargs,

max_iterations=8,

quiet_mode=True,

enabled_toolsets=["memory", "skills"],

session_id=old_entry.session_id,

)

# Build conversation history from transcript

msgs = [

{"role": m.get("role"), "content": m.get("content")}

for m in history

if m.get("role") in ("user", "assistant") and m.get("content")

]

# Give the agent a real turn to think about what to save

flush_prompt = (

"[System: This session is about to be automatically reset due to "

"inactivity or a scheduled daily reset. The conversation context "

"will be cleared after this turn.\n\n"

"Review the conversation above and:\n"

"1. Save any important facts, preferences, or decisions to memory "

"(user profile or your notes) that would be useful in future sessions.\n"

"2. If you discovered a reusable workflow or solved a non-trivial "

"problem, consider saving it as a skill.\n"

"3. If nothing is worth saving, that's fine — just skip.\n\n"

"Do NOT respond to the user. Just use the memory and skill_manage "

"tools if needed, then stop.]"

)

tmp_agent.run_conversation(

user_message=flush_prompt,

conversation_history=msgs,

)

logger.info("Pre-reset save completed for session %s", old_entry.session_id)

except Exception as e:

logger.debug("Pre-reset save failed for session %s: %s", old_entry.session_id, e)

@staticmethod

def _load_prefill_messages() -> List[Dict[str, Any]]:

"""Load ephemeral prefill messages from config or env var.

Checks HERMES_PREFILL_MESSAGES_FILE env var first, then falls back to

the prefill_messages_file key in ~/.hermes/config.yaml.

Relative paths are resolved from ~/.hermes/.

"""

import json as _json

file_path = os.getenv("HERMES_PREFILL_MESSAGES_FILE", "")

if not file_path:

try:

import yaml as _y

cfg_path = _hermes_home / "config.yaml"

if cfg_path.exists():

with open(cfg_path) as _f:

cfg = _y.safe_load(_f) or {}

file_path = cfg.get("prefill_messages_file", "")

except Exception:

pass

if not file_path:

return []

path = Path(file_path).expanduser()

if not path.is_absolute():

path = _hermes_home / path

if not path.exists():

logger.warning("Prefill messages file not found: %s", path)

return []

try:

with open(path, "r", encoding="utf-8") as f:

data = _json.load(f)

if not isinstance(data, list):

logger.warning("Prefill messages file must contain a JSON array: %s", path)

return []

return data

except Exception as e:

logger.warning("Failed to load prefill messages from %s: %s", path, e)

return []

@staticmethod

def _load_ephemeral_system_prompt() -> str:

"""Load ephemeral system prompt from config or env var.

Checks HERMES_EPHEMERAL_SYSTEM_PROMPT env var first, then falls back to

agent.system_prompt in ~/.hermes/config.yaml.

"""

prompt = os.getenv("HERMES_EPHEMERAL_SYSTEM_PROMPT", "")

if prompt:

return prompt

try:

import yaml as _y

cfg_path = _hermes_home / "config.yaml"

if cfg_path.exists():

with open(cfg_path) as _f:

cfg = _y.safe_load(_f) or {}

return (cfg.get("agent", {}).get("system_prompt", "") or "").strip()

except Exception:

pass

return ""

@staticmethod

def _load_reasoning_config() -> dict | None:

"""Load reasoning effort from config or env var.

Checks HERMES_REASONING_EFFORT env var first, then agent.reasoning_effort

in config.yaml. Valid: "xhigh", "high", "medium", "low", "minimal", "none".

Returns None to use default (xhigh).

"""

effort = os.getenv("HERMES_REASONING_EFFORT", "")

if not effort:

try:

import yaml as _y

cfg_path = _hermes_home / "config.yaml"

if cfg_path.exists():

with open(cfg_path) as _f:

cfg = _y.safe_load(_f) or {}

effort = str(cfg.get("agent", {}).get("reasoning_effort", "") or "").strip()

except Exception:

pass

if not effort:

return None

effort = effort.lower().strip()

if effort == "none":

return {"enabled": False}

valid = ("xhigh", "high", "medium", "low", "minimal")

if effort in valid:

return {"enabled": True, "effort": effort}

logger.warning("Unknown reasoning_effort '%s', using default (xhigh)", effort)

return None

@staticmethod

def _load_provider_routing() -> dict:

"""Load OpenRouter provider routing preferences from config.yaml."""

try:

import yaml as _y

cfg_path = _hermes_home / "config.yaml"

if cfg_path.exists():

with open(cfg_path) as _f:

cfg = _y.safe_load(_f) or {}

return cfg.get("provider_routing", {}) or {}

except Exception:

pass

return {}

async def start(self) -> bool:

"""

Start the gateway and all configured platform adapters.

Returns True if at least one adapter connected successfully.

"""

logger.info("Starting Hermes Gateway...")

logger.info("Session storage: %s", self.config.sessions_dir)

# Warn if no user allowlists are configured and open access is not opted in

_any_allowlist = any(

os.getenv(v)

for v in ("TELEGRAM_ALLOWED_USERS", "DISCORD_ALLOWED_USERS",

"WHATSAPP_ALLOWED_USERS", "SLACK_ALLOWED_USERS",

"GATEWAY_ALLOWED_USERS")

)

_allow_all = os.getenv("GATEWAY_ALLOW_ALL_USERS", "").lower() in ("true", "1", "yes")

if not _any_allowlist and not _allow_all:

logger.warning(

"No user allowlists configured. All unauthorized users will be denied. "

"Set GATEWAY_ALLOW_ALL_USERS=true in ~/.hermes/.env to allow open access, "

"or configure platform allowlists (e.g., TELEGRAM_ALLOWED_USERS=your_id)."

)

# Discover and load event hooks

self.hooks.discover_and_load()

# Recover background processes from checkpoint (crash recovery)

try:

from tools.process_registry import process_registry

recovered = process_registry.recover_from_checkpoint()

if recovered:

logger.info("Recovered %s background process(es) from previous run", recovered)

except Exception as e:

logger.warning("Process checkpoint recovery: %s", e)

connected_count = 0

# Initialize and connect each configured platform

for platform, platform_config in self.config.platforms.items():

if not platform_config.enabled:

continue

adapter = self._create_adapter(platform, platform_config)

if not adapter:

logger.warning("No adapter available for %s", platform.value)

continue

# Set up message handler

adapter.set_message_handler(self._handle_message)

# Try to connect

logger.info("Connecting to %s...", platform.value)

try:

success = await adapter.connect()

if success:

self.adapters[platform] = adapter

connected_count += 1

logger.info("✓ %s connected", platform.value)

else:

logger.warning("✗ %s failed to connect", platform.value)

except Exception as e:

logger.error("✗ %s error: %s", platform.value, e)

if connected_count == 0:

logger.warning("No messaging platforms connected.")

logger.info("Gateway will continue running for cron job execution.")

# Update delivery router with adapters

self.delivery_router.adapters = self.adapters

self._running = True

# Emit gateway:startup hook

hook_count = len(self.hooks.loaded_hooks)

if hook_count:

logger.info("%s hook(s) loaded", hook_count)

await self.hooks.emit("gateway:startup", {

"platforms": [p.value for p in self.adapters.keys()],

})

if connected_count > 0:

logger.info("Gateway running with %s platform(s)", connected_count)

# Build initial channel directory for send_message name resolution

try:

from gateway.channel_directory import build_channel_directory

directory = build_channel_directory(self.adapters)

ch_count = sum(len(chs) for chs in directory.get("platforms", {}).values())

logger.info("Channel directory built: %d target(s)", ch_count)

except Exception as e:

logger.warning("Channel directory build failed: %s", e)

logger.info("Press Ctrl+C to stop")

return True

async def stop(self) -> None:

"""Stop the gateway and disconnect all adapters."""

logger.info("Stopping gateway...")

self._running = False

for platform, adapter in self.adapters.items():

try:

await adapter.disconnect()

logger.info("✓ %s disconnected", platform.value)

except Exception as e:

logger.error("✗ %s disconnect error: %s", platform.value, e)

self.adapters.clear()

self._shutdown_event.set()

from gateway.status import remove_pid_file

remove_pid_file()

logger.info("Gateway stopped")

async def wait_for_shutdown(self) -> None:

"""Wait for shutdown signal."""

await self._shutdown_event.wait()

def _create_adapter(

self,

platform: Platform,

config: Any

) -> Optional[BasePlatformAdapter]:

"""Create the appropriate adapter for a platform."""

if platform == Platform.TELEGRAM:

from gateway.platforms.telegram import TelegramAdapter, check_telegram_requirements

if not check_telegram_requirements():

logger.warning("Telegram: python-telegram-bot not installed")

return None

return TelegramAdapter(config)

elif platform == Platform.DISCORD:

from gateway.platforms.discord import DiscordAdapter, check_discord_requirements

if not check_discord_requirements():

logger.warning("Discord: discord.py not installed")

return None

return DiscordAdapter(config)

elif platform == Platform.WHATSAPP:

from gateway.platforms.whatsapp import WhatsAppAdapter, check_whatsapp_requirements

if not check_whatsapp_requirements():

logger.warning("WhatsApp: Node.js not installed or bridge not configured")

return None

return WhatsAppAdapter(config)

elif platform == Platform.SLACK:

from gateway.platforms.slack import SlackAdapter, check_slack_requirements

if not check_slack_requirements():

logger.warning("Slack: slack-bolt not installed. Run: pip install 'hermes-agent[slack]'")

return None

return SlackAdapter(config)

elif platform == Platform.HOMEASSISTANT:

from gateway.platforms.homeassistant import HomeAssistantAdapter, check_ha_requirements

if not check_ha_requirements():

logger.warning("HomeAssistant: aiohttp not installed or HASS_TOKEN not set")

return None

return HomeAssistantAdapter(config)

return None

def _is_user_authorized(self, source: SessionSource) -> bool:

"""

Check if a user is authorized to use the bot.

Checks in order:

1. Per-platform allow-all flag (e.g., DISCORD_ALLOW_ALL_USERS=true)

2. Environment variable allowlists (TELEGRAM_ALLOWED_USERS, etc.)

3. DM pairing approved list

4. Global allow-all (GATEWAY_ALLOW_ALL_USERS=true)

5. Default: deny

"""

# Home Assistant events are system-generated (state changes), not

# user-initiated messages. The HASS_TOKEN already authenticates the

# connection, so HA events are always authorized.

if source.platform == Platform.HOMEASSISTANT:

return True

user_id = source.user_id

if not user_id:

return False

platform_env_map = {

Platform.TELEGRAM: "TELEGRAM_ALLOWED_USERS",

Platform.DISCORD: "DISCORD_ALLOWED_USERS",

Platform.WHATSAPP: "WHATSAPP_ALLOWED_USERS",

Platform.SLACK: "SLACK_ALLOWED_USERS",

}

platform_allow_all_map = {

Platform.TELEGRAM: "TELEGRAM_ALLOW_ALL_USERS",

Platform.DISCORD: "DISCORD_ALLOW_ALL_USERS",

Platform.WHATSAPP: "WHATSAPP_ALLOW_ALL_USERS",

Platform.SLACK: "SLACK_ALLOW_ALL_USERS",

}

# Per-platform allow-all flag (e.g., DISCORD_ALLOW_ALL_USERS=true)

platform_allow_all_var = platform_allow_all_map.get(source.platform, "")

if platform_allow_all_var and os.getenv(platform_allow_all_var, "").lower() in ("true", "1", "yes"):

return True

# Check pairing store (always checked, regardless of allowlists)

platform_name = source.platform.value if source.platform else ""

if self.pairing_store.is_approved(platform_name, user_id):

return True

# Check platform-specific and global allowlists

platform_allowlist = os.getenv(platform_env_map.get(source.platform, ""), "").strip()

global_allowlist = os.getenv("GATEWAY_ALLOWED_USERS", "").strip()

if not platform_allowlist and not global_allowlist:

# No allowlists configured -- check global allow-all flag

return os.getenv("GATEWAY_ALLOW_ALL_USERS", "").lower() in ("true", "1", "yes")

# Check if user is in any allowlist

allowed_ids = set()

if platform_allowlist:

allowed_ids.update(uid.strip() for uid in platform_allowlist.split(",") if uid.strip())

if global_allowlist:

allowed_ids.update(uid.strip() for uid in global_allowlist.split(",") if uid.strip())

# WhatsApp JIDs have @s.whatsapp.net suffix — strip it for comparison

check_ids = {user_id}

if "@" in user_id:

check_ids.add(user_id.split("@")[0])

return bool(check_ids & allowed_ids)

async def _handle_message(self, event: MessageEvent) -> Optional[str]:

"""

Handle an incoming message from any platform.

This is the core message processing pipeline:

1. Check user authorization

2. Check for commands (/new, /reset, etc.)

3. Check for running agent and interrupt if needed

4. Get or create session

5. Build context for agent

6. Run agent conversation

7. Return response

"""

source = event.source

# Check if user is authorized

if not self._is_user_authorized(source):

logger.warning("Unauthorized user: %s (%s) on %s", source.user_id, source.user_name, source.platform.value)

# In DMs: offer pairing code. In groups: silently ignore.

if source.chat_type == "dm":

platform_name = source.platform.value if source.platform else "unknown"

code = self.pairing_store.generate_code(

platform_name, source.user_id, source.user_name or ""

)

if code:

adapter = self.adapters.get(source.platform)

if adapter:

await adapter.send(

source.chat_id,

f"Hi~ I don't recognize you yet!\n\n"

f"Here's your pairing code: `{code}`\n\n"

f"Ask the bot owner to run:\n"

f"`hermes pairing approve {platform_name} {code}`"

)

else:

adapter = self.adapters.get(source.platform)

if adapter:

await adapter.send(

source.chat_id,

"Too many pairing requests right now~ "

"Please try again later!"

)

return None

# PRIORITY: If an agent is already running for this session, interrupt it

# immediately. This is before command parsing to minimize latency -- the

# user's "stop" message reaches the agent as fast as possible.

_quick_key = (

f"agent:main:{source.platform.value}:{source.chat_type}:{source.chat_id}"

if source.chat_type != "dm"

else f"agent:main:{source.platform.value}:dm"

)

if _quick_key in self._running_agents:

running_agent = self._running_agents[_quick_key]

logger.debug("PRIORITY interrupt for session %s", _quick_key[:20])

running_agent.interrupt(event.text)

if _quick_key in self._pending_messages:

self._pending_messages[_quick_key] += "\n" + event.text

else:

self._pending_messages[_quick_key] = event.text

return None

# Check for commands

command = event.get_command()

# Emit command:* hook for any recognized slash command

_known_commands = {"new", "reset", "help", "status", "stop", "model",

"personality", "retry", "undo", "sethome", "set-home",

"compress", "usage", "reload-mcp"}

if command and command in _known_commands:

await self.hooks.emit(f"command:{command}", {

"platform": source.platform.value if source.platform else "",

"user_id": source.user_id,

"command": command,

"args": event.get_command_args().strip(),

})

if command in ["new", "reset"]:

return await self._handle_reset_command(event)

if command == "help":

return await self._handle_help_command(event)

if command == "status":

return await self._handle_status_command(event)

if command == "stop":

return await self._handle_stop_command(event)

if command == "model":

return await self._handle_model_command(event)

if command == "personality":

return await self._handle_personality_command(event)

if command == "retry":

return await self._handle_retry_command(event)

if command == "undo":

return await self._handle_undo_command(event)

if command in ["sethome", "set-home"]:

return await self._handle_set_home_command(event)

if command == "compress":

return await self._handle_compress_command(event)

if command == "usage":

return await self._handle_usage_command(event)

if command == "reload-mcp":

return await self._handle_reload_mcp_command(event)

# Skill slash commands: /skill-name loads the skill and sends to agent

if command:

try:

from agent.skill_commands import get_skill_commands, build_skill_invocation_message

skill_cmds = get_skill_commands()

cmd_key = f"/{command}"

if cmd_key in skill_cmds:

user_instruction = event.get_command_args().strip()

msg = build_skill_invocation_message(cmd_key, user_instruction)

if msg:

event.text = msg

# Fall through to normal message processing with skill content

except Exception as e:

logger.debug("Skill command check failed (non-fatal): %s", e)

# Check for pending exec approval responses

if source.chat_type != "dm":

session_key_preview = f"agent:main:{source.platform.value}:{source.chat_type}:{source.chat_id}"

elif source.platform and source.platform.value == "whatsapp" and source.chat_id:

session_key_preview = f"agent:main:{source.platform.value}:dm:{source.chat_id}"

else:

session_key_preview = f"agent:main:{source.platform.value}:dm"

if session_key_preview in self._pending_approvals:

user_text = event.text.strip().lower()

if user_text in ("yes", "y", "approve", "ok", "go", "do it"):

approval = self._pending_approvals.pop(session_key_preview)

cmd = approval["command"]

pattern_key = approval.get("pattern_key", "")

logger.info("User approved dangerous command: %s...", cmd[:60])

from tools.terminal_tool import terminal_tool

from tools.approval import approve_session

approve_session(session_key_preview, pattern_key)

result = terminal_tool(command=cmd, force=True)

return f"✅ Command approved and executed.\n\n```\n{result[:3500]}\n```"

elif user_text in ("no", "n", "deny", "cancel", "nope"):

self._pending_approvals.pop(session_key_preview)

return "❌ Command denied."

# If it's not clearly an approval/denial, fall through to normal processing

# Get or create session

session_entry = self.session_store.get_or_create_session(source)

session_key = session_entry.session_key

# Emit session:start for new or auto-reset sessions

_is_new_session = (

session_entry.created_at == session_entry.updated_at

or getattr(session_entry, "was_auto_reset", False)

)

if _is_new_session:

await self.hooks.emit("session:start", {

"platform": source.platform.value if source.platform else "",

"user_id": source.user_id,

"session_id": session_entry.session_id,

"session_key": session_key,

})

# Build session context

context = build_session_context(source, self.config, session_entry)

# Set environment variables for tools

self._set_session_env(context)

# Build the context prompt to inject

context_prompt = build_session_context_prompt(context)

# If the previous session expired and was auto-reset, prepend a notice

# so the agent knows this is a fresh conversation (not an intentional /reset).

if getattr(session_entry, 'was_auto_reset', False):

context_prompt = (

"[System note: The user's previous session expired due to inactivity. "

"This is a fresh conversation with no prior context.]\n\n"

+ context_prompt

)

session_entry.was_auto_reset = False

# Load conversation history from transcript

history = self.session_store.load_transcript(session_entry.session_id)

# First-message onboarding -- only on the very first interaction ever

if not history and not self.session_store.has_any_sessions():

context_prompt += (

"\n\n[System note: This is the user's very first message ever. "

"Briefly introduce yourself and mention that /help shows available commands. "

"Keep the introduction concise -- one or two sentences max.]"

)

# One-time prompt if no home channel is set for this platform

if not history and source.platform and source.platform != Platform.LOCAL:

platform_name = source.platform.value

env_key = f"{platform_name.upper()}_HOME_CHANNEL"

if not os.getenv(env_key):

adapter = self.adapters.get(source.platform)

if adapter:

await adapter.send(

source.chat_id,

f"📬 No home channel is set for {platform_name.title()}. "

f"A home channel is where Hermes delivers cron job results "

f"and cross-platform messages.\n\n"

f"Type /sethome to make this chat your home channel, "

f"or ignore to skip."

)

# -----------------------------------------------------------------

# Auto-analyze images sent by the user

#

# If the user attached image(s), we run the vision tool eagerly so

# the conversation model always receives a text description. The

# local file path is also included so the model can re-examine the

# image later with a more targeted question via vision_analyze.

#

# We filter to image paths only (by media_type) so that non-image

# attachments (documents, audio, etc.) are not sent to the vision

# tool even when they appear in the same message.

# -----------------------------------------------------------------

message_text = event.text or ""

if event.media_urls:

image_paths = []

for i, path in enumerate(event.media_urls):

# Check media_types if available; otherwise infer from message type

mtype = event.media_types[i] if i < len(event.media_types) else ""

is_image = (

mtype.startswith("image/")

or event.message_type == MessageType.PHOTO

)

if is_image:

image_paths.append(path)

if image_paths:

message_text = await self._enrich_message_with_vision(

message_text, image_paths

)

# -----------------------------------------------------------------

# Auto-transcribe voice/audio messages sent by the user

# -----------------------------------------------------------------

if event.media_urls:

audio_paths = []

for i, path in enumerate(event.media_urls):

mtype = event.media_types[i] if i < len(event.media_types) else ""

is_audio = (

mtype.startswith("audio/")

or event.message_type in (MessageType.VOICE, MessageType.AUDIO)

)

if is_audio:

audio_paths.append(path)

if audio_paths:

message_text = await self._enrich_message_with_transcription(

message_text, audio_paths

)

# -----------------------------------------------------------------

# Enrich document messages with context notes for the agent

# -----------------------------------------------------------------

if event.media_urls and event.message_type == MessageType.DOCUMENT:

for i, path in enumerate(event.media_urls):

mtype = event.media_types[i] if i < len(event.media_types) else ""

if not (mtype.startswith("application/") or mtype.startswith("text/")):

continue

# Extract display filename by stripping the doc_{uuid12}_ prefix

import os as _os

basename = _os.path.basename(path)

# Format: doc_<12hex>_<original_filename>

parts = basename.split("_", 2)

display_name = parts[2] if len(parts) >= 3 else basename

# Sanitize to prevent prompt injection via filenames

import re as _re

display_name = _re.sub(r'[^\w.\- ]', '_', display_name)

if mtype.startswith("text/"):

context_note = (

f"[The user sent a text document: '{display_name}'. "

f"Its content has been included below. "

f"The file is also saved at: {path}]"

)

else:

context_note = (

f"[The user sent a document: '{display_name}'. "

f"The file is saved at: {path}. "

f"Ask the user what they'd like you to do with it.]"

)

message_text = f"{context_note}\n\n{message_text}"

try:

# Emit agent:start hook

hook_ctx = {

"platform": source.platform.value if source.platform else "",

"user_id": source.user_id,

"session_id": session_entry.session_id,

"message": message_text[:500],

}

await self.hooks.emit("agent:start", hook_ctx)

# Run the agent

agent_result = await self._run_agent(

message=message_text,

context_prompt=context_prompt,

history=history,

source=source,

session_id=session_entry.session_id,

session_key=session_key

)

response = agent_result.get("final_response", "")

agent_messages = agent_result.get("messages", [])

# Emit agent:end hook

await self.hooks.emit("agent:end", {

**hook_ctx,

"response": (response or "")[:500],

})

# Check for pending process watchers (check_interval on background processes)

try:

from tools.process_registry import process_registry

while process_registry.pending_watchers:

watcher = process_registry.pending_watchers.pop(0)

asyncio.create_task(self._run_process_watcher(watcher))

except Exception as e:

logger.error("Process watcher setup error: %s", e)

# Check if the agent encountered a dangerous command needing approval

try:

from tools.approval import pop_pending

pending = pop_pending(session_key)

if pending:

self._pending_approvals[session_key] = pending

except Exception as e:

logger.debug("Failed to check pending approvals: %s", e)

# Save the full conversation to the transcript, including tool calls.

# This preserves the complete agent loop (tool_calls, tool results,

# intermediate reasoning) so sessions can be resumed with full context

# and transcripts are useful for debugging and training data.

ts = datetime.now().isoformat()

# If this is a fresh session (no history), write the full tool

# definitions as the first entry so the transcript is self-describing

# -- the same list of dicts sent as tools=[...] in the API request.

if not history:

tool_defs = agent_result.get("tools", [])

self.session_store.append_to_transcript(

session_entry.session_id,

{

"role": "session_meta",

"tools": tool_defs or [],

"model": os.getenv("HERMES_MODEL", ""),

"platform": source.platform.value if source.platform else "",

"timestamp": ts,

}

)

# Find only the NEW messages from this turn (skip history we loaded)

history_len = len(history)

new_messages = agent_messages[history_len:] if len(agent_messages) > history_len else agent_messages

# If no new messages found (edge case), fall back to simple user/assistant

if not new_messages:

self.session_store.append_to_transcript(

session_entry.session_id,

{"role": "user", "content": message_text, "timestamp": ts}

)

if response:

self.session_store.append_to_transcript(

session_entry.session_id,

{"role": "assistant", "content": response, "timestamp": ts}

)

else:

for msg in new_messages:

# Skip system messages (they're rebuilt each run)

if msg.get("role") == "system":

continue

# Add timestamp to each message for debugging

entry = {**msg, "timestamp": ts}

self.session_store.append_to_transcript(

session_entry.session_id, entry

)

# Update session

self.session_store.update_session(session_entry.session_key)

return response

except Exception as e:

logger.exception("Agent error in session %s", session_key)

return (

"Sorry, I encountered an unexpected error. "

"The details have been logged for debugging. "

"Try again or use /reset to start a fresh session."

)

finally:

# Clear session env

self._clear_session_env()

async def _handle_reset_command(self, event: MessageEvent) -> str:

"""Handle /new or /reset command."""

source = event.source