Merge remote-tracking branch 'origin/main' into feat/honcho-async-memory

Made-with: Cursor

# Conflicts:
#	cli.py
#	tests/test_run_agent.py
This commit is contained in:
Erosika 2026-03-11 12:22:56 -04:00
commit a0b0dbe6b2
138 changed files with 17829 additions and 1109 deletions

405
cli.py
View file

@ -20,6 +20,7 @@ import json
import atexit
import uuid
import textwrap
from contextlib import contextmanager
from pathlib import Path
from datetime import datetime
from typing import List, Dict, Any, Optional
@ -54,6 +55,8 @@ except (ImportError, AttributeError):
import threading
import queue
_COMMAND_SPINNER_FRAMES = ("", "", "", "", "", "", "", "", "", "")
# Load .env from ~/.hermes/.env first, then project root as dev fallback
from dotenv import load_dotenv
@ -202,6 +205,7 @@ def load_cli_config() -> Dict[str, Any]:
"display": {
"compact": False,
"resume_display": "full",
"show_reasoning": False,
"skin": "default",
},
"clarify": {
@ -214,6 +218,8 @@ def load_cli_config() -> Dict[str, Any]:
"delegation": {
"max_iterations": 45, # Max tool-calling turns per child agent
"default_toolsets": ["terminal", "file", "web"], # Default toolsets for subagents
"model": "", # Subagent model override (empty = inherit parent model)
"provider": "", # Subagent provider override (empty = inherit parent provider)
},
}
@ -396,6 +402,7 @@ try:
except Exception:
pass # Skin engine is optional — default skin used if unavailable
from rich import box as rich_box
from rich.console import Console
from rich.panel import Panel
from rich.table import Table
@ -1056,6 +1063,12 @@ def save_config_value(key_path: str, value: any) -> bool:
with open(config_path, 'w') as f:
yaml.dump(config, f, default_flow_style=False, sort_keys=False)
# Enforce owner-only permissions on config files (contain API keys)
try:
os.chmod(config_path, 0o600)
except (OSError, NotImplementedError):
pass
return True
except Exception as e:
logger.error("Failed to save config: %s", e)
@ -1103,6 +1116,7 @@ class HermesCLI:
"""
# Initialize Rich console
self.console = Console()
self.config = CLI_CONFIG
self.compact = compact if compact is not None else CLI_CONFIG["display"].get("compact", False)
# tool_progress: "off", "new", "all", "verbose" (from config.yaml display section)
self.tool_progress_mode = CLI_CONFIG["display"].get("tool_progress", "all")
@ -1110,6 +1124,8 @@ class HermesCLI:
self.resume_display = CLI_CONFIG["display"].get("resume_display", "full")
# bell_on_complete: play terminal bell (\a) when agent finishes a response
self.bell_on_complete = CLI_CONFIG["display"].get("bell_on_complete", False)
# show_reasoning: display model thinking/reasoning before the response
self.show_reasoning = CLI_CONFIG["display"].get("show_reasoning", False)
self.verbose = verbose if verbose is not None else (self.tool_progress_mode == "verbose")
# Configuration - priority: CLI args > env vars > config file
@ -1236,6 +1252,12 @@ class HermesCLI:
self._history_file = Path.home() / ".hermes_history"
self._last_invalidate: float = 0.0 # throttle UI repaints
self._spinner_text: str = "" # thinking spinner text for TUI
self._command_running = False
self._command_status = ""
# Background task tracking: {task_id: threading.Thread}
self._background_tasks: Dict[str, threading.Thread] = {}
self._background_task_counter = 0
def _invalidate(self, min_interval: float = 0.25) -> None:
"""Throttled UI repaint — prevents terminal blinking on slow/SSH connections."""
@ -1304,6 +1326,44 @@ class HermesCLI:
self._spinner_text = text or ""
self._invalidate()
def _slow_command_status(self, command: str) -> str:
"""Return a user-facing status message for slower slash commands."""
cmd_lower = command.lower().strip()
if cmd_lower.startswith("/skills search"):
return "Searching skills..."
if cmd_lower.startswith("/skills browse"):
return "Loading skills..."
if cmd_lower.startswith("/skills inspect"):
return "Inspecting skill..."
if cmd_lower.startswith("/skills install"):
return "Installing skill..."
if cmd_lower.startswith("/skills"):
return "Processing skills command..."
if cmd_lower == "/reload-mcp":
return "Reloading MCP servers..."
return "Processing command..."
def _command_spinner_frame(self) -> str:
"""Return the current spinner frame for slow slash commands."""
import time as _time
frame_idx = int(_time.monotonic() * 10) % len(_COMMAND_SPINNER_FRAMES)
return _COMMAND_SPINNER_FRAMES[frame_idx]
@contextmanager
def _busy_command(self, status: str):
"""Expose a temporary busy state in the TUI while a slash command runs."""
self._command_running = True
self._command_status = status
self._invalidate(min_interval=0.0)
try:
print(f"{status}")
yield
finally:
self._command_running = False
self._command_status = ""
self._invalidate(min_interval=0.0)
def _ensure_runtime_credentials(self) -> bool:
"""
Ensure runtime credentials are resolved before agent use.
@ -1440,6 +1500,7 @@ class HermesCLI:
platform="cli",
session_db=self._session_db,
clarify_callback=self._clarify_callback,
reasoning_callback=self._on_reasoning if self.show_reasoning else None,
honcho_session_key=None, # resolved by run_agent via config sessions map / title
fallback_model=self._fallback_model,
thinking_callback=self._on_thinking,
@ -1898,18 +1959,22 @@ class HermesCLI:
)
def show_help(self):
"""Display help information."""
_cprint(f"\n{_BOLD}+{'-' * 50}+{_RST}")
_cprint(f"{_BOLD}|{' ' * 14}(^_^)? Available Commands{' ' * 10}|{_RST}")
_cprint(f"{_BOLD}+{'-' * 50}+{_RST}\n")
for cmd, desc in COMMANDS.items():
_cprint(f" {_GOLD}{cmd:<15}{_RST} {_DIM}-{_RST} {desc}")
"""Display help information with categorized commands."""
from hermes_cli.commands import COMMANDS_BY_CATEGORY
_cprint(f"\n{_BOLD}+{'-' * 55}+{_RST}")
_cprint(f"{_BOLD}|{' ' * 14}(^_^)? Available Commands{' ' * 15}|{_RST}")
_cprint(f"{_BOLD}+{'-' * 55}+{_RST}")
for category, commands in COMMANDS_BY_CATEGORY.items():
_cprint(f"\n {_BOLD}── {category} ──{_RST}")
for cmd, desc in commands.items():
_cprint(f" {_GOLD}{cmd:<15}{_RST} {_DIM}-{_RST} {desc}")
if _skill_commands:
_cprint(f"\n{_BOLD}Skill Commands{_RST} ({len(_skill_commands)} installed):")
for cmd, info in sorted(_skill_commands.items()):
_cprint(f" {_GOLD}{cmd:<22}{_RST} {_DIM}-{_RST} {info['description']}")
_cprint(f" {_GOLD}{cmd:<22}{_RST} {_DIM}-{_RST} {info['description']}")
_cprint(f"\n {_DIM}Tip: Just type your message to chat with Hermes!{_RST}")
_cprint(f" {_DIM}Multi-line: Alt+Enter for a new line{_RST}")
@ -2249,6 +2314,19 @@ class HermesCLI:
print(" /personality - Use a predefined personality")
print()
@staticmethod
def _resolve_personality_prompt(value) -> str:
"""Accept string or dict personality value; return system prompt string."""
if isinstance(value, dict):
parts = [value.get("system_prompt", "")]
if value.get("tone"):
parts.append(f'Tone: {value["tone"]}' )
if value.get("style"):
parts.append(f'Style: {value["style"]}' )
return "\n".join(p for p in parts if p)
return str(value)
def _handle_personality_command(self, cmd: str):
"""Handle the /personality command to set predefined personalities."""
parts = cmd.split(maxsplit=1)
@ -2257,8 +2335,16 @@ class HermesCLI:
# Set personality
personality_name = parts[1].strip().lower()
if personality_name in self.personalities:
self.system_prompt = self.personalities[personality_name]
if personality_name in ("none", "default", "neutral"):
self.system_prompt = ""
self.agent = None # Force re-init
if save_config_value("agent.system_prompt", ""):
print("(^_^)b Personality cleared (saved to config)")
else:
print("(^_^) Personality cleared (session only)")
print(" No personality overlay — using base agent behavior.")
elif personality_name in self.personalities:
self.system_prompt = self._resolve_personality_prompt(self.personalities[personality_name])
self.agent = None # Force re-init
if save_config_value("agent.system_prompt", self.system_prompt):
print(f"(^_^)b Personality set to '{personality_name}' (saved to config)")
@ -2267,7 +2353,7 @@ class HermesCLI:
print(f" \"{self.system_prompt[:60]}{'...' if len(self.system_prompt) > 60 else ''}\"")
else:
print(f"(._.) Unknown personality: {personality_name}")
print(f" Available: {', '.join(self.personalities.keys())}")
print(f" Available: none, {', '.join(self.personalities.keys())}")
else:
# Show available personalities
print()
@ -2275,8 +2361,13 @@ class HermesCLI:
print("|" + " " * 12 + "(^o^)/ Personalities" + " " * 15 + "|")
print("+" + "-" * 50 + "+")
print()
print(f" {'none':<12} - (no personality overlay)")
for name, prompt in self.personalities.items():
print(f" {name:<12} - \"{prompt}\"")
if isinstance(prompt, dict):
preview = prompt.get("description") or prompt.get("system_prompt", "")[:50]
else:
preview = str(prompt)[:50]
print(f" {name:<12} - {preview}")
print()
print(" Usage: /personality <name>")
print()
@ -2777,11 +2868,14 @@ class HermesCLI:
elif cmd_lower.startswith("/cron"):
self._handle_cron_command(cmd_original)
elif cmd_lower.startswith("/skills"):
self._handle_skills_command(cmd_original)
with self._busy_command(self._slow_command_status(cmd_original)):
self._handle_skills_command(cmd_original)
elif cmd_lower == "/platforms" or cmd_lower == "/gateway":
self._show_gateway_status()
elif cmd_lower == "/verbose":
self._toggle_verbose()
elif cmd_lower.startswith("/reasoning"):
self._handle_reasoning_command(cmd_original)
elif cmd_lower == "/compress":
self._manual_compress()
elif cmd_lower == "/usage":
@ -2791,15 +2885,41 @@ class HermesCLI:
elif cmd_lower == "/paste":
self._handle_paste_command()
elif cmd_lower == "/reload-mcp":
self._reload_mcp()
with self._busy_command(self._slow_command_status(cmd_original)):
self._reload_mcp()
elif cmd_lower.startswith("/rollback"):
self._handle_rollback_command(cmd_original)
elif cmd_lower.startswith("/background"):
self._handle_background_command(cmd_original)
elif cmd_lower.startswith("/skin"):
self._handle_skin_command(cmd_original)
else:
# Check for skill slash commands (/gif-search, /axolotl, etc.)
# Check for user-defined quick commands (bypass agent loop, no LLM call)
base_cmd = cmd_lower.split()[0]
if base_cmd in _skill_commands:
quick_commands = self.config.get("quick_commands", {})
if base_cmd.lstrip("/") in quick_commands:
qcmd = quick_commands[base_cmd.lstrip("/")]
if qcmd.get("type") == "exec":
import subprocess
exec_cmd = qcmd.get("command", "")
if exec_cmd:
try:
result = subprocess.run(
exec_cmd, shell=True, capture_output=True,
text=True, timeout=30
)
output = result.stdout.strip() or result.stderr.strip()
self.console.print(output if output else "[dim]Command returned no output[/]")
except subprocess.TimeoutExpired:
self.console.print("[bold red]Quick command timed out (30s)[/]")
except Exception as e:
self.console.print(f"[bold red]Quick command error: {e}[/]")
else:
self.console.print(f"[bold red]Quick command '{base_cmd}' has no command defined[/]")
else:
self.console.print(f"[bold red]Quick command '{base_cmd}' has unsupported type (only 'exec' is supported)[/]")
# Check for skill slash commands (/gif-search, /axolotl, etc.)
elif base_cmd in _skill_commands:
user_instruction = cmd_original[len(base_cmd):].strip()
msg = build_skill_invocation_message(base_cmd, user_instruction)
if msg:
@ -2815,6 +2935,113 @@ class HermesCLI:
return True
def _handle_background_command(self, cmd: str):
"""Handle /background <prompt> — run a prompt in a separate background session.
Spawns a new AIAgent in a background thread with its own session.
When it completes, prints the result to the CLI without modifying
the active session's conversation history.
"""
parts = cmd.strip().split(maxsplit=1)
if len(parts) < 2 or not parts[1].strip():
_cprint(" Usage: /background <prompt>")
_cprint(" Example: /background Summarize the top HN stories today")
_cprint(" The task runs in a separate session and results display here when done.")
return
prompt = parts[1].strip()
self._background_task_counter += 1
task_num = self._background_task_counter
task_id = f"bg_{datetime.now().strftime('%H%M%S')}_{uuid.uuid4().hex[:6]}"
# Make sure we have valid credentials
if not self._ensure_runtime_credentials():
_cprint(" (>_<) Cannot start background task: no valid credentials.")
return
_cprint(f" 🔄 Background task #{task_num} started: \"{prompt[:60]}{'...' if len(prompt) > 60 else ''}\"")
_cprint(f" Task ID: {task_id}")
_cprint(f" You can continue chatting — results will appear when done.\n")
def run_background():
try:
bg_agent = AIAgent(
model=self.model,
api_key=self.api_key,
base_url=self.base_url,
provider=self.provider,
api_mode=self.api_mode,
max_iterations=self.max_turns,
enabled_toolsets=self.enabled_toolsets,
quiet_mode=True,
verbose_logging=False,
session_id=task_id,
platform="cli",
session_db=self._session_db,
reasoning_config=self.reasoning_config,
providers_allowed=self._providers_only,
providers_ignored=self._providers_ignore,
providers_order=self._providers_order,
provider_sort=self._provider_sort,
provider_require_parameters=self._provider_require_params,
provider_data_collection=self._provider_data_collection,
fallback_model=self._fallback_model,
)
result = bg_agent.run_conversation(
user_message=prompt,
task_id=task_id,
)
response = result.get("final_response", "") if result else ""
if not response and result and result.get("error"):
response = f"Error: {result['error']}"
# Display result in the CLI (thread-safe via patch_stdout)
print()
_cprint(f"{_GOLD}{'' * 40}{_RST}")
_cprint(f" ✅ Background task #{task_num} complete")
_cprint(f" Prompt: \"{prompt[:60]}{'...' if len(prompt) > 60 else ''}\"")
_cprint(f"{_GOLD}{'' * 40}{_RST}")
if response:
try:
from hermes_cli.skin_engine import get_active_skin
_skin = get_active_skin()
label = _skin.get_branding("response_label", "⚕ Hermes")
_resp_color = _skin.get_color("response_border", "#CD7F32")
except Exception:
label = "⚕ Hermes"
_resp_color = "#CD7F32"
_chat_console = ChatConsole()
_chat_console.print(Panel(
response,
title=f"[bold]{label} (background #{task_num})[/bold]",
title_align="left",
border_style=_resp_color,
box=rich_box.HORIZONTALS,
padding=(1, 2),
))
else:
_cprint(" (No response generated)")
# Play bell if enabled
if self.bell_on_complete:
sys.stdout.write("\a")
sys.stdout.flush()
except Exception as e:
print()
_cprint(f" ❌ Background task #{task_num} failed: {e}")
finally:
self._background_tasks.pop(task_id, None)
if self._app:
self._invalidate(min_interval=0)
thread = threading.Thread(target=run_background, daemon=True, name=f"bg-task-{task_id}")
self._background_tasks[task_id] = thread
thread.start()
def _handle_skin_command(self, cmd: str):
"""Handle /skin [name] — show or change the display skin."""
try:
@ -2874,6 +3101,75 @@ class HermesCLI:
}
self.console.print(labels.get(self.tool_progress_mode, ""))
def _handle_reasoning_command(self, cmd: str):
"""Handle /reasoning — manage effort level and display toggle.
Usage:
/reasoning Show current effort level and display state
/reasoning <level> Set reasoning effort (none, low, medium, high, xhigh)
/reasoning show|on Show model thinking/reasoning in output
/reasoning hide|off Hide model thinking/reasoning from output
"""
parts = cmd.strip().split(maxsplit=1)
if len(parts) < 2:
# Show current state
rc = self.reasoning_config
if rc is None:
level = "medium (default)"
elif rc.get("enabled") is False:
level = "none (disabled)"
else:
level = rc.get("effort", "medium")
display_state = "on" if self.show_reasoning else "off"
_cprint(f" {_GOLD}Reasoning effort: {level}{_RST}")
_cprint(f" {_GOLD}Reasoning display: {display_state}{_RST}")
_cprint(f" {_DIM}Usage: /reasoning <none|low|medium|high|xhigh|show|hide>{_RST}")
return
arg = parts[1].strip().lower()
# Display toggle
if arg in ("show", "on"):
self.show_reasoning = True
if self.agent:
self.agent.reasoning_callback = self._on_reasoning
_cprint(f" {_GOLD}Reasoning display: ON{_RST}")
_cprint(f" {_DIM}Model thinking will be shown during and after each response.{_RST}")
return
if arg in ("hide", "off"):
self.show_reasoning = False
if self.agent:
self.agent.reasoning_callback = None
_cprint(f" {_GOLD}Reasoning display: OFF{_RST}")
return
# Effort level change
parsed = _parse_reasoning_config(arg)
if parsed is None:
_cprint(f" {_DIM}(._.) Unknown argument: {arg}{_RST}")
_cprint(f" {_DIM}Valid levels: none, low, minimal, medium, high, xhigh{_RST}")
_cprint(f" {_DIM}Display: show, hide{_RST}")
return
self.reasoning_config = parsed
self.agent = None # Force agent re-init with new reasoning config
if save_config_value("agent.reasoning_effort", arg):
_cprint(f" {_GOLD}Reasoning effort set to '{arg}' (saved to config){_RST}")
else:
_cprint(f" {_GOLD}Reasoning effort set to '{arg}' (session only){_RST}")
def _on_reasoning(self, reasoning_text: str):
"""Callback for intermediate reasoning display during tool-call loops."""
lines = reasoning_text.strip().splitlines()
if len(lines) > 5:
preview = "\n".join(lines[:5])
preview += f"\n ... ({len(lines) - 5} more lines)"
else:
preview = reasoning_text.strip()
_cprint(f" {_DIM}[thinking] {preview}{_RST}")
def _manual_compress(self):
"""Manually trigger context compression on the current conversation."""
if not self.conversation_history or len(self.conversation_history) < 4:
@ -3006,7 +3302,8 @@ class HermesCLI:
with _lock:
old_servers = set(_servers.keys())
print("🔄 Reloading MCP servers...")
if not self._command_running:
print("🔄 Reloading MCP servers...")
# Shutdown existing connections
shutdown_mcp_servers()
@ -3349,6 +3646,24 @@ class HermesCLI:
response = response + "\n\n---\n_[Interrupted - processing new message]_"
response_previewed = result.get("response_previewed", False) if result else False
# Display reasoning (thinking) box if enabled and available
if self.show_reasoning and result:
reasoning = result.get("last_reasoning")
if reasoning:
w = shutil.get_terminal_size().columns
r_label = " Reasoning "
r_fill = w - 2 - len(r_label)
r_top = f"{_DIM}┌─{r_label}{'' * max(r_fill - 1, 0)}{_RST}"
r_bot = f"{_DIM}{'' * (w - 2)}{_RST}"
# Collapse long reasoning: show first 10 lines
lines = reasoning.strip().splitlines()
if len(lines) > 10:
display_reasoning = "\n".join(lines[:10])
display_reasoning += f"\n{_DIM} ... ({len(lines) - 10} more lines){_RST}"
else:
display_reasoning = reasoning.strip()
_cprint(f"\n{r_top}\n{_DIM}{display_reasoning}{_RST}\n{r_bot}")
if response and not response_previewed:
# Use a Rich Panel for the response box — adapts to terminal
# width at render time instead of hard-coding border length.
@ -3367,6 +3682,7 @@ class HermesCLI:
title=f"[bold]{label}[/bold]",
title_align="left",
border_style=_resp_color,
box=rich_box.HORIZONTALS,
padding=(1, 2),
))
@ -3466,6 +3782,10 @@ class HermesCLI:
self._approval_state = None # dict with command, description, choices, selected, response_queue
self._approval_deadline = 0
# Slash command loading state
self._command_running = False
self._command_status = ""
# Clipboard image attachments (paste images into the CLI)
self._attached_images: list[Path] = []
self._image_counter = 0
@ -3738,6 +4058,8 @@ class HermesCLI:
return [('class:clarify-selected', ' ')]
if cli_ref._clarify_state:
return [('class:prompt-working', '? ')]
if cli_ref._command_running:
return [('class:prompt-working', f"{cli_ref._command_spinner_frame()} ")]
if cli_ref._agent_running:
return [('class:prompt-working', ' ')]
return [('class:prompt', ' ')]
@ -3749,6 +4071,7 @@ class HermesCLI:
style='class:input-area',
multiline=True,
wrap_lines=True,
read_only=Condition(lambda: bool(cli_ref._command_running)),
history=FileHistory(str(self._history_file)),
completer=SlashCommandCompleter(skill_commands_provider=lambda: _skill_commands),
complete_while_typing=True,
@ -3833,6 +4156,10 @@ class HermesCLI:
return "type your answer here and press Enter"
if cli_ref._clarify_state:
return ""
if cli_ref._command_running:
frame = cli_ref._command_spinner_frame()
status = cli_ref._command_status or "Processing command..."
return f"{frame} {status}"
if cli_ref._agent_running:
return "type a message + Enter to interrupt, Ctrl+C to cancel"
return ""
@ -3872,10 +4199,16 @@ class HermesCLI:
('class:clarify-countdown', countdown),
]
if cli_ref._command_running:
frame = cli_ref._command_spinner_frame()
return [
('class:hint', f' {frame} command in progress · input temporarily disabled'),
]
return []
def get_hint_height():
if cli_ref._sudo_state or cli_ref._approval_state or cli_ref._clarify_state:
if cli_ref._sudo_state or cli_ref._approval_state or cli_ref._clarify_state or cli_ref._command_running:
return 1
# Keep a 1-line spacer while agent runs so output doesn't push
# right up against the top rule of the input area
@ -4185,6 +4518,19 @@ class HermesCLI:
**({'cursor': _STEADY_CURSOR} if _STEADY_CURSOR is not None else {}),
)
self._app = app # Store reference for clarify_callback
def spinner_loop():
import time as _time
while not self._should_exit:
if self._command_running and self._app:
self._invalidate(min_interval=0.1)
_time.sleep(0.1)
else:
_time.sleep(0.05)
spinner_thread = threading.Thread(target=spinner_loop, daemon=True)
spinner_thread.start()
# Background thread to process inputs and run agent
def process_loop():
@ -4311,6 +4657,7 @@ def main(
base_url: str = None,
max_turns: int = None,
verbose: bool = False,
quiet: bool = False,
compact: bool = False,
list_tools: bool = False,
list_toolsets: bool = False,
@ -4453,10 +4800,22 @@ def main(
# Handle single query mode
if query:
cli.show_banner()
cli.console.print(f"[bold blue]Query:[/] {query}")
cli.chat(query)
cli._print_exit_summary()
if quiet:
# Quiet mode: suppress banner, spinner, tool previews.
# Only print the final response and parseable session info.
cli.tool_progress_mode = "off"
if cli._init_agent():
cli.agent.quiet_mode = True
result = cli.agent.run_conversation(query)
response = result.get("final_response", "") if isinstance(result, dict) else str(result)
if response:
print(response)
print(f"\nsession_id: {cli.session_id}")
else:
cli.show_banner()
cli.console.print(f"[bold blue]Query:[/] {query}")
cli.chat(query)
cli._print_exit_summary()
return
# Run interactive mode