Merge current PR 14 into distributed worker pool

# Conflicts:
#	testing/tests/test_hermes_chat_quality.py
#	testing/tests/test_hermes_cli_lanes.py

services/hermes/scm-common/scripts/scm_broker.py

@@ -23,7 +23,7 @@ from gitea_api import (
     read_token,
 )
 from gitea_api_policy import _reject_forbidden, _validate_ref, _validate_repo
-from scm_broker_io import spool_response
+from scm_broker_io import RejectRedirect, response_status as _status, spool_response
 from scm_broker_server import AbsoluteHeaderDeadlineMixin, BoundedThreadingHTTPServer
 
 BROKER_PORT = 9081
@@ -49,22 +49,9 @@ FEATURE_REF_RE = re.compile(
 )
 
 
-class RejectRedirect(urllib.request.HTTPRedirectHandler):
-    """Reject every upstream redirect before credentials can be forwarded."""
-    def redirect_request(self, req, fp, code, msg, headers, newurl):
-        raise PolicyError("SCM upstream redirects are not allowed")
-
-
 UPSTREAM_OPENER = urllib.request.build_opener(RejectRedirect())
 
 
-def _status(response: object) -> int | None:
-    value = getattr(response, "status", None)
-    if value is None and hasattr(response, "getcode"):
-        value = response.getcode()  # type: ignore[attr-defined]
-    return value
-
-
 def _read_bounded(
     stream,
     maximum: int,

services/hermes/scm-common/scripts/scm_broker_io.py

@@ -5,11 +5,27 @@ from __future__ import annotations
 
 import tempfile
 import time
+import urllib.request
 from typing import BinaryIO
 
 from gitea_api_policy import PolicyError
 
 
+class RejectRedirect(urllib.request.HTTPRedirectHandler):
+    """Reject every upstream redirect before credentials can be forwarded."""
+
+    def redirect_request(self, req, fp, code, msg, headers, newurl):
+        raise PolicyError("SCM upstream redirects are not allowed")
+
+
+def response_status(response: object) -> int | None:
+    """Read an HTTP status from either urllib response interface."""
+    value = getattr(response, "status", None)
+    if value is None and hasattr(response, "getcode"):
+        value = response.getcode()  # type: ignore[attr-defined]
+    return value
+
+
 def spool_response(
     response,
     maximum: int,
@@ -20,7 +36,9 @@ def spool_response(
     deadline_seconds: float,
 ) -> tuple[BinaryIO, int]:
     """Spool an upstream response to disk while scanning chunk boundaries."""
-    spool = tempfile.SpooledTemporaryFile(max_size=memory_limit, dir="/tmp")
+    spool = tempfile.SpooledTemporaryFile(  # noqa: SIM115 - caller owns returned spool
+        max_size=memory_limit, dir="/tmp"
+    )
     carry = b""
     total = 0
     deadline = time.monotonic() + deadline_seconds

services/hermes/scripts/node_account_hardening.py

@@ -11,6 +11,7 @@ import re
 import shlex
 import stat
 import struct
+from contextlib import suppress
 from pathlib import Path
 
 from node_account_audit import audit_membership, audit_privilege_policies
@@ -54,7 +55,9 @@ ACL_ENTRY = struct.Struct("<HHI")
 ACL_XATTR = "system.posix_acl_access"
 
 
-def _read_regular(path: Path, maximum: int = MAX_ACCOUNT_FILE) -> tuple[bytes, os.stat_result]:
+def _read_regular(
+    path: Path, maximum: int = MAX_ACCOUNT_FILE
+) -> tuple[bytes, os.stat_result]:
     snapshot = read_regular(path, maximum)
     return snapshot.value, path.stat(follow_symlinks=False)
 
@@ -162,7 +165,7 @@ def _reconcile_databases() -> None:
             records = _records(_read_regular(path)[0], fields, name)
             if expected[name] not in records:
                 raise HardeningError("dedicated Hermes account validation failed")
-    except Exception:
+    except Exception as error:
         for path, snapshot, updated in reversed(written):
             current = read_regular(path, MAX_ACCOUNT_FILE)
             if (
@@ -174,7 +177,7 @@ def _reconcile_databases() -> None:
             ):
                 raise HardeningError(
                     f"concurrent host account change prevents rollback: {path.name}"
-                )
+                ) from error
             atomic_write(path, snapshot.value, snapshot)
         raise
 
@@ -192,7 +195,9 @@ def _key_identity(line: bytes) -> tuple[str, bytes] | None:
         fields = shlex.split(stripped.decode("ascii"), posix=True)
     except (UnicodeDecodeError, ValueError) as exc:
         raise HardeningError("authorized key line is malformed") from exc
-    indexes = [index for index, field in enumerate(fields) if KEY_TYPE_RE.fullmatch(field)]
+    indexes = [
+        index for index, field in enumerate(fields) if KEY_TYPE_RE.fullmatch(field)
+    ]
     if len(indexes) != 1 or indexes[0] + 1 >= len(fields):
         raise HardeningError("authorized key line has ambiguous key material")
     key_type = fields[indexes[0]]
@@ -226,10 +231,8 @@ def _validated_public_key(path: Path) -> tuple[bytes, tuple[str, bytes]]:
 
 
 def _directory(path: Path, *, mode: int, uid: int, gid: int) -> None:
-    try:
+    with suppress(FileExistsError):
         path.mkdir(mode=mode)
-    except FileExistsError:
-        pass
     metadata = path.lstat()
     if not stat.S_ISDIR(metadata.st_mode) or stat.S_ISLNK(metadata.st_mode):
         raise HardeningError(f"unsafe account directory: {path.name}")

testing/quality_contract.json

@@ -20,6 +20,13 @@
     "ci/scripts/supply_chain_report.py",
     "services/mailu/scripts/mailu_sync.py",
     "services/mailu/scripts/mailu_sync_listener.py",
+    "services/gitea/scripts/gitea_branch_protection_check.py",
+    "services/hermes/scm-common/scripts/gitea_api.py",
+    "services/hermes/scm-common/scripts/gitea_api_policy.py",
+    "services/hermes/scm-common/scripts/scm_broker.py",
+    "services/hermes/scm-common/scripts/scm_broker_client.py",
+    "services/hermes/scm-common/scripts/scm_broker_io.py",
+    "services/hermes/scm-common/scripts/scm_broker_server.py",
     "services/hermes/scripts/execution_pool_client.py",
     "services/hermes/scripts/execution_pool_coordinator.py",
     "services/hermes/scripts/execution_pool_project.py",
@@ -27,6 +34,11 @@
     "services/hermes/scripts/execution_pool_scm.py",
     "services/hermes/scripts/execution_pool_server.py",
     "services/hermes/scripts/execution_pool_worker.py",
+    "services/hermes/scripts/hermes_coordinator.py",
+    "services/hermes/scripts/node_account_audit.py",
+    "services/hermes/scripts/node_account_hardening.py",
+    "services/hermes/scripts/node_account_io.py",
+    "services/hermes/scripts/stage_runtime_access.py",
     "testing/__init__.py",
     "testing/quality_contract.py",
     "testing/quality_docs.py",
@@ -50,6 +62,8 @@
     "services/comms/scripts/tests",
     "services/mailu/scripts/mailu_sync.py",
     "services/mailu/scripts/mailu_sync_listener.py",
+    "services/gitea/scripts/gitea_branch_protection_check.py",
+    "services/hermes/scm-common/scripts",
     "services/hermes/scripts/execution_pool_client.py",
     "services/hermes/scripts/execution_pool_coordinator.py",
     "services/hermes/scripts/execution_pool_project.py",
@@ -57,6 +71,11 @@
     "services/hermes/scripts/execution_pool_scm.py",
     "services/hermes/scripts/execution_pool_server.py",
     "services/hermes/scripts/execution_pool_worker.py",
+    "services/hermes/scripts/hermes_coordinator.py",
+    "services/hermes/scripts/node_account_audit.py",
+    "services/hermes/scripts/node_account_hardening.py",
+    "services/hermes/scripts/node_account_io.py",
+    "services/hermes/scripts/stage_runtime_access.py",
     "testing/tests",
     "testing"
   ],
@@ -72,6 +91,8 @@
       "coverage_sources": [
         "ci/scripts",
         "scripts.render.dashboards_render_atlas",
+        "services/gitea/scripts",
+        "services/hermes/scm-common/scripts",
         "services/hermes/scripts",
         "services/mailu/scripts",
         "testing"
@@ -139,7 +160,14 @@
       "ci/tests/**/*.py",
       "scripts/tests/**/*.py",
       "services/*/scripts/tests/**/*.py",
+      "services/gitea/scripts/gitea_branch_protection_check.py",
+      "services/hermes/scm-common/scripts/*.py",
       "services/hermes/scripts/execution_pool_*.py",
+      "services/hermes/scripts/hermes_coordinator.py",
+      "services/hermes/scripts/node_account_audit.py",
+      "services/hermes/scripts/node_account_hardening.py",
+      "services/hermes/scripts/node_account_io.py",
+      "services/hermes/scripts/stage_runtime_access.py",
       "services/mailu/scripts/mailu_sync.py",
       "services/mailu/scripts/mailu_sync_listener.py"
     ],
