"""Behavioral tests for the supervisor's pure decision state machine.

Every transition, dedup guard, fail-closed branch, and bound is exercised on
real policy logic (no mock-asserting): implementation-done->review, SHIP->ready,
BLOCK->repair, repair->re-review, per-(parent, commit) dedup including a
concurrent external shepherd, ambiguous/unparseable fail-closed, and the
cycle-limit escalation.
"""

from __future__ import annotations

import json

from types import SimpleNamespace

from testing.tests.test_hermes_cli_support import _load

policy = _load("supervisor_policy")

LIMITS = policy.Limits(max_cycles=3, max_chains=5)


def task(**kw):
    base = {
        "id": "t1",
        "status": "done",
        "assignee": "",
        "title": "",
        "body": "",
        "result": None,
        "metadata": {},
        "parents": [],
    }
    result = kw.get("result")
    if "metadata" not in kw and isinstance(result, dict) and result.get("pull_request"):
        base["metadata"] = {"assignment": {"branch": "feature/impl", "base_branch": "main",
                                              "pull_request": result["pull_request"]}}
    base.update(kw)
    return SimpleNamespace(**base)


def review_stamp(root="impl", commit="c1", cycle=1):
    return {
        "supervisor": {
            "kind": "review",
            "root": root,
            "parent": root,
            "head_commit": commit,
            "cycle": cycle,
            "root_task_id": root,
            "branch": "feature/impl",
            "pull_request": "pr/1",
            "base_branch": "main",
        }
    }


def repair_stamp(root="impl", commit="c1", cycle=1):
    return {
        "supervisor": {
            "kind": "repair",
            "root": root,
            "parent": "rev",
            "head_commit": commit,
            "cycle": cycle,
            "root_task_id": root,
            "branch": "feature/impl",
            "pull_request": "pr/1",
            "base_branch": "main",
        }
    }


def trusted_root(root="impl", branch="feature/impl", pull_request="pr/1"):
    return task(id=root, status="ready", metadata={"supervisor_lineage": {
        "root_task_id": root, "branch": branch, "pull_request": pull_request,
        "base_branch": "main"
    }})


# --- primitives -----------------------------------------------------------


def test_parse_result_fails_closed_on_absent_and_non_object():
    assert policy.parse_result(task(result=None))[1]
    assert policy.parse_result(task(result=""))[1]
    assert policy.parse_result(task(result={}))[1]
    assert policy.parse_result(task(result="not json"))[1]
    assert policy.parse_result(task(result="[1, 2]"))[1]


def test_parse_result_accepts_dict_and_json_string():
    parsed, error = policy.parse_result(task(result={"status": "completed"}))
    assert error is None and parsed == {"status": "completed"}
    parsed, error = policy.parse_result(task(result='{"status": "completed"}'))
    assert error is None and parsed["status"] == "completed"


def test_metadata_parses_json_string_and_rejects_scalar():
    assert policy.metadata(task(metadata='{"a": 1}')) == {"a": 1}
    assert policy.metadata(task(metadata="oops")) == {}
    assert policy.metadata(task(metadata="7")) == {}


def test_extract_commit_and_pr_scan_metadata_result_and_nested():
    from_meta = task(metadata={"head_commit": "m"}, result={})
    assert policy.extract_commit(from_meta, {}) == "m"
    nested = task(result={"metadata": {"commit": "n"}})
    parsed, _ = policy.parse_result(nested)
    assert policy.extract_commit(nested, parsed) == "n"
    assert policy.extract_commit(task(result={"status": "x"}), {"status": "x"}) is None
    assert policy.extract_pr(task(result={"branch": "feat"}), {"branch": "feat"}) == "feat"
    assert policy.extract_pr(task(result={"pr_url": "u"}), {"pr_url": "u"}) == "u"
    assert policy.extract_pr(task(result={}), {}) is None


