titan/atlas-iac
3
0
Fork 0
Code Issues Pull Requests 9 Actions Packages Projects Releases Wiki Activity

hermes: recover Kanban and tool execution

Browse Source
This commit is contained in:
jenkins 2026-08-15 03:35:26 -03:00
parent c21203b7ea
commit f7492defdd
8 changed files with 769 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

45
services/hermes/agent-deployment.yaml
View File

@ -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

2
services/hermes/kustomization.yaml
View File

@ -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:

105
services/hermes/scripts/claude_oauth_broker.py
View File

@ -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

48
services/hermes/scripts/patch_main_wrapper.py Normal file
View File

@ -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())

311
services/hermes/scripts/repair_cassandra_kanban.py Normal file
View File

@ -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())

101
testing/tests/test_hermes_claude_broker.py Normal file
View File

@ -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(),
)

118
testing/tests/test_hermes_kanban_recovery.py Normal file
View File

@ -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.*"))

56
testing/tests/test_hermes_main_wrapper.py Normal file
View File

@ -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"