atlas-iac/testing/tests/test_hermes_cli_review_contract.py

"""Upgrade, single-shot and redaction contracts for the Hermes goal gate."""

from __future__ import annotations

import importlib.util
import json
import re
import sys
from pathlib import Path

import pytest


ROOT = Path(__file__).parents[2]
SPEC = importlib.util.spec_from_file_location(
    "cli_lane_review_contract_test",
    ROOT / "services/hermes/scripts/cli_lane_goal.py",
)
assert SPEC and SPEC.loader
goal = importlib.util.module_from_spec(SPEC)
sys.modules[SPEC.name] = goal
SPEC.loader.exec_module(goal)

# A synthetic 40-character lowercase hex string. It has the shape of a Gitea
# personal access token *and* of a git SHA-1; no real credential is read here.
SYNTHETIC_HEX = "0123456789abcdef" * 2 + "01234567"
REAL_SHA = "b080b5f622ae0998213f3287762aea30dc931a73"

REPORTS = {
    "block_review": {
        "status": "completed",
        "summary": (
            "Independent read-only review of PR #18. Verdict: BLOCK. Deployment "
            "verification is pending on two of three nodes, which is exactly why "
            "the reviewed change is unfit to ship."
        ),
        "changed_files": [],
        "tests_run": ["pytest testing/tests: 337 passed"],
        "artifacts": [],
        "findings": ["P0 - coordinator.py:248 compares int to str run ids."],
        "blockers": [],
    },
    "bare_token_review": {
        "status": "completed",
        "summary": (
            "Independent read-only review of PR #18 concludes BLOCK because the "
            "coordinator drops a lease on its first tick and deployment "
            "verification is pending on two of three nodes."
        ),
        "changed_files": [],
        "tests_run": [],
        "artifacts": [],
        "findings": ["P0 - the lease is dropped before the first heartbeat."],
        "blockers": [],
    },
    "terse_ship_audit": {
        "status": "completed",
        "summary": "Decision: ship. Audit done.",
        "changed_files": [],
        "tests_run": ["ruff check: clean"],
        "artifacts": [],
        "findings": [],
        "blockers": [],
    },
    "block_audit_without_findings": {
        "status": "completed",
        "summary": (
            "The generated-strategy audit harness was inspected end to end and "
            "the verdict is BLOCK on the release train for now."
        ),
        "changed_files": [],
        "tests_run": ["pytest -q: 61 passed"],
        "artifacts": [],
        "findings": [],
        "blockers": [],
    },
    "finished_implementation": {
        "status": "completed",
        "summary": "Every acceptance criterion passed and the branch was pushed.",
        "changed_files": ["services/hermes/scripts/cli_lane_goal.py"],
        "tests_run": ["pytest -q: 12 passed"],
        "artifacts": [],
        "findings": [],
        "blockers": [],
    },
    "unfinished_implementation": {
        "status": "completed",
        "summary": "The broad rerun remains active before the final push.",
        "changed_files": ["services/hermes/scripts/cli_lane_goal.py"],
        "tests_run": [],
        "artifacts": [],
        "findings": [],
        "blockers": [],
    },
    "incomplete_turn": {
        "status": "incomplete",
        "summary": "Only two of six boundaries were read.",
        "changed_files": [],
        "tests_run": [],
        "artifacts": [],
        "findings": [],
        "blockers": [],
    },
    "completed_with_blockers": {
        "status": "completed",
        "summary": "The review finished but the head could not be resolved.",
        "changed_files": [],
        "tests_run": [],
        "artifacts": [],
        "findings": [],
        "blockers": ["the Gitea API was unreachable"],
    },
}


