orchestrator/agent-team/tests/test_checker_intake.py

"""Unit tests for agent_team.transport.checker_intake (P5 cross-plane loop).

Fully hermetic: the coordinator is an injected in-memory fake and findings are
plain dicts (or temp report files), so no network call, token, GitHub SDK, model
or live checker is exercised. The tests pin the P5 contract:

* a CONFIRMED finding at/above the threshold creates exactly one task,
* unconfirmed / suppressed / below-threshold findings are ignored,
* re-reading the same finding does not double-ingest (de-dup),
* the rendered task text is safe (no log/text injection from repo-controlled
  fields), and
* the loop is OPT-IN — it is NOT wired into the default coordinator serve path.
"""

from __future__ import annotations

import importlib.util
import json
from pathlib import Path
from typing import Any

import pytest

from agent_team.transport.checker_intake import (
    CHECKER_TRANSPORT_NAME,
    CheckerFindingIntake,
    finding_identity,
    finding_task_text,
    load_report_findings,
    select_findings,
)

# --------------------------------------------------------------------------- #
# Fakes / builders
# --------------------------------------------------------------------------- #


class FakeCoordinator:
    """In-memory coordinator double recording every ``start_task`` call.

    Mirrors the real coordinator's intake entry signature and captures each
    call's keyword args so a test can assert exactly what was ingested.
    """

    def __init__(self) -> None:
        self.calls: list[dict[str, Any]] = []

    def start_task(self, *, task_text: str, transport_name: str) -> str:
        self.calls.append({"task_text": task_text, "transport_name": transport_name})
        return f"thread-{len(self.calls)}"


