"""v2.5 per-agent folders: pub/<folder>/<batch_id>, registry folder slugs, folder_mismatch."""
import json, os, shutil, sys
import pytest
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
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__)))
from aicep_common import b64, key_id, valid_slug, check_slug, validate_registry_folders, active_folders, Ledger
from aicep_pub import Outbox, publish
from aicep_transport import LocalFolderAdapter
from test_delivery import w, send, OBJ, raw  # noqa: F401


def add_sender(w, folder, sender_id="B"):
    sk = Ed25519PrivateKey.generate()
    w["reg"][key_id(raw(sk))] = {"sender_id": sender_id, "ed25519_pub": b64(raw(sk)), "state": "verified",
                                 "verified_by": "human", "verified_at": w["clk"].t - 10,
                                 "expires": w["clk"].t + 86400, "revoked_at": None, "folder": folder}
    ob = Outbox(str(w["t"] / f"ob-{folder}.json"), sk, sender_id, str(w["t"] / f"l-{folder}.jsonl"),
                now=w["clk"], folder=folder)
    return sk, ob


def send_from(w, ob):
    return ob.send(OBJ, [w["recip"]], "c1", w["bus"], str(w["t"] / f"work-{ob.folder}"), ttl=72000)


def test_happy_path_two_senders(w):
    _, ob_b = add_sender(w, "agent-beta")
    a = send(w); b = send_from(w, ob_b)
    assert os.path.isdir(os.path.join(w["bus"].root, "c1", "pub", "agent-a", a["batch_id"]))
    assert os.path.isdir(os.path.join(w["bus"].root, "c1", "pub", "agent-beta", b["batch_id"]))
    out = w["recv"]().run_once()
    assert out[a["batch_id"]][0] == "committed" and out[b["batch_id"]][0] == "committed"
    a2 = send(w); b2 = send_from(w, ob_b)
    out = w["recv"]().run_once()
    assert out[a2["batch_id"]][0] == "committed" and out[b2["batch_id"]][0] == "committed"


def test_batch_in_other_senders_folder_rejected(w):
    add_sender(w, "agent-beta")
    a = send(w)
    src = os.path.join(w["bus"].root, "c1", "pub", "agent-a", a["batch_id"])
    dst = os.path.join(w["bus"].root, "c1", "pub", "agent-beta", a["batch_id"])
    os.makedirs(os.path.dirname(dst), exist_ok=True); shutil.move(src, dst)
    r = w["recv"]()
    assert r.run_once()[a["batch_id"]] == "dlq"
    reason = json.load(open(os.path.join(w["bus"].root, "c1", "dlq", a["batch_id"], "reason.json")))
    assert reason["record"]["reason"] == "folder_mismatch"
    assert any(n["event"] == "folder_mismatch" and n["email"] for n in r.notices)


def test_unregistered_folder_ignored_and_logged_once(w):
    a = send(w)
    src = os.path.join(w["bus"].root, "c1", "pub", "agent-a", a["batch_id"])
    dst = os.path.join(w["bus"].root, "c1", "pub", "stranger-bot", a["batch_id"])
    os.makedirs(os.path.dirname(dst)); shutil.move(src, dst)
    r = w["recv"]()
    assert r.run_once() == {} and r.run_once() == {}
    assert os.path.isdir(dst)                                        # untouched
    L = Ledger(str(w["t"] / "r" / "ledger.jsonl"))
    assert [e["folder"] for e in L.entries() if e["event"] == "unknown_folder_ignored"] == ["stranger-bot"]


def test_revoked_senders_folder_ignored(w):
    _, ob_b = add_sender(w, "agent-beta")
    b = send_from(w, ob_b)
    for ent in w["reg"].values():
        if ent["folder"] == "agent-beta":
            ent["revoked_at"] = w["clk"].t - 1
    assert "agent-beta" not in active_folders(w["reg"], w["clk"].t)
    assert b["batch_id"] not in w["recv"]().run_once()
    assert os.path.isdir(os.path.join(w["bus"].root, "c1", "pub", "agent-beta", b["batch_id"]))