def test_parents_reads_lists_dicts_and_task_links_fallback():
    assert policy.parents(task(parents=["a", "b"])) == ["a", "b"]
    linked = task(parents=None, task_links={"parents": [{"id": "p1"}, {"parent": "p2"}]})
    assert policy.parents(linked) == ["p1", "p2"]
    assert policy.parents(task(parents={"parents": ["z"]})) == ["z"]
    assert policy.parents(task(parents=None, task_links=None)) == []


def test_assert_safe_rejects_unknown_action_and_forbidden_keys():
    assert policy.assert_safe(policy.Decision("none")).action == "none"
    try:
        policy.assert_safe(policy.Decision("merge"))
        raise AssertionError("expected rejection")
    except ValueError:
        pass
    try:
        policy.assert_safe(policy.Decision("spawn", payload={"approve": True}))
        raise AssertionError("expected rejection")
    except ValueError:
        pass


# --- implementation -> review --------------------------------------------


def test_done_implementation_with_pr_spawns_one_review():
    impl = task(
        id="impl",
        result={"changed_files": ["a.py"], "head_commit": "c1", "pull_request": "pr/1"},
    )
    decision = policy.plan(impl, [impl], LIMITS)
    assert decision.action == "spawn" and decision.target_id == "impl"
    payload = decision.payload
    assert payload["assignee"] == LIMITS.review_assignee
    assert payload["parents"] == ["impl"]
    assert payload["idempotency_key"] == "supervisor:review:impl:c1:1"
    stamp = payload["metadata"]["supervisor"]
    assert stamp["root_task_id"] == "impl"
    assert stamp["branch"] == "feature/impl" and stamp["pull_request"] == "pr/1"
    assert "Hermes-Task-Role: review" in payload["body"]
    assert payload["metadata"]["task_role"] == "review"


def test_non_terminal_task_is_never_acted_on():
    impl = task(id="impl", status="running", result={"head_commit": "c1"})
    assert policy.plan(impl, [impl], LIMITS).action == "none"


def test_unparseable_implementation_result_fails_closed():
    impl = task(id="impl", result="broken")
    decision = policy.plan(impl, [impl], LIMITS)
    assert decision.action == "escalate" and decision.target_id == "impl"


def test_done_implementation_with_changes_but_no_commit_fails_closed():
    impl = task(id="impl", result={"changed_files": ["a.py"]})
    assert policy.plan(impl, [impl], LIMITS).action == "escalate"


def test_done_implementation_with_no_commit_and_no_changes_is_idle():
    impl = task(id="impl", result={"status": "completed", "summary": "nothing to do"})
    assert policy.plan(impl, [impl], LIMITS).action == "none"


def test_commit_without_pr_fails_closed():
    impl = task(id="impl", result={"changed_files": ["a.py"], "head_commit": "c1"})
    assert policy.plan(impl, [impl], LIMITS).action == "escalate"


def test_review_shaped_unstamped_card_is_left_to_the_shepherd():
    card = task(
        id="ext",
        body="Hermes-Task-Role: review\nRead-only review, report SHIP or BLOCK.",
        result={"verdict": "SHIP"},
    )
    assert policy.plan(card, [card], LIMITS).action == "none"


def test_existing_supervisor_review_blocks_duplicate():
    impl = task(
        id="impl",
        result={"changed_files": ["a.py"], "head_commit": "c1", "pull_request": "pr/1"},
    )
    review = task(id="rev", status="ready", metadata=review_stamp("impl", "c1"))
    assert policy.plan(impl, [impl, review], LIMITS).action == "none"


def test_existing_external_shepherd_review_blocks_duplicate():
    impl = task(
        id="impl",
        result={"changed_files": ["a.py"], "head_commit": "c1", "pull_request": "pr/1"},
    )
    shepherd = task(
        id="ext",
        status="ready",
        title="Review of the change",
        body="Read-only review of commit c1; report SHIP or BLOCK.",
        parents=["impl"],
        assignee="cli-claude-xhigh",
    )
    assert policy.plan(impl, [impl, shepherd], LIMITS).action == "none"


