diff --git a/services/hermes/agent-deployment.yaml b/services/hermes/agent-deployment.yaml index 2f9be77e..4bcf0be0 100644 --- a/services/hermes/agent-deployment.yaml +++ b/services/hermes/agent-deployment.yaml @@ -25,7 +25,7 @@ spec: ai.bstein.dev/execution: Hermes Kanban with durable direct Codex and Claude Code CLI workers ai.bstein.dev/model-policy: Jetson-assisted AUTO routing, low through xhigh, cross-provider fallback ai.bstein.dev/placement: rpi5 preferred; Jetson deferred until state storage is available - ai.bstein.dev/config-rev: "20260815-atlas-repository-remotes" + ai.bstein.dev/config-rev: "20260815-kanban-and-tool-recovery" vault.hashicorp.com/agent-inject: "true" vault.hashicorp.com/role: hermes-agent vault.hashicorp.com/agent-inject-secret-anthropic-token: kv/data/atlas/hermes/agent-tokens @@ -235,6 +235,46 @@ spec: resources: requests: {cpu: 100m, memory: 256Mi} limits: {cpu: "1", memory: 1Gi} + - name: repair-cassandra-kanban + image: registry.bstein.dev/bstein/hermes-agent@sha256:81970563e542f0720773e72297810b3a844b83e381e278f25c0916c78d930107 + imagePullPolicy: IfNotPresent + command: + - /opt/hermes/.venv/bin/python + - /opt/coordinator/repair_cassandra_kanban.py + - --database + - /opt/data/kanban/boards/cassandra/kanban.db + securityContext: + allowPrivilegeEscalation: false + runAsUser: 10000 + runAsGroup: 10000 + seccompProfile: + type: RuntimeDefault + volumeMounts: + - {name: home, mountPath: /opt/data} + - {name: coordinator, mountPath: /opt/coordinator, readOnly: true} + resources: + requests: {cpu: 25m, memory: 64Mi} + limits: {cpu: 250m, memory: 256Mi} + - name: patch-main-wrapper + image: registry.bstein.dev/bstein/hermes-agent@sha256:81970563e542f0720773e72297810b3a844b83e381e278f25c0916c78d930107 + imagePullPolicy: IfNotPresent + command: + - /opt/hermes/.venv/bin/python + - /opt/coordinator/patch_main_wrapper.py + - /opt/hermes/docker/main-wrapper.sh + - /patched/main-wrapper.sh + securityContext: + allowPrivilegeEscalation: false + runAsUser: 10000 + runAsGroup: 10000 + seccompProfile: + type: RuntimeDefault + volumeMounts: + - {name: coordinator, mountPath: /opt/coordinator, readOnly: true} + - {name: main-wrapper-patch, mountPath: /patched} + resources: + requests: {cpu: 25m, memory: 32Mi} + limits: {cpu: 100m, memory: 64Mi} - name: patch-auth image: registry.bstein.dev/bstein/hermes-agent@sha256:81970563e542f0720773e72297810b3a844b83e381e278f25c0916c78d930107 imagePullPolicy: IfNotPresent @@ -472,6 +512,7 @@ spec: - {name: home, mountPath: /opt/data} - {name: provider-auth, mountPath: /shared-auth} - {name: coordinator, mountPath: /opt/coordinator, readOnly: true} + - {name: main-wrapper-patch, mountPath: /opt/hermes/docker/main-wrapper.sh, subPath: main-wrapper.sh, readOnly: true} - {name: auth-patch, mountPath: /opt/hermes/hermes_cli/auth.py, subPath: auth.py} - {name: codex-runtime-patch, mountPath: /opt/hermes/hermes_cli/runtime_provider.py, subPath: runtime_provider.py} - {name: codex-runtime-patch, mountPath: /opt/hermes/agent/transports/codex_app_server_session.py, subPath: codex_app_server_session.py} @@ -921,6 +962,8 @@ spec: defaultMode: 0444 - name: auth-patch emptyDir: {} + - name: main-wrapper-patch + emptyDir: {} - name: tui-gateway-patch emptyDir: {} - name: api-server-patch diff --git a/services/hermes/kustomization.yaml b/services/hermes/kustomization.yaml index 094921b6..e12c42f7 100644 --- a/services/hermes/kustomization.yaml +++ b/services/hermes/kustomization.yaml @@ -76,10 +76,12 @@ configMapGenerator: - migrate_api_session_lineage.py=scripts/migrate_api_session_lineage.py - patch_api_server_sessions.py=scripts/patch_api_server_sessions.py - patch_hermes_auth.py=scripts/patch_hermes_auth.py + - patch_main_wrapper.py=scripts/patch_main_wrapper.py - patch_codex_runtime.py=scripts/patch_codex_runtime.py - patch_stream_recovery.py=scripts/patch_stream_recovery.py - patch_tui_gateway.py=scripts/patch_tui_gateway.py - patch_ttyd_index.py=scripts/patch_ttyd_index.py + - repair_cassandra_kanban.py=scripts/repair_cassandra_kanban.py - routing_catalog.py=scripts/routing_catalog.py - telegram_media_server.py=scripts/telegram_media_server.py options: diff --git a/services/hermes/scripts/claude_oauth_broker.py b/services/hermes/scripts/claude_oauth_broker.py index 02e24bf7..2a20b4c6 100644 --- a/services/hermes/scripts/claude_oauth_broker.py +++ b/services/hermes/scripts/claude_oauth_broker.py @@ -29,13 +29,9 @@ MAX_BODY_BYTES: Final = int( TIMEOUT_SECONDS: Final = float( os.environ.get("HERMES_CLAUDE_BROKER_READ_TIMEOUT", "1800") ) -MAX_CONCURRENCY: Final = int( - os.environ.get("HERMES_CLAUDE_BROKER_CONCURRENCY", "4") -) +MAX_CONCURRENCY: Final = int(os.environ.get("HERMES_CLAUDE_BROKER_CONCURRENCY", "4")) HEALTH_PATH: Final = Path( - os.environ.get( - "HERMES_CLAUDE_HEALTH_PATH", "/opt/data/provider-health/claude.json" - ) + os.environ.get("HERMES_CLAUDE_HEALTH_PATH", "/opt/data/provider-health/claude.json") ) ROUTED_MODEL_PREFIX: Final = "route/claude/" EFFORTS: Final = {"low", "medium", "high", "xhigh"} @@ -135,14 +131,92 @@ def _prompt(payload: dict[str, Any]) -> str: return ( "You are serving one model boundary for Hermes. The JSON below is the " "complete conversation and the only source of task context. Do not run " - "Claude Code tools or modify files yourself. If a listed external tool " - "is needed, return type=tool_calls with its exact name and a valid input " - "object. Otherwise return type=final and place the complete user-facing " + "Claude Code tools or modify files yourself. Every tool listed in the " + "contract is an available external Hermes tool. If the requested work " + "needs one, return type=tool_calls with its exact name and a valid input " + "object; Hermes executes it after this boundary. Never simulate a tool " + "result or claim that a listed tool succeeded, failed, or is unavailable. " + "Do not return a plan or blocker when an available tool can advance the " + "request. Otherwise return type=final and place the complete user-facing " "answer in text. Do not describe this envelope.\n\n" + json.dumps(contract, ensure_ascii=False, separators=(",", ":")) ) +def _tool_names(payload: dict[str, Any]) -> set[str]: + """Return the external Hermes tool names advertised at this boundary.""" + tools = payload.get("tools") + if not isinstance(tools, list): + return set() + return { + str(tool["name"]) + for tool in tools + if isinstance(tool, dict) and isinstance(tool.get("name"), str) + } + + +def _latest_user_text(payload: dict[str, Any]) -> str: + """Return the newest user text for false-positive-safe result validation.""" + messages = payload.get("messages") + if not isinstance(messages, list): + return "" + for message in reversed(messages): + if not isinstance(message, dict) or message.get("role") != "user": + continue + content = message.get("content") + if isinstance(content, str): + return content + if isinstance(content, list): + return "\n".join( + str(block.get("text") or "") + for block in content + if isinstance(block, dict) + ) + return "" + + +def _validate_structured(structured: dict[str, Any], payload: dict[str, Any]) -> None: + """Reject fabricated tool outcomes so Switchyard can try another lane.""" + response_type = structured.get("type") + tool_calls = structured.get("tool_calls") + if response_type not in {"final", "tool_calls"} or not isinstance(tool_calls, list): + raise RuntimeError("provider: Claude CLI returned an invalid result envelope") + available = _tool_names(payload) + for call in tool_calls: + if ( + not isinstance(call, dict) + or call.get("name") not in available + or not isinstance(call.get("input"), dict) + ): + raise RuntimeError("provider: Claude CLI returned an unavailable tool call") + if response_type == "tool_calls" and not tool_calls: + raise RuntimeError("provider: Claude CLI returned an empty tool request") + if response_type == "final" and tool_calls: + raise RuntimeError("provider: Claude CLI mixed a final answer with tool calls") + text = str(structured.get("text") or "") + latest_user = _latest_user_text(payload).lower() + lowered = text.lower() + unavailable_claims = ( + "no such tool available", + "tools that worked earlier", + "not reachable at this boundary", + ) + claims_tool_failure = any(claim in lowered for claim in unavailable_claims) + user_is_quoting_failure = any(claim in latest_user for claim in unavailable_claims) + names_tool = any( + re.search(rf"\b{re.escape(name.lower())}\b", lowered) for name in available + ) + if ( + response_type == "final" + and claims_tool_failure + and names_tool + and not user_is_quoting_failure + ): + raise RuntimeError( + "provider: Claude CLI fabricated an external tool availability failure" + ) + + def _usage(event: dict[str, Any]) -> dict[str, int]: """Translate Claude Code's result accounting into Anthropic token fields.""" raw = event.get("usage") @@ -153,9 +227,7 @@ def _usage(event: dict[str, Any]) -> dict[str, int]: "cache_creation_input_tokens": max( 0, int(raw.get("cache_creation_input_tokens") or 0) ), - "cache_read_input_tokens": max( - 0, int(raw.get("cache_read_input_tokens") or 0) - ), + "cache_read_input_tokens": max(0, int(raw.get("cache_read_input_tokens") or 0)), } @@ -312,9 +384,7 @@ def _invoke(payload: dict[str, Any]) -> tuple[dict[str, Any], dict[str, int], st "rate_limit": rate_limit, } ) - _atomic_health( - health - ) + _atomic_health(health) kind = "capacity" if CAPACITY_PATTERN.search(error_text) else "provider" raise RuntimeError(f"{kind}: {error_text[-1200:] or 'Claude CLI failed'}") structured = result_event.get("structured_output") @@ -326,6 +396,7 @@ def _invoke(payload: dict[str, Any]) -> tuple[dict[str, Any], dict[str, int], st structured = None if not isinstance(structured, dict): raise RuntimeError("provider: Claude CLI returned no structured result") + _validate_structured(structured, payload) usage = _usage(result_event) actual_model = str( next(iter(result_event.get("modelUsage") or {}), model) @@ -469,7 +540,9 @@ class Handler(BaseHTTPRequestHandler): ) def _check_auth(self) -> bool: - if _authorized(self.headers.get("Authorization"), self.headers.get("x-api-key")): + if _authorized( + self.headers.get("Authorization"), self.headers.get("x-api-key") + ): return True self._error(401, "authentication_error", "unauthorized") return False diff --git a/services/hermes/scripts/patch_main_wrapper.py b/services/hermes/scripts/patch_main_wrapper.py new file mode 100644 index 00000000..c613384a --- /dev/null +++ b/services/hermes/scripts/patch_main_wrapper.py @@ -0,0 +1,48 @@ +#!/usr/bin/env python3 +"""Load the Vault-derived tool environment in Hermes's main gateway.""" + +from __future__ import annotations + +import argparse +import shutil +from pathlib import Path + + +BEFORE = """# HOME comes through with-contenv as /root (the /init context). Override +""" +AFTER = """# The init container writes Vault-derived tool credentials here. Load them +# after with-contenv restores the container environment so dashboard-spawned +# terminal tools receive the same Git identity as durable CLI workers. +if [ -r /opt/data/.env ]; then + set -a + # shellcheck disable=SC1091 + . /opt/data/.env + set +a +fi + +# HOME comes through with-contenv as /root (the /init context). Override +""" + + +def patch(source: Path, destination: Path) -> None: + """Apply the narrow environment load and fail on upstream drift.""" + content = source.read_text(encoding="utf-8") + if BEFORE not in content: + raise RuntimeError("Hermes main-wrapper patch context changed") + destination.parent.mkdir(parents=True, exist_ok=True) + destination.write_text(content.replace(BEFORE, AFTER, 1), encoding="utf-8") + shutil.copymode(source, destination) + + +def main() -> int: + """Patch the wrapper path supplied by the deployment init container.""" + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("source", type=Path) + parser.add_argument("destination", type=Path) + args = parser.parse_args() + patch(args.source, args.destination) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/services/hermes/scripts/repair_cassandra_kanban.py b/services/hermes/scripts/repair_cassandra_kanban.py new file mode 100644 index 00000000..79e97e5d --- /dev/null +++ b/services/hermes/scripts/repair_cassandra_kanban.py @@ -0,0 +1,311 @@ +#!/usr/bin/env python3 +"""Recover the known Cassandra Kanban shared-page corruption at pod startup.""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import os +import re +import shutil +import sqlite3 +import tempfile +from contextlib import closing +from datetime import datetime, timezone +from itertools import islice +from pathlib import Path +from typing import Iterable + + +DEFAULT_DATABASE = Path("/opt/data/kanban/boards/cassandra/kanban.db") +EXPECTED_TABLES = { + "kanban_notify_subs", + "task_attachments", + "task_comments", + "task_events", + "task_links", + "task_runs", + "tasks", +} +COPY_ORDER = ( + "tasks", + "task_links", + "task_comments", + "task_events", + "task_runs", + "task_attachments", + "kanban_notify_subs", +) +COMMENTS_INDEX = "idx_comments_task" + + +class RecoveryRefused(RuntimeError): + """The database failure is not the reviewed, loss-bounded corruption.""" + + +def _quote(identifier: str) -> str: + """Quote one SQLite identifier from the inspected local schema.""" + return '"' + identifier.replace('"', '""') + '"' + + +def _connect_read_only(database: Path) -> sqlite3.Connection: + """Open a database without permitting the audit phase to mutate it.""" + return sqlite3.connect(f"file:{database}?mode=ro", uri=True) + + +def integrity_errors(connection: sqlite3.Connection) -> list[str]: + """Return integrity failures, or an empty list for a healthy database.""" + rows = [str(row[0]) for row in connection.execute("PRAGMA integrity_check")] + return [] if rows == ["ok"] else rows + + +def _schema_tables(connection: sqlite3.Connection) -> set[str]: + """Return application tables while excluding SQLite's internal tables.""" + return { + str(row[0]) + for row in connection.execute( + "SELECT name FROM sqlite_master " + "WHERE type = 'table' AND name NOT LIKE 'sqlite_%'" + ) + } + + +def _known_comments_alias_corruption( + connection: sqlite3.Connection, errors: Iterable[str] +) -> bool: + """Recognize only the observed task-comments page alias failure.""" + row = connection.execute( + "SELECT rootpage FROM sqlite_master " + "WHERE type = 'table' AND name = 'task_comments'" + ).fetchone() + if row is None: + return False + root_page = int(row[0]) + allowed = ( + re.compile(rf"Tree {root_page} page \d+ cell \d+: 2nd reference to page \d+"), + re.compile(rf"Tree {root_page} page \d+ cell \d+: Rowid \d+ out of order"), + re.compile(r"(?:NUMERIC|NULL) value in task_comments\.author"), + re.compile(rf"row \d+ missing from index {COMMENTS_INDEX}"), + ) + lines = [ + line + for error in errors + for line in str(error).splitlines() + if line and line != "*** in database main ***" + ] + if not lines or any( + not any(pattern.fullmatch(line) for pattern in allowed) for line in lines + ): + return False + text = "\n".join(lines) + return all( + marker in text + for marker in ( + "2nd reference to page", + "value in task_comments.author", + f"missing from index {COMMENTS_INDEX}", + ) + ) + + +def _schema_entries( + connection: sqlite3.Connection, +) -> list[tuple[str, str, str, str]]: + """Read the explicit schema objects needed to reconstruct the board.""" + return [ + (str(kind), str(name), str(table), str(statement)) + for kind, name, table, statement in connection.execute( + "SELECT type, name, tbl_name, sql FROM sqlite_master " + "WHERE sql IS NOT NULL AND type IN ('table', 'index', 'trigger', 'view') " + "ORDER BY CASE type WHEN 'table' THEN 0 WHEN 'index' THEN 1 " + "WHEN 'trigger' THEN 2 ELSE 3 END, name" + ) + if name != "sqlite_sequence" + ] + + +def _table_columns(connection: sqlite3.Connection, table: str) -> list[str]: + """Return columns in their persisted insertion order.""" + return [ + str(row[1]) for row in connection.execute(f"PRAGMA table_info({_quote(table)})") + ] + + +def _copy_table( + source: sqlite3.Connection, + destination: sqlite3.Connection, + table: str, +) -> int: + """Copy one table, using the intact comments index as the trusted row set.""" + columns = _table_columns(source, table) + if not columns: + raise RecoveryRefused(f"table {table} has no inspectable columns") + projection = ", ".join(_quote(column) for column in columns) + placeholders = ", ".join("?" for _ in columns) + insert = f"INSERT INTO {_quote(table)} ({projection}) VALUES ({placeholders})" + if table == "task_comments": + # The intact index is the authority for which comments existed, but a + # covering lookup follows the damaged table page and can abort. Walk + # the table once and retain only rowids present in that intact index. + trusted_rowids = { + int(row[0]) + for row in source.execute( + f"SELECT rowid FROM {_quote(table)} " + f"INDEXED BY {_quote(COMMENTS_INDEX)}" + ) + } + cursor = source.execute( + f"SELECT rowid, {projection} FROM {_quote(table)} NOT INDEXED " + "ORDER BY rowid" + ) + rows = (row[1:] for row in cursor if int(row[0]) in trusted_rowids) + else: + cursor = source.execute( + f"SELECT {projection} FROM {_quote(table)} NOT INDEXED ORDER BY rowid" + ) + rows = iter(cursor) + copied = 0 + while batch := list(islice(rows, 500)): + destination.executemany(insert, batch) + copied += len(batch) + if table == "task_comments" and copied != len(trusted_rowids): + raise RecoveryRefused("trusted task-comments rows could not all be recovered") + return copied + + +def _populate_replacement( + source: sqlite3.Connection, replacement: Path +) -> dict[str, int]: + """Build and verify a clean database using only reviewed source rows.""" + entries = _schema_entries(source) + tables = [entry for entry in entries if entry[0] == "table"] + later = [entry for entry in entries if entry[0] != "table"] + counts: dict[str, int] = {} + with closing(sqlite3.connect(replacement)) as destination: + destination.execute("PRAGMA foreign_keys = OFF") + destination.execute("PRAGMA journal_mode = DELETE") + destination.execute("BEGIN IMMEDIATE") + for _, _, _, statement in tables: + destination.execute(statement) + for table in COPY_ORDER: + counts[table] = _copy_table(source, destination, table) + if source.execute( + "SELECT 1 FROM sqlite_master WHERE name = 'sqlite_sequence'" + ).fetchone(): + destination.execute("DELETE FROM sqlite_sequence") + destination.executemany( + "INSERT INTO sqlite_sequence(name, seq) VALUES (?, ?)", + source.execute("SELECT name, seq FROM sqlite_sequence").fetchall(), + ) + for _, _, _, statement in later: + destination.execute(statement) + user_version = int(source.execute("PRAGMA user_version").fetchone()[0]) + destination.execute(f"PRAGMA user_version = {user_version}") + destination.commit() + failures = integrity_errors(destination) + foreign_key_failures = destination.execute( + "PRAGMA foreign_key_check" + ).fetchall() + if failures or foreign_key_failures: + raise RecoveryRefused( + "replacement verification failed: " + f"integrity={failures!r} foreign_keys={foreign_key_failures!r}" + ) + return counts + + +def _backup_path(database: Path) -> Path: + """Return a unique evidence filename containing the damaged file hash.""" + digest = hashlib.sha256(database.read_bytes()).hexdigest()[:16] + timestamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") + return database.with_name( + f"{database.name}.corrupt.recovery-{timestamp}-{digest}.bak" + ) + + +def _durable_copy(source: Path, destination: Path) -> None: + """Copy a recovery artifact and flush it before replacing live data.""" + with source.open("rb") as reader, destination.open("xb") as writer: + shutil.copyfileobj(reader, writer) + writer.flush() + os.fsync(writer.fileno()) + shutil.copystat(source, destination) + + +def recover_database(database: Path, errors: list[str]) -> dict[str, object]: + """Atomically replace the known corrupt board while retaining all evidence.""" + database = database.resolve() + with closing(_connect_read_only(database)) as source: + if _schema_tables(source) != EXPECTED_TABLES: + raise RecoveryRefused("Kanban schema differs from the reviewed table set") + index = source.execute( + "SELECT tbl_name FROM sqlite_master WHERE type = 'index' AND name = ?", + (COMMENTS_INDEX,), + ).fetchone() + if index != ("task_comments",): + raise RecoveryRefused("trusted task-comments index is unavailable") + if not _known_comments_alias_corruption(source, errors): + raise RecoveryRefused( + "integrity failure does not match the reviewed corruption" + ) + descriptor, temporary_name = tempfile.mkstemp( + prefix=f".{database.name}.recovery-", dir=database.parent + ) + os.close(descriptor) + replacement = Path(temporary_name) + replacement.unlink() + try: + counts = _populate_replacement(source, replacement) + except Exception: + replacement.unlink(missing_ok=True) + raise + + backup = _backup_path(database) + _durable_copy(database, backup) + for suffix in ("-wal", "-shm", "-journal"): + journal = database.with_name(database.name + suffix) + if journal.exists(): + os.replace(journal, backup.with_name(backup.name + suffix)) + os.chmod(replacement, database.stat().st_mode & 0o777) + os.replace(replacement, database) + directory_fd = os.open(database.parent, os.O_RDONLY) + try: + os.fsync(directory_fd) + finally: + os.close(directory_fd) + return { + "state": "recovered", + "database": str(database), + "backup": str(backup), + "rows": counts, + } + + +def repair_if_needed(database: Path = DEFAULT_DATABASE) -> dict[str, object]: + """Check one board and recover only its known, evidence-backed failure.""" + if not database.is_file(): + return {"state": "absent", "database": str(database)} + with closing(_connect_read_only(database)) as connection: + errors = integrity_errors(connection) + if not errors: + return {"state": "healthy", "database": str(database)} + return recover_database(database, errors) + + +def main() -> int: + """Run the startup repair and emit a non-secret machine-readable result.""" + parser = argparse.ArgumentParser() + parser.add_argument("--database", type=Path, default=DEFAULT_DATABASE) + args = parser.parse_args() + try: + result = repair_if_needed(args.database) + except (OSError, sqlite3.DatabaseError, RecoveryRefused) as error: + print(f"Kanban recovery refused: {error}", flush=True) + return 1 + print(json.dumps(result, sort_keys=True), flush=True) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/testing/tests/test_hermes_claude_broker.py b/testing/tests/test_hermes_claude_broker.py new file mode 100644 index 00000000..4d1eb398 --- /dev/null +++ b/testing/tests/test_hermes_claude_broker.py @@ -0,0 +1,101 @@ +"""Claude subscription broker tool-boundary regression tests.""" + +from __future__ import annotations + +import importlib.util +import sys +from pathlib import Path +from types import ModuleType + +import pytest + + +ROOT = Path(__file__).resolve().parents[2] +SCRIPTS = ROOT / "services/hermes/scripts" + + +def _module(monkeypatch): + routing = ModuleType("routing_catalog") + routing.load_catalog = lambda: {} + routing.resolve_route = lambda route: route + monkeypatch.setitem(sys.modules, "routing_catalog", routing) + spec = importlib.util.spec_from_file_location( + "claude_broker_tool_test", SCRIPTS / "claude_oauth_broker.py" + ) + assert spec and spec.loader + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def _payload(user: str = "Implement and test the fix") -> dict: + return { + "messages": [{"role": "user", "content": user}], + "tools": [ + {"name": "terminal", "description": "Run a command"}, + {"name": "todo", "description": "Track work"}, + ], + } + + +def test_prompt_declares_external_tools_available(monkeypatch) -> None: + module = _module(monkeypatch) + + prompt = module._prompt(_payload()) + + assert "available external Hermes tool" in prompt + assert "Never simulate a tool result" in prompt + + +def test_valid_external_tool_request_is_accepted(monkeypatch) -> None: + module = _module(monkeypatch) + + module._validate_structured( + { + "type": "tool_calls", + "text": "", + "tool_calls": [{"name": "terminal", "input": {"command": "git status"}}], + }, + _payload(), + ) + + +def test_fabricated_tool_unavailability_is_retryable(monkeypatch) -> None: + module = _module(monkeypatch) + + with pytest.raises(RuntimeError, match="fabricated"): + module._validate_structured( + { + "type": "final", + "text": "The terminal returned No such tool available.", + "tool_calls": [], + }, + _payload(), + ) + + +def test_user_can_ask_about_a_prior_tool_error(monkeypatch) -> None: + module = _module(monkeypatch) + + module._validate_structured( + { + "type": "final", + "text": "No such tool available means terminal was not exposed then.", + "tool_calls": [], + }, + _payload("Why did terminal report No such tool available?"), + ) + + +def test_unadvertised_tool_call_is_retryable(monkeypatch) -> None: + module = _module(monkeypatch) + + with pytest.raises(RuntimeError, match="unavailable tool call"): + module._validate_structured( + { + "type": "tool_calls", + "text": "", + "tool_calls": [{"name": "shell", "input": {}}], + }, + _payload(), + ) diff --git a/testing/tests/test_hermes_kanban_recovery.py b/testing/tests/test_hermes_kanban_recovery.py new file mode 100644 index 00000000..09dbe8e1 --- /dev/null +++ b/testing/tests/test_hermes_kanban_recovery.py @@ -0,0 +1,118 @@ +"""Focused tests for the fail-closed Cassandra Kanban recovery.""" + +from __future__ import annotations + +import importlib.util +import sqlite3 +from pathlib import Path + +import pytest + + +ROOT = Path(__file__).resolve().parents[2] +SCRIPT = ROOT / "services/hermes/scripts/repair_cassandra_kanban.py" +SPEC = importlib.util.spec_from_file_location("repair_cassandra_kanban", SCRIPT) +assert SPEC and SPEC.loader +recovery = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(recovery) + + +def _database(path: Path) -> Path: + connection = sqlite3.connect(path) + connection.executescript( + """ + CREATE TABLE tasks (id TEXT PRIMARY KEY, title TEXT NOT NULL); + CREATE TABLE task_links ( + parent_id TEXT NOT NULL, child_id TEXT NOT NULL, + PRIMARY KEY (parent_id, child_id) + ); + CREATE TABLE task_comments ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + task_id TEXT NOT NULL, + author TEXT NOT NULL, + body TEXT NOT NULL, + created_at INTEGER NOT NULL + ); + CREATE TABLE task_events ( + id INTEGER PRIMARY KEY AUTOINCREMENT, task_id TEXT NOT NULL + ); + CREATE TABLE task_runs ( + id INTEGER PRIMARY KEY AUTOINCREMENT, task_id TEXT NOT NULL + ); + CREATE TABLE task_attachments ( + id INTEGER PRIMARY KEY AUTOINCREMENT, task_id TEXT NOT NULL + ); + CREATE TABLE kanban_notify_subs ( + task_id TEXT NOT NULL, platform TEXT NOT NULL, + PRIMARY KEY (task_id, platform) + ); + CREATE INDEX idx_comments_task ON task_comments(task_id, created_at); + INSERT INTO tasks VALUES ('task-1', 'Preserve me'); + INSERT INTO task_comments(task_id, author, body, created_at) + VALUES ('task-1', 'worker', 'evidence', 1); + INSERT INTO task_events(task_id) VALUES ('task-1'); + INSERT INTO task_runs(task_id) VALUES ('task-1'); + INSERT INTO kanban_notify_subs VALUES ('task-1', 'telegram'); + """ + ) + connection.commit() + connection.close() + return path + + +def _known_errors(database: Path) -> list[str]: + with sqlite3.connect(database) as connection: + root = connection.execute( + "SELECT rootpage FROM sqlite_master WHERE name = 'task_comments'" + ).fetchone()[0] + return [ + "*** in database main ***\n" + f"Tree {root} page {root} cell 0: 2nd reference to page 41\n" + f"Tree {root} page 37 cell 4: Rowid 11 out of order", + "NUMERIC value in task_comments.author", + "row 2 missing from index idx_comments_task", + ] + + +def test_healthy_board_is_untouched(tmp_path: Path) -> None: + database = _database(tmp_path / "kanban.db") + before = database.read_bytes() + + result = recovery.repair_if_needed(database) + + assert result["state"] == "healthy" + assert database.read_bytes() == before + assert not list(tmp_path.glob("*.corrupt.*")) + + +def test_known_corruption_rebuilds_rows_and_retains_backup(tmp_path: Path) -> None: + database = _database(tmp_path / "kanban.db") + before = database.read_bytes() + + result = recovery.recover_database(database, _known_errors(database)) + + assert result["state"] == "recovered" + backup = Path(result["backup"]) + assert backup.read_bytes() == before + with sqlite3.connect(database) as connection: + assert connection.execute("PRAGMA integrity_check").fetchone() == ("ok",) + assert connection.execute("SELECT title FROM tasks").fetchone() == ( + "Preserve me", + ) + assert connection.execute( + "SELECT author, body FROM task_comments" + ).fetchone() == ( + "worker", + "evidence", + ) + + +def test_unknown_corruption_is_preserved_and_refused(tmp_path: Path) -> None: + database = _database(tmp_path / "kanban.db") + before = database.read_bytes() + + with pytest.raises(recovery.RecoveryRefused, match="does not match"): + recovery.recover_database(database, ["freelist leaf count is too big"]) + + assert database.read_bytes() == before + assert not list(tmp_path.glob("*.corrupt.*")) diff --git a/testing/tests/test_hermes_main_wrapper.py b/testing/tests/test_hermes_main_wrapper.py new file mode 100644 index 00000000..5165413c --- /dev/null +++ b/testing/tests/test_hermes_main_wrapper.py @@ -0,0 +1,56 @@ +"""Hermes gateway tool-environment patch and manifest tests.""" + +from __future__ import annotations + +import importlib.util +import stat +from pathlib import Path + +import yaml + + +ROOT = Path(__file__).resolve().parents[2] +HERMES = ROOT / "services/hermes" +SCRIPT = HERMES / "scripts/patch_main_wrapper.py" + + +def _patch_module(): + spec = importlib.util.spec_from_file_location("patch_main_wrapper", SCRIPT) + assert spec and spec.loader + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def _deployment() -> dict: + return yaml.safe_load((HERMES / "agent-deployment.yaml").read_text()) + + +def test_wrapper_loads_tool_environment_after_s6_restore(tmp_path: Path) -> None: + module = _patch_module() + source = tmp_path / "main-wrapper.sh" + destination = tmp_path / "patched.sh" + source.write_text("#!/bin/sh\n" + module.BEFORE + "export HOME=/opt/data\n") + source.chmod(0o755) + + module.patch(source, destination) + + patched = destination.read_text() + assert ". /opt/data/.env" in patched + assert patched.index(". /opt/data/.env") < patched.index("export HOME=/opt/data") + assert destination.stat().st_mode & stat.S_IXUSR + + +def test_gateway_retains_stock_entrypoint_with_patched_wrapper() -> None: + pod = _deployment()["spec"]["template"]["spec"] + init = {item["name"]: item for item in pod["initContainers"]} + containers = {item["name"]: item for item in pod["containers"]} + + assert "repair-cassandra-kanban" in init + assert "patch-main-wrapper" in init + assert containers["hermes"]["command"] == [ + "/init", + "/opt/hermes/docker/main-wrapper.sh", + ] + mounts = {item["name"]: item for item in containers["hermes"]["volumeMounts"]} + assert mounts["main-wrapper-patch"]["subPath"] == "main-wrapper.sh"