Compare commits

...

2 Commits

Author SHA1 Message Date
jenkins
df75479069 hermes: close lane branch coverage gaps
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-08-17 15:03:59 -03:00
jenkins
e3ecc18d0e hermes: refresh capabilities and readiness per loop
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-08-17 15:03:50 -03:00
26 changed files with 1888 additions and 72 deletions

View File

@ -93,9 +93,8 @@ def _sonar_issue_from_finding(finding: dict[str, Any]) -> dict[str, Any] | None:
start = finding.get("start") if isinstance(finding.get("start"), dict) else {}
end = finding.get("end") if isinstance(finding.get("end"), dict) else {}
start_line = _line_number(start.get("line") if isinstance(start, dict) else None)
# _line_number never returns below its default, so end_line >= start_line.
end_line = _line_number(end.get("line") if isinstance(end, dict) else None, start_line)
if end_line < start_line:
end_line = start_line
return {
"engineId": "semgrep",
"ruleId": str(finding.get("check_id") or "semgrep.unknown"),

View File

@ -0,0 +1,96 @@
"""Retry, attribute-corruption, and skip-path edges for the Mailu sync job."""
from test_mailu_sync import _FakeResponse, _FakeSession, load_sync_module
def test_retry_helpers_with_zero_attempts_do_nothing(monkeypatch):
sync = load_sync_module(monkeypatch)
monkeypatch.setattr(
sync.psycopg2,
"connect",
lambda **kwargs: (_ for _ in ()).throw(AssertionError("must not connect")),
)
assert sync.retry_request(
"request",
lambda: (_ for _ in ()).throw(AssertionError("must not run")),
attempts=0,
) is None
assert sync.retry_db_connect(attempts=0) is None
def test_kc_update_attributes_replaces_a_corrupt_attribute_map(monkeypatch):
sync = load_sync_module(monkeypatch)
current_resp = _FakeResponse({"username": "u1", "attributes": "corrupt"})
ok_resp = _FakeResponse({"attributes": {"mailu_app_password": ["abc"]}})
sync.SESSION = _FakeSession(_FakeResponse({}), [current_resp, ok_resp])
sync.kc_update_attributes(
"token",
{"id": "u1", "username": "u1"},
{"mailu_app_password": "abc"},
)
assert sync.SESSION.put_called and sync.SESSION.get_calls == 2
def test_main_skips_disabled_users_and_settled_attribute_sets(monkeypatch):
sync = load_sync_module(monkeypatch)
monkeypatch.setattr(sync.bcrypt_sha256, "hash", lambda password: f"hash:{password}")
users = [
{
"id": "u_disabled",
"username": "disabled",
"email": "disabled@example.com",
"enabled": False,
"attributes": {"mailu_enabled": ["true"]},
},
{
"id": "u_settled",
"username": "settled",
"email": "settled@example.com",
"attributes": {
"mailu_enabled": ["true"],
"mailu_email": ["settled@example.com"],
"mailu_app_password": ["keepme"],
},
},
]
updated = []
class _Cursor:
def __init__(self):
self.executions = []
def execute(self, sql, params):
self.executions.append(params)
def close(self):
return None
class _Conn:
def __init__(self):
self.autocommit = False
self._cursor = _Cursor()
def cursor(self, cursor_factory=None):
return self._cursor
def close(self):
return None
conn = _Conn()
monkeypatch.setattr(sync, "get_kc_token", lambda: "tok")
monkeypatch.setattr(sync, "kc_get_users", lambda token: users)
monkeypatch.setattr(
sync,
"kc_update_attributes",
lambda token, user, attrs: updated.append(user["id"]),
)
monkeypatch.setattr(sync.psycopg2, "connect", lambda **kwargs: conn)
sync.main()
assert updated == []
assert [params["email"] for params in conn._cursor.executions] == [
"settled@example.com"
]

View File

@ -132,3 +132,47 @@ def test_listener_log_message_is_quiet(monkeypatch):
handler = listener.Handler.__new__(listener.Handler)
assert handler.log_message("ignored %s", "value") is None
def test_listener_post_treats_non_object_payloads_as_plain_triggers(monkeypatch):
listener = load_listener_module(monkeypatch)
called = []
monkeypatch.setattr(
listener,
"_trigger_sync_async",
lambda force=False: called.append(force) or True,
)
handler = _handler_for(listener, "[1, 2, 3]")
handler.do_POST()
assert called == [False]
assert handler.responses == [202]
def test_listener_module_serves_forever_when_run_directly(monkeypatch):
import http.server
import runpy
served = []
class _FakeServer:
def __init__(self, address, handler):
served.append((address, handler.__name__))
def serve_forever(self):
served.append("serving")
monkeypatch.setenv("MAILU_SYNC_WAIT_TIMEOUT_SEC", "0")
monkeypatch.setattr(http.server, "ThreadingHTTPServer", _FakeServer)
module_path = (
pathlib.Path(__file__).resolve().parents[2]
/ "services"
/ "mailu"
/ "scripts"
/ "mailu_sync_listener.py"
)
runpy.run_path(str(module_path), run_name="__main__")
assert served == [(("", 8080), "Handler"), "serving"]

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

@ -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")],

View File

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

View File

@ -192,3 +192,38 @@ def test_retention_failure_paths_are_bounded(tmp_path: Path, monkeypatch, capsys
)
assert lanes.maybe_gc_lane_artifacts(now=1000) == 0
assert "retention deferred" in capsys.readouterr().err
def test_evidence_writer_rejects_a_tampered_temporary(tmp_path: Path, monkeypatch):
"""A shared or replaced temporary inode never becomes durable evidence."""
real_fstat = os.fstat
calls = []
def hardlinked_fstat(descriptor):
observed = real_fstat(descriptor)
calls.append(observed)
if len(calls) == 2:
return os.stat_result((*observed[:3], 2, *observed[4:]))
return observed
monkeypatch.setattr(lanes.os, "fstat", hardlinked_fstat)
with pytest.raises(OSError, match="not private and singly linked"):
lanes._write_json_noreplace(tmp_path / "tampered.json", {"value": "x"})
assert list(tmp_path.iterdir()) == []
def test_evidence_writer_releases_descriptor_when_inspection_fails(
tmp_path: Path,
monkeypatch,
):
"""An fstat fault closes the created descriptor without staging cleanup."""
def failing_fstat(_descriptor):
raise OSError("inspection failed")
monkeypatch.setattr(lanes.os, "fstat", failing_fstat)
with pytest.raises(OSError, match="inspection failed"):
lanes._write_json_noreplace(tmp_path / "orphaned.json", {"value": "x"})

View File

@ -216,3 +216,104 @@ def test_unexpected_route_exception_blocks_the_exact_run(
assert blocks[0]["expected_run_id"] == 23
assert "router down" in blocks[0]["reason"]
def test_capacity_fallback_without_evidence_blocks_with_provider_output(
tmp_path: Path,
monkeypatch,
):
"""A fallback turn with no structured result blocks on raw provider output."""
task = SimpleNamespace(
id="t_nofallback",
status="running",
current_run_id=22,
assignee="cli-auto",
max_runtime_seconds=60,
)
blocks = []
db = SimpleNamespace(
scoped_current_board=lambda _board: nullcontext(),
connect=lambda board: _Connection(),
get_task=lambda *_args: task,
worker_log_path=lambda *_args, **_kwargs: tmp_path / "worker.log",
_resolve_worktree_workspace=lambda *_args, **_kwargs: (tmp_path, "branch"),
set_branch_name=lambda *_args: None,
set_workspace_path=lambda *_args: None,
build_worker_context=lambda *_args: "attempt both providers",
add_comment=lambda *_args, **_kwargs: None,
heartbeat_worker=lambda *_args, **_kwargs: True,
block_task=lambda *_args, **kwargs: blocks.append(kwargs),
)
monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=db))
monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "lanes")
monkeypatch.setattr(lanes, "fresh_unavailable_provider", lambda: None)
codex = lanes.Route("codex", "gpt", "high", "p", "c", "r", 1, ())
claude = lanes.Route("claude", "fable", "high", "p", "c", "r", 1, ())
monkeypatch.setattr(
lanes,
"select_route",
lambda _prompt, assignee, **_kwargs: claude
if assignee == "cli-claude-high"
else codex,
)
monkeypatch.setattr(lanes, "git_handoff", lambda *_args: "handoff")
reports = [
lanes.ProcessResult(1, "usage limit reached", None, True),
lanes.ProcessResult(1, "fallback also failed", None, False),
]
monkeypatch.setattr(lanes, "run_provider", lambda *_args, **_kwargs: reports.pop(0))
lanes.execute_claim("cassandra", "t_nofallback")
assert reports == []
assert blocks and "fallback also failed" in blocks[0]["reason"]
assert blocks[0]["kind"] == "capability"
def test_worker_blockers_become_the_block_reason_without_goal_mode(
tmp_path: Path,
monkeypatch,
):
"""A structured non-completed report blocks with its own stated blockers."""
task = SimpleNamespace(
id="t_blocked",
status="running",
current_run_id=23,
assignee="cli-auto",
max_runtime_seconds=60,
)
blocks = []
db = SimpleNamespace(
scoped_current_board=lambda _board: nullcontext(),
connect=lambda board: _Connection(),
get_task=lambda *_args: task,
worker_log_path=lambda *_args, **_kwargs: tmp_path / "worker.log",
_resolve_worktree_workspace=lambda *_args, **_kwargs: (tmp_path, "branch"),
set_branch_name=lambda *_args: None,
set_workspace_path=lambda *_args: None,
build_worker_context=lambda *_args: "report the obstacle",
add_comment=lambda *_args, **_kwargs: None,
heartbeat_worker=lambda *_args, **_kwargs: True,
block_task=lambda *_args, **kwargs: blocks.append(kwargs),
)
monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=db))
monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "lanes")
monkeypatch.setattr(lanes, "fresh_unavailable_provider", lambda: None)
route = lanes.Route("codex", "gpt", "high", "p", "c", "r", 1, ())
monkeypatch.setattr(lanes, "select_route", lambda *_args, **_kwargs: route)
structured = {
**_completed_result("progress only"),
"status": "blocked",
"blockers": ["credentials expired", "remote unreachable"],
}
monkeypatch.setattr(
lanes,
"run_provider",
lambda *_args, **_kwargs: lanes.ProcessResult(0, "out", structured, False),
)
lanes.execute_claim("cassandra", "t_blocked")
assert blocks and blocks[0]["reason"] == (
"credentials expired; remote unreachable"
)

