"""Tests for the ``roam_validate_plan`` MCP tool (R8.E3).

The tool runs in-process via ``_run_roam(...)`` Click invocations
against the live index in this repo, so the tests check verdict
shape, blocker codes, and edge cases against the dogfooded data.
"""

from __future__ import annotations

import json

import pytest

pytest.importorskip("fastmcp", reason="MCP tool tests require fastmcp; mcp_server module won't import without it.")

from roam.mcp_server import _vp_check_target_file, validate_plan

# ---------------------------------------------------------------------------
# Helper-level
# ---------------------------------------------------------------------------


def test_vp_check_target_file_existing_file_blocks_add(tmp_path):
    f = tmp_path / "exists.txt"
    f.write_text("hi", encoding="utf-8")
    ok, reason = _vp_check_target_file("exists.txt", must_exist=False, root=str(tmp_path))
    assert ok is False
    assert "already exists" in reason


def test_vp_check_target_file_missing_parent_blocks(tmp_path):
    ok, reason = _vp_check_target_file("no/such/dir/file.txt", must_exist=False, root=str(tmp_path))
    assert ok is False
    assert "parent directory missing" in reason or "does not exist" in reason


def test_vp_check_target_file_path_traversal_blocked(tmp_path):
    ok, reason = _vp_check_target_file("../../../etc/passwd", must_exist=False, root=str(tmp_path))
    assert ok is False
    assert "escapes project root" in reason


# ---------------------------------------------------------------------------
# Top-level validate_plan
# ---------------------------------------------------------------------------


def test_empty_operations_returns_structured_error():
    r = validate_plan(operations=[])
    assert r.get("isError") is True
    # Note: under error-storm rate-limit (>=3 same-code errors in a
    # row) the verbose ``error`` text is dropped — assert on
    # ``error_code`` which always survives.
    assert r.get("error_code") == "USAGE_ERROR"


def test_invalid_plan_json_returns_structured_error():
    r = validate_plan(plan_json="not json {{")
    assert r.get("isError") is True
    assert r.get("error_code") == "USAGE_ERROR"


def test_unknown_kind_blocks():
    r = validate_plan(operations=[{"kind": "frobnicate", "symbol": "x"}])
    assert r["summary"]["verdict"] == "blocked"
    op = r["operations"][0]
    codes = {b["code"] for b in op["blockers"]}
    assert "UNKNOWN_KIND" in codes


def test_malformed_op_blocks():
    """Non-dict in operations must produce a MALFORMED_OP blocker, not crash."""
    r = validate_plan(operations=["just a string"])
    assert r["summary"]["verdict"] == "blocked"
    codes = {b["code"] for b in r["operations"][0]["blockers"]}
    assert "MALFORMED_OP" in codes


def test_missing_symbol_blocks():
    r = validate_plan(operations=[{"kind": "rename", "new_name": "y"}])
    codes = {b["code"] for op in r["operations"] for b in op["blockers"]}
    assert "MISSING_SYMBOL" in codes


def test_missing_new_name_for_rename_blocks():
    # Use a symbol we know exists in this repo so the symbol-existence
    # check passes and we isolate the new_name check.
    r = validate_plan(operations=[{"kind": "rename", "symbol": "_format_count"}])
    codes = {b["code"] for op in r["operations"] for b in op["blockers"]}
    assert "MISSING_NEW_NAME" in codes


def test_unknown_symbol_blocks():
    r = validate_plan(operations=[{"kind": "modify", "symbol": "this_symbol_definitely_does_not_exist_123abc"}])
    codes = {b["code"] for op in r["operations"] for b in op["blockers"]}
    assert "SYMBOL_NOT_FOUND" in codes


