"""v2.3 delivery-guarantee tests over the LocalFolderAdapter. Throwaway keys; nothing persists."""
import json, os, shutil, sys
import pytest, pyrage
from cryptography.hazmat.primitives import serialization as S
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey

HERE = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, HERE)
from aicep_common import b64, canonical, key_id, Ledger, HOLD_WINDOW, unb64
from aicep_pub import Outbox, publish
from aicep_sub import Receiver, Crash
from aicep_transport import LocalFolderAdapter, TransportAdapter, READY

OBJ = [{"name": "n", "type": "note", "classification": "internal", "body": "hi"}]


class Clock:
    def __init__(self, t=1_800_000_000): self.t = t
    def __call__(self): return self.t


def raw(k): return k.public_key().public_bytes(S.Encoding.Raw, S.PublicFormat.Raw)


@pytest.fixture
def w(tmp_path):
    clk = Clock()
    sk, rk, ident = Ed25519PrivateKey.generate(), Ed25519PrivateKey.generate(), pyrage.x25519.Identity.generate()
    reg = {key_id(raw(sk)): {"sender_id": "A", "ed25519_pub": b64(raw(sk)), "state": "verified",
           "verified_by": "human", "verified_at": clk.t - 10, "expires": clk.t + 86400 * 90, "revoked_at": None,
           "folder": "agent-a"}}
    bus = LocalFolderAdapter(str(tmp_path / "bus"), now=clk)
    ob = Outbox(str(tmp_path / "outbox.json"), sk, "A", str(tmp_path / "s_ledger.jsonl"), now=clk, folder="agent-a")
    def recv(**kw):
        return Receiver(ident, reg, str(tmp_path / "r"), "c1", rk, kw.pop("bus", bus), now=clk, **kw)
    return dict(t=tmp_path, clk=clk, sk=sk, rk=rk, ident=ident, reg=reg, bus=bus, ob=ob, recv=recv,
                recip=str(ident.to_public()), work=str(tmp_path / "work"))


def send(w, ttl=3600 * 20, **kw):
    return w["ob"].send(OBJ, [w["recip"]], "c1", w["bus"], w["work"], ttl=ttl, **kw)


def test_adapter_is_abstract():
    with pytest.raises(TypeError):
        TransportAdapter()
    assert isinstance(LocalFolderAdapter("/tmp/x"), TransportAdapter)


# 1. commit marker / staging
def test_half_uploaded_batch_is_invisible_then_completes(w):
    bad = LocalFolderAdapter(w["bus"].root, now=w["clk"], crash_after_files=2)
    with pytest.raises(IOError):
        w["ob"].send(OBJ, [w["recip"]], "c1", bad, w["work"], ttl=72000)
    assert w["recv"]().run_once() == {}                       # nothing visible
    rows = w["bus"].list_pending("c1")
    assert rows and rows[0]["state"] == "_staging" and not rows[0]["ready"]
    bid = w["ob"].staged()[0]                                  # still staged in outbox
    w["ob"].send(OBJ, [w["recip"]], "c1", w["bus"], w["work"], ttl=72000, batch_id=bid)
    assert w["recv"]().run_once()[bid][0] == "committed"


def test_dir_without_ready_marker_skipped(w):
    res = send(w); d = res["batch_dir"]
    files = {n: open(os.path.join(d, n), "rb").read() for n in os.listdir(d)}
    w["bus"].place_raw("c1", "partial-x", files, ready=False, folder="agent-a")
    out = w["recv"]().run_once()
    assert "partial-x" not in out and out[res["batch_id"]][0] == "committed"


# 2. idempotent redelivery / seq reuse
def test_duplicate_resend_gets_original_receipt(w):
    res = send(w); r = w["recv"]()
    first = r.run_once()[res["batch_id"]][1]
    w["bus"].publish(res["batch_dir"], {"compartment": "c1", "batch_id": res["batch_id"], "folder": "agent-a"})  # re-upload same bytes
    shutil.rmtree(os.path.join(w["bus"].root, "c1", "pub", "agent-a", res["batch_id"]), ignore_errors=True)
    w["bus"].place_raw("c1", res["batch_id"], {n: open(os.path.join(res["batch_dir"], n), "rb").read()
                                               for n in os.listdir(res["batch_dir"])}, folder="agent-a")
    kind, again = r.run_once()[res["batch_id"]]
    assert kind == "duplicate" and again == first
    assert any(n["event"] == "receipt_resent" for n in r.notices)


def test_seq_reuse_with_different_manifest_rejected_with_alarm(w):
    res = send(w); r = w["recv"](); r.run_once()
    st = json.load(open(w["t"] / "outbox.json")); prev = st["last_published"]
    evil = publish([{"name": "x", "type": "note", "classification": "internal", "body": "other"}],
                   [w["recip"]], w["sk"], "A", "c1", str(w["t"] / "evil"), str(w["t"] / "eps.json"),
                   now=w["clk"].t, seq=1, ttl=3600)
    w["bus"].publish(evil["batch_dir"], {"compartment": "c1", "batch_id": evil["batch_id"], "folder": "agent-a"})
    out = r.run_once()
    assert out[evil["batch_id"]] == "dlq"
    assert any(n["event"] == "equivocation" and n["email"] for n in r.notices)