View File

@ -162,3 +162,68 @@ def test_terminal_finalizer_closes_supplied_invalid_snapshots(
empty = _Snapshot(None)
assert lanes._finalize_terminal_record(_db(None), path, snapshot=empty) == "invalid"
assert empty.closed == 1
def test_finalizer_rejects_invalid_names_and_unpinnable_journals(
tmp_path: Path,
monkeypatch,
):
"""Invalid identities and unreadable journals return without side effects."""
monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
(tmp_path / "cli-lanes/cassandra").mkdir(parents=True)
db = _db(None)
foreign = tmp_path / "cli-lanes/cassandra/not-a-journal.json"
assert lanes._finalize_terminal_record(db, foreign) == "invalid"
missing = lanes._terminal_path(
lanes.state_path("cassandra", "t_edge"),
5,
"pending",
)
assert lanes._finalize_terminal_record(db, missing) == "invalid"
class _NullSnapshot:
def __init__(self, document):
self.document = document
self.closed = 0
def close(self):
self.closed += 1
def test_finalizer_closes_supplied_snapshots_on_every_invalid_path(
tmp_path: Path,
monkeypatch,
):
"""Caller-pinned snapshots are released for named and payload rejections."""
monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
(tmp_path / "cli-lanes/cassandra").mkdir(parents=True)
db = _db(None)
pending = lanes._terminal_path(
lanes.state_path("cassandra", "t_edge"),
6,
"pending",
)
misnamed = _NullSnapshot({"kanban_state": "pending"})
assert (
lanes._finalize_terminal_record(
db,
tmp_path / "cli-lanes/cassandra/not-a-journal.json",
snapshot=misnamed,
)
== "invalid"
)
assert misnamed.closed == 1
empty = _NullSnapshot(None)
assert lanes._finalize_terminal_record(db, pending, snapshot=empty) == "invalid"
assert empty.closed == 1
malformed = _NullSnapshot({"kanban_state": "pending"})
assert lanes._finalize_terminal_record(db, pending, snapshot=malformed) == (
"invalid"
)
assert malformed.closed == 1

View File

@ -205,3 +205,50 @@ def test_switchyard_network_failure_is_explicit():
URLError("offline")
),
)
def test_artifact_directories_and_foreign_statuses_are_filtered(
tmp_path: Path,
):
"""String directory paths and non-result statuses never become evidence."""
workspace = tmp_path / "workspace"
(workspace / "directory").mkdir(parents=True)
assert lanes.workspace_artifacts(workspace, ["directory"]) == []
assert lanes._extract_json('{"status": "unknown"}') is None
def test_provider_events_without_sessions_or_results_stay_unparsed(
tmp_path: Path,
):
"""Session persistence and result extraction require explicit fields."""
state_file = tmp_path / "state.json"
state = {}
assert (
lanes._event_payload(
"codex",
json.dumps({"type": "thread.started", "thread": {}}),
state,
state_file,
)
is None
)
assert "codex_thread_id" not in state
assert (
lanes._event_payload(
"codex",
json.dumps({"item": {"text": "no structured result here"}}),
state,
state_file,
)
is None
)
direct = lanes._event_payload(
"codex",
json.dumps({"result": json.dumps(_completed_result("direct"))}),
state,
state_file,
)
assert direct and direct["summary"] == "direct"