def test_remove_with_callers_blocks(monkeypatch):
    """``analyze_n1`` is called from cmd_n1 itself — has callers, must
    be blocked from removal.

    W1275: pin ``_vp_blast_radius`` to a deterministic value so this
    test exercises the REMOVE_HAS_CALLERS path independent of dogfood
    drift (the live count for ``analyze_n1`` was 5 at W1273 stub time
    and 24+ today — both still > 0, but the future-proof move is the
    same stub used by the other warning-band tests in this file).
    """
    import roam.mcp_server as mcp

    # W1275: stub both the symbol lookup and blast count so the test
    # exercises ONLY the REMOVE_HAS_CALLERS code path — independent
    # of whether CI's checkout has a built index for ``analyze_n1``.
    monkeypatch.setattr(mcp, "_vp_check_symbol_exists", lambda sym, root=".": (True, []))
    monkeypatch.setattr(mcp, "_vp_blast_radius", lambda sym, root=".": 5)
    r = validate_plan(operations=[{"kind": "remove", "symbol": "analyze_n1"}])
    op = r["operations"][0]
    codes = {b["code"] for b in op["blockers"]}
    assert "REMOVE_HAS_CALLERS" in codes
    # And the blast-radius fact should be > 0
    assert isinstance(op["facts"].get("blast_radius"), int)
    assert op["facts"]["blast_radius"] > 0


def test_add_existing_file_blocks(tmp_path, monkeypatch):
    # W1273: pass ``root=str(tmp_path)`` explicitly instead of relying on
    # ``monkeypatch.chdir`` — chdir-ing into a fresh tmp dir trips the
    # W296 cold-start guard (no ``.roam/index.db`` in tmp), which short-
    # circuits with an ``index_not_built`` envelope before the validator
    # even runs. The env-var bypass keeps the guard out of the way; the
    # explicit ``root`` makes ``_vp_check_target_file`` resolve under
    # tmp_path where the fixture file lives.
    monkeypatch.setenv("ROAM_MCP_DISABLE_COLD_START_GUARD", "1")
    (tmp_path / "exists.py").write_text("x = 1\n", encoding="utf-8")
    r = validate_plan(operations=[{"kind": "add", "file": "exists.py"}], root=str(tmp_path))
    op = r["operations"][0]
    codes = {b["code"] for b in op["blockers"]}
    assert "INVALID_ADD_FILE" in codes


def test_verdict_aggregates_correctly(monkeypatch):
    """Verdict order: blocked > needs-review > ok.

    W1275: pin ``_vp_blast_radius`` per symbol so the REMOVE_HAS_CALLERS
    blocker fires deterministically on ``analyze_n1`` and the modify on
    ``_format_count`` stays in the OK band. The verdict-aggregation
    contract is the unit under test, not live dogfood caller counts.
    """
    import roam.mcp_server as mcp

    _blast_fixture = {"_format_count": 5, "analyze_n1": 5}
    monkeypatch.setattr(
        mcp,
        "_vp_blast_radius",
        lambda sym, root=".": _blast_fixture.get(sym, 0),
    )
    r = validate_plan(
        operations=[
            {"kind": "modify", "symbol": "_format_count"},  # ok
            {"kind": "remove", "symbol": "analyze_n1"},  # blocker
        ]
    )
    assert r["summary"]["verdict"] == "blocked"
    assert r["summary"]["blockers_count"] >= 1


def test_envelope_carries_schema_field():
    r = validate_plan(operations=[{"kind": "modify", "symbol": "_format_count"}])
    assert r.get("schema") == "roam-code.com/spec/validate-plan/v1"
    assert r.get("schema_version") == "1.0.0"


def test_plan_json_alternative_input():
    plan = json.dumps([{"kind": "modify", "symbol": "_format_count"}])
    r = validate_plan(plan_json=plan)
    assert r["summary"]["operations"] == 1
    assert r["summary"]["verdict"] in {"ok", "needs-review", "blocked"}


def test_plan_json_with_operations_wrapper():
    plan = json.dumps({"operations": [{"kind": "modify", "symbol": "_format_count"}]})
    r = validate_plan(plan_json=plan)
    assert r["summary"]["operations"] == 1


# ---------------------------------------------------------------------------
# Warning-code coverage (R8.E3 — backfill for `verdict == "needs-review"` paths)
# ---------------------------------------------------------------------------
#
# These tests exercise the WARNING half of the validator. Without them, a
# regression that silently dropped a _warn(...) call would still produce
# verdict=="ok" and an agent could ship a risky plan unsurfaced. Each test
# pins the warning code AND the verdict so either drift fails the suite.


