"""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