View File

@ -149,3 +149,99 @@ def test_claim_requires_structured_evidence_and_surfaces_artifacts(
else:
assert calls[0][1]["kind"] == "capability"
assert all(connection.closed for connection in connections)
def test_goal_loop_recovers_from_corrupt_rejection_history(
tmp_path: Path,
monkeypatch,
):
"""Corrupt durable rejection history is replaced, not trusted or fatal."""
task = SimpleNamespace(
id="t_history",
status="running",
result=None,
current_run_id=6,
assignee="cli-auto",
max_runtime_seconds=300,
goal_mode=True,
goal_max_turns=2,
)
calls = []
fake_db = SimpleNamespace(
scoped_current_board=lambda _board: nullcontext(),
connect=lambda board: SimpleNamespace(close=lambda: None),
get_task=lambda _conn, _task_id: task,
worker_log_path=lambda _task_id, board: tmp_path / "worker.log",
_resolve_worktree_workspace=lambda _task, board: (tmp_path, "wt/t_history"),
set_branch_name=lambda *_args: None,
set_workspace_path=lambda *_args: None,
build_worker_context=lambda *_args: "achieve the goal with evidence",
heartbeat_worker=lambda *_args, **_kwargs: True,
add_comment=lambda *_args: None,
complete_task=lambda *_args, **kwargs: (
calls.append(("complete", kwargs)) or True
),
block_task=lambda *_args, **kwargs: calls.append(("block", kwargs)),
)
monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db))
monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
state_file = lanes.state_path("cassandra", "t_history")
lanes.atomic_json(state_file, {"goal_rejections": "corrupt-history"})
probes = []
monkeypatch.setattr(
lanes,
"fresh_unavailable_provider",
lambda: probes.append("health") and None,
)
route = lanes.Route(
"codex", "gpt-5.6-sol", "high", "codex-high", "jetson", "vote", 1, ()
)
monkeypatch.setattr(lanes, "select_route", lambda *_args, **_kwargs: route)
reports = [
lanes.ProcessResult(
0,
"first turn",
{
"status": "completed",
"summary": "done",
"changed_files": ["src/a.py"],
"tests_run": ["pytest -q"],
"artifacts": [],
"findings": [],
"blockers": [],
},
False,
),
lanes.ProcessResult(
0,
"second turn",
{
"status": "completed",
"summary": "done with verification",
"changed_files": ["src/a.py"],
"tests_run": ["pytest -q: passed"],
"artifacts": [],
"findings": [],
"blockers": [],
},
False,
),
]
monkeypatch.setattr(lanes, "run_provider", lambda *_args, **_kwargs: reports.pop(0))
verdicts = iter([(False, "verification evidence missing"), (True, "verified")])
monkeypatch.setattr(
lanes.cli_lane_goal,
"judge_goal_completion",
lambda *_args, **_kwargs: next(verdicts),
)
lanes.execute_claim("cassandra", "t_history")
assert reports == []
assert calls[0][0] == "complete"
assert calls[0][1]["metadata"]["goal_turn"] == 2
assert len(probes) == 2
state = lanes.load_json(state_file)
assert state["goal_rejections"] == [
"local goal judge requested continuation: verification evidence missing"
]