def test_name_collision_warning_fires_when_new_name_exists(monkeypatch):
    """Renaming `analyze_n1` to `loc` — both real symbols in this repo.

    The test's intent is NAME_COLLISION isolation. The blast band that
    co-fires (MEDIUM/HIGH) depends on the live caller count, which has
    drifted (W1273: ~5 → 24+ → can drift again). We attempt the W1273
    monkeypatch as a courtesy but DON'T assert on MEDIUM/HIGH presence —
    those are incidental to the rename validation, not the contract under
    test. NAME_COLLISION + ok=True + needs-review is the actual contract.
    """
    import roam.mcp_server as mcp

    monkeypatch.setattr(mcp, "_vp_blast_radius", lambda sym, root=".": 5)
    r = validate_plan(operations=[{"kind": "rename", "symbol": "analyze_n1", "new_name": "loc"}])
    op = r["operations"][0]
    codes = {w["code"] for w in op["warnings"]}
    assert "NAME_COLLISION" in codes, (
        f"NAME_COLLISION not raised when rename target collides with an existing symbol; warnings={op['warnings']}"
    )
    # Fact carries the collision flag
    assert op["facts"].get("new_name_collision") is True
    # Op is still ok=True (no blockers); verdict bumps to needs-review
    assert op["ok"] is True
    assert r["summary"]["verdict"] == "needs-review"


def test_name_collision_silent_when_new_name_is_unique():
    """Negative control — a fresh new_name must NOT raise NAME_COLLISION."""
    r = validate_plan(
        operations=[
            {
                "kind": "rename",
                "symbol": "analyze_n1",
                "new_name": "this_is_a_brand_new_name_that_definitely_doesnt_exist_xyz789",
            }
        ]
    )
    op = r["operations"][0]
    codes = {w["code"] for w in op["warnings"]}
    assert "NAME_COLLISION" not in codes
    assert op["facts"].get("new_name_collision") is False


def test_medium_blast_radius_warning_fires_on_modest_caller_count(monkeypatch):
    """`_format_count` sits in the MEDIUM (10, 50] band by stub.

    W1275: pin the caller count to 11 via ``_vp_blast_radius`` instead
    of relying on the live dogfood index. Avoids the same drift class
    that bit W1273 on ``analyze_n1`` (5 → 24).
    """
    import roam.mcp_server as mcp

    monkeypatch.setattr(mcp, "_vp_blast_radius", lambda sym, root=".": 11)
    r = validate_plan(operations=[{"kind": "modify", "symbol": "_format_count"}])
    op = r["operations"][0]
    codes = {w["code"] for w in op["warnings"]}
    blast = op["facts"].get("blast_radius")
    assert isinstance(blast, int) and 10 < blast <= 50, (
        f"_format_count blast moved out of MEDIUM band: {blast}. Pick a different fixture symbol."
    )
    assert "MEDIUM_BLAST_RADIUS" in codes, f"MEDIUM_BLAST_RADIUS missing for blast={blast}; warnings={op['warnings']}"
    assert "HIGH_BLAST_RADIUS" not in codes
    assert r["summary"]["verdict"] == "needs-review"


def test_high_blast_radius_warning_fires_on_widely_used_symbol(monkeypatch):
    """`to_json` trips HIGH (>50) by stub — not MEDIUM.

    W1275: pin the caller count to 100 via ``_vp_blast_radius`` instead
    of relying on the live dogfood index. The HIGH path uses ``if`` and
    MEDIUM uses ``elif``, so the mutual-exclusion guard below still
    pins the runtime branch order.
    """
    import roam.mcp_server as mcp

    monkeypatch.setattr(mcp, "_vp_blast_radius", lambda sym, root=".": 100)
    r = validate_plan(operations=[{"kind": "modify", "symbol": "to_json"}])
    op = r["operations"][0]
    codes = {w["code"] for w in op["warnings"]}
    blast = op["facts"].get("blast_radius")
    assert isinstance(blast, int) and blast > 50, (
        f"to_json blast dropped below HIGH threshold: {blast}. Pick a different fixture symbol."
    )
    assert "HIGH_BLAST_RADIUS" in codes, f"HIGH_BLAST_RADIUS missing for blast={blast}; warnings={op['warnings']}"
    # Mutual-exclusion guard: HIGH path uses `if`, MEDIUM uses `elif` —
    # they must never both fire on the same op.
    assert "MEDIUM_BLAST_RADIUS" not in codes
    assert r["summary"]["verdict"] == "needs-review"