# 3. outbox
def test_outbox_reserves_and_logs_staged_before_publish(w):
    bad = LocalFolderAdapter(w["bus"].root, now=w["clk"], crash_after_files=0)
    with pytest.raises(IOError):
        w["ob"].send(OBJ, [w["recip"]], "c1", bad, w["work"])
    st = json.load(open(w["t"] / "outbox.json"))
    (bid, e), = st["batches"].items()
    assert e["status"] == "staged" and e["seq"] == 1
    assert [x["event"] for x in Ledger(str(w["t"] / "s_ledger.jsonl")).entries()] == ["staged"]
    res = send(w)                                               # a new batch takes seq 2, not 1
    assert res["seq"] == 2
    with pytest.raises(ValueError):
        send(w, batch_id=res["batch_id"])                       # published batch can't be resumed


# 4. dlq
def test_poison_batch_goes_to_dlq_with_signed_reason_and_human_redrive(w):
    res = send(w); d = os.path.join(w["bus"].root, "c1", "pub", "agent-a", res["batch_id"])
    f = sorted(x for x in os.listdir(d) if x.endswith(".age") and x != "meta.age")[0]
    p = os.path.join(d, f); b = bytearray(open(p, "rb").read()); b[100] ^= 1; open(p, "wb").write(bytes(b))
    r = w["recv"]()
    assert r.run_once()[res["batch_id"]] == "dlq"
    rr = json.load(open(os.path.join(w["bus"].root, "c1", "dlq", res["batch_id"], "reason.json")))
    assert rr["record"]["reason"] == "hash_mismatch" and rr["record"]["attempts"] == 1
    from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
    Ed25519PublicKey.from_public_bytes(raw(w["rk"])).verify(unb64(rr["sig"]), canonical(rr["record"]))
    assert any(n["event"] == "dlq" and n["email"] for n in r.notices)
    with pytest.raises(PermissionError):
        w["bus"].redrive("c1", res["batch_id"], operator="")
    w["bus"].redrive("c1", res["batch_id"], operator="human-operator")
    assert os.path.isdir(os.path.join(w["bus"].root, "c1", "pub", "agent-a", res["batch_id"]))


def test_transient_errors_retry_three_times_then_dlq(w):
    res = send(w); bid = res["batch_id"]
    for i in (1, 2):
        assert w["recv"](crash_at="transient").run_once()[bid] == f"retry({i})"
        w["clk"].t += 60 * i + 1
    r = w["recv"](crash_at="transient")
    assert r.run_once()[bid] == "dlq"
    rr = json.load(open(os.path.join(w["bus"].root, "c1", "dlq", bid, "reason.json")))
    assert rr["record"]["reason"] == "ingest_error" and rr["record"]["attempts"] == 3


def test_crypto_failure_dlq_on_first_try(w):
    res = send(w); d = os.path.join(w["bus"].root, "c1", "pub", "agent-a", res["batch_id"])
    open(os.path.join(d, "manifest.json.sig"), "w").write(b64(Ed25519PrivateKey.generate().sign(b"x")) + "\n")
    assert w["recv"]().run_once()[res["batch_id"]] == "dlq"
    rr = json.load(open(os.path.join(w["bus"].root, "c1", "dlq", res["batch_id"], "reason.json")))
    assert rr["record"]["reason"] == "sig_fail" and rr["record"]["attempts"] == 1


def test_qop_floor_unencrypted_payload_to_dlq(w):
    res = send(w); d = os.path.join(w["bus"].root, "c1", "pub", "agent-a", res["batch_id"])
    open(os.path.join(d, "plain.txt"), "w").write("plaintext")
    r = w["recv"](); assert r.run_once()[res["batch_id"]] == "dlq"
    rr = json.load(open(os.path.join(w["bus"].root, "c1", "dlq", res["batch_id"], "reason.json")))
    assert rr["record"]["reason"] == "qop"


# 5. retry schedule + cumulative receipts
def test_retry_schedule_then_unconfirmed(w):
    res = send(w, ttl=3600 * 20); ob, t0 = w["ob"], w["clk"].t
    assert ob.due(w["bus"]) == []
    for off in (3600, 4 * 3600, 12 * 3600):
        w["clk"].t = t0 + off
        assert ob.due(w["bus"]) == [("resend", res["batch_id"])]
    w["clk"].t = t0 + 13 * 3600
    assert ob.due(w["bus"]) == [("unconfirmed", res["batch_id"])]


def test_expiry_marks_unconfirmed(w):
    res = send(w, ttl=1800); w["clk"].t += 1800
    assert w["ob"].due(w["bus"]) == [("unconfirmed", res["batch_id"])]