View File

@ -118,3 +118,14 @@ def test_goal_judge_fails_closed_on_invalid_response():
assert accepted is False
assert "local completion judge unavailable" in reason
def test_completed_report_with_corrupt_test_evidence_is_still_inspected():
"""A non-list tests_run field neither crashes nor blocks the summary gate."""
assert goal.unfinished_result_reason(_result(tests_run=None)) is None
assert goal.unfinished_result_reason(
_result(
tests_run={"suite": "still running"},
summary="The remaining work is still running.",
)
)

View File

@ -0,0 +1,238 @@
"""Legacy/patched image compatibility matrix for mixed Hermes rollouts."""
from __future__ import annotations
from contextlib import nullcontext
from pathlib import Path
from types import SimpleNamespace
import pytest
from testing.tests.test_hermes_cli_capabilities import (
_new_reclaim,
_old_reclaim,
_terminal_document,
)
from testing.tests.test_hermes_cli_support import lanes
def _coordinator_db(*, patched: bool):
"""Model the image decomposition API used by the coordinator container."""
if patched:
def specify(
_conn,
_task_id,
*,
title=None,
body=None,
assignee=None,
author=None,
require_no_runs=False,
):
return not require_no_runs
def decompose(
_conn,
_task_id,
*,
root_assignee,
children,
author=None,
auto_promote=True,
require_no_runs=False,
):
return None if require_no_runs else [f"{_task_id}.1"]
else:
def specify(
_conn,
_task_id,
*,
title=None,
body=None,
assignee=None,
author=None,
):
return True
def decompose(
_conn,
_task_id,
*,
root_assignee,
children,
author=None,
auto_promote=True,
):
return [f"{_task_id}.1"]
return SimpleNamespace(
specify_triage_task=specify,
decompose_triage_task=decompose,
)
@pytest.mark.parametrize("worker_patched", [False, True])
@pytest.mark.parametrize("coordinator_patched", [False, True])
def test_mixed_image_rollout_compatibility_matrix(
tmp_path: Path,
monkeypatch,
coordinator_patched: bool,
worker_patched: bool,
):
"""Exact legacy/patched image matrix for decomposition and finalization.
The hermes coordinator container and the cli-lane-runner worker container
can run different image generations during a rollout while sharing one
Kanban SQLite database and these mounted scripts. The behaviors that
differ between the legacy and the patched image on this path are:
* legacy ``complete_task`` accepts ``expected_run_id`` but not
``replay_ended_run_id``: active exact runs complete on both images,
ended-run replay is image-gated.
* legacy ``reclaim_task`` lacks ``expected_run_id``: orphan reclaim is
image-gated (covered by test_old_image_never_calls_unguarded_reclaim).
* legacy ``specify_triage_task``/``decompose_triage_task`` lack
``require_no_runs``: only the patched coordinator refuses to redefine
a task that already has execution history.
Combination behavior proven here:
* legacy coordinator + legacy worker: mid-run decomposition succeeds
unguarded; the worker defers the ended run's accepted result (journal
preserved for a patched replay) and reports deferred health.
* legacy coordinator + patched worker: mid-run decomposition succeeds
unguarded; the patched worker replays under ``replay_ended_run_id``,
the database refuses the voided run, and the result converges to
``stale`` (conflict evidence) instead of completing a redefined task.
* patched coordinator + legacy worker: decomposition is refused while
the run exists, so the still-running exact run completes under
``expected_run_id``; replay/reclaim stay deferred and the readiness
probe keeps the worker pod unready for new claims.
* patched coordinator + patched worker: decomposition is refused while
the run exists, the exact run completes, and the worker is ready.
"""
monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
task = SimpleNamespace(
id="t_matrix",
status="running",
current_run_id=7,
completed_run_id=None,
result=None,
assignee="cli-auto",
)
def old_matrix_complete(
_conn,
_task_id,
*,
result=None,
summary=None,
metadata=None,
expected_run_id=None,
):
if task.status == "running" and task.current_run_id == expected_run_id:
task.status = "done"
task.completed_run_id = expected_run_id
task.current_run_id = None
task.result = result
return True
return False
def new_matrix_complete(
_conn,
_task_id,
*,
result=None,
summary=None,
metadata=None,
expected_run_id=None,
replay_ended_run_id=None,
):
if replay_ended_run_id is not None:
# The decomposed run is no longer the latest ended run.
return False
return old_matrix_complete(
_conn,
_task_id,
result=result,
summary=summary,
metadata=metadata,
expected_run_id=expected_run_id,
)
worker_db = SimpleNamespace(
scoped_current_board=lambda _board: nullcontext(),
connect=lambda board: SimpleNamespace(close=lambda: None),
get_task=lambda _conn, _task_id: task,
complete_task=new_matrix_complete if worker_patched else old_matrix_complete,
reclaim_task=_new_reclaim if worker_patched else _old_reclaim,
)
coordinator_db = _coordinator_db(patched=coordinator_patched)
assert (
lanes._explicit_keyword(
coordinator_db.decompose_triage_task,
"require_no_runs",
)
is coordinator_patched
)
assert (
lanes._explicit_keyword(
coordinator_db.specify_triage_task,
"require_no_runs",
)
is coordinator_patched
)
health = tmp_path / "runtime-health.json"
capabilities = lanes.initialize_kanban_capabilities(
worker_db,
health_path=health,
)
assert capabilities.ready is worker_patched
assert capabilities.exact_run_completion is True
if worker_patched:
assert lanes.readiness_issue(health) is None
else:
assert capabilities.deferred_features == (
"ended-run-replay",
"exact-run-reclaim",
)
assert (
lanes.readiness_issue(health) == "runtime compatibility is deferred"
)
if coordinator_patched:
refused = coordinator_db.decompose_triage_task(
object(),
"t_matrix",
root_assignee="cli-auto",
children=[{"title": "split"}],
require_no_runs=True,
)
assert refused is None
else:
created = coordinator_db.decompose_triage_task(
object(),
"t_matrix",
root_assignee="cli-auto",
children=[{"title": "split"}],
)
assert created == ["t_matrix.1"]
task.status = "triage"
task.current_run_id = None
identity = lanes.TerminalIdentity("cassandra", "t_matrix", 7, "pending")
outcome = lanes._finalize_document_db(worker_db, identity, _terminal_document())
if coordinator_patched:
assert outcome == "committed"
assert task.status == "done"
assert task.completed_run_id == 7
elif worker_patched:
assert outcome == "stale"
assert task.status == "triage"
else:
assert outcome == "deferred"
assert task.status == "triage"

