"""v2.4 tests for design review V1-V8 (beyond the original repro tests in test_review_r1.py)."""
import json, os, shutil, sys
import pytest
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import aicep_sub
from aicep_common import Ledger, Reject, canonical, b64
from aicep_pub import publish
from aicep_sub import Held, ingest
from test_delivery import w, send, OBJ  # noqa: F401
from test_tamper import env, pub, sub, resign, NOW  # noqa: F401


def skid_seq(d):
    m = json.load(open(os.path.join(d, "manifest.json"))); return m["sender_key_id"], m["seq"]


# V1: a forged manifest sitting in holding/ is dead-lettered, never alerted on
def test_forged_batch_in_holding_goes_to_dlq(w):
    res = send(w); d = res["batch_dir"]
    files = {n: open(os.path.join(d, n), "rb").read() for n in os.listdir(d)}
    m = json.loads(files["manifest.json"]); m["seq"] = 50; files["manifest.json"] = canonical(m)
    hp = os.path.join(w["bus"].root, "c1", "holding", "forged-h"); os.makedirs(hp)
    for n, b in files.items():
        open(os.path.join(hp, n), "wb").write(b)
    r = w["recv"](); w["clk"].t += 4000
    out = r.run_once()
    assert out["forged-h"] == "dlq"
    assert not any(n["event"] == "gap_timeout" for n in r.notices)


# V2b: commit merges, never rolls back last_seq written by another receiver
def test_commit_state_never_rolls_back_last_seq(w):
    r = w["recv"]()
    aicep_sub.atomic_write(r.sub_state, json.dumps({"last_seq": {"k": 9, "o": 1}}).encode())
    tmp = os.path.join(r.dir, "sub_state.x.tmp")
    aicep_sub.atomic_write(tmp, json.dumps({"last_seq": {"k": 3, "o": 2}}).encode())
    r._commit_state(tmp)
    assert json.load(open(r.sub_state))["last_seq"] == {"k": 9, "o": 2}


# V2c: losing the pub->processing rename race skips the batch instead of aborting fetch
def test_fetch_rename_race_skips(w, monkeypatch):
    send(w); send(w)
    real = os.rename; calls = []
    def flaky(a, b):
        calls.append(a)
        if len(calls) == 1:
            raise FileNotFoundError(a)
        return real(a, b)
    monkeypatch.setattr(os, "rename", flaky)
    got = w["bus"].fetch("c1", 10, 300, "r1", folders={"agent-a"})
    assert len(got) == 1


# V3: ingest-level downgrade guard (CLI path too)
def test_ingest_rejects_v1_after_chain(env):
    d1 = pub(env, prev_batch_id="genesis"); sub(env, d1)
    d2 = pub(env)  # no prev_batch_id
    with pytest.raises(Reject, match="chain downgrade"):
        sub(env, d2)


def test_ingest_rejects_wrong_prev(env):
    d1 = pub(env, prev_batch_id="genesis"); sub(env, d1)
    d2 = pub(env, prev_batch_id="not-the-last")
    with pytest.raises(Reject, match="chain break"):
        sub(env, d2)


# V4: missing size_padded is a schema reject
def test_missing_size_padded_rejected(env):
    d = pub(env); resign(env, d, lambda m: [f.pop("size_padded") for f in m["files"]])
    with pytest.raises(Reject, match="schema"):
        sub(env, d)


# V5: receiver routes a signed-but-malformed manifest to the DLQ (no exception, no poison loop)
def test_receiver_non_int_size_padded_to_dlq(w):
    res = send(w); d = os.path.join(w["bus"].root, "c1", "pub", "agent-a", res["batch_id"])
    mp = os.path.join(d, "manifest.json"); m = json.load(open(mp))
    for f in m["files"]:
        f["size_padded"] = "big"
    mb = canonical(m); open(mp, "wb").write(mb); open(mp + ".sig", "w").write(b64(w["sk"].sign(mb)))
    assert w["recv"]().run_once()[res["batch_id"]] == "dlq"


def test_non_int_size_padded_reject(env):
    d = pub(env); resign(env, d, lambda m: [f.__setitem__("size_padded", "4096") for f in m["files"]])
    with pytest.raises(Reject):
        sub(env, d)


# V6: expected_head mismatch and malformed lines fail verify (no exception)
def test_ledger_expected_head_and_malformed(tmp_path):
    L = Ledger(str(tmp_path / "l.jsonl")); h1 = L.append("a", t=1); L.append("b", t=2)
    assert L.verify() and L.verify(expected_head=h1)
    assert not L.verify(expected_head="f" * 64)
    open(L.path, "a").write("{not json\n")
    assert L.verify() is False


# V7: created in the future (beyond skew) is rejected
def test_future_created_rejected(env):
    d = pub(env); resign(env, d, lambda m: m.__setitem__("created", NOW + 10_000))
    with pytest.raises(Reject, match="future"):
        sub(env, d)