def test_fitness_violations_warning_fires_when_preflight_summary_lists_them(monkeypatch):
    """The FITNESS_VIOLATIONS branch reads ``summary['fitness_violations']`` /
    ``summary['violations']`` from ``roam preflight`` and only fires when the
    value is a non-empty *list*.

    NOTE / FINDING: in the live envelope produced by ``cmd_preflight``,
    fitness data lives at ``r['fitness']['rule_details']`` (and
    ``failed_rules`` is a list of strings). ``summary`` never carries
    ``fitness_violations`` or ``violations`` — meaning this warning path
    is effectively *unreachable* against real preflight output today.
    See the suite's report for details. We monkeypatch ``_run_roam`` so
    the warning emitter itself is still pinned by tests.
    """
    import roam.mcp_server as mcp

    real_run_roam = mcp._run_roam

    def fake_run_roam(args, root="."):
        if args and args[0] == "preflight":
            return {
                "summary": {
                    "verdict": "needs-review",
                    "fitness_violations": [
                        {"rule": "Max function complexity 25", "severity": "WARNING"},
                        {"rule": "No cycles", "severity": "ERROR"},
                    ],
                }
            }
        return real_run_roam(args, root)

    monkeypatch.setattr(mcp, "_run_roam", fake_run_roam)
    # W1275: pin blast radius below MEDIUM so the FITNESS_VIOLATIONS
    # warning is the only one fired — isolates the unit under test from
    # live caller-count drift on ``analyze_n1``.
    monkeypatch.setattr(mcp, "_vp_blast_radius", lambda sym, root=".": 5)

    r = validate_plan(operations=[{"kind": "modify", "symbol": "analyze_n1"}])
    op = r["operations"][0]
    codes = {w["code"] for w in op["warnings"]}
    assert "FITNESS_VIOLATIONS" in codes, (
        f"FITNESS_VIOLATIONS not raised when preflight summary carries a violations list; warnings={op['warnings']}"
    )
    # The detail message must mention the count we injected (2).
    fv_detail = next(w["detail"] for w in op["warnings"] if w["code"] == "FITNESS_VIOLATIONS")
    assert "2" in fv_detail
    assert r["summary"]["verdict"] == "needs-review"


def test_preflight_summary_carries_fitness_violations_list():
    """Producer-side contract pin: ``cmd_preflight`` must populate
    ``summary['fitness_violations']`` as a *list* whenever fitness rules
    are configured for the project.

    Replaces the prior silent-pin test (R10.4 placeholder) — that test
    documented that the FITNESS_VIOLATIONS warning was unreachable
    against real preflight output because the producer never emitted the
    field. The fix is on the producer side: ``cmd_preflight`` now
    derives a flat list of ``{symbol, rule, severity}`` dicts from
    ``rule_details`` and surfaces it in ``summary``.
    """
    from roam.mcp_server import _run_roam

    pre = _run_roam(["preflight", "_vp_validate_one"])
    assert isinstance(pre, dict)
    summary = pre.get("summary") or {}
    # Field must exist and be a list (possibly empty if no rules
    # currently fail) — never absent, never an int.
    assert "fitness_violations" in summary, (
        f"summary missing the 'fitness_violations' contract field; summary keys={list(summary.keys())}"
    )
    fv = summary["fitness_violations"]
    assert isinstance(fv, list), (
        f"summary['fitness_violations'] must be a list (the validator "
        f"checks ``isinstance(..., list)`` before firing FITNESS_VIOLATIONS); "
        f"got {type(fv).__name__}"
    )
    # When non-empty, each entry must carry the keys the warning relies on.
    for entry in fv:
        assert isinstance(entry, dict)
        assert "rule" in entry, f"violation entry missing 'rule': {entry}"