View File

@ -171,3 +171,168 @@ def test_codex_missing_thread_skips_colliding_restart_result(
)
restart = Path(calls[1][calls[1].index("-o") + 1])
assert restart.name.endswith("provider-3.result.json")
def test_stream_process_trims_the_retained_line_window(tmp_path: Path, monkeypatch):
"""Streamed chatter keeps only the newest bounded output window."""
monkeypatch.setattr(lanes, "_descendant_processes", lambda _pid: {})
monkeypatch.setattr(lanes, "_terminate_worker_process", lambda *_a, **_k: None)
read_fd, write_fd = os.pipe()
os.write(write_fd, "".join(f"{index}\n" for index in range(4100)).encode())
os.close(write_fd)
polls = []
class ChatteringProcess:
pid = 4243
returncode = 0
stdout = os.fdopen(read_fd, "r")
def poll(self):
polls.append(True)
return None if len(polls) < 4300 else 0
monkeypatch.setattr(
lanes.subprocess,
"Popen",
lambda *_args, **_kwargs: ChatteringProcess(),
)
heartbeats = []
result = lanes.stream_process(
["ignored"],
provider="codex",
cwd=tmp_path,
env=dict(os.environ),
log_path=tmp_path / "chatter.log",
state={},
state_file=tmp_path / "chatter-state.json",
heartbeat=lambda note: heartbeats.append(note) or True,
max_runtime=60,
)
assert result.returncode == 0
assert heartbeats
published = result.output.splitlines()
assert len(published) == 4000
assert published[0] == "100"
assert published[-1] == "4099"
def test_stream_process_parses_output_left_after_worker_exit(
tmp_path: Path,
monkeypatch,
):
"""Buffered final output from an already-exited worker is still parsed."""
payload = json.dumps({"result": json.dumps(_completed_result("buffered"))})
read_fd, write_fd = os.pipe()
os.write(write_fd, f"{payload}\n".encode())
os.close(write_fd)
fake = type(
"ExitedProcess",
(),
{
"pid": 4242,
"returncode": 0,
"poll": lambda self: 0,
"stdout": os.fdopen(read_fd, "r"),
},
)()
monkeypatch.setattr(lanes.subprocess, "Popen", lambda *_args, **_kwargs: fake)
monkeypatch.setattr(lanes, "_terminate_worker_process", lambda *_a, **_k: None)
result = lanes.stream_process(
["ignored"],
provider="codex",
cwd=tmp_path,
env={},
log_path=tmp_path / "buffered.log",
state={},
state_file=tmp_path / "buffered-state.json",
heartbeat=lambda _note: True,
max_runtime=60,
)
assert result.returncode == 0
assert result.structured and result.structured["summary"] == "buffered"
assert payload in (tmp_path / "buffered.log").read_text(encoding="utf-8")
def test_descendant_snapshot_ignores_processes_that_vanish(monkeypatch):
entries = [Path("/proc/10"), Path("/proc/11"), Path("/proc/13")]
monkeypatch.setattr(lanes.Path, "iterdir", lambda _path: iter(entries))
records = {10: (1, 10, 100), 11: (10, 11, 101)}
monkeypatch.setattr(lanes, "_process_record", lambda pid: records.get(pid))
assert lanes._descendant_processes(10) == {11: (11, 101)}
def test_signal_tree_targets_only_identity_verified_descendants(monkeypatch):
"""Reused PIDs are never signaled; verified groups and PIDs both are."""
records = {10: (1, 20, 100)}
monkeypatch.setattr(lanes, "_process_record", lambda pid: records.get(pid))
group_signals = []
pid_signals = []
monkeypatch.setattr(
lanes.os,
"killpg",
lambda group, sig: group_signals.append((group, sig)),
)
monkeypatch.setattr(
lanes.os,
"kill",
lambda pid, sig: pid_signals.append((pid, sig)),
)
lanes._signal_worker_tree(
5,
{10: (20, 100), 11: (21, 999)},
signal.SIGTERM,
)
assert sorted(group_signals) == [(5, signal.SIGTERM), (20, signal.SIGTERM)]
assert pid_signals == [(10, signal.SIGTERM)]
def test_claude_worker_without_any_recoverable_session_stays_unstarted(
tmp_path: Path,
monkeypatch,
):
"""A session lost on resume and restart never marks the lane as started."""
state = {"claude_started": True, "claude_session_id": "session-lost"}
state_file = tmp_path / "state.json"
commands = []
def stream(command, **_kwargs):
commands.append(command)
return lanes.ProcessResult(
1,
f"{lanes.NO_CLAUDE_SESSION} session-lost",
None,
False,
)
monkeypatch.setattr(lanes, "stream_process", stream)
persisted = []
monkeypatch.setattr(
sys.modules["cli_lane_provider"],
"atomic_json",
lambda path, value: persisted.append(dict(value)),
)
route = lanes.Route("claude", "claude-fable-5", "high", "p", "c", "r", 1, ())
result = lanes.run_provider(
route,
"prompt",
tmp_path,
state,
state_file,
tmp_path / "worker.log",
lambda _note: True,
60,
)
assert result.returncode == 1
assert len(commands) == 2
assert "--resume" in commands[0]
assert "--session-id" in commands[1]
assert len(persisted) == 1

