hermes: close lane branch coverage gaps

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
jenkins 2026-08-17 15:03:59 -03:00
parent e3ecc18d0e
commit df75479069
16 changed files with 1219 additions and 3 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 {} start = finding.get("start") if isinstance(finding.get("start"), dict) else {}
end = finding.get("end") if isinstance(finding.get("end"), 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) 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) 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 { return {
"engineId": "semgrep", "engineId": "semgrep",
"ruleId": str(finding.get("check_id") or "semgrep.unknown"), "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) handler = listener.Handler.__new__(listener.Handler)
assert handler.log_message("ignored %s", "value") is None 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

@ -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 lanes.maybe_gc_lane_artifacts(now=1000) == 0
assert "retention deferred" in capsys.readouterr().err 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 blocks[0]["expected_run_id"] == 23
assert "router down" in blocks[0]["reason"] 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) empty = _Snapshot(None)
assert lanes._finalize_terminal_record(_db(None), path, snapshot=empty) == "invalid" assert lanes._finalize_terminal_record(_db(None), path, snapshot=empty) == "invalid"
assert empty.closed == 1 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") 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: else:
assert calls[0][1]["kind"] == "capability" assert calls[0][1]["kind"] == "capability"
assert all(connection.closed for connection in connections) 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 accepted is False
assert "local completion judge unavailable" in reason 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]) restart = Path(calls[1][calls[1].index("-o") + 1])
assert restart.name.endswith("provider-3.result.json") 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, record,
lanes.TerminalIdentity("other", "task", 1, "pending"), lanes.TerminalIdentity("other", "task", 1, "pending"),
) is False ) 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 hashlib
import os import os
import sys import sys
from contextlib import nullcontext
from pathlib import Path from pathlib import Path
from types import SimpleNamespace 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) monkeypatch.setattr(Path, "stat", denied)
assert lanes._has_pending_finalization("board", "task", 12) is False 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 from __future__ import annotations
import os
from pathlib import Path from pathlib import Path
from testing.tests.test_hermes_cli_support import ( 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) monkeypatch.setattr(Path, "stat", disappear)
assert lanes.gc_lane_artifacts() == 0 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

@ -4,7 +4,7 @@ from __future__ import annotations
from pathlib import Path 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: 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 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")) payload = json.loads(output.read_text(encoding="utf-8"))
assert payload["status"] == "failed" assert payload["status"] == "failed"
assert payload["blocking_findings"] == 1 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,
}