# V8: exact-match human gap ack, logged to the ledger
def test_ack_gap_to_must_match_and_is_logged(env):
    pub(env); d2 = pub(env); skid, seq = skid_seq(d2)
    kw = dict(now=NOW + 5)
    with pytest.raises(Held):
        ingest(d2, env["ident"], env["registry"], env["sstate"], "c1", ack_gap_to=f"{skid}:{seq + 1}", operator="h", **kw)
    with pytest.raises(Held):
        ingest(d2, env["ident"], env["registry"], env["sstate"], "c1", ack_gap_to=f"ed25519:other:{seq}", operator="h", **kw)
    with pytest.raises(ValueError):
        ingest(d2, env["ident"], env["registry"], env["sstate"], "c1", ack_gap_to=f"{skid}:{seq}", **kw)
    out = ingest(d2, env["ident"], env["registry"], env["sstate"], "c1", ack_gap_to=f"{skid}:{seq}", operator="h", **kw)
    assert out["gap_alarm"] is True
    L = Ledger(os.path.join(os.path.dirname(env["sstate"]), "ledger.jsonl"))
    e = [x for x in L.entries() if x["event"] == "gap_acked"]
    assert e and e[0]["operator"] == "h" and e[0]["skid"] == skid and (e[0]["from"], e[0]["to"]) == (0, seq)
    assert L.verify()


def test_cli_has_no_blanket_ack_gap():
    src = open(aicep_sub.__file__).read()
    assert '"--ack-gap"' not in src and '"--ack-gap-to"' in src


# ---- v2.4 fixes (design review re-review W1-W4)
import subprocess
from aicep_common import LedgerError, MAX_TTL


def test_w1_expires_not_after_created_rejected(env):
    d = pub(env); resign(env, d, lambda m: m.__setitem__("expires", m["created"]))
    with pytest.raises(Reject, match="expires <= created"):
        sub(env, d)


def test_w1_ttl_at_cap_ok_over_cap_rejected(env):
    sub(env, pub(env, ttl=MAX_TTL))
    with pytest.raises(Reject, match="MAX_TTL"):
        sub(env, pub(env, ttl=MAX_TTL + 1))


def test_w3_reanchor_is_human_only_and_logged(tmp_path):
    p = str(tmp_path / "l.jsonl"); L = Ledger(p); L.append("e", t=1); L.append("e", t=2)
    os.remove(L.anchor_path)
    with pytest.raises(LedgerError):
        L.append("e", t=3)
    with pytest.raises(PermissionError):
        L.reanchor("")
    L.reanchor("human-x", t=4)
    assert L.verify() and L.entries()[-1]["event"] == "reanchored" and L.entries()[-1]["operator"] == "human-x"
    L.append("e", t=5); assert L.verify()


def test_w3_reanchor_refuses_broken_links(tmp_path):
    p = str(tmp_path / "l.jsonl"); L = Ledger(p); L.append("e", t=1); L.append("e", t=2); L.append("e", t=3)
    lines = open(p).read().splitlines(True); open(p, "w").writelines([lines[0], lines[2]])
    with pytest.raises(LedgerError):
        L.reanchor("human-x")


def test_w3_cli_reanchor(tmp_path):
    p = str(tmp_path / "l.jsonl"); L = Ledger(p); L.append("e", t=1); os.remove(L.anchor_path)
    sub_py = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "aicep_sub.py")
    assert subprocess.run([sys.executable, sub_py, "--reanchor-ledger", p], capture_output=True).returncode != 0
    r = subprocess.run([sys.executable, sub_py, "--reanchor-ledger", p, "--operator", "human-x"], capture_output=True)
    assert r.returncode == 0 and Ledger(p).verify()


def test_w3_receiver_start_fails_closed_on_truncated_ledger(w):
    send(w); send(w); w["recv"]().run_once()
    p = w["t"] / "r" / "ledger.jsonl"
    lines = open(p).read().splitlines(True); open(p, "w").writelines(lines[:-3])
    with pytest.raises(LedgerError):
        w["recv"]()
    Ledger(str(p)).reanchor("human-x")      # human re-anchor still can't hide receipt heads that are gone
    with pytest.raises(LedgerError):
        w["recv"]()


def test_w4_existing_v1_sender_then_genesis_chain(w):
    def put(seq, prev):
        b = publish(OBJ, [w["recip"]], w["sk"], "A", "c1", str(w["t"] / "f"), str(w["t"] / "fps.json"),
                    now=w["clk"].t, seq=seq, ttl=3600, prev_batch_id=prev)
        w["bus"].publish(b["batch_dir"], {"compartment": "c1", "batch_id": b["batch_id"], "folder": "agent-a"}); return b["batch_id"]
    r = w["recv"]()
    b1 = put(1, None); assert r.run_once()[b1][0] == "committed"          # legacy v1 history
    b2 = put(2, "genesis"); assert r.run_once()[b2][0] == "committed"     # outbox with no last_published
    b3 = put(3, b2); assert r.run_once()[b3][0] == "committed"
    b4 = put(4, "wrong"); assert r.run_once()[b4] == "dlq"


def test_w3_reanchor_blank_operator_refused(tmp_path):
    L = Ledger(str(tmp_path / "l.jsonl")); L.append("e", t=1); os.remove(L.anchor_path)
    for op in ("", "   ", "\t\n", None):
        with pytest.raises(PermissionError):
            L.reanchor(op)
    L.reanchor("  human-x  ", t=2)
    assert L.entries()[-1]["operator"] == "human-x" and L.verify()