def test_chain_ceiling_defers_new_reviews():
    impl = task(
        id="impl",
        result={"changed_files": ["a.py"], "head_commit": "c1", "pull_request": "pr/1"},
    )
    active = [
        task(id=f"r{i}", status="ready", metadata=review_stamp(f"root{i}", "x"))
        for i in range(LIMITS.max_chains)
    ]
    assert policy.plan(impl, [impl, *active], LIMITS).action == "none"


# --- review -> ship / repair ---------------------------------------------


def test_review_ship_marks_parent_ready_without_merging():
    review = task(
        id="rev",
        metadata=review_stamp("impl", "c1", 1),
        result={"verdict": "SHIP", "summary": "clean"},
    )
    decision = policy.plan(review, [trusted_root(), review], LIMITS)
    assert decision.action == "ship" and decision.target_id == "impl"
    assert decision.payload["commit"] == "c1"
    assert decision.payload["pr"] == "pr/1"
    assert decision.payload["branch"] == "feature/impl"


def test_review_block_spawns_bounded_repair_with_findings():
    review = task(
        id="rev",
        metadata=review_stamp("impl", "c1", 1),
        result={"verdict": "BLOCK", "findings": ["null deref", "missing test"]},
    )
    decision = policy.plan(review, [trusted_root(), review], LIMITS)
    assert decision.action == "spawn"
    payload = decision.payload
    assert payload["metadata"]["supervisor"]["kind"] == "repair"
    assert payload["metadata"]["supervisor"]["cycle"] == 1
    assert payload["parents"] == ["impl", "rev"]
    assert payload["assignee"] == LIMITS.repair_assignee
    assert "null deref" in payload["body"] and "missing test" in payload["body"]
    assert payload["idempotency_key"] == "supervisor:repair:impl:c1:1"


def test_review_with_incomplete_stamp_fails_closed():
    review = task(id="rev", metadata={"supervisor": {"kind": "review"}}, result={"verdict": "SHIP"})
    assert policy.plan(review, [review], LIMITS).action == "escalate"


def test_review_unparseable_result_fails_closed():
    review = task(id="rev", metadata=review_stamp(), result="nope")
    assert policy.plan(review, [review], LIMITS).action == "escalate"


def test_review_ambiguous_verdict_fails_closed():
    review = task(
        id="rev",
        metadata=review_stamp(),
        result={"status": "completed", "summary": "looked at it"},
    )
    assert policy.plan(review, [review], LIMITS).action == "escalate"


def test_review_block_at_cycle_limit_escalates_the_source_review_not_impl():
    review = task(
        id="rev",
        metadata=review_stamp("impl", "c1", LIMITS.max_cycles),
        result={"verdict": "BLOCK", "findings": ["x"]},
    )
    decision = policy.plan(review, [trusted_root(), review], LIMITS)
    # Targets the review card (source), not root_id, so it leaves the done state
    # and is not re-planned into the same escalation next tick.
    assert decision.action == "escalate" and decision.target_id == "rev"
    assert "impl" in decision.reason


def test_review_block_skips_when_repair_already_exists():
    review = task(
        id="rev",
        metadata=review_stamp("impl", "c1", 1),
        result={"verdict": "BLOCK", "findings": ["x"]},
    )
    repair = task(id="rep", status="ready", metadata=repair_stamp("impl", "c1", 1))
    assert policy.plan(review, [trusted_root(), review, repair], LIMITS).action == "none"


# --- repair -> re-review --------------------------------------------------


def test_repair_with_new_commit_spawns_re_review_at_next_cycle():
    repair = task(
        id="rep",
        metadata=repair_stamp("impl", "c1", 1),
        result={"changed_files": ["a.py"], "head_commit": "c2", "branch": "feat"},
    )
    parent = task(id="rev", metadata=review_stamp("impl", "c1", 1), result={"verdict": "BLOCK"})
    decision = policy.plan(repair, [trusted_root(), parent, repair], LIMITS)
    assert decision.action == "spawn"
    payload = decision.payload
    stamp = payload["metadata"]["supervisor"]
    assert stamp["kind"] == "review" and stamp["cycle"] == 2 and stamp["head_commit"] == "c2"
    assert payload["parents"] == ["impl", "rep"]
    assert payload["idempotency_key"] == "supervisor:review:impl:c2:2"