def _base_unfinished_result_reason(result):
    """The completion gate exactly as main d8f2d818 shipped it.

    Kept as a local oracle so the upgrade properties below are checked against
    the previous semantics rather than against the implementation under test.
    """
    status = str(result.get("status") or "")
    summary = str(result.get("summary") or "").strip()
    blockers = result.get("blockers")
    if status == "incomplete":
        return summary or "worker explicitly reported incomplete work"
    if status != "completed":
        return None
    if isinstance(blockers, list) and any(str(item).strip() for item in blockers):
        return "worker reported blockers while claiming completion"
    tests = result.get("tests_run")
    evidence = [summary]
    if isinstance(tests, list):
        evidence.extend(str(item) for item in tests)
    match = goal.UNFINISHED_EVIDENCE.search("\n".join(evidence))
    if match:
        return f"completion evidence says work is unfinished: {match.group(0).strip()}"
    return None


def _accepted(result, *, role):
    return goal.unfinished_result_reason(result, role=role) is None


@pytest.mark.parametrize("name", sorted(REPORTS))
def test_a_role_blind_replay_accepts_whatever_the_lane_accepted(name):
    """PR15 re-runs this gate on journalled terminal records without a card.

    A record accepted by any version of the lane must stay valid, or an upgrade
    quarantines it and re-dispatches an already-accepted task.
    """
    result = REPORTS[name]
    accepted_somewhere = _base_unfinished_result_reason(result) is None or any(
        _accepted(result, role=role)
        for role in (goal.REVIEW_ROLE, goal.IMPLEMENTATION_ROLE)
    )

    if accepted_somewhere:
        assert goal.unfinished_result_reason(result) is None


def test_role_blind_replay_keeps_the_unfinished_work_integrity_check():
    """PR15's ``_terminal_record_valid`` relies on this without a card."""
    assert goal.unfinished_result_reason(REPORTS["completed_with_blockers"])
    assert goal.unfinished_result_reason(REPORTS["incomplete_turn"])
    assert goal.unfinished_result_reason(REPORTS["unfinished_implementation"])
    assert goal.unfinished_result_reason(
        {**REPORTS["terse_ship_audit"], "summary": "tests are still running"}
    )


def test_role_blind_replay_exempts_only_a_complete_review_deliverable():
    """The exemption is exactly the shape the role-aware lane calls a review."""
    complete = REPORTS["bare_token_review"]
    mutated = {**complete, "changed_files": ["src/a.py"]}
    verdictless = {**complete, "summary": complete["summary"].replace("BLOCK", "no")}

    assert _base_unfinished_result_reason(complete)
    assert _accepted(complete, role=goal.REVIEW_ROLE)
    assert goal.unfinished_result_reason(complete) is None
    assert goal.unfinished_result_reason(mutated)
    assert goal.unfinished_result_reason(verdictless)


@pytest.mark.parametrize("name", sorted(REPORTS))
def test_implementation_completion_is_byte_for_byte_the_previous_gate(name):
    result = REPORTS[name]

    assert goal.unfinished_result_reason(
        result, role=goal.IMPLEMENTATION_ROLE
    ) == (
        goal.sanitize_reason(_base_unfinished_result_reason(result))
        if _base_unfinished_result_reason(result)
        else None
    )


def test_the_two_single_shot_outcomes_that_change_are_the_documented_ones():
    """Single-shot runs this gate as its only completion check.

    The verdict contract binds there too, so a review's own deliverable decides
    the outcome rather than prose about the artifact it reviewed.
    """
    now_completes = REPORTS["bare_token_review"]
    now_blocks = (REPORTS["terse_ship_audit"], REPORTS["block_audit_without_findings"])

    assert _base_unfinished_result_reason(now_completes)
    assert _accepted(now_completes, role=goal.REVIEW_ROLE)
    for result in now_blocks:
        assert _base_unfinished_result_reason(result) is None
        problem = goal.unfinished_result_reason(result, role=goal.REVIEW_ROLE)
        assert problem and problem.startswith(goal.READ_ONLY_GUARD)