@@ -183,13 +211,20 @@
   },
   "coverage": {
     "minimum_percent": 95.0,
-    "tracked_files": [
+    "minimum_branch_percent": 95.0,
-      "ci/scripts/publish_test_metrics.py",
+    "branch_tracked_files": [
-      "ci/scripts/publish_test_metrics_quality.py",
+      "services/gitea/scripts/gitea_branch_protection_check.py",
-      "ci/scripts/semgrep_report.py",
+      "services/hermes/scm-common/scripts/gitea_api.py",
-      "ci/scripts/supply_chain_report.py",
+      "services/hermes/scm-common/scripts/gitea_api_policy.py",
-      "services/mailu/scripts/mailu_sync.py",
+      "services/hermes/scm-common/scripts/scm_broker.py",
-      "services/mailu/scripts/mailu_sync_listener.py",
+      "services/hermes/scm-common/scripts/scm_broker_client.py",
+      "services/hermes/scm-common/scripts/scm_broker_io.py",
+      "services/hermes/scm-common/scripts/scm_broker_server.py",
+      "services/hermes/scripts/hermes_coordinator.py",
+      "services/hermes/scripts/node_account_audit.py",
+      "services/hermes/scripts/node_account_hardening.py",
+      "services/hermes/scripts/node_account_io.py",
+      "services/hermes/scripts/stage_runtime_access.py",
       "services/hermes/scripts/execution_pool_client.py",
       "services/hermes/scripts/execution_pool_coordinator.py",
       "services/hermes/scripts/execution_pool_project.py",
@@ -197,6 +232,34 @@
       "services/hermes/scripts/execution_pool_scm.py",
       "services/hermes/scripts/execution_pool_server.py",
       "services/hermes/scripts/execution_pool_worker.py",
+      "testing/quality_coverage.py"
+    ],
+    "tracked_files": [
+      "ci/scripts/publish_test_metrics.py",
+      "ci/scripts/publish_test_metrics_quality.py",
+      "ci/scripts/semgrep_report.py",
+      "ci/scripts/supply_chain_report.py",
+      "services/mailu/scripts/mailu_sync.py",
+      "services/mailu/scripts/mailu_sync_listener.py",
+      "services/gitea/scripts/gitea_branch_protection_check.py",
+      "services/hermes/scm-common/scripts/gitea_api.py",
+      "services/hermes/scm-common/scripts/gitea_api_policy.py",
+      "services/hermes/scm-common/scripts/scm_broker.py",
+      "services/hermes/scm-common/scripts/scm_broker_client.py",
+      "services/hermes/scm-common/scripts/scm_broker_io.py",
+      "services/hermes/scm-common/scripts/scm_broker_server.py",
+      "services/hermes/scripts/execution_pool_client.py",
+      "services/hermes/scripts/execution_pool_coordinator.py",
+      "services/hermes/scripts/execution_pool_project.py",
+      "services/hermes/scripts/execution_pool_protocol.py",
+      "services/hermes/scripts/execution_pool_scm.py",
+      "services/hermes/scripts/execution_pool_server.py",
+      "services/hermes/scripts/execution_pool_worker.py",
+      "services/hermes/scripts/hermes_coordinator.py",
+      "services/hermes/scripts/node_account_audit.py",
+      "services/hermes/scripts/node_account_hardening.py",
+      "services/hermes/scripts/node_account_io.py",
+      "services/hermes/scripts/stage_runtime_access.py",
       "testing/quality_contract.py",
       "testing/quality_docs.py",
       "testing/quality_hygiene.py",

testing/quality_coverage.py

@@ -3,23 +3,31 @@
 from __future__ import annotations
 
 import xml.etree.ElementTree as ET
+from dataclasses import dataclass
 from pathlib import Path
 from typing import Any
 
 
-def _load_percentages(xml_path: Path, root: Path) -> dict[str, float]:
+@dataclass(frozen=True)
-    """Load per-file line-rate percentages from a Cobertura XML report."""
+class CoverageRates:
+    """Line and branch percentages reported for one source file."""
+
+    line: float
+    branch: float | None
+
+
+def _load_percentages(xml_path: Path, root: Path) -> dict[str, CoverageRates]:
+    """Load per-file line and branch percentages from a Cobertura report."""
     tree = ET.parse(xml_path)
     xml_root = tree.getroot()
     source_roots = [
-        Path(node.text)
+        Path(node.text) for node in xml_root.findall("./sources/source") if node.text
-        for node in xml_root.findall("./sources/source")
-        if node.text
     ]
-    percentages: dict[str, float] = {}
+    percentages: dict[str, CoverageRates] = {}
     for class_node in xml_root.findall(".//class"):
         filename = class_node.attrib.get("filename")
         line_rate = class_node.attrib.get("line-rate")
+        branch_rate = class_node.attrib.get("branch-rate")
         if not filename or line_rate is None:
             continue
         normalized = filename.replace("\\", "/")
@@ -32,7 +40,10 @@ def _load_percentages(xml_path: Path, root: Path) -> dict[str, float]:
                 if candidate.exists():
                     key = candidate.relative_to(root).as_posix()
                     break
-        percentages[key] = float(line_rate) * 100.0
+        percentages[key] = CoverageRates(
+            line=float(line_rate) * 100.0,
+            branch=float(branch_rate) * 100.0 if branch_rate is not None else None,
+        )
     return percentages
 
 
@@ -47,17 +58,39 @@ def run_check(contract: dict[str, Any], root: Path, xml_path: Path) -> list[str]
 
     percentages = _load_percentages(xml_path, root)
     minimum = float(contract.get("coverage", {}).get("minimum_percent", 95.0))
+    coverage_contract = contract.get("coverage", {})
+    branch_minimum = coverage_contract.get("minimum_branch_percent")
+    if branch_minimum is not None:
+        branch_minimum = float(branch_minimum)
+    branch_tracked = {
+        path.replace("\\", "/")
+        for path in coverage_contract.get(
+            "branch_tracked_files",
+            coverage_contract.get("tracked_files", [])
+            if branch_minimum is not None
+            else [],
+        )
+    }
     issues: list[str] = []
 
     for relative_path in contract.get("coverage", {}).get("tracked_files", []):
         normalized = relative_path.replace("\\", "/")
-        percent = percentages.get(normalized)
+        rates = percentages.get(normalized)
-        if percent is None:
+        if rates is None:
             issues.append(f"coverage missing for tracked file: {relative_path}")
             continue
-        if percent + 1e-9 < minimum:
+        if rates.line + 1e-9 < minimum:
             issues.append(
-                f"coverage below {minimum:.1f}%: {relative_path} ({percent:.1f}%)"
+                f"coverage below {minimum:.1f}%: {relative_path} ({rates.line:.1f}%)"
+            )
+        if branch_minimum is None or normalized not in branch_tracked:
+            continue
+        if rates.branch is None:
+            issues.append(f"branch coverage missing for tracked file: {relative_path}")
+        elif rates.branch + 1e-9 < branch_minimum:
+            issues.append(
+                "branch coverage below "
+                f"{branch_minimum:.1f}%: {relative_path} ({rates.branch:.1f}%)"
             )
 
     return issues
@@ -77,9 +110,9 @@ def compute_workspace_line_coverage(
     samples: list[float] = []
     for relative_path in contract.get("coverage", {}).get("tracked_files", []):
         normalized = relative_path.replace("\\", "/")
-        percent = percentages.get(normalized)
+        rates = percentages.get(normalized)
-        if percent is not None:
+        if rates is not None:
-            samples.append(percent)
+            samples.append(rates.line)
     if not samples:
         return 0.0
     return round(sum(samples) / len(samples), 3)

testing/tests/test_hermes_coordinator_cassandra.py

@@ -123,6 +123,9 @@ def test_cassandra_sync_repairs_origin_through_broker_without_token(
     tmp_path: Path, monkeypatch
 ):
     """Repository sync uses the credential-isolated broker remote."""
+    monkeypatch.setenv(
+        "HERMES_GITEA_TOKEN_FILE", str(tmp_path / "explicitly-unavailable-token")
+    )
     workspace = tmp_path / "cassandra"
     (workspace / ".git").mkdir(parents=True)
     monkeypatch.setattr(coordinator, "CASSANDRA_BASE_PATH", workspace)

testing/tests/test_hermes_coordinator_coverage.py

@@ -0,0 +1,267 @@
+"""Behavioral branch coverage for Hermes project coordination."""
+
+from __future__ import annotations
+
+import json
+import subprocess
+import sys
+import types
+from contextlib import nullcontext
+from pathlib import Path
+
+import pytest
+
+from testing.tests.test_hermes_coordinator_support import coordinator, routing
+
+
+def test_workspace_falls_back_when_unset_or_invalid(tmp_path: Path, monkeypatch):
+    monkeypatch.delenv("HERMES_CASSANDRA_ACTIVE_WORKTREE", raising=False)
+    assert coordinator.cassandra_workspace() == coordinator.CASSANDRA_BASE_PATH
+    monkeypatch.setenv("HERMES_CASSANDRA_ACTIVE_WORKTREE", str(tmp_path / "missing"))
+    assert coordinator.cassandra_workspace() == coordinator.CASSANDRA_BASE_PATH
+
+
+def test_task_migration_is_noop_for_base_workspace():
+    class Board:
+        @staticmethod
+        def connect_closing(**_kwargs):
+            raise AssertionError("base workspace must not access Kanban")
+
+    coordinator._migrate_open_cassandra_tasks(Board, coordinator.CASSANDRA_BASE_PATH)
+
+
+class _Project:
+    id = "project-id"
+
+
+def _fake_hermes_modules(monkeypatch, *, existing: bool, active: bool):
+    events = []
+    kb = types.ModuleType("hermes_cli.kanban_db")
+    kb.board_exists = lambda slug: existing
+    kb.create_board = lambda *args, **kwargs: events.append(
+        ("create-board", args, kwargs)
+    )
+    kb.set_current_board = lambda slug: events.append(("current", slug))
+    kb.connect_closing = lambda **kwargs: nullcontext(object())
+    kb.list_tasks = lambda *_a, **_k: []
+
+    pdb = types.ModuleType("hermes_cli.projects_db")
+    pdb.connect_closing = lambda: nullcontext(object())
+    pdb.get_project = lambda *_a: _Project() if existing else None
+    pdb.get_active_id = lambda *_a: "active" if active else None
+    pdb.create_project = (
+        lambda *args, **kwargs: events.append(("create-project", kwargs)) or "new-id"
+    )
+    pdb.set_active = lambda *args: events.append(("active", args[-1]))
+    pdb.update_project = lambda *args, **kwargs: events.append(("update", kwargs))
+    pdb.add_folder = lambda *args, **kwargs: events.append(("folder", args[-1], kwargs))
+
+    package = types.ModuleType("hermes_cli")
+    package.kanban_db = kb
+    package.projects_db = pdb
+    monkeypatch.setitem(sys.modules, "hermes_cli", package)
+    monkeypatch.setitem(sys.modules, "hermes_cli.kanban_db", kb)
+    monkeypatch.setitem(sys.modules, "hermes_cli.projects_db", pdb)
+    return events
+
+
+@pytest.mark.parametrize(
+    ("existing", "active", "expected"),
+    [
+        (False, False, {"current", "create-project", "active"}),
+        (False, True, {"current", "create-project"}),
+        (True, True, {"update", "folder"}),
+    ],
+)
+def test_bootstrap_cassandra_creates_or_updates_state(
+    tmp_path: Path, monkeypatch, existing, active, expected
+):
+    events = _fake_hermes_modules(monkeypatch, existing=existing, active=active)
+    base = tmp_path / "projects/cassandra"
+    worktree = tmp_path / "projects/cassandra-worktree"
+    worktree.mkdir(parents=True)
+    (worktree / ".git").write_text("gitdir: elsewhere\n", encoding="utf-8")
+    monkeypatch.setattr(coordinator, "CASSANDRA_BASE_PATH", base)
+    monkeypatch.setenv("HERMES_CASSANDRA_ACTIVE_WORKTREE", str(worktree))
+    coordinator.bootstrap_cassandra(tmp_path / "hermes")
+    names = {event[0] for event in events}
+    assert expected <= names
+    assert "create-board" in names
+
+
+def test_bootstrap_state_ready_path(tmp_path: Path, monkeypatch):
+    monkeypatch.setattr(coordinator, "bootstrap_cassandra", lambda _root: None)
+    assert coordinator.bootstrap_cassandra_state(tmp_path) == {"state": "ready"}
+
+
+def test_corruption_type_imports_pinned_exception(monkeypatch):
+    class Expected(Exception):
+        pass
+
+    package = types.ModuleType("hermes_cli")
+    kanban = types.ModuleType("hermes_cli.kanban_db")
+    kanban.KanbanDbCorruptError = Expected
+    package.kanban_db = kanban
+    monkeypatch.setitem(sys.modules, "hermes_cli", package)
+    monkeypatch.setitem(sys.modules, "hermes_cli.kanban_db", kanban)
+    assert coordinator._kanban_corruption_type() is Expected
+
+
+def test_repo_sync_reports_missing_git_and_unmanaged_directory(
+    tmp_path: Path, monkeypatch
+):
+    monkeypatch.setattr(coordinator.shutil, "which", lambda _name: None)
+    assert coordinator.sync_cassandra_repo({}) == "git-unavailable"
+
+    workspace = tmp_path / "cassandra"
+    workspace.mkdir()
+    (workspace / "user-file").write_text("preserve", encoding="utf-8")
+    monkeypatch.setattr(coordinator, "CASSANDRA_BASE_PATH", workspace)
+    monkeypatch.setattr(coordinator.shutil, "which", lambda _name: "/usr/bin/git")
+    assert coordinator.sync_cassandra_repo({}) == "unmanaged-nonempty-directory"
+
+
+@pytest.mark.parametrize(
+    ("failure", "expected"),
+    [
+        ("get-url-error", "remote-repair-failed"),
+        ("repair-exit", "remote-repair-failed-9"),
+        ("repair-error", "remote-repair-failed"),
+        ("fetch-exit", "sync-failed-8"),
+        ("fetch-error", "sync-failed"),
+    ],
+)
+def test_existing_repo_sync_reports_each_failure(
+    tmp_path: Path, monkeypatch, failure, expected
+):
+    workspace = tmp_path / "cassandra"
+    (workspace / ".git").mkdir(parents=True)
+    monkeypatch.setattr(coordinator, "CASSANDRA_BASE_PATH", workspace)
+    monkeypatch.setattr(coordinator.shutil, "which", lambda _name: "/usr/bin/git")
+
+    def run(command, **_kwargs):
+        action = command[-2:] if len(command) >= 2 else []
+        if action == ["get-url", "origin"]:
+            if failure == "get-url-error":
+                raise OSError("git unavailable")
+            return subprocess.CompletedProcess(command, 0, stdout="old\n")
+        if "set-url" in command:
+            if failure == "repair-error":
+                raise subprocess.TimeoutExpired(command, 30)
+            return subprocess.CompletedProcess(
+                command, 9 if failure == "repair-exit" else 0
+            )
+        if failure == "fetch-error":
+            raise OSError("fetch unavailable")
+        return subprocess.CompletedProcess(command, 8 if failure == "fetch-exit" else 0)
+
+    monkeypatch.setattr(coordinator.subprocess, "run", run)
+    assert coordinator.sync_cassandra_repo({}) == expected
+
+
+@pytest.mark.parametrize(
+    ("exit_code", "expected"), [(0, "ready"), (7, "sync-failed-7")]
+)
+def test_empty_repo_clones_through_broker(
+    tmp_path: Path, monkeypatch, exit_code, expected
+):
+    workspace = tmp_path / "cassandra"
+    monkeypatch.setattr(coordinator, "CASSANDRA_BASE_PATH", workspace)
+    monkeypatch.setattr(coordinator.shutil, "which", lambda _name: "/usr/bin/git")
+    seen = []
+
+    def run(command, **kwargs):
+        seen.append((command, kwargs["env"]))
+        return subprocess.CompletedProcess(command, exit_code)
+
+    monkeypatch.setattr(coordinator.subprocess, "run", run)
+    assert (
+        coordinator.sync_cassandra_repo(
+            {"API_SERVER_KEY": "not-forwarded", "SAFE": "yes"}
+        )
+        == expected
+    )
+    assert coordinator.CASSANDRA_REMOTE in seen[0][0]
+    assert seen[0][1]["SAFE"] == "yes"
+    assert "API_SERVER_KEY" not in seen[0][1]
+
+
+def _catalog(provider):
+    return routing.Catalog(provider, ["model"], True, True, "connected")
+
+
+def test_refresh_once_writes_complete_status(tmp_path: Path, monkeypatch):
+    monkeypatch.setattr(coordinator, "_read_env", lambda _path: {"SAFE": "yes"})
+    monkeypatch.setattr(coordinator, "discover_codex_models", lambda: _catalog("codex"))
+    monkeypatch.setattr(
+        coordinator, "discover_claude_models", lambda: _catalog("claude")
+    )
+    monkeypatch.setattr(
+        coordinator, "configure_routes", lambda *_a: {"coordinator": ["auto"]}
+    )
+    monkeypatch.setattr(
+        coordinator, "bootstrap_cassandra_state", lambda _root: {"state": "ready"}
+    )
+    monkeypatch.setattr(coordinator, "sync_cassandra_repo", lambda _env: "ready")
+    monkeypatch.setattr(
+        coordinator, "cassandra_workspace", lambda: tmp_path / "worktree"
+    )
+    written = []
+    monkeypatch.setattr(
+        coordinator, "_atomic_write", lambda path, value: written.append((path, value))
+    )
+    status = coordinator.refresh_once(tmp_path)
+    assert status["projects"]["cassandra"]["state"] == "ready"
+    assert json.loads(written[0][1])["providers"]["codex"]["connected"] is True
+
+
+def test_main_once_success_and_failure(monkeypatch, capsys):
+    status = {"providers": {"codex": {"state": "ready"}}}
+    monkeypatch.setattr(sys, "argv", ["coordinator", "--once"])
+    monkeypatch.setattr(coordinator, "refresh_once", lambda _root: status)
+    assert coordinator.main() == 0
+    assert "codex=ready" in capsys.readouterr().out
+    monkeypatch.setattr(
+        coordinator,
+        "refresh_once",
+        lambda _root: (_ for _ in ()).throw(RuntimeError("failed")),
+    )
+    assert coordinator.main() == 1
+    assert "RuntimeError" in capsys.readouterr().out
+
+
+@pytest.mark.parametrize(("interval", "expected"), [(1, 300), (400, 400)])
+def test_main_loop_sleeps_with_floor(monkeypatch, interval, expected):
+    calls = iter(({"providers": {}}, KeyboardInterrupt()))
+
+    def refresh(_root):
+        value = next(calls)
+        if isinstance(value, BaseException):
+            raise value
+        return value
+
+    sleeps = []
+    monkeypatch.setattr(coordinator, "refresh_once", refresh)
+    monkeypatch.setattr(coordinator.time, "sleep", sleeps.append)
+    monkeypatch.setattr(
+        sys, "argv", ["coordinator", "--loop", "--interval", str(interval)]
+    )
+    with pytest.raises(KeyboardInterrupt):
+        coordinator.main()
+    assert sleeps == [expected]
+
+
+def test_main_loop_recovers_from_refresh_failure(monkeypatch):
+    calls = iter((RuntimeError("temporary"), KeyboardInterrupt()))
+
+    def refresh(_root):
+        error = next(calls)
+        raise error
+
+    sleeps = []
+    monkeypatch.setattr(coordinator, "refresh_once", refresh)
+    monkeypatch.setattr(coordinator.time, "sleep", sleeps.append)
+    monkeypatch.setattr(sys, "argv", ["coordinator", "--loop", "--interval", "300"])
+    with pytest.raises(KeyboardInterrupt):
+        coordinator.main()
+    assert sleeps == [300]

testing/tests/test_hermes_gitea_branch_coverage.py

@@ -0,0 +1,183 @@
+"""Behavioral branch coverage for the Gitea protection policy checker."""
+
+from __future__ import annotations
+
+import json
+import os
+import stat
+import sys
+from pathlib import Path
+
+import pytest
+
+from testing.tests.test_hermes_scm_broker_support import ROOT, _load_path
+
+
+def _module(name: str = "branch_protection_coverage"):
+    return _load_path(
+        name,
+        ROOT / "services/gitea/scripts/gitea_branch_protection_check.py",
+    )
+
+
+def _rule(module, name: str = "main", **updates):
+    value = {
+        "rule_name": name,
+        "priority": 1,
+        "created_at": "2026-01-01T00:00:00Z",
+        **module._required("bstein"),
+    }
+    value.update(updates)
+    return value
+
+
+def test_read_bounded_accepts_regular_file_and_closes_descriptor(
+    tmp_path: Path, monkeypatch
+):
+    module = _module("branch_read_regular")
+    source = tmp_path / "rules.json"
+    source.write_bytes(b"[]")
+    closed: list[int] = []
+    real_close = os.close
+
+    def close(descriptor: int) -> None:
+        closed.append(descriptor)
+        real_close(descriptor)
+
+    monkeypatch.setattr(module.os, "close", close)
+    assert module._read_bounded(source) == b"[]"
+    assert len(closed) == 1
+
+
+@pytest.mark.parametrize("kind", ["directory", "oversized", "short"])
+def test_read_bounded_rejects_unsafe_or_changed_input(
+    tmp_path: Path, monkeypatch, kind
+):
+    module = _module(f"branch_read_{kind}")
+    source = tmp_path / "rules"
+    source.write_bytes(b"[]")
+    real_fstat = module.os.fstat
+    real_read = module.os.read
+
+    if kind == "directory":
+        monkeypatch.setattr(
+            module.os,
+            "fstat",
+            lambda descriptor: os.stat_result(
+                (stat.S_IFDIR | 0o700, 0, 0, 0, 0, 0, 2, 0, 0, 0)
+            ),
+        )
+    elif kind == "oversized":
+        monkeypatch.setattr(
+            module.os,
+            "fstat",
+            lambda descriptor: os.stat_result(
+                (stat.S_IFREG | 0o600, 0, 0, 0, 0, 0, module.MAX_INPUT + 1, 0, 0, 0)
+            ),
+        )
+    else:
+        monkeypatch.setattr(module.os, "fstat", real_fstat)
+        monkeypatch.setattr(module.os, "read", lambda descriptor, maximum: b"[")
+    with pytest.raises(module.PolicyError):
+        module._read_bounded(source)
+    monkeypatch.setattr(module.os, "read", real_read)
+
+
+@pytest.mark.parametrize(
+    ("pattern", "branch", "expected"),
+    [
+        ("m?in", "main", True),
+        ("m[ai]in", "main", True),
+        ("m[!z]in", "main", True),
+        ("m]ain", "m]ain", True),
+        ("m}ain", "m}ain", True),
+        ("m[a-z]in", "main", True),
+        ("m{ain,aster}", "main", True),
+        ("m\\ain", "main", True),
+    ],
+)
+def test_glob_grammar_accepts_supported_constructs(pattern, branch, expected):
+    module = _module("branch_glob_supported")
+    assert module._matches(pattern, branch) is expected
+
+
+@pytest.mark.parametrize(
+    "pattern",
+    [
+        "m\\",
+        "m[",
+        "m[]",
+        "m[!]",
+        "m[[a]",
+        "m[\\a]",
+        "m[-a]",
+        "m[a-]",
+        "m[a-b-c]",
+        "m[z-a]",
+        "m[aa-b]",
+        "m{ain}",
+        "m{ain,}",
+        "m{ain,aster",
+    ],
+)
+def test_invalid_globs_are_literal_nonmatches(pattern):
+    module = _module("branch_glob_invalid")
+    assert module._matches(pattern, "main") is False
+
+
+@pytest.mark.parametrize("pattern", ["", "x" * 256, "máin"])
+def test_rule_names_are_ascii_and_bounded(pattern):
+    module = _module("branch_name_bounds")
+    with pytest.raises(module.PolicyError, match="rule name"):
+        module._matches(pattern, "main")
+
+
+@pytest.mark.parametrize("value", [None, "x" * 65, "invalid", "2026-01-01T00:00:00"])
+def test_created_at_requires_short_timezone_aware_iso(value):
+    module = _module("branch_created_bounds")
+    with pytest.raises(module.PolicyError, match="creation time"):
+        module._created(value)
+
+
+@pytest.mark.parametrize(
+    "value",
+    [b"x" * (1024 * 1024 + 1), b"{}", json.dumps([None]).encode()],
+)
+def test_evaluate_rejects_invalid_top_level_inputs(value):
+    module = _module("branch_evaluate_top")
+    with pytest.raises((module.PolicyError, json.JSONDecodeError)):
+        module.evaluate(value, "main", "bstein")
+
+
+def test_evaluate_rejects_unknown_branch_and_excess_rules():
+    module = _module("branch_evaluate_bounds")
+    with pytest.raises(module.PolicyError, match="input"):
+        module.evaluate(b"[]", "release", "bstein")
+    with pytest.raises(module.PolicyError, match="response"):
+        module.evaluate(json.dumps([{}] * 101).encode(), "main", "bstein")
+
+
+@pytest.mark.parametrize("priority", [True, 0, 1_000_001, "1"])
+def test_evaluate_rejects_ambiguous_priority_types(priority):
+    module = _module("branch_priority_bounds")
+    rule = _rule(module, priority=priority)
+    with pytest.raises(module.PolicyError, match="priority"):
+        module.evaluate(json.dumps([rule]).encode(), "main", "bstein")
+
+
+def test_main_reports_success_and_failures(tmp_path: Path, monkeypatch, capsys):
+    module = _module("branch_main_paths")
+    source = tmp_path / "rules.json"
+    source.write_text(json.dumps([_rule(module)]), encoding="utf-8")
+    monkeypatch.setattr(sys, "argv", ["checker", str(source), "main", "bstein"])
+    assert module.main() == 0
+    assert capsys.readouterr().out.strip() == "PRESENT"
+
+    source.write_text("not-json", encoding="utf-8")
+    assert module.main() == 1
+    assert "branch protection check failed" in capsys.readouterr().err
+
+    monkeypatch.setattr(
+        module, "_read_bounded", lambda _path: (_ for _ in ()).throw(OSError())
+    )
+    assert module.main() == 1

testing/tests/test_hermes_gitea_internal_coverage.py

@@ -0,0 +1,375 @@
+"""Behavioral coverage for internal safe-Gitea client and policy paths."""
+
+from __future__ import annotations
+
+import json
+import subprocess
+import sys
+import types
+import urllib.error
+import urllib.parse
+from email.message import Message
+from pathlib import Path
+
+import pytest
+
+from testing.tests.test_hermes_gitea_support import CLIENT_PATH, Response, _load
+from testing.tests.test_hermes_scm_broker_support import _load_path
+
+
+def _policy(name: str = "gitea_policy_coverage"):
+    return _load_path(name, CLIENT_PATH.parent / "gitea_api_policy.py")
+
+
+def test_runtime_token_reader_and_fixed_origin(tmp_path: Path, monkeypatch):
+    module = _load()
+    token = tmp_path / "token"
+    token.write_text("  synthetic-runtime-value  \n", encoding="utf-8")
+    assert module.read_token(token) == "synthetic-runtime-value"
+    token.write_text(" \n", encoding="utf-8")
+    with pytest.raises(ValueError, match="empty"):
+        module.read_token(token)
+
+    monkeypatch.setenv("GITEA_BASE_URL", module.CANONICAL_BASE_URL)
+    assert module.configured_base_url() == module.CANONICAL_BASE_URL
+    monkeypatch.setenv("GITEA_BASE_URL", "https://example.invalid")
+    with pytest.raises(module.PolicyError, match="fixed"):
+        module.configured_base_url()
+
+
+def test_safe_urlopen_delegates_only_to_redirect_rejecting_opener(monkeypatch):
+    module = _load()
+    sentinel = object()
+    monkeypatch.setattr(
+        module._SAFE_OPENER,
+        "open",
+        lambda request, timeout: (request, timeout, sentinel),
+    )
+    request = object()
+    assert module._safe_urlopen(request, 7) == (request, 7, sentinel)
+
+
+def test_api_target_and_request_body_edge_paths(monkeypatch):
+    module = _load()
+    with pytest.raises(module.PolicyError, match="path exceeds"):
+        module._split_api_path("/api/v1/" + "x" * 510)
+    with pytest.raises(module.PolicyError, match="request body"):
+        module.authorize_request(
+            "GET", "/api/v1/repos/atlas/cassandra", {"unexpected": True}
+        )
+    with pytest.raises(module.PolicyError, match="query"):
+        module.authorize_request(
+            "POST",
+            "/api/v1/repos/atlas/cassandra/pulls?page=1",
+            {},
+        )
+    with pytest.raises(module.PolicyError, match="only read"):
+        module.authorize_request("TRACE", "/api/v1/repos/atlas/cassandra/pulls", None)
+    with pytest.raises(module.PolicyError, match="accepts only"):
+        module.authorize_request("POST", "/api/v1/repos/atlas/cassandra/pulls", [])
+
+
+def test_response_status_fallback_and_nested_fail_closed():
+    module = _load()
+
+    class GetCodeResponse(Response):
+        def __init__(self):
+            super().__init__({"name": "cassandra"}, status=200)
+            del self.status
+
+        def getcode(self):
+            return 200
+
+    assert (
+        module.read(
+            "/api/v1/repos/atlas/cassandra",
+            token="synthetic",
+            opener=lambda *_a, **_k: GetCodeResponse(),
+        )
+        == b'{"name": "cassandra"}'
+    )
+    with pytest.raises(module.PolicyError, match="omitted required"):
+        module._nested({"base": None}, "base", "ref")
+
+
+@pytest.mark.parametrize("body", [b"not-json", b"[]"])
+def test_create_response_requires_json_object(body: bytes):
+    module = _load()
+    with pytest.raises(module.PolicyError, match="invalid pull-request metadata"):
+        module._require_create_response(
+            body,
+            repo="cassandra",
+            base="main",
+            head="feature/test",
+            head_sha="a" * 40,
+            title="WIP: Test",
+            body="Evidence",
+        )
+
+
+def test_write_body_handles_empty_newline_and_missing_newline(monkeypatch):
+    module = _load()
+
+    class Buffer:
+        value = bytearray()
+
+        @classmethod
+        def write(cls, value):
+            cls.value.extend(value)
+
+    monkeypatch.setattr(module.sys, "stdout", types.SimpleNamespace(buffer=Buffer))
+    module._write_body(b"")
+    module._write_body(b"one\n")
+    module._write_body(b"two")
+    assert bytes(Buffer.value) == b"one\ntwo\n"
+
+
+def test_main_executes_broker_read_and_create_paths(monkeypatch):
+    module = _load()
+    outputs: list[bytes] = []
+    client = types.ModuleType("scm_broker_client")
+    client.read = lambda path: json.dumps({"path": path}).encode()
+    client.create_draft = lambda repo, **data: json.dumps(
+        {"repo": repo, **data}, sort_keys=True
+    ).encode()
+    monkeypatch.setitem(sys.modules, "scm_broker_client", client)
+    monkeypatch.setattr(module, "_write_body", outputs.append)
+
+    assert module.main(["read", "/api/v1/repos/atlas/cassandra"]) == 0
+    assert json.loads(outputs.pop()) == {"path": "/api/v1/repos/atlas/cassandra"}
+    assert (
+        module.main(
+            [
+                "create-draft",
+                "cassandra",
+                "--base",
+                "main",
+                "--head",
+                "feature/coverage",
+                "--head-sha",
+                "a" * 40,
+                "--title",
+                "Coverage repair",
+                "--body",
+                "Review the focused tests",
+            ]
+        )
+        == 0
+    )
+    assert json.loads(outputs.pop())["repo"] == "cassandra"
+
+
+def test_main_handles_http_and_policy_failures(monkeypatch, capsys):
+    module = _load()
+    client = types.ModuleType("scm_broker_client")
+
+    def http_failure(_path):
+        headers = Message()
+        raise urllib.error.HTTPError("url", 503, "unavailable", headers, None)
+
+    client.read = http_failure
+    client.create_draft = lambda *_a, **_k: b"{}"
+    monkeypatch.setitem(sys.modules, "scm_broker_client", client)
+    assert module.main(["read", "/api/v1/repos/atlas/cassandra"]) == 1
+    assert "HTTP 503" in capsys.readouterr().err
+    client.read = lambda _path: (_ for _ in ()).throw(module.PolicyError("rejected"))
+    assert module.main(["read", "/not-allowed"]) == 1
+    assert "no credential" in capsys.readouterr().err
+
+
+@pytest.mark.parametrize(
+    ("function", "value"),
+    [
+        ("_validate_repo", "."),
+        ("_validate_repo", ".."),
+        ("_validate_sha", "short"),
+        ("_validate_pr_number", True),
+        ("_validate_pr_number", 0),
+        ("_validate_pr_number", 2_147_483_648),
+    ],
+)
+def test_policy_scalar_validators_reject_ambiguous_values(function, value):
+    module = _policy("gitea_policy_scalar")
+    with pytest.raises(module.PolicyError):
+        getattr(module, function)(value)
+
+
+def test_ref_bounds_cover_utf8_controls_git_failure_and_success(monkeypatch):
+    module = _policy("gitea_policy_refs")
+    with pytest.raises(module.PolicyError, match="valid UTF-8"):
+        module._validate_ref_bounds("\ud800", "head")
+    with pytest.raises(module.PolicyError, match="UTF-8"):
+        module._validate_ref_bounds("🧪" * 64, "head")
+    with pytest.raises(module.PolicyError, match="safe same"):
+        module._validate_ref("bad\nref", "head")
+    monkeypatch.setattr(
+        module.subprocess,
+        "run",
+        lambda *_a, **_k: subprocess.CompletedProcess([], 1),
+    )
+    with pytest.raises(module.PolicyError, match="same-repository"):
+        module._validate_ref("invalid-ref", "head")
+    monkeypatch.setattr(
+        module.subprocess,
+        "run",
+        lambda *_a, **_k: subprocess.CompletedProcess([], 0),
+    )
+    assert module._validate_ref("feature/valid", "head") == "feature/valid"
+
+
+@pytest.mark.parametrize(
+    ("value", "required", "match"),
+    [
+        (None, False, "must be text"),
+        ("  ", True, "must not be empty"),
+        ("\ud800", False, "valid UTF-8"),
+        ("x\x00y", False, "safe request limit"),
+    ],
+)
+def test_text_validation_rejects_nontext_empty_invalid_and_controls(
+    value, required, match
+):
+    module = _policy("gitea_policy_text")
+    with pytest.raises(module.PolicyError, match=match):
+        module._validate_text(value, "field", 20, 20, required=required)
+
+
+@pytest.mark.parametrize(
+    ("key", "expected"),
+    [
+        ("password", True),
+        ("accountKey", True),
+        ("registry-key", True),
+        ("clientEmail", True),
+        ("clientId", True),
+        ("accessId", True),
+        ("connectionString", True),
+        ("dockerConfigJson", True),
+        ("release_note", False),
+    ],
+)
+def test_sensitive_key_semantics_cover_generic_forms(key, expected):
+    module = _policy("gitea_policy_keys")
+    assert module._is_sensitive_key(key) is expected
+
+
+@pytest.mark.parametrize(
+    ("value", "expected"),
+    [
+        ("", False),
+        ('"synthetic"', True),
+        ("{encoded}", True),
+        ("Bearer value-12345678", True),
+        ("ghp_" + "a" * 24, True),
+        ("https://example.invalid/value", True),
+        ("name@example.invalid", True),
+        ("${RUNTIME_VALUE}", True),
+        ("AbCdEf0123456789", True),
+        ("reject empty values", False),
+        ("ordinary", True),
+    ],
+)
+def test_assignment_value_shape_distinguishes_prose_from_credentials(value, expected):
+    module = _policy("gitea_policy_assignment_values")
+    assert module._looks_sensitive_assignment_value(value) is expected
+
+
+@pytest.mark.parametrize(
+    ("value", "expected"),
+    [
+        ("single", False),
+        ("add regression coverage", True),
+        ("add regression_coverage", False),
+        ("ordinary engineering prose", False),
+    ],
+)
+def test_prose_classifier_covers_length_punctuation_and_lead_words(value, expected):
+    module = _policy("gitea_policy_prose")
+    assert module._looks_like_prose(value) is expected
+
+
+def test_structured_json_walk_covers_lists_nonstring_keys_and_limits():
+    module = _policy("gitea_policy_json_walk")
+    assert module._json_value_has_sensitive_assignment({1: "ignored"}) is False
+    assert module._json_value_has_sensitive_assignment([{"release": "safe"}]) is False
+    assert (
+        module._json_value_has_sensitive_assignment({"client_secret": "value"}) is True
+    )
+    assert (
+        module._json_value_has_sensitive_assignment(
+            {"outer": {"client_secret": "value"}}
+        )
+        is True
+    )
+    assert (
+        module._json_value_has_sensitive_assignment({"type": "service-account"}) is True
+    )
+    with pytest.raises(module.PolicyError, match="scan limit"):
+        module._json_value_has_sensitive_assignment([], depth=33)
+    with pytest.raises(module.PolicyError, match="scan limit"):
+        module._json_value_has_sensitive_assignment([], nodes=[2048])
+
+
+def test_json_key_decoder_and_embedded_document_scanner_cover_failures():
+    module = _policy("gitea_policy_json_decoder")
+    assert module._decode_json_key("client\\u005fsecret") == ("client_secret", True)
+    decoded, valid = module._decode_json_key("client\\qsecret")
+    assert valid is False and "client" in decoded
+    assert list(module._decoded_json_documents("prose only")) == []
+    assert list(module._decoded_json_documents("x {bad y [1, 2]")) == [[1, 2]]
+    with pytest.raises(module.PolicyError, match="scan limit"):
+        list(module._decoded_json_documents("{" * 33))
+    assert (
+        module._has_structured_sensitive_assignment('"client\\qsecret": value') is True
+    )
+    assert (
+        module._has_structured_sensitive_assignment("type:\n  service-account") is True
+    )
+
+
+def test_high_entropy_detector_covers_long_mixed_and_low_entropy_values():
+    module = _policy("gitea_policy_entropy")
+    assert module._has_high_entropy_token("A1_" * 100) is True
+    assert (
+        module._has_high_entropy_token(
+            "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789+/_="
+        )
+        is True
+    )
+    assert module._has_high_entropy_token("a" * 64) is False
+    assert module._has_high_entropy_token("short prose") is False
+
+
+@pytest.mark.parametrize(
+    "query",
+    [
+        "bad",
+        "a=1&b=2&c=3&d=4",
+        "page=1&page=2",
+        "=value",
+        "unexpected=1",
+        "page=abc",
+        "page=01",
+        "page=10001",
+        "limit=51",
+        "state=merged",
+    ],
+)
+def test_query_validator_covers_each_rejection_class(query):
+    module = _policy("gitea_policy_query")
+    target = urllib.parse.urlsplit("/api/v1/repos/atlas/cassandra/pulls?" + query)
+    with pytest.raises(module.PolicyError):
+        module._validate_query(target, {"page", "limit", "state"})
+
+
+def test_draft_title_rejects_prefix_without_content():
+    module = _policy("gitea_policy_empty_draft")
+    with pytest.raises(module.PolicyError, match="after the draft prefix"):
+        module._draft_title("WIP:")
+
+
+def test_query_validator_rejects_noncanonical_raw_form():
+    module = _policy("gitea_policy_raw_query")
+    target = urllib.parse.SplitResult("", "", "/api/v1/repos/atlas/cassandra", "%", "")
+    with pytest.raises(module.PolicyError, match="canonical ASCII"):
+        module._validate_query(target, set())

testing/tests/test_hermes_node_acl_coverage.py

@@ -0,0 +1,206 @@
+"""Behavioral branch coverage for node-storage ACL hardening."""
+
+from __future__ import annotations
+
+import errno
+import os
+import pytest
+
+from testing.tests.test_hermes_node_account_support import _fixture, _load
+
+
+def test_acl_decoder_rejects_version_permissions_and_adds_missing_mask():
+    module = _load()
+    valid = [
+        (module.ACL_USER_OBJ, 7, module.ACL_UNDEFINED_ID),
+        (module.ACL_GROUP_OBJ, 5, module.ACL_UNDEFINED_ID),
+        (module.ACL_OTHER, 5, module.ACL_UNDEFINED_ID),
+    ]
+    wrong_version = module.ACL_HEADER.pack(99) + b"".join(
+        module.ACL_ENTRY.pack(*entry) for entry in valid
+    )
+    with pytest.raises(module.HardeningError, match="version"):
+        module._decode_acl(wrong_version, 0o755)
+    bad_permission = module._encode_acl([*valid, (module.ACL_USER, 8, 1200)])
+    with pytest.raises(module.HardeningError, match="permission"):
+        module._decode_acl(bad_permission, 0o755)
+    updated = module._decode_acl(
+        module._acl_with_deny(module._encode_acl(valid), 0o755), 0o755
+    )
+    assert (module.ACL_MASK, 5, module.ACL_UNDEFINED_ID) in updated
+    assert (module.ACL_USER, 0, module.ACCOUNT_UID) in updated
+
+
+def test_read_acl_handles_absence_and_propagates_other_errors(monkeypatch, tmp_path):
+    module = _load()
+    path = tmp_path / "root"
+    path.mkdir()
+
+    def missing(*_args, **_kwargs):
+        raise OSError(errno.ENODATA, "missing")
+
+    monkeypatch.setattr(module.os, "getxattr", missing)
+    assert module._read_acl(path) == b""
+
+    def denied(*_args, **_kwargs):
+        raise OSError(errno.EPERM, "denied")
+
+    monkeypatch.setattr(module.os, "getxattr", denied)
+    with pytest.raises(OSError):
+        module._read_acl(path)
+
+
+@pytest.mark.parametrize("value", [b"", b"X", b"A", b"Nextra"])
+def test_acl_backup_validator_rejects_malformed_encodings(value):
+    module = _load()
+    with pytest.raises(module.HardeningError, match="backup is malformed"):
+        module._validate_acl_backup(value)
+    module._validate_acl_backup(b"N")
+    valid = module._encode_acl(
+        [
+            (module.ACL_USER_OBJ, 7, module.ACL_UNDEFINED_ID),
+            (module.ACL_GROUP_OBJ, 5, module.ACL_UNDEFINED_ID),
+            (module.ACL_OTHER, 5, module.ACL_UNDEFINED_ID),
+        ]
+    )
+    module._validate_acl_backup(b"A" + valid)
+
+
+def test_acl_backup_reuses_valid_existing_copy(tmp_path, monkeypatch):
+    module = _load()
+    host_etc = tmp_path / "etc"
+    root = host_etc / "hermes-node-boundary"
+    root.mkdir(parents=True)
+    backup = root / "k3s.acl"
+    backup.write_bytes(b"N")
+    monkeypatch.setattr(module, "HOST_ETC", host_etc)
+    monkeypatch.setattr(module, "HOST_ROOT_UID", os.getuid())
+    monkeypatch.setattr(module, "HOST_ROOT_GID", os.getgid())
+    module._acl_backup_once(tmp_path / "k3s", b"")
+    assert backup.read_bytes() == b"N"
+
+
+def test_acl_backup_short_write_cleans_temporary(tmp_path, monkeypatch):
+    module = _load()
+    host_etc = tmp_path / "etc"
+    host_etc.mkdir()
+    monkeypatch.setattr(module, "HOST_ETC", host_etc)
+    monkeypatch.setattr(module, "HOST_ROOT_UID", os.getuid())
+    monkeypatch.setattr(module, "HOST_ROOT_GID", os.getgid())
+    monkeypatch.setattr(module.os, "write", lambda _fd, _value: 0)
+    with pytest.raises(module.HardeningError, match="short sensitive"):
+        module._acl_backup_once(tmp_path / "k3s", b"")
+    assert not list((host_etc / "hermes-node-boundary").glob(".*.hermes-*"))
+
+
+def test_acl_backup_tolerates_concurrent_first_writer(tmp_path, monkeypatch):
+    module = _load()
+    host_etc = tmp_path / "etc"
+    host_etc.mkdir()
+    monkeypatch.setattr(module, "HOST_ETC", host_etc)
+    monkeypatch.setattr(module, "HOST_ROOT_UID", os.getuid())
+    monkeypatch.setattr(module, "HOST_ROOT_GID", os.getgid())
+    real_link = os.link
+
+    def race(source, target, **kwargs):
+        real_link(source, target, **kwargs)
+        raise FileExistsError()
+
+    monkeypatch.setattr(module.os, "link", race)
+    module._acl_backup_once(tmp_path / "k3s", b"")
+    assert (host_etc / "hermes-node-boundary/k3s.acl").read_bytes() == b"N"
+
+
+def test_sensitive_root_failure_restores_existing_acl(tmp_path, monkeypatch):
+    module = _load()
+    root = tmp_path / "k3s"
+    root.mkdir()
+    monkeypatch.setattr(module, "HOST_ROOT_UID", os.getuid())
+    current = module._encode_acl(
+        [
+            (module.ACL_USER_OBJ, 7, module.ACL_UNDEFINED_ID),
+            (module.ACL_GROUP_OBJ, 5, module.ACL_UNDEFINED_ID),
+            (module.ACL_MASK, 5, module.ACL_UNDEFINED_ID),
+            (module.ACL_OTHER, 5, module.ACL_UNDEFINED_ID),
+        ]
+    )
+    reads = iter((current, current))
+    writes = []
+    monkeypatch.setattr(module, "_read_acl", lambda _path: next(reads))
+    monkeypatch.setattr(module, "_acl_backup_once", lambda *_a, **_k: None)
+    monkeypatch.setattr(
+        module.os, "setxattr", lambda *args, **kwargs: writes.append(args[2])
+    )
+    with pytest.raises(module.HardeningError, match="validation failed"):
+        module._deny_sensitive_root(root)
+    assert writes[-1] == current
+
+
+@pytest.mark.parametrize("remove_errno", [errno.ENODATA, errno.EPERM])
+def test_sensitive_root_failure_removes_new_acl_or_propagates_cleanup_error(
+    tmp_path, monkeypatch, remove_errno
+):
+    module = _load()
+    root = tmp_path / "k3s"
+    root.mkdir()
+    monkeypatch.setattr(module, "HOST_ROOT_UID", os.getuid())
+    monkeypatch.setattr(module, "_read_acl", lambda _path: b"")
+    monkeypatch.setattr(module, "_acl_backup_once", lambda *_a, **_k: None)
+    monkeypatch.setattr(
+        module.os,
+        "setxattr",
+        lambda *_a, **_k: (_ for _ in ()).throw(RuntimeError("set failed")),
+    )
+    monkeypatch.setattr(
+        module.os,
+        "removexattr",
+        lambda *_a, **_k: (_ for _ in ()).throw(OSError(remove_errno, "remove")),
+    )
+    expected = OSError if remove_errno == errno.EPERM else RuntimeError
+    with pytest.raises(expected):
+        module._deny_sensitive_root(root)
+
+
+def test_database_validation_failure_rolls_back_all_writes(tmp_path, monkeypatch):
+    module, originals, _key_value, _other, _public = _fixture(tmp_path, monkeypatch)
+    real_read = module._read_regular
+    calls = 0
+
+    def missing_expected(path, maximum=module.MAX_ACCOUNT_FILE):
+        nonlocal calls
+        calls += 1
+        if calls == 1:
+            return originals["passwd"].encode(), path.stat(follow_symlinks=False)
+        return real_read(path, maximum)
+
+    monkeypatch.setattr(module, "_read_regular", missing_expected)
+    with pytest.raises(module.HardeningError, match="validation failed"):
+        module._reconcile_databases()
+    for name, value in originals.items():
+        assert (module.HOST_ETC / name).read_text() == value
+
+
+def test_preexisting_target_key_removal_is_verified(tmp_path, monkeypatch):
+    module, _originals, key, _other, public = _fixture(tmp_path, monkeypatch)
+    target = module.HOST_HOME / module.ACCOUNT / ".ssh/authorized_keys"
+    target.parent.mkdir(parents=True)
+    target.write_text(key + "\n", encoding="utf-8")
+    real_write = module.atomic_write
+
+    def ignore_preinstall_target(path, value, metadata):
+        if path == target and value == b"":
+            return
+        real_write(path, value, metadata)
+
+    monkeypatch.setattr(module, "atomic_write", ignore_preinstall_target)
+    with pytest.raises(module.HardeningError, match="preexisting Hermes authorization"):
+        module._move_key(public)
+
+
+def test_existing_target_without_key_does_not_need_precleanup(tmp_path, monkeypatch):
+    module, _originals, _key_value, other, public = _fixture(tmp_path, monkeypatch)
+    target = module.HOST_HOME / module.ACCOUNT / ".ssh/authorized_keys"
+    target.parent.mkdir(parents=True)
+    target.write_text(other + "\n", encoding="utf-8")
+    module._move_key(public)
+    assert target.read_text().endswith("\n")

testing/tests/test_hermes_node_audit_coverage.py

@@ -0,0 +1,221 @@
+"""Behavioral branch coverage for dedicated node-account privilege audits."""
+
+from __future__ import annotations
+
+import os
+import struct
+import sys
+from types import SimpleNamespace
+
+import pytest
+
+from testing.tests.test_hermes_node_account_support import _load
+
+
+def _audit_module():
+    hardening = _load()
+    return sys.modules[hardening.audit_membership.__module__]
+
+
+def test_member_parser_handles_empty_valid_malformed_and_duplicates():
+    module = _audit_module()
+    assert module._members("", "group") == set()
+    assert module._members("one,two", "group") == {"one", "two"}
+    for value in ("one,", " one", "one,one"):
+        with pytest.raises(module.HardeningError):
+            module._members(value, "group")
+
+
+def test_membership_audit_accepts_unrelated_records_and_rejects_account():
+    module = _audit_module()
+    module.audit_membership(
+        "hermes-agent", [["disk", "x", "6", "atlas"]], [["disk", "!", "atlas", ""]]
+    )
+    with pytest.raises(module.HardeningError, match="supplementary"):
+        module.audit_membership(
+            "hermes-agent", [["disk", "x", "6", "hermes-agent"]], []
+        )
+    with pytest.raises(module.HardeningError, match="gshadow"):
+        module.audit_membership("hermes-agent", [], [["disk", "!", "hermes-agent", ""]])
+
+
+@pytest.mark.parametrize("kind", ["mode", "symlink"])
+def test_policy_files_reject_unsafe_directories(tmp_path, kind):
+    module = _audit_module()
+    host_etc = tmp_path / "etc"
+    host_etc.mkdir()
+    root = host_etc / "sudoers.d"
+    if kind == "mode":
+        root.mkdir(mode=0o777)
+        root.chmod(0o777)
+    else:
+        target = tmp_path / "target"
+        target.mkdir()
+        root.symlink_to(target)
+    with pytest.raises(module.HardeningError, match="unsafe sudo/polkit"):
+        module._policy_files(host_etc, tmp_path / "share", os.getuid())
+
+
+def test_policy_file_count_is_bounded(tmp_path):
+    module = _audit_module()
+    root = tmp_path / "etc/sudoers.d"
+    root.mkdir(parents=True)
+    for index in range(257):
+        (root / str(index)).write_text("# safe\n", encoding="utf-8")
+    with pytest.raises(module.HardeningError, match="too many"):
+        module._policy_files(tmp_path / "etc", tmp_path / "share", os.getuid())
+
+
+def _acl(*entries):
+    return struct.pack("<I", 2) + b"".join(
+        struct.pack("<HHI", *entry) for entry in entries
+    )
+
+
+def test_policy_metadata_accepts_safe_acl_and_rejects_malformed_or_writable():
+    module = _audit_module()
+    safe = SimpleNamespace(
+        uid=0,
+        mode=0o600,
+        xattrs=(("system.posix_acl_access", _acl((1, 6, 0xFFFFFFFF))),),
+    )
+    module._audit_policy_metadata(safe, 0, 1200)
+    malformed = SimpleNamespace(
+        uid=0, mode=0o600, xattrs=(("system.posix_acl_access", b"bad"),)
+    )
+    with pytest.raises(module.HardeningError, match="ACL is malformed"):
+        module._audit_policy_metadata(malformed, 0, 1200)
+    writable = SimpleNamespace(
+        uid=0, mode=0o600, xattrs=(("system.posix_acl_access", _acl((2, 2, 1200))),)
+    )
+    with pytest.raises(module.HardeningError, match="grants Hermes write"):
+        module._audit_policy_metadata(writable, 0, 1200)
+
+
+def test_active_sudo_policy_handles_blank_comments_numeric_and_includes():
+    module = _audit_module()
+    text = (
+        "\n# ordinary comment\n#1200 ALL=(ALL) ALL\n"
+        "#includedir /etc/sudoers.d\nroot ALL=(ALL) ALL\n"
+    )
+    active = module._active_sudo_policy(text)
+    assert "#1200" in active and "root ALL" in active
+    assert "ordinary" not in active and "includedir" not in active
+    with pytest.raises(module.HardeningError, match="unaudited"):
+        module._active_sudo_policy("@include /external/policy\n")
+
+
+def test_identity_matcher_covers_account_numeric_and_unrelated_values():
+    module = _audit_module()
+    assert module._mentions_dedicated_identity("hermes-agent ALL", "hermes-agent", 1200)
+    assert module._mentions_dedicated_identity("#1200 ALL", "hermes-agent", 1200)
+    assert not module._mentions_dedicated_identity(
+        "hermes-agent-extra 12001", "hermes-agent", 1200
+    )
+
+
+def _snapshot(value: bytes, *, mode=0o600, uid=0, xattrs=()):
+    return SimpleNamespace(value=value, mode=mode, uid=uid, xattrs=tuple(xattrs))
+
+
+def test_privilege_audit_accepts_files_systemd_nss_and_comments(tmp_path, monkeypatch):
+    module = _audit_module()
+    host_etc = tmp_path / "etc"
+    host_etc.mkdir()
+    nss = host_etc / "nsswitch.conf"
+    nss.write_text(
+        "passwd: files systemd\nhosts: files dns\n# comment\n", encoding="utf-8"
+    )
+    sudoers = host_etc / "sudoers"
+    sudoers.write_text("# safe comment\nroot ALL=(ALL) ALL\n", encoding="utf-8")
+    monkeypatch.setattr(
+        module,
+        "read_regular",
+        lambda path, maximum: _snapshot(path.read_bytes(), uid=os.getuid()),
+    )
+    module.audit_privilege_policies(
+        "hermes-agent", 1200, os.getuid(), host_etc, tmp_path / "share"
+    )
+
+
+@pytest.mark.parametrize(
+    ("value", "match"),
+    [
+        (b"passwd: \n", "external group/account"),
+        (b"passwd: files\xff\n", "not UTF-8"),
+    ],
+)
+def test_nsswitch_audit_rejects_empty_or_non_utf8_sources(
+    tmp_path, monkeypatch, value, match
+):
+    module = _audit_module()
+    host_etc = tmp_path / "etc"
+    host_etc.mkdir()
+    path = host_etc / "nsswitch.conf"
+    path.write_bytes(value)
+    monkeypatch.setattr(
+        module,
+        "read_regular",
+        lambda _path, _maximum: _snapshot(value, uid=os.getuid()),
+    )
+    with pytest.raises(module.HardeningError, match=match):
+        module.audit_privilege_policies(
+            "hermes-agent", 1200, os.getuid(), host_etc, tmp_path / "share"
+        )
+
+
+def test_policy_input_rejects_non_utf8_and_total_size(tmp_path, monkeypatch):
+    module = _audit_module()
+    host_etc = tmp_path / "etc"
+    host_etc.mkdir()
+    sudoers = host_etc / "sudoers"
+    sudoers.write_bytes(b"\xff")
+    monkeypatch.setattr(
+        module,
+        "read_regular",
+        lambda _path, _maximum: _snapshot(b"\xff", uid=os.getuid()),
+    )
+    with pytest.raises(module.HardeningError, match="not UTF-8"):
+        module.audit_privilege_policies(
+            "hermes-agent", 1200, os.getuid(), host_etc, tmp_path / "share"
+        )
+
+    monkeypatch.setattr(module, "_policy_files", lambda *_a: [sudoers] * 9)
+    monkeypatch.setattr(
+        module,
+        "read_regular",
+        lambda _path, _maximum: _snapshot(b"#" * (256 * 1024), uid=os.getuid()),
+    )
+    with pytest.raises(module.HardeningError, match="exceeds safe limit"):
+        module.audit_privilege_policies(
+            "hermes-agent", 1200, os.getuid(), host_etc, tmp_path / "share"
+        )
+
+
+def test_broad_sudo_and_pkla_paths_are_rejected(tmp_path, monkeypatch):
+    module = _audit_module()
+    host_etc = tmp_path / "etc"
+    host_etc.mkdir()
+    sudoers = host_etc / "sudoers"
+    sudoers.write_text("%ALL ALL=(ALL) ALL\n", encoding="utf-8")
+    monkeypatch.setattr(
+        module,
+        "read_regular",
+        lambda path, maximum: _snapshot(path.read_bytes(), uid=os.getuid()),
+    )
+    with pytest.raises(module.HardeningError, match="broad sudo"):
+        module.audit_privilege_policies(
+            "hermes-agent", 1200, os.getuid(), host_etc, tmp_path / "share"
+        )
+
+    sudoers.unlink()
+    policy = host_etc / "polkit-1/localauthority/policy.pkla"
+    policy.parent.mkdir(parents=True)
+    policy.write_text(
+        "Identity=unix-user:*\nAction=org.freedesktop.policykit.exec\nResultAny=yes\n",
+        encoding="utf-8",
+    )
+    with pytest.raises(module.HardeningError, match="root-equivalent broad polkit"):
+        module.audit_privilege_policies(
+            "hermes-agent", 1200, os.getuid(), host_etc, tmp_path / "share"
+        )

testing/tests/test_hermes_node_hardening_coverage.py

@@ -0,0 +1,211 @@
+"""Behavioral branch coverage for node account/key reconciliation."""
+
+from __future__ import annotations
+
+import base64
+import os
+import sys
+import pytest
+
+from testing.tests.test_hermes_node_account_support import (
+    _fixture,
+    _key,
+    _load,
+    _typed_key,
+)
+
+
+@pytest.mark.parametrize(
+    ("value", "fields", "match"),
+    [
+        (b"name:\xff\n", 2, "not UTF-8"),
+        (b"name:x", 2, "final newline"),
+        (b":x\n", 2, "invalid record"),
+        (b"name:x\nname:y\n", 2, "invalid record"),
+    ],
+)
+def test_account_record_parser_rejects_encoding_shape_and_duplicates(
+    value, fields, match
+):
+    module = _load()
+    with pytest.raises(module.HardeningError, match=match):
+        module._records(value, fields, "database")
+
+
+def test_reconcile_record_accepts_exact_existing_identity():
+    module = _load()
+    expected = ["hermes-agent", "x", "1200", "1200"]
+    records = [["root", "x", "0", "0"], expected]
+    assert module._reconcile_record(records, expected, identity_index=2) is records
+
+
+def test_database_transaction_rolls_back_prior_writes(tmp_path, monkeypatch):
+    module, originals, _key_value, _other, _public_key = _fixture(tmp_path, monkeypatch)
+    real_write = module.atomic_write
+    writes = 0
+
+    def fail_second(path, value, metadata):
+        nonlocal writes
+        writes += 1
+        if writes == 2:
+            raise module.HardeningError("synthetic database failure")
+        real_write(path, value, metadata)
+
+    monkeypatch.setattr(module, "atomic_write", fail_second)
+    with pytest.raises(module.HardeningError, match="synthetic database failure"):
+        module._reconcile_databases()
+    assert (module.HOST_ETC / "passwd").read_text(encoding="utf-8") == originals[
+        "passwd"
+    ]
+
+
+def test_database_rollback_rejects_concurrent_postwrite_change(tmp_path, monkeypatch):
+    module, _originals, _key_value, _other, _public_key = _fixture(
+        tmp_path, monkeypatch
+    )
+    real_write = module.atomic_write
+    writes = 0
+
+    def race_then_fail(path, value, metadata):
+        nonlocal writes
+        writes += 1
+        if writes == 2:
+            (module.HOST_ETC / "passwd").write_text(
+                "raced:x:1:1:r:/r:/bin/sh\n", encoding="utf-8"
+            )
+            raise module.HardeningError("synthetic failure")
+        real_write(path, value, metadata)
+
+    monkeypatch.setattr(module, "atomic_write", race_then_fail)
+    with pytest.raises(module.HardeningError, match="prevents rollback"):
+        module._reconcile_databases()
+
+
+@pytest.mark.parametrize(
+    ("line", "match"),
+    [
+        (b"", None),
+        (b"# comment", None),
+        (b"\xff", "malformed"),
+        (b'command="unterminated ssh-ed25519 data', "malformed"),
+        (b"ordinary text", "ambiguous"),
+        (b"ssh-ed25519", "ambiguous"),
+        (b"ssh-ed25519 bad-base64!", "payload"),
+        (b"ssh-ed25519 YWJj", "blob"),
+    ],
+)
+def test_key_identity_rejects_malformed_authorized_key_lines(line, match):
+    module = _load()
+    if match is None:
+        assert module._key_identity(line) is None
+    else:
+        with pytest.raises(module.HardeningError, match=match):
+            module._key_identity(line)
+
+
+def test_key_identity_rejects_type_blob_mismatch():
+    module = _load()
+    actual = b"ssh-rsaXXXX"
+    blob = len(actual).to_bytes(4, "big") + actual + b"material"
+    line = b"ssh-ed25519 " + base64.b64encode(blob)
+    with pytest.raises(module.HardeningError, match="does not match"):
+        module._key_identity(line)
+
+
+@pytest.mark.parametrize("kind", ["multiline", "unsupported"])
+def test_public_key_validator_rejects_multiline_and_unsupported(tmp_path, kind):
+    module = _load()
+    path = tmp_path / "key"
+    value = (
+        "line-one\nline-two\n"
+        if kind == "multiline"
+        else _typed_key("ssh-dss", b"synthetic-material") + "\n"
+    )
+    path.write_text(value, encoding="utf-8")
+    with pytest.raises(module.HardeningError):
+        module._validated_public_key(path)
+
+
+def test_directory_rejects_file_and_conflicting_owner(tmp_path):
+    module = _load()
+    file_path = tmp_path / "file"
+    file_path.write_text("not-directory", encoding="utf-8")
+    with pytest.raises(module.HardeningError, match="unsafe account directory"):
+        module._directory(file_path, mode=0o700, uid=os.getuid(), gid=os.getgid())
+    directory = tmp_path / "directory"
+    directory.mkdir()
+    with pytest.raises(module.HardeningError, match="ownership conflicts"):
+        module._directory(directory, mode=0o700, uid=123456, gid=123456)
+
+
+def test_move_key_skips_missing_legacy_files(tmp_path, monkeypatch):
+    module = _load()
+    key = _key(b"synthetic-key")
+    public = tmp_path / "public"
+    public.write_text(key + "\n", encoding="utf-8")
+    home = tmp_path / "home"
+    home.mkdir()
+    monkeypatch.setattr(module, "HOST_HOME", home)
+    monkeypatch.setattr(module, "ACCOUNT_UID", os.getuid())
+    monkeypatch.setattr(module, "ACCOUNT_GID", os.getgid())
+    module._move_key(public)
+    assert (home / module.ACCOUNT / ".ssh/authorized_keys").read_text() == key + "\n"
+
+
+def test_move_key_backs_up_different_existing_target(tmp_path, monkeypatch):
+    module, _originals, key, other, public = _fixture(tmp_path, monkeypatch)
+    for legacy in module.LEGACY_ACCOUNTS:
+        (module.HOST_HOME / legacy / ".ssh/authorized_keys").write_text(other + "\n")
+    target = module.HOST_HOME / module.ACCOUNT / ".ssh/authorized_keys"
+    target.parent.mkdir(parents=True)
+    target.write_text(other + "\n")
+    module._move_key(public)
+    assert target.read_text() == key + "\n"
+    assert (
+        target.with_name(target.name + ".hermes-boundary-backup").read_text()
+        == other + "\n"
+    )
+
+
+def test_move_key_fails_if_legacy_removal_or_target_install_does_not_persist(
+    tmp_path, monkeypatch
+):
+    module, _originals, _key_value, _other, public = _fixture(tmp_path, monkeypatch)
+    real_write = module.atomic_write
+
+    def ignore_legacy(path, value, metadata):
+        if module.ACCOUNT not in path.parts and path.name == "authorized_keys":
+            return
+        real_write(path, value, metadata)
+
+    monkeypatch.setattr(module, "atomic_write", ignore_legacy)
+    with pytest.raises(module.HardeningError, match="legacy Hermes authorization"):
+        module._move_key(public)
+
+    second = tmp_path / "second"
+    second.mkdir()
+    module, _originals, _key_value, _other, public = _fixture(second, monkeypatch)
+    real_write = module.atomic_write
+
+    def ignore_target(path, value, metadata):
+        if module.ACCOUNT in path.parts and path.name == "authorized_keys":
+            path.parent.mkdir(parents=True, exist_ok=True)
+            path.write_bytes(b"wrong\n")
+            return
+        real_write(path, value, metadata)
+
+    monkeypatch.setattr(module, "atomic_write", ignore_target)
+    with pytest.raises(module.HardeningError, match="validation failed"):
+        module._move_key(public)
+
+
+def test_main_reconciles_requested_key(monkeypatch, tmp_path, capsys):
+    module = _load()
+    key = tmp_path / "key"
+    key.write_text("value", encoding="utf-8")
+    seen = []
+    monkeypatch.setattr(module, "reconcile", seen.append)
+    monkeypatch.setattr(sys, "argv", ["hardener", "--public-key-file", str(key)])
+    assert module.main() == 0
+    assert seen == [key]
+    assert "reconciled" in capsys.readouterr().out

testing/tests/test_hermes_node_io_coverage.py

@@ -0,0 +1,189 @@
+"""Behavioral branch coverage for crash-safe node account file I/O."""
+
+from __future__ import annotations
+
+import os
+import stat
+import sys
+from pathlib import Path
+
+import pytest
+
+from testing.tests.test_hermes_node_account_support import _load
+
+
+def _io_module():
+    hardening = _load()
+    return sys.modules[hardening.atomic_write.__module__]
+
+
+def _snapshot(module, path: Path, *, value=None, xattrs=()):
+    metadata = path.stat()
+    content = path.read_bytes() if value is None else value
+    return module.FileSnapshot(
+        value=content,
+        device=metadata.st_dev,
+        inode=metadata.st_ino,
+        mode=stat.S_IMODE(metadata.st_mode),
+        uid=metadata.st_uid,
+        gid=metadata.st_gid,
+        size=len(content),
+        mtime_ns=metadata.st_mtime_ns,
+        ctime_ns=metadata.st_ctime_ns,
+        xattrs=tuple(xattrs),
+    )
+
+
+def test_xattr_reader_uses_compatibility_fallback(monkeypatch):
+    module = _io_module()
+    calls = []
+
+    def listxattr(target, **kwargs):
+        calls.append((target, kwargs))
+        if kwargs:
+            raise TypeError("fd API")
+        return ["user.safe"]
+
+    def getxattr(target, name, **kwargs):
+        if kwargs:
+            raise TypeError("fd API")
+        return b"value"
+
+    monkeypatch.setattr(module.os, "listxattr", listxattr)
+    monkeypatch.setattr(module.os, "getxattr", getxattr)
+    assert module._xattrs(7) == (("user.safe", b"value"),)
+    assert len(calls) == 2
+
+
+@pytest.mark.parametrize("failure", ["count", "name", "value", "total"])
+def test_xattr_reader_enforces_all_bounds(monkeypatch, failure):
+    module = _io_module()
+    if failure == "count":
+        names = [f"user.{index}" for index in range(module.MAX_XATTRS + 1)]
+    elif failure == "name":
+        names = ["x" * (module.MAX_XATTR_NAME + 1)]
+    elif failure == "total":
+        names = ["user.one", "user.two", "user.three", "user.four", "user.five"]
+    else:
+        names = ["user.large"]
+    monkeypatch.setattr(module.os, "listxattr", lambda *_a, **_k: names)
+    size = module.MAX_XATTR_VALUE + 1 if failure == "value" else module.MAX_XATTR_VALUE
+    monkeypatch.setattr(module.os, "getxattr", lambda *_a, **_k: b"x" * size)
+    with pytest.raises(module.HardeningError):
+        module._xattrs(7)
+
+
+def test_restore_xattrs_removes_unexpected_and_sets_expected(monkeypatch):
+    module = _io_module()
+    monkeypatch.setattr(
+        module, "_xattrs", lambda _fd: (("user.old", b"old"), ("user.keep", b"before"))
+    )
+    removed = []
+    written = []
+    monkeypatch.setattr(
+        module.os, "removexattr", lambda fd, name: removed.append((fd, name))
+    )
+    monkeypatch.setattr(
+        module.os, "setxattr", lambda fd, name, value: written.append((fd, name, value))
+    )
+    module._restore_xattrs(4, (("user.keep", b"after"),))
+    assert removed == [(4, "user.old")]
+    assert written == [(4, "user.keep", b"after")]
+
+
+def test_read_regular_rejects_directory_oversize_and_changed_read(
+    tmp_path, monkeypatch
+):
+    module = _io_module()
+    with pytest.raises(module.HardeningError, match="unsafe regular"):
+        module.read_regular(tmp_path, 10)
+    source = tmp_path / "file"
+    source.write_bytes(b"12345")
+    with pytest.raises(module.HardeningError, match="unsafe regular"):
+        module.read_regular(source, 4)
+    monkeypatch.setattr(module.os, "read", lambda _fd, _size: b"1")
+    with pytest.raises(module.HardeningError, match="short file read"):
+        module.read_regular(source, 10)
+
+
+def test_assert_unchanged_rejects_changed_snapshot(tmp_path):
+    module = _io_module()
+    source = tmp_path / "passwd"
+    source.write_bytes(b"old")
+    snapshot = module.read_regular(source, 16)
+    source.write_bytes(b"new")
+    with pytest.raises(module.HardeningError, match="concurrent"):
+        module.assert_unchanged(source, snapshot, 16)
+
+
+def test_atomic_write_short_write_unlinks_temporary(tmp_path, monkeypatch):
+    module = _io_module()
+    target = tmp_path / "passwd"
+    target.write_bytes(b"old")
+    snapshot = module.read_regular(target, 16)
+    monkeypatch.setattr(module.os, "write", lambda _fd, _value: 0)
+    with pytest.raises(module.HardeningError, match="short atomic write"):
+        module.atomic_write(target, b"new", snapshot)
+    assert target.read_bytes() == b"old"
+    assert not list(tmp_path.glob(".passwd.hermes-*"))
+
+
+def test_backup_existing_is_validated_and_short_write_is_removed(tmp_path, monkeypatch):
+    module = _io_module()
+    source = tmp_path / "passwd"
+    source.write_bytes(b"old")
+    snapshot = module.read_regular(source, 16)
+    backup = tmp_path / "passwd.hermes-boundary-backup"
+    backup.write_bytes(b"preserved")
+    assert module.backup_once(source, snapshot, 16) == backup
+
+    backup.unlink()
+    monkeypatch.setattr(module.os, "write", lambda _fd, _value: 0)
+    with pytest.raises(module.HardeningError, match="short backup write"):
+        module.backup_once(source, snapshot, 16)
+    assert not backup.exists()
+
+
+def test_account_lock_rejects_unsafe_metadata(tmp_path):
+    module = _io_module()
+    lock = tmp_path / ".pwd.lock"
+    lock.write_text("", encoding="utf-8")
+    lock.chmod(0o644)
+    with (
+        pytest.raises(module.HardeningError, match="unsafe standard account lock"),
+        module.account_lock(tmp_path, expected_uid=os.getuid()),
+    ):
+        pass
+
+
+def test_account_lock_retries_then_succeeds(tmp_path, monkeypatch):
+    module = _io_module()
+    attempts = []
+
+    def lockf(_descriptor, operation):
+        attempts.append(operation)
+        if len(attempts) == 1:
+            raise BlockingIOError()
+
+    monkeypatch.setattr(module.fcntl, "lockf", lockf)
+    monkeypatch.setattr(module.time, "sleep", lambda _seconds: None)
+    with module.account_lock(tmp_path, expected_uid=os.getuid()):
+        assert attempts
+    assert len(attempts) == 3
+
+
+def test_account_lock_reports_timeout(tmp_path, monkeypatch):
+    module = _io_module()
+
+    def lockf(_descriptor, operation):
+        if operation != module.fcntl.LOCK_UN:
+            raise BlockingIOError()
+
+    monkeypatch.setattr(module.fcntl, "lockf", lockf)
+    ticks = iter((0.0, 16.0))
+    monkeypatch.setattr(module.time, "monotonic", lambda: next(ticks))
+    with (
+        pytest.raises(module.HardeningError, match="lock is busy"),
+        module.account_lock(tmp_path, expected_uid=os.getuid()),
+    ):
+        pass

testing/tests/test_hermes_runtime_stage_coverage.py

@@ -0,0 +1,137 @@
+"""Behavioral branch coverage for Vault-to-memory runtime access staging."""
+
+from __future__ import annotations
+
+import json
+import sys
+from pathlib import Path
+
+import pytest
+
+from testing.tests.test_hermes_runtime_access import _load
+
+
+def _stage(tmp_path: Path, monkeypatch):
+    module = _load("stage_runtime_access")
+    vault = tmp_path / "vault"
+    runtime = tmp_path / "runtime"
+    home = tmp_path / "home"
+    vault.mkdir()
+    home.mkdir()
+    monkeypatch.setattr(module, "VAULT_ROOT", vault)
+    monkeypatch.setattr(module, "RUNTIME_ROOT", runtime)
+    monkeypatch.setattr(module, "PERSISTENT_HOME", home)
+    monkeypatch.setattr(module.os, "chown", lambda *_args: None)
+    return module, vault, runtime, home
+
+
+def test_private_directory_and_secret_copy_validate_mode_and_content(
+    tmp_path, monkeypatch
+):
+    module, vault, runtime, _home = _stage(tmp_path, monkeypatch)
+    module._owned_directory(runtime)
+    assert runtime.stat().st_mode & 0o777 == 0o700
+    (vault / "empty").write_text(" \n", encoding="utf-8")
+    with pytest.raises(RuntimeError, match="is empty"):
+        module._copy_secret("empty", runtime / "empty")
+    (vault / "value").write_text("  staged  \n", encoding="utf-8")
+    assert module._copy_secret("value", runtime / "value") == "staged"
+    assert (runtime / "value").read_text(encoding="utf-8") == "staged\n"
+
+
+@pytest.mark.parametrize(
+    ("document", "required", "match"),
+    [
+        ({"outer": {}}, ("outer", "token"), "invalid shape"),
+        ({"outer": {"token": ""}}, ("outer", "token"), "empty credential"),
+        ({"outer": {"token": 7}}, ("outer", "token"), "empty credential"),
+    ],
+)
+def test_json_staging_rejects_missing_nontext_and_empty_credentials(
+    tmp_path, monkeypatch, document, required, match
+):
+    module, vault, runtime, _home = _stage(tmp_path, monkeypatch)
+    runtime.mkdir()
+    (vault / "credential").write_text(json.dumps(document), encoding="utf-8")
+    destination = runtime / "credential.json"
+    with pytest.raises(RuntimeError, match=match):
+        module._validated_json("credential", destination, required)
+    assert not destination.exists()
+
+
+def test_noncredential_linker_skips_absent_symlink_and_existing_entries(
+    tmp_path, monkeypatch
+):
+    module, _vault, runtime, home = _stage(tmp_path, monkeypatch)
+    runtime.mkdir()
+    source = home / "state"
+    destination = runtime / "state"
+    destination.mkdir()
+    module._link_noncredential_state(source, destination, ("allowed",))
+
+    source.mkdir()
+    (source / "allowed").write_text("safe", encoding="utf-8")
+    (source / "existing").write_text("new", encoding="utf-8")
+    (source / "target-link").write_text("new", encoding="utf-8")
+    (source / "linked").symlink_to(source / "allowed")
+    (destination / "existing").write_text("preserve", encoding="utf-8")
+    (destination / "target-link").symlink_to(destination / "existing")
+    module._link_noncredential_state(
+        source,
+        destination,
+        ("missing", "linked", "existing", "target-link", "allowed"),
+    )
+    assert (destination / "allowed").is_symlink()
+    assert (destination / "existing").read_text(encoding="utf-8") == "preserve"
+
+
+def test_empty_auth_store_is_private_and_noncredential(tmp_path, monkeypatch):
+    module, _vault, runtime, _home = _stage(tmp_path, monkeypatch)
+    runtime.mkdir()
+    module._write_empty_auth_store()
+    path = runtime / "hermes-auth.json"
+    assert json.loads(path.read_text(encoding="utf-8")) == {
+        "version": 1,
+        "providers": {},
+        "credential_pool": {},
+    }
+    assert path.stat().st_mode & 0o777 == 0o600
+
+
+def test_node_ssh_config_rejects_nul_and_removes_partial_file(tmp_path, monkeypatch):
+    module, vault, runtime, _home = _stage(tmp_path, monkeypatch)
+    runtime.mkdir()
+    (vault / "node-ssh-config").write_bytes(b"Host titan-01\x00\n")
+    with pytest.raises(RuntimeError, match="control data"):
+        module._stage_node_ssh_config()
+    assert not (runtime / "node-ssh-config").exists()
+
+
+@pytest.mark.parametrize(
+    ("mode", "secret_name"),
+    [("chat", "chat-relay-key"), ("triage", "triage-api-key")],
+)
+def test_isolated_tenant_staging_exposes_only_required_secret(
+    tmp_path, monkeypatch, mode, secret_name
+):
+    module, vault, runtime, _home = _stage(tmp_path, monkeypatch)
+    (vault / secret_name).write_text("synthetic-value\n", encoding="utf-8")
+    getattr(module, f"stage_{mode}")()
+    assert (runtime / secret_name).read_text(encoding="utf-8") == "synthetic-value\n"
+    assert (runtime / "hermes-auth.json").exists()
+    assert sorted(path.name for path in runtime.iterdir()) == sorted(
+        ["hermes-auth.json", secret_name]
+    )
+
+
+@pytest.mark.parametrize("mode", ["agent", "chat", "triage"])
+def test_main_dispatches_each_mode(mode, monkeypatch, capsys):
+    module = _load("stage_runtime_access")
+    calls = []
+    monkeypatch.setattr(module, "stage_agent", lambda: calls.append("agent"))
+    monkeypatch.setattr(module, "stage_chat", lambda: calls.append("chat"))
+    monkeypatch.setattr(module, "stage_triage", lambda: calls.append("triage"))
+    monkeypatch.setattr(sys, "argv", ["stage-runtime-access", mode])
+    assert module.main() == 0
+    assert calls == [mode]
+    assert "staged from Vault" in capsys.readouterr().out

testing/tests/test_hermes_scm_broker_client_coverage.py

@@ -0,0 +1,157 @@
+"""Behavioral coverage for the credential-free SCM broker client and I/O."""
+
+from __future__ import annotations
+
+import io
+import json
+import pytest
+
+from testing.tests.test_hermes_scm_broker_support import Response, _load
+
+
+def test_client_redirect_and_default_opener_are_fail_closed(monkeypatch):
+    client = _load("scm_broker_client")
+    with pytest.raises(client.PolicyError, match="redirects"):
+        client.RejectRedirectHandler().redirect_request(
+            object(), None, 302, "redirect", {}, "http://elsewhere.invalid"
+        )
+    sentinel = object()
+    monkeypatch.setattr(
+        client._OPENER, "open", lambda request, timeout: (request, timeout, sentinel)
+    )
+    assert client._open("request", 4) == ("request", 4, sentinel)
+
+
+@pytest.mark.parametrize(
+    ("endpoint", "payload", "match"),
+    [
+        ("/v1/admin", {}, "outside"),
+        ("/v1/metadata", {"path": "x" * (64 * 1024)}, "safe size"),
+    ],
+)
+def test_client_rejects_unknown_or_oversized_operations(endpoint, payload, match):
+    client = _load("scm_broker_client")
+    with pytest.raises(client.PolicyError, match=match):
+        client.request(endpoint, payload, opener=lambda *_a, **_k: None)
+
+
+class _GetCodeResponse(Response):
+    def __init__(self, body: bytes, *, content_type="application/json", code=200):
+        super().__init__(body, content_type=content_type)
+        del self.status
+        self._code = code
+
+    def getcode(self):
+        return self._code
+
+
+def test_client_accepts_getcode_fallback_and_create_payload():
+    client = _load("scm_broker_client")
+    seen = []
+
+    def opener(request, timeout):
+        seen.append((request, timeout))
+        return _GetCodeResponse(b"{}")
+
+    result = client.create_draft(
+        "cassandra",
+        base="main",
+        head="feature/coverage",
+        head_sha="a" * 40,
+        title="WIP: Coverage",
+        body="Review evidence",
+        opener=opener,
+    )
+    assert result == b"{}"
+    assert json.loads(seen[0][0].data)["repo"] == "cassandra"
+
+
+@pytest.mark.parametrize(
+    ("response", "match"),
+    [
+        (_GetCodeResponse(b"{}", code=201), "HTTP status"),
+        (_GetCodeResponse(b"{}", content_type="text/plain"), "response type"),
+        (
+            Response(b"x" * (2 * 1024 * 1024 + 1), content_type="application/json"),
+            "safe size",
+        ),
+        (Response(b"not-json", content_type="application/json"), "Expecting value"),
+    ],
+)
+def test_client_rejects_bad_status_type_size_and_json(response, match):
+    client = _load("scm_broker_client")
+    expected = (
+        json.JSONDecodeError if match == "Expecting value" else client.PolicyError
+    )
+    with pytest.raises(expected, match=match):
+        client.request(
+            "/v1/metadata",
+            {"path": "/api/v1/repos/atlas/cassandra"},
+            opener=lambda *_a, **_k: response,
+        )
+
+
+class _Socket:
+    def __init__(self):
+        self.timeouts = []
+
+    def settimeout(self, value):
+        self.timeouts.append(value)
+
+
+class _Stream:
+    def __init__(self, body: bytes, socket_shape: str = "direct"):
+        self.body = io.BytesIO(body)
+        self.socket = _Socket()
+        if socket_shape == "nested":
+            self.fp = type(
+                "FP", (), {"raw": type("Raw", (), {"_sock": self.socket})()}
+            )()
+        elif socket_shape == "direct":
+            self.fp = type("FP", (), {"raw": self.socket})()
+        else:
+            self.fp = None
+
+    def read(self, size):
+        return self.body.read(size)
+
+
+def test_response_spool_handles_direct_and_absent_socket_shapes():
+    module = _load("scm_broker_io")
+    for shape in ("direct", "absent"):
+        stream = _Stream(b"safe", shape)
+        spool, length = module.spool_response(
+            stream,
+            16,
+            (b"forbidden",),
+            memory_limit=8,
+            chunk_size=2,
+            deadline_seconds=5,
+        )
+        try:
+            assert length == 4 and spool.read() == b"safe"
+        finally:
+            spool.close()
+
+
+@pytest.mark.parametrize("failure", ["deadline", "size", "credential", "read"])
+def test_response_spool_closes_and_rejects_each_failure(monkeypatch, failure):
+    module = _load("scm_broker_io")
+    stream = _Stream(b"safe-forbidden-value")
+    kwargs = {
+        "maximum": 128,
+        "forbidden": (b"forbidden",),
+        "memory_limit": 8,
+        "chunk_size": 5,
+        "deadline_seconds": 5,
+    }
+    if failure == "deadline":
+        ticks = iter((10.0, 16.0))
+        monkeypatch.setattr(module.time, "monotonic", lambda: next(ticks))
+    elif failure == "size":
+        kwargs["maximum"] = 2
+    elif failure == "read":
+        stream.read = lambda _size: (_ for _ in ()).throw(OSError("read failed"))
+    expected = module.PolicyError if failure != "read" else OSError
+    with pytest.raises(expected):
+        module.spool_response(stream, **kwargs)

testing/tests/test_hermes_scm_broker_handler_coverage.py

@@ -0,0 +1,265 @@
+"""Behavioral branch coverage for SCM broker request handling."""
+
+from __future__ import annotations
+
+import io
+import json
+import sys
+from email.message import Message
+
+import pytest
+
+from testing.tests.test_hermes_scm_broker_support import _load, _receive_command
+
+
+class _Connection:
+    def __init__(self):
+        self.timeouts = []
+
+    def settimeout(self, value):
+        self.timeouts.append(value)
+
+
+def _headers(*, content_type="application/json", body=b""):
+    value = Message()
+    value["Content-Type"] = content_type
+    value["Content-Length"] = str(len(body))
+    return value
+
+
+def _handler(broker, *, path="/healthz", body=b"", content_type="application/json"):
+    handler = object.__new__(broker.BrokerHandler)
+    handler.path = path
+    handler.headers = _headers(content_type=content_type, body=body)
+    handler.rfile = io.BytesIO(body)
+    handler.wfile = io.BytesIO()
+    handler.connection = _Connection()
+    handler.events = []
+    handler.send_response = lambda status: handler.events.append(("status", status))
+    handler.send_header = lambda name, value: handler.events.append((name, value))
+    handler.end_headers = lambda: handler.events.append(("headers", "done"))
+    return handler
+
+
+def test_header_validation_accepts_safe_ascii_and_rejects_controls():
+    broker = _load("scm_broker")
+    handler = _handler(broker)
+    handler._validate_headers()
+
+    class HeaderBag:
+        def items(self):
+            return [("X-Test", "bad\x01value")]
+
+        def get_all(self, _name, default):
+            return default
+
+    handler.headers = HeaderBag()
+    with pytest.raises(broker.PolicyError, match="controls"):
+        handler._validate_headers()
+
+    class UnicodeHeader(HeaderBag):
+        def items(self):
+            return [("X-Test", "máin")]
+
+    handler.headers = UnicodeHeader()
+    with pytest.raises(broker.PolicyError, match="ASCII"):
+        handler._validate_headers()
+
+
+def test_json_reject_and_stream_response_methods(monkeypatch):
+    broker = _load("scm_broker")
+    handler = _handler(broker)
+    handler._json(202, b'{"ok":true}')
+    assert ("status", 202) in handler.events
+    assert handler.wfile.getvalue() == b'{"ok":true}'
+
+    handler = _handler(broker)
+    handler._reject(403)
+    assert ("status", 403) in handler.events
+    assert json.loads(handler.wfile.getvalue()) == {"error": "request rejected"}
+
+    handler = _handler(broker)
+    handler._stream(200, "application/test", io.BytesIO(b"streamed"), 8)
+    assert handler.wfile.getvalue() == b"streamed"
+    assert handler.connection.timeouts
+    handler.log_message("ignored %s", "value")
+
+    handler = _handler(broker)
+    ticks = iter((0.0, 121.0))
+    monkeypatch.setattr(broker.time, "monotonic", lambda: next(ticks))
+    with pytest.raises(broker.PolicyError, match="write deadline"):
+        handler._stream(200, "application/test", io.BytesIO(b"x"), 1)
+
+
+def test_get_health_and_git_discovery_paths(monkeypatch):
+    broker = _load("scm_broker")
+    handler = _handler(broker)
+    handler.do_GET()
+    assert json.loads(handler.wfile.getvalue()) == {"status": "ok"}
+
+    closed = []
+
+    class Body(io.BytesIO):
+        def close(self):
+            closed.append(True)
+            super().close()
+
+    handler = _handler(
+        broker,
+        path="/git/atlas/cassandra.git/info/refs?service=git-upload-pack",
+    )
+    monkeypatch.setattr(broker, "read_token", lambda: "sentinel")
+    monkeypatch.setattr(
+        broker,
+        "_upstream_git_request",
+        lambda *args, **kwargs: (Body(b"advertisement"), 13),
+    )
+    handler.do_GET()
+    assert handler.wfile.getvalue() == b"advertisement"
+    assert closed == [True]
+
+    handler = _handler(broker, path="/git/atlas/cassandra.git/git-upload-pack")
+    handler.do_GET()
+    assert json.loads(handler.wfile.getvalue()) == {"error": "request rejected"}
+
+
+def test_post_dispatches_control_and_git_or_rejects(monkeypatch):
+    broker = _load("scm_broker")
+    control = _handler(broker, path="/v1/metadata", body=b"{}")
+    git = _handler(broker, path="/git/atlas/cassandra.git/git-upload-pack", body=b"")
+    calls = []
+    control._control = lambda: calls.append("control")
+    control._git_rpc = lambda: calls.append("git")
+    git._control = lambda: calls.append("control")
+    git._git_rpc = lambda: calls.append("git")
+    control.do_POST()
+    git.do_POST()
+    assert calls == ["control", "git"]
+
+    rejected = _handler(broker, path="/v1/metadata", body=b"{}")
+    rejected._control = lambda: (_ for _ in ()).throw(broker.PolicyError("no"))
+    rejected.do_POST()
+    assert json.loads(rejected.wfile.getvalue()) == {"error": "request rejected"}
+
+
+def test_control_metadata_and_draft_fields(monkeypatch):
+    broker = _load("scm_broker")
+    monkeypatch.setattr(broker, "read_token", lambda: "sentinel")
+    monkeypatch.setattr(
+        broker, "read", lambda path, token: json.dumps({"path": path}).encode()
+    )
+    monkeypatch.setattr(
+        broker,
+        "create_draft",
+        lambda token, **data: json.dumps({"token_used": bool(token), **data}).encode(),
+    )
+
+    metadata_body = b'{"path":"/api/v1/repos/atlas/cassandra"}'
+    metadata = _handler(broker, path="/v1/metadata", body=metadata_body)
+    metadata._control()
+    assert json.loads(metadata.wfile.getvalue())["path"].endswith("cassandra")
+
+    draft_data = {
+        "base": "main",
+        "body": "Review evidence",
+        "head": "feature/coverage",
+        "head_sha": "a" * 40,
+        "repo": "cassandra",
+        "title": "WIP: Coverage",
+    }
+    draft_body = json.dumps(draft_data).encode()
+    draft = _handler(broker, path="/v1/drafts", body=draft_body)
+    draft._control()
+    assert json.loads(draft.wfile.getvalue())["repo"] == "cassandra"
+
+    for path, body, match in (
+        ("/v1/metadata", b'{"wrong":"field"}', "fields"),
+        ("/v1/drafts", b'{"repo":1}', "fields"),
+    ):
+        with pytest.raises(broker.PolicyError, match=match):
+            _handler(broker, path=path, body=body)._control()
+
+    monkeypatch.setattr(broker, "read", lambda *_a, **_k: b"sentinel")
+    with pytest.raises(broker.PolicyError, match="reflected"):
+        _handler(broker, path="/v1/metadata", body=metadata_body)._control()
+
+
+@pytest.mark.parametrize("service", ["git-upload-pack", "git-receive-pack"])
+def test_git_rpc_streams_allowed_upload_and_receive(monkeypatch, service):
+    broker = _load("scm_broker")
+    if service == "git-receive-pack":
+        body = _receive_command(b"0" * 40, b"1" * 40, b"refs/heads/hermes/coverage")
+    else:
+        body = b"upload-request"
+    handler = _handler(
+        broker,
+        path=f"/git/atlas/cassandra.git/{service}",
+        body=body,
+        content_type=f"application/x-{service}-request",
+    )
+    monkeypatch.setattr(broker, "read_token", lambda: "sentinel")
+    seen = []
+
+    def upstream(target, **kwargs):
+        seen.append((target, kwargs))
+        return io.BytesIO(b"result"), 6
+
+    monkeypatch.setattr(broker, "_upstream_git_request", upstream)
+    handler._git_rpc()
+    assert handler.wfile.getvalue() == b"result"
+    assert seen[0][0].endswith(service)
+    assert seen[0][1]["body_length"] == len(body)
+
+
+@pytest.mark.parametrize(
+    ("path", "content_type", "transfer", "match"),
+    [
+        (
+            "/git/atlas/cassandra.git/info/refs?service=git-upload-pack",
+            "application/x-git-upload-pack-request",
+            None,
+            "operation",
+        ),
+        (
+            "/git/atlas/cassandra.git/git-upload-pack",
+            "text/plain",
+            None,
+            "request type",
+        ),
+        (
+            "/git/atlas/cassandra.git/git-upload-pack",
+            "application/x-git-upload-pack-request",
+            "chunked",
+            "request type",
+        ),
+    ],
+)
+def test_git_rpc_rejects_discovery_wrong_type_and_chunking(
+    path, content_type, transfer, match
+):
+    broker = _load("scm_broker")
+    handler = _handler(broker, path=path, content_type=content_type)
+    if transfer:
+        handler.headers["Transfer-Encoding"] = transfer
+    with pytest.raises(broker.PolicyError, match=match):
+        handler._git_rpc()
+
+
+def test_broker_main_constructs_bounded_server(monkeypatch):
+    broker = _load("scm_broker")
+    seen = []
+
+    class Server:
+        def __init__(self, address, handler):
+            seen.append((address, handler))
+
+        def serve_forever(self):
+            seen.append("served")
+
+    monkeypatch.setattr(broker, "BoundedThreadingHTTPServer", Server)
+    monkeypatch.setattr(
+        sys, "argv", ["broker", "--listen", "127.0.0.1", "--port", "9191"]
+    )
+    assert broker.main() == 0
+    assert seen[0] == (("127.0.0.1", 9191), broker.BrokerHandler)
+    assert seen[1] == "served"

testing/tests/test_hermes_scm_broker_internal_coverage.py

@@ -0,0 +1,263 @@
+"""Behavioral branch coverage for SCM broker protocol helpers."""
+
+from __future__ import annotations
+
+import io
+from email.message import Message
+
+import pytest
+
+from testing.tests.test_hermes_scm_broker_support import Response, _load
+
+
+def test_broker_redirect_status_and_bounded_read_helpers(monkeypatch):
+    broker = _load("scm_broker")
+    with pytest.raises(broker.PolicyError, match="redirects"):
+        broker.RejectRedirect().redirect_request(
+            object(), None, 302, "redirect", {}, "https://elsewhere.invalid"
+        )
+
+    class CodeOnly:
+        def getcode(self):
+            return 204
+
+    assert broker._status(type("Status", (), {"status": 200})()) == 200
+    assert broker._status(CodeOnly()) == 204
+    assert broker._status(object()) is None
+
+    with pytest.raises(broker.PolicyError, match="safe size"):
+        broker._read_bounded(io.BytesIO(b"x"), 4, -1)
+    with pytest.raises(broker.PolicyError, match="safe size"):
+        broker._read_bounded(io.BytesIO(b"xxxxx"), 4)
+    with pytest.raises(broker.PolicyError, match="safe size"):
+        broker._read_bounded(io.BytesIO(b"x"), 4, 2)
+    with pytest.raises(broker.PolicyError, match="deadline"):
+        broker._read_bounded(io.BytesIO(b"x"), 4, 1, deadline=0)
+
+    timeouts = []
+    assert (
+        broker._read_bounded(
+            io.BytesIO(b"xy"),
+            4,
+            2,
+            deadline=broker.time.monotonic() + 5,
+            set_timeout=timeouts.append,
+        )
+        == b"xy"
+    )
+    assert timeouts
+
+
+def test_spool_bounded_rejects_bounds_early_eof_and_closes_on_read_error():
+    broker = _load("scm_broker")
+    for length in (-1, 5):
+        with pytest.raises(broker.PolicyError, match="safe size"):
+            broker._spool_bounded(
+                io.BytesIO(b"x"),
+                4,
+                length,
+                token="sentinel",
+                context="request",
+                deadline=999999999,
+            )
+    with pytest.raises(broker.PolicyError, match="ended early"):
+        broker._spool_bounded(
+            io.BytesIO(b"x"),
+            4,
+            2,
+            token="sentinel",
+            context="request",
+            deadline=999999999,
+        )
+
+    class Broken:
+        def read(self, _size):
+            raise OSError("broken stream")
+
+    with pytest.raises(OSError, match="broken"):
+        broker._spool_bounded(
+            Broken(), 4, 1, token="sentinel", context="request", deadline=999999999
+        )
+
+
+class _HandlerInput:
+    def __init__(self, body: bytes, *, content_type="application/json", length=None):
+        self.headers = Message()
+        self.headers["Content-Type"] = content_type
+        if length is None:
+            length = len(body)
+        self.headers["Content-Length"] = str(length)
+        self.rfile = io.BytesIO(body)
+        self.connection = type(
+            "Connection", (), {"settimeout": lambda self, value: None}
+        )()
+
+
+def test_control_json_loader_requires_fixed_json_object():
+    broker = _load("scm_broker")
+    handler = _HandlerInput(b"{}")
+    handler.headers["Transfer-Encoding"] = "chunked"
+    with pytest.raises(broker.PolicyError, match="chunked"):
+        broker._load_json(handler)
+    with pytest.raises(broker.PolicyError, match="must be JSON"):
+        broker._load_json(_HandlerInput(b"{}", content_type="text/plain"))
+    with pytest.raises(broker.PolicyError, match="must be an object"):
+        broker._load_json(_HandlerInput(b"[]"))
+    assert broker._load_json(_HandlerInput(b'{"path":"safe"}')) == {"path": "safe"}
+
+
+@pytest.mark.parametrize(
+    ("target", "expected"),
+    [
+        (
+            "/git/atlas/cassandra.git/info/refs?service=git-upload-pack",
+            ("cassandra", "info/refs", "git-upload-pack"),
+        ),
+        (
+            "/git/atlas/cassandra.git/info/refs?service=git-receive-pack",
+            ("cassandra", "info/refs", "git-receive-pack"),
+        ),
+        (
+            "/git/atlas/cassandra.git/git-upload-pack",
+            ("cassandra", "git-upload-pack", "git-upload-pack"),
+        ),
+        (
+            "/git/atlas/cassandra.git/git-receive-pack",
+            ("cassandra", "git-receive-pack", "git-receive-pack"),
+        ),
+    ],
+)
+def test_git_target_accepts_each_exact_smart_http_route(target, expected):
+    broker = _load("scm_broker")
+    assert broker._git_target(target) == expected
+
+
+def test_content_length_and_credential_helpers_cover_safe_and_rejected_values():
+    broker = _load("scm_broker")
+    headers = Message()
+    headers["Content-Length"] = "0"
+    assert broker._content_length(headers, 1) == 0
+    forms = broker._credential_forms("sentinel")
+    assert len(forms) == 3 and forms[0] == b"sentinel"
+    broker._reject_credential_bytes(b"safe", "sentinel", "response")
+    with pytest.raises(broker.PolicyError, match="credential material"):
+        broker._reject_credential_bytes(b"prefix " + forms[2], "sentinel", "response")
+
+
+def test_receive_prefix_preserves_stream_position_and_accepts_bytes():
+    broker = _load("scm_broker")
+    assert broker._receive_prefix(b"bytes") == b"bytes"
+    stream = io.BytesIO(b"prefix")
+    stream.seek(3)
+    assert broker._receive_prefix(stream) == b"prefix"
+    assert stream.tell() == 3
+
+
+@pytest.mark.parametrize(
+    ("body", "match"),
+    [
+        (b"bad!", "framing"),
+        (b"0000", "no ref command"),
+        (b"0003", "framing"),
+        (b"0008abc", "framing"),
+        (b"0008a b\n0000", "ref command"),
+    ],
+)
+def test_receive_pack_rejects_bad_framing_and_commands(body, match):
+    broker = _load("scm_broker")
+    with pytest.raises(broker.PolicyError, match=match):
+        broker._validate_receive_pack(body, "sentinel")
+
+
+def _packet(old: bytes, new: bytes, ref: bytes, *, terminator=True) -> bytes:
+    command = old + b" " + new + b" " + ref + b"\n"
+    framed = f"{len(command) + 4:04x}".encode() + command
+    return framed + (b"0000" if terminator else b"")
+
+
+def test_receive_pack_rejects_nonascii_many_commands_and_missing_terminator(
+    monkeypatch,
+):
+    broker = _load("scm_broker")
+    zero = b"0" * 40
+    commit = b"1" * 40
+    with pytest.raises(broker.PolicyError, match="canonical ASCII"):
+        broker._validate_receive_pack(
+            _packet(zero, commit, b"refs/heads/hermes/\xff"), "sentinel"
+        )
+    command = _packet(zero, commit, b"refs/heads/hermes/fix", terminator=False)
+    with pytest.raises(broker.PolicyError, match="terminator"):
+        broker._validate_receive_pack(command, "sentinel")
+    many = command * (broker.MAX_PUSH_COMMANDS + 1) + b"0000"
+    with pytest.raises(broker.PolicyError, match="too many"):
+        broker._validate_receive_pack(many, "sentinel")
+
+
+def test_upstream_request_infers_byte_length_and_requires_stream_length():
+    broker = _load("scm_broker")
+    seen = []
+
+    def opener(request, timeout):
+        seen.append((request, timeout))
+        return Response(b"result", content_type="application/x-git-upload-pack-result")
+
+    assert (
+        broker._upstream_git_request(
+            "/atlas/cassandra.git/git-upload-pack",
+            method="POST",
+            body=b"request",
+            content_type=None,
+            expected_type="application/x-git-upload-pack-result",
+            token="sentinel",
+            opener=opener,
+        )
+        == b"result"
+    )
+    assert seen[0][0].get_header("Content-length") == "7"
+    with pytest.raises(broker.PolicyError, match="length is missing"):
+        broker._upstream_git_request(
+            "/atlas/cassandra.git/git-upload-pack",
+            method="POST",
+            body=io.BytesIO(b"request"),
+            content_type=None,
+            expected_type="application/x-git-upload-pack-result",
+            token="sentinel",
+            opener=opener,
+        )
+
+
+def test_upstream_request_rejects_wrong_type_and_supports_getcode_status():
+    broker = _load("scm_broker")
+
+    class CodeOnly(Response):
+        def __init__(self, body, content_type):
+            super().__init__(body, content_type=content_type)
+            del self.status
+
+        def getcode(self):
+            return 200
+
+    assert (
+        broker._upstream_git_request(
+            "/atlas/cassandra.git/info/refs?service=git-upload-pack",
+            method="GET",
+            body=None,
+            content_type=None,
+            expected_type="application/x-git-upload-pack-advertisement",
+            token="sentinel",
+            opener=lambda *_a, **_k: CodeOnly(
+                b"advertisement", "application/x-git-upload-pack-advertisement"
+            ),
+        )
+        == b"advertisement"
+    )
+    with pytest.raises(broker.PolicyError, match="response type"):
+        broker._upstream_git_request(
+            "/atlas/cassandra.git/info/refs?service=git-upload-pack",
+            method="GET",
+            body=None,
+            content_type=None,
+            expected_type="application/x-git-upload-pack-advertisement",
+            token="sentinel",
+            opener=lambda *_a, **_k: Response(b"bad", content_type="text/plain"),
+        )

testing/tests/test_hermes_scm_server_coverage.py

@@ -0,0 +1,180 @@
+"""Behavioral branch coverage for the SCM broker HTTP server primitives."""
+
+from __future__ import annotations
+
+import io
+
+import pytest
+
+from testing.tests.test_hermes_scm_broker_support import _load
+
+
+class _Connection:
+    def __init__(self):
+        self.timeouts = []
+
+    def settimeout(self, value):
+        self.timeouts.append(value)
+
+
+class _FlushBuffer(io.BytesIO):
+    def __init__(self):
+        super().__init__()
+        self.flushes = 0
+
+    def flush(self):
+        self.flushes += 1
+
+
+def test_absolute_reader_supports_idle_limit_newline_eof_and_attribute_delegation():
+    module = _load("scm_broker_server")
+    stream = io.BytesIO(b"first\nsecond")
+    reader = module._AbsoluteDeadlineReader(stream, _Connection())
+    assert reader.readline(3) == b"fir"
+    reader.begin(5)
+    assert reader.readline() == b"st\n"
+    assert reader.readline() == b"second"
+    assert reader.readline() == b""
+    assert reader.closed is False
+    reader.end()
+    assert reader.readline() == b""
+
+
+def _handler_type(module):
+    class StubHandler:
+        request_version = "HTTP/1.1"
+        command = "GET"
+
+        def setup(self):
+            self.connection = _Connection()
+            self.rfile = io.BytesIO(b"")
+            self.wfile = _FlushBuffer()
+
+        def parse_request(self):
+            return self.parse_result
+
+        def send_error(self, code, *_args):
+            self.errors.append(code)
+
+        def do_GET(self):
+            self.calls.append("GET")
+
+    return type(
+        "DeadlineHandler", (module.AbsoluteHeaderDeadlineMixin, StubHandler), {}
+    )
+
+
+def _handler(module, raw: bytes, *, parse=True):
+    handler_type = _handler_type(module)
+    handler = object.__new__(handler_type)
+    handler.setup()
+    handler._header_reader._stream = io.BytesIO(raw)
+    handler.parse_result = parse
+    handler.errors = []
+    handler.calls = []
+    return handler
+
+
+def test_header_mixin_setup_and_known_request_dispatch():
+    module = _load("scm_broker_server")
+    handler = _handler(module, b"GET /healthz HTTP/1.1\r\n")
+    handler.handle_one_request()
+    assert handler.calls == ["GET"]
+    assert handler.wfile.flushes == 1
+    assert handler._header_reader._deadline is None
+
+
+@pytest.mark.parametrize(
+    ("raw", "parse", "command", "expected_error", "closed"),
+    [
+        (b"x" * 65537, True, "GET", 414, False),
+        (b"", True, "GET", None, True),
+        (b"GET / HTTP/1.1\r\n", False, "GET", None, False),
+        (b"TRACE / HTTP/1.1\r\n", True, "TRACE", 501, False),
+    ],
+)
+def test_header_mixin_rejects_long_empty_unparsed_and_unknown_requests(
+    raw, parse, command, expected_error, closed
+):
+    module = _load("scm_broker_server")
+    handler = _handler(module, raw, parse=parse)
+    handler.command = command
+    handler.close_connection = False
+    handler.handle_one_request()
+    assert handler.errors == ([] if expected_error is None else [expected_error])
+    assert handler.close_connection is closed
+
+
+def test_header_mixin_closes_on_absolute_timeout(monkeypatch):
+    module = _load("scm_broker_server")
+    handler = _handler(module, b"GET / HTTP/1.1\r\n")
+    handler.close_connection = False
+    monkeypatch.setattr(
+        handler._header_reader,
+        "readline",
+        lambda *_a, **_k: (_ for _ in ()).throw(TimeoutError()),
+    )
+    handler.handle_one_request()
+    assert handler.close_connection is True
+
+
+class _Slots:
+    def __init__(self, available=True):
+        self.available = available
+        self.releases = 0
+
+    def acquire(self, *, blocking):
+        assert blocking is False
+        return self.available
+
+    def release(self):
+        self.releases += 1
+
+
+def _server_shell(module, available=True):
+    server = object.__new__(module.BoundedThreadingHTTPServer)
+    server._slots = _Slots(available)
+    server.shutdowns = []
+    server.shutdown_request = server.shutdowns.append
+    return server
+
+
+def test_bounded_server_delegates_accepted_request(monkeypatch):
+    module = _load("scm_broker_server")
+    server = _server_shell(module)
+    calls = []
+    monkeypatch.setattr(
+        module.ThreadingHTTPServer,
+        "process_request",
+        lambda self, request, address: calls.append((request, address)),
+    )
+    server.process_request("socket", ("127.0.0.1", 1))
+    assert calls == [("socket", ("127.0.0.1", 1))]
+    assert server._slots.releases == 0
+
+
+def test_bounded_server_releases_on_dispatch_failure(monkeypatch):
+    module = _load("scm_broker_server")
+    server = _server_shell(module)
+    monkeypatch.setattr(
+        module.ThreadingHTTPServer,
+        "process_request",
+        lambda *_a, **_k: (_ for _ in ()).throw(RuntimeError("dispatch")),
+    )
+    with pytest.raises(RuntimeError, match="dispatch"):
+        server.process_request("socket", ("127.0.0.1", 1))
+    assert server._slots.releases == 1
+    assert server.shutdowns == ["socket"]
+
+
+def test_bounded_server_thread_always_releases(monkeypatch):
+    module = _load("scm_broker_server")
+    server = _server_shell(module)
+    monkeypatch.setattr(
+        module.ThreadingHTTPServer,
+        "process_request_thread",
+        lambda *_a, **_k: (_ for _ in ()).throw(RuntimeError("handler")),
+    )
+    with pytest.raises(RuntimeError, match="handler"):
+        server.process_request_thread("socket", ("127.0.0.1", 1))
+    assert server._slots.releases == 1

testing/tests/test_quality_coverage_helpers.py

@@ -12,10 +12,15 @@ def test_compute_workspace_line_coverage_handles_missing_xml(tmp_path: Path) ->
     """Missing coverage XML should produce a zero workspace coverage score."""
 
     contract = {"coverage": {"tracked_files": ["managed.py"]}}
-    assert compute_workspace_line_coverage(contract, tmp_path, tmp_path / "missing.xml") == 0.0
+    assert (
+        compute_workspace_line_coverage(contract, tmp_path, tmp_path / "missing.xml")
+        == 0.0
+    )
 
 
-def test_compute_workspace_line_coverage_averages_present_tracked_files(tmp_path: Path) -> None:
+def test_compute_workspace_line_coverage_averages_present_tracked_files(
+    tmp_path: Path,
+) -> None:
     """Workspace coverage should average only tracked files that appear in the report."""
 
     coverage_xml = tmp_path / "coverage.xml"
@@ -26,8 +31,8 @@ def test_compute_workspace_line_coverage_averages_present_tracked_files(tmp_path
               <packages>
                 <package>
                   <classes>
-                    <class filename="alpha.py" line-rate="1.0" />
+                    <class filename="alpha.py" line-rate="1.0" branch-rate="1.0" />
-                    <class filename="beta.py" line-rate="0.5" />
+                    <class filename="beta.py" line-rate="0.5" branch-rate="0.5" />
                   </classes>
                 </package>
               </packages>
@@ -41,7 +46,9 @@ def test_compute_workspace_line_coverage_averages_present_tracked_files(tmp_path
     assert compute_workspace_line_coverage(contract, tmp_path, coverage_xml) == 75.0
 
 
-def test_run_check_keeps_relative_names_when_source_roots_do_not_match(tmp_path: Path) -> None:
+def test_run_check_keeps_relative_names_when_source_roots_do_not_match(
+    tmp_path: Path,
+) -> None:
     """Relative filenames should remain relative when no declared source root contains them."""
 
     coverage_xml = tmp_path / "coverage.xml"
@@ -74,3 +81,73 @@ def test_run_check_keeps_relative_names_when_source_roots_do_not_match(tmp_path:
     )
 
     assert issues == ["coverage below 95.0%: relative.py (80.0%)"]
+
+
+def test_run_check_enforces_branch_floor_and_requires_branch_data(
+    tmp_path: Path,
+) -> None:
+    """Tracked modules must report and meet the configured branch floor."""
+
+    coverage_xml = tmp_path / "coverage.xml"
+    coverage_xml.write_text(
+        textwrap.dedent(
+            """\
+            <coverage>
+              <packages>
+                <package>
+                  <classes>
+                    <class filename="low.py" line-rate="1.0" branch-rate="0.90" />
+                    <class filename="missing.py" line-rate="1.0" />
+                  </classes>
+                </package>
+              </packages>
+            </coverage>
+            """
+        ),
+        encoding="utf-8",
+    )
+
+    issues = run_check(
+        {
+            "coverage": {
+                "minimum_percent": 95.0,
+                "minimum_branch_percent": 95.0,
+                "tracked_files": ["low.py", "missing.py"],
+            }
+        },
+        tmp_path,
+        coverage_xml,
+    )
+
+    assert issues == [
+        "branch coverage below 95.0%: low.py (90.0%)",
+        "branch coverage missing for tracked file: missing.py",
+    ]
+
+
+def test_run_check_limits_branch_floor_to_explicit_branch_tracked_files(
+    tmp_path: Path,
+) -> None:
+    """A branch rollout can cover selected security modules without hiding lines."""
+
+    coverage_xml = tmp_path / "coverage.xml"
+    coverage_xml.write_text(
+        """<coverage><packages><package><classes>
+        <class filename="new.py" line-rate="1.0" branch-rate="1.0" />
+        <class filename="legacy.py" line-rate="1.0" branch-rate="0.5" />
+        </classes></package></packages></coverage>""",
+        encoding="utf-8",
+    )
+    issues = run_check(
+        {
+            "coverage": {
+                "minimum_percent": 95,
+                "minimum_branch_percent": 95,
+                "tracked_files": ["new.py", "legacy.py"],
+                "branch_tracked_files": ["new.py"],
+            }
+        },
+        tmp_path,
+        coverage_xml,
+    )
+    assert issues == []