def _finding(
    *,
    fid: str | None = "repo-a-branchprot-no-pr",
    repo: str = "repo-a",
    title: str = "main does not require a PR for merge",
    severity: str = "high",
    status: str = "confirmed",
    check: str = "branch-protection",
    proof: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """Build a checker-finding-shaped mapping matching the bash checker emit."""
    finding: dict[str, Any] = {
        "repo": repo,
        "title": title,
        "severity": severity,
        "category": "other",
        "check": check,
        "status": status,
        "proof": proof if proof is not None else {"outcome": "require a PR for merges"},
    }
    if fid is not None:
        finding["id"] = fid
    return finding


def _report(
    findings: list[dict[str, Any]], *, checker: str = "compliance-drift"
) -> dict[str, Any]:
    """Wrap findings in the top-level report object the checkers emit."""
    return {
        "checker": checker,
        "generated": "2026-06-18T00:00:00Z",
        "org": "Sea-Haven-Industries",
        "findings": findings,
    }


# --------------------------------------------------------------------------- #
# select_findings — the gate
# --------------------------------------------------------------------------- #


def test_selects_confirmed_at_or_above_threshold() -> None:
    findings = [
        _finding(fid="r-high", severity="high"),
        _finding(fid="r-crit", severity="critical"),
    ]
    selected = select_findings(findings, threshold="high")
    assert [f["id"] for f in selected] == ["r-high", "r-crit"]


def test_ignores_below_threshold() -> None:
    findings = [
        _finding(fid="r-low", severity="low"),
        _finding(fid="r-med", severity="medium"),
        _finding(fid="r-high", severity="high"),
    ]
    selected = select_findings(findings, threshold="high")
    assert [f["id"] for f in selected] == ["r-high"]


def test_ignores_unconfirmed_and_suppressed() -> None:
    findings = [
        _finding(fid="r-unver", status="unverified", severity="critical"),
        _finding(fid="r-supp", status="suppressed", severity="critical"),
        _finding(fid="r-ok", status="confirmed", severity="high"),
    ]
    selected = select_findings(findings, threshold="high")
    assert [f["id"] for f in selected] == ["r-ok"]


def test_unverified_severity_never_meets_threshold() -> None:
    # status confirmed but severity is the schema's downgrade marker.
    findings = [_finding(fid="r-x", status="confirmed", severity="unverified")]
    assert select_findings(findings, threshold="low") == []


def test_threshold_lower_admits_more() -> None:
    findings = [
        _finding(fid="r-low", severity="low"),
        _finding(fid="r-med", severity="medium"),
    ]
    selected = select_findings(findings, threshold="low")
    assert [f["id"] for f in selected] == ["r-low", "r-med"]


def test_invalid_threshold_rejected() -> None:
    for bad in ("info", "unverified", "nope", ""):
        with pytest.raises(ValueError):
            select_findings([], threshold=bad)


# --------------------------------------------------------------------------- #
# finding_task_text / sanitization — untrusted-input hygiene
# --------------------------------------------------------------------------- #


def test_task_text_renders_checker_repo_severity_title_and_hint() -> None:
    text = finding_task_text(
        _finding(
            repo="orchestrator",
            title="vulnerable dependency",
            severity="critical",
            check="vulnerable-dependency",
            proof={
                "package": "requests",
                "version": "2.0.0",
                "fixed_version": "2.32.0",
                "advisory_id": "GHSA-xxxx",
                "summary": "RCE in requests",
            },
        ),
        checker="dependency-cve",
    )
    assert "[Plane-1 dependency-cve]" in text
    assert "orchestrator" in text
    assert "severity=critical" in text
    assert "Finding: vulnerable dependency" in text
    assert "package=requests" in text
    assert "fixed_version=2.32.0" in text


def test_task_text_neutralises_newline_injection_in_title() -> None:
    # A forged title that tries to inject a fake task line / log line.
    evil = "real title\nFinding: SPOOFED\nadmin=true"
    text = finding_task_text(_finding(title=evil))
    # No raw newline from the untrusted field reaches the task text body: the
    # title occupies exactly one line, so the only real newlines are the ones
    # WE add between the headline / finding / hint lines.
    lines = text.split("\n")
    finding_lines = [ln for ln in lines if ln.startswith("Finding: ")]
    assert finding_lines == ["Finding: real title\\nFinding: SPOOFED\\nadmin=true"]


def test_task_text_neutralises_carriage_return_and_controls() -> None:
    evil = "x\r\ty\x00z\x1b[31m"
    text = finding_task_text(_finding(title=evil))
    assert "\r" not in text
    assert "\x00" not in text
    assert "\x1b" not in text
    assert "\\r" in text and "\\t" in text and "\\x00" in text


def test_task_text_bounds_megastring() -> None:
    text = finding_task_text(_finding(title="A" * 10_000))
    assert "…(truncated)" in text
    assert len(text) < 1_000


def test_proof_hint_sanitises_repo_controlled_summary() -> None:
    text = finding_task_text(
        _finding(proof={"outcome": "line1\nline2\rline3"}),
    )
    assert "outcome=line1\\nline2\\rline3" in text
    assert "\n line2" not in text


# --------------------------------------------------------------------------- #
# finding_identity / de-dup
# --------------------------------------------------------------------------- #


def test_identity_prefers_id() -> None:
    assert finding_identity(_finding(fid="repo-a-x")) == "repo-a-x"


def test_identity_falls_back_to_content_hash_without_id() -> None:
    f = _finding(fid=None)
    ident = finding_identity(f)
    assert ident.startswith("sha256:")
    # Stable for identical content.
    assert finding_identity(_finding(fid=None)) == ident


# --------------------------------------------------------------------------- #
# CheckerFindingIntake.ingest_findings — the core contract
# --------------------------------------------------------------------------- #


def test_confirmed_finding_creates_exactly_one_task() -> None:
    coordinator = FakeCoordinator()
    intake = CheckerFindingIntake(coordinator=coordinator)
    ingested = intake.ingest_findings([_finding(fid="repo-a-x")])

    assert ingested == ["repo-a-x"]
    assert len(coordinator.calls) == 1
    call = coordinator.calls[0]
    assert call["transport_name"] == CHECKER_TRANSPORT_NAME
    assert "repo-a" in call["task_text"]


def test_below_threshold_and_unconfirmed_ignored_by_intake() -> None:
    coordinator = FakeCoordinator()
    intake = CheckerFindingIntake(coordinator=coordinator)
    ingested = intake.ingest_findings(
        [
            _finding(fid="r-low", severity="low"),
            _finding(fid="r-unver", status="unverified", severity="critical"),
            _finding(fid="r-ok", severity="high"),
        ]
    )
    assert ingested == ["r-ok"]
    assert len(coordinator.calls) == 1


def test_repeated_finding_does_not_double_ingest() -> None:
    coordinator = FakeCoordinator()
    intake = CheckerFindingIntake(coordinator=coordinator)

    first = intake.ingest_findings([_finding(fid="repo-a-x")])
    second = intake.ingest_findings([_finding(fid="repo-a-x")])

    assert first == ["repo-a-x"]
    assert second == []  # already ingested -> no second task
    assert len(coordinator.calls) == 1
    assert intake.ingested_ids == frozenset({"repo-a-x"})


def test_dedup_within_single_batch() -> None:
    coordinator = FakeCoordinator()
    intake = CheckerFindingIntake(coordinator=coordinator)
    # Same finding present twice in one report batch.
    ingested = intake.ingest_findings(
        [_finding(fid="repo-a-x"), _finding(fid="repo-a-x")]
    )
    assert ingested == ["repo-a-x"]
    assert len(coordinator.calls) == 1


def test_new_finding_on_second_pass_is_ingested() -> None:
    coordinator = FakeCoordinator()
    intake = CheckerFindingIntake(coordinator=coordinator)

    first = intake.ingest_findings([_finding(fid="repo-a-x")])
    second = intake.ingest_findings(
        [_finding(fid="repo-a-x"), _finding(fid="repo-b-y", repo="repo-b")]
    )
    assert first == ["repo-a-x"]
    assert second == ["repo-b-y"]
    assert len(coordinator.calls) == 2


def test_custom_threshold_admits_medium() -> None:
    coordinator = FakeCoordinator()
    intake = CheckerFindingIntake(coordinator=coordinator, threshold="medium")
    ingested = intake.ingest_findings([_finding(fid="r-med", severity="medium")])
    assert ingested == ["r-med"]


def test_failed_start_task_leaves_finding_eligible_for_retry() -> None:
    class FlakyCoordinator:
        def __init__(self) -> None:
            self.attempts = 0

        def start_task(self, *, task_text: str, transport_name: str) -> str:
            self.attempts += 1
            if self.attempts == 1:
                raise RuntimeError("transient intake failure")
            return "thread-ok"

    coordinator = FlakyCoordinator()
    intake = CheckerFindingIntake(coordinator=coordinator)

    with pytest.raises(RuntimeError):
        intake.ingest_findings([_finding(fid="repo-a-x")])
    assert intake.ingested_ids == frozenset()  # not recorded -> retryable

    ingested = intake.ingest_findings([_finding(fid="repo-a-x")])
    assert ingested == ["repo-a-x"]
    assert coordinator.attempts == 2


def test_invalid_threshold_rejected_at_construction() -> None:
    with pytest.raises(ValueError):
        CheckerFindingIntake(coordinator=FakeCoordinator(), threshold="bogus")


# --------------------------------------------------------------------------- #
# load_report_findings / ingest_reports — report-file path
# --------------------------------------------------------------------------- #


def test_load_report_findings_reads_top_level_object(tmp_path: Path) -> None:
    report = tmp_path / "compliance-drift.json"
    report.write_text(json.dumps(_report([_finding(fid="repo-a-x")])), encoding="utf-8")
    findings = load_report_findings(report)
    assert [f["id"] for f in findings] == ["repo-a-x"]


def test_load_report_findings_accepts_bare_array(tmp_path: Path) -> None:
    report = tmp_path / "bare.json"
    report.write_text(json.dumps([_finding(fid="repo-a-x")]), encoding="utf-8")
    assert [f["id"] for f in load_report_findings(report)] == ["repo-a-x"]


def test_load_report_findings_reads_directory_of_reports(tmp_path: Path) -> None:
    (tmp_path / "a.json").write_text(
        json.dumps(_report([_finding(fid="repo-a-x")])), encoding="utf-8"
    )
    (tmp_path / "b.json").write_text(
        json.dumps(_report([_finding(fid="repo-b-y", repo="repo-b")])),
        encoding="utf-8",
    )
    ids = sorted(f["id"] for f in load_report_findings(tmp_path))
    assert ids == ["repo-a-x", "repo-b-y"]


def test_ingest_reports_selects_dedups_across_files(tmp_path: Path) -> None:
    # Two reports both carrying the same high finding plus a unique one each.
    (tmp_path / "a.json").write_text(
        json.dumps(
            _report(
                [
                    _finding(fid="shared", severity="high"),
                    _finding(fid="only-a", repo="repo-a"),
                    _finding(fid="low-a", severity="low"),
                ]
            )
        ),
        encoding="utf-8",
    )
    (tmp_path / "b.json").write_text(
        json.dumps(
            _report([_finding(fid="shared", severity="high"), _finding(fid="only-b")])
        ),
        encoding="utf-8",
    )
    coordinator = FakeCoordinator()
    intake = CheckerFindingIntake(coordinator=coordinator)
    ingested = intake.ingest_reports([tmp_path])

    assert sorted(ingested) == ["only-a", "only-b", "shared"]
    assert len(coordinator.calls) == 3  # low-a dropped, shared once


def test_empty_report_file_yields_nothing(tmp_path: Path) -> None:
    report = tmp_path / "empty.json"
    report.write_text("", encoding="utf-8")
    assert load_report_findings(report) == []


# --------------------------------------------------------------------------- #
# OPT-IN: not wired into the default serve path
# --------------------------------------------------------------------------- #


def _load_run_team():
    """Import run-team.py (hyphenated, so loaded by path) as a module."""
    cli_path = Path(__file__).resolve().parent.parent / "run-team.py"
    spec = importlib.util.spec_from_file_location("run_team_cli", cli_path)
    assert spec is not None and spec.loader is not None
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


def test_intake_checker_is_a_subcommand_not_the_default() -> None:
    cli = _load_run_team()
    parser = cli.build_parser()
    # The subcommand exists...
    args = parser.parse_args(
        ["intake-checker", "--report", "/tmp/x.json", "--threshold", "high"]
    )
    assert args.func is cli._cmd_intake_checker
    assert args.command == "intake-checker"


def test_serve_path_does_not_invoke_checker_intake() -> None:
    """The always-on serve loop must not reference checker intake (opt-in only)."""
    cli = _load_run_team()
    import inspect

    serve_src = inspect.getsource(cli._cmd_serve)
    assert "checker" not in serve_src.lower()
    assert "CheckerFindingIntake" not in serve_src

    # And coordinator.serve itself does not pull in the checker intake module.
    from agent_team import coordinator as coordinator_mod

    coord_src = inspect.getsource(coordinator_mod.Coordinator.serve)
    assert "checker_intake" not in coord_src
    assert "CheckerFindingIntake" not in coord_src