#!/usr/bin/env python3
"""AICEP v2.3 transport-adapter interface (proposed Phase 2 extension).
The AICEP batch is the message. Adapters move opaque bytes and report delivery state; they never see
plaintext or keys. Their guarantees are hints only: the core still enforces seq, dedup, expiry, signatures.
"""
import json, os, shutil, time
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from aicep_common import atomic_write, canonical, check_slug, load_json
READY = "READY"
@dataclass
class PublishReceipt:
transport_ref: str
durable: bool
@dataclass
class Delivery:
batch_id: str
path: str # local path of the fetched batch dir (opaque files)
transport_ref: str
delivery_count: int
lease_until: int
attrs: dict = field(default_factory=dict)
class TransportAdapter(ABC):
@abstractmethod
def publish(self, batch_dir: str, attrs: dict) -> PublishReceipt:
"""Durable, atomic-visibility, idempotent on batch_id."""
@abstractmethod
def fetch(self, compartment: str, max_items: int, lease_secs: int, consumer: str, folders=()) -> list:
"""Lease complete deliveries from the given sender folders only (v2.5); incomplete/staged
batches and unlisted folders are never returned. Delivery.attrs['folder'] = source folder."""
@abstractmethod
def ack(self, delivery: Delivery) -> None:
"""Final transport settle after core commit."""
@abstractmethod
def nack(self, delivery: Delivery, retry_after_secs: int, reason: str) -> None:
"""Transient failure: release lease, re-visible after delay, delivery_count++."""
@abstractmethod
def dead_letter(self, delivery: Delivery, reason_record: dict) -> None:
"""Move to DLQ with signed reason record. Never deletes."""
@abstractmethod
def hold(self, delivery: Delivery) -> None:
"""Move an out-of-order batch to the holding area."""
def lease_valid(self, delivery: Delivery, consumer: str) -> bool:
"""Re-check the lease right before commit (v2.4). Adapters without leases may return True."""
return True
@abstractmethod
def list_pending(self, compartment: str) -> list:
"""Read-only view: [{batch_id, state, delivery_count}]."""
@abstractmethod
def capabilities(self) -> dict:
...
class LocalFolderAdapter(TransportAdapter):
"""Folder bus (v2.5): senders publish to <root>/<compartment>/pub/<folder>/<batch_id>/ where <folder> is the
sender's registered slug; <root>/<compartment>/{_staging/<folder>,processing,holding,done,dlq}/<batch_id>/.
The source folder is kept in _meta so hold/redrive return a batch to the same folder.
Commit marker: READY written LAST (after manifest.json.sig); fetch skips dirs without it."""
def __init__(self, root, now=time.time, crash_after_files=None):
self.root, self.now = root, now
self.crash_after_files = crash_after_files # test hook: simulate half-upload
def _d(self, comp, area, bid=""):
return os.path.join(self.root, comp, area, bid)
def _meta(self, comp, bid):
return os.path.join(self.root, comp, "_meta", bid + ".json")
def capabilities(self):
return {"atomic_publish": "emulated(READY marker)", "native_dedup_window": 0, "native_ordering": False,
"native_dlq": False, "native_lease": "emulated(lease file)", "max_blob_bytes": None}
def publish(self, batch_dir, attrs):
comp, bid = attrs["compartment"], attrs["batch_id"]
folder = check_slug(attrs.get("folder")) # v2.5: per-agent folder required
dest = os.path.join(self._d(comp, "pub"), folder, bid)
if os.path.exists(os.path.join(dest, READY)):
return PublishReceipt(dest, True) # idempotent
stage = os.path.join(self._d(comp, "_staging"), folder, bid)
if os.path.exists(stage):
shutil.rmtree(stage) # discard a half-upload
os.makedirs(stage)
names = sorted(n for n in os.listdir(batch_dir) if n != "manifest.json.sig") + ["manifest.json.sig"]
for i, n in enumerate(names): # signature uploaded last
if self.crash_after_files is not None and i >= self.crash_after_files:
raise IOError("simulated crash mid-upload")
shutil.copyfile(os.path.join(batch_dir, n), os.path.join(stage, n))
open(os.path.join(stage, READY), "w").write("ok\n") # commit marker last
os.makedirs(os.path.dirname(dest), exist_ok=True)
if os.path.exists(dest):
shutil.rmtree(dest)
os.rename(stage, dest)
return PublishReceipt(dest, True)
def place_raw(self, comp, bid, files: dict, ready=True, folder=None):
"""Test/ops helper: drop a (possibly incomplete) batch straight into pub/<folder>/ (no slug check)."""
d = os.path.join(self._d(comp, "pub"), folder, bid) if folder else self._d(comp, "pub", bid)
os.makedirs(d, exist_ok=True)
for n, b in files.items():
open(os.path.join(d, n), "wb").write(b)
if ready:
open(os.path.join(d, READY), "w").write("ok\n")
def unknown_folders(self, comp, folders):
"""pub/ entries that are not a watched sender folder (ignored by fetch)."""
base = self._d(comp, "pub")
return sorted(n for n in os.listdir(base) if n not in set(folders)) if os.path.isdir(base) else []
def _candidates(self, comp, folders):
folders = set(folders or ())
base = self._d(comp, "processing")
if os.path.isdir(base):
for bid in sorted(os.listdir(base)):
yield "processing", None, bid, os.path.join(base, bid)
for f in sorted(folders):
fb = os.path.join(self._d(comp, "pub"), f)
if os.path.isdir(fb) and not os.path.islink(fb):
for bid in sorted(os.listdir(fb)):
yield "pub", f, bid, os.path.join(fb, bid)
def fetch(self, comp, max_items=10, lease_secs=300, consumer="r1", folders=()):
"""folders: iterable of slugs, or {slug: sender_key_id} (v2.5.1: lets processing items that carry
a verified skid in _meta resolve their CURRENT folder from the registry, not the stale one)."""
out, now, taken = [], int(self.now()), set()
by_skid = {v: k for k, v in folders.items()} if isinstance(folders, dict) else {}
folders = set(folders or ())
for area, folder, bid, d in self._candidates(comp, folders):
if True:
if bid in taken:
continue
if not os.path.exists(os.path.join(d, READY)):
continue # incomplete: skip
meta = load_json(self._meta(comp, bid), default={"delivery_count": 0, "retry_after": 0})
if os.path.islink(d):
continue # v2.5.1: never follow a symlinked batch dir
if area == "pub":
meta["folder"] = folder
elif meta.get("skid") in by_skid:
meta["folder"] = by_skid[meta["skid"]] # X3: resolve from registry by verified skid
elif meta.get("folder") not in folders:
continue # sender no longer watched (revoked)
lease = meta.get("lease")
if lease and lease["until"] > now and lease["consumer"] != consumer:
continue # leased by someone else
if meta.get("retry_after", 0) > now:
continue
meta["delivery_count"] += 1
meta["lease"] = {"consumer": consumer, "until": now + lease_secs}
if area == "pub":
dst = self._d(comp, "processing", bid); os.makedirs(os.path.dirname(dst), exist_ok=True)
try:
os.rename(d, dst); d = dst
except OSError:
continue # V2c: lost the race; skip, don't abort
if not os.path.isdir(d):
continue
atomic_write(self._meta(comp, bid), json.dumps(meta).encode())
taken.add(bid)
out.append(Delivery(bid, d, d, meta["delivery_count"], now + lease_secs,
{"folder": meta.get("folder")}))
if len(out) >= max_items:
return out
return out
def lease_valid(self, delivery, consumer):
"""V2a: True only if `consumer` still holds an unexpired lease on this delivery."""
try:
meta = load_json(self._meta(self._comp(delivery), delivery.batch_id), default={})
except (OSError, ValueError):
return False
lease = meta.get("lease") or {}
return lease.get("consumer") == consumer and int(lease.get("until", 0)) > int(self.now()) \
and os.path.isdir(delivery.path)
def batch_files(self, d):
return {n for n in os.listdir(d) if n != READY}
def _move(self, delivery, comp, area):
dst = self._d(comp, area, delivery.batch_id); os.makedirs(os.path.dirname(dst), exist_ok=True)
if os.path.exists(dst):
shutil.rmtree(dst)
os.rename(delivery.path, dst)
return dst
def _comp(self, delivery):
return os.path.basename(os.path.dirname(os.path.dirname(delivery.path)))
def ack(self, delivery):
comp = self._comp(delivery); self._move(delivery, comp, "done")
meta = load_json(self._meta(comp, delivery.batch_id), default={}); meta.pop("lease", None)
atomic_write(self._meta(comp, delivery.batch_id), json.dumps(meta).encode())
def nack(self, delivery, retry_after_secs, reason):
comp = self._comp(delivery)
meta = load_json(self._meta(comp, delivery.batch_id), default={"delivery_count": 0})
meta.pop("lease", None); meta["retry_after"] = int(self.now()) + retry_after_secs; meta["last_reason"] = reason
atomic_write(self._meta(comp, delivery.batch_id), json.dumps(meta).encode())
def dead_letter(self, delivery, reason_record):
comp = self._comp(delivery); dst = self._move(delivery, comp, "dlq")
atomic_write(os.path.join(dst, "reason.json"), canonical(reason_record))
def hold(self, delivery):
comp = self._comp(delivery); self._move(delivery, comp, "holding")
meta = load_json(self._meta(comp, delivery.batch_id), default={}); meta.pop("lease", None)
meta["held_since"] = meta.get("held_since", int(self.now()))
atomic_write(self._meta(comp, delivery.batch_id), json.dumps(meta).encode())
def set_skid(self, delivery, skid):
"""X3: record the VERIFIED sender_key_id so later moves resolve the folder from the registry."""
comp = self._comp(delivery)
meta = load_json(self._meta(comp, delivery.batch_id), default={}); meta["skid"] = skid
atomic_write(self._meta(comp, delivery.batch_id), json.dumps(meta).encode())
def _pub_dst(self, comp, bid, folder=None):
meta = load_json(self._meta(comp, bid), default={})
if folder is not None: # X3: caller resolved it from the registry
meta["folder"] = check_slug(folder); atomic_write(self._meta(comp, bid), json.dumps(meta).encode())
return os.path.join(self._d(comp, "pub"), check_slug(meta.get("folder")), bid)
def release_held(self, comp, bid, folder=None):
src, dst = self._d(comp, "holding", bid), self._pub_dst(comp, bid, folder)
os.makedirs(os.path.dirname(dst), exist_ok=True); os.rename(src, dst)
def redrive(self, comp, bid, operator: str, folder=None):
"""Human-only: move a dlq batch back to pub/<folder>/. Requires a named operator. Prefer
Receiver.redrive, which resolves the folder from the registry by the verified signing key (X3)."""
if not operator or not str(operator).strip():
raise PermissionError("redrive requires a named human operator")
src, dst = self._d(comp, "dlq", bid), self._pub_dst(comp, bid, folder)
os.makedirs(os.path.dirname(dst), exist_ok=True)
os.remove(os.path.join(src, "reason.json")); os.rename(src, dst)
meta = load_json(self._meta(comp, bid), default={}); meta.update(delivery_count=0, redriven_by=operator)
atomic_write(self._meta(comp, bid), json.dumps(meta).encode())
def list_pending(self, comp):
rows = []
for area in ("_staging", "pub", "processing", "holding", "dlq", "done"):
base = self._d(comp, area)
if not os.path.isdir(base):
continue
items = []
if area in ("_staging", "pub"):
for f in sorted(os.listdir(base)):
if os.path.isdir(os.path.join(base, f)):
items += [(f, b, os.path.join(base, f, b)) for b in sorted(os.listdir(os.path.join(base, f)))]
else:
items = [(None, b, os.path.join(base, b)) for b in sorted(os.listdir(base))]
for f, bid, p in items:
meta = load_json(self._meta(comp, bid), default={})
rows.append({"batch_id": bid, "state": area, "folder": f or meta.get("folder"),
"delivery_count": meta.get("delivery_count", 0),
"ready": os.path.exists(os.path.join(p, READY))})
return rows