View File

@ -228,3 +228,66 @@ def test_terminal_schema_validates_optional_identity():
record,
lanes.TerminalIdentity("other", "task", 1, "pending"),
) is False
def test_terminal_identity_rejects_noncanonical_run_ids():
"""Identity construction fails closed before SQLite can see the run."""
with pytest.raises(ValueError, match="positive SQLite int64"):
lanes.TerminalIdentity("cassandra", "t_identity", 0, "pending")
with pytest.raises(ValueError, match="positive SQLite int64"):
lanes.TerminalIdentity("cassandra", "t_identity", "7", "pending")
def test_small_snapshot_rejects_payloads_that_grow_past_the_stat_bound(
tmp_path: Path,
monkeypatch,
):
"""A file appended between fstat and read never yields a snapshot."""
path = tmp_path / "grown.json"
path.write_text("{}", encoding="utf-8")
monkeypatch.setattr(
lanes,
"_read_bounded",
lambda _descriptor, limit: b"x" * (lanes.MAX_TERMINAL_RECORD_BYTES + 1),
)
assert lanes._open_small_json_snapshot(path) is None
def test_recovery_snapshot_marks_racing_writers_as_unstable(
tmp_path: Path,
monkeypatch,
):
"""A payload that changes under the pinned read cannot be classified."""
path = tmp_path / "unstable.json"
path.write_text('{"kanban_state": "pending"}', encoding="utf-8")
real_read = lanes._read_bounded
monkeypatch.setattr(
lanes,
"_read_bounded",
lambda descriptor, limit: real_read(descriptor, limit) + b" ",
)
snapshot = lanes._open_terminal_recovery_snapshot(path)
assert snapshot is not None
assert snapshot.document is None
assert snapshot.invalid_reason == "unstable-payload"
snapshot.close()
snapshot.close()
def test_recovery_snapshot_releases_descriptors_on_read_faults(
tmp_path: Path,
monkeypatch,
):
"""An I/O fault during the pinned read closes both held descriptors."""
path = tmp_path / "faulted.json"
path.write_text("{}", encoding="utf-8")
monkeypatch.setattr(
lanes,
"_read_bounded",
lambda _descriptor, _limit: (_ for _ in ()).throw(OSError("read fault")),
)
assert lanes._open_terminal_recovery_snapshot(path) is None

View File

