diff --git a/ci/scripts/semgrep_report.py b/ci/scripts/semgrep_report.py index b4ed5a2d..5175e2e6 100644 --- a/ci/scripts/semgrep_report.py +++ b/ci/scripts/semgrep_report.py @@ -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"), diff --git a/scripts/tests/test_mailu_sync_edges.py b/scripts/tests/test_mailu_sync_edges.py new file mode 100644 index 00000000..db1d93d6 --- /dev/null +++ b/scripts/tests/test_mailu_sync_edges.py @@ -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" + ] diff --git a/scripts/tests/test_mailu_sync_listener.py b/scripts/tests/test_mailu_sync_listener.py index eff7ff34..58285591 100644 --- a/scripts/tests/test_mailu_sync_listener.py +++ b/scripts/tests/test_mailu_sync_listener.py @@ -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"] diff --git a/testing/tests/test_hermes_cli_evidence_edges.py b/testing/tests/test_hermes_cli_evidence_edges.py index c28ff43f..86b02a06 100644 --- a/testing/tests/test_hermes_cli_evidence_edges.py +++ b/testing/tests/test_hermes_cli_evidence_edges.py @@ -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"}) diff --git a/testing/tests/test_hermes_cli_execution_edges.py b/testing/tests/test_hermes_cli_execution_edges.py index 07b9035d..7873c86c 100644 --- a/testing/tests/test_hermes_cli_execution_edges.py +++ b/testing/tests/test_hermes_cli_execution_edges.py @@ -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" + ) diff --git a/testing/tests/test_hermes_cli_finalization_edges.py b/testing/tests/test_hermes_cli_finalization_edges.py index 0f2c2a2c..92b69621 100644 --- a/testing/tests/test_hermes_cli_finalization_edges.py +++ b/testing/tests/test_hermes_cli_finalization_edges.py @@ -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 diff --git a/testing/tests/test_hermes_cli_foundation_coverage.py b/testing/tests/test_hermes_cli_foundation_coverage.py index 71f0c44a..9184aae3 100644 --- a/testing/tests/test_hermes_cli_foundation_coverage.py +++ b/testing/tests/test_hermes_cli_foundation_coverage.py @@ -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" diff --git a/testing/tests/test_hermes_cli_goal_execution.py b/testing/tests/test_hermes_cli_goal_execution.py index ce4ebc37..749a3d12 100644 --- a/testing/tests/test_hermes_cli_goal_execution.py +++ b/testing/tests/test_hermes_cli_goal_execution.py @@ -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" + ] diff --git a/testing/tests/test_hermes_cli_lane_goal.py b/testing/tests/test_hermes_cli_lane_goal.py index 24bea93d..0a2d7180 100644 --- a/testing/tests/test_hermes_cli_lane_goal.py +++ b/testing/tests/test_hermes_cli_lane_goal.py @@ -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.", + ) + ) diff --git a/testing/tests/test_hermes_cli_matrix_capabilities.py b/testing/tests/test_hermes_cli_matrix_capabilities.py new file mode 100644 index 00000000..0e9498f3 --- /dev/null +++ b/testing/tests/test_hermes_cli_matrix_capabilities.py @@ -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" diff --git a/testing/tests/test_hermes_cli_provider_edges.py b/testing/tests/test_hermes_cli_provider_edges.py index 21e57453..9b5b6c14 100644 --- a/testing/tests/test_hermes_cli_provider_edges.py +++ b/testing/tests/test_hermes_cli_provider_edges.py @@ -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 diff --git a/testing/tests/test_hermes_cli_records_edges.py b/testing/tests/test_hermes_cli_records_edges.py index 3362466a..f6864918 100644 --- a/testing/tests/test_hermes_cli_records_edges.py +++ b/testing/tests/test_hermes_cli_records_edges.py @@ -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 diff --git a/testing/tests/test_hermes_cli_recovery_edges.py b/testing/tests/test_hermes_cli_recovery_edges.py index 27ba6773..94cd35ca 100644 --- a/testing/tests/test_hermes_cli_recovery_edges.py +++ b/testing/tests/test_hermes_cli_recovery_edges.py @@ -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 diff --git a/testing/tests/test_hermes_cli_retention_edges.py b/testing/tests/test_hermes_cli_retention_edges.py index f7deb6a2..7e56f592 100644 --- a/testing/tests/test_hermes_cli_retention_edges.py +++ b/testing/tests/test_hermes_cli_retention_edges.py @@ -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() diff --git a/testing/tests/test_quality_hygiene_helpers.py b/testing/tests/test_quality_hygiene_helpers.py index a303c650..9902f445 100644 --- a/testing/tests/test_quality_hygiene_helpers.py +++ b/testing/tests/test_quality_hygiene_helpers.py @@ -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) == [] diff --git a/testing/tests/test_semgrep_report.py b/testing/tests/test_semgrep_report.py index e0cf1735..89a15251 100644 --- a/testing/tests/test_semgrep_report.py +++ b/testing/tests/test_semgrep_report.py @@ -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, + }