def test_block_repair_and_rereview_keep_one_immutable_pr_lineage():
    impl = task(
        id="impl",
        metadata={"assignment": {"root_task_id": "impl", "project": "atlas/iac",
                                  "branch": "feature/fix", "pull_request": "pr/9", "base_branch": "main"}},
        result={"changed_files": ["a.py"], "head_commit": "c1"},
    )
    review_payload = policy.plan(impl, [impl], LIMITS).payload
    review = task(id="review", metadata=review_payload["metadata"],
                  result={"verdict": "BLOCK", "findings": ["add regression"]})
    repair_payload = policy.plan(review, [impl, review], LIMITS).payload
    repair = task(id="repair", metadata=repair_payload["metadata"],
                  result={"changed_files": ["test.py"], "head_commit": "c2"})
    rereview = policy.plan(repair, [impl, review, repair], LIMITS)
    stamped = rereview.payload["metadata"]["supervisor"]
    assert rereview.payload["idempotency_key"] == "supervisor:review:impl:c2:2"
    assert {key: stamped[key] for key in ("root_task_id", "project", "branch", "pull_request")} == {
        "root_task_id": "impl", "project": "atlas/iac", "branch": "feature/fix", "pull_request": "pr/9"
    }
    assert "do not create a branch or PR" in repair_payload["body"]


def test_stale_ship_clears_readiness_when_repair_has_a_newer_head():
    chain = {"root_task_id": "impl", "branch": "feature/fix", "pull_request": "pr/9", "base_branch": "main"}
    old_review = task(id="old", metadata={"supervisor": {"kind": "review", "root": "impl",
                      "parent": "impl", "head_commit": "c1", "cycle": 1, **chain}},
                      result={"verdict": "SHIP"})
    repair = task(id="repair", metadata={"supervisor": {"kind": "repair", "root": "impl",
                  "parent": "old", "head_commit": "c1", "cycle": 1, **chain}},
                  result={"head_commit": "c2"})
    decision = policy.plan(old_review, [trusted_root(branch="feature/fix", pull_request="pr/9"), old_review, repair], LIMITS)
    assert decision.action == "clear_ready"
    assert decision.payload == {"commit": "c2", "stale_commit": "c1"}


def test_repair_incomplete_stamp_fails_closed():
    repair = task(id="rep", metadata={"supervisor": {"kind": "repair"}}, result={"head_commit": "c2"})
    assert policy.plan(repair, [repair], LIMITS).action == "escalate"


def test_repair_unparseable_result_fails_closed():
    repair = task(id="rep", metadata=repair_stamp(), result="broken")
    assert policy.plan(repair, [repair], LIMITS).action == "escalate"


def test_repair_without_new_commit_fails_closed():
    repair = task(id="rep", metadata=repair_stamp("impl", "c1", 1), result={"status": "completed"})
    assert policy.plan(repair, [repair], LIMITS).action == "escalate"


def test_repair_producing_same_commit_fails_closed():
    repair = task(
        id="rep",
        metadata=repair_stamp("impl", "c1", 1),
        result={"head_commit": "c1"},
    )
    assert policy.plan(repair, [repair], LIMITS).action == "escalate"


def test_repair_at_cycle_limit_escalates_the_source_repair_not_impl():
    repair = task(
        id="rep",
        metadata=repair_stamp("impl", "c1", LIMITS.max_cycles),
        result={"head_commit": "c2"},
    )
    parent = task(id="rev", metadata=review_stamp("impl", "c1", LIMITS.max_cycles), result={"verdict": "BLOCK"})
    decision = policy.plan(repair, [trusted_root(), parent, repair], LIMITS)
    assert decision.action == "escalate" and decision.target_id == "rep"
    assert "impl" in decision.reason


