Compare commits

...

7 Commits

Author SHA1 Message Date
Sam Herring 545809d09b Tau2 bench changes 2026-04-06 15:30:54 -07:00
Sam Herring c32efc2885 Initial taubench implementation 2026-04-02 09:23:42 -07:00
Dean Kerr e905768ffd fix(gateway): remap HERMES_HOME to target user in system service unit
When `sudo hermes gateway install --system --run-as-user <user>` generates
the systemd unit, get_hermes_home() resolves to /root/.hermes because
Path.home() returns root's home under sudo. The unit correctly sets
HOME= and User= via _system_service_identity(), but HERMES_HOME was
computed independently and pointed to root's config directory.

Add _hermes_home_for_target_user() which remaps the current HERMES_HOME
to the equivalent path under the target user's home. This handles:
- Default ~/.hermes → target user's ~/.hermes
- Profiles (e.g. ~/.hermes/profiles/coder) → preserves relative structure
- Custom paths (e.g. /opt/hermes) → kept as-is

Supersedes #3861 which only handled the default case and left profiles
broken (also flagged by Copilot review).

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-04-01 06:09:33 -07:00
Teknium e0abf2416d fix: restore _config_version to 11 (reverted by stale-branch merge in #4419) (#4440)
PR #4419 was based on pre-credential-pools main where _config_version was 10.
The squash merge downgraded it from 11 (set by #2647) back to 10.
Also fixes the test assertion.
2026-04-01 04:34:04 -07:00
Teknium f6ada27d1c feat(skills): size limits for agent writes + fuzzy matching for patch (#4414)
* feat(skills): add content size limits for agent-created skills

Agent writes via skill_manage (create/edit/patch/write_file) are now
constrained to prevent unbounded growth:

- SKILL.md and supporting files: 100,000 character limit
- Supporting files: additional 1 MiB byte limit
- Patches on oversized hand-placed skills that reduce the size are
  allowed (shrink path), but patches that grow beyond the limit are
  rejected

Hand-placed skills and hub-installed skills have NO hard limit —
they load and function normally regardless of size. Hub installs
get a warning in the log if SKILL.md exceeds 100k chars.

This mirrors the memory system's char_limit pattern. Without this,
the agent auto-grows skills indefinitely through iterative patches
(hermes-agent-dev reached 197k chars / 72k tokens — 40x larger than
the largest skill in the entire skills.sh ecosystem).

Constants: MAX_SKILL_CONTENT_CHARS (100k), MAX_SKILL_FILE_BYTES (1MiB)
Tests: 14 new tests covering all write paths and edge cases

* feat(skills): add fuzzy matching to skill patch

_patch_skill now uses the same 8-strategy fuzzy matching engine
(tools/fuzzy_match.py) as the file patch tool. Handles whitespace
normalization, indentation differences, escape sequences, and
block-anchor matching. Eliminates exact-match failures when agents
patch skills with minor formatting mismatches.
2026-04-01 04:19:19 -07:00
Teknium 70744add15 feat(browser): add persistent Camofox sessions and VNC URL discovery (salvage #4400) (#4419)
Adds two Camofox features:

1. Persistent browser sessions: new `browser.camofox.managed_persistence`
   config option. When enabled, Hermes sends a deterministic profile-scoped
   userId to Camofox so the server maps it to a persistent browser profile
   directory. Cookies, logins, and browser state survive across restarts.
   Default remains ephemeral (random userId per session).

2. VNC URL discovery: Camofox /health endpoint returns vncPort when running
   in headed mode. Hermes constructs the VNC URL and includes it in navigate
   responses so the agent can share it with users.

Also fixes camofox_vision bug where call_llm response object was passed
directly to json.dumps instead of extracting .choices[0].message.content.

Changes from original PR:
- Removed browser_evaluate tool (separate feature, needs own PR)
- Removed snapshot truncation limit change (unrelated)
- Config.yaml only for managed_persistence (no env var, no version bump)
- Rewrote tests to use config mock instead of env var
- Reverted package-lock.json churn

Co-authored-by: analista <psikonetik@gmail.com.com>
2026-04-01 04:18:50 -07:00
Teknium 85e96a4638 fix(skills): move unified hermes-agent skill into autonomous-ai-agents category (#4435)
The unified skill from PR #4332 was placed at a top-level
skills/hermes-agent/ directory, creating a redundant standalone
category. Move it to skills/autonomous-ai-agents/hermes-agent/
alongside claude-code, codex, and opencode where it belongs.
2026-04-01 03:39:25 -07:00
20 changed files with 1686 additions and 29 deletions
@@ -0,0 +1,324 @@
"""
HermesAgent for tau2-bench evaluation.
Implements the tau2 HalfDuplexAgent interface using litellm with OpenRouter,
matching the inference path used across the rest of the Hermes Agent codebase.
Usage:
python environments/benchmarks/taubench/run_eval.py \\
--model anthropic/claude-sonnet-4-5 \\
--base-url openrouter \\
--env retail
"""
import json
import os
import sys
from pathlib import Path
from typing import Optional
import litellm
from pydantic import BaseModel
_repo_root = Path(__file__).resolve().parent.parent.parent.parent
if str(_repo_root) not in sys.path:
sys.path.insert(0, str(_repo_root))
from environments.tool_call_parsers import get_parser
from tau2.agent.base_agent import HalfDuplexAgent, ValidAgentInputMessage
from tau2.data_model.message import (
AssistantMessage,
Message,
MultiToolMessage,
SystemMessage,
ToolCall,
ToolMessage,
UserMessage,
)
from tau2.environment.tool import Tool
class HermesAgentState(BaseModel):
system_messages: list[SystemMessage]
messages: list
class HermesAgent(HalfDuplexAgent[HermesAgentState]):
"""
tau2 HalfDuplexAgent backed by litellm, using OpenRouter (or any
OpenAI-compatible endpoint).
Registered as "hermes_agent" in the tau2 registry by run_eval.py.
"""
SYSTEM_PROMPT = (
"You are a customer service agent that helps the user according to the "
"<policy> provided below.\n"
"In each turn you can either:\n"
"- Send a message to the user.\n"
"- Make a tool call.\n"
"You cannot do both at the same time.\n\n"
"Try to be helpful and always follow the policy. "
"Always make sure you generate valid JSON only.\n\n"
"<policy>\n{domain_policy}\n</policy>"
)
# System prompt variant for qwen3_coder tool format — tools are embedded
# directly in the system prompt as <tools> XML instead of passed via the
# OpenAI tools= parameter.
SYSTEM_PROMPT_QWEN3_CODER = (
"You are a customer service agent that helps the user according to the "
"<policy> provided below.\n"
"In each turn you can either:\n"
"- Send a message to the user.\n"
"- Make a tool call.\n"
"You cannot do both at the same time.\n\n"
"Try to be helpful and always follow the policy. "
"Always make sure you generate valid JSON only.\n\n"
"You may call one or more functions to assist with the user query.\n\n"
"You are provided with function signatures within <tools></tools> XML tags:\n"
"<tools>\n{tools_json}\n</tools>\n\n"
"<policy>\n{domain_policy}\n</policy>"
)
def __init__(
self,
tools: list[Tool],
domain_policy: str,
model: str,
base_url: Optional[str] = None,
api_key: Optional[str] = None,
temperature: float = 0.0,
max_tokens: Optional[int] = None,
top_p: Optional[float] = None,
thinking: bool = False,
tool_parser: Optional[str] = None,
):
super().__init__(tools=tools, domain_policy=domain_policy)
self.model = model
self.base_url = base_url
self.api_key = api_key
self.temperature = temperature
self.max_tokens = max_tokens
self.top_p = top_p
self.thinking = thinking
self.tool_parser = tool_parser
self._parser = get_parser(tool_parser) if tool_parser else None
# OpenRouter requires specific headers; pass them via litellm extra_headers
self._extra_headers: dict = {}
if base_url and "openrouter" in base_url.lower():
self._extra_headers = {
"HTTP-Referer": "https://hermes-agent.nousresearch.com",
"X-Title": "Hermes Agent",
}
@property
def system_prompt(self) -> str:
if self.tool_parser == "qwen3_coder" and self.tools:
tools_json = json.dumps(
[t.openai_schema for t in self.tools], indent=2, ensure_ascii=False
)
return self.SYSTEM_PROMPT_QWEN3_CODER.format(
tools_json=tools_json,
domain_policy=self.domain_policy,
)
return self.SYSTEM_PROMPT.format(domain_policy=self.domain_policy)
def get_init_state(
self, message_history: Optional[list[Message]] = None
) -> HermesAgentState:
return HermesAgentState(
system_messages=[SystemMessage(role="system", content=self.system_prompt)],
messages=list(message_history or []),
)
def generate_next_message(
self, message: ValidAgentInputMessage, state: HermesAgentState
) -> tuple[AssistantMessage, HermesAgentState]:
# Append incoming message(s) to history
if isinstance(message, MultiToolMessage):
state.messages.extend(message.tool_messages)
else:
state.messages.append(message)
# Build litellm-compatible message list
all_messages = state.system_messages + state.messages
lm_messages = [_to_litellm_message(m) for m in all_messages]
kwargs = dict(
model=self.model,
messages=lm_messages,
temperature=self.temperature,
)
if self.tools:
kwargs["tools"] = [t.openai_schema for t in self.tools]
if self.max_tokens is not None:
kwargs["max_tokens"] = self.max_tokens
if self.top_p is not None:
kwargs["top_p"] = self.top_p
# Enable thinking/reasoning mode. OpenRouter exposes this as
# `include_reasoning` for nemotron (per supported_parameters in the
# model metadata). Pass via extra_body to bypass litellm filtering.
if self.thinking:
kwargs["extra_body"] = {"include_reasoning": True}
# Only pass base_url when model doesn't already have a provider prefix
# (litellm uses either the prefix OR base_url, not both)
if self.base_url and not self.model.startswith("openrouter/"):
kwargs["base_url"] = self.base_url
if self.api_key:
kwargs["api_key"] = self.api_key
if self._extra_headers:
kwargs["extra_headers"] = self._extra_headers
response = litellm.completion(**kwargs)
assistant_msg = _litellm_response_to_assistant_message(response, parser=self._parser)
state.messages.append(assistant_msg)
return assistant_msg, state
# ---------------------------------------------------------------------------
# Conversion helpers
# ---------------------------------------------------------------------------
def _to_litellm_message(msg) -> dict:
"""Convert a tau2 message object to a litellm-compatible dict."""
if isinstance(msg, SystemMessage):
return {"role": "system", "content": msg.content or ""}
if isinstance(msg, UserMessage):
if msg.tool_calls:
# User tool calls (tau2 v2 feature — user has tools too)
return {
"role": "user",
"content": msg.content or "",
"tool_calls": [_tool_call_to_dict(tc) for tc in msg.tool_calls],
}
return {"role": "user", "content": msg.content or ""}
if isinstance(msg, AssistantMessage):
d: dict = {"role": "assistant", "content": msg.content or ""}
if msg.tool_calls:
d["tool_calls"] = [_tool_call_to_dict(tc) for tc in msg.tool_calls]
return d
if isinstance(msg, ToolMessage):
return {
"role": "tool",
"tool_call_id": msg.id,
"content": msg.content or "",
}
# Fallback
return {"role": getattr(msg, "role", "user"), "content": str(getattr(msg, "content", ""))}
def _tool_call_to_dict(tc: ToolCall) -> dict:
import json
return {
"id": tc.id or "call_0",
"type": "function",
"function": {
"name": tc.name,
"arguments": json.dumps(tc.arguments),
},
}
def _litellm_response_to_assistant_message(response, parser=None) -> AssistantMessage:
"""Convert a litellm ModelResponse to a tau2 AssistantMessage."""
import json
choice = response.choices[0]
msg = choice.message
content = msg.content or ""
tool_calls_raw = getattr(msg, "tool_calls", None)
tau2_tool_calls: Optional[list[ToolCall]] = None
if parser and content:
# Use the custom tool parser (e.g. qwen3_coder) to extract tool calls
# from the raw text response.
parsed_content, parsed_tool_calls = parser.parse(content)
if parsed_tool_calls:
content = parsed_content or ""
tau2_tool_calls = []
for tc in parsed_tool_calls:
try:
arguments = json.loads(tc.function.arguments or "{}")
except json.JSONDecodeError:
arguments = {}
tau2_tool_calls.append(
ToolCall(
id=tc.id or "call_0",
name=tc.function.name,
arguments=arguments,
requestor="assistant",
)
)
elif tool_calls_raw:
tau2_tool_calls = []
for tc in tool_calls_raw:
if hasattr(tc, "function"):
name = tc.function.name
try:
arguments = json.loads(tc.function.arguments or "{}")
except json.JSONDecodeError:
arguments = {}
tau2_tool_calls.append(
ToolCall(
id=tc.id or "call_0",
name=name,
arguments=arguments,
requestor="assistant",
)
)
cost = None
try:
cost = litellm.completion_cost(response)
except Exception:
pass
usage = None
if hasattr(response, "usage") and response.usage:
usage = dict(response.usage)
return AssistantMessage(
role="assistant",
content=content if not tau2_tool_calls else None,
tool_calls=tau2_tool_calls,
cost=cost,
usage=usage,
)
def create_hermes_agent(tools: list[Tool], domain_policy: str, **kwargs) -> HermesAgent:
"""
Factory function registered with the tau2 registry.
Expected kwargs:
model (str): litellm model string
base_url (str): API base URL (optional)
api_key (str): API key (optional)
temperature (float): sampling temperature (default 0.0)
top_p (float): nucleus sampling (optional)
max_tokens (int): max tokens (optional)
thinking (bool): enable reasoning/thinking mode (default False)
"""
return HermesAgent(
tools=tools,
domain_policy=domain_policy,
model=kwargs["model"],
base_url=kwargs.get("base_url"),
api_key=kwargs.get("api_key"),
temperature=kwargs.get("temperature", 0.0),
top_p=kwargs.get("top_p"),
max_tokens=kwargs.get("max_tokens"),
thinking=kwargs.get("thinking", False),
tool_parser=kwargs.get("tool_parser"),
)
@@ -0,0 +1,288 @@
"""
tau2-bench evaluation runner for Hermes Agent.
Runs the tau2-bench retail, airline, telecom, or banking_knowledge evaluation
using HermesAgent backed by litellm — the same inference path used across the
rest of the Hermes Agent codebase.
Usage:
# Against OpenRouter (auto-detects OPENROUTER_API_KEY)
python environments/benchmarks/taubench/run_eval.py \\
--model openrouter/anthropic/claude-sonnet-4-5 \\
--base-url openrouter \\
--env retail
# Against OpenAI directly
python environments/benchmarks/taubench/run_eval.py \\
--model gpt-4o \\
--env retail
# Local vLLM
python environments/benchmarks/taubench/run_eval.py \\
--model openai/NousResearch/Hermes-3-Llama-3.1-70B \\
--base-url http://localhost:8000/v1 \\
--env retail \\
--num-trials 3
# Specific tasks only
python environments/benchmarks/taubench/run_eval.py \\
--model openrouter/anthropic/claude-sonnet-4-5 \\
--base-url openrouter \\
--env retail \\
--task-ids task_1 task_2 task_5
Results are saved to results/tau2bench/ as JSON.
Dependencies (requires Python 3.12+):
pip install "tau2 @ git+https://github.com/sierra-research/tau2-bench.git"
# or: pip install -e ".[tau2bench]"
"""
import argparse
import logging
import os
import sys
from pathlib import Path
from typing import Optional
_repo_root = Path(__file__).resolve().parent.parent.parent.parent
if str(_repo_root) not in sys.path:
sys.path.insert(0, str(_repo_root))
from tau2.data_model.simulation import Results, TextRunConfig
from tau2.evaluator.evaluator import EvaluationType
from tau2.registry import registry
from tau2.runner.batch import run_tasks
from tau2.runner.helpers import get_tasks
from environments.benchmarks.taubench.hermes_agent import create_hermes_agent
logging.basicConfig(
level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s"
)
logger = logging.getLogger(__name__)
OPENROUTER_BASE_URL = "https://openrouter.ai/api/v1"
AGENT_NAME = "hermes_agent"
def _register_agent(
model: str,
base_url: Optional[str],
api_key: Optional[str],
temperature: float,
top_p: Optional[float],
max_tokens: Optional[int],
thinking: bool,
tool_parser: Optional[str],
) -> None:
"""Register the HermesAgent factory with the tau2 registry (idempotent)."""
if registry.get_agent_factory(AGENT_NAME) is not None:
return
def factory(tools, domain_policy, **kwargs):
return create_hermes_agent(
tools=tools,
domain_policy=domain_policy,
model=model,
base_url=base_url,
api_key=api_key,
temperature=temperature,
top_p=top_p,
max_tokens=max_tokens,
thinking=thinking,
tool_parser=tool_parser,
)
registry.register_agent_factory(factory=factory, name=AGENT_NAME)
logger.info("Registered agent factory: %s (model=%s, thinking=%s, tool_parser=%s)", AGENT_NAME, model, thinking, tool_parser)
def run_eval(
model: str,
base_url: Optional[str],
api_key: Optional[str],
user_model: str,
env_name: str,
task_split: Optional[str],
num_trials: int,
max_concurrency: int,
max_steps: int,
temperature: float,
top_p: Optional[float],
max_tokens: Optional[int],
thinking: bool,
tool_parser: Optional[str],
task_ids: Optional[list],
start_index: int,
end_index: int,
log_dir: str,
seed: int,
) -> Results:
# Resolve OpenRouter shorthand
if base_url and base_url.strip().lower() == "openrouter":
base_url = OPENROUTER_BASE_URL
is_openrouter = base_url and "openrouter" in base_url.lower()
# litellm requires the "openrouter/" prefix to route correctly
if is_openrouter and not model.startswith("openrouter/"):
model = f"openrouter/{model}"
if is_openrouter and not user_model.startswith("openrouter/"):
user_model = f"openrouter/{user_model}"
# Resolve API key
if is_openrouter:
api_key = api_key or os.environ.get("OPENROUTER_API_KEY") or os.environ.get("OPENAI_API_KEY")
# litellm reads OPENAI_API_KEY for base_url overrides; set it so the
# user simulator's generate() call also authenticates correctly.
if api_key and not os.environ.get("OPENAI_API_KEY"):
os.environ["OPENAI_API_KEY"] = api_key
else:
api_key = api_key or os.environ.get("OPENAI_API_KEY")
_register_agent(
model=model,
base_url=base_url,
api_key=api_key,
temperature=temperature,
top_p=top_p,
max_tokens=max_tokens,
thinking=thinking,
tool_parser=tool_parser,
)
# Load tasks — task_ids in tau2 are strings like "task_1"
tasks = get_tasks(
task_set_name=env_name,
task_split_name=task_split,
task_ids=[str(i) for i in task_ids] if task_ids else None,
)
if not task_ids and (end_index != -1 or start_index != 0):
end = end_index if end_index != -1 else len(tasks)
tasks = tasks[start_index:end]
logger.info(
"Running tau2-%s eval: %d tasks, %d trial(s), concurrency=%d",
env_name, len(tasks), num_trials, max_concurrency,
)
save_path = Path(log_dir) / f"tau2-{env_name}-{model.split('/')[-1]}.json"
save_path.parent.mkdir(parents=True, exist_ok=True)
# Pass api_key/base_url to user sim via llm_args so tau2's generate() authenticates.
# When using OpenRouter for the user sim, mirror the agent's key + endpoint.
user_llm_args: dict = {}
if is_openrouter and api_key:
user_llm_args["api_key"] = api_key
user_llm_args["base_url"] = base_url
config = TextRunConfig(
domain=env_name,
agent=AGENT_NAME,
user="user_simulator",
llm_agent=model,
llm_args_agent={},
llm_user=user_model,
llm_args_user=user_llm_args,
num_trials=num_trials,
max_steps=max_steps,
max_concurrency=max_concurrency,
seed=seed,
)
results = run_tasks(
config,
tasks,
save_path=save_path,
console_display=True,
# ALL: respects each task's reward_basis. NL assertions are skipped
# gracefully (scored as pass) rather than raising an error, so tasks
# are evaluated only on their actual basis components (DB, ACTION, etc.)
evaluation_type=EvaluationType.ALL,
)
logger.info("Results saved to %s", save_path)
return results
def main():
parser = argparse.ArgumentParser(
description="Run tau2-bench evaluation with Hermes Agent (requires Python 3.12+)",
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
)
parser.add_argument(
"--model", required=True,
help="litellm model string, e.g. 'openrouter/anthropic/claude-sonnet-4-5' or 'gpt-4o'",
)
parser.add_argument(
"--base-url", default=None,
help="API base URL. Use 'openrouter' as shorthand for https://openrouter.ai/api/v1.",
)
parser.add_argument("--api-key", default=None, help="API key (falls back to OPENROUTER_API_KEY / OPENAI_API_KEY)")
parser.add_argument("--temperature", type=float, default=1.0,
help="Sampling temperature. NVIDIA used 1.0 for nemotron-super.")
parser.add_argument("--top-p", type=float, default=0.95,
help="Nucleus sampling. NVIDIA used 0.95 for nemotron-super.")
parser.add_argument("--max-tokens", type=int, default=None)
parser.add_argument("--thinking", action="store_true", default=False,
help="Enable reasoning/thinking mode (use_reasoning=true). "
"Required to match NVIDIA's reported nemotron-super scores.")
parser.add_argument("--tool-parser", default=None,
help="Tool call parser to use (e.g. 'qwen3_coder'). When set, tools are "
"embedded in the system prompt as <tools> XML and responses are parsed "
"from raw text instead of using OpenAI function calling format.")
parser.add_argument(
"--user-model", default="qwen/qwen3-235b-a22b-2507:nitro",
help="litellm model string for the tau2 user simulator. "
"Defaults to qwen/qwen3-235b-a22b-2507:nitro (instruct, non-thinking) to match NVIDIA's eval setup. "
"When using --base-url openrouter the openrouter/ prefix is added automatically.",
)
parser.add_argument(
"--env", default="retail",
choices=["retail", "airline", "telecom", "banking_knowledge", "mock"],
)
parser.add_argument(
"--task-split", default=None,
help="Task split name (e.g. 'base'). Defaults to the domain default.",
)
parser.add_argument("--num-trials", type=int, default=1)
parser.add_argument("--max-concurrency", type=int, default=8)
parser.add_argument("--max-steps", type=int, default=50)
parser.add_argument(
"--task-ids", nargs="*", default=None,
help="Specific task IDs to run (tau2 task IDs are strings like 'task_1')",
)
parser.add_argument("--start-index", type=int, default=0)
parser.add_argument("--end-index", type=int, default=-1)
parser.add_argument("--seed", type=int, default=10)
parser.add_argument("--log-dir", default="results/tau2bench")
args = parser.parse_args()
run_eval(
model=args.model,
base_url=args.base_url,
api_key=args.api_key,
user_model=args.user_model,
env_name=args.env,
task_split=args.task_split,
num_trials=args.num_trials,
max_concurrency=args.max_concurrency,
max_steps=args.max_steps,
temperature=args.temperature,
top_p=args.top_p,
max_tokens=args.max_tokens,
thinking=args.thinking,
tool_parser=args.tool_parser,
task_ids=args.task_ids,
start_index=args.start_index,
end_index=args.end_index,
log_dir=args.log_dir,
seed=args.seed,
)
if __name__ == "__main__":
main()
+7
View File
@@ -247,6 +247,13 @@ DEFAULT_CONFIG = {
"command_timeout": 30, # Timeout for browser commands in seconds (screenshot, navigate, etc.)
"record_sessions": False, # Auto-record browser sessions as WebM videos
"allow_private_urls": False, # Allow navigating to private/internal IPs (localhost, 192.168.x.x, etc.)
"camofox": {
# When true, Hermes sends a stable profile-scoped userId to Camofox
# so the server can map it to a persistent browser profile directory.
# Requires Camofox server to be configured with CAMOFOX_PROFILE_DIR.
# When false (default), each session gets a random userId (ephemeral).
"managed_persistence": False,
},
},
# Filesystem checkpoints — automatic snapshots before destructive file ops.
+28 -2
View File
@@ -463,6 +463,32 @@ def _build_user_local_paths(home: Path, path_entries: list[str]) -> list[str]:
return [p for p in candidates if p not in path_entries and Path(p).exists()]
def _hermes_home_for_target_user(target_home_dir: str) -> str:
"""Remap the current HERMES_HOME to the equivalent under a target user's home.
When installing a system service via sudo, get_hermes_home() resolves to
root's home. This translates it to the target user's equivalent path:
/root/.hermes /home/alice/.hermes
/root/.hermes/profiles/coder /home/alice/.hermes/profiles/coder
/opt/custom-hermes /opt/custom-hermes (kept as-is)
"""
current_hermes = get_hermes_home().resolve()
current_default = (Path.home() / ".hermes").resolve()
target_default = Path(target_home_dir) / ".hermes"
# Default ~/.hermes → remap to target user's default
if current_hermes == current_default:
return str(target_default)
# Profile or subdir of ~/.hermes → preserve the relative structure
try:
relative = current_hermes.relative_to(current_default)
return str(target_default / relative)
except ValueError:
# Completely custom path (not under ~/.hermes) — keep as-is
return str(current_hermes)
def generate_systemd_unit(system: bool = False, run_as_user: str | None = None) -> str:
python_path = get_python_path()
working_dir = str(PROJECT_ROOT)
@@ -478,12 +504,11 @@ def generate_systemd_unit(system: bool = False, run_as_user: str | None = None)
if resolved_node_dir not in path_entries:
path_entries.append(resolved_node_dir)
hermes_home = str(get_hermes_home().resolve())
common_bin_paths = ["/usr/local/sbin", "/usr/local/bin", "/usr/sbin", "/usr/bin", "/sbin", "/bin"]
if system:
username, group_name, home_dir = _system_service_identity(run_as_user)
hermes_home = _hermes_home_for_target_user(home_dir)
path_entries.extend(_build_user_local_paths(Path(home_dir), path_entries))
path_entries.extend(common_bin_paths)
sane_path = ":".join(path_entries)
@@ -518,6 +543,7 @@ StandardError=journal
WantedBy=multi-user.target
"""
hermes_home = str(get_hermes_home().resolve())
path_entries.extend(_build_user_local_paths(Path.home(), path_entries))
path_entries.extend(common_bin_paths)
sane_path = ":".join(path_entries)
+2
View File
@@ -72,6 +72,8 @@ rl = [
"wandb>=0.15.0,<1",
]
yc-bench = ["yc-bench @ git+https://github.com/collinear-ai/yc-bench.git ; python_version >= '3.12'"]
taubench = ["tau-bench @ git+https://github.com/sierra-research/tau-bench.git"]
tau2bench = ["tau2 @ git+https://github.com/sierra-research/tau2-bench.git"]
all = [
"hermes-agent[modal]",
"hermes-agent[daytona]",
+96
View File
@@ -339,6 +339,102 @@ class TestDetectVenvDir:
assert result is None
class TestSystemUnitHermesHome:
"""HERMES_HOME in system units must reference the target user, not root."""
def test_system_unit_uses_target_user_home_not_calling_user(self, monkeypatch):
# Simulate sudo: Path.home() returns /root, target user is alice
monkeypatch.setattr(Path, "home", staticmethod(lambda: Path("/root")))
monkeypatch.delenv("HERMES_HOME", raising=False)
monkeypatch.setattr(
gateway_cli, "_system_service_identity",
lambda run_as_user=None: ("alice", "alice", "/home/alice"),
)
monkeypatch.setattr(
gateway_cli, "_build_user_local_paths",
lambda home, existing: [],
)
unit = gateway_cli.generate_systemd_unit(system=True, run_as_user="alice")
assert 'HERMES_HOME=/home/alice/.hermes' in unit
assert '/root/.hermes' not in unit
def test_system_unit_remaps_profile_to_target_user(self, monkeypatch):
# Simulate sudo with a profile: HERMES_HOME was resolved under root
monkeypatch.setattr(Path, "home", staticmethod(lambda: Path("/root")))
monkeypatch.setenv("HERMES_HOME", "/root/.hermes/profiles/coder")
monkeypatch.setattr(
gateway_cli, "_system_service_identity",
lambda run_as_user=None: ("alice", "alice", "/home/alice"),
)
monkeypatch.setattr(
gateway_cli, "_build_user_local_paths",
lambda home, existing: [],
)
unit = gateway_cli.generate_systemd_unit(system=True, run_as_user="alice")
assert 'HERMES_HOME=/home/alice/.hermes/profiles/coder' in unit
assert '/root/' not in unit
def test_system_unit_preserves_custom_hermes_home(self, monkeypatch):
# Custom HERMES_HOME not under any user's home — keep as-is
monkeypatch.setattr(Path, "home", staticmethod(lambda: Path("/root")))
monkeypatch.setenv("HERMES_HOME", "/opt/hermes-shared")
monkeypatch.setattr(
gateway_cli, "_system_service_identity",
lambda run_as_user=None: ("alice", "alice", "/home/alice"),
)
monkeypatch.setattr(
gateway_cli, "_build_user_local_paths",
lambda home, existing: [],
)
unit = gateway_cli.generate_systemd_unit(system=True, run_as_user="alice")
assert 'HERMES_HOME=/opt/hermes-shared' in unit
def test_user_unit_unaffected_by_change(self):
# User-scope units should still use the calling user's HERMES_HOME
unit = gateway_cli.generate_systemd_unit(system=False)
hermes_home = str(gateway_cli.get_hermes_home().resolve())
assert f'HERMES_HOME={hermes_home}' in unit
class TestHermesHomeForTargetUser:
"""Unit tests for _hermes_home_for_target_user()."""
def test_remaps_default_home(self, monkeypatch):
monkeypatch.setattr(Path, "home", staticmethod(lambda: Path("/root")))
monkeypatch.delenv("HERMES_HOME", raising=False)
result = gateway_cli._hermes_home_for_target_user("/home/alice")
assert result == "/home/alice/.hermes"
def test_remaps_profile_path(self, monkeypatch):
monkeypatch.setattr(Path, "home", staticmethod(lambda: Path("/root")))
monkeypatch.setenv("HERMES_HOME", "/root/.hermes/profiles/coder")
result = gateway_cli._hermes_home_for_target_user("/home/alice")
assert result == "/home/alice/.hermes/profiles/coder"
def test_keeps_custom_path(self, monkeypatch):
monkeypatch.setattr(Path, "home", staticmethod(lambda: Path("/root")))
monkeypatch.setenv("HERMES_HOME", "/opt/hermes")
result = gateway_cli._hermes_home_for_target_user("/home/alice")
assert result == "/opt/hermes"
def test_noop_when_same_user(self, monkeypatch):
monkeypatch.setattr(Path, "home", staticmethod(lambda: Path("/home/alice")))
monkeypatch.delenv("HERMES_HOME", raising=False)
result = gateway_cli._hermes_home_for_target_user("/home/alice")
assert result == "/home/alice/.hermes"
class TestGeneratedUnitUsesDetectedVenv:
def test_systemd_unit_uses_dot_venv_when_detected(self, tmp_path, monkeypatch):
dot_venv = tmp_path / ".venv"
@@ -0,0 +1,242 @@
"""Persistence tests for the Camofox browser backend.
Tests that managed persistence uses stable identity while default mode
uses random identity. The actual browser profile persistence is handled
by the Camofox server (when CAMOFOX_PROFILE_DIR is set).
"""
import json
from unittest.mock import MagicMock, patch
import pytest
from tools.browser_camofox import (
_drop_session,
_get_session,
_managed_persistence_enabled,
camofox_close,
camofox_navigate,
check_camofox_available,
cleanup_all_camofox_sessions,
get_vnc_url,
)
from tools.browser_camofox_state import get_camofox_identity
def _mock_response(status=200, json_data=None):
resp = MagicMock()
resp.status_code = status
resp.json.return_value = json_data or {}
resp.raise_for_status = MagicMock()
return resp
def _enable_persistence():
"""Return a patch context that enables managed persistence via config."""
config = {"browser": {"camofox": {"managed_persistence": True}}}
return patch("tools.browser_camofox.load_config", return_value=config)
@pytest.fixture(autouse=True)
def _clear_session_state():
import tools.browser_camofox as mod
yield
with mod._sessions_lock:
mod._sessions.clear()
mod._vnc_url = None
mod._vnc_url_checked = False
class TestManagedPersistenceToggle:
def test_disabled_by_default(self):
config = {"browser": {"camofox": {"managed_persistence": False}}}
with patch("tools.browser_camofox.load_config", return_value=config):
assert _managed_persistence_enabled() is False
def test_enabled_via_config_yaml(self):
config = {"browser": {"camofox": {"managed_persistence": True}}}
with patch("tools.browser_camofox.load_config", return_value=config):
assert _managed_persistence_enabled() is True
def test_disabled_when_key_missing(self):
config = {"browser": {}}
with patch("tools.browser_camofox.load_config", return_value=config):
assert _managed_persistence_enabled() is False
def test_disabled_on_config_load_error(self):
with patch("tools.browser_camofox.load_config", side_effect=Exception("fail")):
assert _managed_persistence_enabled() is False
class TestEphemeralMode:
"""Default behavior: random userId, no persistence."""
def test_session_gets_random_user_id(self, tmp_path, monkeypatch):
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
monkeypatch.setenv("CAMOFOX_URL", "http://localhost:9377")
session = _get_session("task-1")
assert session["user_id"].startswith("hermes_")
assert session["managed"] is False
def test_different_tasks_get_different_user_ids(self, tmp_path, monkeypatch):
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
monkeypatch.setenv("CAMOFOX_URL", "http://localhost:9377")
s1 = _get_session("task-1")
s2 = _get_session("task-2")
assert s1["user_id"] != s2["user_id"]
def test_session_reuse_within_same_task(self, tmp_path, monkeypatch):
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
monkeypatch.setenv("CAMOFOX_URL", "http://localhost:9377")
s1 = _get_session("task-1")
s2 = _get_session("task-1")
assert s1 is s2
class TestManagedPersistenceMode:
"""With managed_persistence: stable userId derived from Hermes profile."""
def test_session_gets_stable_user_id(self, tmp_path, monkeypatch):
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
monkeypatch.setenv("CAMOFOX_URL", "http://localhost:9377")
with _enable_persistence():
session = _get_session("task-1")
expected = get_camofox_identity("task-1")
assert session["user_id"] == expected["user_id"]
assert session["session_key"] == expected["session_key"]
assert session["managed"] is True
def test_same_user_id_after_session_drop(self, tmp_path, monkeypatch):
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
monkeypatch.setenv("CAMOFOX_URL", "http://localhost:9377")
with _enable_persistence():
s1 = _get_session("task-1")
uid1 = s1["user_id"]
_drop_session("task-1")
s2 = _get_session("task-1")
assert s2["user_id"] == uid1
def test_same_user_id_across_tasks(self, tmp_path, monkeypatch):
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
monkeypatch.setenv("CAMOFOX_URL", "http://localhost:9377")
with _enable_persistence():
s1 = _get_session("task-a")
s2 = _get_session("task-b")
# Same profile = same userId, different session keys
assert s1["user_id"] == s2["user_id"]
assert s1["session_key"] != s2["session_key"]
def test_different_profiles_get_different_user_ids(self, tmp_path, monkeypatch):
monkeypatch.setenv("CAMOFOX_URL", "http://localhost:9377")
with _enable_persistence():
monkeypatch.setenv("HERMES_HOME", str(tmp_path / "profile-a"))
s1 = _get_session("task-1")
uid_a = s1["user_id"]
_drop_session("task-1")
monkeypatch.setenv("HERMES_HOME", str(tmp_path / "profile-b"))
s2 = _get_session("task-1")
assert s2["user_id"] != uid_a
def test_navigate_uses_stable_identity(self, tmp_path, monkeypatch):
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
monkeypatch.setenv("CAMOFOX_URL", "http://localhost:9377")
requests_seen = []
def _capture_post(url, json=None, timeout=None):
requests_seen.append(json)
return _mock_response(
json_data={"tabId": "tab-1", "url": "https://example.com"}
)
with _enable_persistence(), \
patch("tools.browser_camofox.requests.post", side_effect=_capture_post):
result = json.loads(camofox_navigate("https://example.com", task_id="task-1"))
assert result["success"] is True
expected = get_camofox_identity("task-1")
assert requests_seen[0]["userId"] == expected["user_id"]
def test_navigate_reuses_identity_after_close(self, tmp_path, monkeypatch):
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
monkeypatch.setenv("CAMOFOX_URL", "http://localhost:9377")
requests_seen = []
def _capture_post(url, json=None, timeout=None):
requests_seen.append(json)
return _mock_response(
json_data={"tabId": f"tab-{len(requests_seen)}", "url": "https://example.com"}
)
with (
_enable_persistence(),
patch("tools.browser_camofox.requests.post", side_effect=_capture_post),
patch("tools.browser_camofox.requests.delete", return_value=_mock_response()),
):
first = json.loads(camofox_navigate("https://example.com", task_id="task-1"))
camofox_close("task-1")
second = json.loads(camofox_navigate("https://example.com", task_id="task-1"))
assert first["success"] is True
assert second["success"] is True
tab_requests = [req for req in requests_seen if "userId" in req]
assert len(tab_requests) == 2
assert tab_requests[0]["userId"] == tab_requests[1]["userId"]
class TestVncUrlDiscovery:
"""VNC URL is derived from the Camofox health endpoint."""
def test_vnc_url_from_health_port(self, monkeypatch):
monkeypatch.setenv("CAMOFOX_URL", "http://myhost:9377")
health_resp = _mock_response(json_data={"ok": True, "vncPort": 6080})
with patch("tools.browser_camofox.requests.get", return_value=health_resp):
assert check_camofox_available() is True
assert get_vnc_url() == "http://myhost:6080"
def test_vnc_url_none_when_headless(self, monkeypatch):
monkeypatch.setenv("CAMOFOX_URL", "http://localhost:9377")
health_resp = _mock_response(json_data={"ok": True})
with patch("tools.browser_camofox.requests.get", return_value=health_resp):
check_camofox_available()
assert get_vnc_url() is None
def test_vnc_url_rejects_invalid_port(self, monkeypatch):
monkeypatch.setenv("CAMOFOX_URL", "http://localhost:9377")
health_resp = _mock_response(json_data={"ok": True, "vncPort": "bad"})
with patch("tools.browser_camofox.requests.get", return_value=health_resp):
check_camofox_available()
assert get_vnc_url() is None
def test_vnc_url_only_probed_once(self, monkeypatch):
monkeypatch.setenv("CAMOFOX_URL", "http://localhost:9377")
health_resp = _mock_response(json_data={"ok": True, "vncPort": 6080})
with patch("tools.browser_camofox.requests.get", return_value=health_resp) as mock_get:
check_camofox_available()
check_camofox_available()
# Second call still hits /health for availability but doesn't re-parse vncPort
assert get_vnc_url() == "http://localhost:6080"
def test_navigate_includes_vnc_hint(self, tmp_path, monkeypatch):
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
monkeypatch.setenv("CAMOFOX_URL", "http://localhost:9377")
import tools.browser_camofox as mod
mod._vnc_url = "http://localhost:6080"
mod._vnc_url_checked = True
with patch("tools.browser_camofox.requests.post", return_value=_mock_response(
json_data={"tabId": "t1", "url": "https://example.com"}
)):
result = json.loads(camofox_navigate("https://example.com", task_id="vnc-test"))
assert result["vnc_url"] == "http://localhost:6080"
assert "vnc_hint" in result
+66
View File
@@ -0,0 +1,66 @@
"""Tests for Hermes-managed Camofox state helpers."""
from unittest.mock import patch
import pytest
def _load_module():
from tools import browser_camofox_state as state
return state
class TestCamofoxStatePaths:
def test_paths_are_profile_scoped(self, tmp_path):
state = _load_module()
with patch.object(state, "get_hermes_home", return_value=tmp_path):
assert state.get_camofox_state_dir() == tmp_path / "browser_auth" / "camofox"
class TestCamofoxIdentity:
def test_identity_is_deterministic(self, tmp_path):
state = _load_module()
with patch.object(state, "get_hermes_home", return_value=tmp_path):
first = state.get_camofox_identity("task-1")
second = state.get_camofox_identity("task-1")
assert first == second
def test_identity_differs_by_task(self, tmp_path):
state = _load_module()
with patch.object(state, "get_hermes_home", return_value=tmp_path):
a = state.get_camofox_identity("task-a")
b = state.get_camofox_identity("task-b")
# Same user (same profile), different session keys
assert a["user_id"] == b["user_id"]
assert a["session_key"] != b["session_key"]
def test_identity_differs_by_profile(self, tmp_path):
state = _load_module()
with patch.object(state, "get_hermes_home", return_value=tmp_path / "profile-a"):
a = state.get_camofox_identity("task-1")
with patch.object(state, "get_hermes_home", return_value=tmp_path / "profile-b"):
b = state.get_camofox_identity("task-1")
assert a["user_id"] != b["user_id"]
def test_default_task_id(self, tmp_path):
state = _load_module()
with patch.object(state, "get_hermes_home", return_value=tmp_path):
identity = state.get_camofox_identity()
assert "user_id" in identity
assert "session_key" in identity
assert identity["user_id"].startswith("hermes_")
assert identity["session_key"].startswith("task_")
class TestCamofoxConfigDefaults:
def test_default_config_includes_managed_persistence_toggle(self):
from hermes_cli.config import DEFAULT_CONFIG
browser_cfg = DEFAULT_CONFIG["browser"]
assert browser_cfg["camofox"]["managed_persistence"] is False
def test_config_version_unchanged(self):
from hermes_cli.config import DEFAULT_CONFIG
# managed_persistence is auto-merged by _deep_merge, no version bump needed
assert DEFAULT_CONFIG["_config_version"] == 11
+174
View File
@@ -0,0 +1,174 @@
"""Tests for skill fuzzy patching via tools.fuzzy_match."""
import json
import os
from pathlib import Path
from unittest.mock import patch
import pytest
from tools.skill_manager_tool import (
_create_skill,
_patch_skill,
_write_file,
skill_manage,
)
SKILL_CONTENT = """\
---
name: test-skill
description: A test skill for unit testing.
---
# Test Skill
Step 1: Do the thing.
Step 2: Do another thing.
Step 3: Final step.
"""
# ---------------------------------------------------------------------------
# Fuzzy patching
# ---------------------------------------------------------------------------
class TestFuzzyPatchSkill:
@pytest.fixture(autouse=True)
def setup_skills(self, tmp_path, monkeypatch):
skills_dir = tmp_path / "skills"
skills_dir.mkdir()
monkeypatch.setattr("tools.skill_manager_tool.SKILLS_DIR", skills_dir)
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
self.skills_dir = skills_dir
def test_exact_match_still_works(self):
_create_skill("test-skill", SKILL_CONTENT)
result = _patch_skill("test-skill", "Step 1: Do the thing.", "Step 1: Done!")
assert result["success"] is True
content = (self.skills_dir / "test-skill" / "SKILL.md").read_text()
assert "Step 1: Done!" in content
def test_whitespace_trimmed_match(self):
"""Patch with extra leading whitespace should still find the target."""
skill = """\
---
name: ws-skill
description: Whitespace test
---
# Commands
def hello():
print("hi")
"""
_create_skill("ws-skill", skill)
# Agent sends patch with no leading whitespace (common LLM behaviour)
result = _patch_skill("ws-skill", "def hello():\n print(\"hi\")", "def hello():\n print(\"hello world\")")
assert result["success"] is True
content = (self.skills_dir / "ws-skill" / "SKILL.md").read_text()
assert 'print("hello world")' in content
def test_indentation_flexible_match(self):
"""Patch where only indentation differs should succeed."""
skill = """\
---
name: indent-skill
description: Indentation test
---
# Steps
1. First step
2. Second step
3. Third step
"""
_create_skill("indent-skill", skill)
# Agent sends with different indentation
result = _patch_skill(
"indent-skill",
"1. First step\n2. Second step",
"1. Updated first\n2. Updated second"
)
assert result["success"] is True
content = (self.skills_dir / "indent-skill" / "SKILL.md").read_text()
assert "Updated first" in content
def test_multiple_matches_blocked_without_replace_all(self):
"""Multiple fuzzy matches should return an error without replace_all."""
skill = """\
---
name: dup-skill
description: Duplicate test
---
# Steps
word word word
"""
_create_skill("dup-skill", skill)
result = _patch_skill("dup-skill", "word", "replaced")
assert result["success"] is False
assert "match" in result["error"].lower()
def test_replace_all_with_fuzzy(self):
skill = """\
---
name: dup-skill
description: Duplicate test
---
# Steps
word word word
"""
_create_skill("dup-skill", skill)
result = _patch_skill("dup-skill", "word", "replaced", replace_all=True)
assert result["success"] is True
content = (self.skills_dir / "dup-skill" / "SKILL.md").read_text()
assert "word" not in content
assert "replaced" in content
def test_no_match_returns_preview(self):
_create_skill("test-skill", SKILL_CONTENT)
result = _patch_skill("test-skill", "this does not exist anywhere", "replacement")
assert result["success"] is False
assert "file_preview" in result
def test_fuzzy_patch_on_supporting_file(self):
"""Fuzzy matching should also work on supporting files."""
_create_skill("test-skill", SKILL_CONTENT)
ref_content = " function hello() {\n console.log('hi');\n }"
_write_file("test-skill", "references/code.js", ref_content)
# Patch with stripped indentation
result = _patch_skill(
"test-skill",
"function hello() {\nconsole.log('hi');\n}",
"function hello() {\nconsole.log('hello world');\n}",
file_path="references/code.js"
)
assert result["success"] is True
content = (self.skills_dir / "test-skill" / "references" / "code.js").read_text()
assert "hello world" in content
def test_patch_preserves_frontmatter_validation(self):
"""Fuzzy matching should still run frontmatter validation on SKILL.md."""
_create_skill("test-skill", SKILL_CONTENT)
# Try to destroy the frontmatter via patch
result = _patch_skill("test-skill", "---\nname: test-skill", "BROKEN")
assert result["success"] is False
assert "structure" in result["error"].lower() or "frontmatter" in result["error"].lower()
def test_skill_manage_patch_uses_fuzzy(self):
"""The dispatcher should route to the fuzzy-matching patch."""
_create_skill("test-skill", SKILL_CONTENT)
raw = skill_manage(
action="patch",
name="test-skill",
old_string=" Step 1: Do the thing.", # extra leading space
new_string="Step 1: Updated.",
)
result = json.loads(raw)
# Should succeed via line-trimmed or indentation-flexible matching
assert result["success"] is True
+2 -2
View File
@@ -271,7 +271,7 @@ class TestPatchSkill:
_create_skill("my-skill", VALID_SKILL_CONTENT)
result = _patch_skill("my-skill", "this text does not exist", "replacement")
assert result["success"] is False
assert "not found" in result["error"]
assert "not found" in result["error"].lower() or "could not find" in result["error"].lower()
def test_patch_ambiguous_match_rejected(self, tmp_path):
content = """\
@@ -288,7 +288,7 @@ word word
_create_skill("my-skill", content)
result = _patch_skill("my-skill", "word", "replaced")
assert result["success"] is False
assert "matched" in result["error"]
assert "match" in result["error"].lower()
def test_patch_replace_all(self, tmp_path):
content = """\
+215
View File
@@ -0,0 +1,215 @@
"""Tests for skill content size limits.
Agent writes (create/edit/patch/write_file) are constrained to
MAX_SKILL_CONTENT_CHARS (100k) and MAX_SKILL_FILE_BYTES (1 MiB).
Hand-placed and hub-installed skills have no hard limit.
"""
import json
import os
from pathlib import Path
from unittest.mock import patch
import pytest
from tools.skill_manager_tool import (
MAX_SKILL_CONTENT_CHARS,
MAX_SKILL_FILE_BYTES,
_validate_content_size,
skill_manage,
)
@pytest.fixture(autouse=True)
def isolate_skills(tmp_path, monkeypatch):
"""Redirect SKILLS_DIR to a temp directory."""
skills_dir = tmp_path / "skills"
skills_dir.mkdir()
monkeypatch.setattr("tools.skill_manager_tool.SKILLS_DIR", skills_dir)
monkeypatch.setattr("tools.skills_tool.SKILLS_DIR", skills_dir)
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
return skills_dir
def _make_skill_content(body_chars: int) -> str:
"""Generate valid SKILL.md content with a body of the given character count."""
frontmatter = (
"---\n"
"name: test-skill\n"
"description: A test skill\n"
"---\n"
)
body = "# Test Skill\n\n" + ("x" * max(0, body_chars - 15))
return frontmatter + body
class TestValidateContentSize:
"""Unit tests for _validate_content_size."""
def test_within_limit(self):
assert _validate_content_size("a" * 1000) is None
def test_at_limit(self):
assert _validate_content_size("a" * MAX_SKILL_CONTENT_CHARS) is None
def test_over_limit(self):
err = _validate_content_size("a" * (MAX_SKILL_CONTENT_CHARS + 1))
assert err is not None
assert "100,001" in err
assert "100,000" in err
def test_custom_label(self):
err = _validate_content_size("a" * (MAX_SKILL_CONTENT_CHARS + 1), label="references/api.md")
assert "references/api.md" in err
class TestCreateSkillSizeLimit:
"""create action rejects oversized content."""
def test_create_within_limit(self, isolate_skills):
content = _make_skill_content(5000)
result = json.loads(skill_manage(action="create", name="small-skill", content=content))
assert result["success"] is True
def test_create_over_limit(self, isolate_skills):
content = _make_skill_content(MAX_SKILL_CONTENT_CHARS + 100)
result = json.loads(skill_manage(action="create", name="huge-skill", content=content))
assert result["success"] is False
assert "100,000" in result["error"]
def test_create_at_limit(self, isolate_skills):
# Content at exactly the limit should succeed
frontmatter = "---\nname: edge-skill\ndescription: Edge case\n---\n# Edge\n\n"
body_budget = MAX_SKILL_CONTENT_CHARS - len(frontmatter)
content = frontmatter + ("x" * body_budget)
assert len(content) == MAX_SKILL_CONTENT_CHARS
result = json.loads(skill_manage(action="create", name="edge-skill", content=content))
assert result["success"] is True
class TestEditSkillSizeLimit:
"""edit action rejects oversized content."""
def test_edit_over_limit(self, isolate_skills):
# Create a small skill first
small = _make_skill_content(1000)
json.loads(skill_manage(action="create", name="grow-me", content=small))
# Try to edit it to be oversized
big = _make_skill_content(MAX_SKILL_CONTENT_CHARS + 100)
# Fix the name in frontmatter
big = big.replace("name: test-skill", "name: grow-me")
result = json.loads(skill_manage(action="edit", name="grow-me", content=big))
assert result["success"] is False
assert "100,000" in result["error"]
class TestPatchSkillSizeLimit:
"""patch action checks resulting size, not just the new_string."""
def test_patch_that_would_exceed_limit(self, isolate_skills):
# Create a skill near the limit
near_limit = _make_skill_content(MAX_SKILL_CONTENT_CHARS - 50)
json.loads(skill_manage(action="create", name="near-limit", content=near_limit))
# Patch that adds enough to go over
result = json.loads(skill_manage(
action="patch",
name="near-limit",
old_string="# Test Skill",
new_string="# Test Skill\n" + ("y" * 200),
))
assert result["success"] is False
assert "100,000" in result["error"]
def test_patch_that_reduces_size_on_oversized_skill(self, isolate_skills, tmp_path):
"""Patches that shrink an already-oversized skill should succeed."""
# Manually create an oversized skill (simulating hand-placed)
skill_dir = tmp_path / "skills" / "bloated"
skill_dir.mkdir(parents=True)
oversized = _make_skill_content(MAX_SKILL_CONTENT_CHARS + 5000)
oversized = oversized.replace("name: test-skill", "name: bloated")
(skill_dir / "SKILL.md").write_text(oversized, encoding="utf-8")
assert len(oversized) > MAX_SKILL_CONTENT_CHARS
# Patch that removes content to bring it under the limit.
# Use replace_all to replace the repeated x's with a shorter string.
result = json.loads(skill_manage(
action="patch",
name="bloated",
old_string="x" * 100,
new_string="y",
replace_all=True,
))
# Should succeed because the result is well within limits
assert result["success"] is True
def test_patch_supporting_file_size_limit(self, isolate_skills):
"""Patch on a supporting file also checks size."""
small = _make_skill_content(1000)
json.loads(skill_manage(action="create", name="with-ref", content=small))
# Create a supporting file
json.loads(skill_manage(
action="write_file",
name="with-ref",
file_path="references/data.md",
file_content="# Data\n\nSmall content.",
))
# Try to patch it to be oversized
result = json.loads(skill_manage(
action="patch",
name="with-ref",
old_string="Small content.",
new_string="x" * (MAX_SKILL_CONTENT_CHARS + 100),
file_path="references/data.md",
))
assert result["success"] is False
assert "references/data.md" in result["error"]
class TestWriteFileSizeLimit:
"""write_file action enforces both char and byte limits."""
def test_write_file_over_char_limit(self, isolate_skills):
small = _make_skill_content(1000)
json.loads(skill_manage(action="create", name="file-test", content=small))
result = json.loads(skill_manage(
action="write_file",
name="file-test",
file_path="references/huge.md",
file_content="x" * (MAX_SKILL_CONTENT_CHARS + 1),
))
assert result["success"] is False
assert "100,000" in result["error"]
def test_write_file_within_limit(self, isolate_skills):
small = _make_skill_content(1000)
json.loads(skill_manage(action="create", name="file-ok", content=small))
result = json.loads(skill_manage(
action="write_file",
name="file-ok",
file_path="references/normal.md",
file_content="# Normal\n\n" + ("x" * 5000),
))
assert result["success"] is True
class TestHandPlacedSkillsNoLimit:
"""Skills dropped directly on disk are not constrained."""
def test_oversized_handplaced_skill_loads(self, isolate_skills, tmp_path):
"""A hand-placed 200k skill can still be read via skill_view."""
from tools.skills_tool import skill_view
skill_dir = tmp_path / "skills" / "manual-giant"
skill_dir.mkdir(parents=True)
huge = _make_skill_content(200_000)
huge = huge.replace("name: test-skill", "name: manual-giant")
(skill_dir / "SKILL.md").write_text(huge, encoding="utf-8")
result = json.loads(skill_view("manual-giant"))
assert "content" in result
# The full content is returned — no truncation at the storage layer
assert len(result["content"]) > MAX_SKILL_CONTENT_CHARS
+74 -9
View File
@@ -34,6 +34,9 @@ from typing import Any, Dict, Optional
import requests
from hermes_cli.config import load_config
from tools.browser_camofox_state import get_camofox_identity
logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
@@ -42,6 +45,8 @@ logger = logging.getLogger(__name__)
_DEFAULT_TIMEOUT = 30 # seconds per HTTP request
_SNAPSHOT_MAX_CHARS = 80_000 # camofox paginates at this limit
_vnc_url: Optional[str] = None # cached from /health response
_vnc_url_checked = False # only probe once per process
def get_camofox_url() -> str:
@@ -56,16 +61,52 @@ def is_camofox_mode() -> bool:
def check_camofox_available() -> bool:
"""Verify the Camofox server is reachable."""
global _vnc_url, _vnc_url_checked
url = get_camofox_url()
if not url:
return False
try:
resp = requests.get(f"{url}/health", timeout=5)
if resp.status_code == 200 and not _vnc_url_checked:
try:
data = resp.json()
vnc_port = data.get("vncPort")
if isinstance(vnc_port, int) and 1 <= vnc_port <= 65535:
from urllib.parse import urlparse
parsed = urlparse(url)
host = parsed.hostname or "localhost"
_vnc_url = f"http://{host}:{vnc_port}"
except (ValueError, KeyError):
pass
_vnc_url_checked = True
return resp.status_code == 200
except Exception:
return False
def get_vnc_url() -> Optional[str]:
"""Return the VNC URL if the Camofox server exposes one, or None."""
if not _vnc_url_checked:
check_camofox_available()
return _vnc_url
def _managed_persistence_enabled() -> bool:
"""Return whether Hermes-managed persistence is enabled for Camofox.
When enabled, sessions use a stable profile-scoped userId so the
Camofox server can map it to a persistent browser profile directory.
When disabled (default), each session gets a random userId (ephemeral).
Controlled by ``browser.camofox.managed_persistence`` in config.yaml.
"""
try:
camofox_cfg = load_config().get("browser", {}).get("camofox", {})
except Exception:
return False
return bool(camofox_cfg.get("managed_persistence"))
# ---------------------------------------------------------------------------
# Session management
# ---------------------------------------------------------------------------
@@ -75,16 +116,31 @@ _sessions_lock = threading.Lock()
def _get_session(task_id: Optional[str]) -> Dict[str, Any]:
"""Get or create a camofox session for the given task."""
"""Get or create a camofox session for the given task.
When managed persistence is enabled, uses a deterministic userId
derived from the Hermes profile so the Camofox server can map it
to the same persistent browser profile across restarts.
"""
task_id = task_id or "default"
with _sessions_lock:
if task_id in _sessions:
return _sessions[task_id]
session = {
"user_id": f"hermes_{uuid.uuid4().hex[:10]}",
"tab_id": None,
"session_key": f"task_{task_id[:16]}",
}
if _managed_persistence_enabled():
identity = get_camofox_identity(task_id)
session = {
"user_id": identity["user_id"],
"tab_id": None,
"session_key": identity["session_key"],
"managed": True,
}
else:
session = {
"user_id": f"hermes_{uuid.uuid4().hex[:10]}",
"tab_id": None,
"session_key": f"task_{task_id[:16]}",
"managed": False,
}
_sessions[task_id] = session
return session
@@ -172,11 +228,19 @@ def camofox_navigate(url: str, task_id: Optional[str] = None) -> str:
{"userId": session["user_id"], "url": url},
timeout=60,
)
return json.dumps({
result = {
"success": True,
"url": data.get("url", url),
"title": data.get("title", ""),
})
}
vnc = get_vnc_url()
if vnc:
result["vnc_url"] = vnc
result["vnc_hint"] = (
"Browser is visible via VNC. "
"Share this link with the user so they can watch the browser live."
)
return json.dumps(result)
except requests.HTTPError as e:
return json.dumps({"success": False, "error": f"Navigation failed: {e}"})
except requests.ConnectionError:
@@ -436,7 +500,7 @@ def camofox_vision(question: str, annotate: bool = False,
except Exception:
_vision_timeout = 120
analysis = call_llm(
response = call_llm(
messages=[{
"role": "user",
"content": [
@@ -452,6 +516,7 @@ def camofox_vision(question: str, annotate: bool = False,
task="vision",
timeout=_vision_timeout,
)
analysis = response.choices[0].message.content if response.choices else ""
return json.dumps({
"success": True,
+47
View File
@@ -0,0 +1,47 @@
"""Hermes-managed Camofox state helpers.
Provides profile-scoped identity and state directory paths for Camofox
persistent browser profiles. When managed persistence is enabled, Hermes
sends a deterministic userId derived from the active profile so that
Camofox can map it to the same persistent browser profile directory
across restarts.
"""
from __future__ import annotations
import uuid
from pathlib import Path
from typing import Dict, Optional
from hermes_constants import get_hermes_home
CAMOFOX_STATE_DIR_NAME = "browser_auth"
CAMOFOX_STATE_SUBDIR = "camofox"
def get_camofox_state_dir() -> Path:
"""Return the profile-scoped root directory for Camofox persistence."""
return get_hermes_home() / CAMOFOX_STATE_DIR_NAME / CAMOFOX_STATE_SUBDIR
def get_camofox_identity(task_id: Optional[str] = None) -> Dict[str, str]:
"""Return the stable Hermes-managed Camofox identity for this profile.
The user identity is profile-scoped (same Hermes profile = same userId).
The session key is scoped to the logical browser task so newly created
tabs within the same profile reuse the same identity contract.
"""
scope_root = str(get_camofox_state_dir())
logical_scope = task_id or "default"
user_digest = uuid.uuid5(
uuid.NAMESPACE_URL,
f"camofox-user:{scope_root}",
).hex[:10]
session_digest = uuid.uuid5(
uuid.NAMESPACE_URL,
f"camofox-session:{scope_root}:{logical_scope}",
).hex[:16]
return {
"user_id": f"hermes_{user_digest}",
"session_key": f"task_{session_digest}",
}
+57 -16
View File
@@ -82,6 +82,8 @@ SKILLS_DIR = HERMES_HOME / "skills"
MAX_NAME_LENGTH = 64
MAX_DESCRIPTION_LENGTH = 1024
MAX_SKILL_CONTENT_CHARS = 100_000 # ~36k tokens at 2.75 chars/token
MAX_SKILL_FILE_BYTES = 1_048_576 # 1 MiB per supporting file
# Characters allowed in skill names (filesystem-safe, URL-friendly)
VALID_NAME_RE = re.compile(r'^[a-z0-9][a-z0-9._-]*$')
@@ -177,6 +179,21 @@ def _validate_frontmatter(content: str) -> Optional[str]:
return None
def _validate_content_size(content: str, label: str = "SKILL.md") -> Optional[str]:
"""Check that content doesn't exceed the character limit for agent writes.
Returns an error message or None if within bounds.
"""
if len(content) > MAX_SKILL_CONTENT_CHARS:
return (
f"{label} content is {len(content):,} characters "
f"(limit: {MAX_SKILL_CONTENT_CHARS:,}). "
f"Consider splitting into a smaller SKILL.md with supporting files "
f"in references/ or templates/."
)
return None
def _resolve_skill_dir(name: str, category: str = None) -> Path:
"""Build the directory path for a new skill, optionally under a category."""
if category:
@@ -275,6 +292,10 @@ def _create_skill(name: str, content: str, category: str = None) -> Dict[str, An
if err:
return {"success": False, "error": err}
err = _validate_content_size(content)
if err:
return {"success": False, "error": err}
# Check for name collisions across all directories
existing = _find_skill(name)
if existing:
@@ -318,6 +339,10 @@ def _edit_skill(name: str, content: str) -> Dict[str, Any]:
if err:
return {"success": False, "error": err}
err = _validate_content_size(content)
if err:
return {"success": False, "error": err}
existing = _find_skill(name)
if not existing:
return {"success": False, "error": f"Skill '{name}' not found. Use skills_list() to see available skills."}
@@ -379,27 +404,29 @@ def _patch_skill(
content = target.read_text(encoding="utf-8")
count = content.count(old_string)
if count == 0:
# Use the same fuzzy matching engine as the file patch tool.
# This handles whitespace normalization, indentation differences,
# escape sequences, and block-anchor matching — saving the agent
# from exact-match failures on minor formatting mismatches.
from tools.fuzzy_match import fuzzy_find_and_replace
new_content, match_count, match_error = fuzzy_find_and_replace(
content, old_string, new_string, replace_all
)
if match_error:
# Show a short preview of the file so the model can self-correct
preview = content[:500] + ("..." if len(content) > 500 else "")
return {
"success": False,
"error": "old_string not found in the file.",
"error": match_error,
"file_preview": preview,
}
if count > 1 and not replace_all:
return {
"success": False,
"error": (
f"old_string matched {count} times. Provide more surrounding context "
f"to make the match unique, or set replace_all=true to replace all occurrences."
),
"match_count": count,
}
new_content = content.replace(old_string, new_string) if replace_all else content.replace(old_string, new_string, 1)
# Check size limit on the result
target_label = "SKILL.md" if not file_path else file_path
err = _validate_content_size(new_content, label=target_label)
if err:
return {"success": False, "error": err}
# If patching SKILL.md, validate frontmatter is still intact
if not file_path:
@@ -419,10 +446,9 @@ def _patch_skill(
_atomic_write_text(target, original_content)
return {"success": False, "error": scan_error}
replacements = count if replace_all else 1
return {
"success": True,
"message": f"Patched {'SKILL.md' if not file_path else file_path} in skill '{name}' ({replacements} replacement{'s' if replacements > 1 else ''}).",
"message": f"Patched {'SKILL.md' if not file_path else file_path} in skill '{name}' ({match_count} replacement{'s' if match_count > 1 else ''}).",
}
@@ -455,6 +481,21 @@ def _write_file(name: str, file_path: str, file_content: str) -> Dict[str, Any]:
if not file_content and file_content != "":
return {"success": False, "error": "file_content is required."}
# Check size limits
content_bytes = len(file_content.encode("utf-8"))
if content_bytes > MAX_SKILL_FILE_BYTES:
return {
"success": False,
"error": (
f"File content is {content_bytes:,} bytes "
f"(limit: {MAX_SKILL_FILE_BYTES:,} bytes / 1 MiB). "
f"Consider splitting into smaller files."
),
}
err = _validate_content_size(file_content, label=file_path)
if err:
return {"success": False, "error": err}
existing = _find_skill(name)
if not existing:
return {"success": False, "error": f"Skill '{name}' not found. Create it first with action='create'."}
+16
View File
@@ -2525,6 +2525,22 @@ def install_from_quarantine(
if install_dir.exists():
shutil.rmtree(install_dir)
# Warn (but don't block) if SKILL.md is very large
skill_md = quarantine_path / "SKILL.md"
if skill_md.exists():
try:
skill_size = skill_md.stat().st_size
if skill_size > 100_000:
logger.warning(
"Skill '%s' has a large SKILL.md (%s chars). "
"Large skills consume significant context when loaded. "
"Consider asking the author to split it into smaller files.",
safe_skill_name,
f"{skill_size:,}",
)
except OSError:
pass
install_dir.parent.mkdir(parents=True, exist_ok=True)
shutil.move(str(quarantine_path), str(install_dir))
@@ -85,6 +85,7 @@ For native Anthropic auth, Hermes prefers Claude Code's own credential files whe
| `BROWSERBASE_PROJECT_ID` | Browserbase project ID |
| `BROWSER_USE_API_KEY` | Browser Use cloud browser API key ([browser-use.com](https://browser-use.com/)) |
| `BROWSER_CDP_URL` | Chrome DevTools Protocol URL for local browser (set via `/browser connect`, e.g. `ws://localhost:9222`) |
| `CAMOFOX_URL` | Camofox local anti-detection browser URL (default: `http://localhost:9377`) |
| `BROWSER_INACTIVITY_TIMEOUT` | Browser session inactivity timeout in seconds |
| `FAL_KEY` | Image generation ([fal.ai](https://fal.ai/)) |
| `GROQ_API_KEY` | Groq Whisper STT API key ([groq.com](https://groq.com/)) |
+2
View File
@@ -1016,6 +1016,8 @@ browser:
inactivity_timeout: 120 # Seconds before auto-closing idle sessions
command_timeout: 30 # Timeout in seconds for browser commands (screenshot, navigate, etc.)
record_sessions: false # Auto-record browser sessions as WebM videos to ~/.hermes/browser_recordings/
camofox:
managed_persistence: false # When true, Camofox sessions persist cookies/logins across restarts
```
The browser toolset supports multiple providers. See the [Browser feature page](/docs/user-guide/features/browser) for details on Browserbase, Browser Use, and local Chrome CDP setup.
@@ -11,6 +11,7 @@ Hermes Agent includes a full browser automation toolset with multiple backend op
- **Browserbase cloud mode** via [Browserbase](https://browserbase.com) for managed cloud browsers and anti-bot tooling
- **Browser Use cloud mode** via [Browser Use](https://browser-use.com) as an alternative cloud browser provider
- **Camofox local mode** via [Camofox](https://github.com/jo-inc/camofox-browser) for local anti-detection browsing (Firefox-based fingerprint spoofing)
- **Local Chrome via CDP** — connect browser tools to your own Chrome instance using `/browser connect`
- **Local browser mode** via the `agent-browser` CLI and a local Chromium installation
@@ -54,6 +55,50 @@ BROWSER_USE_API_KEY=***
Get your API key at [browser-use.com](https://browser-use.com). Browser Use provides a cloud browser via its REST API. If both Browserbase and Browser Use credentials are set, Browserbase takes priority.
### Camofox local mode
[Camofox](https://github.com/jo-inc/camofox-browser) is a self-hosted Node.js server wrapping Camoufox (a Firefox fork with C++ fingerprint spoofing). It provides local anti-detection browsing without cloud dependencies.
```bash
# Install and run
git clone https://github.com/jo-inc/camofox-browser && cd camofox-browser
npm install && npm start # downloads Camoufox (~300MB) on first run
# Or via Docker
docker run -d --network host -e CAMOFOX_PORT=9377 jo-inc/camofox-browser
```
Then set in `~/.hermes/.env`:
```bash
CAMOFOX_URL=http://localhost:9377
```
Or configure via `hermes tools` → Browser Automation → Camofox.
When `CAMOFOX_URL` is set, all browser tools automatically route through Camofox instead of Browserbase or agent-browser.
#### Persistent browser sessions
By default, each Camofox session gets a random identity — cookies and logins don't survive across agent restarts. To enable persistent browser sessions:
```yaml
# In ~/.hermes/config.yaml
browser:
camofox:
managed_persistence: true
```
When enabled, Hermes sends a stable profile-scoped identity to Camofox. The Camofox server maps this identity to a persistent browser profile directory, so cookies, logins, and localStorage survive across restarts. Different Hermes profiles get different browser profiles (profile isolation).
:::note
The Camofox server must also be configured with `CAMOFOX_PROFILE_DIR` on the server side for persistence to work.
:::
#### VNC live view
When Camofox runs in headed mode (with a visible browser window), it exposes a VNC port in its health check response. Hermes automatically discovers this and includes the VNC URL in navigation responses, so the agent can share a link for you to watch the browser live.
### Local Chrome via CDP (`/browser connect`)
Instead of a cloud provider, you can attach Hermes browser tools to your own running Chrome instance via the Chrome DevTools Protocol (CDP). This is useful when you want to see what the agent is doing in real-time, interact with pages that require your own cookies/sessions, or avoid cloud browser costs.