@ -5,6 +5,7 @@ from __future__ import annotations
import hashlib
import os
import sys
from contextlib import nullcontext
from pathlib import Path
from types import SimpleNamespace
@ -296,3 +297,148 @@ def test_pending_guard_rejects_invalid_run_and_inaccessible_or_bad_artifacts(
monkeypatch.setattr(Path, "stat", denied)
assert lanes._has_pending_finalization("board", "task", 12) is False
def test_staged_authority_rejects_a_mismatched_pathname_digest(
tmp_path: Path,
monkeypatch,
):
"""A staged entry whose name does not bind its payload has no authority."""
monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "lanes")
document = _pending_terminal_record("board", "task", 5, "accepted")
board = lanes.STATE_ROOT / "board"
board.mkdir(parents=True)
forged = board / f".retire.{'f' * 16}.{'a' * 16}.0"
assert lanes._staged_terminal_authority(forged, _Snapshot(document)) is None
def test_drain_staging_defers_unbounded_replacement_churn(
tmp_path: Path,
monkeypatch,
):
"""A staging surface that keeps changing stops after bounded attempts."""
monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "lanes")
board = lanes.STATE_ROOT / "board"
board.mkdir(parents=True)
churning = board / ".retire.churning"
churning.write_text("0", encoding="utf-8")
def churn():
churning.write_text(churning.read_text(encoding="utf-8") + "x")
return 1
monkeypatch.setattr(lanes, "_recover_retirement_staging", churn)
assert lanes._drain_retirement_staging() == 16
def test_prepared_recovery_orders_and_bounds_disappearing_journals(
tmp_path: Path,
monkeypatch,
):
"""A journal whose stat fails is ordered last instead of aborting replay."""
path, _document = _prepared_path(tmp_path, monkeypatch)
real_stat = Path.stat
calls = []
def vanishing(candidate, *args, **kwargs):
if candidate == path and not calls:
calls.append("ordered")
raise OSError("stat raced")
return real_stat(candidate, *args, **kwargs)
monkeypatch.setattr(Path, "stat", vanishing)
monkeypatch.setattr(lanes, "_finalize_document_db", lambda *_args: "deferred")
assert lanes._recover_prepared_finalizations(object()) == 0
assert calls == ["ordered"]
assert path.exists()
def test_prepared_recovery_defers_a_journal_that_cannot_be_quarantined(
tmp_path: Path,
monkeypatch,
):
"""An unquarantinable malformed name never spins recovery forever."""
monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "lanes")
board = lanes.STATE_ROOT / "board"
board.mkdir(parents=True)
misnamed = board / "task.run-9.terminal.prepared-zz.json"
misnamed.write_text("{}", encoding="utf-8")
attempts = []
monkeypatch.setattr(
lanes,
"_quarantine_terminal",
lambda *_args, **_kwargs: attempts.append("quarantine"),
)
assert lanes._recover_prepared_finalizations(object()) == 0
assert len(attempts) == 16
assert misnamed.exists()
def test_prepared_recovery_counts_one_recovery_per_exact_run(
tmp_path: Path,
monkeypatch,
):
"""Duplicate prepared evidence for one run converges to a single winner."""
monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "lanes")
identity = lanes.TerminalIdentity("board", "task", 9, "pending")
first = _pending_terminal_record("board", "task", 9, "first accepted")
second = _pending_terminal_record("board", "task", 9, "second accepted")
lanes._persist_prepared_evidence(identity, first)
lanes._persist_prepared_evidence(identity, second)
task = SimpleNamespace(
id="task",
status="running",
current_run_id=9,
completed_run_id=None,
result=None,
assignee="cli-auto",
)
db = SimpleNamespace(
scoped_current_board=lambda _board: nullcontext(),
connect=lambda board: SimpleNamespace(close=lambda: None),
get_task=lambda _conn, _task_id: task,
complete_task=lambda *_args, **_kwargs: True,
)
assert lanes._recover_prepared_finalizations(db) == 1
conflicts = list((lanes.STATE_ROOT / "board").glob("*.terminal.conflict-*.json"))
assert len(conflicts) == 1
def test_pending_guard_ignores_symlinked_boards_and_invalid_artifacts(
tmp_path: Path,
monkeypatch,
):
"""Invalid journals, evidence, and staging never hold a claim open."""
monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "lanes")
board = lanes.STATE_ROOT / "board"
board.mkdir(parents=True)
pending = lanes._terminal_path(lanes.state_path("board", "task"), 13)
pending.write_text("{}", encoding="utf-8")
misnamed = board / "task.run-13.terminal.prepared-zz.json"
misnamed.write_text("{}", encoding="utf-8")
invalid = board / f"task.run-13.terminal.prepared-{'a' * 32}.json"
invalid.write_text("{}", encoding="utf-8")
(board / ".retire.dangling").symlink_to(board / "missing-target")
foreign = _pending_terminal_record("board", "other", 13, "accepted")
staged, _canonical = _staged_path(
lanes.STATE_ROOT,
lanes.TerminalIdentity("board", "other", 13, "pending"),
foreign,
)
lanes.atomic_json(staged, foreign)
assert lanes._has_pending_finalization("board", "task", 13) is False
assert lanes._has_pending_finalization("board", "other", 13) is True
aliased_root = tmp_path / "aliased"
aliased_root.mkdir()
(aliased_root / "linked").symlink_to(board, target_is_directory=True)
monkeypatch.setattr(lanes, "STATE_ROOT", aliased_root)
linked_pending = aliased_root / "linked" / "task.run-13.terminal.pending.json"
assert linked_pending.exists()
assert lanes._has_pending_finalization("linked", "task", 13) is False

View File