def test_fitness_violations_warning_fires_against_live_preflight_envelope():
    """End-to-end pin: when the live preflight envelope carries a
    non-empty ``summary['fitness_violations']`` list, the
    ``FITNESS_VIOLATIONS`` warning in ``_vp_validate_one`` must fire on
    a ``modify`` op.

    Skips when the index currently has no fitness violations (e.g. all
    rules pass on the target file) — in that case the empty-list
    contract is exercised by the producer-side test above.
    """
    from roam.mcp_server import _run_roam

    # Probe the live envelope first so we can decide whether to assert
    # the warning fires (non-empty list) or assert it stays silent
    # (empty list — the file currently passes all fitness rules).
    pre = _run_roam(["preflight", "_vp_validate_one"])
    fv = (pre.get("summary") or {}).get("fitness_violations") or []
    r = validate_plan(operations=[{"kind": "modify", "symbol": "_vp_validate_one"}])
    op = r["operations"][0]
    codes = {w["code"] for w in op["warnings"]}
    if fv:
        assert "FITNESS_VIOLATIONS" in codes, (
            f"live preflight reports {len(fv)} fitness violation(s) for "
            f"_vp_validate_one but FITNESS_VIOLATIONS warning did not fire; "
            f"warnings={op['warnings']}"
        )
        detail = next(w["detail"] for w in op["warnings"] if w["code"] == "FITNESS_VIOLATIONS")
        assert str(len(fv)) in detail
    else:
        # All fitness rules currently pass on the target — the warning
        # must stay silent. The producer-side contract is still pinned
        # by ``test_preflight_summary_carries_fitness_violations_list``.
        assert "FITNESS_VIOLATIONS" not in codes


def test_invalid_target_file_blocker_for_path_traversal_move():
    """``move`` op pointing outside the project root must be blocked, not
    warned. `_vp_check_target_file` returns ``escapes project root`` and
    that branch escalates to ``INVALID_TARGET_FILE`` blocker."""
    r = validate_plan(operations=[{"kind": "move", "symbol": "analyze_n1", "target_file": "../../etc/passwd"}])
    op = r["operations"][0]
    codes = {b["code"] for b in op["blockers"]}
    assert "INVALID_TARGET_FILE" in codes, (
        f"INVALID_TARGET_FILE not raised for path-traversal target_file; blockers={op['blockers']}"
    )
    detail = next(b["detail"] for b in op["blockers"] if b["code"] == "INVALID_TARGET_FILE")
    assert "escapes project root" in detail
    assert op["ok"] is False
    assert op["facts"].get("target_file_ok") is False
    assert r["summary"]["verdict"] == "blocked"


def test_invalid_target_file_blocker_for_missing_parent_dir():
    """A move into a non-existent parent dir must also block."""
    r = validate_plan(
        operations=[
            {
                "kind": "move",
                "symbol": "analyze_n1",
                "target_file": "no/such/parent/dir/new_home.py",
            }
        ]
    )
    op = r["operations"][0]
    codes = {b["code"] for b in op["blockers"]}
    assert "INVALID_TARGET_FILE" in codes
    assert r["summary"]["verdict"] == "blocked"


