diff --git a/services/hermes/agent-configmap.yaml b/services/hermes/agent-configmap.yaml index 53715c6b..19a540af 100644 --- a/services/hermes/agent-configmap.yaml +++ b/services/hermes/agent-configmap.yaml @@ -369,10 +369,14 @@ data: repository/pull-request evidence or to create a review-ready draft PR, load `$manage-atlas-pull-requests` and use `/opt/scm/gitea_api.py`. The client carries no repository credential; a separate least-authority - workload performs the allowed operation. It permits only Atlas-scoped - metadata reads and verified same-repository draft creation. Updates, merge, + workload performs the allowed operation. The generic client permits only + Atlas-scoped metadata reads and verified same-repository draft creation; it + cannot update an existing PR. Updates, merge, approve, close, delete, comments, repository administration, and force-push - are unavailable; never bypass the client with raw HTTP. Leave every PR + are unavailable to the + generic client. A coordinator-issued signed continuation worker may update + the recorded owned PR through the broker path below; never bypass the + client with raw HTTP. Leave every PR unmerged for Brad's review. Independently verify the exact base/head ancestry and diff, run the relevant tests, and confirm the candidate Jenkins result before presenting a handoff. Prefer diff --git a/services/hermes/agent-deployment.yaml b/services/hermes/agent-deployment.yaml index b36e05aa..37384f57 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: primary amd64 accelerator titan-22; arm64 rpi5 fleet fallback; storage-backbone nodes excluded - ai.bstein.dev/config-rev: "20260913-capability-effort-v3" + ai.bstein.dev/config-rev: "20260913-soteria-kanban-recovery-v1" prometheus.io/scrape: "true" prometheus.io/path: /metrics prometheus.io/port: "9010" @@ -287,6 +287,26 @@ spec: resources: requests: {cpu: 25m, memory: 32Mi} limits: {cpu: 100m, memory: 64Mi} + - name: recover-soteria-kanban + image: registry.bstein.dev/bstein/hermes-agent@sha256:81970563e542f0720773e72297810b3a844b83e381e278f25c0916c78d930107 + imagePullPolicy: IfNotPresent + command: + - /opt/hermes/.venv/bin/python + - /opt/coordinator/recover_soteria_kanban.py + securityContext: + allowPrivilegeEscalation: false + capabilities: + drop: [ALL] + 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} # Worker HUX identity provisioning. Mirrors the chat tenants' # init-hux-runtime, adapted to the single-replica Deployment: the # subtree name is the fixed literal "hux" (Deployment pod names churn, diff --git a/services/hermes/kustomization.yaml b/services/hermes/kustomization.yaml index 3e529b01..94e33911 100644 --- a/services/hermes/kustomization.yaml +++ b/services/hermes/kustomization.yaml @@ -4,7 +4,8 @@ kind: Kustomization namespace: hermes images: - name: registry.bstein.dev/bstein/hermes-agent - digest: sha256:a68d1c4d5517cc5e6719661e77be4f18d4964b07a85dcebaf6f62368646e4e6c # {"$imagepolicy": "hermes:hermes-agent-release:digest"} + # Hold this verified SQLite fix until the normal image release includes it. + digest: sha256:bebfe9c24cdd877ed2c16335027ccf393892161642756f54e8d12984eefdd903 resources: - namespace.yaml - image.yaml @@ -189,6 +190,7 @@ configMapGenerator: - 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 + - recover_soteria_kanban.py=scripts/recover_soteria_kanban.py - recover_cassandra_workers.py=scripts/recover_cassandra_workers.py - routing_catalog.py=scripts/routing_catalog.py - telegram_media_server.py=scripts/telegram_media_server.py diff --git a/services/hermes/scripts/kanban_supervisor.py b/services/hermes/scripts/kanban_supervisor.py index 210cda22..060ffa08 100644 --- a/services/hermes/scripts/kanban_supervisor.py +++ b/services/hermes/scripts/kanban_supervisor.py @@ -283,7 +283,7 @@ def _escalate(kanban_db: Any, conn: Any, task_id: str, reason: str) -> None: "no follow-up card was auto-created." ) try: - kanban_db.block_task(conn, task_id, reason=body, kind="supervisor") + kanban_db.block_task(conn, task_id, reason=body, kind="needs_input") except Exception as error: # noqa: BLE001 - still record the human-visible comment _log(f"could not block {task_id}: {error}") _comment(kanban_db, conn, task_id, body) diff --git a/services/hermes/scripts/model_capability_evaluator.py b/services/hermes/scripts/model_capability_evaluator.py index 04fb677c..7bcf0b1d 100644 --- a/services/hermes/scripts/model_capability_evaluator.py +++ b/services/hermes/scripts/model_capability_evaluator.py @@ -31,7 +31,7 @@ from model_evaluation_evidence import ( from provider_model_catalog import ( CAPABILITY_ROLES, EFFORTS, - REVIEWED_CAPABILITY_POLICY, + capability_role, is_eligible, proposed_candidate_role, supported_efforts, @@ -297,7 +297,7 @@ def _candidate_records(catalog: Mapping[str, Any]) -> list[tuple[str, str, Mappi details = metadata.get(model, {}) if isinstance(metadata.get(model), Mapping) else {} eligible, _ = is_eligible(provider, model, dict(details), None) if isinstance(model, str) else (False, "") role, provenance = proposed_candidate_role(provider, model, dict(details)) if isinstance(model, str) else (None, "") - reviewed = model in REVIEWED_CAPABILITY_POLICY.get(provider, {}) + reviewed, _ = capability_role(provider, model, dict(details)) if isinstance(model, str) else (None, "") if isinstance(model, str) and eligible and not reviewed and role in CAPABILITY_ROLES and provenance: candidates.append((provider, model, details, role)) return candidates diff --git a/services/hermes/scripts/model_evaluation_evidence.py b/services/hermes/scripts/model_evaluation_evidence.py index 5beb491c..f5a4b67c 100644 --- a/services/hermes/scripts/model_evaluation_evidence.py +++ b/services/hermes/scripts/model_evaluation_evidence.py @@ -13,6 +13,7 @@ from typing import Any, Iterable, Mapping SCHEMA_VERSION = 1 EVALUATION_VERSION = "capability-v1" RETRY_SECONDS = 6 * 60 * 60 +TRANSIENT_RETRY_SECONDS = 15 * 60 FAILURE_CLASSES = frozenset( { "auth", "network", "timeout", "rate_limited", "provider", @@ -101,6 +102,9 @@ def unavailable_record( """Represent transport trouble without treating it as model quality evidence.""" if failure_class not in FAILURE_CLASSES: failure_class = "provider" + # The steward starts alongside its loopback brokers. Retry a local startup + # race on the next hourly refresh, while keeping provider failures backoffed. + retry_seconds = TRANSIENT_RETRY_SECONDS if failure_class == "network" else RETRY_SECONDS return { "provider": provider, "model": model, @@ -112,7 +116,7 @@ def unavailable_record( "role_fit": "pending", "failure_class": failure_class, "last_attempt_at": now, - "retry_after": now + RETRY_SECONDS, + "retry_after": now + retry_seconds, "tokens": { "input": max(0, input_tokens), "output": max(0, output_tokens), diff --git a/services/hermes/scripts/provider_model_catalog.py b/services/hermes/scripts/provider_model_catalog.py index d089e076..7495130f 100644 --- a/services/hermes/scripts/provider_model_catalog.py +++ b/services/hermes/scripts/provider_model_catalog.py @@ -242,6 +242,14 @@ def capability_role( reviewed = REVIEWED_CAPABILITY_POLICY.get(provider, {}).get(model.strip().lower()) if reviewed: return reviewed, REVIEWED_POLICY_PROVENANCE + # Claude Code exposes executable aliases (for example ``opus[1m]``) and + # separately attests their concrete native model. Admit an alias only + # when that attested value is itself an exact reviewed model ID. + resolved = metadata.get("resolvedModel") if provider == "claude" else None + if isinstance(resolved, str): + reviewed = REVIEWED_CAPABILITY_POLICY["claude"].get(resolved.strip().lower()) + if reviewed: + return reviewed, "reviewed native Claude alias" evaluated = metadata.get("evaluated_capability_role") if isinstance(evaluated, str) and evaluated in CAPABILITY_ROLES: return evaluated, str(metadata.get("evaluation_provenance") or "verified evaluation") diff --git a/services/hermes/scripts/recover_soteria_kanban.py b/services/hermes/scripts/recover_soteria_kanban.py new file mode 100644 index 00000000..8c642b1c --- /dev/null +++ b/services/hermes/scripts/recover_soteria_kanban.py @@ -0,0 +1,200 @@ +#!/usr/bin/env python3 +"""Atomically install the reviewed Soteria Kanban recovery candidate once.""" +from __future__ import annotations + +import argparse +import hashlib +import json +import os +import shutil +import sqlite3 +import tempfile +import time +from pathlib import Path + + +SOURCE_SHA = "6ce2088512ae5bffe5a379aab00447086ab7cd74f4822b478702e55df1a9fb7d" +CANDIDATE_SHA = "a81c6dab96832578d9181db6b3935df5b7d1937569e5c59aa177bd19ead42e7e" +DEFAULT_DATABASE = Path("/opt/data/kanban/boards/soteria/kanban.db") +DEFAULT_CANDIDATE = Path("/opt/data/kanban/recovery/soteria/recovered.db") +COUNTS = {"tasks": 4, "task_events": 159, "task_comments": 8, "task_runs": 4, + "supervisor_roots": 3, "supervisor_children": 0} +ROOTS = {"t_f1593f8c", "t_c7c42600", "t_f4f726e1"} +REQUIRED_TABLES = { + "tasks", "task_events", "task_comments", "task_runs", "task_links", + "task_attachments", "kanban_notify_subs", "supervisor_roots", "supervisor_children", +} + + +class RecoveryError(RuntimeError): + """The candidate or current board did not match the reviewed recovery plan.""" + + +def _sha(path: Path) -> str: + """Hash one local artifact without accepting symlink indirection.""" + if path.is_symlink() or not path.is_file(): + raise RecoveryError("recovery artifact is unavailable") + digest = hashlib.sha256() + with path.open("rb") as source: + for block in iter(lambda: source.read(1024 * 1024), b""): + digest.update(block) + return digest.hexdigest() + + +def _integrity(path: Path) -> bool: + """Return whether SQLite can fully validate a database without writing it.""" + try: + connection = sqlite3.connect(f"file:{path}?mode=ro", uri=True) + try: + return [row[0] for row in connection.execute("PRAGMA integrity_check")] == ["ok"] + finally: + connection.close() + except sqlite3.Error: + return False + + +def _verify_candidate(path: Path) -> None: + """Require the exact forensic artifact and its reviewed data contract.""" + if _sha(path) != CANDIDATE_SHA or not _integrity(path): + raise RecoveryError("recovery candidate does not match reviewed evidence") + try: + connection = sqlite3.connect(f"file:{path}?mode=ro", uri=True) + try: + tables = {str(row[0]) for row in connection.execute( + "SELECT name FROM sqlite_master WHERE type='table'" + )} + if not REQUIRED_TABLES <= tables: + raise RecoveryError("recovery candidate is missing required tables") + counts = {table: int(connection.execute(f"SELECT count(*) FROM {table}").fetchone()[0]) + for table in COUNTS} + roots = {str(row[0]) for row in connection.execute( + "SELECT root_task_id FROM supervisor_roots" + )} + finally: + connection.close() + except sqlite3.Error as error: + raise RecoveryError("recovery candidate schema is unreadable") from error + if counts != COUNTS or roots != ROOTS: + raise RecoveryError("recovery candidate data contract differs") + + +def _copy(source: Path, destination: Path) -> None: + """Durably copy one evidence artifact into a new destination.""" + with source.open("rb") as reader, destination.open("xb") as writer: + shutil.copyfileobj(reader, writer) + writer.flush() + os.fsync(writer.fileno()) + + +def _marker(database: Path) -> Path: + return database.with_name(f"{database.name}.recovery-{CANDIDATE_SHA[:16]}.json") + + +def _write_marker(database: Path) -> None: + """Durably publish completion only after the replacement is in place.""" + marker = _marker(database) + descriptor, temporary = tempfile.mkstemp(prefix=f".{marker.name}.", dir=database.parent) + try: + with os.fdopen(descriptor, "w", encoding="utf-8") as handle: + json.dump({"candidate_sha": CANDIDATE_SHA, "source_sha": SOURCE_SHA}, handle, sort_keys=True) + handle.write("\n") + handle.flush() + os.fsync(handle.fileno()) + os.chmod(temporary, 0o600) + os.replace(temporary, marker) + directory = os.open(database.parent, os.O_RDONLY) + try: + os.fsync(directory) + finally: + os.close(directory) + except Exception: + Path(temporary).unlink(missing_ok=True) + raise + + +def _completed(database: Path) -> bool: + """Allow later restarts only after the recorded replacement remains healthy.""" + try: + value = json.loads(_marker(database).read_text()) + except (OSError, ValueError, TypeError): + return False + return value == {"candidate_sha": CANDIDATE_SHA, "source_sha": SOURCE_SHA} and _integrity(database) + + +def _has_nonempty_journal(database: Path) -> bool: + """Refuse a source with unapplied writes whose provenance was not reviewed.""" + for suffix in ("-wal", "-journal"): + sidecar = database.with_name(database.name + suffix) + if sidecar.exists() and sidecar.stat().st_size: + return True + return False + + +def _retained_source(database: Path) -> bool: + """Require the exact original evidence before finalizing an interrupted install.""" + pattern = f"{database.name}.corrupt.recovery-*-{SOURCE_SHA[:16]}.bak" + for backup in database.parent.glob(pattern): + try: + if _sha(backup) == SOURCE_SHA: + return True + except RecoveryError: + continue + return False + + +def recover(database: Path = DEFAULT_DATABASE, candidate: Path = DEFAULT_CANDIDATE) -> dict[str, str]: + """Replace only the exact quarantined source with the verified candidate.""" + if _marker(database).exists(): + if _completed(database): + return {"state": "already-recovered", "database": str(database)} + raise RecoveryError("completed recovery marker does not match healthy board") + if _has_nonempty_journal(database): + raise RecoveryError("current board has unreviewed journal writes") + current = _sha(database) + if current == CANDIDATE_SHA and _integrity(database): + if not _retained_source(database): + raise RecoveryError("interrupted recovery lacks retained source evidence") + _write_marker(database) + return {"state": "recovery-finalized", "database": str(database)} + _verify_candidate(candidate) + if current != SOURCE_SHA: + raise RecoveryError("current board does not match quarantined source") + stamp = f"{int(time.time())}-{SOURCE_SHA[:16]}" + backup = database.with_name(f"{database.name}.corrupt.recovery-{stamp}.bak") + _copy(database, backup) + for suffix in ("-wal", "-shm", "-journal"): + sidecar = database.with_name(database.name + suffix) + if sidecar.exists(): + os.replace(sidecar, backup.with_name(backup.name + suffix)) + descriptor, temporary = tempfile.mkstemp(prefix=f".{database.name}.recovery-", dir=database.parent) + os.close(descriptor) + replacement = Path(temporary) + replacement.unlink() + try: + _copy(candidate, replacement) + if not _integrity(replacement): + raise RecoveryError("copied candidate failed integrity verification") + os.chmod(replacement, database.stat().st_mode & 0o777) + os.replace(replacement, database) + except Exception: + replacement.unlink(missing_ok=True) + raise + _write_marker(database) + return {"state": "recovered", "database": str(database), "backup": str(backup)} + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--database", type=Path, default=DEFAULT_DATABASE) + parser.add_argument("--candidate", type=Path, default=DEFAULT_CANDIDATE) + args = parser.parse_args() + try: + print(json.dumps(recover(args.database, args.candidate), sort_keys=True)) + except (OSError, RecoveryError) as error: + print(f"Soteria Kanban recovery refused: {error}") + return 1 + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/services/hermes/scripts/supervisor_policy.py b/services/hermes/scripts/supervisor_policy.py index 7b5f4af5..2df13557 100644 --- a/services/hermes/scripts/supervisor_policy.py +++ b/services/hermes/scripts/supervisor_policy.py @@ -76,6 +76,15 @@ def supervised_kind(task: Any) -> str | None: return kind if kind in (REVIEW_KIND, REPAIR_KIND) else None def is_done(task: Any) -> bool: return str(field(task, "status", "") or "") in DONE_STATUSES +def is_lineage_only_anchor(task: Any) -> bool: + """Keep only trusted, unparented legacy roots out of automated chains.""" + meta = metadata(task) + root = meta.get("supervisor_lineage") + trusted = lineage.initial({"id": task_id(task), "metadata": {"supervisor_lineage": root}}) + return (isinstance(root, dict) and STAMP_KEY not in meta and trusted is not None + and trusted.root_task_id == task_id(task) + and field(task, "result") == "Lineage-only migration anchor." + and "Lineage-only anchor" in str(field(task, "body", "") or "")) def objective(task: Any) -> str: title = str(field(task, "title", "") or "") body = str(field(task, "body", "") or "") @@ -483,4 +492,6 @@ def plan(task: Any, tasks: list[Any], limits: Limits) -> Decision: return plan_review(task, tasks, limits) if kind == REPAIR_KIND: return plan_repair(task, tasks, limits) + if is_lineage_only_anchor(task): + return Decision("none", "lineage-only migration anchors require an explicit continuation") return plan_implementation(task, tasks, limits) diff --git a/services/hermes/scripts/supervisor_state.py b/services/hermes/scripts/supervisor_state.py index 0038edd0..fbd60978 100644 --- a/services/hermes/scripts/supervisor_state.py +++ b/services/hermes/scripts/supervisor_state.py @@ -4,15 +4,21 @@ from __future__ import annotations import hashlib +import fcntl import os import sqlite3 +import tempfile +from contextlib import contextmanager from pathlib import Path -from typing import Any +from typing import Any, Iterator from supervisor_lineage import Lineage -KANBAN_ROOT = Path("/opt/data/kanban/boards") +KANBAN_ROOT = Path( + os.environ.get("HERMES_KANBAN_HOME", os.environ.get("HERMES_HOME", "/opt/data")) +) / "kanban/boards" +STATE_FILE = "supervisor-state.db" class SupervisorStateError(ValueError): @@ -20,22 +26,16 @@ class SupervisorStateError(ValueError): def _path(board: str, path: Path | None) -> Path: - """Keep integrity rows beside the board's real Kanban transaction store.""" + """Keep integrity rows board-local without sharing Kanban's SQLite file.""" if path is not None: return path if not board or "/" in board or "\\" in board: raise ValueError("board name is invalid") - return KANBAN_ROOT / board / "kanban.db" + return KANBAN_ROOT / board / STATE_FILE -def _connect(board: str, path: Path | None = None) -> sqlite3.Connection: - path = _path(board, path) - if not path.exists() and path is None: - raise ValueError("Kanban board database does not exist") - path.parent.mkdir(mode=0o700, parents=True, exist_ok=True) - connection = sqlite3.connect(path) - connection.execute("PRAGMA journal_mode=WAL") - connection.execute("PRAGMA foreign_keys=ON") +def _schema(connection: sqlite3.Connection) -> None: + """Install the small coordinator schema in its isolated state database.""" connection.executescript( """ CREATE TABLE IF NOT EXISTS supervisor_roots ( @@ -53,22 +53,125 @@ def _connect(board: str, path: Path | None = None) -> sqlite3.Connection: ); """ ) - columns = {row[1] for row in connection.execute("PRAGMA table_info(supervisor_roots)")} - if "ready_for_human_merge" not in columns: - connection.execute( - "ALTER TABLE supervisor_roots ADD COLUMN ready_for_human_merge INTEGER NOT NULL DEFAULT 0" - ) - if "ready_commit" not in columns: - connection.execute("ALTER TABLE supervisor_roots ADD COLUMN ready_commit TEXT NOT NULL DEFAULT ''") - child_columns = {row[1] for row in connection.execute("PRAGMA table_info(supervisor_children)")} - if "cycle" not in child_columns: - connection.execute("ALTER TABLE supervisor_children ADD COLUMN cycle INTEGER NOT NULL DEFAULT 1") + + +def _legacy_rows(board: str, database: Path) -> tuple[list[tuple[Any, ...]], list[tuple[Any, ...]]]: + """Read a complete, healthy legacy state table set or refuse migration.""" + if not database.exists(): + return [], [] try: - if path.name == "state.db": + source = sqlite3.connect(f"file:{database}?mode=ro", uri=True) + try: + checked = [str(row[0]) for row in source.execute("PRAGMA integrity_check")] + if checked != ["ok"]: + raise SupervisorStateError("legacy supervisor state is unreadable") + names = {str(row[0]) for row in source.execute( + "SELECT name FROM sqlite_master WHERE type='table'" + )} + required = {"supervisor_roots", "supervisor_children"} + if not names & required: + return [], [] + if not required <= names: + raise SupervisorStateError("legacy supervisor state is malformed") + root_columns = {str(row[1]) for row in source.execute("PRAGMA table_info(supervisor_roots)")} + child_columns = {str(row[1]) for row in source.execute("PRAGMA table_info(supervisor_children)")} + root_base = {"board", "root_task_id", "project", "branch", "pull_request", "base_branch", "live_pr_head"} + child_base = {"board", "child_task_id", "root_task_id", "parent_task_id", "kind", "head_commit", "objective_digest"} + if not root_base <= root_columns or not child_base <= child_columns: + raise SupervisorStateError("legacy supervisor state schema is malformed") + ready = "ready_for_human_merge" if "ready_for_human_merge" in root_columns else "0" + commit = "ready_commit" if "ready_commit" in root_columns else "''" + cycle = "cycle" if "cycle" in child_columns else "1" + roots = source.execute( + "SELECT board,root_task_id,project,branch,pull_request,base_branch,live_pr_head," + f"{ready},{commit} FROM supervisor_roots" + ).fetchall() + children = source.execute( + "SELECT board,child_task_id,root_task_id,parent_task_id,kind,head_commit,objective_digest," + f"{cycle} FROM supervisor_children" + ).fetchall() + finally: + source.close() + except sqlite3.Error as error: + raise SupervisorStateError("legacy supervisor state is unreadable") from error + root_ids = {str(row[1]) for row in roots} + if (any(not row or row[0] != board for row in [*roots, *children]) + or any(str(row[2]) not in root_ids for row in children)): + raise SupervisorStateError("legacy supervisor state is malformed") + return roots, children + + +def _bootstrap_sidecar(board: str, state: Path) -> None: + """Atomically import valid legacy rows once, never creating an empty fallback.""" + lock = state.with_name(f".{STATE_FILE}.lock") + state.parent.mkdir(mode=0o700, parents=True, exist_ok=True) + with lock.open("a+") as handle: + fcntl.flock(handle.fileno(), fcntl.LOCK_EX) + if state.exists(): + return + roots, children = _legacy_rows(board, state.with_name("kanban.db")) + descriptor, temporary = tempfile.mkstemp(prefix=f".{STATE_FILE}.", dir=state.parent) + os.close(descriptor) + replacement = Path(temporary) + try: + connection = sqlite3.connect(replacement) + try: + _schema(connection) + if roots: + connection.executemany("INSERT INTO supervisor_roots VALUES(?,?,?,?,?,?,?,?,?)", roots) + connection.executemany("INSERT INTO supervisor_children VALUES(?,?,?,?,?,?,?,?)", children) + connection.commit() + finally: + connection.close() + descriptor = os.open(replacement, os.O_RDONLY) + try: + os.fsync(descriptor) + finally: + os.close(descriptor) + os.chmod(replacement, 0o600) + os.replace(replacement, state) + directory = os.open(state.parent, os.O_RDONLY) + try: + os.fsync(directory) + finally: + os.close(directory) + except Exception: + replacement.unlink(missing_ok=True) + raise + + +@contextmanager +def _connect(board: str, path: Path | None = None) -> Iterator[sqlite3.Connection]: + """Open one sidecar transaction and always close its SQLite handle.""" + default_path = path is None + path = _path(board, path) + if default_path and not path.exists(): + _bootstrap_sidecar(board, path) + path.parent.mkdir(mode=0o700, parents=True, exist_ok=True) + connection = sqlite3.connect(path, timeout=5) + try: + connection.execute("PRAGMA busy_timeout=5000") + connection.execute("PRAGMA foreign_keys=ON") + _schema(connection) + columns = {row[1] for row in connection.execute("PRAGMA table_info(supervisor_roots)")} + if "ready_for_human_merge" not in columns: + connection.execute( + "ALTER TABLE supervisor_roots ADD COLUMN ready_for_human_merge INTEGER NOT NULL DEFAULT 0" + ) + if "ready_commit" not in columns: + connection.execute("ALTER TABLE supervisor_roots ADD COLUMN ready_commit TEXT NOT NULL DEFAULT ''") + child_columns = {row[1] for row in connection.execute("PRAGMA table_info(supervisor_children)")} + if "cycle" not in child_columns: + connection.execute("ALTER TABLE supervisor_children ADD COLUMN cycle INTEGER NOT NULL DEFAULT 1") + if path.name == STATE_FILE: os.chmod(path, 0o600) - except OSError: - pass - return connection + yield connection + connection.commit() + except Exception: + connection.rollback() + raise + finally: + connection.close() def _lineage(row: tuple[Any, ...] | None) -> Lineage | None: diff --git a/testing/tests/test_hermes_agent_layout.py b/testing/tests/test_hermes_agent_layout.py index 4fe3edeb..9475fa90 100644 --- a/testing/tests/test_hermes_agent_layout.py +++ b/testing/tests/test_hermes_agent_layout.py @@ -403,12 +403,7 @@ def test_cli_lane_config_refresh_does_not_restart_active_work(): assert "startupProbe" not in lane assert "livenessProbe" not in lane assert lane["readinessProbe"] == { - "exec": { - "command": [ - "/opt/hermes/.venv/bin/python", - "/opt/coordinator/cli_lane_capabilities.py", - ] - }, + "httpGet": {"path": "/healthz", "port": "lane-metrics"}, "initialDelaySeconds": 2, "periodSeconds": 5, "timeoutSeconds": 5, diff --git a/testing/tests/test_hermes_chat_quality.py b/testing/tests/test_hermes_chat_quality.py index 0f4404fa..dfea1604 100644 --- a/testing/tests/test_hermes_chat_quality.py +++ b/testing/tests/test_hermes_chat_quality.py @@ -167,10 +167,15 @@ def test_sandbox_shares_only_the_tenant_workspace_without_credentials(): def test_gateway_image_honors_ui_model_and_caps_reasoning(): dockerfile = (ROOT / "dockerfiles" / "Dockerfile.hermes-agent").read_text() - assert ( - "FROM harbor-core.harbor.svc.cluster.local/mirror/hermes-agent@sha256:" - "9c841866021c54c4596849f6135717e8a4d52ba510b7f52c50aef1de1a283973" - ) in dockerfile + base_arg = next( + line for line in dockerfile.splitlines() if line.startswith("ARG HERMES_AGENT_BASE=") + ) + assert base_arg.startswith( + "ARG HERMES_AGENT_BASE=harbor-core.harbor.svc.cluster.local/mirror/hermes-agent@sha256:" + ) + assert len(base_arg.rsplit("@sha256:", 1)[1]) == 64 + assert "FROM ${HERMES_AGENT_BASE} AS sqlite-build" in dockerfile + assert "FROM sqlite-runtime" in dockerfile assert "_resolve_request_route" in dockerfile assert 'allowed_providers = {"atlas-switchyard"}' in dockerfile assert 'reasoning_effort=body.get("reasoning_effort")' in dockerfile diff --git a/testing/tests/test_hermes_coordinator.py b/testing/tests/test_hermes_coordinator.py index 65e340ee..39870d3d 100644 --- a/testing/tests/test_hermes_coordinator.py +++ b/testing/tests/test_hermes_coordinator.py @@ -297,6 +297,31 @@ def test_live_claude_effort_gap_uses_nearest_higher_declared_tier(): assert catalog["providers"]["claude"]["candidates"]["haiku"]["eligible"] is False +def test_live_claude_alias_requires_an_exact_reviewed_resolved_model(): + """Claude's native aliases inherit only an exact reviewed concrete ID.""" + metadata = { + "opus[1m]": { + "resolvedModel": "claude-opus-5[1m]", + "supportsEffort": True, + "supportedEffortLevels": ["low", "medium", "high", "xhigh"], + }, + "unknown": { + "resolvedModel": "claude-fable-6", + "supportsEffort": True, + "supportedEffortLevels": ["low", "medium", "high", "xhigh"], + }, + } + catalog = routing.build_routing_catalog( + routing.Catalog("openai-codex", [], True, True, "connected"), + routing.Catalog("anthropic", list(metadata), True, True, "connected", metadata), + ) + + claude = catalog["providers"]["claude"] + assert claude["capability_pools"]["advanced"] == ["opus[1m]"] + assert claude["capability_pools"]["frontier"] == [] + assert claude["candidates"]["unknown"]["tier"] is None + + def test_live_unavailable_capability_does_not_promote_to_a_higher_role(): """A fresh disabled economy model is an availability failure, not a tier gap.""" metadata = { diff --git a/testing/tests/test_hermes_image_builder.py b/testing/tests/test_hermes_image_builder.py index d5967f05..c2af55d2 100644 --- a/testing/tests/test_hermes_image_builder.py +++ b/testing/tests/test_hermes_image_builder.py @@ -61,6 +61,7 @@ def test_sqlite_runtime_keeps_hermes_required_extensions() -> None: """The WAL fix must not remove state and search capabilities from SQLite.""" source = AGENT_DOCKERFILE.read_text(encoding="utf-8") assert "./configure --prefix=/opt/sqlite --disable-static --all --scanstatus" in source + assert "/etc/ld.so.conf.d/00-hermes-sqlite.conf" in source for option in ( "SQLITE_ENABLE_COLUMN_METADATA", "SQLITE_ENABLE_PREUPDATE_HOOK", diff --git a/testing/tests/test_hermes_image_builder_vault.py b/testing/tests/test_hermes_image_builder_vault.py index 6dadf10c..7c993446 100644 --- a/testing/tests/test_hermes_image_builder_vault.py +++ b/testing/tests/test_hermes_image_builder_vault.py @@ -53,10 +53,13 @@ def test_flux_orders_seed_then_jenkins_then_hermes() -> None: ) jenkins = yaml.safe_load((app_root / "jenkins/kustomization.yaml").read_text()) hermes = yaml.safe_load((app_root / "hermes/kustomization.yaml").read_text()) + role_job = yaml.safe_load( + (REPO_ROOT / "services/vault/hermes-auth-role-bootstrap-job.yaml").read_text() + ) role_check = { "apiVersion": "batch/v1", "kind": "Job", - "name": "vault-k8s-auth-hermes-10", + "name": role_job["metadata"]["name"], "namespace": "vault", } seed_check = { diff --git a/testing/tests/test_hermes_kanban_continuation.py b/testing/tests/test_hermes_kanban_continuation.py index c30269f2..c890a4de 100644 --- a/testing/tests/test_hermes_kanban_continuation.py +++ b/testing/tests/test_hermes_kanban_continuation.py @@ -5,6 +5,7 @@ from __future__ import annotations from contextlib import nullcontext import json from pathlib import Path +import sqlite3 import sys import pytest @@ -74,7 +75,8 @@ def trusted_root(tmp_path: Path, monkeypatch): ) state.record_submission(board, "root", lineage, "a" * 40, path=path) state.set_ready(board, "root", "a" * 40, path=path) - return board, lineage, path + assert state.get_root(board, "root") == lineage + return board, lineage, state.KANBAN_ROOT / board / state.STATE_FILE def _pr(head: str) -> bytes: @@ -156,3 +158,79 @@ def test_present_malformed_continuation_state_fails_closed(trusted_root): ) with pytest.raises(state.SupervisorStateError, match="malformed"): state.get_child(board, "child", path=path) + + +def test_default_state_sidecar_imports_healthy_legacy_rows_once(tmp_path, monkeypatch): + """The first isolated state DB preserves native rows without sharing its file.""" + board = "titan-iac" + monkeypatch.setattr(state, "KANBAN_ROOT", tmp_path / "boards") + legacy = state.KANBAN_ROOT / board / "kanban.db" + lineage = state.Lineage("root", "feature/root", "https://scm.bstein.dev/titan/atlas-iac/pulls/9", "atlas-iac", "main") + state.record_submission(board, "root", lineage, "a" * 40, path=legacy) + state.set_ready(board, "root", "a" * 40, path=legacy) + state.record_child(board, "repair", "root", "root", "repair", "a" * 40, "repair", path=legacy) + + assert state.get_root(board, "root") == lineage + assert state.get_child(board, "repair")["parent_task_id"] == "root" + sidecar = state.KANBAN_ROOT / board / state.STATE_FILE + assert sidecar.is_file() and sidecar != legacy + state.record_submission(board, "root", lineage, "b" * 40) + assert state.get_live_head(board, "root") == "b" * 40 + assert state.get_live_head(board, "root", path=legacy) == "a" * 40 + + +def test_default_state_refuses_corrupt_legacy_without_empty_sidecar(tmp_path, monkeypatch): + """A damaged native DB cannot silently erase supervisor authority.""" + board = "soteria" + monkeypatch.setattr(state, "KANBAN_ROOT", tmp_path / "boards") + legacy = state.KANBAN_ROOT / board / "kanban.db" + legacy.parent.mkdir(parents=True) + legacy.write_bytes(b"not a sqlite database") + + with pytest.raises(state.SupervisorStateError, match="unreadable"): + state.get_root(board, "root") + assert not (state.KANBAN_ROOT / board / state.STATE_FILE).exists() + + +def test_default_state_imports_additive_legacy_schema_and_rejects_orphans(tmp_path, monkeypatch): + """Older valid rows keep their safe defaults; orphan children stop migration.""" + board = "atlas-iac" + monkeypatch.setattr(state, "KANBAN_ROOT", tmp_path / "boards") + legacy = state.KANBAN_ROOT / board / "kanban.db" + legacy.parent.mkdir(parents=True) + schema = """ + CREATE TABLE supervisor_roots ( + board TEXT, root_task_id TEXT, project TEXT, branch TEXT, pull_request TEXT, + base_branch TEXT, live_pr_head TEXT + ); + CREATE TABLE supervisor_children ( + board TEXT, child_task_id TEXT, root_task_id TEXT, parent_task_id TEXT, + kind TEXT, head_commit TEXT, objective_digest TEXT + ); + """ + with sqlite3.connect(legacy) as connection: + connection.executescript(schema) + connection.execute( + "INSERT INTO supervisor_roots VALUES(?,?,?,?,?,?,?)", + (board, "root", "atlas-iac", "feature/root", "pull/1", "main", "a" * 40), + ) + connection.execute( + "INSERT INTO supervisor_children VALUES(?,?,?,?,?,?,?)", + (board, "child", "root", "root", "repair", "a" * 40, "digest"), + ) + assert state.get_child(board, "child")["cycle"] == 1 + with state._connect(board) as connection: + assert connection.execute("SELECT ready_for_human_merge,ready_commit FROM supervisor_roots").fetchone() == (0, "") + + orphan_board = "soteria" + orphan = state.KANBAN_ROOT / orphan_board / "kanban.db" + orphan.parent.mkdir() + with sqlite3.connect(orphan) as connection: + connection.executescript(schema) + connection.execute( + "INSERT INTO supervisor_children VALUES(?,?,?,?,?,?,?)", + (orphan_board, "child", "missing", "missing", "repair", "a" * 40, "digest"), + ) + with pytest.raises(state.SupervisorStateError, match="malformed"): + state.get_root(orphan_board, "missing") + assert not (state.KANBAN_ROOT / orphan_board / state.STATE_FILE).exists() diff --git a/testing/tests/test_hermes_kanban_supervisor.py b/testing/tests/test_hermes_kanban_supervisor.py index aac1f1e5..63d20742 100644 --- a/testing/tests/test_hermes_kanban_supervisor.py +++ b/testing/tests/test_hermes_kanban_supervisor.py @@ -273,7 +273,37 @@ def test_escalation_blocks_card_and_creates_no_followup(): db = RecordingDb([impl]) supervisor.supervise_once(db, policy.Limits()) assert db.created == [] - assert db.blocked and db.blocked[0][0] == "impl" and db.blocked[0][2] == "supervisor" + assert db.blocked and db.blocked[0][0] == "impl" and db.blocked[0][2] == "needs_input" + + +def test_lineage_only_anchor_never_parses_or_blocks_legacy_pr_root(): + anchor = _task( + id="legacy-root", + title="Migration anchor: soteria PR 3 continuation", + body=("Lineage-only anchor for an existing open pull request. It records no " + "implementation or review outcome and does not authorize automatic follow-up work."), + result="Lineage-only migration anchor.", + metadata={"supervisor_lineage": {"root_task_id": "legacy-root", "project": "soteria", + "branch": "legacy/pr-3", "pull_request": "pull/3", "base_branch": "main"}}, + ) + db = RecordingDb([anchor]) + + assert supervisor.supervise_once(db, policy.Limits()) == 0 + assert db.created == [] and db.blocked == [] and db.comments == [] + + +def test_anchor_words_never_skip_a_supervised_review_contract(): + review = _task( + id="review", body="Lineage-only anchor", result="Lineage-only migration anchor.", + metadata={"supervisor": {"kind": "review", "root": "impl", "parent": "impl", + "head_commit": "c1", "cycle": 1, "root_task_id": "impl", + "project": "atlas-iac", "branch": "feature/impl", + "pull_request": "https://scm.bstein.dev/titan/atlas-iac/pulls/1", + "base_branch": "main"}}, + ) + db = RecordingDb([_trusted_root(), review]) + supervisor.supervise_once(db, policy.Limits()) + assert db.blocked and db.blocked[0][0] == "review" def test_spawn_retries_without_idempotency_key_on_typeerror(): diff --git a/testing/tests/test_hermes_soteria_kanban_recovery.py b/testing/tests/test_hermes_soteria_kanban_recovery.py new file mode 100644 index 00000000..4903051d --- /dev/null +++ b/testing/tests/test_hermes_soteria_kanban_recovery.py @@ -0,0 +1,101 @@ +"""Fail-closed installation tests for the reviewed Soteria recovery artifact.""" +from __future__ import annotations + +import importlib.util +import shutil +import sqlite3 +from pathlib import Path + +import pytest + + +ROOT = Path(__file__).parents[2] +SPEC = importlib.util.spec_from_file_location( + "recover_soteria_kanban", ROOT / "services/hermes/scripts/recover_soteria_kanban.py" +) +assert SPEC and SPEC.loader +recovery = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(recovery) + + +def _candidate(path: Path) -> Path: + """Build the reviewed-shape candidate without relying on production data.""" + with sqlite3.connect(path) as connection: + connection.executescript(""" + CREATE TABLE tasks (id TEXT); CREATE TABLE task_events (id INTEGER); + CREATE TABLE task_comments (id INTEGER); CREATE TABLE task_runs (id INTEGER); + CREATE TABLE task_links (id INTEGER); CREATE TABLE task_attachments (id INTEGER); + CREATE TABLE kanban_notify_subs (id INTEGER); CREATE TABLE supervisor_roots (root_task_id TEXT); + CREATE TABLE supervisor_children (id TEXT); + """) + connection.executemany("INSERT INTO tasks VALUES(?)", [(str(value),) for value in range(4)]) + connection.executemany("INSERT INTO task_events VALUES(?)", [(value,) for value in range(159)]) + connection.executemany("INSERT INTO task_comments VALUES(?)", [(value,) for value in range(8)]) + connection.executemany("INSERT INTO task_runs VALUES(?)", [(value,) for value in range(4)]) + connection.executemany("INSERT INTO supervisor_roots VALUES(?)", [(value,) for value in recovery.ROOTS]) + return path + + +def test_recovery_replaces_only_verified_source_and_is_restart_safe(tmp_path, monkeypatch): + candidate = _candidate(tmp_path / "candidate.db") + database = tmp_path / "kanban.db" + shutil.copyfile(candidate, database) + with sqlite3.connect(database) as connection: + connection.execute("CREATE TABLE forensic_source (value TEXT)") + monkeypatch.setattr(recovery, "CANDIDATE_SHA", recovery._sha(candidate)) + monkeypatch.setattr(recovery, "SOURCE_SHA", recovery._sha(database)) + + result = recovery.recover(database, candidate) + assert result["state"] == "recovered" + assert recovery._integrity(database) + assert list(tmp_path.glob("kanban.db.corrupt.recovery-*.bak")) + candidate.unlink() + assert recovery.recover(database, candidate)["state"] == "already-recovered" + + +def test_recovery_refuses_an_unexpected_live_database(tmp_path, monkeypatch): + candidate = _candidate(tmp_path / "candidate.db") + database = _candidate(tmp_path / "kanban.db") + with sqlite3.connect(database) as connection: + connection.execute("CREATE TABLE unexpected_live_state (value TEXT)") + monkeypatch.setattr(recovery, "CANDIDATE_SHA", recovery._sha(candidate)) + monkeypatch.setattr(recovery, "SOURCE_SHA", "0" * 64) + + with pytest.raises(recovery.RecoveryError, match="quarantined source"): + recovery.recover(database, candidate) + assert not list(tmp_path.glob("*.bak")) + + +def test_recovery_rejects_unreviewed_wal_before_replacement(tmp_path, monkeypatch): + candidate = _candidate(tmp_path / "candidate.db") + database = tmp_path / "kanban.db" + shutil.copyfile(candidate, database) + with sqlite3.connect(database) as connection: + connection.execute("CREATE TABLE forensic_source (value TEXT)") + monkeypatch.setattr(recovery, "CANDIDATE_SHA", recovery._sha(candidate)) + monkeypatch.setattr(recovery, "SOURCE_SHA", recovery._sha(database)) + database.with_name("kanban.db-wal").write_bytes(b"unreviewed writes") + + with pytest.raises(recovery.RecoveryError, match="unreviewed journal"): + recovery.recover(database, candidate) + assert not list(tmp_path.glob("*.bak")) + + +def test_recovery_finalizes_an_interrupted_atomic_replacement(tmp_path, monkeypatch): + candidate = _candidate(tmp_path / "candidate.db") + database = tmp_path / "kanban.db" + shutil.copyfile(candidate, database) + with sqlite3.connect(database) as connection: + connection.execute("CREATE TABLE forensic_source (value TEXT)") + monkeypatch.setattr(recovery, "CANDIDATE_SHA", recovery._sha(candidate)) + monkeypatch.setattr(recovery, "SOURCE_SHA", recovery._sha(database)) + original_marker = recovery._write_marker + monkeypatch.setattr(recovery, "_write_marker", lambda _database: (_ for _ in ()).throw(OSError("crash"))) + + with pytest.raises(OSError, match="crash"): + recovery.recover(database, candidate) + assert recovery._sha(database) == recovery.CANDIDATE_SHA + assert list(tmp_path.glob("kanban.db.corrupt.recovery-*.bak")) + monkeypatch.setattr(recovery, "_write_marker", original_marker) + assert recovery.recover(database, candidate)["state"] == "recovery-finalized" + assert recovery.recover(database, candidate)["state"] == "already-recovered" diff --git a/testing/tests/test_model_capability_evaluator.py b/testing/tests/test_model_capability_evaluator.py index 4e41a9a0..23c2acc3 100644 --- a/testing/tests/test_model_capability_evaluator.py +++ b/testing/tests/test_model_capability_evaluator.py @@ -14,7 +14,13 @@ sys.path.insert(0, str(SCRIPTS)) import model_capability_evaluator as evaluator # noqa: E402 from hermes_model_routing import Catalog, write_routing_catalog # noqa: E402 -from model_evaluation_evidence import accepted_outcomes, load_store # noqa: E402 +from model_evaluation_evidence import ( # noqa: E402 + TRANSIENT_RETRY_SECONDS, + accepted_outcomes, + load_store, + retry_due, + unavailable_record, +) from routing_catalog import load_catalog # noqa: E402 @@ -183,6 +189,23 @@ def test_refresh_is_capped_and_transport_failure_is_not_quality_failure(tmp_path assert retry.calls == 0 +def test_loopback_transport_failure_retries_before_the_next_hourly_refresh(): + """A broker startup race does not suppress a future model for six hours.""" + record = unavailable_record( + provider="claude", + model="unknown", + fingerprint="fingerprint", + role="advanced", + efforts=["high"], + failure_class="network", + now=100, + ) + + assert record["retry_after"] == 100 + TRANSIENT_RETRY_SECONDS + assert retry_due(record, 100 + TRANSIENT_RETRY_SECONDS - 1) is False + assert retry_due(record, 100 + TRANSIENT_RETRY_SECONDS) is True + + def test_only_verified_acceptance_outcomes_are_retained(tmp_path): path = tmp_path / "model-evaluations.json" result = evaluator.evaluate_catalog_candidates( @@ -249,6 +272,33 @@ def test_malformed_output_and_reviewed_policy_records_do_not_promote(tmp_path): assert "gpt-5.6-sol" not in result["evaluations"]["codex"] +def test_reviewed_native_claude_alias_does_not_spend_an_evaluation(tmp_path): + """A first-party alias resolving to reviewed Opus is already classified.""" + document = { + "providers": { + "claude": { + "live": True, + "models": ["opus[1m]"], + "model_metadata": { + "opus[1m]": { + "resolvedModel": "claude-opus-5[1m]", + "supportsEffort": True, + "supportedEffortLevels": ["high", "xhigh"], + } + }, + } + } + } + transport = Unavailable() + + result = evaluator.evaluate_catalog_candidates( + document, evidence_path=tmp_path / "evidence.json", transport=transport, now=100 + ) + + assert transport.calls == 0 + assert result["evaluations"]["claude"] == {} + + def test_native_broker_envelopes_keep_literal_models_and_account_usage(monkeypatch): class Response: def __init__(self, body):