def test_revoked_sender_batch_already_in_processing_not_redelivered(w):
    _, ob_b = add_sender(w, "agent-beta")
    b = send_from(w, ob_b)
    assert w["bus"].fetch("c1", 10, 1, "r9", folders={"agent-beta"})
    w["clk"].t += 5
    assert w["bus"].fetch("c1", 10, 300, "r1", folders={"agent-a"}) == []


@pytest.mark.parametrize("slug", ["ab", "a" * 65, "Agent-alpha", "agent.alpha", "grok/bot", "..", "grok_bot",
                                  "", None, "agent alpha"])
def test_bad_slug_refused(w, slug):
    assert not valid_slug(slug)
    with pytest.raises(ValueError):
        check_slug(slug)
    with pytest.raises(ValueError):
        w["bus"].publish(str(w["t"]), {"compartment": "c1", "batch_id": "x", "folder": slug})
    with pytest.raises(ValueError):
        Outbox(str(w["t"] / "o.json"), Ed25519PrivateKey.generate(), "Z", str(w["t"] / "l.jsonl"), folder=slug)


def test_good_slugs():
    for s in ("agent-alpha", "agent-beta", "abc", "a" * 64, "agent-007"):
        assert valid_slug(s)


def test_registry_folder_must_be_unique_and_present(w):
    add_sender(w, "agent-a")                                         # duplicate of the fixture sender
    with pytest.raises(ValueError, match="two keys"):
        validate_registry_folders(w["reg"])
    with pytest.raises(ValueError):
        w["recv"]()


def test_registry_entry_without_folder_refused(w):
    for ent in w["reg"].values():
        ent.pop("folder")
    with pytest.raises(ValueError):
        w["recv"]()


def test_redrive_and_hold_return_to_same_folder(w):
    res = send(w); d = os.path.join(w["bus"].root, "c1", "pub", "agent-a", res["batch_id"])
    open(os.path.join(d, "meta.age"), "ab").write(b"x")             # poison -> dlq
    assert w["recv"]().run_once()[res["batch_id"]] == "dlq"
    w["bus"].redrive("c1", res["batch_id"], operator="human-operator")
    assert os.path.isdir(d)


from test_tamper import env, pub, NOW  # noqa: E402,F401
from aicep_sub import ingest  # noqa: E402
from aicep_common import Reject  # noqa: E402


def test_ingest_folder_label_checked_against_verified_key(env):
    d = pub(env)
    with pytest.raises(Reject, match="folder mismatch"):
        ingest(d, env["ident"], env["registry"], env["sstate"], "c1", now=NOW + 5, folder="other-agent")
    assert ingest(d, env["ident"], env["registry"], env["sstate"], "c1", now=NOW + 5, folder="swarm-a")["status"] == "accepted"


# ---- v2.5.1 (design review v2.5 review X1-X3 + minor)
import subprocess  # noqa: E402
from aicep_common import Ledger as _L  # noqa: E402


def _P(w, *a):
    return os.path.join(w["bus"].root, "c1", *a)


def test_x1_overflow_summary_row_written_once(w):
    for i in range(100):
        os.makedirs(_P(w, "pub", f"junk-{i}"))
    os.makedirs(_P(w, "pub", "Bad.Name"))
    r = w["recv"](); r.run_once(); r.run_once()
    es = _L(str(w["t"] / "r" / "ledger.jsonl")).entries()
    assert sum(e["event"] == "unknown_folder_ignored" for e in es) == 64
    assert sum(e["event"] == "unknown_folder_overflow" for e in es) == 1
    assert all(len(h) == 16 for h in r._r()["unknown_folders_logged"])