@pytest.mark.parametrize(
    ("text", "secret"),
    [
        (
            f"cloned https://hermes:{SYNTHETIC_HEX}@scm.bstein.dev/titan/x.git",
            SYNTHETIC_HEX,
        ),
        (f"exported GITEA_TOKEN={SYNTHETIC_HEX} into the lane", SYNTHETIC_HEX),
        (f"the pat {SYNTHETIC_HEX} was reused by the askpass helper", SYNTHETIC_HEX),
        (f"Authorization: token {SYNTHETIC_HEX}", SYNTHETIC_HEX),
        ("token ghp_" + "A" * 24, "ghp_" + "A" * 24),
        ("the secret is xoxb-1234567890-abcdefghij", "xoxb-1234567890-abcdefghij"),
        (f"reviewed exact head {REAL_SHA} against base main", None),
    ],
)
def test_judge_reasons_redact_credentials_but_keep_commit_evidence(text, secret):
    """A 40-hex PAT and a 40-hex commit SHA are the same string in isolation.

    Redaction is anchored on the surrounding syntax so credentials disappear
    and the exact head a review pins its findings to survives as evidence.
    """
    reason = goal.sanitize_reason(text)

    if secret is None:
        assert reason == text
    else:
        assert secret not in reason
        assert "[redacted]" in reason or "***" in reason


def test_redaction_falls_back_when_the_agent_runtime_is_absent(monkeypatch):
    monkeypatch.setattr(goal, "_canonical_redact", None)

    reason = goal.sanitize_reason(
        f"token ghp_{'A' * 24} and https://ci:{SYNTHETIC_HEX}@scm.bstein.dev/x.git"
    )

    assert "ghp_" not in reason
    assert SYNTHETIC_HEX not in reason
    assert reason.count("[redacted]") == 2


def test_redaction_delegates_to_the_canonical_helper_when_it_is_installed(monkeypatch):
    seen = []

    def spy(text, **kwargs):
        seen.append(kwargs)
        return text.replace("OPAQUE", "***")

    monkeypatch.setattr(goal, "_canonical_redact", spy)

    assert goal.sanitize_reason("value OPAQUE here") == "value *** here"
    assert seen == [{"force": True}]


def test_the_canonical_redactor_is_the_one_the_lane_runtime_ships():
    canonical = pytest.importorskip("agent.redact")

    assert goal._canonical_redact is canonical.redact_sensitive_text


def test_the_upstream_worker_context_headings_are_still_the_ones_we_cut_on():
    """Pin ``hermes_cli.kanban_db.build_worker_context``'s H2 contract.

    ``card_scope`` allow-lists CARD_SECTIONS and drops every other H2. A rename
    upstream fails closed rather than leaking history into role resolution, but
    it would also truncate real cards, so the coupling is asserted explicitly.
    """
    kanban_db = pytest.importorskip("hermes_cli.kanban_db")
    source = Path(kanban_db.__file__).read_text(encoding="utf-8")
    emitted = {
        match.group(1).replace("@", "").strip()
        for match in re.finditer(r'lines\.append\(f?"## ([^"{]+)', source)
    }

    assert emitted == {*goal.CARD_SECTIONS, *goal.HISTORY_SECTIONS}


def test_the_controller_evidence_heading_is_not_a_card_section():
    assert goal.CONTROLLER_EVIDENCE_HEADING.startswith("## ")
    assert goal.CONTROLLER_EVIDENCE_HEADING[3:] not in goal.CARD_SECTIONS
    assert goal.card_scope(
        f"## Body\nreview only\n\n{goal.CONTROLLER_EVIDENCE_HEADING}\npush it"
    ) == "## Body\nreview only\n\n"


def test_every_report_shape_round_trips_through_the_result_schema():
    for name, result in REPORTS.items():
        assert set(result) == {
            "status",
            "summary",
            "changed_files",
            "tests_run",
            "artifacts",
            "findings",
            "blockers",
        }, name
        assert result["status"] in goal.RESULT_STATUSES
        assert json.loads(json.dumps(result)) == result