Compare commits

..

20 Commits

Author SHA1 Message Date
Sam Herring e3123be445 Removing old patches 2026-03-30 10:06:08 -07:00
Sam Herring e46d5b2c13 Removing old files 2026-03-30 09:58:05 -07:00
Sam Herring 34cc666105 Updating with trainer config pieces 2026-03-30 09:46:24 -07:00
Sam Herring d6832260f9 Fixing eval steps to be a set number of tasks 2026-03-30 09:46:24 -07:00
Sam Herring d2652e980f Adding random jitter for agent temp to add variance into rollouts 2026-03-30 09:46:24 -07:00
Sam Herring 89cea9fd2d Test basic Atropos trainer 2026-03-30 09:46:24 -07:00
Sam Herring 143e72c145 Updating endless terminals env with silenced warnings 2026-03-30 09:46:24 -07:00
Sam Herring 51305b3f3d Tool call changes 2026-03-30 09:46:24 -07:00
Sam Herring 570e52b342 Monkey patching chat template kwargs 2026-03-30 09:46:24 -07:00
Sam Herring d6e874491d Env changes for tool use 2026-03-30 09:46:24 -07:00
Sam Herring dd3812dffe Adding tool call parser default 2026-03-30 09:46:24 -07:00
Sam Herring 6e17630bac Eval splits for holdout sets 2026-03-30 09:46:24 -07:00
Sam Herring 53b710b13f Changing return type to be ScoredDataGroup to account for multiple trajectories 2026-03-30 09:46:24 -07:00
Sam Herring 5b1e8059cb Added task sppecific metris and evals 2026-03-30 09:46:24 -07:00
Sam Herring ff16a33cdd Wandb changes 2026-03-30 09:46:24 -07:00
Sam Herring 7cfb9eb1f6 Updating config 2026-03-30 09:46:24 -07:00
Sam Herring c7b15f8ce1 Adding config init method 2026-03-30 09:46:24 -07:00
Sam Herring 7602c462ee Updating path vars and dataset loading 2026-03-30 09:46:24 -07:00
Sam Herring e38c24363c Updating to use hermes-agent backend and parse container definition out of provided .sif files 2026-03-30 09:46:24 -07:00
Sam Herring d768b244a5 Adding endless terminal environment after rebase: 2026-03-30 09:46:24 -07:00
152 changed files with 1497 additions and 4880 deletions
-15
View File
@@ -231,21 +231,6 @@ VOICE_TOOLS_OPENAI_KEY=
# Slack allowed users (comma-separated Slack user IDs)
# SLACK_ALLOWED_USERS=
# =============================================================================
# TELEGRAM INTEGRATION
# =============================================================================
# Telegram Bot Token - From @BotFather (https://t.me/BotFather)
# TELEGRAM_BOT_TOKEN=
# TELEGRAM_ALLOWED_USERS= # Comma-separated user IDs
# TELEGRAM_HOME_CHANNEL= # Default chat for cron delivery
# TELEGRAM_HOME_CHANNEL_NAME= # Display name for home channel
# Webhook mode (optional — for cloud deployments like Fly.io/Railway)
# Default is long polling. Setting TELEGRAM_WEBHOOK_URL switches to webhook mode.
# TELEGRAM_WEBHOOK_URL=https://my-app.fly.dev/telegram
# TELEGRAM_WEBHOOK_PORT=8443
# TELEGRAM_WEBHOOK_SECRET= # Recommended for production
# WhatsApp (built-in Baileys bridge — run `hermes whatsapp` to pair)
# WHATSAPP_ENABLED=false
# WHATSAPP_ALLOWED_USERS=15551234567
-2
View File
@@ -19,8 +19,6 @@ concurrency:
jobs:
build-and-deploy:
# Only run on the upstream repository, not on forks
if: github.repository == 'NousResearch/hermes-agent'
runs-on: ubuntu-latest
environment:
name: github-pages
-2
View File
@@ -12,8 +12,6 @@ concurrency:
jobs:
build-and-push:
# Only run on the upstream repository, not on forks
if: github.repository == 'NousResearch/hermes-agent'
runs-on: ubuntu-latest
timeout-minutes: 30
steps:
-4
View File
@@ -1,4 +0,0 @@
graft skills
graft optional-skills
global-exclude __pycache__
global-exclude *.py[cod]
+2 -31
View File
@@ -152,35 +152,16 @@ def _is_oauth_token(key: str) -> bool:
Regular API keys start with 'sk-ant-api'. Everything else (setup-tokens
starting with 'sk-ant-oat', managed keys, JWTs, etc.) needs Bearer auth.
Azure AI Foundry keys (non sk-ant- prefixed) should use x-api-key, not Bearer.
"""
if not key:
return False
# Regular Console API keys use x-api-key header
if key.startswith("sk-ant-api"):
return False
# Azure AI Foundry keys don't start with sk-ant- at all — treat as regular API key
if not key.startswith("sk-ant-"):
return False
# Everything else (setup-tokens sk-ant-oat, managed keys, JWTs) uses Bearer auth
# Everything else (setup-tokens, managed keys, JWTs) uses Bearer auth
return True
def _requires_bearer_auth(base_url: str | None) -> bool:
"""Return True for Anthropic-compatible providers that require Bearer auth.
Some third-party /anthropic endpoints implement Anthropic's Messages API but
require Authorization: Bearer instead of Anthropic's native x-api-key header.
MiniMax's global and China Anthropic-compatible endpoints follow this pattern.
"""
if not base_url:
return False
normalized = base_url.rstrip("/").lower()
return normalized.startswith("https://api.minimax.io/anthropic") or normalized.startswith(
"https://api.minimaxi.com/anthropic"
)
def build_anthropic_client(api_key: str, base_url: str = None):
"""Create an Anthropic client, auto-detecting setup-tokens vs API keys.
@@ -199,17 +180,7 @@ def build_anthropic_client(api_key: str, base_url: str = None):
if base_url:
kwargs["base_url"] = base_url
if _requires_bearer_auth(base_url):
# Some Anthropic-compatible providers (e.g. MiniMax) expect the API key in
# Authorization: Bearer even for regular API keys. Route those endpoints
# through auth_token so the SDK sends Bearer auth instead of x-api-key.
# Check this before OAuth token shape detection because MiniMax secrets do
# not use Anthropic's sk-ant-api prefix and would otherwise be misread as
# Anthropic OAuth/setup tokens.
kwargs["auth_token"] = api_key
if _COMMON_BETAS:
kwargs["default_headers"] = {"anthropic-beta": ",".join(_COMMON_BETAS)}
elif _is_oauth_token(api_key):
if _is_oauth_token(api_key):
# OAuth access token / setup-token → Bearer auth + Claude Code identity.
# Anthropic routes OAuth requests based on user-agent and headers;
# without Claude Code's fingerprint, requests get intermittent 500s.
-39
View File
@@ -2837,28 +2837,6 @@ class HermesCLI:
print(" Example: python cli.py --toolsets web,terminal")
print()
def _handle_profile_command(self):
"""Display active profile name and home directory."""
from hermes_constants import get_hermes_home, display_hermes_home
home = get_hermes_home()
display = display_hermes_home()
profiles_parent = Path.home() / ".hermes" / "profiles"
try:
rel = home.relative_to(profiles_parent)
profile_name = str(rel).split("/")[0]
except ValueError:
profile_name = None
print()
if profile_name:
print(f" Profile: {profile_name}")
else:
print(" Profile: default")
print(f" Home: {display}")
print()
def show_config(self):
"""Display current configuration with kawaii ASCII art."""
# Get terminal config from environment (which was set from cli-config.yaml)
@@ -3701,8 +3679,6 @@ class HermesCLI:
return False
elif canonical == "help":
self.show_help()
elif canonical == "profile":
self._handle_profile_command()
elif canonical == "tools":
self._handle_tools_command(cmd_original)
elif canonical == "toolsets":
@@ -3860,8 +3836,6 @@ class HermesCLI:
self.console.print(f" Status bar {state}")
elif canonical == "verbose":
self._toggle_verbose()
elif canonical == "yolo":
self._toggle_yolo()
elif canonical == "reasoning":
self._handle_reasoning_command(cmd_original)
elif canonical == "compress":
@@ -4460,17 +4434,6 @@ class HermesCLI:
}
_cprint(labels.get(self.tool_progress_mode, ""))
def _toggle_yolo(self):
"""Toggle YOLO mode — skip all dangerous command approval prompts."""
import os
current = bool(os.environ.get("HERMES_YOLO_MODE"))
if current:
os.environ.pop("HERMES_YOLO_MODE", None)
self.console.print(" ⚠ YOLO mode [bold red]OFF[/] — dangerous commands will require approval.")
else:
os.environ["HERMES_YOLO_MODE"] = "1"
self.console.print(" ⚡ YOLO mode [bold green]ON[/] — all commands auto-approved. Use with caution.")
def _handle_reasoning_command(self, cmd: str):
"""Handle /reasoning — manage effort level and display toggle.
@@ -5597,8 +5560,6 @@ class HermesCLI:
self.agent = None
# Initialize agent if needed
if self.agent is None:
_cprint(f"{_DIM}Initializing agent...{_RST}")
if not self._init_agent(
model_override=turn_route["model"],
runtime_override=turn_route["runtime"],
+1
View File
@@ -13,6 +13,7 @@ Core layers:
Concrete environments:
- terminal_test_env/: Simple file-creation tasks for testing the stack
- hermes_swe_env/: SWE-bench style tasks with Modal sandboxes
- endless_terminals/: Terminal tasks from HuggingFace dataset with Apptainer containers
Benchmarks (eval-only):
- benchmarks/terminalbench_2/: Terminal-Bench 2.0 evaluation
@@ -0,0 +1,5 @@
"""Endless Terminals Environment - Terminal task training from HuggingFace dataset."""
from .endless_terminals_env import EndlessTerminalsEnv, EndlessTerminalsEnvConfig
__all__ = ["EndlessTerminalsEnv", "EndlessTerminalsEnvConfig"]
File diff suppressed because it is too large Load Diff
@@ -0,0 +1,91 @@
# Endless Terminals - Qwen3-4B-Instruct-2507
# Single config for both trainer (launch_training.py) and env (endless_terminals_env.py serve)
#
# Usage:
# Terminal 1: run-api
# Terminal 2: cd tinker-atropos && python launch_training.py --config ../environments/endless_terminals/tinker_qwen.yaml
# Terminal 3: python environments/endless_terminals/endless_terminals_env.py serve --config environments/endless_terminals/tinker_qwen.yaml
env:
# Toolsets
enabled_toolsets: ["terminal", "file"]
# Model / tokenizer
tokenizer_name: "Qwen/Qwen3-4B-Instruct-2507"
# Agent configuration
max_agent_turns: 16
max_token_length: 2048
agent_temperature: 0.6
extra_body:
chat_template_kwargs:
enable_thinking: false
tool_call_parser: "hermes"
# Terminal backend
terminal_backend: "docker"
# Dataset settings
use_dataset: true
dataset_name: "obiwan96/endless-terminals"
dataset_split: "train"
dataset_cache_dir: "~/.cache/huggingface/datasets"
tasks_base_dir: "/Users/samherring/Desktop/Projects/Hermes-Agent/endless-terminals"
# Test execution
test_timeout_s: 180
default_docker_image: "ubuntu:22.04"
max_concurrent_containers: 16
# Training configuration
group_size: 16
batch_size: 64 # 4 groups × 16 rollouts per step
total_steps: 500
steps_per_eval: 5
min_items_sent_before_logging: 1
ensure_scores_are_not_same: true
max_num_workers: 2048
worker_timeout: 3600
inference_weight: 1.0
eval_limit_ratio: 0.1
rollout_server_url: "http://localhost:8000"
# Evaluation configuration
num_eval_tasks: 20
eval_split_ratio: 0.1
# Logging
use_wandb: true
wandb_name: "endless-terminals-qwen3-4b"
# System prompt
system_prompt: >
You are a skilled Linux system administrator and programmer.
You have access to a terminal and file tools to complete system administration
and programming tasks. Use the tools effectively to solve the given task,
and verify your solution works correctly before finishing.
Keep each command short and focused — break complex tasks into multiple steps
rather than writing long one-liners.
tinker:
lora_rank: 32
learning_rate: 0.0000005
max_token_trainer_length: 32768
checkpoint_dir: "./temp/"
save_checkpoint_interval: 50
wandb_project: "endless-terminals"
wandb_group: null
wandb_run_name: "qwen3-4b"
tool_call_parser: "hermes"
openai:
- model_name: "Qwen/Qwen3-4B-Instruct-2507"
base_url: "http://localhost:8001/v1"
api_key: "x"
weight: 1.0
num_requests_for_eval: 64
timeout: 600
server_type: "sglang"
slurm: false
testing: false
-1
View File
@@ -298,7 +298,6 @@ class HermesAgentBaseEnv(BaseEnv):
return False
server = self.server.servers[0]
# If the server is an OpenAI server (not VLLM/SGLang), use direct mode
from atroposlib.envs.server_handling.openai_server import OpenAIServer
return not isinstance(server, OpenAIServer)
@@ -48,7 +48,13 @@ class HermesToolCallParser(ToolCallParser):
if not raw_json.strip():
continue
tc_data = json.loads(raw_json)
try:
tc_data = json.loads(raw_json)
except json.JSONDecodeError:
# Fix invalid backslash escapes from shell commands in JSON strings
# e.g. \s \w \d \n (unescaped) → \\s \\w \\d \\n
fixed = re.sub(r'\\([^"\\/bfnrtu0-9\n])', r'\\\\\1', raw_json)
tc_data = json.loads(fixed)
tool_calls.append(
ChatCompletionMessageToolCall(
id=f"call_{uuid.uuid4().hex[:8]}",
+2 -9
View File
@@ -27,16 +27,9 @@ def _coerce_bool(value: Any, default: bool = True) -> bool:
return default
if isinstance(value, bool):
return value
if isinstance(value, int):
return value != 0
if isinstance(value, str):
lowered = value.strip().lower()
if lowered in ("true", "1", "yes", "on"):
return True
if lowered in ("false", "0", "no", "off"):
return False
return default
return default
return value.strip().lower() in ("true", "1", "yes", "on")
return bool(value)
def _normalize_unauthorized_dm_behavior(value: Any, default: str = "pair") -> str:
+4 -164
View File
@@ -49,14 +49,6 @@ _STORE_DIR = _get_hermes_dir("platforms/matrix/store", "matrix/store")
# Grace period: ignore messages older than this many seconds before startup.
_STARTUP_GRACE_SECONDS = 5
# E2EE key export file for persistence across restarts.
_KEY_EXPORT_FILE = _STORE_DIR / "exported_keys.txt"
_KEY_EXPORT_PASSPHRASE = "hermes-matrix-e2ee-keys"
# Pending undecrypted events: cap and TTL for retry buffer.
_MAX_PENDING_EVENTS = 100
_PENDING_EVENT_TTL = 300 # seconds — stop retrying after 5 min
def check_matrix_requirements() -> bool:
"""Return True if the Matrix adapter can be used."""
@@ -119,10 +111,6 @@ class MatrixAdapter(BasePlatformAdapter):
self._processed_events: deque = deque(maxlen=1000)
self._processed_events_set: set = set()
# Buffer for undecrypted events pending key receipt.
# Each entry: (room, event, timestamp)
self._pending_megolm: list = []
def _is_duplicate_event(self, event_id) -> bool:
"""Return True if this event was already processed. Tracks the ID otherwise."""
if not event_id:
@@ -244,16 +232,6 @@ class MatrixAdapter(BasePlatformAdapter):
logger.info("Matrix: E2EE crypto initialized")
except Exception as exc:
logger.warning("Matrix: crypto init issue: %s", exc)
# Import previously exported Megolm keys (survives restarts).
if _KEY_EXPORT_FILE.exists():
try:
await client.import_keys(
str(_KEY_EXPORT_FILE), _KEY_EXPORT_PASSPHRASE,
)
logger.info("Matrix: imported Megolm keys from backup")
except Exception as exc:
logger.debug("Matrix: could not import keys: %s", exc)
elif self._encryption:
logger.warning(
"Matrix: E2EE requested but crypto store is not loaded; "
@@ -308,18 +286,6 @@ class MatrixAdapter(BasePlatformAdapter):
except (asyncio.CancelledError, Exception):
pass
# Export Megolm keys before closing so the next restart can decrypt
# events that used sessions from this run.
if self._client and self._encryption and getattr(self._client, "olm", None):
try:
_STORE_DIR.mkdir(parents=True, exist_ok=True)
await self._client.export_keys(
str(_KEY_EXPORT_FILE), _KEY_EXPORT_PASSPHRASE,
)
logger.info("Matrix: exported Megolm keys for next restart")
except Exception as exc:
logger.debug("Matrix: could not export keys on disconnect: %s", exc)
if self._client:
await self._client.close()
self._client = None
@@ -699,22 +665,17 @@ class MatrixAdapter(BasePlatformAdapter):
Hermes uses a custom sync loop instead of matrix-nio's sync_forever(),
so we need to explicitly drive the key management work that sync_forever()
normally handles for encrypted rooms.
Also auto-trusts all devices (so senders share session keys with us)
and retries decryption for any buffered MegolmEvents.
"""
client = self._client
if not client or not self._encryption or not getattr(client, "olm", None):
return
did_query_keys = client.should_query_keys
tasks = [asyncio.create_task(client.send_to_device_messages())]
if client.should_upload_keys:
tasks.append(asyncio.create_task(client.keys_upload()))
if did_query_keys:
if client.should_query_keys:
tasks.append(asyncio.create_task(client.keys_query()))
if client.should_claim_keys:
@@ -730,111 +691,6 @@ class MatrixAdapter(BasePlatformAdapter):
except Exception as exc:
logger.warning("Matrix: E2EE maintenance task failed: %s", exc)
# After key queries, auto-trust all devices so senders share keys with
# us. For a bot this is the right default — we want to decrypt
# everything, not enforce manual verification.
if did_query_keys:
self._auto_trust_devices()
# Retry any buffered undecrypted events now that new keys may have
# arrived (from key requests, key queries, or to-device forwarding).
if self._pending_megolm:
await self._retry_pending_decryptions()
def _auto_trust_devices(self) -> None:
"""Trust/verify all unverified devices we know about.
When other clients see our device as verified, they proactively share
Megolm session keys with us. Without this, many clients will refuse
to include an unverified device in key distributions.
"""
client = self._client
if not client:
return
device_store = getattr(client, "device_store", None)
if not device_store:
return
own_device = getattr(client, "device_id", None)
trusted_count = 0
try:
# DeviceStore.__iter__ yields OlmDevice objects directly.
for device in device_store:
if getattr(device, "device_id", None) == own_device:
continue
if not getattr(device, "verified", False):
client.verify_device(device)
trusted_count += 1
except Exception as exc:
logger.debug("Matrix: auto-trust error: %s", exc)
if trusted_count:
logger.info("Matrix: auto-trusted %d new device(s)", trusted_count)
async def _retry_pending_decryptions(self) -> None:
"""Retry decrypting buffered MegolmEvents after new keys arrive."""
import nio
client = self._client
if not client or not self._pending_megolm:
return
now = time.time()
still_pending: list = []
for room, event, ts in self._pending_megolm:
# Drop events that have aged past the TTL.
if now - ts > _PENDING_EVENT_TTL:
logger.debug(
"Matrix: dropping expired pending event %s (age %.0fs)",
getattr(event, "event_id", "?"), now - ts,
)
continue
try:
decrypted = client.decrypt_event(event)
except Exception:
# Still missing the key — keep in buffer.
still_pending.append((room, event, ts))
continue
if isinstance(decrypted, nio.MegolmEvent):
# decrypt_event returned the same undecryptable event.
still_pending.append((room, event, ts))
continue
logger.info(
"Matrix: decrypted buffered event %s (%s)",
getattr(event, "event_id", "?"),
type(decrypted).__name__,
)
# Route to the appropriate handler based on decrypted type.
try:
if isinstance(decrypted, nio.RoomMessageText):
await self._on_room_message(room, decrypted)
elif isinstance(
decrypted,
(nio.RoomMessageImage, nio.RoomMessageAudio,
nio.RoomMessageVideo, nio.RoomMessageFile),
):
await self._on_room_message_media(room, decrypted)
else:
logger.debug(
"Matrix: decrypted event %s has unhandled type %s",
getattr(event, "event_id", "?"),
type(decrypted).__name__,
)
except Exception as exc:
logger.warning(
"Matrix: error processing decrypted event %s: %s",
getattr(event, "event_id", "?"), exc,
)
self._pending_megolm = still_pending
# ------------------------------------------------------------------
# Event callbacks
# ------------------------------------------------------------------
@@ -856,29 +712,13 @@ class MatrixAdapter(BasePlatformAdapter):
if event_ts and event_ts < self._startup_ts - _STARTUP_GRACE_SECONDS:
return
# Handle undecryptable MegolmEvents: request the missing session key
# and buffer the event for retry once the key arrives.
# Handle decrypted MegolmEvents — extract the inner event.
if isinstance(event, nio.MegolmEvent):
# Failed to decrypt.
logger.warning(
"Matrix: could not decrypt event %s in %s — requesting key",
"Matrix: could not decrypt event %s in %s",
event.event_id, room.room_id,
)
# Ask other devices in the room to forward the session key.
try:
resp = await self._client.request_room_key(event)
if hasattr(resp, "event_id") or not isinstance(resp, Exception):
logger.debug(
"Matrix: room key request sent for session %s",
getattr(event, "session_id", "?"),
)
except Exception as exc:
logger.debug("Matrix: room key request failed: %s", exc)
# Buffer for retry on next maintenance cycle.
self._pending_megolm.append((room, event, time.time()))
if len(self._pending_megolm) > _MAX_PENDING_EVENTS:
self._pending_megolm = self._pending_megolm[-_MAX_PENDING_EVENTS:]
return
# Skip edits (m.replace relation).
+2 -11
View File
@@ -622,19 +622,10 @@ class TelegramAdapter(BasePlatformAdapter):
# gateway command there automatically adds it to the Telegram menu.
try:
from telegram import BotCommand
from hermes_cli.commands import telegram_menu_commands
# Telegram allows up to 100 commands but has an undocumented
# payload size limit. Skill descriptions are truncated to 40
# chars in telegram_menu_commands() to fit 100 commands safely.
menu_commands, hidden_count = telegram_menu_commands(max_commands=100)
from hermes_cli.commands import telegram_bot_commands
await self._bot.set_my_commands([
BotCommand(name, desc) for name, desc in menu_commands
BotCommand(name, desc) for name, desc in telegram_bot_commands()
])
if hidden_count:
logger.info(
"[%s] Telegram menu: %d commands registered, %d hidden (over 100 limit). Use /commands for full list.",
self.name, len(menu_commands), hidden_count,
)
except Exception as e:
logger.warning(
"[%s] Could not register Telegram command menu: %s",
+4 -209
View File
@@ -301,50 +301,6 @@ def _resolve_runtime_agent_kwargs() -> dict:
}
def _check_unavailable_skill(command_name: str) -> str | None:
"""Check if a command matches a known-but-inactive skill.
Returns a helpful message if the skill exists but is disabled or only
available as an optional install. Returns None if no match found.
"""
# Normalize: command uses hyphens, skill names may use hyphens or underscores
normalized = command_name.lower().replace("_", "-")
try:
from tools.skills_tool import SKILLS_DIR, _get_disabled_skill_names
disabled = _get_disabled_skill_names()
# Check disabled built-in skills
for skill_md in SKILLS_DIR.rglob("SKILL.md"):
if any(part in ('.git', '.github', '.hub') for part in skill_md.parts):
continue
name = skill_md.parent.name.lower().replace("_", "-")
if name == normalized and name in disabled:
return (
f"The **{command_name}** skill is installed but disabled.\n"
f"Enable it with: `hermes skills config`"
)
# Check optional skills (shipped with repo but not installed)
from hermes_constants import get_hermes_home, get_optional_skills_dir
repo_root = Path(__file__).resolve().parent.parent
optional_dir = get_optional_skills_dir(repo_root / "optional-skills")
if optional_dir.exists():
for skill_md in optional_dir.rglob("SKILL.md"):
name = skill_md.parent.name.lower().replace("_", "-")
if name == normalized:
# Build install path: official/<category>/<name>
rel = skill_md.parent.relative_to(optional_dir)
parts = list(rel.parts)
install_path = f"official/{'/'.join(parts)}"
return (
f"The **{command_name}** skill is available but not installed.\n"
f"Install it with: `hermes skills install {install_path}`"
)
except Exception:
pass
return None
def _platform_config_key(platform: "Platform") -> str:
"""Map a Platform enum to its config.yaml key (LOCAL→"cli", rest→enum value)."""
return "cli" if platform == Platform.LOCAL else platform.value
@@ -476,13 +432,6 @@ class GatewayRunner:
self._honcho_managers: Dict[str, Any] = {}
self._honcho_configs: Dict[str, Any] = {}
# Rate-limit compression warning messages sent to users.
# Keyed by chat_id — value is the timestamp of the last warning sent.
# Prevents the warning from firing on every message when a session
# remains above the threshold after compression.
self._compression_warn_sent: Dict[str, float] = {}
self._compression_warn_cooldown: int = 3600 # seconds (1 hour)
# Ensure tirith security scanner is available (downloads if needed)
try:
from tools.tirith_security import ensure_installed
@@ -1702,11 +1651,6 @@ class GatewayRunner:
# In DMs: offer pairing code. In groups: silently ignore.
if source.chat_type == "dm" and self._get_unauthorized_dm_behavior(source.platform) == "pair":
platform_name = source.platform.value if source.platform else "unknown"
# Rate-limit ALL pairing responses (code or rejection) to
# prevent spamming the user with repeated messages when
# multiple DMs arrive in quick succession.
if self.pairing_store._is_rate_limited(platform_name, source.user_id):
return None
code = self.pairing_store.generate_code(
platform_name, source.user_id, source.user_name or ""
)
@@ -1728,8 +1672,6 @@ class GatewayRunner:
"Too many pairing requests right now~ "
"Please try again later!"
)
# Record rate limit so subsequent messages are silently ignored
self.pairing_store._record_rate_limit(platform_name, source.user_id)
return None
# PRIORITY handling when an agent is already running for this session.
@@ -1875,13 +1817,7 @@ class GatewayRunner:
if canonical == "help":
return await self._handle_help_command(event)
if canonical == "commands":
return await self._handle_commands_command(event)
if canonical == "profile":
return await self._handle_profile_command(event)
if canonical == "status":
return await self._handle_status_command(event)
@@ -1894,9 +1830,6 @@ class GatewayRunner:
if canonical == "verbose":
return await self._handle_verbose_command(event)
if canonical == "yolo":
return await self._handle_yolo_command(event)
if canonical == "provider":
return await self._handle_provider_command(event)
@@ -2041,12 +1974,6 @@ class GatewayRunner:
if msg:
event.text = msg
# Fall through to normal message processing with skill content
else:
# Not an active skill — check if it's a known-but-disabled or
# uninstalled skill and give actionable guidance.
_unavail_msg = _check_unavailable_skill(command)
if _unavail_msg:
return _unavail_msg
except Exception as e:
logger.debug("Skill command check failed (non-fatal): %s", e)
@@ -2284,29 +2211,6 @@ class GatewayRunner:
_hyg_api_key = _hyg_runtime.get("api_key")
except Exception:
pass
# Check custom_providers per-model context_length
# (same fallback as run_agent.py lines 1171-1189).
# Must run after runtime resolution so _hyg_base_url is set.
if _hyg_config_context_length is None and _hyg_base_url:
try:
_hyg_custom_providers = _hyg_data.get("custom_providers")
if isinstance(_hyg_custom_providers, list):
for _cp in _hyg_custom_providers:
if not isinstance(_cp, dict):
continue
_cp_url = (_cp.get("base_url") or "").rstrip("/")
if _cp_url and _cp_url == _hyg_base_url.rstrip("/"):
_cp_models = _cp.get("models", {})
if isinstance(_cp_models, dict):
_cp_model_cfg = _cp_models.get(_hyg_model, {})
if isinstance(_cp_model_cfg, dict):
_cp_ctx = _cp_model_cfg.get("context_length")
if _cp_ctx is not None:
_hyg_config_context_length = int(_cp_ctx)
break
except (TypeError, ValueError):
pass
except Exception:
pass
@@ -2440,18 +2344,13 @@ class GatewayRunner:
pass
# Still too large after compression — warn user
# Rate-limited to once per cooldown period per
# chat to avoid spamming on every message.
if _new_tokens >= _warn_token_threshold:
logger.warning(
"Session hygiene: still ~%s tokens after "
"compression — suggesting /reset",
f"{_new_tokens:,}",
)
_now = time.time()
_last_warn = self._compression_warn_sent.get(source.chat_id, 0)
if _hyg_adapter and _now - _last_warn >= self._compression_warn_cooldown:
self._compression_warn_sent[source.chat_id] = _now
if _hyg_adapter:
try:
await _hyg_adapter.send(
source.chat_id,
@@ -2473,10 +2372,7 @@ class GatewayRunner:
if _approx_tokens >= _warn_token_threshold:
_hyg_adapter = self.adapters.get(source.platform)
_hyg_meta = {"thread_id": source.thread_id} if source.thread_id else None
_now = time.time()
_last_warn = self._compression_warn_sent.get(source.chat_id, 0)
if _hyg_adapter and _now - _last_warn >= self._compression_warn_cooldown:
self._compression_warn_sent[source.chat_id] = _now
if _hyg_adapter:
try:
await _hyg_adapter.send(
source.chat_id,
@@ -3103,36 +2999,6 @@ class GatewayRunner:
return f"{header}\n\n{session_info}"
return header
async def _handle_profile_command(self, event: MessageEvent) -> str:
"""Handle /profile — show active profile name and home directory."""
from hermes_constants import get_hermes_home, display_hermes_home
from pathlib import Path
home = get_hermes_home()
display = display_hermes_home()
# Detect profile name from HERMES_HOME path
# Profile paths look like: ~/.hermes/profiles/<name>
profiles_parent = Path.home() / ".hermes" / "profiles"
try:
rel = home.relative_to(profiles_parent)
profile_name = str(rel).split("/")[0]
except ValueError:
profile_name = None
if profile_name:
lines = [
f"👤 **Profile:** `{profile_name}`",
f"📂 **Home:** `{display}`",
]
else:
lines = [
"👤 **Profile:** default",
f"📂 **Home:** `{display}`",
]
return "\n".join(lines)
async def _handle_status_command(self, event: MessageEvent) -> str:
"""Handle /status command."""
source = event.source
@@ -3199,68 +3065,11 @@ class GatewayRunner:
from agent.skill_commands import get_skill_commands
skill_cmds = get_skill_commands()
if skill_cmds:
lines.append(f"\n⚡ **Skill Commands** ({len(skill_cmds)} active):")
# Show first 10, then point to /commands for the rest
sorted_cmds = sorted(skill_cmds)
for cmd in sorted_cmds[:10]:
lines.append(f"`{cmd}` — {skill_cmds[cmd]['description']}")
if len(sorted_cmds) > 10:
lines.append(f"\n... and {len(sorted_cmds) - 10} more. Use `/commands` for the full paginated list.")
except Exception:
pass
return "\n".join(lines)
async def _handle_commands_command(self, event: MessageEvent) -> str:
"""Handle /commands [page] - paginated list of all commands and skills."""
from hermes_cli.commands import gateway_help_lines
raw_args = event.get_command_args().strip()
if raw_args:
try:
requested_page = int(raw_args)
except ValueError:
return "Usage: `/commands [page]`"
else:
requested_page = 1
# Build combined entry list: built-in commands + skill commands
entries = list(gateway_help_lines())
try:
from agent.skill_commands import get_skill_commands
skill_cmds = get_skill_commands()
if skill_cmds:
entries.append("")
entries.append("⚡ **Skill Commands**:")
lines.append(f"\n⚡ **Skill Commands** ({len(skill_cmds)} installed):")
for cmd in sorted(skill_cmds):
desc = skill_cmds[cmd].get("description", "").strip() or "Skill command"
entries.append(f"`{cmd}` — {desc}")
lines.append(f"`{cmd}` — {skill_cmds[cmd]['description']}")
except Exception:
pass
if not entries:
return "No commands available."
from gateway.config import Platform
page_size = 15 if event.source.platform == Platform.TELEGRAM else 20
total_pages = max(1, (len(entries) + page_size - 1) // page_size)
page = max(1, min(requested_page, total_pages))
start = (page - 1) * page_size
page_entries = entries[start:start + page_size]
lines = [
f"📚 **Commands** ({len(entries)} total, page {page}/{total_pages})",
"",
*page_entries,
]
if total_pages > 1:
nav_parts = []
if page > 1:
nav_parts.append(f"`/commands {page - 1}` ← prev")
if page < total_pages:
nav_parts.append(f"next → `/commands {page + 1}`")
lines.extend(["", " | ".join(nav_parts)])
if page != requested_page:
lines.append(f"_(Requested page {requested_page} was out of range, showing page {page}.)_")
return "\n".join(lines)
async def _handle_provider_command(self, event: MessageEvent) -> str:
@@ -4190,16 +3999,6 @@ class GatewayRunner:
else:
return f"🧠 ✓ Reasoning effort set to `{effort}` (this session only)"
async def _handle_yolo_command(self, event: MessageEvent) -> str:
"""Handle /yolo — toggle dangerous command approval bypass."""
current = bool(os.environ.get("HERMES_YOLO_MODE"))
if current:
os.environ.pop("HERMES_YOLO_MODE", None)
return "⚠️ YOLO mode **OFF** — dangerous commands will require approval."
else:
os.environ["HERMES_YOLO_MODE"] = "1"
return "⚡ YOLO mode **ON** — all commands auto-approved. Use with caution."
async def _handle_verbose_command(self, event: MessageEvent) -> str:
"""Handle /verbose command — cycle tool progress display mode.
@@ -4695,10 +4494,6 @@ class GatewayRunner:
import shutil
import subprocess
from datetime import datetime
from hermes_cli.config import is_managed, format_managed_message
if is_managed():
return f"{format_managed_message('update Hermes Agent')}"
project_root = Path(__file__).parent.parent.resolve()
git_dir = project_root / '.git'
+1 -2
View File
@@ -432,11 +432,10 @@ def build_welcome_banner(console: Console, model: str, cwd: str,
try:
behind = get_update_result(timeout=0.5)
if behind and behind > 0:
from hermes_cli.config import recommended_update_command
commits_word = "commit" if behind == 1 else "commits"
right_lines.append(
f"[bold yellow]⚠ {behind} {commits_word} behind[/]"
f"[dim yellow] — run [bold]{recommended_update_command()}[/bold] to update[/]"
f"[dim yellow] — run [bold]hermes update[/bold] to update[/]"
)
except Exception:
pass # Never break the banner over an update check
+4 -252
View File
@@ -4,19 +4,14 @@ Usage:
hermes claw migrate # Interactive migration from ~/.openclaw
hermes claw migrate --dry-run # Preview what would be migrated
hermes claw migrate --preset full --overwrite # Full migration, overwrite conflicts
hermes claw cleanup # Archive leftover OpenClaw directories
hermes claw cleanup --dry-run # Preview what would be archived
"""
import importlib.util
import logging
import shutil
import sys
from datetime import datetime
from pathlib import Path
from hermes_cli.config import get_hermes_home, get_config_path, load_config, save_config
from hermes_constants import get_optional_skills_dir
from hermes_cli.setup import (
Colors,
color,
@@ -24,7 +19,6 @@ from hermes_cli.setup import (
print_info,
print_success,
print_error,
print_warning,
prompt_yes_no,
)
@@ -33,7 +27,8 @@ logger = logging.getLogger(__name__)
PROJECT_ROOT = Path(__file__).parent.parent.resolve()
_OPENCLAW_SCRIPT = (
get_optional_skills_dir(PROJECT_ROOT / "optional-skills")
PROJECT_ROOT
/ "optional-skills"
/ "migration"
/ "openclaw-migration"
/ "scripts"
@@ -50,18 +45,6 @@ _OPENCLAW_SCRIPT_INSTALLED = (
/ "openclaw_to_hermes.py"
)
# Known OpenClaw directory names (current + legacy)
_OPENCLAW_DIR_NAMES = (".openclaw", ".clawdbot", ".moldbot")
# State files commonly found in OpenClaw workspace directories that cause
# confusion after migration (the agent discovers them and writes to them)
_WORKSPACE_STATE_GLOBS = (
"*/todo.json",
"*/sessions/*",
"*/memory/*.json",
"*/logs/*",
)
def _find_migration_script() -> Path | None:
"""Find the openclaw_to_hermes.py script in known locations."""
@@ -88,88 +71,19 @@ def _load_migration_module(script_path: Path):
return mod
def _find_openclaw_dirs() -> list[Path]:
"""Find all OpenClaw directories on disk."""
found = []
for name in _OPENCLAW_DIR_NAMES:
candidate = Path.home() / name
if candidate.is_dir():
found.append(candidate)
return found
def _scan_workspace_state(source_dir: Path) -> list[tuple[Path, str]]:
"""Scan an OpenClaw directory for workspace state files that cause confusion.
Returns a list of (path, description) tuples.
"""
findings: list[tuple[Path, str]] = []
# Direct state files in the root
for name in ("todo.json", "sessions", "logs"):
candidate = source_dir / name
if candidate.exists():
kind = "directory" if candidate.is_dir() else "file"
findings.append((candidate, f"Root {kind}: {name}"))
# State files inside workspace directories
for child in sorted(source_dir.iterdir()):
if not child.is_dir() or child.name.startswith("."):
continue
# Check for workspace-like subdirectories
for state_name in ("todo.json", "sessions", "logs", "memory"):
state_path = child / state_name
if state_path.exists():
kind = "directory" if state_path.is_dir() else "file"
rel = state_path.relative_to(source_dir)
findings.append((state_path, f"Workspace {kind}: {rel}"))
return findings
def _archive_directory(source_dir: Path, dry_run: bool = False) -> Path:
"""Rename an OpenClaw directory to .pre-migration.
Returns the archive path.
"""
timestamp = datetime.now().strftime("%Y%m%d")
archive_name = f"{source_dir.name}.pre-migration"
archive_path = source_dir.parent / archive_name
# If archive already exists, add timestamp
if archive_path.exists():
archive_name = f"{source_dir.name}.pre-migration-{timestamp}"
archive_path = source_dir.parent / archive_name
# If still exists (multiple runs same day), add counter
counter = 2
while archive_path.exists():
archive_name = f"{source_dir.name}.pre-migration-{timestamp}-{counter}"
archive_path = source_dir.parent / archive_name
counter += 1
if not dry_run:
source_dir.rename(archive_path)
return archive_path
def claw_command(args):
"""Route hermes claw subcommands."""
action = getattr(args, "claw_action", None)
if action == "migrate":
_cmd_migrate(args)
elif action in ("cleanup", "clean"):
_cmd_cleanup(args)
else:
print("Usage: hermes claw <command> [options]")
print("Usage: hermes claw migrate [options]")
print()
print("Commands:")
print(" migrate Migrate settings from OpenClaw to Hermes")
print(" cleanup Archive leftover OpenClaw directories after migration")
print()
print("Run 'hermes claw <command> --help' for options.")
print("Run 'hermes claw migrate --help' for migration options.")
def _cmd_migrate(args):
@@ -296,168 +210,6 @@ def _cmd_migrate(args):
# Print results
_print_migration_report(report, dry_run)
# After successful non-dry-run migration, offer to archive the source directory
if not dry_run and report.get("summary", {}).get("migrated", 0) > 0:
_offer_source_archival(source_dir, getattr(args, "yes", False))
def _offer_source_archival(source_dir: Path, auto_yes: bool = False):
"""After migration, offer to rename the source directory to prevent state fragmentation.
OpenClaw workspace directories contain state files (todo.json, sessions, etc.)
that the agent may discover and write to, causing confusion. Renaming the
directory prevents this.
"""
if not source_dir.is_dir():
return
# Scan for state files that could cause problems
state_files = _scan_workspace_state(source_dir)
print()
print_header("Post-Migration Cleanup")
print_info("The OpenClaw directory still exists and contains workspace state files")
print_info("that can confuse the agent (todo lists, sessions, logs).")
if state_files:
print()
print(color(" Found state files:", Colors.YELLOW))
# Show up to 10 most relevant findings
for path, desc in state_files[:10]:
print(f" {desc}")
if len(state_files) > 10:
print(f" ... and {len(state_files) - 10} more")
print()
print_info(f"Recommend: rename {source_dir.name}/ to {source_dir.name}.pre-migration/")
print_info("This prevents the agent from discovering old workspace directories.")
print_info("You can always rename it back if needed.")
print()
if auto_yes or prompt_yes_no(f"Archive {source_dir} now?", default=True):
try:
archive_path = _archive_directory(source_dir)
print_success(f"Archived: {source_dir}{archive_path}")
print_info("The original directory has been renamed, not deleted.")
print_info(f"To undo: mv {archive_path} {source_dir}")
except OSError as e:
print_error(f"Could not archive: {e}")
print_info(f"You can do it manually: mv {source_dir} {source_dir}.pre-migration")
else:
print_info("Skipped. You can archive later with: hermes claw cleanup")
def _cmd_cleanup(args):
"""Archive leftover OpenClaw directories after migration.
Scans for OpenClaw directories that still exist after migration and offers
to rename them to .pre-migration to prevent state fragmentation.
"""
dry_run = getattr(args, "dry_run", False)
auto_yes = getattr(args, "yes", False)
explicit_source = getattr(args, "source", None)
print()
print(
color(
"┌─────────────────────────────────────────────────────────┐",
Colors.MAGENTA,
)
)
print(
color(
"│ ⚕ Hermes — OpenClaw Cleanup │",
Colors.MAGENTA,
)
)
print(
color(
"└─────────────────────────────────────────────────────────┘",
Colors.MAGENTA,
)
)
# Find OpenClaw directories
if explicit_source:
dirs_to_check = [Path(explicit_source)]
else:
dirs_to_check = _find_openclaw_dirs()
if not dirs_to_check:
print()
print_success("No OpenClaw directories found. Nothing to clean up.")
return
total_archived = 0
for source_dir in dirs_to_check:
print()
print_header(f"Found: {source_dir}")
# Scan for state files
state_files = _scan_workspace_state(source_dir)
# Show directory stats
try:
workspace_dirs = [
d for d in source_dir.iterdir()
if d.is_dir() and not d.name.startswith(".")
and any((d / name).exists() for name in ("todo.json", "SOUL.md", "MEMORY.md", "USER.md"))
]
except OSError:
workspace_dirs = []
if workspace_dirs:
print_info(f"Workspace directories: {len(workspace_dirs)}")
for ws in workspace_dirs[:5]:
items = []
if (ws / "todo.json").exists():
items.append("todo.json")
if (ws / "sessions").is_dir():
items.append("sessions/")
if (ws / "SOUL.md").exists():
items.append("SOUL.md")
if (ws / "MEMORY.md").exists():
items.append("MEMORY.md")
detail = ", ".join(items) if items else "empty"
print(f" {ws.name}/ ({detail})")
if len(workspace_dirs) > 5:
print(f" ... and {len(workspace_dirs) - 5} more")
if state_files:
print()
print(color(f" {len(state_files)} state file(s) that could cause confusion:", Colors.YELLOW))
for path, desc in state_files[:8]:
print(f" {desc}")
if len(state_files) > 8:
print(f" ... and {len(state_files) - 8} more")
print()
if dry_run:
archive_path = _archive_directory(source_dir, dry_run=True)
print_info(f"Would archive: {source_dir}{archive_path}")
else:
if auto_yes or prompt_yes_no(f"Archive {source_dir}?", default=True):
try:
archive_path = _archive_directory(source_dir)
print_success(f"Archived: {source_dir}{archive_path}")
total_archived += 1
except OSError as e:
print_error(f"Could not archive: {e}")
print_info(f"Try manually: mv {source_dir} {source_dir}.pre-migration")
else:
print_info("Skipped.")
# Summary
print()
if dry_run:
print_info(f"Dry run complete. {len(dirs_to_check)} directory(ies) would be archived.")
print_info("Run without --dry-run to archive them.")
elif total_archived:
print_success(f"Cleaned up {total_archived} OpenClaw directory(ies).")
print_info("Directories were renamed, not deleted. You can undo by renaming them back.")
else:
print_info("No directories were archived.")
def _print_migration_report(report: dict, dry_run: bool):
"""Print a formatted migration report."""
+2 -18
View File
@@ -1,24 +1,8 @@
"""Shared ANSI color utilities for Hermes CLI modules."""
import os
import sys
def should_use_color() -> bool:
"""Return True when colored output is appropriate.
Respects the NO_COLOR environment variable (https://no-color.org/)
and TERM=dumb, in addition to the existing TTY check.
"""
if os.environ.get("NO_COLOR") is not None:
return False
if os.environ.get("TERM") == "dumb":
return False
if not sys.stdout.isatty():
return False
return True
class Colors:
RESET = "\033[0m"
BOLD = "\033[1m"
@@ -32,7 +16,7 @@ class Colors:
def color(text: str, *codes) -> str:
"""Apply color codes to text (only when color output is appropriate)."""
if not should_use_color():
"""Apply color codes to text (only when output is a TTY)."""
if not sys.stdout.isatty():
return text
return "".join(codes) + text + Colors.RESET
-68
View File
@@ -71,7 +71,6 @@ COMMAND_REGISTRY: list[CommandDef] = [
aliases=("q",), args_hint="<prompt>"),
CommandDef("status", "Show session info", "Session",
gateway_only=True),
CommandDef("profile", "Show active profile name and home directory", "Info"),
CommandDef("sethome", "Set this chat as the home channel", "Session",
gateway_only=True, aliases=("set-home",)),
CommandDef("resume", "Resume a previously-named session", "Session",
@@ -91,8 +90,6 @@ COMMAND_REGISTRY: list[CommandDef] = [
CommandDef("verbose", "Cycle tool progress display: off -> new -> all -> verbose",
"Configuration", cli_only=True,
gateway_config_gate="display.tool_progress_command"),
CommandDef("yolo", "Toggle YOLO mode (skip all dangerous command approvals)",
"Configuration"),
CommandDef("reasoning", "Manage reasoning effort and display", "Configuration",
args_hint="[level|show|hide]",
subcommands=("none", "low", "minimal", "medium", "high", "xhigh", "show", "hide", "on", "off")),
@@ -121,8 +118,6 @@ COMMAND_REGISTRY: list[CommandDef] = [
"Tools & Skills", cli_only=True),
# Info
CommandDef("commands", "Browse all commands and skills (paginated)", "Info",
gateway_only=True, args_hint="[page]"),
CommandDef("help", "Show available commands", "Info"),
CommandDef("usage", "Show token usage for the current session", "Info"),
CommandDef("insights", "Show usage insights and analytics", "Info",
@@ -366,69 +361,6 @@ def telegram_bot_commands() -> list[tuple[str, str]]:
return result
def telegram_menu_commands(max_commands: int = 100) -> tuple[list[tuple[str, str]], int]:
"""Return Telegram menu commands capped to the Bot API limit.
Priority order (higher priority = never bumped by overflow):
1. Core CommandDef commands (always included)
2. Plugin slash commands (take precedence over skills)
3. Built-in skill commands (fill remaining slots, alphabetical)
Skills are the only tier that gets trimmed when the cap is hit.
User-installed hub skills are excluded — accessible via /skills.
Returns:
(menu_commands, hidden_count) where hidden_count is the number of
skill commands omitted due to the cap.
"""
all_commands = list(telegram_bot_commands())
# Plugin slash commands get priority over skills
try:
from hermes_cli.plugins import get_plugin_manager
pm = get_plugin_manager()
plugin_cmds = getattr(pm, "_plugin_commands", {})
for cmd_name in sorted(plugin_cmds):
tg_name = cmd_name.replace("-", "_")
desc = "Plugin command"
if len(desc) > 40:
desc = desc[:37] + "..."
all_commands.append((tg_name, desc))
except Exception:
pass
# Remaining slots go to built-in skill commands (not hub-installed).
skill_entries: list[tuple[str, str]] = []
try:
from agent.skill_commands import get_skill_commands
from tools.skills_tool import SKILLS_DIR
_skills_dir = str(SKILLS_DIR.resolve())
_hub_dir = str((SKILLS_DIR / ".hub").resolve())
skill_cmds = get_skill_commands()
for cmd_key in sorted(skill_cmds):
info = skill_cmds[cmd_key]
skill_path = info.get("skill_md_path", "")
if not skill_path.startswith(_skills_dir):
continue
if skill_path.startswith(_hub_dir):
continue
name = cmd_key.lstrip("/").replace("-", "_")
desc = info.get("description", "")
# Keep descriptions short — setMyCommands has an undocumented
# total payload limit. 40 chars fits 100 commands safely.
if len(desc) > 40:
desc = desc[:37] + "..."
skill_entries.append((name, desc))
except Exception:
pass
# Skills fill remaining slots — they're the only tier that gets trimmed
remaining_slots = max(0, max_commands - len(all_commands))
hidden_count = max(0, len(skill_entries) - remaining_slots)
all_commands.extend(skill_entries[:remaining_slots])
return all_commands[:max_commands], hidden_count
def slack_subcommand_map() -> dict[str, str]:
"""Return subcommand -> /command mapping for Slack /hermes handler.
+11 -79
View File
@@ -52,86 +52,26 @@ from hermes_cli.default_soul import DEFAULT_SOUL_MD
# Managed mode (NixOS declarative config)
# =============================================================================
_MANAGED_TRUE_VALUES = ("true", "1", "yes")
_MANAGED_SYSTEM_NAMES = {
"brew": "Homebrew",
"homebrew": "Homebrew",
"nix": "NixOS",
"nixos": "NixOS",
}
def get_managed_system() -> Optional[str]:
"""Return the package manager owning this install, if any."""
raw = os.getenv("HERMES_MANAGED", "").strip()
if raw:
normalized = raw.lower()
if normalized in _MANAGED_TRUE_VALUES:
return "NixOS"
return _MANAGED_SYSTEM_NAMES.get(normalized, raw)
managed_marker = get_hermes_home() / ".managed"
if managed_marker.exists():
return "NixOS"
return None
def is_managed() -> bool:
"""Check if Hermes is running in package-manager-managed mode.
"""Check if hermes is running in Nix-managed mode.
Two signals: the HERMES_MANAGED env var (set by the systemd service),
or a .managed marker file in HERMES_HOME (set by the NixOS activation
script, so interactive shells also see it).
"""
return get_managed_system() is not None
def get_managed_update_command() -> Optional[str]:
"""Return the preferred upgrade command for a managed install."""
managed_system = get_managed_system()
if managed_system == "Homebrew":
return "brew upgrade hermes-agent"
if managed_system == "NixOS":
return "sudo nixos-rebuild switch"
return None
def recommended_update_command() -> str:
"""Return the best update command for the current installation."""
return get_managed_update_command() or "hermes update"
def format_managed_message(action: str = "modify this Hermes installation") -> str:
"""Build a user-facing error for managed installs."""
managed_system = get_managed_system() or "a package manager"
raw = os.getenv("HERMES_MANAGED", "").strip().lower()
if managed_system == "NixOS":
env_hint = "true" if raw in _MANAGED_TRUE_VALUES else raw or "true"
return (
f"Cannot {action}: this Hermes installation is managed by NixOS "
f"(HERMES_MANAGED={env_hint}).\n"
"Edit services.hermes-agent.settings in your configuration.nix and run:\n"
" sudo nixos-rebuild switch"
)
if managed_system == "Homebrew":
env_hint = raw or "homebrew"
return (
f"Cannot {action}: this Hermes installation is managed by Homebrew "
f"(HERMES_MANAGED={env_hint}).\n"
"Use:\n"
" brew upgrade hermes-agent"
)
return (
f"Cannot {action}: this Hermes installation is managed by {managed_system}.\n"
"Use your package manager to upgrade or reinstall Hermes."
)
if os.getenv("HERMES_MANAGED", "").lower() in ("true", "1", "yes"):
return True
managed_marker = get_hermes_home() / ".managed"
return managed_marker.exists()
def managed_error(action: str = "modify configuration"):
"""Print user-friendly error for managed mode."""
print(format_managed_message(action), file=sys.stderr)
print(
f"Cannot {action}: configuration is managed by NixOS (HERMES_MANAGED=true).\n"
"Edit services.hermes-agent.settings in your configuration.nix and run:\n"
" sudo nixos-rebuild switch",
file=sys.stderr,
)
# =============================================================================
@@ -766,14 +706,6 @@ OPTIONAL_ENV_VARS = {
"password": True,
"category": "tool",
},
"CAMOFOX_URL": {
"description": "Camofox browser server URL for local anti-detection browsing (e.g. http://localhost:9377)",
"prompt": "Camofox server URL",
"url": "https://github.com/jo-inc/camofox-browser",
"tools": ["browser_navigate", "browser_click"],
"password": False,
"category": "tool",
},
"FAL_KEY": {
"description": "FAL API key for image generation",
"prompt": "FAL API key",
+8 -15
View File
@@ -406,11 +406,8 @@ def run_doctor(args):
if terminal_env == "docker":
if shutil.which("docker"):
# Check if docker daemon is running
try:
result = subprocess.run(["docker", "info"], capture_output=True, timeout=10)
except subprocess.TimeoutExpired:
result = None
if result is not None and result.returncode == 0:
result = subprocess.run(["docker", "info"], capture_output=True)
if result.returncode == 0:
check_ok("docker", "(daemon running)")
else:
check_fail("docker daemon not running")
@@ -429,16 +426,12 @@ def run_doctor(args):
ssh_host = os.getenv("TERMINAL_SSH_HOST")
if ssh_host:
# Try to connect
try:
result = subprocess.run(
["ssh", "-o", "ConnectTimeout=5", "-o", "BatchMode=yes", ssh_host, "echo ok"],
capture_output=True,
text=True,
timeout=15
)
except subprocess.TimeoutExpired:
result = None
if result is not None and result.returncode == 0:
result = subprocess.run(
["ssh", "-o", "ConnectTimeout=5", "-o", "BatchMode=yes", ssh_host, "echo ok"],
capture_output=True,
text=True
)
if result.returncode == 0:
check_ok(f"SSH connection to {ssh_host}")
else:
check_fail(f"SSH connection to {ssh_host}")
+2 -33
View File
@@ -2467,14 +2467,10 @@ def cmd_version(args):
# Show update status (synchronous — acceptable since user asked for version info)
try:
from hermes_cli.banner import check_for_updates
from hermes_cli.config import recommended_update_command
behind = check_for_updates()
if behind and behind > 0:
commits_word = "commit" if behind == 1 else "commits"
print(
f"Update available: {behind} {commits_word} behind — "
f"run '{recommended_update_command()}'"
)
print(f"Update available: {behind} {commits_word} behind — run 'hermes update'")
elif behind == 0:
print("Up to date")
except Exception:
@@ -2825,11 +2821,6 @@ def _invalidate_update_cache():
def cmd_update(args):
"""Update Hermes Agent to the latest version."""
import shutil
from hermes_cli.config import is_managed, managed_error
if is_managed():
managed_error("update Hermes Agent")
return
print("⚕ Updating Hermes Agent...")
print()
@@ -3038,7 +3029,7 @@ def cmd_update(args):
import shutil
if shutil.which("npm"):
print("→ Updating Node.js dependencies...")
subprocess.run(["npm", "ci", "--silent"], cwd=PROJECT_ROOT, check=False)
subprocess.run(["npm", "install", "--silent"], cwd=PROJECT_ROOT, check=False)
print()
print("✓ Code updated!")
@@ -4712,28 +4703,6 @@ For more help on a command:
help="Skip confirmation prompts"
)
# claw cleanup
claw_cleanup = claw_subparsers.add_parser(
"cleanup",
aliases=["clean"],
help="Archive leftover OpenClaw directories after migration",
description="Scan for and archive leftover OpenClaw directories to prevent state fragmentation"
)
claw_cleanup.add_argument(
"--source",
help="Path to a specific OpenClaw directory to clean up"
)
claw_cleanup.add_argument(
"--dry-run",
action="store_true",
help="Preview what would be archived without making changes"
)
claw_cleanup.add_argument(
"--yes", "-y",
action="store_true",
help="Skip confirmation prompts"
)
def cmd_claw(args):
from hermes_cli.claw import claw_command
claw_command(args)
+1 -2
View File
@@ -265,11 +265,10 @@ def cmd_install(identifier: str, force: bool = False) -> None:
)
sys.exit(1)
if mv_int > _SUPPORTED_MANIFEST_VERSION:
from hermes_cli.config import recommended_update_command
console.print(
f"[red]Error:[/red] Plugin '{plugin_name}' requires manifest_version "
f"{mv}, but this installer only supports up to {_SUPPORTED_MANIFEST_VERSION}.\n"
f"Run [bold]{recommended_update_command()}[/bold] to get a newer installer."
f"Run [bold]hermes update[/bold] to get a newer installer."
)
sys.exit(1)
+4 -7
View File
@@ -18,8 +18,6 @@ import sys
from pathlib import Path
from typing import Optional, Dict, Any
from hermes_constants import get_optional_skills_dir
logger = logging.getLogger(__name__)
PROJECT_ROOT = Path(__file__).parent.parent.resolve()
@@ -603,15 +601,13 @@ def _print_setup_summary(config: dict, hermes_home):
Path(__file__).parent.parent / "node_modules" / ".bin" / "agent-browser"
).exists()
)
if get_env_value("CAMOFOX_URL"):
tool_status.append(("Browser Automation (Camofox)", True, None))
elif get_env_value("BROWSERBASE_API_KEY"):
if get_env_value("BROWSERBASE_API_KEY"):
tool_status.append(("Browser Automation (Browserbase)", True, None))
elif _ab_found:
tool_status.append(("Browser Automation (local)", True, None))
else:
tool_status.append(
("Browser Automation", False, "npm install -g agent-browser or set CAMOFOX_URL")
("Browser Automation", False, "npm install -g agent-browser")
)
# FAL (image generation)
@@ -3123,7 +3119,8 @@ def _skip_configured_section(
_OPENCLAW_SCRIPT = (
get_optional_skills_dir(PROJECT_ROOT / "optional-skills")
PROJECT_ROOT
/ "optional-skills"
/ "migration"
/ "openclaw-migration"
/ "scripts"
+12 -20
View File
@@ -285,31 +285,23 @@ def show_status(args):
_gw_svc = get_service_name()
except Exception:
_gw_svc = "hermes-gateway"
try:
result = subprocess.run(
["systemctl", "--user", "is-active", _gw_svc],
capture_output=True,
text=True,
timeout=5
)
is_active = result.stdout.strip() == "active"
except subprocess.TimeoutExpired:
is_active = False
result = subprocess.run(
["systemctl", "--user", "is-active", _gw_svc],
capture_output=True,
text=True
)
is_active = result.stdout.strip() == "active"
print(f" Status: {check_mark(is_active)} {'running' if is_active else 'stopped'}")
print(" Manager: systemd (user)")
elif sys.platform == 'darwin':
from hermes_cli.gateway import get_launchd_label
try:
result = subprocess.run(
["launchctl", "list", get_launchd_label()],
capture_output=True,
text=True,
timeout=5
)
is_loaded = result.returncode == 0
except subprocess.TimeoutExpired:
is_loaded = False
result = subprocess.run(
["launchctl", "list", get_launchd_label()],
capture_output=True,
text=True
)
is_loaded = result.returncode == 0
print(f" Status: {check_mark(is_loaded)} {'loaded' if is_loaded else 'not loaded'}")
print(" Manager: launchd")
else:
+1 -35
View File
@@ -273,16 +273,6 @@ TOOL_CATEGORIES = {
"browser_provider": "browser-use",
"post_setup": "browserbase",
},
{
"name": "Camofox",
"tag": "Local anti-detection browser (Firefox/Camoufox)",
"env_vars": [
{"key": "CAMOFOX_URL", "prompt": "Camofox server URL", "default": "http://localhost:9377",
"url": "https://github.com/jo-inc/camofox-browser"},
],
"browser_provider": "camofox",
"post_setup": "camofox",
},
],
},
"homeassistant": {
@@ -347,28 +337,6 @@ def _run_post_setup(post_setup_key: str):
elif not node_modules.exists():
_print_warning(" Node.js not found - browser tools require: npm install (in hermes-agent directory)")
elif post_setup_key == "camofox":
camofox_dir = PROJECT_ROOT / "node_modules" / "@askjo" / "camoufox-browser"
if not camofox_dir.exists() and shutil.which("npm"):
_print_info(" Installing Camofox browser server...")
import subprocess
result = subprocess.run(
["npm", "install", "--silent"],
capture_output=True, text=True, cwd=str(PROJECT_ROOT)
)
if result.returncode == 0:
_print_success(" Camofox installed")
else:
_print_warning(" npm install failed - run manually: npm install")
if camofox_dir.exists():
_print_info(" Start the Camofox server:")
_print_info(" npx @askjo/camoufox-browser")
_print_info(" First run downloads the Camoufox engine (~300MB)")
_print_info(" Or use Docker: docker run -p 9377:9377 jo-inc/camofox-browser")
elif not shutil.which("npm"):
_print_warning(" Node.js not found. Install Camofox via Docker:")
_print_info(" docker run -p 9377:9377 jo-inc/camofox-browser")
elif post_setup_key == "rl_training":
try:
__import__("tinker_atropos")
@@ -597,9 +565,7 @@ def _toolset_has_keys(ts_key: str) -> bool:
if cat:
for provider in cat.get("providers", []):
env_vars = provider.get("env_vars", [])
if not env_vars:
return True # No-key provider (e.g. Local Browser, Edge TTS)
if all(get_env_value(e["key"]) for e in env_vars):
if env_vars and all(get_env_value(e["key"]) for e in env_vars):
return True
return False
-14
View File
@@ -17,20 +17,6 @@ def get_hermes_home() -> Path:
return Path(os.getenv("HERMES_HOME", Path.home() / ".hermes"))
def get_optional_skills_dir(default: Path | None = None) -> Path:
"""Return the optional-skills directory, honoring package-manager wrappers.
Packaged installs may ship ``optional-skills`` outside the Python package
tree and expose it via ``HERMES_OPTIONAL_SKILLS``.
"""
override = os.getenv("HERMES_OPTIONAL_SKILLS", "").strip()
if override:
return Path(override)
if default is not None:
return default
return get_hermes_home() / "optional-skills"
def get_hermes_dir(new_subpath: str, old_name: str) -> Path:
"""Resolve a Hermes subdirectory with backward compatibility.
+8 -22
View File
@@ -10,27 +10,16 @@ import os
import sys
from pathlib import Path
from hermes_constants import get_hermes_home
from honcho_integration.client import resolve_config_path, GLOBAL_CONFIG_PATH
HOST = "hermes"
def _config_path() -> Path:
"""Return the active Honcho config path for reading (instance-local or global)."""
"""Return the active Honcho config path (instance-local or global)."""
return resolve_config_path()
def _local_config_path() -> Path:
"""Return the instance-local Honcho config path for writing.
Always returns $HERMES_HOME/honcho.json so each profile/instance gets
its own config file. The global ~/.honcho/config.json is only used as
a read fallback (via resolve_config_path) for cross-app interop.
"""
return get_hermes_home() / "honcho.json"
def _read_config() -> dict:
path = _config_path()
if path.exists():
@@ -42,7 +31,7 @@ def _read_config() -> dict:
def _write_config(cfg: dict, path: Path | None = None) -> None:
path = path or _local_config_path()
path = path or _config_path()
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(
json.dumps(cfg, indent=2, ensure_ascii=False) + "\n",
@@ -106,13 +95,13 @@ def cmd_setup(args) -> None:
"""Interactive Honcho setup wizard."""
cfg = _read_config()
write_path = _local_config_path()
read_path = _config_path()
active_path = _config_path()
print("\nHoncho memory setup\n" + "" * 40)
print(" Honcho gives Hermes persistent cross-session memory.")
print(f" Config: {write_path}")
if read_path != write_path and read_path.exists():
print(f" (seeding from existing config at {read_path})")
if active_path != GLOBAL_CONFIG_PATH:
print(f" Instance config: {active_path}")
else:
print(" Config is shared with other hosts at ~/.honcho/config.json")
print()
if not _ensure_sdk_installed():
@@ -200,7 +189,7 @@ def cmd_setup(args) -> None:
hermes_host.setdefault("saveMessages", True)
_write_config(cfg)
print(f"\n Config written to {write_path}")
print(f"\n Config written to {active_path}")
# Test connection
print(" Testing connection... ", end="", flush=True)
@@ -248,7 +237,6 @@ def cmd_status(args) -> None:
cfg = _read_config()
active_path = _config_path()
write_path = _local_config_path()
if not cfg:
print(f" No Honcho config found at {active_path}")
@@ -271,8 +259,6 @@ def cmd_status(args) -> None:
print(f" Workspace: {hcfg.workspace_id}")
print(f" Host: {hcfg.host}")
print(f" Config path: {active_path}")
if write_path != active_path:
print(f" Write path: {write_path} (instance-local)")
print(f" AI peer: {hcfg.ai_peer}")
print(f" User peer: {hcfg.peer_name or 'not set'}")
print(f" Session key: {hcfg.resolve_session_name()}")
@@ -2455,24 +2455,9 @@ class Migrator:
notes.append("")
notes.extend([
"## IMPORTANT: Archive the OpenClaw Directory",
"",
"After migration, your OpenClaw directory still exists on disk with workspace",
"state files (todo.json, sessions, logs). If the Hermes agent discovers these",
"directories, it may read/write to them instead of the Hermes state, causing",
"confusion (e.g., cron jobs reading a different todo list than interactive sessions).",
"",
"**Strongly recommended:** Run `hermes claw cleanup` to rename the OpenClaw",
"directory to `.openclaw.pre-migration`. This prevents the agent from finding it.",
"The directory is renamed, not deleted — you can undo this at any time.",
"",
"If you skip this step and notice the agent getting confused about workspaces",
"or todo lists, run `hermes claw cleanup` to fix it.",
"",
"## Hermes-Specific Setup",
"",
"After migration, you may want to:",
"- Run `hermes claw cleanup` to archive the OpenClaw directory (prevents state confusion)",
"- Run `hermes setup` to configure any remaining settings",
"- Run `hermes mcp list` to verify MCP servers were imported correctly",
"- Run `hermes cron` to recreate scheduled tasks (see archive/cron-config.json)",
+1 -2
View File
@@ -16,8 +16,7 @@
},
"homepage": "https://github.com/NousResearch/Hermes-Agent#readme",
"dependencies": {
"agent-browser": "^0.13.0",
"@askjo/camoufox-browser": "^1.0.0"
"agent-browser": "^0.13.0"
},
"engines": {
"node": ">=18.0.0"
-14
View File
@@ -1,14 +0,0 @@
Homebrew packaging notes for Hermes Agent.
Use `packaging/homebrew/hermes-agent.rb` as a tap or `homebrew-core` starting point.
Key choices:
- Stable builds should target the semver-named sdist asset attached to each GitHub release, not the CalVer tag tarball.
- `faster-whisper` now lives in the `voice` extra, which keeps wheel-only transitive dependencies out of the base Homebrew formula.
- The wrapper exports `HERMES_BUNDLED_SKILLS`, `HERMES_OPTIONAL_SKILLS`, and `HERMES_MANAGED=homebrew` so packaged installs keep runtime assets and defer upgrades to Homebrew.
Typical update flow:
1. Bump the formula `url`, `version`, and `sha256`.
2. Refresh Python resources with `brew update-python-resources --print-only hermes-agent`.
3. Keep `ignore_packages: %w[certifi cryptography pydantic]`.
4. Verify `brew audit --new --strict hermes-agent` and `brew test hermes-agent`.
-48
View File
@@ -1,48 +0,0 @@
class HermesAgent < Formula
include Language::Python::Virtualenv
desc "Self-improving AI agent that creates skills from experience"
homepage "https://hermes-agent.nousresearch.com"
# Stable source should point at the semver-named sdist asset attached by
# scripts/release.py, not the CalVer tag tarball.
url "https://github.com/NousResearch/hermes-agent/releases/download/v2026.3.30/hermes_agent-0.6.0.tar.gz"
sha256 "<replace-with-release-asset-sha256>"
license "MIT"
depends_on "certifi" => :no_linkage
depends_on "cryptography" => :no_linkage
depends_on "libyaml"
depends_on "python@3.14"
pypi_packages ignore_packages: %w[certifi cryptography pydantic]
# Refresh resource stanzas after bumping the source url/version:
# brew update-python-resources --print-only hermes-agent
def install
venv = virtualenv_create(libexec, "python3.14")
venv.pip_install resources
venv.pip_install buildpath
pkgshare.install "skills", "optional-skills"
%w[hermes hermes-agent hermes-acp].each do |exe|
next unless (libexec/"bin"/exe).exist?
(bin/exe).write_env_script(
libexec/"bin"/exe,
HERMES_BUNDLED_SKILLS: pkgshare/"skills",
HERMES_OPTIONAL_SKILLS: pkgshare/"optional-skills",
HERMES_MANAGED: "homebrew"
)
end
end
test do
assert_match "Hermes Agent v#{version}", shell_output("#{bin}/hermes version")
managed = shell_output("#{bin}/hermes update 2>&1")
assert_match "managed by Homebrew", managed
assert_match "brew upgrade hermes-agent", managed
end
end
+2 -7
View File
@@ -32,6 +32,7 @@ dependencies = [
"fal-client>=0.13.1,<1",
# Text-to-speech (Edge TTS is free, no API key needed)
"edge-tts>=7.2.7,<8",
"faster-whisper>=1.0.0,<2",
# Skills Hub (GitHub App JWT auth — optional, only needed for bot identity)
"PyJWT[crypto]>=2.12.0,<3", # CVE-2026-32597
]
@@ -46,13 +47,7 @@ slack = ["slack-bolt>=1.18.0,<2", "slack-sdk>=3.27.0,<4"]
matrix = ["matrix-nio[e2e]>=0.24.0,<1"]
cli = ["simple-term-menu>=1.0,<2"]
tts-premium = ["elevenlabs>=1.0,<2"]
voice = [
# Local STT pulls in wheel-only transitive deps (ctranslate2, onnxruntime),
# so keep it out of the base install for source-build packagers like Homebrew.
"faster-whisper>=1.0.0,<2",
"sounddevice>=0.4.6,<1",
"numpy>=1.24.0,<3",
]
voice = ["sounddevice>=0.4.6,<1", "numpy>=1.24.0,<3"]
pty = [
"ptyprocess>=0.7.0,<1; sys_platform != 'win32'",
"pywinpty>=2.0.0,<3; sys_platform == 'win32'",
+5 -12
View File
@@ -5221,8 +5221,11 @@ class AIAgent:
except Exception as e:
logger.warning("Session DB compression split failed — new session will NOT be indexed: %s", e)
# Update token estimate after compaction so pressure calculations
# use the post-compression count, not the stale pre-compression one.
# Reset context pressure warning and token estimate — usage drops
# after compaction. Without this, the stale last_prompt_tokens from
# the previous API call causes the pressure calculation to stay at
# >1000% and spam warnings / re-trigger compression in a loop.
self._context_pressure_warned = False
_compressed_est = (
estimate_tokens_rough(new_system_prompt)
+ estimate_messages_tokens_rough(compressed)
@@ -5230,16 +5233,6 @@ class AIAgent:
self.context_compressor.last_prompt_tokens = _compressed_est
self.context_compressor.last_completion_tokens = 0
# Only reset the pressure warning if compression actually brought
# us below the warning level (85% of threshold). When compression
# can't reduce enough (e.g. threshold is very low, or system prompt
# alone exceeds the warning level), keep the flag set to prevent
# spamming the user with repeated warnings every loop iteration.
if self.context_compressor.threshold_tokens > 0:
_post_progress = _compressed_est / self.context_compressor.threshold_tokens
if _post_progress < 0.85:
self._context_pressure_warned = False
return compressed, new_system_prompt
def _execute_tool_calls(self, assistant_message, messages: list, effective_task_id: str, api_call_count: int = 0) -> None:
+3 -15
View File
@@ -94,7 +94,7 @@ print_banner() {
echo ""
echo -e "${MAGENTA}${BOLD}"
echo "┌─────────────────────────────────────────────────────────┐"
echo "│ ⚕ Hermes Agent Installer │"
echo "│ ⚕ Hermes Agent Installer │"
echo "├─────────────────────────────────────────────────────────┤"
echo "│ An open source AI agent by Nous Research. │"
echo "└─────────────────────────────────────────────────────────┘"
@@ -699,19 +699,14 @@ install_deps() {
# Install the main package in editable mode with all extras.
# Try [all] first, fall back to base install if extras have issues.
ALL_INSTALL_LOG=$(mktemp)
if ! $UV_CMD pip install -e ".[all]" 2>"$ALL_INSTALL_LOG"; then
if ! $UV_CMD pip install -e ".[all]" 2>/dev/null; then
log_warn "Full install (.[all]) failed, trying base install..."
log_info "Reason: $(tail -5 "$ALL_INSTALL_LOG" | head -3)"
rm -f "$ALL_INSTALL_LOG"
if ! $UV_CMD pip install -e "."; then
log_error "Package installation failed."
log_info "Check that build tools are installed: sudo apt install build-essential python3-dev"
log_info "Then re-run: cd $INSTALL_DIR && uv pip install -e '.[all]'"
exit 1
fi
else
rm -f "$ALL_INSTALL_LOG"
fi
log_success "Main package installed"
@@ -1075,14 +1070,7 @@ print_success() {
echo ""
echo -e "${YELLOW}⚡ Reload your shell to use 'hermes' command:${NC}"
echo ""
LOGIN_SHELL="$(basename "${SHELL:-/bin/bash}")"
if [ "$LOGIN_SHELL" = "zsh" ]; then
echo " source ~/.zshrc"
elif [ "$LOGIN_SHELL" = "bash" ]; then
echo " source ~/.bashrc"
else
echo " source ~/.bashrc # or ~/.zshrc"
fi
echo " source ~/.bashrc # or ~/.zshrc"
echo ""
# Show Node.js warning if auto-install failed
+32 -124
View File
@@ -24,7 +24,6 @@ import argparse
import json
import os
import re
import shutil
import subprocess
import sys
from collections import defaultdict
@@ -129,16 +128,6 @@ def git(*args, cwd=None):
return result.stdout.strip()
def git_result(*args, cwd=None):
"""Run a git command and return the full CompletedProcess."""
return subprocess.run(
["git"] + list(args),
capture_output=True,
text=True,
cwd=cwd or str(REPO_ROOT),
)
def get_last_tag():
"""Get the most recent CalVer tag."""
tags = git("tag", "--list", "v20*", "--sort=-v:refname")
@@ -147,18 +136,6 @@ def get_last_tag():
return None
def next_available_tag(base_tag: str) -> tuple[str, str]:
"""Return a tag/calver pair, suffixing same-day releases when needed."""
if not git("tag", "--list", base_tag):
return base_tag, base_tag.removeprefix("v")
suffix = 2
while git("tag", "--list", f"{base_tag}.{suffix}"):
suffix += 1
tag_name = f"{base_tag}.{suffix}"
return tag_name, tag_name.removeprefix("v")
def get_current_version():
"""Read current semver from __init__.py."""
content = VERSION_FILE.read_text()
@@ -215,41 +192,6 @@ def update_version_files(semver: str, calver_date: str):
PYPROJECT_FILE.write_text(pyproject)
def build_release_artifacts(semver: str) -> list[Path]:
"""Build sdist/wheel artifacts for the current release.
Returns the artifact paths when the local environment has ``python -m build``
available. If build tooling is missing or the build fails, returns an empty
list and lets the release proceed without attached Python artifacts.
"""
dist_dir = REPO_ROOT / "dist"
shutil.rmtree(dist_dir, ignore_errors=True)
result = subprocess.run(
[sys.executable, "-m", "build", "--sdist", "--wheel"],
cwd=str(REPO_ROOT),
capture_output=True,
text=True,
)
if result.returncode != 0:
print(" ⚠ Could not build Python release artifacts.")
stderr = result.stderr.strip()
stdout = result.stdout.strip()
if stderr:
print(f" {stderr.splitlines()[-1]}")
elif stdout:
print(f" {stdout.splitlines()[-1]}")
print(" Install the 'build' package to attach semver-named sdist/wheel assets.")
return []
artifacts = sorted(p for p in dist_dir.iterdir() if p.is_file())
matching = [p for p in artifacts if semver in p.name]
if not matching:
print(" ⚠ Built artifacts did not match the expected release version.")
return []
return matching
def resolve_author(name: str, email: str) -> str:
"""Resolve a git author to a GitHub @mention."""
# Try email lookup first
@@ -482,10 +424,18 @@ def main():
now = datetime.now()
calver_date = f"{now.year}.{now.month}.{now.day}"
base_tag = f"v{calver_date}"
tag_name, calver_date = next_available_tag(base_tag)
if tag_name != base_tag:
print(f"Note: Tag {base_tag} already exists, using {tag_name}")
tag_name = f"v{calver_date}"
# Check for existing tag with same date
existing = git("tag", "--list", tag_name)
if existing and not args.publish:
# Append a suffix for same-day releases
suffix = 2
while git("tag", "--list", f"{tag_name}.{suffix}"):
suffix += 1
tag_name = f"{tag_name}.{suffix}"
calver_date = f"{calver_date}.{suffix}"
print(f"Note: Tag {tag_name[:-2]} already exists, using {tag_name}")
# Determine semver
current_version = get_current_version()
@@ -544,83 +494,41 @@ def main():
print(f" ✓ Updated version files to v{new_version} ({calver_date})")
# Commit version bump
add_result = git_result("add", str(VERSION_FILE), str(PYPROJECT_FILE))
if add_result.returncode != 0:
print(f" ✗ Failed to stage version files: {add_result.stderr.strip()}")
return
commit_result = git_result(
"commit", "-m", f"chore: bump version to v{new_version} ({calver_date})"
)
if commit_result.returncode != 0:
print(f" ✗ Failed to commit version bump: {commit_result.stderr.strip()}")
return
git("add", str(VERSION_FILE), str(PYPROJECT_FILE))
git("commit", "-m", f"chore: bump version to v{new_version} ({calver_date})")
print(f" ✓ Committed version bump")
# Create annotated tag
tag_result = git_result(
"tag", "-a", tag_name, "-m",
f"Hermes Agent v{new_version} ({calver_date})\n\nWeekly release"
)
if tag_result.returncode != 0:
print(f" ✗ Failed to create tag {tag_name}: {tag_result.stderr.strip()}")
return
git("tag", "-a", tag_name, "-m",
f"Hermes Agent v{new_version} ({calver_date})\n\nWeekly release")
print(f" ✓ Created tag {tag_name}")
# Push
push_result = git_result("push", "origin", "HEAD", "--tags")
if push_result.returncode == 0:
print(f" ✓ Pushed to origin")
else:
print(f" ✗ Failed to push to origin: {push_result.stderr.strip()}")
print(" Continue manually after fixing access:")
print(" git push origin HEAD --tags")
# Build semver-named Python artifacts so downstream packagers
# (e.g. Homebrew) can target them without relying on CalVer tag names.
artifacts = build_release_artifacts(new_version)
if artifacts:
print(" ✓ Built release artifacts:")
for artifact in artifacts:
print(f" - {artifact.relative_to(REPO_ROOT)}")
push_result = git("push", "origin", "HEAD", "--tags")
print(f" ✓ Pushed to origin")
# Create GitHub release
changelog_file = REPO_ROOT / ".release_notes.md"
changelog_file.write_text(changelog)
gh_cmd = [
"gh", "release", "create", tag_name,
"--title", f"Hermes Agent v{new_version} ({calver_date})",
"--notes-file", str(changelog_file),
]
gh_cmd.extend(str(path) for path in artifacts)
result = subprocess.run(
["gh", "release", "create", tag_name,
"--title", f"Hermes Agent v{new_version} ({calver_date})",
"--notes-file", str(changelog_file)],
capture_output=True, text=True,
cwd=str(REPO_ROOT),
)
gh_bin = shutil.which("gh")
if gh_bin:
result = subprocess.run(
gh_cmd,
capture_output=True, text=True,
cwd=str(REPO_ROOT),
)
else:
result = None
changelog_file.unlink(missing_ok=True)
if result and result.returncode == 0:
changelog_file.unlink(missing_ok=True)
if result.returncode == 0:
print(f" ✓ GitHub release created: {result.stdout.strip()}")
print(f"\n 🎉 Release v{new_version} ({tag_name}) published!")
else:
if result is None:
print(" ✗ GitHub release skipped: `gh` CLI not found.")
else:
print(f" ✗ GitHub release failed: {result.stderr.strip()}")
print(f" Release notes kept at: {changelog_file}")
print(f" Tag was created locally. Create the release manually:")
print(
f" gh release create {tag_name} --title 'Hermes Agent v{new_version} ({calver_date})' "
f"--notes-file .release_notes.md {' '.join(str(path) for path in artifacts)}"
)
print(f"\n ✓ Release artifacts prepared for manual publish: v{new_version} ({tag_name})")
print(f" ✗ GitHub release failed: {result.stderr}")
print(f" Tag was created. Create the release manually:")
print(f" gh release create {tag_name} --title 'Hermes Agent v{new_version} ({calver_date})'")
print(f"\n 🎉 Release v{new_version} ({tag_name}) published!")
else:
print(f"\n{'='*60}")
print(f" Dry run complete. To publish, add --publish")
@@ -744,149 +744,3 @@ class PixelBlendStack:
result = blend_canvas(result, canvas, mode, opacity)
return result
```
## Text Backdrop (Readability Mask)
When placing readable text over busy multi-grid ASCII backgrounds, the text will blend into the background and become illegible. **Always apply a dark backdrop behind text regions.**
The technique: compute the bounding box of all text glyphs, create a gaussian-blurred dark mask covering that area with padding, and multiply the background by `(1 - mask * darkness)` before rendering text on top.
```python
from scipy.ndimage import gaussian_filter
def apply_text_backdrop(canvas, glyphs, padding=80, darkness=0.75):
"""Darken the background behind text for readability.
Call AFTER rendering background, BEFORE rendering text.
Args:
canvas: (VH, VW, 3) uint8 background
glyphs: list of {"x": float, "y": float, ...} glyph positions
padding: pixel padding around text bounding box
darkness: 0.0 = no darkening, 1.0 = fully black
Returns:
darkened canvas (uint8)
"""
if not glyphs:
return canvas
xs = [g['x'] for g in glyphs]
ys = [g['y'] for g in glyphs]
x0 = max(0, int(min(xs)) - padding)
y0 = max(0, int(min(ys)) - padding)
x1 = min(VW, int(max(xs)) + padding + 50) # extra for char width
y1 = min(VH, int(max(ys)) + padding + 60) # extra for char height
# Soft dark mask with gaussian blur for feathered edges
mask = np.zeros((VH, VW), dtype=np.float32)
mask[y0:y1, x0:x1] = 1.0
mask = gaussian_filter(mask, sigma=padding * 0.6)
factor = 1.0 - mask * darkness
return (canvas.astype(np.float32) * factor[:, :, np.newaxis]).astype(np.uint8)
```
### Usage in render pipeline
Insert between background rendering and text rendering:
```python
# 1. Render background (multi-grid ASCII effects)
bg = render_background(cfg, t)
# 2. Darken behind text region
bg = apply_text_backdrop(bg, frame_glyphs, padding=80, darkness=0.75)
# 3. Render text on top (now readable against dark backdrop)
bg = text_renderer.render(bg, frame_glyphs, color=(255, 255, 255))
```
Combine with **reverse vignette** (see shaders.md) for scenes where text is always centered — the reverse vignette provides a persistent center-dark zone, while the backdrop handles per-frame glyph positions.
## External Layout Oracle Pattern
For text-heavy videos where text needs to dynamically reflow around obstacles (shapes, icons, other text), use an external layout engine to pre-compute glyph positions and feed them into the Python renderer via JSON.
### Architecture
```
Layout Engine (browser/Node.js) → layouts.json → Python ASCII Renderer
↑ ↑
Computes per-frame Reads glyph positions,
glyph (x,y) positions renders as ASCII chars
with obstacle-aware reflow with full effect pipeline
```
### JSON interchange format
```json
{
"meta": {
"canvas_width": 1080, "canvas_height": 1080,
"fps": 24, "total_frames": 1248,
"fonts": {
"body": {"charW": 12.04, "charH": 24, "fontSize": 20},
"hero": {"charW": 24.08, "charH": 48, "fontSize": 40}
}
},
"scenes": [
{
"id": "scene_name",
"start_frame": 0, "end_frame": 96,
"frames": {
"0": {
"glyphs": [
{"char": "H", "x": 287.1, "y": 400.0, "alpha": 1.0},
{"char": "e", "x": 311.2, "y": 400.0, "alpha": 1.0}
],
"obstacles": [
{"type": "circle", "cx": 540, "cy": 540, "r": 80},
{"type": "rect", "x": 300, "y": 500, "w": 120, "h": 80}
]
}
}
}
]
}
```
### When to use
- Text that dynamically reflows around moving objects
- Per-glyph animation (reveal, scatter, physics)
- Variable typography that needs precise measurement
- Any case where Python's Pillow text layout is insufficient
### When NOT to use
- Static centered text (just use PIL `draw.text()` directly)
- Text that only fades in/out without spatial animation
- Simple typewriter effects (handle in Python with a character counter)
### Running the oracle
Use Playwright to run the layout engine in a headless browser:
```javascript
// extract.mjs
import { chromium } from 'playwright';
const browser = await chromium.launch({ headless: true });
const page = await browser.newPage();
await page.goto(`file://${oraclePath}`);
await page.waitForFunction(() => window.__ORACLE_DONE__ === true, null, { timeout: 60000 });
const result = await page.evaluate(() => window.__ORACLE_RESULT__);
writeFileSync('layouts.json', JSON.stringify(result));
await browser.close();
```
### Consuming in Python
```python
# In the renderer, map pixel positions to the canvas:
for glyph in frame_data['glyphs']:
char, px, py = glyph['char'], glyph['x'], glyph['y']
alpha = glyph.get('alpha', 1.0)
# Render using PIL draw.text() at exact pixel position
draw.text((px, py), char, fill=(int(255*alpha),)*3, font=font)
```
Obstacles from the JSON can also be rendered as glowing ASCII shapes (circles, rectangles) to visualize the reflow zones.
@@ -834,39 +834,6 @@ def sh_vignette(c, s=0.22):
return np.clip(c * _vig_cache[k][:,:,None], 0, 255).astype(np.uint8)
```
#### Reverse Vignette
Inverted vignette: darkens the **center** and leaves edges bright. Useful when text is centered over busy backgrounds — creates a natural dark zone for readability without a hard-edged box.
Combine with `apply_text_backdrop()` (see composition.md) for per-frame glyph-aware darkening.
```python
_rvignette_cache = {}
def sh_reverse_vignette(c, strength=0.5):
"""Center darkening, edge brightening. Cached."""
k = ('rv', c.shape[0], c.shape[1], round(strength, 2))
if k not in _rvignette_cache:
h, w = c.shape[:2]
Y = np.linspace(-1, 1, h)[:, None]
X = np.linspace(-1, 1, w)[None, :]
d = np.sqrt(X**2 + Y**2)
# Invert: bright at edges, dark at center
mask = np.clip(1.0 - (1.0 - d * 0.7) * strength, 0.2, 1.0)
_rvignette_cache[k] = mask[:, :, np.newaxis].astype(np.float32)
return np.clip(c.astype(np.float32) * _rvignette_cache[k], 0, 255).astype(np.uint8)
```
| Param | Default | Effect |
|-------|---------|--------|
| `strength` | 0.5 | 0 = no effect, 1.0 = center nearly black |
Add to ShaderChain dispatch:
```python
elif name == "reverse_vignette":
return sh_reverse_vignette(canvas, kwargs.get("strength", 0.5))
```
#### Contrast
```python
def sh_contrast(c, factor=1.3):
@@ -14,8 +14,6 @@
| Random dark holes in output | Font missing Unicode glyphs | Validate palettes at init |
| Audio-visual desync | Frame timing accumulation | Use integer frame counter, compute t fresh each frame |
| Single-color flat output | Hue field shape mismatch | Ensure h,s,v arrays all (rows,cols) before hsv2rgb |
| Text unreadable over busy bg | No contrast between text and background | Use `apply_text_backdrop()` (composition.md) + `reverse_vignette` shader (shaders.md) |
| Text garbled/mirrored | Kaleidoscope or mirror shader applied to text scene | **Never apply kaleidoscope, mirror_h/v/quad/diag to scenes with readable text** — radial folding destroys legibility. Apply these only to background layers or text-free scenes |
Common bugs, gotchas, and platform-specific issues encountered during ASCII video development.

Some files were not shown because too many files have changed in this diff Show More