@ -2,6 +2,7 @@
from __future__ import annotations
import os
from pathlib import Path
from testing.tests.test_hermes_cli_support import (
@ -87,3 +88,66 @@ def test_gc_skips_symlink_board_and_disappearing_candidate(
monkeypatch.setattr(Path, "stat", disappear)
assert lanes.gc_lane_artifacts() == 0
def test_gc_reports_quarantine_that_left_the_entry_in_place(
tmp_path: Path,
monkeypatch,
):
"""An invalid retained journal that survives quarantine is not counted."""
root = tmp_path / "lanes"
board = root / "board"
board.mkdir(parents=True)
retained = board / "task.run-1.terminal.committed.json"
retained.write_text("{}", encoding="utf-8")
monkeypatch.setattr(lanes, "STATE_ROOT", root)
monkeypatch.setattr(lanes, "_quarantine_terminal", lambda *_a, **_k: retained)
assert lanes.gc_lane_artifacts() == 0
assert retained.exists()
def test_gc_bounds_unpinnable_staging_and_failed_nonregular_unlinks(
tmp_path: Path,
monkeypatch,
):
"""Unsnapshotable staging entries fall through to identity-safe removal."""
root = tmp_path / "lanes"
board = root / "board"
board.mkdir(parents=True)
dangling = board / ".retire.dangling"
dangling.symlink_to(board / "missing-target")
monkeypatch.setattr(lanes, "STATE_ROOT", root)
attempts = []
monkeypatch.setattr(
lanes,
"_unlink_artifact_if_same",
lambda path, _observed: attempts.append(path.name) and False,
)
assert lanes.gc_lane_artifacts() == 0
assert attempts == [".retire.dangling"]
def test_gc_keeps_entries_whose_capped_unlink_loses_the_race(
tmp_path: Path,
monkeypatch,
):
"""A lost inode race during count capping removes nothing else."""
root = tmp_path / "lanes"
board = root / "board"
board.mkdir(parents=True)
newer = board / "task.candidate-2.json"
newer.write_text("{}", encoding="utf-8")
older = board / "task.candidate-1.json"
older.write_text("{}", encoding="utf-8")
os.utime(older, (1, 1))
monkeypatch.setattr(lanes, "STATE_ROOT", root)
monkeypatch.setattr(
lanes,
"_unlink_artifact_if_same",
lambda _path, _observed: False,
)
assert lanes.gc_lane_artifacts(now=100.0, max_age_seconds=10**9, max_count=1) == 0
assert older.exists() and newer.exists()

View File

@ -98,8 +98,8 @@ def test_coverage_check_enforces_per_file_floor(tmp_path: Path):
<packages>
<package>
<classes>
<class filename="ok.py" line-rate="1.0" />
<class filename="low.py" line-rate="0.90" />
<class filename="ok.py" line-rate="1.0" branch-rate="1.0" />
<class filename="low.py" line-rate="0.90" branch-rate="1.0" />
</classes>
</package>
</packages>
@ -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
<packages>
<package>
<classes>
<class filename="mapped.py" line-rate="1.0" />
<class filename="{(tmp_path / 'absolute.py').as_posix()}" line-rate="1.0" />
<class filename="skip.py" />
<class filename="mapped.py" line-rate="1.0" branch-rate="1.0" />
<class filename="{(tmp_path / 'absolute.py').as_posix()}" line-rate="1.0" branch-rate="1.0" />
</classes>
</package>
</packages>

View File

@ -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
<packages>
<package>
<classes>
<class filename="alpha.py" line-rate="1.0" />
<class filename="beta.py" line-rate="0.5" />
<class filename="alpha.py" line-rate="1.0" branch-rate="1.0" />
<class filename="beta.py" line-rate="0.5" branch-rate="1.0" />
</classes>
</package>
</packages>
@ -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:
<packages>
<package>
<classes>
<class filename="relative.py" line-rate="0.80" />
<class filename="relative.py" line-rate="0.80" branch-rate="1.0" />
</classes>
</package>
</packages>
@ -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'<coverage><class filename="managed.py" line-rate="1"{attribute}/></coverage>',
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(
'<coverage><class filename="managed.py" line-rate="1" branch-rate="0.5"/></coverage>',
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(
'<coverage><class filename="managed.py" line-rate="1" branch-rate="1"/></coverage>',
encoding="utf-8",
)
assert (
run_check(
{"coverage": {"minimum_percent": 95, "tracked_files": ["managed.py"]}},
tmp_path,
coverage_xml,
)
== []
)

View File

@ -4,7 +4,7 @@ from __future__ import annotations
from pathlib import Path
from testing.quality_hygiene import count_files_over_line_limit
from testing.quality_hygiene import count_files_over_line_limit, run_check
def test_count_files_over_line_limit_counts_only_long_matches(tmp_path: Path) -> None:
@ -24,3 +24,28 @@ def test_count_files_over_line_limit_counts_only_long_matches(tmp_path: Path) ->
}
assert count_files_over_line_limit(contract, tmp_path) == 1
def test_run_check_exempts_conftest_from_naming_rules(tmp_path: Path) -> None:
"""Shared conftest.py fixtures never violate test naming rules."""
tests_dir = tmp_path / "tests"
tests_dir.mkdir()
(tests_dir / "conftest.py").write_text("fixtures = True\n", encoding="utf-8")
(tests_dir / "test_named.py").write_text("checked = True\n", encoding="utf-8")
contract = {
"hygiene": {
"max_lines": 3,
"line_limit_globs": [],
"naming_rules": [
{
"glob": "tests/*.py",
"pattern": "^test_[a-z0-9_]+\\.py$",
"description": "pytest files use test_*.py names",
}
],
}
}
assert run_check(contract, tmp_path) == []

View File

@ -168,3 +168,24 @@ def test_main_honors_custom_blocking_severity_without_sonar_output(tmp_path: Pat
payload = json.loads(output.read_text(encoding="utf-8"))
assert payload["status"] == "failed"
assert payload["blocking_findings"] == 1
def test_build_sonar_issues_normalizes_inverted_text_ranges() -> None:
"""An end line before the start line collapses to the start line."""
issues = semgrep_report.build_sonar_issues(
{
"results": [
{
"path": "app/config.yaml",
"start": {"line": 9},
"end": {"line": 4},
}
]
}
)
assert issues["issues"][0]["primaryLocation"]["textRange"] == {
"startLine": 9,
"endLine": 9,
}