def test_repair_skips_when_re_review_already_exists():
    repair = task(
        id="rep",
        metadata=repair_stamp("impl", "c1", 1),
        result={"head_commit": "c2"},
    )
    existing = task(id="rev2", status="ready", metadata=review_stamp("impl", "c2", 2))
    parent = task(id="rev", metadata=review_stamp("impl", "c1", 1), result={"verdict": "BLOCK"})
    assert policy.plan(repair, [trusted_root(), parent, repair, existing], LIMITS).action == "none"


# --- dedup / classification helpers --------------------------------------


def test_existing_followup_matches_external_repair_by_title_and_commit():
    repair = task(
        id="ext-rep",
        title="Repair the regression",
        body="fixes commit c1",
        parents=["impl"],
    )
    assert policy.existing_followup([repair], "repair", "impl", "c1") is True
    assert policy.existing_followup([repair], "repair", "impl", "other") is False


def test_active_chain_count_only_counts_in_flight_supervised_cards():
    tasks = [
        task(id="a", status="ready", metadata=review_stamp("r1", "c")),
        task(id="b", status="done", metadata=review_stamp("r2", "c")),
        task(id="c", status="blocked", metadata=repair_stamp("r3", "c")),
        task(id="d", status="ready", metadata={}),
    ]
    assert policy.active_chain_count(tasks) == 1


def test_looks_like_helpers_cover_role_metadata_and_titles():
    assert policy._looks_like_review(task(metadata={"task_role": "review"}))
    assert policy._looks_like_review(task(title="Review pass"))
    assert policy._looks_like_repair(task(metadata={"task_role": "implementation"}))
    assert policy._looks_like_repair(task(title="fix the bug"))
    assert not policy._looks_like_repair(task(title="ship it"))


def test_helper_edge_branches_are_defensive():
    assert policy.field({"id": "d"}, "id") == "d"
    assert policy.parents(task(parents=["ok", "", {"id": None}])) == ["ok"]
    assert policy._first_key(None, policy.COMMIT_KEYS) is None
    assert policy._result_sources(task(), None) == [{}]
    assert policy._strings("scalar") == []
    assert policy._references(task(), "") is False
    assert policy.existing_followup([], "review", "impl", "c1") is False


def test_active_chain_count_ignores_supervised_card_without_root():
    tasks = [task(id="a", status="ready", metadata={"supervisor": {"kind": "review"}})]
    assert policy.active_chain_count(tasks) == 0


def test_review_block_without_findings_still_spawns_repair():
    review = task(
        id="rev",
        metadata=review_stamp("impl", "c1", 1),
        result={"verdict": "BLOCK", "summary": "must not ship BLOCK"},
    )
    decision = policy.plan(review, [trusted_root(), review], LIMITS)
    assert decision.action == "spawn"
    assert "findings to address" not in decision.payload["body"]


def test_no_decision_ever_yields_a_metered_or_merge_action():
    corpus = [
        task(id="impl", result={"changed_files": ["a"], "head_commit": "c1", "pull_request": "p"}),
        task(id="rev", metadata=review_stamp(), result={"verdict": "SHIP"}),
        task(id="rev2", metadata=review_stamp(), result={"verdict": "BLOCK", "findings": ["x"]}),
        task(id="rep", metadata=repair_stamp(), result={"head_commit": "c9"}),
        task(id="bad", metadata=review_stamp(), result="garbage"),
    ]
    for candidate in corpus:
        decision = policy.plan(candidate, corpus, LIMITS)
        assert decision.action in policy.SAFE_ACTIONS
        payload = json.dumps(decision.payload or {})
        for forbidden in ("merge", "approve", "provider", "api_key", "deploy"):
            assert forbidden not in payload
        policy.assert_safe(decision)