def test_cumulative_receipt_acks_all_lower_seqs(w):
    a, b = send(w), send(w); r = w["recv"]()
    out = r.run_once()
    rc = out[b["batch_id"]][1]
    assert rc["record"]["acked_through_seq"] == 2 and len(rc["record"]["ledger_head"]) == 64
    assert w["ob"].on_receipt(rc["record"], rc["sig"], raw(w["rk"])) == 2
    st = json.load(open(w["t"] / "outbox.json"))
    assert all(e["status"] == "acked" for e in st["batches"].values())
    assert w["ob"].due(w["bus"]) == []
    bad = dict(rc["record"], acked_through_seq=99)
    with pytest.raises(Exception):
        w["ob"].on_receipt(bad, rc["sig"], raw(w["rk"]))


# 6. state machine, crash recovery, lease
@pytest.mark.parametrize("point", ["committing", "after_state"])
def test_crash_injection_recovers_exactly_once(w, point):
    res = send(w); bid = res["batch_id"]
    with pytest.raises(Crash):
        w["recv"](crash_at=point).run_once()
    r2 = w["recv"]()
    out = r2.run_once()
    assert out == {}                                          # recovered in run_once, nothing redone
    st = json.load(open(w["t"] / "r" / "receiver.json"))
    assert st["phase"][bid]["phase"] == "committed"
    assert json.load(open(w["t"] / "r" / "sub_state.json"))["last_seq"][key_id(raw(w["sk"]))] == 1
    events = [e["event"] for e in Ledger(str(w["t"] / "r" / "ledger.jsonl")).entries()]
    assert events.count("committed") == 1 and ("recovered_commit" in events or "recovered_apply" in events)


def test_crash_before_verify_redelivered_after_lease_expiry(w):
    res = send(w); bid = res["batch_id"]
    w["bus"].fetch("c1", 10, 300, "r1", folders={"agent-a"})                     # r1 leases then dies
    assert w["recv"](consumer="r2").run_once() == {}        # lease still held: r2 can't take it
    w["clk"].t += 301                                        # lease expires
    assert w["recv"](consumer="r2").run_once()[bid][0] == "committed"


def test_lease_blocks_second_consumer(w):
    send(w)
    a = w["bus"].fetch("c1", 10, 300, "r1", folders={"agent-a"}); b = w["bus"].fetch("c1", 10, 300, "r2", folders={"agent-a"})
    assert len(a) == 1 and b == []


# 7. reorder holding, chain, ledgers
def test_out_of_order_held_then_released_in_order(w):
    a = send(w); b = send(w)
    shutil.move(os.path.join(w["bus"].root, "c1", "pub", "agent-a", a["batch_id"]), str(w["t"] / "delayed"))
    r = w["recv"]()
    assert r.run_once()[b["batch_id"]] == "held"
    shutil.move(str(w["t"] / "delayed"), os.path.join(w["bus"].root, "c1", "pub", "agent-a", a["batch_id"]))
    out = r.run_once()
    assert out[a["batch_id"]][0] == "committed" and out[b["batch_id"]][0] == "committed"


def test_gap_persisting_past_window_alerts_human(w):
    a = send(w); b = send(w)
    shutil.rmtree(os.path.join(w["bus"].root, "c1", "pub", "agent-a", a["batch_id"]))
    r = w["recv"](); r.run_once()
    w["clk"].t += HOLD_WINDOW + 1
    assert r.run_once()[b["batch_id"]] == "held_human"
    assert any(n["event"] == "gap_timeout" and n["email"] for n in r.notices)


def test_prev_batch_chain_break_rejected(w):
    a = send(w); r = w["recv"](); r.run_once()
    forged = publish(OBJ, [w["recip"]], w["sk"], "A", "c1", str(w["t"] / "f"), str(w["t"] / "fps.json"),
                     now=w["clk"].t, seq=2, prev_batch_id="00000000-0000-4000-8000-000000000000", ttl=3600)
    w["bus"].publish(forged["batch_dir"], {"compartment": "c1", "batch_id": forged["batch_id"], "folder": "agent-a"})
    assert r.run_once()[forged["batch_id"]] == "dlq"


def test_ledgers_hash_chained_and_tamper_evident(w):
    send(w); send(w); w["recv"]().run_once()
    for p in (w["t"] / "s_ledger.jsonl", w["t"] / "r" / "ledger.jsonl"):
        L = Ledger(str(p)); assert L.verify() and len(L.entries()) >= 4
        lines = open(p).read().splitlines(); e = json.loads(lines[1]); e["seq"] = 99
        lines[1] = json.dumps(e); open(p, "w").write("\n".join(lines) + "\n")
        assert not Ledger(str(p)).verify()


def test_no_content_or_keys_in_ledgers_or_notices(w):
    send(w); r = w["recv"](); r.run_once()
    blob = open(w["t"] / "r" / "ledger.jsonl").read() + json.dumps(r.notices) + open(w["t"] / "s_ledger.jsonl").read()
    assert '"hi"' not in blob and "AGE-SECRET-KEY" not in blob and "PRIVATE" not in blob