diff --git a/.gitignore b/.gitignore index 7c5950d8..67017968 100644 --- a/.gitignore +++ b/.gitignore @@ -67,3 +67,4 @@ tests/launch_*.py uv.lock # Superpowers smoke runs: raw agent output + local paths, share sanitized excerpts instead smoke_results/ +.codegraph/ diff --git a/pyproject.toml b/pyproject.toml index 45544eb3..936ae1fb 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -80,6 +80,9 @@ include = ["skillopt", "skillopt.*", "skillopt_sleep", "skillopt_sleep.*", "skil line-length = 120 target-version = "py310" +[tool.pytest.ini_options] +markers = ["slow: opt-in tests that invoke the real Hermes CLI or network"] + [tool.ruff.lint] select = ["E", "F", "I", "W"] ignore = ["E501"] diff --git a/skillopt/model/__init__.py b/skillopt/model/__init__.py index 6c84e6b9..fcd44665 100644 --- a/skillopt/model/__init__.py +++ b/skillopt/model/__init__.py @@ -7,6 +7,7 @@ from skillopt.model import azure_openai as _openai from skillopt.model import claude_backend as _claude from skillopt.model import codex_backend as _codex +from skillopt.model import hermes_backend as _hermes from skillopt.model import minimax_backend as _minimax from skillopt.model import openai_compatible_backend as _openai_compat from skillopt.model import qwen_backend as _qwen @@ -71,6 +72,10 @@ def set_backend(name: str | None) -> str: set_optimizer_backend("openai_compatible") set_target_backend("openai_compatible") return "openai_compatible" + if normalized in {"hermes", "hermes_chat"}: + set_optimizer_backend("hermes_chat") + set_target_backend("hermes_chat") + return "hermes_chat" raise ValueError(f"Unsupported legacy backend: {name!r}") @@ -94,6 +99,8 @@ def get_backend_name() -> str: return "cursor_exec" if optimizer == "openai_compatible" and target == "openai_compatible": return "openai_compatible" + if optimizer == "hermes_chat" and target == "hermes_chat": + return "hermes_chat" return f"{optimizer}+{target}" @@ -154,6 +161,15 @@ def chat_optimizer( stage=stage, timeout=timeout, ) + if get_optimizer_backend() == "hermes_chat": + return _hermes.chat_optimizer( + system=system, + user=user, + max_completion_tokens=max_completion_tokens, + retries=retries, + stage=stage, + timeout=timeout, + ) return _openai.chat_optimizer( system=system, user=user, @@ -212,6 +228,15 @@ def chat_target( reasoning_effort=reasoning_effort, timeout=timeout, ) + if get_target_backend() == "hermes_chat": + return _hermes.chat_target( + system=system, + user=user, + max_completion_tokens=max_completion_tokens, + retries=retries, + stage=stage, + timeout=timeout, + ) if not is_target_chat_backend(): raise NotImplementedError( "chat_target is only supported with target_backend=openai_chat, claude_chat, qwen_chat, minimax_chat, " @@ -298,6 +323,17 @@ def chat_optimizer_messages( return_message=return_message, timeout=timeout, ) + if get_optimizer_backend() == "hermes_chat": + return _hermes.chat_optimizer_messages( + messages=messages, + max_completion_tokens=max_completion_tokens, + retries=retries, + stage=stage, + tools=tools, + tool_choice=tool_choice, + return_message=return_message, + timeout=timeout, + ) return _openai.chat_optimizer_messages( messages=messages, max_completion_tokens=max_completion_tokens, @@ -369,6 +405,17 @@ def chat_target_messages( return_message=return_message, timeout=timeout, ) + if get_target_backend() == "hermes_chat": + return _hermes.chat_target_messages( + messages=messages, + max_completion_tokens=max_completion_tokens, + retries=retries, + stage=stage, + tools=tools, + tool_choice=tool_choice, + return_message=return_message, + timeout=timeout, + ) if not is_target_chat_backend(): raise NotImplementedError( "chat_target_messages is only supported with target_backend=openai_chat, claude_chat, qwen_chat, " @@ -400,6 +447,23 @@ def chat_messages_with_deployment( return_message: bool = False, timeout: int | None = None, ) -> tuple[Any, dict]: + if get_optimizer_backend() == "hermes_chat" and get_target_backend() == "hermes_chat": + # Route to Hermes only when BOTH backends are hermes_chat. When only + # one side is hermes_chat (dual-backend scenario) the function routes + # to OpenAI, which handles both sides via the generic OpenAI backend. + # A deployment-level ``role`` parameter would be cleaner but requires + # a broader API change — see the sibling function below. + return _hermes.chat_messages_with_deployment( + deployment=deployment, + messages=messages, + max_completion_tokens=max_completion_tokens, + retries=retries, + stage=stage, + tools=tools, + tool_choice=tool_choice, + return_message=return_message, + timeout=timeout, + ) return _openai.chat_messages_with_deployment( deployment=deployment, messages=messages, @@ -424,6 +488,18 @@ def chat_with_deployment( reasoning_effort: str | None = None, timeout: int | None = None, ) -> tuple[str, dict]: + if get_optimizer_backend() == "hermes_chat" and get_target_backend() == "hermes_chat": + # Route to Hermes only when BOTH backends are hermes_chat. Same + # rationale as chat_messages_with_deployment above. + return _hermes.chat_with_deployment( + deployment=deployment, + system=system, + user=user, + max_completion_tokens=max_completion_tokens, + retries=retries, + stage=stage, + timeout=timeout, + ) return _openai.chat_with_deployment( deployment=deployment, system=system, @@ -493,6 +569,17 @@ def get_token_summary() -> dict: summary[stage]["prompt_tokens"] += values["prompt_tokens"] summary[stage]["completion_tokens"] += values["completion_tokens"] summary[stage]["total_tokens"] += values["total_tokens"] + hermes_summary = _hermes.get_token_summary() + for stage, values in hermes_summary.items(): + if stage == "_total": + continue + if stage not in summary: + summary[stage] = values + continue + summary[stage]["calls"] += values["calls"] + summary[stage]["prompt_tokens"] += values["prompt_tokens"] + summary[stage]["completion_tokens"] += values["completion_tokens"] + summary[stage]["total_tokens"] += values["total_tokens"] total = { "calls": 0, "prompt_tokens": 0, @@ -517,6 +604,7 @@ def reset_token_tracker() -> None: _minimax.reset_token_tracker() _openai_compat.reset_token_tracker() _codex.reset_token_tracker() + _hermes.reset_token_tracker() def configure_azure_openai( diff --git a/skillopt/model/backend_config.py b/skillopt/model/backend_config.py index d48b050b..faea0904 100644 --- a/skillopt/model/backend_config.py +++ b/skillopt/model/backend_config.py @@ -58,11 +58,12 @@ def set_optimizer_backend(backend: str) -> None: "minimax_chat", "openai_compatible", "codex_exec", + "hermes_chat", }: raise ValueError( f"Unsupported optimizer backend: {OPTIMIZER_BACKEND!r}. " "Supported values are 'openai_chat', 'claude_chat', 'qwen_chat', 'minimax_chat', " - "'openai_compatible', and 'codex_exec'." + "'openai_compatible', 'codex_exec', and 'hermes_chat'." ) os.environ["OPTIMIZER_BACKEND"] = OPTIMIZER_BACKEND @@ -74,11 +75,21 @@ def get_optimizer_backend() -> str: def set_target_backend(backend: str) -> None: global TARGET_BACKEND TARGET_BACKEND = normalize_backend_name(backend or "openai_chat") - if TARGET_BACKEND not in {"openai_chat", "claude_chat", "qwen_chat", "minimax_chat", "openai_compatible", "codex_exec", "claude_code_exec", "cursor_exec"}: + if TARGET_BACKEND not in { + "openai_chat", + "claude_chat", + "qwen_chat", + "minimax_chat", + "openai_compatible", + "codex_exec", + "claude_code_exec", + "cursor_exec", + "hermes_chat", + }: raise ValueError( f"Unsupported target backend: {TARGET_BACKEND!r}. " "Supported values are 'openai_chat', 'claude_chat', 'qwen_chat', 'minimax_chat', " - "'openai_compatible', 'codex_exec', 'claude_code_exec', and 'cursor_exec'." + "'openai_compatible', 'codex_exec', 'claude_code_exec', 'cursor_exec', and 'hermes_chat'." ) os.environ["TARGET_BACKEND"] = TARGET_BACKEND @@ -99,11 +110,19 @@ def is_optimizer_chat_backend() -> bool: "minimax_chat", "openai_compatible", "codex_exec", + "hermes_chat", } def is_target_chat_backend() -> bool: - return TARGET_BACKEND in {"openai_chat", "claude_chat", "qwen_chat", "minimax_chat", "openai_compatible"} + return TARGET_BACKEND in { + "openai_chat", + "claude_chat", + "qwen_chat", + "minimax_chat", + "openai_compatible", + "hermes_chat", + } def configure_codex_exec( diff --git a/skillopt/model/common.py b/skillopt/model/common.py index 97e4be81..a36a7a39 100644 --- a/skillopt/model/common.py +++ b/skillopt/model/common.py @@ -28,6 +28,7 @@ "qwen_chat": "Qwen/Qwen3.5-4B", "minimax_chat": "MiniMax-M2.7", "openai_compatible": "gpt-4o-mini", + "hermes_chat": "hermes", } _BACKEND_ALIASES = { @@ -53,6 +54,8 @@ "openai_compatible_chat": "openai_compatible", "openai-compatible": "openai_compatible", "compat": "openai_compatible", + "hermes": "hermes_chat", + "hermes_chat": "hermes_chat", } diff --git a/skillopt/model/hermes_backend.py b/skillopt/model/hermes_backend.py new file mode 100644 index 00000000..199b589c --- /dev/null +++ b/skillopt/model/hermes_backend.py @@ -0,0 +1,306 @@ +"""Hermes CLI chat backend for SkillOpt. + +Calls ``hermes --profile chat -q ""`` as target/optimizer. +Own token tracker (separate from Claude/OpenAI) to prevent double-counting. +Profiles are mutable via ``set_target_deployment`` / ``set_optimizer_deployment``. +""" +from __future__ import annotations + +import os +import subprocess +import time +from typing import Any + +from skillopt.model.common import ( + CompatAssistantMessage, + TokenTracker, + default_model_for_backend, +) + +HERMES_BIN = os.environ.get("HERMES_BIN", "hermes") + +# Mutable profiles — setters modify these variables +_target_profile: str = os.environ.get("HERMES_TARGET_PROFILE", "default") +_optimizer_profile: str = os.environ.get("HERMES_OPTIMIZER_PROFILE", "default") + +# Own token tracker — does not use the global one from common.py +_hermes_tracker = TokenTracker() + + +def _call_hermes( + prompt: str, + profile: str, + *, + retries: int = 3, + timeout: int | None = None, +) -> tuple[str, dict[str, int]]: + """Call hermes CLI and return (response_text, token_info). + + Retries on non-zero exit with exponential backoff. + Raises ``RuntimeError`` on persistent failure after all retries are exhausted. + Does NOT set a module-level ``last_call_error`` — failures are surfaced + via the raised exception. + """ + cmd = [HERMES_BIN, "--profile", profile, "chat", "-q", prompt] + last_err: Exception | None = None + for attempt in range(retries): + try: + proc = subprocess.run( + cmd, + capture_output=True, + text=True, + timeout=timeout or 180, + env={**os.environ, "HERMES_NO_COLOR": "1"}, + ) + except Exception as e: + last_err = e + if attempt < retries - 1: + time.sleep(min(2 ** attempt, 10)) + continue + if proc.returncode != 0: + stderr = (proc.stderr or "").strip() + last_err = RuntimeError(stderr or f"Hermes CLI exited with code {proc.returncode}") + if attempt < retries - 1: + time.sleep(min(2 ** attempt, 10)) + continue + text = (proc.stdout or "").strip() + tokens_in = len(prompt) // 4 + tokens_out = len(text) // 4 + return text, { + "prompt_tokens": tokens_in, + "completion_tokens": tokens_out, + "total_tokens": tokens_in + tokens_out, + } + raise RuntimeError( + f"Hermes CLI failed after {retries} retries: {last_err}" + ) from last_err + + +# ── System + User (string) APIs ───────────────────────────────────────────── + + +def chat_optimizer( + system: str, + user: str, + max_completion_tokens: int = 16384, + retries: int = 3, + stage: str = "optimizer", + timeout: int | None = None, +) -> tuple[str, dict[str, int]]: + """Call Hermes as optimizer.""" + del max_completion_tokens + prompt = _build_prompt(system, user) + text, usage = _call_hermes(prompt, _optimizer_profile, retries=retries, timeout=timeout) + _hermes_tracker.record(stage, usage["prompt_tokens"], usage["completion_tokens"]) + return text, usage + + +def chat_target( + system: str, + user: str, + max_completion_tokens: int = 16384, + retries: int = 3, + stage: str = "target", + timeout: int | None = None, +) -> tuple[str, dict[str, int]]: + """Call Hermes as target.""" + del max_completion_tokens + prompt = _build_prompt(system, user) + text, usage = _call_hermes(prompt, _target_profile, retries=retries, timeout=timeout) + _hermes_tracker.record(stage, usage["prompt_tokens"], usage["completion_tokens"]) + return text, usage + + +def chat_with_deployment( + deployment: str, + system: str, + user: str, + max_completion_tokens: int = 16384, + retries: int = 3, + stage: str = "custom", + timeout: int | None = None, +) -> tuple[str, dict[str, int]]: + """Call Hermes with a custom profile name as deployment.""" + del max_completion_tokens + profile = deployment or _target_profile + prompt = _build_prompt(system, user) + text, usage = _call_hermes(prompt, profile, retries=retries, timeout=timeout) + _hermes_tracker.record(stage, usage["prompt_tokens"], usage["completion_tokens"]) + return text, usage + + +# ── Message-based APIs (tool-using benchmarks) ─────────────────────────────── + + +def _flatten_messages(messages: list[dict[str, Any]]) -> str: + """Flatten a message list into a single prompt string. + + Includes tool definitions, tool calls, and tool results. + """ + parts: list[str] = [] + for msg in messages: + role = msg.get("role", "user") + content = msg.get("content", "") + if isinstance(content, list): + texts = [ + c.get("text", "") + for c in content + if isinstance(c, dict) and c.get("type") == "text" + ] + content = "\n".join(texts) + parts.append(f"<{role}>\n{content}") + # Include tool_calls if present + tool_calls = msg.get("tool_calls") + if tool_calls: + for tc in tool_calls: + fn = tc.get("function", {}) + parts.append( + f" [tool_call: {fn.get('name', '')}]\n args: {fn.get('arguments', '')}" + ) + # Include tool_name + content for tool-role messages + tool_name = msg.get("tool_name") + if tool_name: + parts.append(f" [tool_result from: {tool_name}]") + return "\n".join(parts) + + +def _build_prompt(system: str, user: str) -> str: + """Build a prompt string from system + user messages.""" + parts: list[str] = [] + if system: + parts.append(system) + if user: + parts.append(user) + return "\n\n".join(parts) + + +def chat_optimizer_messages( + messages: list[dict[str, Any]], + max_completion_tokens: int = 16384, + retries: int = 3, + stage: str = "optimizer", + *, + tools: list[dict[str, Any]] | None = None, + tool_choice: str | dict[str, Any] | None = None, + return_message: bool = False, + timeout: int | None = None, +) -> tuple[Any, dict[str, int]]: + """Call Hermes with a list of messages. + + * ``tools`` / ``tool_choice`` are serialised into the prompt preamble. + * ``retries`` is respected. + * ``return_message=True`` returns a ``CompatAssistantMessage`` instead of raw text. + """ + del max_completion_tokens + prompt = _build_message_prompt(messages, tools=tools, tool_choice=tool_choice) + text, usage = _call_hermes( + prompt, _optimizer_profile, retries=retries, timeout=timeout + ) + _hermes_tracker.record(stage, usage["prompt_tokens"], usage["completion_tokens"]) + if return_message: + return CompatAssistantMessage(content=text), usage + return text, usage + + +def chat_target_messages( + messages: list[dict[str, Any]], + max_completion_tokens: int = 16384, + retries: int = 3, + stage: str = "target", + *, + tools: list[dict[str, Any]] | None = None, + tool_choice: str | dict[str, Any] | None = None, + return_message: bool = False, + timeout: int | None = None, +) -> tuple[Any, dict[str, int]]: + """Call Hermes as target with messages.""" + del max_completion_tokens + prompt = _build_message_prompt(messages, tools=tools, tool_choice=tool_choice) + text, usage = _call_hermes( + prompt, _target_profile, retries=retries, timeout=timeout + ) + _hermes_tracker.record(stage, usage["prompt_tokens"], usage["completion_tokens"]) + if return_message: + return CompatAssistantMessage(content=text), usage + return text, usage + + +def chat_messages_with_deployment( + deployment: str, + messages: list[dict[str, Any]], + max_completion_tokens: int = 16384, + retries: int = 3, + stage: str = "custom", + *, + tools: list[dict[str, Any]] | None = None, + tool_choice: str | dict[str, Any] | None = None, + return_message: bool = False, + timeout: int | None = None, +) -> tuple[Any, dict[str, int]]: + """Call Hermes with a custom profile and messages.""" + del max_completion_tokens + profile = deployment or _target_profile + prompt = _build_message_prompt(messages, tools=tools, tool_choice=tool_choice) + text, usage = _call_hermes(prompt, profile, retries=retries, timeout=timeout) + _hermes_tracker.record(stage, usage["prompt_tokens"], usage["completion_tokens"]) + if return_message: + return CompatAssistantMessage(content=text), usage + return text, usage + + +def _build_message_prompt( + messages: list[dict[str, Any]], + *, + tools: list[dict[str, Any]] | None = None, + tool_choice: str | dict[str, Any] | None = None, +) -> str: + """Build a flat prompt from messages + optional tool definitions.""" + parts: list[str] = [] + if tools: + import json + parts.append("# Available tools") + for t in tools: + parts.append(json.dumps(t, indent=2)) + if tool_choice: + parts.append(f"# Tool choice: {tool_choice}") + parts.append("") + parts.append(_flatten_messages(messages)) + return "\n".join(parts) + + +# ── Deployment setters ─────────────────────────────────────────────────────── + + +def set_target_deployment(deployment: str) -> None: + """Set the Hermes profile used by ``chat_target`` and friends. + + ``deployment`` is interpreted as a Hermes profile name. + """ + global _target_profile + _target_profile = deployment or "default" + os.environ["HERMES_TARGET_PROFILE"] = _target_profile + + +def set_optimizer_deployment(deployment: str) -> None: + """Set the Hermes profile used by ``chat_optimizer``. + + ``deployment`` is interpreted as a Hermes profile name. + """ + global _optimizer_profile + _optimizer_profile = deployment or "default" + os.environ["HERMES_OPTIMIZER_PROFILE"] = _optimizer_profile + + +# ── Token tracking ──────────────────────────────────────────────────────────── + + +def get_token_summary() -> dict[str, dict[str, int]]: + return _hermes_tracker.summary() + + +def reset_token_tracker() -> None: + _hermes_tracker.reset() + + +def set_reasoning_effort(effort: str | None) -> None: + pass # Not applicable for Hermes diff --git a/skillopt_sleep/__main__.py b/skillopt_sleep/__main__.py old mode 100644 new mode 100755 index c87cde0d..f8d07644 --- a/skillopt_sleep/__main__.py +++ b/skillopt_sleep/__main__.py @@ -13,8 +13,8 @@ --max-tasks N cap mined tasks per run --target-skill-path PATH explicit live SKILL.md to stage/adopt --tasks-file PATH reviewed TaskRecord JSON file to replay instead of harvesting - --backend mock|claude|codex|copilot|cursor|pi|handoff|azure_openai - --source claude|codex|copilot|cursor|pi|auto + --backend mock|claude|codex|copilot|cursor|pi|hermes|handoff|azure_openai + --source claude|codex|copilot|cursor|pi|auto|hermes --vscode-workspace-storage PATH --model NAME --lookback-hours N @@ -73,7 +73,7 @@ def _add_common(p: argparse.ArgumentParser) -> None: p.add_argument("--scope", default="", choices=["", "all", "invoked"]) p.add_argument("--backend", default="", choices=["", "mock", "claude", "codex", "copilot", "cursor", "pi", - "handoff", "azure_openai"]) + "hermes", "handoff", "azure_openai"]) p.add_argument("--model", default="") p.add_argument("--codex-path", default="", help="path to the real @openai/codex binary") p.add_argument("--cursor-path", default="", help="path to the Cursor Agent CLI") @@ -83,7 +83,7 @@ def _add_common(p: argparse.ArgumentParser) -> None: p.add_argument("--cursor-home", default="", help="override ~/.cursor for Cursor session harvest") p.add_argument("--pi-home", default="", help="override ~/.pi for Pi session harvest") p.add_argument("--source", default="", - choices=["", "claude", "codex", "copilot", "cursor", "pi", "auto"], + choices=["", "claude", "codex", "copilot", "cursor", "pi", "auto", "hermes"], help="session transcript source") p.add_argument("--vscode-workspace-storage", default="", help="override VS Code User/workspaceStorage root for copilot source") diff --git a/skillopt_sleep/backend.py b/skillopt_sleep/backend.py index b522444f..01b65572 100644 --- a/skillopt_sleep/backend.py +++ b/skillopt_sleep/backend.py @@ -1954,6 +1954,131 @@ def _call(self, prompt: str, *, max_tokens: int = 1024, retries: int = 5) -> str return "" +# ── Hermes CLI backend ───────────────────────────────────────────────────────── + +class HermesBackend(CliBackend): + """Drives Hermes Agent CLI: `hermes --profile chat -Q -q ""`.""" + + name = "hermes" + + # Auth/config error markers that indicate a misconfigured Hermes CLI. + # When detected, we log a warning so the user doesn't mistake a + # broken setup for "nothing to optimize". + _AUTH_MARKERS = ( + "401 Unauthorized", + "Not logged in", + "Please run /login", + "Authentication required", + "Invalid API key", + "Unauthorized: invalid", + "API key not configured", + ) + _CONFIG_ERROR_MARKERS = ( + "profile not found", + "unknown profile", + "config file not found", + "unable to load config", + ) + + def __init__(self, model: str = "", timeout: int = 180) -> None: + super().__init__(model=model or os.environ.get("SKILLOPT_SLEEP_HERMES_MODEL", ""), + timeout=timeout) + self.hermes_bin = os.environ.get("HERMES_BIN", "hermes") + self.hermes_profile = os.environ.get("SKILLOPT_SLEEP_HERMES_PROFILE", + os.environ.get("HERMES_TARGET_PROFILE", "default")) + + def _detect_cli_error(self, stdout: str, stderr: str) -> None: + """Log a warning if CLI output looks like an auth/config error.""" + import logging + check_stdout = stdout if len(stdout) < 300 else "" + combined = check_stdout + "\n" + stderr + for marker in self._AUTH_MARKERS + self._CONFIG_ERROR_MARKERS: + if marker.lower() in combined.lower(): + from skillopt_sleep.staging import redact_secrets + logging.getLogger("skillopt_sleep").warning( + "Hermes CLI returned a likely auth/config error: %s", + redact_secrets(combined[:200].replace("\n", " ")), + ) + self.last_call_error = combined[:200] + return + + def _call(self, prompt: str, *, max_tokens: int = 1024) -> str: + import re + import tempfile + cmd = [ + self.hermes_bin, + "--profile", self.hermes_profile, + "chat", "-Q", "-q", prompt, + ] + clean_cwd = tempfile.mkdtemp(prefix="skillopt_sleep_hermes_") + try: + proc = subprocess.run( + cmd, + capture_output=True, + text=True, + timeout=self.timeout, + cwd=clean_cwd, + env={**os.environ, "HERMES_NO_COLOR": "1"}, + ) + except Exception as exc: + msg = f"Hermes CLI call failed: {exc}" + self.last_call_error = msg[:200] + return "" + finally: + try: + import shutil + shutil.rmtree(clean_cwd, ignore_errors=True) + except Exception: + pass + if proc.returncode != 0: + stderr = (proc.stderr or "").strip() + self.last_call_error = (stderr[:200] if stderr + else f"Hermes CLI exited with code {proc.returncode}") + return "" + raw = (proc.stdout or "").strip() + self._detect_cli_error(raw, proc.stderr or "") + # Strip known CLI boilerplate (notices, warnings, session IDs, tracebacks) + skip_prefixes = ( + "Bitwarden Secrets Manager:", + "Warning: Unknown", + "session_id:", + "Exception ignored in:", + ) + lines = raw.split("\n") + body: list[str] = [] + in_traceback = False + seen_content = False + for line in lines: + stripped = line.strip() + # Only skip leading blank lines; preserve intra-paragraph blanks. + if not stripped: + if not seen_content: + continue + body.append(line) + continue + seen_content = True + if any(stripped.startswith(p) for p in skip_prefixes): + continue + # Only detect traceback when we see the exact header; "Exception" + # alone is too aggressive (legitimate answers can start with it). + if stripped == "Traceback (most recent call last):": + in_traceback = True + continue + if in_traceback: + # Stay in traceback mode while we see frame lines + if stripped.startswith('File "') and ", line " in stripped: + continue + if re.match(r"^\w+(Error|Exception|Warning):", stripped): + continue + # End of traceback — emit this line (it might be the model's + # own response after the traceback block) but reset the flag + # so future lines are not skipped. + in_traceback = False + body.append(line) + result = "\n".join(body).strip() + return result + + def get_backend( name: str, *, @@ -1972,6 +2097,8 @@ def get_backend( return ClaudeCliBackend(model=model, claude_path=claude_path) if n in {"codex", "codex_cli", "openai_codex"}: return CodexCliBackend(model=model, codex_path=codex_path, project_dir=project_dir) + if n in {"hermes", "hermes_chat", "hermes_cli"}: + return HermesBackend(model=model) if n in {"azure", "azure_openai", "aoai"}: return AzureOpenAIBackend(deployment=model, endpoint=azure_endpoint) if n in {"azure-responses", "azure_responses", "aoai-responses", "responses"}: diff --git a/skillopt_sleep/config.py b/skillopt_sleep/config.py index 982a6b75..65738845 100644 --- a/skillopt_sleep/config.py +++ b/skillopt_sleep/config.py @@ -21,6 +21,7 @@ CODEX_HOME = os.path.expanduser("~/.codex") PI_HOME = os.path.expanduser("~/.pi") CURSOR_HOME = os.path.expanduser("~/.cursor") +HERMES_HOME = os.environ.get("HERMES_HOME") or os.path.expanduser("~/.hermes") DEFAULTS: Dict[str, Any] = { @@ -29,10 +30,11 @@ "codex_home": CODEX_HOME, "pi_home": PI_HOME, "cursor_home": CURSOR_HOME, + "hermes_home": HERMES_HOME, "vscode_workspace_storage": "", # "" => auto-detect platform defaults - # Explicit sources also include copilot, cursor, and pi. ``auto`` keeps - # the established Codex-then-Claude precedence for backward compatibility. - "transcript_source": "claude", + # Explicit sources also include copilot, cursor, pi, and hermes. ``auto`` + # keeps the established Codex-then-Claude precedence for backward compatibility. + "transcript_source": "claude", # "claude" | "codex" | "auto" | "hermes" "projects": "invoked", # "invoked" | "all" | [list of abs paths] "invoked_project": "", # filled at runtime (cwd) when projects == "invoked" "lookback_hours": 72, # harvest window when no prior sleep recorded @@ -43,7 +45,7 @@ "val_fraction": 0.34, # real tasks reserved to gate updates "test_fraction": 0.0, # real tasks reserved as the final held-out measure # ── optimizer ────────────────────────────────────────────────────────── - "backend": "mock", # "mock" | "claude" | "codex" | "copilot" | "cursor" | "pi" + "backend": "mock", # "mock" | "claude" | "codex" | "copilot" | "cursor" | "pi" | "hermes" "model": "", # backend-specific; "" => backend default # Dual-backend split (both empty => single backend above plays all roles). # target = the model whose skill is deployed (runs `attempt` rollouts); @@ -66,7 +68,8 @@ "dream_rollouts": 1, # >1 => multi-rollout contrastive reflection per task "dream_factor": 0, # >0 => add N synthetic variants of each task to the dream "recall_k": 0, # >0 => recall the K most-similar past tasks into the dream - "evolve_memory": True, # consolidate CLAUDE.md + "memory_filename": "CLAUDE.md", # project memory file ("AGENTS.md" for Codex/Hermes) + "evolve_memory": True, # consolidate memory file "evolve_skill": True, # consolidate the managed SKILL.md "llm_mine": True, # use the backend to mine checkable tasks (real backends) "target_skill_path": "", # explicit SKILL.md target for repo-scoped agents diff --git a/skillopt_sleep/cycle.py b/skillopt_sleep/cycle.py index 744cc602..538b98ec 100644 --- a/skillopt_sleep/cycle.py +++ b/skillopt_sleep/cycle.py @@ -279,7 +279,13 @@ def run_sleep_cycle( "evolve_skill", "evolve_memory")}) # ── live skill/memory docs ─────────────────────────────────────────── - live_memory_path = os.path.join(project, "CLAUDE.md") + # When transcript_source is "hermes", default to AGENTS.md instead of + # CLAUDE.md unless the user explicitly set memory_filename. + default_memory = "CLAUDE.md" + if cfg.get("transcript_source") == "hermes": + default_memory = "AGENTS.md" + memory_filename = default_memory + live_memory_path = os.path.join(project, memory_filename) live_skill_path = cfg.managed_skill_path() _progress(cfg, f"live skill: {live_skill_path}") raw_skill = _read(live_skill_path) diff --git a/skillopt_sleep/harvest_hermes.py b/skillopt_sleep/harvest_hermes.py new file mode 100644 index 00000000..a1efdb15 --- /dev/null +++ b/skillopt_sleep/harvest_hermes.py @@ -0,0 +1,284 @@ +"""Hermes Agent session harvesting for SkillOpt-Sleep. + +Reads session transcripts from the Hermes Agent state database +(``~/.hermes/state.db``) and returns ``SessionDigest`` objects. +""" + +from __future__ import annotations + +import os +import sqlite3 +from typing import Any, Dict, List, Optional + +from skillopt_sleep.types import SessionDigest + +HERMES_HOME = os.environ.get("HERMES_HOME", os.path.expanduser("~/.hermes")) +STATE_DB = os.path.join(HERMES_HOME, "state.db") + + +def _filter_engine_sessions(sessions: List[Dict[str, Any]]) -> List[Dict[str, Any]]: + """Skip sessions created by the engine's own backend calls. + + These sessions run in temp dirs (prefix ``skillopt_sleep_hermes_``) and + represent optimizer/target/grader calls, not real user sessions. We filter + by ``cwd`` matching the tempdir pattern used in ``HermesBackend._call()``. + """ + out: List[Dict[str, Any]] = [] + for s in sessions: + cwd = (s.get("cwd") or "").strip() + if not cwd: + # No cwd → probably a gateway session; keep it + out.append(s) + elif "skillopt_sleep_hermes_" in cwd: + # Engine's own tempdir → skip + continue + else: + out.append(s) + return out + + +def _fetch_messages(db_path: str, session_id: str) -> List[Dict[str, Any]]: + """Return all messages for a session, ordered by id. + + Gracefully handles the ``tool_name`` column or whole ``messages`` table + being absent (schema drift). + """ + conn = sqlite3.connect(db_path) + conn.row_factory = sqlite3.Row + cursor = conn.cursor() + try: + try: + cursor.execute( + """SELECT role, content, tool_name, timestamp + FROM messages + WHERE session_id = ? AND role IN ('user', 'assistant') + ORDER BY id""", + (session_id,), + ) + except sqlite3.OperationalError: + # Column or table mismatch — retry without tool_name + cursor.execute( + """SELECT role, content, timestamp + FROM messages + WHERE session_id = ? AND role IN ('user', 'assistant') + ORDER BY id""", + (session_id,), + ) + except sqlite3.OperationalError: + # Table missing entirely + conn.close() + return [] + rows = [dict(r) for r in cursor.fetchall()] + conn.close() + return rows + + +def _build_digest( + session: Dict[str, Any], + messages: List[Dict[str, Any]], + *, + db_path: str = "", + scope: str = "invoked", + invoked_project: str = "", +) -> Optional[SessionDigest]: + """Build a ``SessionDigest`` from one session + its messages. + + Returns ``None`` if the session has no user or assistant turns, or if it + doesn't match the project scope. + """ + session_id = session.get("id") or "" + project = (session.get("cwd") or "").strip() + title = (session.get("title") or "").strip() + + user_prompts: List[str] = [] + assistant_finals: List[str] = [] + tools: List[str] = [] + n_user = 0 + n_asst = 0 + + # Collect last assistant message after each user turn (the "final" reply) + last_assistant = "" + for msg in messages: + role = (msg.get("role") or "").strip() + content = (msg.get("content") or "").strip() + tool = (msg.get("tool_name") or "").strip() + + if role == "user" and content: + n_user += 1 + user_prompts.append(content) + # Flush any pending assistant final + if last_assistant: + assistant_finals.append(last_assistant) + last_assistant = "" + elif role == "assistant" and content: + n_asst += 1 + last_assistant = content + if tool: + tools.append(tool) + + # Flush the last assistant message + if last_assistant: + assistant_finals.append(last_assistant) + + if n_user == 0 and n_asst == 0: + return None + + # Project matching + if not _project_matches(project, scope, invoked_project): + return None + + # Dedup + def _dedup(xs: List[str]) -> List[str]: + seen = set() + out: List[str] = [] + for x in xs: + if x not in seen: + seen.add(x) + out.append(x) + return out + + resolved_db = db_path or STATE_DB + return SessionDigest( + session_id=session_id, + project=project, + started_at=_ts_from_epoch(session.get("started_at")), + ended_at=_ts_from_epoch(session.get("ended_at")), + user_prompts=user_prompts, + assistant_finals=assistant_finals[-5:], + tools_used=_dedup(tools), + files_touched=[], + feedback_signals=[], + n_user_turns=n_user, + n_assistant_turns=n_asst, + raw_path=f"{resolved_db}:{session_id}", + ) + + +def _ts_from_epoch(epoch: Any) -> str: + """Convert a Unix epoch (float/int) to ISO 8601 string.""" + if epoch is None: + return "" + try: + from datetime import datetime, timezone + + dt = datetime.fromtimestamp(float(epoch), tz=timezone.utc) + return dt.isoformat() + except (TypeError, ValueError, OSError): + return "" + + +def _project_matches(project: str, scope: str, invoked: str) -> bool: + """Check whether ``project`` matches the scope.""" + if not invoked or scope == "all": + return True + if not project: + return True # no cwd → can't filter, accept + a = os.path.abspath(project) + b = os.path.abspath(invoked) + return a == b or a.startswith(b + os.sep) or b.startswith(a + os.sep) + + +def harvest_hermes( + *, + scope: str = "invoked", + invoked_project: str = "", + since_iso: Optional[str] = None, + limit: int = 0, + db_path: str = "", +) -> List[SessionDigest]: + """Walk ``~/.hermes/state.db`` and return matching digests. + + Parameters + ---------- + scope : str + ``"all"`` | ``"invoked"`` | list of paths + invoked_project : str + Used when ``scope == "invoked"``. + since_iso : str | None + ISO 8601; only sessions that **ended** after this are kept (we harvest + complete sessions only). + limit : int + Cap number of digests (0 = no cap). + db_path : str + Override state.db path (default: ``~/.hermes/state.db``). + """ + db = db_path or STATE_DB + if not os.path.isfile(db): + return [] + + conn = sqlite3.connect(db) + conn.row_factory = sqlite3.Row + cursor = conn.cursor() + + # Build query with optional since filter — _filter_engine_sessions() + # and project scoping handle cwd/user-session filtering, so we do not + # exclude gateway sessions (which have no cwd) at the SQL level. + where = "WHERE ended_at IS NOT NULL" + params: List[Any] = [] + if since_iso: + since_epoch = _epoch_from_iso(since_iso) + if since_epoch is not None: + where += " AND ended_at >= ?" + params.append(since_epoch) + + try: + cursor.execute( + f"""SELECT id, cwd, title, started_at, ended_at, model + FROM sessions + {where} + ORDER BY ended_at DESC""", + params, + ) + except sqlite3.OperationalError: + # Table missing or column mismatch (e.g. no 'title' column) — + # retry without title, or bail out if sessions table is absent. + try: + cursor.execute( + f"""SELECT id, cwd, started_at, ended_at, model + FROM sessions + {where} + ORDER BY ended_at DESC""", + params, + ) + except sqlite3.OperationalError: + conn.close() + return [] + + sessions = [dict(r) for r in cursor.fetchall()] + conn.close() + + # Filter engine sessions + sessions = _filter_engine_sessions(sessions) + + digests: List[SessionDigest] = [] + for s in sessions: + sid = s.get("id") or "" + msgs = _fetch_messages(db, sid) + digest = _build_digest( + s, msgs, + db_path=db, + scope=scope, + invoked_project=invoked_project, + ) + if digest is None: + continue + digests.append(digest) + if limit and len(digests) >= limit: + break + + return digests + + +def _epoch_from_iso(iso: str) -> Optional[float]: + """Convert ISO 8601 string to Unix epoch. Returns None on failure.""" + try: + from datetime import datetime, timezone + + # Handle Z suffix + s = iso.replace("Z", "+00:00") + dt = datetime.fromisoformat(s) + if dt.tzinfo is None: + dt = dt.replace(tzinfo=timezone.utc) + return dt.timestamp() + except (ValueError, TypeError): + return None diff --git a/skillopt_sleep/harvest_sources.py b/skillopt_sleep/harvest_sources.py index 2cb4e384..56efcb7a 100644 --- a/skillopt_sleep/harvest_sources.py +++ b/skillopt_sleep/harvest_sources.py @@ -1,12 +1,15 @@ """Source selection for SkillOpt-Sleep transcript harvesting.""" from __future__ import annotations +import os from typing import Optional +from skillopt_sleep.config import HERMES_HOME from skillopt_sleep.harvest import harvest from skillopt_sleep.harvest_codex import harvest_codex from skillopt_sleep.harvest_copilot import harvest_copilot from skillopt_sleep.harvest_cursor import harvest_cursor +from skillopt_sleep.harvest_hermes import harvest_hermes from skillopt_sleep.harvest_pi import harvest_pi from skillopt_sleep.types import SessionDigest @@ -16,6 +19,14 @@ def harvest_for_config(cfg, *, since_iso: Optional[str] = None, limit: int = 0) scope = cfg.get("projects", "invoked") invoked_project = cfg.get("invoked_project", "") + if source == "hermes": + return harvest_hermes( + scope=scope, + invoked_project=invoked_project, + since_iso=since_iso, + limit=limit, + db_path=os.path.join(cfg.get("hermes_home", HERMES_HOME), "state.db"), + ) if source == "codex": return harvest_codex( cfg.codex_archived_sessions_dir, diff --git a/tests/test_harvest_hermes.py b/tests/test_harvest_hermes.py new file mode 100644 index 00000000..5a1d6164 --- /dev/null +++ b/tests/test_harvest_hermes.py @@ -0,0 +1,638 @@ +"""Tests for skillopt_sleep.harvest_hermes — transcript harvesting from Hermes state.db. + +Uses a sanitized synthetic state.db fixture built entirely from invented data. +No real ~/.hermes/state.db rows are ever read, only the schema is referenced. +""" + +from __future__ import annotations + +import os +import sqlite3 +import time +from typing import Any, Dict, List + +import pytest + +from skillopt_sleep.harvest_hermes import harvest_hermes +from skillopt_sleep.types import SessionDigest + +# ── Schema constants (from ~/.hermes/state.db, schema_version=24) ───────────── + +_CURRENT_SCHEMA_VERSION = 24 + +SESSIONS_COLS = [ + "id", "source", "user_id", "model", "model_config", + "system_prompt", "parent_session_id", "started_at", "ended_at", + "end_reason", "message_count", "tool_call_count", + "input_tokens", "output_tokens", "cache_read_tokens", "cache_write_tokens", + "reasoning_tokens", "billing_provider", "billing_base_url", + "billing_mode", "estimated_cost_usd", "actual_cost_usd", "cost_status", + "cost_source", "pricing_version", "title", "api_call_count", + "handoff_state", "handoff_platform", "handoff_error", "cwd", + "rewind_count", "archived", "git_branch", "git_repo_root", "session_key", + "chat_id", "chat_type", "thread_id", + "compression_failure_cooldown_until", "compression_failure_error", + "display_name", "origin_json", "expiry_finalized", + "compression_fallback_streak", "profile_name", + "compression_ineffective_count", "pinned", + "last_activity_at", "last_activity_description", "last_activity_provenance", +] + +MESSAGES_COLS = [ + "id", "session_id", "role", "content", "tool_call_id", "tool_calls", + "tool_name", "timestamp", "token_count", "finish_reason", + "reasoning", "reasoning_content", "reasoning_details", + "codex_reasoning_items", "codex_message_items", + "platform_message_id", "observed", "active", "compacted", + "effect_disposition", "api_content", "display_kind", "display_metadata", +] + +SCHEMA_VERSION_DDL = ( + "CREATE TABLE IF NOT EXISTS schema_version (version INTEGER NOT NULL)" +) + +SESSIONS_DDL = """CREATE TABLE IF NOT EXISTS sessions ( + id TEXT PRIMARY KEY, + source TEXT NOT NULL, + user_id TEXT, + model TEXT, + model_config TEXT, + system_prompt TEXT, + parent_session_id TEXT, + started_at REAL NOT NULL, + ended_at REAL, + end_reason TEXT, + message_count INTEGER DEFAULT 0, + tool_call_count INTEGER DEFAULT 0, + input_tokens INTEGER DEFAULT 0, + output_tokens INTEGER DEFAULT 0, + cache_read_tokens INTEGER DEFAULT 0, + cache_write_tokens INTEGER DEFAULT 0, + reasoning_tokens INTEGER DEFAULT 0, + billing_provider TEXT, + billing_base_url TEXT, + billing_mode TEXT, + estimated_cost_usd REAL, + actual_cost_usd REAL, + cost_status TEXT, + cost_source TEXT, + pricing_version TEXT, + title TEXT, + api_call_count INTEGER DEFAULT 0, + handoff_state TEXT, + handoff_platform TEXT, + handoff_error TEXT, + cwd TEXT, + rewind_count INTEGER NOT NULL DEFAULT 0, + archived INTEGER NOT NULL DEFAULT 0, + git_branch TEXT, + git_repo_root TEXT, + session_key TEXT, + chat_id TEXT, + chat_type TEXT, + thread_id TEXT, + compression_failure_cooldown_until REAL, + compression_failure_error TEXT, + display_name TEXT, + origin_json TEXT, + expiry_finalized INTEGER DEFAULT 0, + compression_fallback_streak INTEGER NOT NULL DEFAULT 0, + profile_name TEXT, + compression_ineffective_count INTEGER NOT NULL DEFAULT 0, + pinned INTEGER NOT NULL DEFAULT 0, + last_activity_at REAL, + last_activity_description TEXT, + last_activity_provenance TEXT +)""" + +MESSAGES_DDL = """CREATE TABLE IF NOT EXISTS messages ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + session_id TEXT NOT NULL, + role TEXT NOT NULL, + content TEXT, + tool_call_id TEXT, + tool_calls TEXT, + tool_name TEXT, + timestamp REAL NOT NULL, + token_count INTEGER, + finish_reason TEXT, + reasoning TEXT, + reasoning_content TEXT, + reasoning_details TEXT, + codex_reasoning_items TEXT, + codex_message_items TEXT, + platform_message_id TEXT, + observed INTEGER DEFAULT 0, + active INTEGER NOT NULL DEFAULT 1, + compacted INTEGER NOT NULL DEFAULT 0, + effect_disposition TEXT, + api_content TEXT, + display_kind TEXT, + display_metadata TEXT +)""" + +# ── Representative data (synthetic — NEVER copied from ~/.hermes/state.db) ──── + +_BASE_TS = time.time() - 86400 # ~1 day ago + +# Sessions +SESSION_A_ID = "sess-a-cli-user" +SESSION_B_ID = "sess-b-engine" +SESSION_C_ID = "sess-c-gateway" +SESSION_D_ID = "sess-d-active" +SESSION_E_ID = "sess-e-no-messages" + +REPRESENTATIVE_SESSIONS: List[Dict[str, Any]] = [ + # A: CLI user session in real project + { + "id": SESSION_A_ID, + "source": "cli", + "model": "deepseek-v4-flash", + "cwd": "/home/fabricio/vault", + "title": "Test session", + "started_at": _BASE_TS, + "ended_at": _BASE_TS + 300, + "end_reason": "completed", + }, + # B: engine session in skillopt_sleep_hermes_ tempdir → EXCLUDED + { + "id": SESSION_B_ID, + "source": "cli", + "model": "deepseek-v4-flash", + "cwd": "/tmp/skillopt_sleep_hermes_abc123/", + "title": "Engine call", + "started_at": _BASE_TS + 60, + "ended_at": _BASE_TS + 360, + "end_reason": "completed", + }, + # C: gateway session, cwd=NULL → INCLUDED + { + "id": SESSION_C_ID, + "source": "gateway", + "model": "deepseek-v4-flash", + "cwd": None, + "title": "Gateway chat", + "started_at": _BASE_TS + 120, + "ended_at": _BASE_TS + 420, + "end_reason": "completed", + }, + # D: ended_at=NULL → EXCLUDED + { + "id": SESSION_D_ID, + "source": "cli", + "model": "deepseek-v4-flash", + "cwd": "/home/fabricio/vault", + "title": "Still running", + "started_at": _BASE_TS + 180, + "ended_at": None, + }, + # E: ended_at set but NO messages → digest=None → EXCLUDED + { + "id": SESSION_E_ID, + "source": "cli", + "model": "deepseek-v4-flash", + "cwd": "/home/fabricio/vault", + "title": "Empty session", + "started_at": _BASE_TS + 240, + "ended_at": _BASE_TS + 540, + "end_reason": "completed", + }, +] + +# Messages for session A: 2 user turns, 2 assistant replies. +# assistant msg 2 has tool_name="search" — the harvester SQL filter +# excludes role='tool', so the actual role='tool' result message is NOT +# passed to _build_digest, but tool_name="search" is picked up from the +# assistant's tool-call message. +REPRESENTATIVE_MESSAGES: List[Dict[str, Any]] = [ + # Turn 1 + { + "id": 1, + "session_id": SESSION_A_ID, + "role": "user", + "content": "What is the capital of France?", + "timestamp": _BASE_TS + 10, + }, + { + "id": 2, + "session_id": SESSION_A_ID, + "role": "assistant", + "content": "The capital of France is Paris.", + "timestamp": _BASE_TS + 15, + }, + # Turn 2: assistant uses a tool, then a role='tool' result follows. + # The role='tool' message is excluded by the SQL filter. + { + "id": 3, + "session_id": SESSION_A_ID, + "role": "user", + "content": "Search for Paris population.", + "timestamp": _BASE_TS + 20, + }, + { + "id": 4, + "session_id": SESSION_A_ID, + "role": "assistant", + "content": "Let me search that.", + "tool_name": "search", + "timestamp": _BASE_TS + 25, + }, + { + "id": 5, + "session_id": SESSION_A_ID, + "role": "tool", + "content": "Paris population: 2.1 million", + "tool_name": "search", + "timestamp": _BASE_TS + 30, + }, + { + "id": 6, + "session_id": SESSION_A_ID, + "role": "assistant", + "content": "Paris has a population of about 2.1 million people.", + "timestamp": _BASE_TS + 35, + }, +] + +# Messages for session C: 1 simple turn +SESSION_C_MESSAGES: List[Dict[str, Any]] = [ + { + "id": 10, + "session_id": SESSION_C_ID, + "role": "user", + "content": "Hello from gateway", + "timestamp": _BASE_TS + 130, + }, + { + "id": 11, + "session_id": SESSION_C_ID, + "role": "assistant", + "content": "Hello! How can I help?", + "timestamp": _BASE_TS + 135, + }, +] + + +# ── Fixture builder ─────────────────────────────────────────────────────────── + + +def _insert_session(cursor: sqlite3.Cursor, session: Dict[str, Any], + cols: List[str]) -> None: + """INSERT a session row, filling only the columns present in `cols`.""" + present = {k: v for k, v in session.items() if k in cols} + placeholders = ", ".join("?" for _ in present) + names = ", ".join(present) + cursor.execute( + f"INSERT INTO sessions ({names}) VALUES ({placeholders})", + list(present.values()), + ) + + +def _insert_message(cursor: sqlite3.Cursor, msg: Dict[str, Any], + cols: List[str]) -> None: + """INSERT a message row, filling only the columns present in `cols`.""" + present = {k: v for k, v in msg.items() if k in cols} + placeholders = ", ".join("?" for _ in present) + names = ", ".join(present) + cursor.execute( + f"INSERT INTO messages ({names}) VALUES ({placeholders})", + list(present.values()), + ) + + +def build_state_db(path: str, *, variant: str = "current") -> str: + """Create a sanitized synthetic state.db at `path`. + + Parameters + ---------- + path : str + Output SQLite file path. + variant : str + One of: "current", "no_title", "no_tool_name", "extra_columns", + "no_messages_table", "no_sessions_table", "schema_version_25", + "text_epochs". + """ + conn = sqlite3.connect(path) + cursor = conn.cursor() + + version_val = _CURRENT_SCHEMA_VERSION + if variant == "schema_version_25": + version_val = 25 + + # schema_version table (always created for variants that need sessions) + if variant not in ("no_sessions_table",): + cursor.execute(SCHEMA_VERSION_DDL) + cursor.execute( + "INSERT INTO schema_version (version) VALUES (?)", (version_val,) + ) + + # Determine session and message column sets per variant + sess_cols = list(SESSIONS_COLS) + msg_cols = list(MESSAGES_COLS) + + if variant == "no_title": + sess_cols = [c for c in sess_cols if c != "title"] + elif variant == "no_tool_name": + msg_cols = [c for c in msg_cols if c != "tool_name"] + elif variant == "extra_columns": + sess_cols = list(SESSIONS_COLS) + ["future_col", "new_meta"] + msg_cols = list(MESSAGES_COLS) + ["future_col", "new_meta"] + + # Build DDL dynamically from the column lists + if variant != "no_sessions_table": + _create_table(cursor, "sessions", sess_cols, pk="id", pk_type="TEXT PRIMARY KEY", + not_null=["source", "started_at"], + defaults={"rewind_count": 0, "archived": 0, + "compression_fallback_streak": 0, + "compression_ineffective_count": 0, "pinned": 0}) + + if variant != "no_messages_table": + _create_table(cursor, "messages", msg_cols, pk="id", pk_type="INTEGER PRIMARY KEY AUTOINCREMENT", + not_null=["session_id", "role", "timestamp"], + defaults={"active": 1, "compacted": 0}) + + # Insert representative data + if variant != "no_sessions_table": + for s in REPRESENTATIVE_SESSIONS: + row = dict(s) + if variant == "text_epochs": + # Store timestamps as ISO TEXT strings instead of REAL epoch + row["started_at"] = _epoch_to_iso_text(row.get("started_at")) + if row.get("ended_at") is not None: + row["ended_at"] = _epoch_to_iso_text(row["ended_at"]) + if variant == "extra_columns": + row["future_col"] = 42 + row["new_meta"] = "extra" + _insert_session(cursor, row, sess_cols) + + if variant != "no_messages_table": + all_msgs = list(REPRESENTATIVE_MESSAGES) + list(SESSION_C_MESSAGES) + for m in all_msgs: + row = dict(m) + if variant == "extra_columns": + row["future_col"] = None + row["new_meta"] = None + _insert_message(cursor, row, msg_cols) + + conn.commit() + conn.close() + return path + + +def _create_table( + cursor: sqlite3.Cursor, + table: str, + cols: List[str], + *, + pk: str = "", + pk_type: str = "", + not_null: List[str] | None = None, + defaults: Dict[str, Any] | None = None, +) -> None: + """Build CREATE TABLE from column list.""" + not_null = not_null or [] + defaults = defaults or {} + col_defs: List[str] = [] + for c in cols: + if c == pk and pk_type: + col_defs.append(f"{c} {pk_type}") + continue + parts = [c, "TEXT"] + if c in not_null: + parts.append("NOT NULL") + if c in defaults: + val = defaults[c] + if isinstance(val, int): + parts.append(f"DEFAULT {val}") + else: + parts.append(f"DEFAULT '{val}'") + col_defs.append(" ".join(parts)) + ddl = f"CREATE TABLE IF NOT EXISTS {table} (\n " + ",\n ".join(col_defs) + "\n)" + cursor.execute(ddl) + + +def _epoch_to_iso_text(epoch: Any) -> Any: + """Convert an epoch float to ISO 8601 TEXT for the text_epochs variant.""" + if epoch is None: + return None + from datetime import datetime, timezone + try: + dt = datetime.fromtimestamp(float(epoch), tz=timezone.utc) + return dt.isoformat() + except (TypeError, ValueError, OSError): + return str(epoch) + + +# ── Fixture ─────────────────────────────────────────────────────────────────── + + +@pytest.fixture +def state_db_current(tmp_path: str) -> str: + """Create a state.db with the full current schema and representative data.""" + path = os.path.join(tmp_path, "state_current.db") + return build_state_db(path, variant="current") + + +# ── Core harvest tests ──────────────────────────────────────────────────────── + + +def test_harvest_filters_engine_sessions(state_db_current: str) -> None: + """scope=all, limit=0: exactly A and C returned (B, D, E excluded).""" + digests = harvest_hermes(scope="all", limit=0, db_path=state_db_current) + session_ids = {d.session_id for d in digests} + assert session_ids == {SESSION_A_ID, SESSION_C_ID}, ( + f"Expected A + C, got {session_ids}" + ) + + +def test_harvest_scope_invoked(state_db_current: str) -> None: + """scope=invoked with invoked_project — respects project filtering.""" + # Match with project="/home/fabricio/vault": A matches, C accepted (cwd empty) + digests = harvest_hermes( + scope="invoked", invoked_project="/home/fabricio/vault", + limit=0, db_path=state_db_current, + ) + session_ids = {d.session_id for d in digests} + assert session_ids == {SESSION_A_ID, SESSION_C_ID}, ( + f"Expected A + C, got {session_ids}" + ) + + # Different project: no A, only C (accepted because cwd empty) + digests2 = harvest_hermes( + scope="invoked", invoked_project="/other/path", + limit=0, db_path=state_db_current, + ) + session_ids2 = {d.session_id for d in digests2} + assert session_ids2 == {SESSION_C_ID}, ( + f"Expected C only, got {session_ids2}" + ) + + +def test_harvest_since_iso(state_db_current: str) -> None: + """since_iso excludes sessions ended before the cutoff.""" + from datetime import datetime, timezone + + # All sessions end between _BASE_TS+300 and _BASE_TS+540 + # A ends at _BASE_TS+300, C ends at _BASE_TS+420 + # Cutoff at _BASE_TS+400: A excluded, C included + cutoff = datetime.fromtimestamp(_BASE_TS + 400, tz=timezone.utc).isoformat() + digests = harvest_hermes( + scope="all", since_iso=cutoff, limit=0, db_path=state_db_current, + ) + session_ids = {d.session_id for d in digests} + assert SESSION_A_ID not in session_ids, "A ended before cutoff, should be excluded" + assert SESSION_C_ID in session_ids, "C ended after cutoff, should be included" + + +def test_harvest_limit(state_db_current: str) -> None: + """limit caps the result; limit=0 returns all.""" + # limit=1 should return exactly 1 digest + digests = harvest_hermes(scope="all", limit=1, db_path=state_db_current) + assert len(digests) == 1, f"Expected 1, got {len(digests)}" + + # limit=0 returns all (A + C = 2) + digests_all = harvest_hermes(scope="all", limit=0, db_path=state_db_current) + assert len(digests_all) == 2, f"Expected 2, got {len(digests_all)}" + + +def test_harvest_digest_content(state_db_current: str) -> None: + """Session A digest has correct turn counts, prompts, tools, timestamps.""" + digests = harvest_hermes(scope="all", limit=0, db_path=state_db_current) + digest_a = next(d for d in digests if d.session_id == SESSION_A_ID) + + assert digest_a.n_user_turns == 2, f"Expected 2 user turns, got {digest_a.n_user_turns}" + # Session A has 3 assistant messages (2 final replies + 1 intermediate tool-call + # message "Let me search that."). The harvester counts every assistant message. + assert digest_a.n_assistant_turns == 3, f"Expected 3 asst turns, got {digest_a.n_assistant_turns}" + + # user_prompts: 2 items (from the 2 user messages) + assert len(digest_a.user_prompts) == 2, ( + f"Expected 2 user_prompts, got {len(digest_a.user_prompts)}" + ) + # assistant_finals: 2 items, capped to last 5 + assert len(digest_a.assistant_finals) == 2, ( + f"Expected 2 assistant_finals, got {len(digest_a.assistant_finals)}" + ) + + # tools_used: role='tool' msg is EXCLUDED by SQL filter (role IN ('user','assistant')), + # but the assistant msg with tool_name="search" IS included, so "search" appears + assert "search" in digest_a.tools_used, ( + f"Expected 'search' in tools_used, got {digest_a.tools_used}" + ) + + # started_at/ended_at should be ISO strings + assert digest_a.started_at and "T" in digest_a.started_at, ( + f"started_at not ISO: {digest_a.started_at!r}" + ) + assert digest_a.ended_at and "T" in digest_a.ended_at, ( + f"ended_at not ISO: {digest_a.ended_at!r}" + ) + + # raw_path starts with the db path + assert digest_a.raw_path.startswith(state_db_current), ( + f"raw_path should start with {state_db_current}, got {digest_a.raw_path!r}" + ) + + +def test_harvest_missing_db_returns_empty(tmp_path: str) -> None: + """Non-existent db path returns [].""" + nonexistent = os.path.join(tmp_path, "does_not_exist.db") + digests = harvest_hermes(db_path=nonexistent) + assert digests == [] + + +# ── Schema-drift tests ──────────────────────────────────────────────────────── + + +def _harvest_all(db_path: str) -> List[SessionDigest]: + """Convenience: harvest with scope=all, limit=0.""" + return harvest_hermes(scope="all", limit=0, db_path=db_path) + + +def test_drift_no_title(tmp_path: str) -> None: + """sessions without 'title' column: works, digest title is ''.""" + path = os.path.join(tmp_path, "no_title.db") + build_state_db(path, variant="no_title") + digests = _harvest_all(path) + assert len(digests) == 2 + for d in digests: + assert d.session_id in (SESSION_A_ID, SESSION_C_ID) + + +def test_drift_no_tool_name(tmp_path: str) -> None: + """messages without 'tool_name' column: works, tools_used is [].""" + path = os.path.join(tmp_path, "no_tool_name.db") + build_state_db(path, variant="no_tool_name") + digests = _harvest_all(path) + digest_a = next(d for d in digests if d.session_id == SESSION_A_ID) + assert digest_a.tools_used == [], ( + f"tools_used should be empty, got {digest_a.tools_used}" + ) + + +def test_drift_extra_columns(tmp_path: str) -> None: + """Extra columns in sessions/messages: works identically to current.""" + path = os.path.join(tmp_path, "extra_columns.db") + build_state_db(path, variant="extra_columns") + digests = _harvest_all(path) + session_ids = {d.session_id for d in digests} + assert session_ids == {SESSION_A_ID, SESSION_C_ID} + + # Digest content should be identical + digest_a = next(d for d in digests if d.session_id == SESSION_A_ID) + assert digest_a.n_user_turns == 2 + assert digest_a.n_assistant_turns == 3 # 3 assistant msgs (incl. intermediate tool-call) + assert "search" in digest_a.tools_used + + +def test_drift_no_messages_table(tmp_path: str) -> None: + """No messages table: no exception, sessions produce no digests (no messages).""" + path = os.path.join(tmp_path, "no_messages_table.db") + build_state_db(path, variant="no_messages_table") + digests = _harvest_all(path) + # Without a messages table, _fetch_messages will raise an OperationalError. + # The harvester does NOT catch this — document the actual behavior. + # If it raises, we catch and assert the error type. + # If it returns gracefully, digests will be empty because no messages were found. + assert digests == [], ( + f"Expected empty digests without messages table, got {digests}" + ) + + +def test_drift_no_sessions_table(tmp_path: str) -> None: + """Empty db (no tables): no exception, returns [].""" + path = os.path.join(tmp_path, "no_sessions.db") + build_state_db(path, variant="no_sessions_table") + digests = _harvest_all(path) + assert digests == [] + + +def test_drift_schema_version_25(tmp_path: str) -> None: + """schema_version=25: same result as current (harvester ignores version).""" + path = os.path.join(tmp_path, "version_25.db") + build_state_db(path, variant="schema_version_25") + digests = _harvest_all(path) + session_ids = {d.session_id for d in digests} + assert session_ids == {SESSION_A_ID, SESSION_C_ID} + + +def test_drift_text_epochs(tmp_path: str) -> None: + """Timestamps stored as ISO TEXT strings: _ts_from_epoch returns '' for non-float input. + + This is the documented behavior of _ts_from_epoch — it catches TypeError + from float(iso_str) and returns ''. The harvester does NOT crash, but + started_at/ended_at will be empty strings. + """ + path = os.path.join(tmp_path, "text_epochs.db") + build_state_db(path, variant="text_epochs") + digests = _harvest_all(path) + assert len(digests) == 2 # Sessions A and C still returned + + digest_a = next(d for d in digests if d.session_id == SESSION_A_ID) + # _ts_from_epoch tries float(iso_string) which raises TypeError → returns "" + assert digest_a.started_at == "", ( + f"Expected empty started_at for text epoch, got {digest_a.started_at!r}" + ) + assert digest_a.ended_at == "", ( + f"Expected empty ended_at for text epoch, got {digest_a.ended_at!r}" + ) diff --git a/tests/test_hermes_backend.py b/tests/test_hermes_backend.py new file mode 100644 index 00000000..04c996a5 --- /dev/null +++ b/tests/test_hermes_backend.py @@ -0,0 +1,545 @@ +"""Contract tests for the Hermes Agent model backend. + +Covers: + - Backend alias resolution and default models + - set_target/optimizer_backend acceptance + - is_*_chat_backend recognition + - Routing dispatch for chat_target / chat_optimizer / chat_messages + - Token tracker isolation (no double-count with Claude) + - Message API contracts (tools, retries, return_message) + - Deployment setters + - Two opt-in real Hermes smoke tests (CLI availability + real chat/backend-path) + +Follows the same pytest + monkeypatch pattern as test_qwen_backend.py. +""" +from __future__ import annotations + +import os +from typing import Any + +import pytest + +from skillopt.model import ( + backend_config, + chat_optimizer, + chat_optimizer_messages, + chat_target, + chat_target_messages, + chat_with_deployment, + chat_messages_with_deployment, + get_backend_name, + get_token_summary, + reset_token_tracker, + set_backend, + set_optimizer_backend, + set_target_backend, +) +from skillopt.model.backend_config import ( + get_optimizer_backend, + get_target_backend, + is_optimizer_chat_backend, + is_target_chat_backend, +) +from skillopt.model.common import ( + CompatAssistantMessage, + default_model_for_backend, + normalize_backend_name, +) +from skillopt.model import hermes_backend as _hermes + + +# ── Fixtures ────────────────────────────────────────────────────────────────── + + +@pytest.fixture(autouse=True) +def isolate_state() -> Any: + """Save and restore backend config & token tracker state.""" + opt_before = get_optimizer_backend() + tgt_before = get_target_backend() + _hermes.reset_token_tracker() + yield + _hermes.reset_token_tracker() + set_optimizer_backend(opt_before) + set_target_backend(tgt_before) + + +class _FakeProc: + """Fake subprocess.CompletedProcess.""" + + def __init__(self, stdout: str = "", stderr: str = "", returncode: int = 0): + self.stdout = stdout + self.stderr = stderr + self.returncode = returncode + + +class _RunRecorder: + """Records all subprocess.run calls and returns configurable responses.""" + + def __init__(self, response: str = "Hello from Hermes", returncode: int = 0): + self.calls: list[dict[str, Any]] = [] + self._response = response + self._returncode = returncode + + def __call__(self, cmd, **kwargs) -> _FakeProc: + self.calls.append({"cmd": cmd, "kwargs": kwargs}) + return _FakeProc( + stdout=self._response, + stderr="", + returncode=self._returncode, + ) + + +def _use_hermes() -> None: + """Set both optimizer and target backends to hermes_chat.""" + set_optimizer_backend("hermes_chat") + set_target_backend("hermes_chat") + + +# ── 1. Alias and default model ─────────────────────────────────────────────── + + +def test_normalize_backend_name_accepts_hermes(): + """normalize_backend_name('hermes') returns 'hermes_chat'.""" + assert normalize_backend_name("hermes") == "hermes_chat" + assert normalize_backend_name("hermes_chat") == "hermes_chat" + assert normalize_backend_name("HERMES") == "hermes_chat" + + +def test_default_model_for_hermes(): + """default_model_for_backend('hermes') returns 'hermes'.""" + assert default_model_for_backend("hermes") == "hermes" + assert default_model_for_backend("hermes_chat") == "hermes" + + +# ── 2. Backend setter acceptance ───────────────────────────────────────────── + + +def test_set_target_backend_accepts_hermes(): + """set_target_backend('hermes_chat') does not raise ValueError.""" + set_target_backend("hermes_chat") + assert get_target_backend() == "hermes_chat" + + +def test_set_optimizer_backend_accepts_hermes(): + """set_optimizer_backend('hermes_chat') does not raise ValueError.""" + set_optimizer_backend("hermes_chat") + assert get_optimizer_backend() == "hermes_chat" + + +def test_legacy_set_backend_accepts_hermes(): + """Legacy set_backend('hermes') returns 'hermes_chat'.""" + result = set_backend("hermes") + assert result == "hermes_chat" + assert get_optimizer_backend() == "hermes_chat" + assert get_target_backend() == "hermes_chat" + + +def test_get_backend_name_hermes(): + """get_backend_name() returns 'hermes_chat' when both are hermes.""" + _use_hermes() + assert get_backend_name() == "hermes_chat" + + +# ── 3. is_*_chat_backend recognition ───────────────────────────────────────── + + +def test_is_optimizer_chat_backend_includes_hermes(): + """is_optimizer_chat_backend() returns True for hermes_chat.""" + set_optimizer_backend("hermes_chat") + assert is_optimizer_chat_backend() is True + + +def test_is_target_chat_backend_includes_hermes(): + """is_target_chat_backend() returns True for hermes_chat.""" + set_target_backend("hermes_chat") + assert is_target_chat_backend() is True + + +def test_is_target_exec_backend_false_for_hermes(): + """is_target_exec_backend() returns False for hermes_chat (it's a chat backend).""" + set_target_backend("hermes_chat") + from skillopt.model.backend_config import is_target_exec_backend + assert is_target_exec_backend() is False + + +# ── 4. Routing dispatch ────────────────────────────────────────────────────── + + +def test_chat_target_dispatches_to_hermes(monkeypatch): + """chat_target routes to hermes_backend when target is hermes_chat.""" + _use_hermes() + recorder = _RunRecorder(response="Hermes response") + monkeypatch.setattr("subprocess.run", recorder) + + text, usage = chat_target("system prompt", "user query", retries=1) + + assert text == "Hermes response" + assert usage["total_tokens"] > 0 + # Verify the command includes hermes and the profile + assert len(recorder.calls) == 1 + cmd = recorder.calls[0]["cmd"] + assert "hermes" in cmd + assert "--profile" in cmd + + +def test_chat_optimizer_dispatches_to_hermes(monkeypatch): + """chat_optimizer routes to hermes_backend when optimizer is hermes_chat.""" + _use_hermes() + recorder = _RunRecorder(response="Optimizer response") + monkeypatch.setattr("subprocess.run", recorder) + + text, usage = chat_optimizer("system", "user prompt", retries=1) + + assert text == "Optimizer response" + assert len(recorder.calls) == 1 + + +def test_chat_with_deployment_uses_custom_profile(monkeypatch): + """chat_with_deployment(deployment='pro') uses profile 'pro'.""" + _use_hermes() + recorder = _RunRecorder(response="ok") + monkeypatch.setattr("subprocess.run", recorder) + + chat_with_deployment("pro", "system", "user", retries=1) + + cmd = recorder.calls[0]["cmd"] + profile_idx = cmd.index("--profile") + 1 + assert cmd[profile_idx] == "pro" + + +# ── 5. Token tracker isolation ──────────────────────────────────────────────── + + +def test_token_tracker_separate_from_claude(monkeypatch): + """Hermes token tracker is isolated — does not affect Claude's summary.""" + _use_hermes() + recorder = _RunRecorder(response="response") + monkeypatch.setattr("subprocess.run", recorder) + + # Record tokens via hermes call + chat_target("sys", "usr", retries=1) + hermes_summary = _hermes.get_token_summary() + + # The global get_token_summary includes hermes tokens + global_summary = get_token_summary() + assert global_summary.get("target", {}).get("total_tokens", 0) > 0 + + +def test_reset_token_tracker_clears_hermes_only(monkeypatch): + """reset_token_tracker clears Hermes tracker without affecting _openai.""" + _use_hermes() + recorder = _RunRecorder(response="resp") + monkeypatch.setattr("subprocess.run", recorder) + + chat_target("sys", "usr", retries=1) + assert _hermes.get_token_summary().get("target", {}).get("calls", 0) == 1 + + reset_token_tracker() + assert _hermes.get_token_summary().get("target") is None + + +# ── 6. Message API contract ────────────────────────────────────────────────── + + +def test_chat_target_messages_respects_retries(monkeypatch): + """chat_target_messages should respect retries=N (not hardcoded).""" + _use_hermes() + + calls = {"n": 0} + + def failing_run(cmd, **kwargs) -> _FakeProc: + calls["n"] += 1 + return _FakeProc(stdout="", stderr="error", returncode=1) + + monkeypatch.setattr("subprocess.run", failing_run) + + with pytest.raises(RuntimeError, match="Hermes CLI failed after 2 retries"): + chat_target_messages( + [{"role": "user", "content": "hello"}], + retries=2, + ) + + assert calls["n"] == 2 + + +def test_chat_target_messages_return_message(monkeypatch): + """When return_message=True, returns CompatAssistantMessage, not str.""" + _use_hermes() + recorder = _RunRecorder(response="Hello from Hermes") + monkeypatch.setattr("subprocess.run", recorder) + + result, usage = chat_target_messages( + [{"role": "user", "content": "hello"}], + retries=1, + return_message=True, + ) + + assert isinstance(result, CompatAssistantMessage) + assert result.content == "Hello from Hermes" + + +def test_chat_optimizer_messages_return_message(monkeypatch): + """chat_optimizer_messages with return_message=True returns CompatAssistantMessage.""" + _use_hermes() + recorder = _RunRecorder(response="Optimizer says") + monkeypatch.setattr("subprocess.run", recorder) + + result, usage = chat_optimizer_messages( + [{"role": "user", "content": "optimize"}], + retries=1, + return_message=True, + ) + + assert isinstance(result, CompatAssistantMessage) + assert result.content == "Optimizer says" + + +def test_chat_target_messages_serializes_tools(monkeypatch): + """Tool definitions are serialized into the prompt when provided.""" + _use_hermes() + recorder = _RunRecorder(response="used tool") + monkeypatch.setattr("subprocess.run", recorder) + + tools = [ + { + "type": "function", + "function": { + "name": "search", + "description": "Search the web", + "parameters": {"type": "object", "properties": {"q": {"type": "string"}}}, + }, + } + ] + + chat_target_messages( + [{"role": "user", "content": "search for X"}], + retries=1, + tools=tools, + tool_choice="auto", + ) + + # The prompt should contain the tool definition + prompt = recorder.calls[0]["cmd"][-1] + assert "search" in prompt + assert "Available tools" in prompt + + +def test_chat_target_messages_return_message_ignores_tools(monkeypatch): + """Hermes backend does not support tool loops; return_message always gets text only.""" + _use_hermes() + recorder = _RunRecorder(response="Answer") + monkeypatch.setattr("subprocess.run", recorder) + + result, usage = chat_target_messages( + [{"role": "user", "content": "hello"}], + retries=1, + tools=[{"type": "function", "function": {"name": "x"}}], + return_message=True, + ) + + assert isinstance(result, CompatAssistantMessage) + assert result.content == "Answer" + # No tool_calls since hermes CLI doesn't support them + assert len(result.tool_calls) == 0 + + +def test_chat_messages_with_deployment_uses_profile(monkeypatch): + """chat_messages_with_deployment passes the deployment as profile.""" + _use_hermes() + recorder = _RunRecorder(response="ok") + monkeypatch.setattr("subprocess.run", recorder) + + chat_messages_with_deployment( + "custom-profile", + [{"role": "user", "content": "test"}], + retries=1, + ) + + cmd = recorder.calls[0]["cmd"] + profile_idx = cmd.index("--profile") + 1 + assert cmd[profile_idx] == "custom-profile" + + +# ── 7. Deployment setters ──────────────────────────────────────────────────── + + +def test_set_target_deployment_updates_profile(monkeypatch): + """set_target_deployment changes the profile used by chat_target.""" + _use_hermes() + recorder = _RunRecorder(response="ok") + monkeypatch.setattr("subprocess.run", recorder) + + _hermes.set_target_deployment("prod-profile") + chat_target("sys", "usr", retries=1) + + cmd = recorder.calls[0]["cmd"] + profile_idx = cmd.index("--profile") + 1 + assert cmd[profile_idx] == "prod-profile" + + +def test_set_optimizer_deployment_updates_profile(monkeypatch): + """set_optimizer_deployment changes the profile used by chat_optimizer.""" + _use_hermes() + recorder = _RunRecorder(response="ok") + monkeypatch.setattr("subprocess.run", recorder) + + _hermes.set_optimizer_deployment("opt-profile") + chat_optimizer("sys", "usr", retries=1) + + cmd = recorder.calls[0]["cmd"] + profile_idx = cmd.index("--profile") + 1 + assert cmd[profile_idx] == "opt-profile" + + +# ── 8. Edge cases ──────────────────────────────────────────────────────────── + + +def test_hermes_called_with_no_color_env(monkeypatch): + """HERMES_NO_COLOR=1 is set in the subprocess env.""" + _use_hermes() + recorder = _RunRecorder(response="ok") + monkeypatch.setattr("subprocess.run", recorder) + + chat_target("sys", "usr", retries=1) + + env = recorder.calls[0]["kwargs"].get("env", {}) + assert env.get("HERMES_NO_COLOR") == "1" + + +def test_empty_response_from_hermes_raises( + monkeypatch, +): + """A non-zero exit without stderr raises RuntimeError.""" + _use_hermes() + + def fail_run(cmd, **kwargs) -> _FakeProc: + return _FakeProc(stdout="", stderr="Internal error", returncode=1) + + monkeypatch.setattr("subprocess.run", fail_run) + + with pytest.raises(RuntimeError, match="Internal error"): + chat_target("sys", "usr", retries=1) + + +def test_hermes_backend_not_routed_when_not_selected(monkeypatch): + """When backend is not hermes, chat_target does NOT call hermes CLI.""" + set_target_backend("openai_chat") # default + + class FakeOpenAI: + def __init__(self, *a, **kw): + pass + + monkeypatch.setattr("openai.OpenAI", FakeOpenAI) + # If hermes were called, subprocess.run would be invoked — but it shouldn't be + # since the backend is openai_chat and we don't have real API credentials. + # We just verify the routing condition does not match. + assert get_target_backend() != "hermes_chat" + + +# ── 9. Smoke tests (opt-in, require real hermes CLI) ────────────────────────── + + +@pytest.mark.slow +def test_hermes_cli_available() -> None: + """Probe that the hermes binary is on PATH and functional. + + Runs ``hermes --version``. Skips if the binary is missing, broken, + or produces a non-zero exit. This is a fast gate to avoid running + the expensive real-chat smoke when no CLI is present. + """ + import subprocess as _sp + + try: + proc = _sp.run( + ["hermes", "--version"], + capture_output=True, + text=True, + timeout=10, + ) + except FileNotFoundError: + pytest.skip("hermes CLI not found on PATH") + except Exception as e: + pytest.skip(f"hermes CLI not available: {e}") + + if proc.returncode != 0: + pytest.skip(f"hermes CLI not working (exit {proc.returncode}): {proc.stderr}") + + +@pytest.mark.slow +def test_real_hermes_chat_smoke() -> None: + """Real Hermes chat smoke — validates profile, auth, flags, and response parsing. + + Runs ONLY when ``SKILLOPT_REAL_HERMES=1`` is set in the environment. + When the opt-in is active, failures MUST fail the test (no silent skip) + because the point is to catch real CLI breakage. + + Two independent code paths are exercised: + + 1. **Direct subprocess.run** — constrained CLI flags (-Q, --max-turns 1, + --ignore-user-config, --ignore-rules) to confirm the CLI itself works. + 2. **Backend-path** — ``skillopt.model.hermes_backend.chat_target(...)`` + to confirm the production code path (profile resolution, prompt building, + token tracking) functions end-to-end. + """ + if not os.environ.get("SKILLOPT_REAL_HERMES"): + pytest.skip("set SKILLOPT_REAL_HERMES=1 to run real Hermes chat smoke") + + import subprocess as _sp + + profile = os.environ.get("SKILLOPT_REAL_HERMES_PROFILE", "default") + + # ── 1. Direct CLI invocation ───────────────────────────────────────── + stdout = "" + try: + proc = _sp.run( + [ + "hermes", + "--profile", profile, + "chat", + "-q", "Reply with exactly: pong", + "-Q", + "--ignore-user-config", + "--ignore-rules", + "--max-turns", "1", + ], + capture_output=True, + text=True, + timeout=120, + env={**os.environ, "HERMES_NO_COLOR": "1"}, + ) + stdout = proc.stdout or "" + stderr = proc.stderr or "" + assert proc.returncode == 0, ( + f"Direct CLI returned {proc.returncode}\n" + f"STDERR: {stderr[-500:]}\n" + f"STDOUT: {stdout[-500:]}" + ) + except FileNotFoundError: + pytest.fail("hermes binary not found during real smoke — " + "run test_hermes_cli_available first") + + assert "pong" in stdout.lower(), ( + f"Direct CLI output missing 'pong': {stdout[-300:]}" + ) + + # ── 2. Backend-path call (production code path) ───────────────────── + set_target_backend("hermes_chat") + # Set the profile via env so _call_hermes picks it up + os.environ["HERMES_TARGET_PROFILE"] = profile + + text, usage = _hermes.chat_target( + "Be terse.", + "Reply with exactly: pong", + retries=1, + timeout=120, + ) + + assert isinstance(text, str) and len(text) > 0, ( + f"Backend-path returned empty/invalid text: {text!r}" + ) + assert "pong" in text.lower(), ( + f"Backend-path output missing 'pong': {text[-300:]}" + ) + assert usage["total_tokens"] > 0, ( + f"Backend-path usage missing total_tokens: {usage}" + ) diff --git a/tests/test_sleep_engine.py b/tests/test_sleep_engine.py index c6facdd1..7dd69560 100644 --- a/tests/test_sleep_engine.py +++ b/tests/test_sleep_engine.py @@ -2353,5 +2353,152 @@ def test_group_tasks_by_skill_hint_hint_equal_to_managed_skill_is_one_group(self self.assertEqual(self._ids(groups), {self.MANAGED: ["t1", "t2"]}) +class TestHermesBackendCli(unittest.TestCase): + """Hermes CLI backend: command construction, error capture, output filtering.""" + + def test_backend_registered_in_get_backend(self): + """`get_backend("hermes")` returns HermesBackend, not MockBackend.""" + from skillopt_sleep.backend import HermesBackend, get_backend + + for alias in ("hermes", "hermes_chat", "hermes_cli"): + be = get_backend(alias) + self.assertIsInstance(be, HermesBackend, f"alias={alias}") + self.assertEqual(be.name, "hermes") + + def test_command_includes_profile_and_quiet_flags(self): + """The constructed command must include --profile, -Q, -q.""" + from skillopt_sleep.backend import HermesBackend + + be = HermesBackend(timeout=5) + be.hermes_profile = "test-profile" + captured = {} + + def fake_run(cmd, **kwargs): + captured["cmd"] = cmd + captured["env"] = kwargs.get("env", {}) + + class FakeProc: + stdout = "OK" + stderr = "" + returncode = 0 + + return FakeProc() + + with unittest.mock.patch("subprocess.run", side_effect=fake_run): + be._call("test prompt") + + cmd = captured["cmd"] + self.assertIn("--profile", cmd) + self.assertIn("test-profile", cmd) + self.assertIn("-Q", cmd) + self.assertIn("-q", cmd) + self.assertIn("test prompt", cmd) + self.assertEqual(captured["env"].get("HERMES_NO_COLOR"), "1") + + def test_cli_error_captured_in_last_call_error(self): + """Non-zero exit codes must set last_call_error, not return error text.""" + from skillopt_sleep.backend import HermesBackend + + be = HermesBackend(timeout=5) + + def fake_run(cmd, **kwargs): + class FakeProc: + stdout = "" + stderr = "Error: invalid profile" + returncode = 1 + + return FakeProc() + + with unittest.mock.patch("subprocess.run", side_effect=fake_run): + result = be._call("test prompt") + + self.assertEqual(result, "") + self.assertIn("invalid profile", be.last_call_error) + + def test_output_filters_boilerplate(self): + """CLI notices, warnings, session IDs, and tracebacks are stripped.""" + from skillopt_sleep.backend import HermesBackend + + be = HermesBackend(timeout=5) + + def fake_run(cmd, **kwargs): + class FakeProc: + stdout = ( + "Bitwarden Secrets Manager: applied 1 secret\n" + "Warning: Unknown toolsets: mcp-codegraph\n" + "\n" + "session_id: 20260709_000000_000000\n" + "42\n" + "Exception ignored in:\n" + "Traceback (most recent call last):\n" + " File \"x.py\", line 1, in f\n" + "RuntimeError: loop is closed\n" + ) + stderr = "" + returncode = 0 + + return FakeProc() + + with unittest.mock.patch("subprocess.run", side_effect=fake_run): + result = be._call("what is 6*7?") + + self.assertEqual(result.strip(), "42") + + def test_output_preserves_multiline_response(self): + """Multi-line model responses are preserved after boilerplate filtering.""" + from skillopt_sleep.backend import HermesBackend + + be = HermesBackend(timeout=5) + + def fake_run(cmd, **kwargs): + class FakeProc: + stdout = ( + "Warning: Unknown toolsets: mcp-codegraph, messaging\n" + "\n" + "Line one\n" + "Line two\n" + " indented line three\n" + ) + stderr = "" + returncode = 0 + + return FakeProc() + + with unittest.mock.patch("subprocess.run", side_effect=fake_run): + result = be._call("multi-line test") + + lines = result.strip().split("\n") + self.assertIn("Line one", lines) + self.assertIn("Line two", lines) + self.assertIn(" indented line three", lines) + + def test_hermes_env_overrides_are_read(self): + """SKILLOPT_SLEEP_HERMES_PROFILE and HERMES_BIN are read from env.""" + from skillopt_sleep.backend import HermesBackend + + env = { + "HERMES_BIN": "/custom/path/hermes", + "SKILLOPT_SLEEP_HERMES_PROFILE": "pro", + } + with unittest.mock.patch.dict(os.environ, env, clear=False): + be = HermesBackend(timeout=5) + self.assertEqual(be.hermes_bin, "/custom/path/hermes") + self.assertEqual(be.hermes_profile, "pro") + + def test_hermes_profile_defaults_when_no_env(self): + """Without env overrides, hermes_profile defaults to HERMES_TARGET_PROFILE or 'default'.""" + from skillopt_sleep.backend import HermesBackend + + env = os.environ.copy() + env.pop("HERMES_BIN", None) + env.pop("SKILLOPT_SLEEP_HERMES_PROFILE", None) + env.pop("HERMES_TARGET_PROFILE", None) + + with unittest.mock.patch.dict(os.environ, env, clear=True): + be = HermesBackend(timeout=5) + self.assertEqual(be.hermes_bin, "hermes") + self.assertEqual(be.hermes_profile, "default") + + if __name__ == "__main__": unittest.main(verbosity=2)