def test_x2_coalesced_alerts_one_per_folder_with_summary(w):
    add_sender(w, "agent-beta"); add_sender(w, "other-bot", sender_id="C")
    a = send(w)
    for i in range(5):
        shutil.copytree(_P(w, "pub", "agent-a", a["batch_id"]), _P(w, "pub", "agent-beta", f"x{i}"))
        shutil.copytree(_P(w, "pub", "agent-a", a["batch_id"]), _P(w, "pub", "other-bot", f"y{i}"))
    r = w["recv"](); r.run_once()
    assert sum(1 for n in r.notices if n.get("email") and n["event"] == "folder_mismatch") == 2
    s = [n for n in r.notices if n["event"] == "alerts_coalesced"]
    assert s and not s[0]["email"] and sum(s[0]["counts"].values()) == 8
    r2 = w["recv"](); send(w)
    shutil.copytree(_P(w, "pub", "agent-a", sorted(os.listdir(_P(w, "pub", "agent-a")))[0]), _P(w, "pub", "agent-beta", "z"))
    r2.run_once()                                                      # new run -> alert again
    assert any(n.get("email") and n["event"] == "folder_mismatch" for n in r2.notices)


def test_x3_processing_resolves_folder_from_verified_skid(w):
    a = send(w); r = w["recv"]()
    (d,) = w["bus"].fetch("c1", 10, 1, "r1", folders={"agent-a"})
    w["bus"].set_skid(d, next(iter(w["reg"])))                         # verified once, then crash
    for e in w["reg"].values():
        e["folder"] = "agent-a2"
    w["clk"].t += 5
    assert r.run_once()[a["batch_id"]][0] == "committed"


def test_x3_receiver_redrive_uses_registry_folder(w):
    res = send(w); d = _P(w, "pub", "agent-a", res["batch_id"])
    open(os.path.join(d, "meta.age"), "ab").write(b"x")
    r = w["recv"](); assert r.run_once()[res["batch_id"]] == "dlq"
    for e in w["reg"].values():
        e["folder"] = "agent-a2"
    with pytest.raises(PermissionError):
        r.redrive(res["batch_id"], "  ")
    r.redrive(res["batch_id"], "human-operator")
    assert os.path.isdir(_P(w, "pub", "agent-a2", res["batch_id"]))


def test_symlinked_batch_file_rejected_by_receiver(w, tmp_path):
    res = send(w); d = _P(w, "pub", "agent-a", res["batch_id"])
    real = tmp_path / "outside.age"; shutil.move(os.path.join(d, "meta.age"), real)
    os.symlink(real, os.path.join(d, "meta.age"))
    assert w["recv"]().run_once()[res["batch_id"]] == "dlq"


def test_symlinked_batch_dir_skipped_by_fetch(w, tmp_path):
    res = send(w); d = _P(w, "pub", "agent-a", res["batch_id"])
    shutil.move(d, tmp_path / "real"); os.symlink(tmp_path / "real", d)
    assert w["bus"].fetch("c1", 10, 300, "r1", folders={"agent-a"}) == []


def test_ingest_rejects_symlinked_dir_and_file(env, tmp_path):
    d = pub(env)
    link = tmp_path / "lnk"; os.symlink(d, link)
    with pytest.raises(Reject, match="symlink"):
        ingest(str(link), env["ident"], env["registry"], env["sstate"], "c1", now=NOW + 5, folder="swarm-a")
    real = tmp_path / "m.age"; shutil.move(os.path.join(d, "meta.age"), real); os.symlink(real, os.path.join(d, "meta.age"))
    with pytest.raises(Reject, match="symlink"):
        ingest(d, env["ident"], env["registry"], env["sstate"], "c1", now=NOW + 5, folder="swarm-a")


def test_ingest_cli_requires_folder(tmp_path):
    sub_py = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "aicep_sub.py")
    r = subprocess.run([sys.executable, sub_py, str(tmp_path), "--compartment", "c1"], capture_output=True, text=True)
    assert r.returncode != 0 and "--folder" in r.stderr