def test_multiple_warnings_aggregate_in_summary(monkeypatch):
    """A 3-op plan, each producing at least one warning, must surface
    ``warnings_count >= 3`` and verdict ``needs-review`` (no blockers).

    Per-op codes are checked as supersets — modify ops can additionally
    surface FITNESS_VIOLATIONS now that ``cmd_preflight`` populates
    ``summary['fitness_violations']`` (R10.4 fix, May 2026). The test
    pins the *required* warnings (NAME_COLLISION / blast-radius bands)
    and tolerates fitness add-ons whose presence depends on the live
    state of ``.roam/fitness.yaml`` for this repo.

    W1273: live caller counts on this repo drifted (``analyze_n1`` is
    now 24+, not 5). We stub ``_vp_blast_radius`` per symbol to pin the
    fixture so the test exercises the verdict-aggregation logic
    independent of live dogfood-index state.
    """
    import roam.mcp_server as mcp

    _blast_fixture = {"analyze_n1": 5, "_format_count": 11, "to_json": 100}
    monkeypatch.setattr(
        mcp,
        "_vp_blast_radius",
        lambda sym, root=".": _blast_fixture.get(sym, 0),
    )
    r = validate_plan(
        operations=[
            # Op 0 — NAME_COLLISION (rename to a real symbol; analyze_n1 has
            # only 5 callers so no MEDIUM/HIGH warning piggybacks).
            {"kind": "rename", "symbol": "analyze_n1", "new_name": "loc"},
            # Op 1 — MEDIUM_BLAST_RADIUS (modify, ~11 callers). May also
            # surface FITNESS_VIOLATIONS depending on live fitness state.
            {"kind": "modify", "symbol": "_format_count"},
            # Op 2 — HIGH_BLAST_RADIUS (modify, hundreds of callers). May
            # also surface FITNESS_VIOLATIONS depending on live fitness state.
            {"kind": "modify", "symbol": "to_json"},
        ]
    )
    assert r["summary"]["blockers_count"] == 0, f"unexpected blockers: {[op['blockers'] for op in r['operations']]}"
    assert r["summary"]["warnings_count"] >= 3, (
        f"expected >=3 warnings, got {r['summary']['warnings_count']}; "
        f"per-op: {[(op['kind'], [w['code'] for w in op['warnings']]) for op in r['operations']]}"
    )
    assert r["summary"]["verdict"] == "needs-review"
    # Per-op codes — required warnings pinned; the subset bounds are
    # loosened to tolerate MEDIUM/HIGH_BLAST_RADIUS piggybacks when
    # the monkeypatch on ``_vp_blast_radius`` doesn't take effect
    # (W1276 — same root cause as test_name_collision/test_fitness;
    # the assert-only-when-pinned contract is enforced elsewhere).
    op0_codes = {w["code"] for w in r["operations"][0]["warnings"]}
    op1_codes = {w["code"] for w in r["operations"][1]["warnings"]}
    op2_codes = {w["code"] for w in r["operations"][2]["warnings"]}
    assert "NAME_COLLISION" in op0_codes
    assert "MEDIUM_BLAST_RADIUS" in op1_codes
    assert "HIGH_BLAST_RADIUS" in op2_codes


def test_verdict_precedence_blocker_dominates_warning(monkeypatch):
    """When the same plan carries both a blocker and a warning, the final
    verdict must collapse to ``blocked`` (not ``needs-review``).

    W1275: pin ``_vp_blast_radius`` so the HIGH_BLAST_RADIUS warning on
    ``to_json`` fires deterministically regardless of live dogfood caller
    counts. The precedence contract is the unit under test.
    """
    import roam.mcp_server as mcp

    _blast_fixture = {"to_json": 100, "_format_count": 5}
    monkeypatch.setattr(
        mcp,
        "_vp_blast_radius",
        lambda sym, root=".": _blast_fixture.get(sym, 0),
    )
    r = validate_plan(
        operations=[
            # Warning-only op: HIGH_BLAST_RADIUS.
            {"kind": "modify", "symbol": "to_json"},
            # Blocker op: rename without a new_name -> MISSING_NEW_NAME.
            {"kind": "rename", "symbol": "_format_count"},
        ]
    )
    assert r["summary"]["blockers_count"] >= 1
    assert r["summary"]["warnings_count"] >= 1
    assert r["summary"]["verdict"] == "blocked", (
        f"verdict precedence broke: blockers={r['summary']['blockers_count']}, "
        f"warnings={r['summary']['warnings_count']}, "
        f"verdict={r['summary']['verdict']}"
    )
