diff --git a/services/hermes/agent-deployment.yaml b/services/hermes/agent-deployment.yaml index 5ed758cf..57611a51 100644 --- a/services/hermes/agent-deployment.yaml +++ b/services/hermes/agent-deployment.yaml @@ -875,6 +875,7 @@ spec: - {name: KUBECONFIG, value: /opt/data/home/.kube/config} - {name: PYTHONPATH, value: /opt/hermes} - {name: HERMES_CLI_LANE_CONCURRENCY, value: "2"} + - {name: HERMES_CLI_HEALTH_MAX_AGE_SECONDS, value: "60"} - {name: HERMES_AUTO_ROUTER_PROFILE, value: agent} - {name: PATH, value: /opt/coordinator:/opt/data/tools/bin:/opt/data/home/.local/bin:/opt/hermes/.venv/bin:/usr/local/bin:/usr/bin:/bin} securityContext: @@ -895,6 +896,15 @@ spec: - {name: kubeconfig, mountPath: /opt/data/home/.kube/config, subPath: config, readOnly: true} - {name: auto-router-plugin, mountPath: /opt/data/plugins/auto-router, readOnly: true} - {name: tmp, mountPath: /tmp} + readinessProbe: + exec: + command: + - /opt/hermes/.venv/bin/python + - /opt/coordinator/cli_lane_capabilities.py + initialDelaySeconds: 2 + periodSeconds: 5 + timeoutSeconds: 2 + failureThreshold: 1 resources: requests: {cpu: 100m, memory: 256Mi} limits: {cpu: "2", memory: 6Gi} diff --git a/services/hermes/scripts/cli_lane_capabilities.py b/services/hermes/scripts/cli_lane_capabilities.py index c54c8bb0..6db205a6 100644 --- a/services/hermes/scripts/cli_lane_capabilities.py +++ b/services/hermes/scripts/cli_lane_capabilities.py @@ -1,11 +1,38 @@ #!/usr/bin/env python3 -"""Detect and publish the Kanban API boundary required by CLI lanes.""" +"""Detect and publish the Kanban API boundary required by CLI lanes. + +During a rollout the coordinator container and the cli-lane-runner worker can +run different hermes-agent image generations against one shared Kanban DB. +The image APIs that differ on the decomposition/finalization path: + +* ``complete_task``: both generations accept ``expected_run_id`` (active + exact-run completion); only the patched image accepts + ``replay_ended_run_id`` (ended-run replay). +* ``reclaim_task``: only the patched image accepts ``expected_run_id`` + (guarded orphan reclaim after a runner restart). +* ``specify_triage_task``/``decompose_triage_task``: only the patched image + accepts ``require_no_runs`` (refuses redefining tasks with run history). + +Worker-side handling per combination (proven in +testing/tests/test_hermes_cli_capabilities.py): + +* legacy worker: health/readiness stay deferred, no new claims, no unguarded + reclaim, replay-needing results stay journaled for a patched successor. +* patched worker: ready health, exact completion, guarded replay/reclaim; a + run voided by a legacy coordinator's unguarded mid-run decomposition + converges to stale/conflict evidence instead of completing. +* patched coordinator: mid-run decomposition is refused entirely, so the + active exact run completes on either worker generation. +""" from __future__ import annotations import inspect +import json +import os import sys from dataclasses import asdict, dataclass +from datetime import datetime, timezone from pathlib import Path from typing import Any, Callable @@ -85,6 +112,7 @@ def _health_document(capabilities: KanbanCapabilities) -> dict[str, Any]: """Return a bounded operator-readable compatibility health document.""" return { "component": "cli-lane-runner", + "schema_version": 1, "state": "ready" if capabilities.ready else "deferred", "ready": capabilities.ready, "active_run_completion_safe": capabilities.exact_run_completion, @@ -117,6 +145,21 @@ def initialize_kanban_capabilities( return capabilities +def refresh_kanban_capabilities( + kanban_db: Any, + *, + health_path: Path | None = None, +) -> KanbanCapabilities: + """Reinspect the live callables and refresh bounded readiness evidence. + + ConfigMap script updates and image changes are deliberately not imported or + reloaded here. A running process observes only the already-loaded module, + which prevents a half-old/half-new Python module graph. Tests may replace + the two callables to model an image API transition at a loop boundary. + """ + return initialize_kanban_capabilities(kanban_db, health_path=health_path) + + def kanban_capabilities(kanban_db: Any) -> KanbanCapabilities: """Return startup evidence, redetecting only when the loaded API changes.""" if _CAPABILITIES is None or _source_identity(kanban_db) != _CAPABILITY_SOURCE: @@ -128,3 +171,68 @@ def runtime_health(path: Path | None = None) -> dict[str, Any]: """Read the bounded health document for probes and operator diagnostics.""" document = load_json(path or STATE_ROOT / "runtime-health.json") return document if isinstance(document, dict) else {} + + +def readiness_issue( + path: Path | None = None, + *, + now: datetime | None = None, + maximum_age_seconds: float = 60.0, +) -> str | None: + """Return a bounded readiness reason without exposing health contents.""" + destination = path or STATE_ROOT / "runtime-health.json" + try: + document = json.loads(destination.read_text(encoding="utf-8")) + except FileNotFoundError: + return "runtime health is missing" + except (OSError, UnicodeError, json.JSONDecodeError): + return "runtime health is malformed" + if not isinstance(document, dict) or document.get("schema_version") != 1: + return "runtime health is malformed" + if document.get("state") == "deferred" and document.get("ready") is False: + return "runtime compatibility is deferred" + capabilities = document.get("capabilities") + if ( + document.get("state") != "ready" + or document.get("ready") is not True + or not isinstance(capabilities, dict) + or any( + capabilities.get(name) is not True + for name in ( + "exact_run_completion", + "ended_run_replay", + "exact_run_reclaim", + ) + ) + ): + return "runtime health is malformed" + observed_at = document.get("observed_at") + if not isinstance(observed_at, str): + return "runtime health is malformed" + try: + observed = datetime.fromisoformat(observed_at.replace("Z", "+00:00")) + except ValueError: + return "runtime health is malformed" + if observed.tzinfo is None: + return "runtime health is malformed" + current = now or datetime.now(timezone.utc) + age = ( + current.astimezone(timezone.utc) - observed.astimezone(timezone.utc) + ).total_seconds() + if age < -5.0 or age > maximum_age_seconds: + return "runtime health is stale" + return None + + +def main() -> int: + """Provide the non-mutating Kubernetes readiness command.""" + maximum_age = float(os.environ.get("HERMES_CLI_HEALTH_MAX_AGE_SECONDS", "60")) + issue = readiness_issue(maximum_age_seconds=maximum_age) + if issue: + print(issue, file=sys.stderr) + return 1 + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/services/hermes/scripts/cli_lane_dispatch.py b/services/hermes/scripts/cli_lane_dispatch.py index 8a740dbf..3ed56c2f 100644 --- a/services/hermes/scripts/cli_lane_dispatch.py +++ b/services/hermes/scripts/cli_lane_dispatch.py @@ -10,7 +10,11 @@ import time from typing import Any from cli_lane_board import _external, _record_board_access_error, _task_value -from cli_lane_capabilities import initialize_kanban_capabilities, kanban_capabilities +from cli_lane_capabilities import ( + initialize_kanban_capabilities, + kanban_capabilities, + refresh_kanban_capabilities, +) from cli_lane_config import ( BOARD_CORRUPTION_ERRORS, DEFAULT_CLAIM_TTL, @@ -29,6 +33,7 @@ def _board_slug(board: Any) -> str: return str(board.get("slug") or board.get("id") or "") return str(getattr(board, "slug", None) or getattr(board, "id", None) or board) + def _connect_healthy_board(kanban_db: Any, board: str) -> Any | None: """Open one board without letting localized storage faults stop other lanes.""" try: @@ -37,6 +42,7 @@ def _connect_healthy_board(kanban_db: Any, board: str) -> Any | None: _record_board_access_error(board, error) return None + def recover_orphans() -> None: """Return external running tasks to ready after a runner/pod restart.""" from hermes_cli import kanban_db @@ -59,7 +65,10 @@ def recover_orphans() -> None: continue try: for task in kanban_db.list_tasks(conn): - if _external(task) and str(_task_value(task, "status", "")) == "running": + if ( + _external(task) + and str(_task_value(task, "status", "")) == "running" + ): task_id = str(_task_value(task, "id")) run_id = _task_value(task, "current_run_id", None) if not isinstance(run_id, int): @@ -78,6 +87,7 @@ def recover_orphans() -> None: finally: conn.close() + def claim_ready(active: set[tuple[str, str]], limit: int) -> list[tuple[str, str]]: """Atomically claim external ready tasks across all non-archived boards.""" from hermes_cli import kanban_db @@ -134,8 +144,13 @@ def claim_ready(active: set[tuple[str, str]], limit: int) -> list[tuple[str, str conn.close() return claimed + def main() -> int: - """Continuously bridge external Kanban lanes to provider CLIs.""" + """Continuously bridge external Kanban lanes to provider CLIs. + + The loop is exited only by process signals or unrecoverable exceptions; + every per-board and per-claim fault is bounded inside one pass. + """ from hermes_cli import kanban_db RESULT_SCHEMA_PATH.parent.mkdir(parents=True, exist_ok=True) @@ -146,12 +161,20 @@ def main() -> int: futures: dict[concurrent.futures.Future[None], tuple[str, str]] = {} with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as pool: while True: + previous_capabilities = capabilities + capabilities = refresh_kanban_capabilities(kanban_db) + if capabilities.ready and not previous_capabilities.ready: + recover_orphans() for future in list(futures): if future.done(): try: future.result() except Exception as error: - print(f"worker future failed: {error}", file=sys.stderr, flush=True) + print( + f"worker future failed: {error}", + file=sys.stderr, + flush=True, + ) del futures[future] recover_pending_finalizations() maybe_gc_lane_artifacts() @@ -170,4 +193,3 @@ def main() -> int: future = pool.submit(execute_claim, board, task_id) futures[future] = (board, task_id) time.sleep(5) - return 0 diff --git a/services/hermes/scripts/cli_lane_runner.py b/services/hermes/scripts/cli_lane_runner.py index ed32e3b7..84aa185d 100644 --- a/services/hermes/scripts/cli_lane_runner.py +++ b/services/hermes/scripts/cli_lane_runner.py @@ -18,6 +18,8 @@ from cli_lane_capabilities import ( detect_kanban_capabilities, initialize_kanban_capabilities, kanban_capabilities, + readiness_issue, + refresh_kanban_capabilities, runtime_health, ) from cli_lane_board import ( diff --git a/testing/quality_coverage.py b/testing/quality_coverage.py index 78d6649f..7339baa1 100644 --- a/testing/quality_coverage.py +++ b/testing/quality_coverage.py @@ -3,24 +3,42 @@ from __future__ import annotations import xml.etree.ElementTree as ET +import math +from dataclasses import dataclass from pathlib import Path from typing import Any -def _load_percentages(xml_path: Path, root: Path) -> dict[str, float]: - """Load per-file line-rate percentages from a Cobertura XML report.""" +@dataclass(frozen=True) +class CoverageRates: + """Validated line and branch percentages for one source file.""" + + line: float + branch: float + + +def _percentage(class_node: ET.Element, attribute: str) -> float: + """Parse one finite Cobertura rate and fail closed outside its domain.""" + raw = class_node.attrib.get(attribute) + if raw is None: + raise ValueError(f"coverage class missing {attribute}") + value = float(raw) + if not math.isfinite(value) or value < 0.0 or value > 1.0: + raise ValueError(f"coverage class has invalid {attribute}: {raw}") + return value * 100.0 + + +def _load_rates(xml_path: Path, root: Path) -> dict[str, CoverageRates]: + """Load validated per-file line and branch rates from Cobertura XML.""" tree = ET.parse(xml_path) xml_root = tree.getroot() source_roots = [ - Path(node.text) - for node in xml_root.findall("./sources/source") - if node.text + Path(node.text) for node in xml_root.findall("./sources/source") if node.text ] - percentages: dict[str, float] = {} + rates: 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") - if not filename or line_rate is None: + if not filename: continue normalized = filename.replace("\\", "/") if normalized.startswith("/"): @@ -32,8 +50,16 @@ 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 - return percentages + rates[key] = CoverageRates( + line=_percentage(class_node, "line-rate"), + branch=_percentage(class_node, "branch-rate"), + ) + return rates + + +def _load_percentages(xml_path: Path, root: Path) -> dict[str, float]: + """Load per-file line percentages for backward-compatible callers.""" + return {path: rates.line for path, rates in _load_rates(xml_path, root).items()} def run_check(contract: dict[str, Any], root: Path, xml_path: Path) -> list[str]: @@ -45,19 +71,28 @@ def run_check(contract: dict[str, Any], root: Path, xml_path: Path) -> list[str] if not xml_path.exists(): return [f"coverage xml missing: {xml_path.relative_to(root)}"] - percentages = _load_percentages(xml_path, root) + try: + rates_by_path = _load_rates(xml_path, root) + except (ET.ParseError, OSError, UnicodeError, ValueError) as error: + return [f"coverage xml invalid: {error}"] minimum = float(contract.get("coverage", {}).get("minimum_percent", 95.0)) issues: list[str] = [] for relative_path in contract.get("coverage", {}).get("tracked_files", []): normalized = relative_path.replace("\\", "/") - percent = percentages.get(normalized) - if percent is None: + rates = rates_by_path.get(normalized) + 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"line coverage below {minimum:.1f}%: {relative_path} " + f"({rates.line:.1f}%)" + ) + if rates.branch + 1e-9 < minimum: + issues.append( + f"branch coverage below {minimum:.1f}%: {relative_path} " + f"({rates.branch:.1f}%)" ) return issues @@ -73,7 +108,10 @@ def compute_workspace_line_coverage( if not xml_path.exists(): return 0.0 - percentages = _load_percentages(xml_path, root) + try: + percentages = _load_percentages(xml_path, root) + except (ET.ParseError, OSError, UnicodeError, ValueError): + return 0.0 samples: list[float] = [] for relative_path in contract.get("coverage", {}).get("tracked_files", []): normalized = relative_path.replace("\\", "/") diff --git a/testing/tests/test_hermes_agent_layout.py b/testing/tests/test_hermes_agent_layout.py index 2e2ef65e..72ab7f05 100644 --- a/testing/tests/test_hermes_agent_layout.py +++ b/testing/tests/test_hermes_agent_layout.py @@ -36,7 +36,9 @@ def test_claude_pretool_hook_allows_normal_engineering(): assert policy.denial_reason("vault kv get kv/atlas/hermes") is None -def test_claude_settings_preserve_state_and_install_three_guardrail_layers(tmp_path: Path): +def test_claude_settings_preserve_state_and_install_three_guardrail_layers( + tmp_path: Path, +): state = tmp_path / ".claude.json" settings = tmp_path / "settings.json" state.write_text('{"promptQueueUseCount": 4}\n', encoding="utf-8") @@ -78,7 +80,9 @@ def test_claude_settings_preserve_state_and_install_three_guardrail_layers(tmp_p def test_legacy_state_is_archived_without_removing_provider_transcripts(tmp_path: Path): session = tmp_path / "home/.config/herdr/session.json" session.parent.mkdir(parents=True) - session.write_text('{"agents":[{"agent":"claude","session_id":"abc"}]}', encoding="utf-8") + session.write_text( + '{"agents":[{"agent":"claude","session_id":"abc"}]}', encoding="utf-8" + ) binary = tmp_path / "tools/bin/herdr" binary.parent.mkdir(parents=True) binary.write_text("legacy", encoding="utf-8") @@ -156,9 +160,7 @@ def test_agent_avoids_unhealthy_nodes_and_fits_its_remaining_capacity(): assert hostnames["operator"] == "NotIn" assert set(hostnames["values"]) >= {"titan-04", "titan-19"} - hermes = next( - item for item in pod["containers"] if item["name"] == "hermes" - ) + hermes = next(item for item in pod["containers"] if item["name"] == "hermes") assert hermes["resources"]["requests"]["cpu"] == "300m" @@ -223,8 +225,7 @@ def test_agent_root_is_stock_dashboard_and_terminal_is_a_separate_path(): ) patch_init = next( - item for item in pod["initContainers"] - if item["name"] == "patch-tui-gateway" + item for item in pod["initContainers"] if item["name"] == "patch-tui-gateway" ) assert patch_init["command"][-1] == "/patched/server.py" for name in ("hermes", "terminal"): @@ -248,20 +249,29 @@ def test_agent_root_is_stock_dashboard_and_terminal_is_a_separate_path(): assert middlewares["hermes-agent-terminal-slash"]["spec"]["redirectRegex"][ "replacement" ].endswith("/terminal/") - assert middlewares["hermes-agent-stock-dashboard-headers"]["spec"]["headers"][ - "customRequestHeaders" - ]["Origin"] == "http://127.0.0.1:9119" + assert ( + middlewares["hermes-agent-stock-dashboard-headers"]["spec"]["headers"][ + "customRequestHeaders" + ]["Origin"] + == "http://127.0.0.1:9119" + ) ingresses = { item["metadata"]["name"]: item for item in ingress_documents if item["kind"] == "Ingress" } - assert ingresses["hermes-agent-dashboard"]["metadata"]["annotations"][ - "traefik.ingress.kubernetes.io/router.middlewares" - ] == "hermes-hermes-agent-stock-dashboard-headers@kubernetescrd" - assert ingresses["hermes-agent-terminal"]["metadata"]["annotations"][ - "traefik.ingress.kubernetes.io/router.middlewares" - ] == "hermes-hermes-agent-terminal-slash@kubernetescrd" + assert ( + ingresses["hermes-agent-dashboard"]["metadata"]["annotations"][ + "traefik.ingress.kubernetes.io/router.middlewares" + ] + == "hermes-hermes-agent-stock-dashboard-headers@kubernetescrd" + ) + assert ( + ingresses["hermes-agent-terminal"]["metadata"]["annotations"][ + "traefik.ingress.kubernetes.io/router.middlewares" + ] + == "hermes-hermes-agent-terminal-slash@kubernetescrd" + ) def test_broker_services_survive_sibling_container_readiness_loss(): @@ -322,9 +332,7 @@ def test_cli_lane_config_refresh_does_not_restart_active_work(): for item in deployment["spec"]["template"]["spec"]["containers"] if item["name"] == "cli-lane-runner" ) - mount = next( - item for item in lane["volumeMounts"] if item["name"] == "coordinator" - ) + mount = next(item for item in lane["volumeMounts"] if item["name"] == "coordinator") assert coordinator["options"]["disableNameSuffixHash"] is True assert mount == { @@ -335,12 +343,28 @@ def test_cli_lane_config_refresh_does_not_restart_active_work(): assert "checksum/hermes-coordinator" not in deployment["spec"]["template"].get( "metadata", {} ).get("annotations", {}) + assert "startupProbe" not in lane + assert "livenessProbe" not in lane + assert lane["readinessProbe"] == { + "exec": { + "command": [ + "/opt/hermes/.venv/bin/python", + "/opt/coordinator/cli_lane_capabilities.py", + ] + }, + "initialDelaySeconds": 2, + "periodSeconds": 5, + "timeoutSeconds": 2, + "failureThreshold": 1, + } + environment = {item["name"]: item["value"] for item in lane["env"]} + assert environment["HERMES_CLI_HEALTH_MAX_AGE_SECONDS"] == "60" def test_agent_dashboard_reconnects_all_transient_websockets(): - dockerfile = ( - HERMES.parents[1] / "dockerfiles/Dockerfile.hermes-agent" - ).read_text(encoding="utf-8") + dockerfile = (HERMES.parents[1] / "dockerfiles/Dockerfile.hermes-agent").read_text( + encoding="utf-8" + ) assert "eventsRetryAttempt.current" in dockerfile assert "if (!unmounting) setVersion((v) => v + 1);" in dockerfile assert "events feed rejected (${ev.code}) — reload the page" in dockerfile diff --git a/testing/tests/test_hermes_cli_capabilities.py b/testing/tests/test_hermes_cli_capabilities.py index adc6e3ca..fd243267 100644 --- a/testing/tests/test_hermes_cli_capabilities.py +++ b/testing/tests/test_hermes_cli_capabilities.py @@ -3,14 +3,16 @@ from __future__ import annotations import json +import runpy import sys from contextlib import nullcontext +from datetime import datetime, timedelta, timezone from pathlib import Path from types import SimpleNamespace import pytest -from testing.tests.test_hermes_cli_support import _completed_result, lanes +from testing.tests.test_hermes_cli_support import SCRIPTS, _completed_result, lanes def _old_complete( @@ -94,6 +96,18 @@ def test_capability_detection_requires_explicit_safety_keywords(): assert lanes._explicit_keyword(object(), "expected_run_id") is False +def test_deferred_features_name_every_absent_capability(): + """A fully legacy image defers all three safety features by exact name.""" + fully_legacy = lanes.KanbanCapabilities(False, False, False) + + assert fully_legacy.ready is False + assert fully_legacy.deferred_features == ( + "exact-run-completion", + "ended-run-replay", + "exact-run-reclaim", + ) + + def test_startup_health_moves_from_bounded_deferred_to_ready( tmp_path: Path, capsys, @@ -106,6 +120,7 @@ def test_startup_health_moves_from_bounded_deferred_to_ready( assert old.ready is False assert old_document["state"] == "deferred" + assert old_document["schema_version"] == 1 assert old_document["ready"] is False assert old_document["active_run_completion_safe"] is True assert old_document["deferred_features"] == [ @@ -123,6 +138,107 @@ def test_startup_health_moves_from_bounded_deferred_to_ready( assert lanes.runtime_health(health)["state"] == "ready" +def test_readiness_fails_closed_without_disclosing_health_values( + tmp_path: Path, + capsys, + monkeypatch, +): + """Readiness distinguishes startup, malformed, deferred, stale, and ready.""" + health = tmp_path / "runtime-health.json" + now = datetime(2026, 8, 17, 12, 0, tzinfo=timezone.utc) + assert lanes.readiness_issue(health, now=now) == "runtime health is missing" + + health.write_text("not-json", encoding="utf-8") + assert lanes.readiness_issue(health, now=now) == "runtime health is malformed" + + deferred = lanes._health_document(lanes.KanbanCapabilities(True, False, False)) + deferred["observed_at"] = now.isoformat() + health.write_text(json.dumps(deferred), encoding="utf-8") + assert lanes.readiness_issue(health, now=now) == "runtime compatibility is deferred" + + ready = lanes._health_document(lanes.KanbanCapabilities(True, True, True)) + ready["observed_at"] = (now - timedelta(seconds=61)).isoformat() + health.write_text(json.dumps(ready), encoding="utf-8") + assert lanes.readiness_issue(health, now=now) == "runtime health is stale" + + ready["observed_at"] = now.isoformat() + health.write_text(json.dumps(ready), encoding="utf-8") + assert lanes.readiness_issue(health, now=now) is None + + monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path) + monkeypatch.setenv("HERMES_CLI_HEALTH_MAX_AGE_SECONDS", "31536000") + assert sys.modules["cli_lane_capabilities"].main() == 0 + assert capsys.readouterr().out == "" + + +@pytest.mark.parametrize( + "mutation", + [ + {"schema_version": 2}, + {"state": "ready", "ready": False}, + {"state": "ready", "capabilities": []}, + {"observed_at": 5}, + {"observed_at": "not-a-date"}, + {"observed_at": "2026-08-17T12:00:00"}, + ], +) +def test_readiness_rejects_inconsistent_documents( + tmp_path: Path, + mutation: dict, +): + """Inconsistent ready documents remain not-ready.""" + now = datetime(2026, 8, 17, 12, 0, tzinfo=timezone.utc) + document = lanes._health_document(lanes.KanbanCapabilities(True, True, True)) + document["observed_at"] = now.isoformat() + document.update(mutation) + health = tmp_path / "runtime-health.json" + health.write_text(json.dumps(document), encoding="utf-8") + + assert lanes.readiness_issue(health, now=now) == "runtime health is malformed" + + +def test_readiness_command_fails_with_only_a_bounded_reason( + tmp_path: Path, + capsys, + monkeypatch, +): + """The probe command exits nonzero and discloses nothing but the reason.""" + monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path) + + assert sys.modules["cli_lane_capabilities"].main() == 1 + + captured = capsys.readouterr() + assert captured.out == "" + assert captured.err.strip() == "runtime health is missing" + + +def test_capabilities_module_is_the_deployed_probe_entrypoint( + tmp_path: Path, + monkeypatch, +): + """`python cli_lane_capabilities.py` runs the readiness probe directly.""" + monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path) + + with pytest.raises(SystemExit) as excinfo: + runpy.run_path( + str(SCRIPTS / "cli_lane_capabilities.py"), + run_name="__main__", + ) + + assert excinfo.value.code == 1 + + +def test_readiness_rejects_future_health(tmp_path: Path): + """A clock-skewed future observation cannot make a stale process ready.""" + now = datetime(2026, 8, 17, 12, 0, tzinfo=timezone.utc) + document = lanes._health_document(lanes.KanbanCapabilities(True, True, True)) + document["observed_at"] = (now + timedelta(seconds=6)).isoformat() + health = tmp_path / "runtime-health.json" + health.write_text(json.dumps(document), encoding="utf-8") + + assert lanes.readiness_issue(health, now=now) == "runtime health is stale" + + @pytest.mark.parametrize( ("status", "run_id", "expected"), [("running", 7, "committed"), ("blocked", None, "deferred")], diff --git a/testing/tests/test_hermes_cli_dispatch_runtime.py b/testing/tests/test_hermes_cli_dispatch_runtime.py index 74684105..06873f0a 100644 --- a/testing/tests/test_hermes_cli_dispatch_runtime.py +++ b/testing/tests/test_hermes_cli_dispatch_runtime.py @@ -2,6 +2,7 @@ from __future__ import annotations +import runpy import sys from contextlib import nullcontext from pathlib import Path @@ -9,7 +10,7 @@ from types import SimpleNamespace import pytest -from testing.tests.test_hermes_cli_support import lanes +from testing.tests.test_hermes_cli_support import SCRIPTS, lanes def test_board_slug_and_connection_failure_paths(monkeypatch): @@ -129,6 +130,28 @@ class _Pool: return self.future +class _HeldFuture: + def done(self): + return False + + +class _RecordingPool: + submissions = [] + + def __init__(self, max_workers): + self.max_workers = max_workers + + def __enter__(self): + return self + + def __exit__(self, *_args): + return False + + def submit(self, _function, board, task_id): + self.submissions.append((board, task_id)) + return _HeldFuture() + + def test_ready_dispatch_loop_submits_and_reaps_failed_workers( tmp_path: Path, monkeypatch, @@ -138,11 +161,9 @@ def test_ready_dispatch_loop_submits_and_reaps_failed_workers( monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=db)) monkeypatch.setattr(lanes, "RESULT_SCHEMA_PATH", tmp_path / "schema.json") monkeypatch.setattr(lanes, "recover_orphans", lambda: None) - monkeypatch.setattr( - lanes, - "initialize_kanban_capabilities", - lambda _db: lanes.KanbanCapabilities(True, True, True), - ) + ready = lanes.KanbanCapabilities(True, True, True) + monkeypatch.setattr(lanes, "initialize_kanban_capabilities", lambda _db: ready) + monkeypatch.setattr(lanes, "refresh_kanban_capabilities", lambda _db: ready) monkeypatch.setattr(lanes, "recover_pending_finalizations", lambda: 0) monkeypatch.setattr(lanes, "maybe_gc_lane_artifacts", lambda: 0) claims = [[("cassandra", "t_loop")], []] @@ -175,11 +196,9 @@ def test_deferred_dispatch_health_never_claims_new_work( monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=db)) monkeypatch.setattr(lanes, "RESULT_SCHEMA_PATH", tmp_path / "schema.json") monkeypatch.setattr(lanes, "recover_orphans", lambda: None) - monkeypatch.setattr( - lanes, - "initialize_kanban_capabilities", - lambda _db: lanes.KanbanCapabilities(True, False, False), - ) + deferred = lanes.KanbanCapabilities(True, False, False) + monkeypatch.setattr(lanes, "initialize_kanban_capabilities", lambda _db: deferred) + monkeypatch.setattr(lanes, "refresh_kanban_capabilities", lambda _db: deferred) monkeypatch.setattr(lanes, "recover_pending_finalizations", lambda: 0) monkeypatch.setattr(lanes, "maybe_gc_lane_artifacts", lambda: 0) monkeypatch.setattr( @@ -195,3 +214,192 @@ def test_deferred_dispatch_health_never_claims_new_work( ) with pytest.raises(RuntimeError, match="stop loop"): lanes.main() + + +@pytest.mark.parametrize("start_patched", [False, True]) +def test_two_loop_api_transition_refreshes_dispatch_and_health( + tmp_path: Path, + monkeypatch, + start_patched: bool, +): + """Each loop observes legacy/patched API transitions without reloading modules.""" + + def old_complete(_conn, _task_id, *, expected_run_id=None): + return expected_run_id is not None + + def new_complete( + _conn, + _task_id, + *, + expected_run_id=None, + replay_ended_run_id=None, + ): + return expected_run_id is not None or replay_ended_run_id is not None + + def old_reclaim(_conn, _task_id, *, reason): + pytest.fail(f"unguarded legacy reclaim invoked: {reason}") + + def new_reclaim(_conn, _task_id, *, reason, expected_run_id=None): + return bool(reason and expected_run_id) + + db = SimpleNamespace( + complete_task=new_complete if start_patched else old_complete, + reclaim_task=new_reclaim if start_patched else old_reclaim, + ) + monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=db)) + monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes") + monkeypatch.setattr(lanes, "RESULT_SCHEMA_PATH", tmp_path / "schema.json") + recoveries = [] + monkeypatch.setattr(lanes, "recover_orphans", lambda: recoveries.append("recover")) + monkeypatch.setattr(lanes, "recover_pending_finalizations", lambda: 0) + monkeypatch.setattr(lanes, "maybe_gc_lane_artifacts", lambda: 0) + claim_states = [] + + def claim(_active, _limit): + ready = lanes.kanban_capabilities(db).ready + claim_states.append(ready) + return [("cassandra", "t_loop")] if ready and not _active else [] + + monkeypatch.setattr(lanes, "claim_ready", claim) + _RecordingPool.submissions = [] + monkeypatch.setattr(lanes.concurrent.futures, "ThreadPoolExecutor", _RecordingPool) + journal = tmp_path / "cli-lanes/cassandra/t_loop.terminal.json" + journal.parent.mkdir(parents=True) + journal.write_text("journal", encoding="utf-8") + sleeps = [] + + def transition_then_stop(_seconds): + sleeps.append(True) + if len(sleeps) == 1: + db.complete_task = old_complete if start_patched else new_complete + db.reclaim_task = old_reclaim if start_patched else new_reclaim + return + raise RuntimeError("two loops complete") + + monkeypatch.setattr(lanes.time, "sleep", transition_then_stop) + + with pytest.raises(RuntimeError, match="two loops complete"): + lanes.main() + + assert claim_states == ([True] if start_patched else [True]) + assert _RecordingPool.submissions == ( + [("cassandra", "t_loop")] if start_patched else [("cassandra", "t_loop")] + ) + assert len(recoveries) == (1 if start_patched else 2) + assert journal.read_text(encoding="utf-8") == "journal" + expected_ready = not start_patched + assert lanes.runtime_health()["ready"] is expected_ready + + +def test_orphan_recovery_skips_unreachable_boards_and_settled_tasks( + monkeypatch, + capsys, +): + """One faulted board or settled task never stops restart reclamation.""" + running = SimpleNamespace( + id="t_run", + status="running", + current_run_id=4, + assignee="cli-auto", + ) + settled = SimpleNamespace( + id="t_done", + status="done", + current_run_id=None, + assignee="cli-auto", + ) + reclaims = [] + + def connect(*, board): + if board == "broken": + raise OSError("volume stall") + return SimpleNamespace(close=lambda: None) + + db = SimpleNamespace( + list_boards=lambda include_archived=False: [ + {"slug": "broken"}, + {"slug": "healthy"}, + ], + scoped_current_board=lambda _board: nullcontext(), + connect=connect, + list_tasks=lambda _conn: [settled, running], + reclaim_task=lambda _conn, task_id, **kwargs: reclaims.append( + (task_id, kwargs["expected_run_id"]) + ), + ) + monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=db)) + monkeypatch.setattr(lanes, "recover_pending_finalizations", lambda: 0) + monkeypatch.setattr(lanes, "_has_pending_finalization", lambda *_args: False) + + lanes.recover_orphans() + + assert reclaims == [("t_run", 4)] + assert "temporarily skipping Kanban board 'broken'" in capsys.readouterr().err + + +def test_claim_scan_skips_foreign_lanes_and_lost_claim_races(monkeypatch): + """Foreign assignees and lost atomic claims never enter dispatch.""" + foreign = SimpleNamespace(id="t_manual", assignee="human", status="ready") + lost = SimpleNamespace(id="t_lost", assignee="cli-auto", status="ready") + won = SimpleNamespace(id="t_won", assignee="cli-auto", status="ready") + + def claim_task(_conn, task_id, **_kwargs): + return None if task_id == "t_lost" else SimpleNamespace(id=task_id) + + db = SimpleNamespace( + list_boards=lambda include_archived=False: [{"slug": "cassandra"}], + scoped_current_board=lambda _board: nullcontext(), + connect=lambda board: SimpleNamespace(close=lambda: None), + recompute_ready=lambda _conn: None, + list_tasks=lambda _conn: [foreign, lost, won], + claim_task=claim_task, + ) + monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=db)) + + assert lanes.claim_ready(set(), 3) == [("cassandra", "t_won")] + + +def test_dispatch_loop_survives_board_registry_scan_failure( + tmp_path: Path, + monkeypatch, + capsys, +): + """A failing registry scan defers claiming without stopping the loop.""" + db = SimpleNamespace() + monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=db)) + monkeypatch.setattr(lanes, "RESULT_SCHEMA_PATH", tmp_path / "schema.json") + monkeypatch.setattr(lanes, "recover_orphans", lambda: None) + ready = lanes.KanbanCapabilities(True, True, True) + monkeypatch.setattr(lanes, "initialize_kanban_capabilities", lambda _db: ready) + monkeypatch.setattr(lanes, "refresh_kanban_capabilities", lambda _db: ready) + monkeypatch.setattr(lanes, "recover_pending_finalizations", lambda: 0) + monkeypatch.setattr(lanes, "maybe_gc_lane_artifacts", lambda: 0) + + def failing_claim(_active, _limit): + raise RuntimeError("registry scan failed") + + monkeypatch.setattr(lanes, "claim_ready", failing_claim) + monkeypatch.setattr( + lanes.time, + "sleep", + lambda _seconds: (_ for _ in ()).throw(RuntimeError("stop loop")), + ) + + with pytest.raises(RuntimeError, match="stop loop"): + lanes.main() + + assert "temporarily skipping Kanban board 'board-registry'" in ( + capsys.readouterr().err + ) + + +def test_runner_module_is_the_deployed_dispatch_entrypoint(monkeypatch): + """`python cli_lane_runner.py` hands control to the dispatch loop.""" + calls = [] + monkeypatch.setattr(lanes, "main", lambda: calls.append(True) or 0) + + with pytest.raises(SystemExit) as excinfo: + runpy.run_path(str(SCRIPTS / "cli_lane_runner.py"), run_name="__main__") + + assert excinfo.value.code == 0 + assert calls == [True] diff --git a/testing/tests/test_quality_contract.py b/testing/tests/test_quality_contract.py index 6e260a1b..59417eff 100644 --- a/testing/tests/test_quality_contract.py +++ b/testing/tests/test_quality_contract.py @@ -98,8 +98,8 @@ def test_coverage_check_enforces_per_file_floor(tmp_path: Path): - - + + @@ -118,7 +118,7 @@ def test_coverage_check_enforces_per_file_floor(tmp_path: Path): issues = run_coverage_check(contract, tmp_path, coverage_xml) - assert "coverage below 95.0%: low.py (90.0%)" in issues + assert "line coverage below 95.0%: low.py (90.0%)" in issues assert "coverage missing for tracked file: missing.py" in issues @@ -142,9 +142,8 @@ def test_coverage_check_handles_missing_xml_and_source_root_mapping(tmp_path: Pa - - - + + diff --git a/testing/tests/test_quality_coverage_helpers.py b/testing/tests/test_quality_coverage_helpers.py index 5eaa624e..beae7d0a 100644 --- a/testing/tests/test_quality_coverage_helpers.py +++ b/testing/tests/test_quality_coverage_helpers.py @@ -5,6 +5,8 @@ from __future__ import annotations import textwrap from pathlib import Path +import pytest + from testing.quality_coverage import compute_workspace_line_coverage, run_check @@ -12,10 +14,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 +33,8 @@ def test_compute_workspace_line_coverage_averages_present_tracked_files(tmp_path - - + + @@ -41,7 +48,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" @@ -57,7 +66,7 @@ def test_run_check_keeps_relative_names_when_source_roots_do_not_match(tmp_path: - + @@ -73,4 +82,68 @@ def test_run_check_keeps_relative_names_when_source_roots_do_not_match(tmp_path: coverage_xml, ) - assert issues == ["coverage below 95.0%: relative.py (80.0%)"] + assert issues == ["line coverage below 95.0%: relative.py (80.0%)"] + + +@pytest.mark.parametrize("branch_rate", [None, "nan", "inf", "-0.1", "1.1", "broken"]) +def test_run_check_rejects_missing_or_invalid_branch_rates( + tmp_path: Path, + branch_rate: str | None, +) -> None: + """Tracked files must carry finite Cobertura branch evidence.""" + attribute = "" if branch_rate is None else f' branch-rate="{branch_rate}"' + coverage_xml = tmp_path / "coverage.xml" + coverage_xml.write_text( + f'', + encoding="utf-8", + ) + + issues = run_check( + {"coverage": {"tracked_files": ["managed.py"]}}, + tmp_path, + coverage_xml, + ) + + assert len(issues) == 1 + assert issues[0].startswith("coverage xml invalid:") + assert ( + compute_workspace_line_coverage( + {"coverage": {"tracked_files": ["managed.py"]}}, + tmp_path, + coverage_xml, + ) + == 0.0 + ) + + +def test_run_check_enforces_line_and_branch_rates_independently(tmp_path: Path) -> None: + """A strong line score must not hide weak branch coverage.""" + coverage_xml = tmp_path / "coverage.xml" + coverage_xml.write_text( + '', + encoding="utf-8", + ) + + assert run_check( + {"coverage": {"minimum_percent": 95, "tracked_files": ["managed.py"]}}, + tmp_path, + coverage_xml, + ) == ["branch coverage below 95.0%: managed.py (50.0%)"] + + +def test_run_check_accepts_cobertura_branchless_file_semantics(tmp_path: Path) -> None: + """Coverage.py reports branchless files with branch-rate one.""" + coverage_xml = tmp_path / "coverage.xml" + coverage_xml.write_text( + '', + encoding="utf-8", + ) + + assert ( + run_check( + {"coverage": {"minimum_percent": 95, "tracked_files": ["managed.py"]}}, + tmp_path, + coverage_xml, + ) + == [] + )