armiger_dht_db_sync.py

"""Modular, CSAM-Protected, Encrypted, Poison/Spoof-Proof BEP-44 / BEP-46 DHT DB Sync Plugin.

Enforces:
  1. Multi-Layer CSAM & Abuse Protection:
     - Multi-stage ContentFilter normalization parity (normalization, leetspeak, zero-width stripping)
     - Deterministic & structural pre-ingest CSAM/abuse blocker
     - Ed25519-signed blinded SHA-256 tombstone purge engine
     - Armiger Mondrian conformal safety gates (junk/malware, out-of-scope, adult containment)
  2. Modular Sharded Subscription Architecture:
     - Partitioned by Category (10 modules, xxx strictly isolated & opt-in) x
       Language (14 modules) x Quality Tier (7 tiers) x Time Epoch
     - Fine-grained subscription filters: nodes only fetch and store subscribed shards
  3. X25519 + ChaCha20-Poly1305 AEAD Encryption:
     - Ephemeral / topic-derived key exchange via HKDF-SHA256
     - 12-byte random nonces and AAD binding shard ID, epoch, sequence, and Merkle root
     - Tamper-proof ciphertext with replay and cross-shard transplant defense
  4. BEP-44 / BEP-46 Root-of-Trust + K-of-N Distinct-ASN Quorum:
     - Strictly monotonic sequence counters (anti-replay/anti-downgrade)
     - Offline Ed25519 root signatures over canonical BEP-44 mutable manifests
     - Sybil-resistant K-of-N attestation quorum requiring independent distinct ASNs
     - Binary Merkle tree root verification and individual record inclusion proofs
  5. Anonymity & Per-Module Opt-In/Opt-Out:
     - Zero PII, peer IPs, or tracker credentials in payloads
     - Blinded infohash lookups and sanitized titles
     - Full sync modes: OFF, PULL_ONLY, SUBSCRIBE_AND_SEED, PUBLISHER
"""
from __future__ import annotations

import dataclasses
import enum
import hashlib
import hmac
import io
import json
import os
import pathlib
import sqlite3
import time
import zlib
from typing import Any, Iterable, Mapping, Sequence
from armiger_dht_rsi import (
    AnonymityGuard,
    AnonymityViolationError,
    _bep44_canonical_sign_buffer,
    resolve_opt_in_sharing,
)

from cryptography.exceptions import InvalidSignature, InvalidTag
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.asymmetric.ed25519 import (
    Ed25519PrivateKey,
    Ed25519PublicKey,
)
from cryptography.hazmat.primitives.asymmetric.x25519 import (
    X25519PrivateKey,
    X25519PublicKey,
)
from cryptography.hazmat.primitives.ciphers.aead import ChaCha20Poly1305
from cryptography.hazmat.primitives.kdf.hkdf import HKDF
from cryptography.hazmat.primitives.serialization import (
    Encoding,
    NoEncryption,
    PrivateFormat,
    PublicFormat,
)

from csam_guard import (
    ArmigerSafetyGate,
    CSAMGuard,
    CSAMViolationError,
    CSAMViolationType,
    ContentFilterResult,
    MLKEM768,
    SignedTombstoneList,
    TombstoneStore,
    WOTSPlusSHAKE256,
    blind_infohash,
    generate_ed25519_keypair,
    normalize_text,
    sanitize_title,
    sign_tombstone_list,
)

# Optional zstandard import with graceful zlib fallback
try:
    import zstandard as zstd  # type: ignore

    _HAS_ZSTD = True
except ImportError:
    _HAS_ZSTD = False


# ---------------------------------------------------------------------------
# Constants & Partition Taxonomies
# ---------------------------------------------------------------------------

CATEGORIES_MAINSTREAM: tuple[str, ...] = (
    "movie",
    "tv_show",
    "music",
    "game",
    "software",
    "anime",
    "book",
    "comic",
    "course",
)

CATEGORY_ADULT = "xxx"

ALL_CATEGORIES: tuple[str, ...] = CATEGORIES_MAINSTREAM + (CATEGORY_ADULT,)

LANGUAGES: tuple[str, ...] = (
    "en",
    "ja",
    "es",
    "fr",
    "de",
    "it",
    "pt",
    "ru",
    "zh",
    "ko",
    "hi",
    "ar",
    "multi",
    "other",
)

QUALITY_TIERS: tuple[str, ...] = (
    "uhd_4k",
    "hd_1080p",
    "hd_720p",
    "sd",
    "lossless_audio",
    "standard_audio",
    "any",
)

DEFAULT_EPOCH = "2026_q4"
SQLITE_MAX_INT64 = (1 << 63) - 1


class SyncMode(str, enum.Enum):
    """Operational mode for decentralized DHT sync."""

    OFF = "off"
    PULL_ONLY = "pull_only"
    SUBSCRIBE_AND_SEED = "subscribe_and_seed"
    PUBLISHER = "publisher"


# ---------------------------------------------------------------------------
# Data Models
# ---------------------------------------------------------------------------

@dataclasses.dataclass(frozen=True)
class TorrentRecord:
    """Canonical representation of an indexed torrent entry."""

    infohash: str  # 40-char hex
    title: str
    category: str
    language: str
    quality_tier: str
    epoch: str
    size_bytes: int
    file_count: int
    seeders: int
    leechers: int
    content_sha256: str
    file_list: list[str]
    safety_gate_passed: bool
    created_at: int

    def canonical_dict(self) -> dict[str, Any]:
        """Deterministic dictionary for serialization and hashing."""
        return {
            "category": self.category.strip().lower(),
            "content_sha256": self.content_sha256.strip().lower(),
            "created_at": int(self.created_at),
            "epoch": self.epoch.strip().lower(),
            "file_count": int(self.file_count),
            "file_list": sorted(self.file_list),
            "infohash": self.infohash.strip().lower(),
            "language": self.language.strip().lower(),
            "leechers": int(self.leechers),
            "quality_tier": self.quality_tier.strip().lower(),
            "safety_gate_passed": bool(self.safety_gate_passed),
            "seeders": int(self.seeders),
            "size_bytes": int(self.size_bytes),
            "title": self.title,
        }

    def canonical_bytes(self) -> bytes:
        return json.dumps(self.canonical_dict(), sort_keys=True, separators=(",", ":")).encode("utf-8")

    def leaf_hash(self) -> bytes:
        return hashlib.sha256(self.canonical_bytes()).digest()


@dataclasses.dataclass(frozen=True)
class QuorumAttestation:
    """Individual node attestation with Hybrid Ed25519 + WOTS+-SHAKE256 PQ signatures."""

    node_id: str  # 40-char hex DHT node ID
    asn: int  # Autonomous System Number (e.g. AS13335)
    ip_digest: str  # Blinded SHA256 of IP/subnet
    pubkey_hex: str  # Ed25519 public key (64 hex chars)
    signature_hex: str  # Ed25519 signature over manifest canonical hash
    timestamp: int
    pq_pubkey_hex: str = ""  # WOTS+-SHAKE256 PQ public key (64 hex chars)
    pq_signature_hex: str = ""  # WOTS+-SHAKE256 PQ signature (4352 hex chars)

    def canonical_dict(self) -> dict[str, Any]:
        return {
            "asn": int(self.asn),
            "ip_digest": self.ip_digest.lower(),
            "node_id": self.node_id.lower(),
            "pubkey_hex": self.pubkey_hex.lower(),
            "timestamp": int(self.timestamp),
        }

    def signing_payload(self, manifest_digest: bytes) -> bytes:
        meta = json.dumps(self.canonical_dict(), sort_keys=True, separators=(",", ":")).encode("utf-8")
        return hashlib.sha256(b"armiger-attestation-v2:" + manifest_digest + b":" + meta).digest()


@dataclasses.dataclass(frozen=True)
class DHTShardManifest:
    """BEP-44 / BEP-46 Hybrid Post-Quantum (Ed25519 + WOTS+-SHAKE256) signed manifest for an encrypted shard."""

    seq: int
    shard_id: str
    epoch: str
    record_count: int
    compressed_size_bytes: int
    uncompressed_size_bytes: int
    merkle_root: str  # 64-char hex
    ciphertext_sha256: str  # 64-char hex
    bep46_infohash: str  # 40-char hex
    timestamp: int
    root_pubkey_hex: str
    ed25519_signature_hex: str
    attestations: list[QuorumAttestation] = dataclasses.field(default_factory=list)
    root_pq_pubkey_hex: str = ""
    pq_signature_hex: str = ""

    def with_attestations(self, attestations: Sequence[QuorumAttestation]) -> DHTShardManifest:
        """Return a copy of the manifest with attached quorum attestations."""
        return dataclasses.replace(self, attestations=list(attestations))

    def canonical_dict_without_sig(self) -> dict[str, Any]:
        d: dict[str, Any] = {
            "bep46_infohash": self.bep46_infohash.lower(),
            "ciphertext_sha256": self.ciphertext_sha256.lower(),
            "compressed_size_bytes": int(self.compressed_size_bytes),
            "epoch": self.epoch.lower(),
            "merkle_root": self.merkle_root.lower(),
            "record_count": int(self.record_count),
            "root_pubkey_hex": self.root_pubkey_hex.lower(),
            "seq": int(self.seq),
            "shard_id": self.shard_id.lower(),
            "timestamp": int(self.timestamp),
            "uncompressed_size_bytes": int(self.uncompressed_size_bytes),
        }
        if self.root_pq_pubkey_hex:
            d["root_pq_pubkey_hex"] = self.root_pq_pubkey_hex.strip().lower()
        return d

    def canonical_payload(self) -> bytes:
        return json.dumps(self.canonical_dict_without_sig(), sort_keys=True, separators=(",", ":")).encode("utf-8")

    def canonical_digest(self) -> bytes:
        return hashlib.sha256(self.canonical_payload()).digest()

    def bep44_target(self, salt: str | None = None) -> str:
        """Compute BEP-44 DHT target lookup key: SHA1(pubkey || salt)."""
        pub_bytes = bytes.fromhex(self.root_pubkey_hex.strip().lower())
        salt_bytes = (salt.lower() if salt else f"armiger-shard:{self.shard_id.strip().lower()}").encode("utf-8")
        return hashlib.sha1(pub_bytes + salt_bytes).hexdigest()


# ---------------------------------------------------------------------------
# Merkle Tree Implementation
# ---------------------------------------------------------------------------

class MerkleTree:
    """Binary Merkle tree over canonical record leaf hashes with domain-separated odd-node padding."""

    ODD_NODE_SENTINEL: bytes = hashlib.sha256(b"armiger-merkle-odd-node-sentinel-v1").digest()

    def __init__(self, leaves: Sequence[bytes]) -> None:
        self.leaves = list(leaves)
        self.levels: list[list[bytes]] = []
        self._build()

    def _build(self) -> None:
        if not self.leaves:
            self.root = b"\x00" * 32
            self.levels = [[]]
            return

        current = self.leaves[:]
        self.levels.append(current)

        while len(current) > 1:
            next_level: list[bytes] = []
            for i in range(0, len(current), 2):
                left = current[i]
                right = current[i + 1] if i + 1 < len(current) else self.ODD_NODE_SENTINEL
                parent = hashlib.sha256(left + right).digest()
                next_level.append(parent)
            self.levels.append(next_level)
            current = next_level

        self.root = self.levels[-1][0]

    @property
    def root_hex(self) -> str:
        return self.root.hex()

    def get_proof(self, index: int) -> list[tuple[str, bool]]:
        """Return list of (sibling_hash_hex, is_right) proof pairs for leaf at index."""
        if index < 0 or index >= len(self.leaves):
            raise IndexError(f"Leaf index {index} out of range (total {len(self.leaves)})")

        proof: list[tuple[str, bool]] = []
        cur_idx = index
        for level in self.levels[:-1]:
            is_right = (cur_idx % 2 == 0)
            sib_idx = cur_idx + 1 if is_right else cur_idx - 1
            if sib_idx >= len(level):
                proof.append((self.ODD_NODE_SENTINEL.hex(), True))
            else:
                proof.append((level[sib_idx].hex(), is_right))
            cur_idx //= 2

        return proof

    @staticmethod
    def verify_proof(leaf_hash: bytes, proof: Sequence[tuple[str, bool]], root_hex: str) -> bool:
        """Verify an inclusion proof against expected Merkle root."""
        if not isinstance(leaf_hash, (bytes, bytearray)) or len(leaf_hash) != 32:
            return False
        try:
            expected_root = bytes.fromhex(str(root_hex).strip())
            if len(expected_root) != 32:
                return False
        except (ValueError, TypeError):
            return False

        current = bytes(leaf_hash)
        for item in proof:
            if not isinstance(item, (tuple, list)) or len(item) != 2:
                return False
            sib_hex, is_right = item
            try:
                sib = bytes.fromhex(str(sib_hex).strip())
            except (ValueError, TypeError):
                return False
            if len(sib) != 32:
                return False
            if is_right:
                current = hashlib.sha256(current + sib).digest()
            else:
                current = hashlib.sha256(sib + current).digest()
        return hmac.compare_digest(current, expected_root)


# ---------------------------------------------------------------------------
# X25519 & ChaCha20-Poly1305 AEAD Cryptographic Engine
# ---------------------------------------------------------------------------

class ShardCryptoEngine:
    """Post-Quantum Hybrid (X25519 + ML-KEM-768 + SHAKE-256 / HKDF-SHA3-256) & ChaCha20-Poly1305 AEAD Engine."""

    HYBRID_PRIV_LEN = 32 + MLKEM768.DK_BYTES  # 2,432 bytes
    HYBRID_PUB_LEN = 32 + MLKEM768.EK_BYTES   # 1,216 bytes
    HYBRID_ENC_LEN = 32 + MLKEM768.CT_BYTES   # 1,120 bytes

    @staticmethod
    def derive_shard_key(
        master_secret: bytes,
        shard_id: str,
        epoch: str,
        pq_shared_secret: bytes = b"",
    ) -> bytes:
        """Derive 256-bit ChaCha20-Poly1305 key via Dual-PRF (HKDF-SHA256 XOR SHAKE-256/SHA3-256)."""
        salt_bytes = shard_id.lower().encode("utf-8")
        info_bytes = f"armiger-shard-chacha20-poly1305-pq-v2:{epoch.lower()}".encode("utf-8")
        ikm = master_secret + pq_shared_secret
        k_hkdf = HKDF(
            algorithm=hashes.SHA256(),
            length=32,
            salt=salt_bytes,
            info=info_bytes,
        ).derive(ikm)
        k_shake = hashlib.shake_256(
            b"armiger-pq-shard-kdf-v2:" + salt_bytes + b":" + info_bytes + b":" + ikm
        ).digest(32)
        return bytes(a ^ b for a, b in zip(k_hkdf, k_shake))

    @classmethod
    def generate_pq_keypair(cls) -> tuple[bytes, bytes]:
        """Generate a Hybrid Post-Quantum (X25519 + ML-KEM-768) keypair: (priv_2432B, pub_1216B)."""
        priv = X25519PrivateKey.generate()
        pub = priv.public_key()
        x_priv = priv.private_bytes(Encoding.Raw, PrivateFormat.Raw, NoEncryption())
        x_pub = pub.public_bytes(Encoding.Raw, PublicFormat.Raw)
        pq_ek, pq_dk = MLKEM768.keygen()
        return x_priv + pq_dk, x_pub + pq_ek

    @classmethod
    def encapsulate_shard_key(
        cls,
        recipient_hybrid_pubkey: bytes,
        shard_id: str,
        epoch: str,
    ) -> tuple[bytes, bytes]:
        """Encapsulate a 256-bit shard key using Hybrid X25519 + ML-KEM-768, returning (kem_token_1120B, key_32B)."""
        if len(recipient_hybrid_pubkey) != cls.HYBRID_PUB_LEN:
            raise ValueError(f"Expected {cls.HYBRID_PUB_LEN}-byte hybrid public key")
        recip_x_pub = recipient_hybrid_pubkey[:32]
        recip_pq_ek = recipient_hybrid_pubkey[32:]
        ephem_priv = X25519PrivateKey.generate()
        ephem_x_pub = ephem_priv.public_key().public_bytes(Encoding.Raw, PublicFormat.Raw)
        ss_classical = ephem_priv.exchange(X25519PublicKey.from_public_bytes(recip_x_pub))
        pq_ct, ss_pq = MLKEM768.encaps(recip_pq_ek)
        kem_token = ephem_x_pub + pq_ct
        ctx = hashlib.sha3_256(kem_token).digest()
        key = cls.derive_shard_key(ss_classical, shard_id, epoch, pq_shared_secret=ss_pq + ctx)
        return kem_token, key

    @classmethod
    def decapsulate_shard_key(
        cls,
        recipient_hybrid_privkey: bytes,
        kem_token: bytes,
        shard_id: str,
        epoch: str,
    ) -> bytes:
        """Decapsulate a 1,120-byte X25519 + ML-KEM-768 token to recover the 256-bit shard key."""
        if len(recipient_hybrid_privkey) != cls.HYBRID_PRIV_LEN or len(kem_token) != cls.HYBRID_ENC_LEN:
            raise ValueError("Invalid hybrid private key or KEM encapsulation token length")
        x_priv = X25519PrivateKey.from_private_bytes(recipient_hybrid_privkey[:32])
        pq_dk = recipient_hybrid_privkey[32:]
        ephem_x_pub = kem_token[:32]
        pq_ct = kem_token[32:]
        ss_classical = x_priv.exchange(X25519PublicKey.from_public_bytes(ephem_x_pub))
        ss_pq = MLKEM768.decaps(pq_dk, pq_ct)
        ctx = hashlib.sha3_256(kem_token).digest()
        return cls.derive_shard_key(ss_classical, shard_id, epoch, pq_shared_secret=ss_pq + ctx)

    @staticmethod
    def construct_aad(shard_id: str, epoch: str, seq: int, merkle_root: str) -> bytes:
        """Deterministic Authenticated Associated Data (AAD) binding ciphertext to shard metadata."""
        return f"armiger-shard-aad-v1:{shard_id.lower()}:{epoch.lower()}:{seq}:{merkle_root.lower()}".encode("utf-8")

    @classmethod
    def encrypt_shard(
        cls,
        plaintext: bytes,
        key: bytes,
        shard_id: str,
        epoch: str,
        seq: int,
        merkle_root: str,
    ) -> bytes:
        """Encrypt shard payload with 256-bit ChaCha20-Poly1305 AEAD (128-bit Grover PQ security).

        Format: nonce (12 bytes) + ciphertext_with_poly1305_tag
        """
        if len(key) != 32:
            raise ValueError(f"ChaCha20 key must be 32 bytes, got {len(key)}")
        nonce = os.urandom(12)
        aad = cls.construct_aad(shard_id, epoch, seq, merkle_root)
        aead = ChaCha20Poly1305(key)
        ciphertext = aead.encrypt(nonce, plaintext, aad)
        return nonce + ciphertext

    @classmethod
    def decrypt_shard(
        cls,
        bundle: bytes,
        key: bytes,
        shard_id: str,
        epoch: str,
        seq: int,
        merkle_root: str,
    ) -> bytes:
        """Decrypt shard payload and verify Poly1305 tag against AAD.

        Raises InvalidTag if tampered or transplanted across different shard_id/epoch/seq.
        """
        if len(bundle) < 12 + 16:
            raise ValueError("Encrypted bundle is too short to contain nonce and tag")
        nonce = bundle[:12]
        ciphertext = bundle[12:]
        aad = cls.construct_aad(shard_id, epoch, seq, merkle_root)
        aead = ChaCha20Poly1305(key)
        return aead.decrypt(nonce, ciphertext, aad)

    @classmethod
    def encrypt_shard_pq(
        cls,
        plaintext: bytes,
        recipient_hybrid_pubkey: bytes,
        shard_id: str,
        epoch: str,
        seq: int,
        merkle_root: str,
    ) -> bytes:
        """Encrypt shard with ephemeral X25519 + ML-KEM-768 encapsulation + 256-bit ChaCha20-Poly1305 AEAD.

        Format: kem_token (1,120 bytes) + nonce (12 bytes) + ciphertext_with_poly1305_tag
        """
        kem_token, key = cls.encapsulate_shard_key(recipient_hybrid_pubkey, shard_id, epoch)
        nonce = os.urandom(12)
        aad = cls.construct_aad(shard_id, epoch, seq, merkle_root) + b":kem:" + hashlib.sha3_256(kem_token).digest()
        aead = ChaCha20Poly1305(key)
        ciphertext = aead.encrypt(nonce, plaintext, aad)
        return kem_token + nonce + ciphertext

    @classmethod
    def decrypt_shard_pq(
        cls,
        pq_bundle: bytes,
        recipient_hybrid_privkey: bytes,
        shard_id: str,
        epoch: str,
        seq: int,
        merkle_root: str,
    ) -> bytes:
        """Decrypt an X25519 + ML-KEM-768 + ChaCha20-Poly1305 shard bundle."""
        min_len = cls.HYBRID_ENC_LEN + 12 + 16
        if len(pq_bundle) < min_len:
            raise ValueError("PQ encrypted shard bundle too short")
        kem_token = pq_bundle[: cls.HYBRID_ENC_LEN]
        nonce = pq_bundle[cls.HYBRID_ENC_LEN : cls.HYBRID_ENC_LEN + 12]
        ciphertext = pq_bundle[cls.HYBRID_ENC_LEN + 12 :]
        key = cls.decapsulate_shard_key(recipient_hybrid_privkey, kem_token, shard_id, epoch)
        aad = cls.construct_aad(shard_id, epoch, seq, merkle_root) + b":kem:" + hashlib.sha3_256(kem_token).digest()
        aead = ChaCha20Poly1305(key)
        return aead.decrypt(nonce, ciphertext, aad)


# ---------------------------------------------------------------------------
# Compression Utilities (Deterministic Mandatory Zlib Wire Codec + Optional Zstd)
# ---------------------------------------------------------------------------

MANDATORY_WIRE_CODEC = "zlib"
ZSTD_MAGIC = b"\x28\xb5\x2f\xfd"
MAX_DECOMPRESSED_SHARD_BYTES = 4 * 1024 * 1024
MAX_DECOMPRESSION_RATIO = 50


def compress_payload(data: bytes, codec: str = MANDATORY_WIRE_CODEC) -> tuple[bytes, str]:
    """Compress payload using the mandatory deterministic wire codec ('zlib' level 9) by default.

    All peers in the swarm MUST produce identical compressed bytes for the same record set
    so that content-addressed shard digests and BEP-46 infohashes match across hosts regardless
    of whether optional `zstandard` is installed. Explicit `codec='zstd'` is supported only when
    `zstandard` is available.
    """
    norm_codec = str(codec or MANDATORY_WIRE_CODEC).strip().lower()
    if norm_codec == "zlib":
        return zlib.compress(data, level=9), "zlib"
    if norm_codec == "zstd":
        if not _HAS_ZSTD:
            raise RuntimeError("zstd_codec_unavailable: optional zstandard package is not installed")
        cctx = zstd.ZstdCompressor(level=7)
        return cctx.compress(data), "zstd"
    raise ValueError(f"Unsupported compression codec: {codec}")


def decompress_payload(
    compressed: bytes,
    codec: str = MANDATORY_WIRE_CODEC,
    max_output_bytes: int = MAX_DECOMPRESSED_SHARD_BYTES,
    max_ratio: int = MAX_DECOMPRESSION_RATIO,
    *,
    max_uncompressed_bytes: int | None = None,
) -> bytes:
    """Decompress payload using specified codec with strict zip-bomb size & ratio caps."""
    if not compressed:
        return b""
    if max_uncompressed_bytes is not None:
        max_output_bytes = int(max_uncompressed_bytes)
    hard_cap = min(max_output_bytes, max(65536, len(compressed) * max_ratio))

    def _zlib_bounded(data: bytes) -> bytes:
        try:
            dobj = zlib.decompressobj()
            out = dobj.decompress(data, hard_cap + 1)
            if dobj.unconsumed_tail or len(out) > hard_cap:
                return b""
            out += dobj.flush()
            if len(out) > hard_cap:
                return b""
            return out
        except Exception:
            return b""

    norm_codec = str(codec or MANDATORY_WIRE_CODEC).strip().lower()
    is_zstd_frame = compressed.startswith(ZSTD_MAGIC)

    if norm_codec == "zlib" and not is_zstd_frame:
        return _zlib_bounded(compressed)
    if norm_codec in ("zstd", "zlib") and is_zstd_frame:
        if not _HAS_ZSTD:
            return b""
        try:
            params = zstd.get_frame_parameters(compressed)
            if (
                params.content_size not in (0, zstd.CONTENTSIZE_UNKNOWN)
                and params.content_size > hard_cap
            ):
                return b""
            dctx = zstd.ZstdDecompressor()
            with dctx.stream_reader(io.BytesIO(compressed)) as reader:
                out = reader.read(hard_cap + 1)
            if len(out) > hard_cap:
                return b""
            return out
        except Exception:
            return b""
    if norm_codec == "zstd":
        return _zlib_bounded(compressed)
    raise ValueError(f"Unsupported compression codec: {codec}")


# ---------------------------------------------------------------------------
# Modular Sharded Subscription Config
# ---------------------------------------------------------------------------

@dataclasses.dataclass
class SubscriptionConfig:
    """User-selected subscription preferences for modular DHT sharding."""

    categories: set[str] = dataclasses.field(default_factory=lambda: {"movie", "tv_show"})
    languages: set[str] = dataclasses.field(default_factory=lambda: {"en", "ja"})
    quality_tiers: set[str] = dataclasses.field(default_factory=lambda: {"uhd_4k", "hd_1080p"})
    epochs: set[str] = dataclasses.field(default_factory=lambda: {DEFAULT_EPOCH})
    enable_xxx: bool = False  # Strictly isolated, default OFF
    min_quorum: int = 5
    min_distinct_asns: int = 5
    sync_mode: SyncMode = SyncMode.PULL_ONLY

    @classmethod
    def parse_shard_id(cls, shard_id: str) -> tuple[str, str, str, str]:
        """Deconstruct 'category/language/quality_tier/epoch' shard identifier."""
        parts = shard_id.strip().lower().split("/")
        if len(parts) != 4:
            raise ValueError(f"Malformed shard_id: '{shard_id}'. Expected 'category/lang/quality/epoch'")
        return parts[0], parts[1], parts[2], parts[3]

    @classmethod
    def format_shard_id(cls, category: str, language: str, quality_tier: str, epoch: str) -> str:
        return f"{category.lower()}/{language.lower()}/{quality_tier.lower()}/{epoch.lower()}"

    def is_subscribed(self, shard_id: str) -> bool:
        """Check whether a given shard matches user subscription filters.

        Invariant: If category == 'xxx' and enable_xxx is False, unconditionally REJECT.
        """
        try:
            cat, lang, quality, ep = self.parse_shard_id(shard_id)
        except ValueError:
            return False

        # Strict isolation gate for adult content
        if cat == CATEGORY_ADULT and not self.enable_xxx:
            return False

        return (
            cat in self.categories
            and lang in self.languages
            and quality in self.quality_tiers
            and ep in self.epochs
        )


# ---------------------------------------------------------------------------
# SQLite Columnar Shard Database Engine
# ---------------------------------------------------------------------------

class ShardDatabase:
    """Local SQLite database storing partitioned torrent metadata."""

    def __init__(self, db_path: str | pathlib.Path = ":memory:", enable_xxx: bool = False) -> None:
        self.db_path = str(db_path)
        self.enable_xxx = enable_xxx
        self.conn = sqlite3.connect(self.db_path, timeout=10.0)
        self.conn.row_factory = sqlite3.Row
        self._init_schema()

    def _init_schema(self) -> None:
        with self.conn:
            # Mainstream torrents table
            self.conn.execute("""
                CREATE TABLE IF NOT EXISTS mainstream_torrents (
                    infohash TEXT PRIMARY KEY,
                    title TEXT NOT NULL,
                    category TEXT NOT NULL,
                    language TEXT NOT NULL,
                    quality_tier TEXT NOT NULL,
                    epoch TEXT NOT NULL,
                    size_bytes INTEGER NOT NULL,
                    file_count INTEGER NOT NULL,
                    seeders INTEGER NOT NULL,
                    leechers INTEGER NOT NULL,
                    content_sha256 TEXT NOT NULL,
                    file_list_json TEXT NOT NULL,
                    safety_gate_passed INTEGER NOT NULL DEFAULT 1,
                    created_at INTEGER NOT NULL
                )
            """)
            try:
                self.conn.execute("ALTER TABLE mainstream_torrents ADD COLUMN safety_gate_passed INTEGER NOT NULL DEFAULT 1")
            except sqlite3.OperationalError:
                pass
            self.conn.execute("CREATE INDEX IF NOT EXISTS idx_main_cat_lang ON mainstream_torrents(category, language)")
            self.conn.execute("CREATE INDEX IF NOT EXISTS idx_main_epoch ON mainstream_torrents(epoch)")
            self.conn.execute("""
                CREATE TABLE IF NOT EXISTS shard_sequences (
                    shard_id TEXT PRIMARY KEY,
                    seq INTEGER NOT NULL,
                    updated_at INTEGER NOT NULL
                )
            """)

            # Isolated adult table
            if self.enable_xxx:
                self.conn.execute("""
                    CREATE TABLE IF NOT EXISTS adult_torrents (
                        infohash TEXT PRIMARY KEY,
                        title TEXT NOT NULL,
                        category TEXT NOT NULL,
                        language TEXT NOT NULL,
                        quality_tier TEXT NOT NULL,
                        epoch TEXT NOT NULL,
                        size_bytes INTEGER NOT NULL,
                        file_count INTEGER NOT NULL,
                        seeders INTEGER NOT NULL,
                        leechers INTEGER NOT NULL,
                        content_sha256 TEXT NOT NULL,
                        file_list_json TEXT NOT NULL,
                        safety_gate_passed INTEGER NOT NULL DEFAULT 1,
                        created_at INTEGER NOT NULL
                    )
                """)
                try:
                    self.conn.execute("ALTER TABLE adult_torrents ADD COLUMN safety_gate_passed INTEGER NOT NULL DEFAULT 1")
                except sqlite3.OperationalError:
                    pass
                self.conn.execute("CREATE INDEX IF NOT EXISTS idx_adult_lang ON adult_torrents(language)")

    def get_shard_sequences(self) -> dict[str, int]:
        """Return durable per-shard monotonic sequence high-water marks from SQLite."""
        cur = self.conn.cursor()
        cur.execute("SELECT shard_id, seq FROM shard_sequences")
        return {str(row["shard_id"]): int(row["seq"]) for row in cur.fetchall()}

    def insert_record(self, record: TorrentRecord) -> bool:
        """Insert a torrent record into the appropriate partitioned table."""
        if not record.safety_gate_passed:
            return False
        for fname, fval in (
            ("size_bytes", record.size_bytes),
            ("file_count", record.file_count),
            ("seeders", record.seeders),
            ("leechers", record.leechers),
            ("created_at", record.created_at),
        ):
            if isinstance(fval, bool) or not isinstance(fval, int) or fval < 0:
                raise ValueError(f"invalid_shard_record_int_field:{fname}")
            if fval > SQLITE_MAX_INT64:
                raise ValueError(f"integer_overflow_for_sqlite_int64:{fname}")
        cat_clean = record.category.strip().lower()
        if cat_clean not in ALL_CATEGORIES:
            return False
        is_adult = (cat_clean == CATEGORY_ADULT)
        if is_adult and not self.enable_xxx:
            return False  # Drop un-opted adult record

        table = "adult_torrents" if is_adult else "mainstream_torrents"
        with self.conn:
            self.conn.execute(
                f"""
                INSERT OR REPLACE INTO {table} (
                    infohash, title, category, language, quality_tier, epoch,
                    size_bytes, file_count, seeders, leechers, content_sha256,
                    file_list_json, safety_gate_passed, created_at
                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                """,
                (
                    record.infohash.strip().lower(),
                    record.title,
                    cat_clean,
                    record.language.strip().lower(),
                    record.quality_tier.strip().lower(),
                    record.epoch.strip().lower(),
                    record.size_bytes,
                    record.file_count,
                    record.seeders,
                    record.leechers,
                    record.content_sha256.lower(),
                    json.dumps(record.file_list),
                    1 if record.safety_gate_passed else 0,
                    record.created_at,
                ),
            )
        return True

    def insert_records_batch(
        self,
        records: Sequence[TorrentRecord],
        shard_id: str | None = None,
        seq: int | None = None,
    ) -> int:
        """Batch-insert validated torrent records and advance shard sequence within a single SQLite transaction."""
        if not records and (shard_id is None or seq is None):
            return 0
        main_rows: list[tuple[Any, ...]] = []
        adult_rows: list[tuple[Any, ...]] = []
        for record in records:
            if not record.safety_gate_passed:
                raise ValueError(f"record_failed_safety_gate:{record.infohash}")
            for fname, fval in (
                ("size_bytes", record.size_bytes),
                ("file_count", record.file_count),
                ("seeders", record.seeders),
                ("leechers", record.leechers),
                ("created_at", record.created_at),
            ):
                if isinstance(fval, bool) or not isinstance(fval, int) or fval < 0:
                    raise ValueError(f"invalid_shard_record_int_field:{fname}")
                if fval > SQLITE_MAX_INT64:
                    raise ValueError(f"integer_overflow_for_sqlite_int64:{fname}")
            cat_clean = record.category.strip().lower()
            if cat_clean not in ALL_CATEGORIES:
                continue
            is_adult = (cat_clean == CATEGORY_ADULT)
            if is_adult and not self.enable_xxx:
                continue
            row = (
                record.infohash.strip().lower(),
                record.title,
                cat_clean,
                record.language.strip().lower(),
                record.quality_tier.strip().lower(),
                record.epoch.strip().lower(),
                record.size_bytes,
                record.file_count,
                record.seeders,
                record.leechers,
                record.content_sha256.lower(),
                json.dumps(record.file_list),
                1 if record.safety_gate_passed else 0,
                record.created_at,
            )
            if is_adult:
                adult_rows.append(row)
            else:
                main_rows.append(row)

        if self.conn.in_transaction:
            self.conn.commit()
        self.conn.execute("BEGIN IMMEDIATE")
        try:
            if shard_id is not None and seq is not None:
                norm_shard = shard_id.strip().lower()
                cur_row = self.conn.execute(
                    "SELECT seq FROM shard_sequences WHERE shard_id = ?",
                    (norm_shard,),
                ).fetchone()
                if cur_row is not None and int(seq) <= int(cur_row["seq"]):
                    raise ValueError(f"stale_or_downgrade_sequence:{seq}<={cur_row['seq']}")
            if main_rows:
                self.conn.executemany(
                    """
                    INSERT OR REPLACE INTO mainstream_torrents (
                        infohash, title, category, language, quality_tier, epoch,
                        size_bytes, file_count, seeders, leechers, content_sha256,
                        file_list_json, safety_gate_passed, created_at
                    ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                    """,
                    main_rows,
                )
            if adult_rows and self.enable_xxx:
                self.conn.executemany(
                    """
                    INSERT OR REPLACE INTO adult_torrents (
                        infohash, title, category, language, quality_tier, epoch,
                        size_bytes, file_count, seeders, leechers, content_sha256,
                        file_list_json, safety_gate_passed, created_at
                    ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                    """,
                    adult_rows,
                )
            if shard_id is not None and seq is not None:
                self.conn.execute(
                    """
                    INSERT INTO shard_sequences (shard_id, seq, updated_at)
                    VALUES (?, ?, ?)
                    ON CONFLICT(shard_id) DO UPDATE SET
                        seq = excluded.seq,
                        updated_at = excluded.updated_at
                    WHERE excluded.seq > shard_sequences.seq
                    """,
                    (shard_id.strip().lower(), int(seq), int(time.time())),
                )
            self.conn.commit()
        except Exception:
            if self.conn.in_transaction:
                self.conn.rollback()
            raise
        return len(main_rows) + len(adult_rows)

    def get_record(self, infohash: str) -> TorrentRecord | None:
        """Retrieve record by infohash across accessible tables."""
        clean = infohash.lower().strip()
        cur = self.conn.cursor()
        cur.execute("SELECT * FROM mainstream_torrents WHERE infohash = ?", (clean,))
        row = cur.fetchone()
        if row is None and self.enable_xxx:
            cur.execute("SELECT * FROM adult_torrents WHERE infohash = ?", (clean,))
            row = cur.fetchone()

        if row is None:
            return None

        return TorrentRecord(
            infohash=row["infohash"],
            title=row["title"],
            category=row["category"],
            language=row["language"],
            quality_tier=row["quality_tier"],
            epoch=row["epoch"],
            size_bytes=row["size_bytes"],
            file_count=row["file_count"],
            seeders=row["seeders"],
            leechers=row["leechers"],
            content_sha256=row["content_sha256"],
            file_list=json.loads(row["file_list_json"]),
            safety_gate_passed=bool(row["safety_gate_passed"]),
            created_at=row["created_at"],
        )

    def purge_blinded_hashes(self, tombstone_store: TombstoneStore) -> int:
        """Purge any records whose blinded infohashes are registered on tombstone list."""
        purged = 0
        cur = self.conn.cursor()

        # Check mainstream
        cur.execute("SELECT infohash FROM mainstream_torrents")
        main_hashes = [r[0] for r in cur.fetchall()]
        to_delete_main = [h for h in main_hashes if tombstone_store.is_tombstoned(h)]
        if to_delete_main:
            with self.conn:
                self.conn.executemany("DELETE FROM mainstream_torrents WHERE infohash = ?", [(h,) for h in to_delete_main])
            purged += len(to_delete_main)

        # Check adult if enabled
        if self.enable_xxx:
            cur.execute("SELECT infohash FROM adult_torrents")
            adult_hashes = [r[0] for r in cur.fetchall()]
            to_delete_adult = [h for h in adult_hashes if tombstone_store.is_tombstoned(h)]
            if to_delete_adult:
                with self.conn:
                    self.conn.executemany("DELETE FROM adult_torrents WHERE infohash = ?", [(h,) for h in to_delete_adult])
                purged += len(to_delete_adult)

        return purged

    def count_records(self, category: str | None = None) -> int:
        cur = self.conn.cursor()
        if category == CATEGORY_ADULT:
            if not self.enable_xxx:
                return 0
            cur.execute("SELECT COUNT(*) FROM adult_torrents")
            return int(cur.fetchone()[0])
        elif category is not None:
            cur.execute("SELECT COUNT(*) FROM mainstream_torrents WHERE category = ?", (category.lower(),))
            return int(cur.fetchone()[0])
        else:
            cur.execute("SELECT COUNT(*) FROM mainstream_torrents")
            tot = int(cur.fetchone()[0])
            if self.enable_xxx:
                cur.execute("SELECT COUNT(*) FROM adult_torrents")
                tot += int(cur.fetchone()[0])
            return tot

    def search(self, query: str = "", category: str | None = None, language: str | None = None) -> list[TorrentRecord]:
        """Search records by title query, category, and language."""
        cur = self.conn.cursor()
        is_adult_query = (category == CATEGORY_ADULT)
        if is_adult_query and not self.enable_xxx:
            return []

        table = "adult_torrents" if is_adult_query else "mainstream_torrents"
        sql = f"SELECT * FROM {table} WHERE 1=1"
        params: list[Any] = []

        if query:
            sql += " AND title LIKE ?"
            params.append(f"%{query}%")
        if category and not is_adult_query:
            sql += " AND category = ?"
            params.append(category.lower())
        if language:
            sql += " AND language = ?"
            params.append(language.lower())

        cur.execute(sql, params)
        results: list[TorrentRecord] = []
        for row in cur.fetchall():
            results.append(TorrentRecord(
                infohash=row["infohash"],
                title=row["title"],
                category=row["category"],
                language=row["language"],
                quality_tier=row["quality_tier"],
                epoch=row["epoch"],
                size_bytes=row["size_bytes"],
                file_count=row["file_count"],
                seeders=row["seeders"],
                leechers=row["leechers"],
                content_sha256=row["content_sha256"],
                file_list=json.loads(row["file_list_json"]),
                safety_gate_passed=bool(row["safety_gate_passed"]),
                created_at=row["created_at"],
            ))
        return results

    def close(self) -> None:
        self.conn.close()

    def __del__(self) -> None:
        try:
            self.close()
        except Exception:
            pass


# ---------------------------------------------------------------------------
# Shard Serialization & Deserialization
# ---------------------------------------------------------------------------

def serialize_shard_records(
    records: Sequence[TorrentRecord],
    *,
    max_output_bytes: int = MAX_DECOMPRESSED_SHARD_BYTES,
    max_ratio: int = MAX_DECOMPRESSION_RATIO,
) -> tuple[bytes, MerkleTree]:
    """Serialize a list of TorrentRecords into a deterministic compressed micro-bundle.

    Enforces `MAX_DECOMPRESSED_SHARD_BYTES` and `MAX_DECOMPRESSION_RATIO` at build time
    before any one-time WOTS+ signature leaf is reserved or consumed.
    """
    # Deterministic sorting by infohash
    sorted_records = sorted(records, key=lambda r: r.infohash.lower())
    leaves = [r.leaf_hash() for r in sorted_records]
    tree = MerkleTree(leaves)

    payload_dict = {
        "format": "armiger-shard-v1",
        "record_count": len(sorted_records),
        "records": [r.canonical_dict() for r in sorted_records],
    }
    raw_json = json.dumps(payload_dict, sort_keys=True, separators=(",", ":")).encode("utf-8")
    if len(raw_json) > max_output_bytes:
        raise ValueError(
            f"shard_exceeds_decompression_bounds:uncompressed_bytes={len(raw_json)}>{max_output_bytes}"
        )
    compressed, _codec = compress_payload(raw_json, codec=MANDATORY_WIRE_CODEC)
    hard_cap = min(max_output_bytes, max(65536, len(compressed) * max_ratio))
    if len(raw_json) > hard_cap:
        raise ValueError(
            f"shard_exceeds_decompression_bounds:ratio_cap_exceeded:{len(raw_json)}>{hard_cap}"
        )
    return compressed, tree


def split_records_into_bounded_shards(
    records: Sequence[TorrentRecord],
    *,
    max_output_bytes: int = MAX_DECOMPRESSED_SHARD_BYTES,
    max_ratio: int = MAX_DECOMPRESSION_RATIO,
) -> list[list[TorrentRecord]]:
    """Partition a sequence of TorrentRecords into bounded micro-shards that each satisfy receiver decompression caps."""
    sorted_records = sorted(records, key=lambda r: r.infohash.lower())
    if not sorted_records:
        return [[]]
    batches: list[list[TorrentRecord]] = []
    current_batch: list[TorrentRecord] = []
    for rec in sorted_records:
        candidate = current_batch + [rec]
        try:
            serialize_shard_records(
                candidate,
                max_output_bytes=max_output_bytes,
                max_ratio=max_ratio,
            )
            current_batch = candidate
        except ValueError:
            if not current_batch:
                raise
            batches.append(current_batch)
            serialize_shard_records(
                [rec],
                max_output_bytes=max_output_bytes,
                max_ratio=max_ratio,
            )
            current_batch = [rec]
    if current_batch:
        batches.append(current_batch)
    return batches


def deserialize_shard_records(compressed: bytes) -> list[TorrentRecord]:
    """Decompress and parse shard records with strict schema and type validation."""
    raw_json = decompress_payload(compressed)
    payload_dict = json.loads(raw_json.decode("utf-8"))
    if not isinstance(payload_dict, dict):
        raise ValueError("shard_payload_not_dict")
    raw_records = payload_dict.get("records")
    if not isinstance(raw_records, list):
        raise ValueError("shard_records_not_list")
    records: list[TorrentRecord] = []
    for d in raw_records:
        if not isinstance(d, dict):
            raise ValueError("shard_record_item_not_dict")
        for str_field in ("infohash", "title", "category", "language", "quality_tier", "epoch", "content_sha256"):
            val = d.get(str_field)
            if not isinstance(val, str) or not val.strip():
                raise ValueError(f"invalid_shard_record_string_field:{str_field}")
        infohash_norm = d["infohash"].strip().lower()
        if len(infohash_norm) != 40 or any(ch not in "0123456789abcdef" for ch in infohash_norm):
            raise ValueError("invalid_shard_record_infohash_hex")
        sha_norm = d["content_sha256"].strip().lower()
        if len(sha_norm) != 64 or any(ch not in "0123456789abcdef" for ch in sha_norm):
            raise ValueError("invalid_shard_record_content_sha256_hex")
        for int_field in ("size_bytes", "file_count", "seeders", "leechers"):
            ival = d.get(int_field)
            if isinstance(ival, bool) or not isinstance(ival, int) or ival < 0:
                raise ValueError(f"invalid_shard_record_int_field:{int_field}")
            if ival > SQLITE_MAX_INT64:
                raise ValueError(f"integer_overflow_for_sqlite_int64:{int_field}")
        created_val = d.get("created_at", int(time.time()))
        if isinstance(created_val, bool) or not isinstance(created_val, int) or created_val < 0:
            raise ValueError("invalid_shard_record_int_field:created_at")
        if created_val > SQLITE_MAX_INT64:
            raise ValueError("integer_overflow_for_sqlite_int64:created_at")
        flist = d.get("file_list", [])
        if not isinstance(flist, list) or any(not isinstance(item, str) for item in flist):
            raise ValueError("invalid_shard_record_file_list")
        sg_val = d.get("safety_gate_passed", True)
        if not isinstance(sg_val, bool):
            raise ValueError("invalid_shard_record_safety_gate_passed")
        records.append(TorrentRecord(
            infohash=d["infohash"],
            title=d["title"],
            category=d["category"],
            language=d["language"],
            quality_tier=d["quality_tier"],
            epoch=d["epoch"],
            size_bytes=d["size_bytes"],
            file_count=d["file_count"],
            seeders=d["seeders"],
            leechers=d["leechers"],
            content_sha256=d["content_sha256"],
            file_list=list(flist),
            safety_gate_passed=sg_val,
            created_at=created_val,
        ))
    return records


# ---------------------------------------------------------------------------
# Quorum & Attestation Verification Engine
# ---------------------------------------------------------------------------

class QuorumVerifier:
    """Validates K-of-N distinct-ASN quorum attestations and Hybrid Ed25519 + WOTS+-SHAKE256 root signatures."""

    _GLOBAL_ATTESTOR_ASN_REGISTRY: dict[str, int] = {}

    @classmethod
    def register_trusted_attestor(cls, pubkey_hex: str, asn: int) -> None:
        """Bind an attestor Ed25519 public key to its authenticated Autonomous System Number."""
        norm_pk = str(pubkey_hex).strip().lower()
        if len(norm_pk) == 64:
            cls._GLOBAL_ATTESTOR_ASN_REGISTRY[norm_pk] = int(asn)

    def __init__(
        self,
        trusted_root_pubkeys_hex: Sequence[str],
        min_quorum: int = 5,
        min_distinct_asns: int = 5,
        trusted_root_pq_pubkeys_hex: Sequence[str] | None = None,
        *,
        require_pq: bool = True,
        trusted_attestor_asns: Mapping[str, int] | None = None,
    ) -> None:
        self.trusted_roots: list[bytes] = []
        self.trusted_pq_roots: list[bytes] = [
            bytes.fromhex(k.strip().lower())
            for k in (trusted_root_pq_pubkeys_hex or [])
            if len(k.strip()) == 64
        ]
        self.ed_to_pq_roots: dict[bytes, list[bytes]] = {}
        raw_ed_entries = list(trusted_root_pubkeys_hex or [])
        for idx, raw_entry in enumerate(raw_ed_entries):
            cleaned = str(raw_entry).strip().lower()
            if ":" in cleaned:
                ed_part, pq_part = cleaned.split(":", 1)
                ed_b: bytes | None = None
                if len(ed_part.strip()) == 64:
                    ed_b = bytes.fromhex(ed_part.strip())
                    if ed_b not in self.trusted_roots:
                        self.trusted_roots.append(ed_b)
                if len(pq_part.strip()) == 64:
                    pq_b = bytes.fromhex(pq_part.strip())
                    if pq_b not in self.trusted_pq_roots:
                        self.trusted_pq_roots.append(pq_b)
                    if ed_b is not None:
                        bound_list = self.ed_to_pq_roots.setdefault(ed_b, [])
                        if pq_b not in bound_list:
                            bound_list.append(pq_b)
            elif len(cleaned) == 64:
                ed_b = bytes.fromhex(cleaned)
                if ed_b not in self.trusted_roots:
                    self.trusted_roots.append(ed_b)
                bound_list = self.ed_to_pq_roots.setdefault(ed_b, [])
                if self.trusted_pq_roots:
                    if len(self.trusted_pq_roots) == len(raw_ed_entries) and idx < len(self.trusted_pq_roots):
                        pq_b = self.trusted_pq_roots[idx]
                        if pq_b not in bound_list:
                            bound_list.append(pq_b)
                    else:
                        bound_root_pq = WOTSPlusSHAKE256.get_root_wots_pub(cleaned)
                        if bound_root_pq and len(bound_root_pq) == 64:
                            pq_b = bytes.fromhex(bound_root_pq)
                            if pq_b in self.trusted_pq_roots and pq_b not in bound_list:
                                bound_list.append(pq_b)
                        elif len(raw_ed_entries) == 1:
                            for pq_b in self.trusted_pq_roots:
                                if pq_b not in bound_list:
                                    bound_list.append(pq_b)
                else:
                    bound_root_pq = WOTSPlusSHAKE256.get_root_wots_pub(cleaned)
                    if bound_root_pq and len(bound_root_pq) == 64:
                        pq_b = bytes.fromhex(bound_root_pq)
                        if pq_b not in self.trusted_pq_roots:
                            self.trusted_pq_roots.append(pq_b)
                        if pq_b not in bound_list:
                            bound_list.append(pq_b)
        self.min_quorum = min_quorum
        self.min_distinct_asns = min_distinct_asns
        self.require_pq = bool(require_pq)
        self.trusted_attestor_asns: dict[str, int] = {
            str(k).strip().lower(): int(v)
            for k, v in (trusted_attestor_asns or {}).items()
            if len(str(k).strip()) == 64
        }
        self._used_pq_pubkeys: dict[str, bytes] = {}

    def verify_manifest(
        self,
        manifest: DHTShardManifest,
        current_seq: int = -1,
    ) -> tuple[bool, str]:
        """Verify:
          1. Monotonic sequence: seq > current_seq
          2. Root Hybrid Ed25519 + WOTS+-SHAKE256 Post-Quantum signature
          3. K-of-N attestation count >= min_quorum
          4. Distinct ASNs >= min_distinct_asns (with authenticated ASN bindings)
          5. Each attestation Hybrid Ed25519 + WOTS+-SHAKE256 signature
        """
        # 0. Policy & type sanity checks
        if isinstance(manifest.seq, bool) or not isinstance(manifest.seq, int) or manifest.seq < 0:
            return False, "invalid_manifest_seq_type"
        if self.min_quorum <= 0 or self.min_distinct_asns <= 0:
            return False, "invalid_quorum_policy_threshold"
        if not self.trusted_roots:
            return False, "no_trusted_root_pubkeys_configured"

        # 1. Monotonic sequence & timestamp freshness check
        if manifest.seq <= current_seq:
            return False, f"stale_or_downgrade_sequence:{manifest.seq}<={current_seq}"
        if manifest.seq > (2**63 - 1) or (current_seq >= 0 and manifest.seq > current_seq + 1024):
            return False, f"sequence_jump_exceeds_max_window:{manifest.seq}"
        now_manifest_ts = int(time.time())
        if (
            isinstance(manifest.timestamp, bool)
            or not isinstance(manifest.timestamp, int)
            or manifest.timestamp <= 0
            or (manifest.timestamp - now_manifest_ts) > 300
        ):
            return False, f"manifest_timestamp_out_of_bounds:{manifest.timestamp}"

        # 2. Root Ed25519 signature check
        try:
            root_pub_bytes = bytes.fromhex(manifest.root_pubkey_hex.strip().lower())
        except ValueError:
            return False, "invalid_root_pubkey_hex"
        if root_pub_bytes not in self.trusted_roots:
            return False, f"untrusted_root_pubkey:{manifest.root_pubkey_hex}"

        payload = manifest.canonical_payload()
        try:
            vk = Ed25519PublicKey.from_public_bytes(root_pub_bytes)
            vk.verify(bytes.fromhex(manifest.ed25519_signature_hex.strip()), payload)
        except (InvalidSignature, ValueError) as exc:
            return False, f"invalid_root_ed25519_signature:{exc}"

        # 2b. Root WOTS+-SHAKE256 Post-Quantum signature check (mandatory when require_pq=True)
        has_root_pq_fields = bool(manifest.pq_signature_hex or manifest.root_pq_pubkey_hex)
        if self.require_pq or has_root_pq_fields:
            if not manifest.root_pq_pubkey_hex.strip() or not manifest.pq_signature_hex.strip():
                return False, "missing_root_pq_wots_signature"
            norm_root_pq = manifest.root_pq_pubkey_hex.strip().lower()
            try:
                pq_root_bytes = bytes.fromhex(norm_root_pq)
                pq_sig_bytes = bytes.fromhex(manifest.pq_signature_hex.strip())
            except ValueError as exc:
                return False, f"invalid_root_pq_wots_signature:{exc}"
            bound_pq_roots = list(self.ed_to_pq_roots.get(root_pub_bytes, []))
            if not bound_pq_roots:
                dyn_pq_hex = WOTSPlusSHAKE256.get_root_wots_pub(root_pub_bytes.hex())
                if dyn_pq_hex and len(dyn_pq_hex) == 64:
                    dyn_b = bytes.fromhex(dyn_pq_hex)
                    if not self.trusted_pq_roots or dyn_b in self.trusted_pq_roots:
                        bound_pq_roots.append(dyn_b)
            ok_pq, pq_reason = WOTSPlusSHAKE256.verify_against_trusted_roots(
                bound_pq_roots,
                pq_root_bytes,
                payload,
                pq_sig_bytes,
                used_keys=self._used_pq_pubkeys,
            )
            if not ok_pq:
                if pq_reason == "untrusted_pq_root_pubkey":
                    return False, f"untrusted_root_pq_pubkey:{manifest.root_pq_pubkey_hex}"
                if pq_reason == "wots_one_time_key_reused":
                    return False, f"wots_one_time_key_reused:{norm_root_pq[:16]}"
                return False, "invalid_root_pq_wots_signature:verification_failed"

        # 3. Quorum threshold check
        if len(manifest.attestations) < self.min_quorum:
            return False, f"insufficient_attestations:{len(manifest.attestations)}<{self.min_quorum}"

        # 4. Validate individual attestations & count verified attestations and distinct ASNs
        manifest_digest = manifest.canonical_digest()
        verified_attestations_count = 0
        verified_asns: set[int] = set()
        seen_node_ids: set[str] = set()
        seen_pubkeys: set[str] = set()
        seen_pq_pubkeys: set[str] = set()
        seen_sigs: set[str] = set()
        seen_ip_digests: set[str] = set()
        now = int(time.time())

        for att in manifest.attestations:
            norm_node = str(att.node_id).strip().lower()
            norm_pk = str(att.pubkey_hex).strip().lower()
            norm_sig = str(att.signature_hex).strip().lower()
            norm_ip = str(att.ip_digest).strip().lower()

            # Reject duplicate node_id, pubkey, signature, or IP digest Sybil replay
            if norm_pk in seen_pubkeys or norm_sig in seen_sigs:
                return False, f"sybil_duplicate_attestor_pubkey_or_signature:{norm_pk[:16]}"
            if norm_node in seen_node_ids or (norm_ip and norm_ip in seen_ip_digests):
                continue

            seen_node_ids.add(norm_node)
            seen_pubkeys.add(norm_pk)
            seen_sigs.add(norm_sig)
            if norm_ip:
                seen_ip_digests.add(norm_ip)

            if self.require_pq and (not att.pq_pubkey_hex.strip() or not att.pq_signature_hex.strip()):
                continue

            if att.pq_pubkey_hex:
                norm_pq_pk = str(att.pq_pubkey_hex).strip().lower()
                if norm_pq_pk in seen_pq_pubkeys:
                    return False, f"sybil_duplicate_attestor_pq_pubkey:{norm_pq_pk[:16]}"
                seen_pq_pubkeys.add(norm_pq_pk)

            # Enforce attestation timestamp freshness (reject ts<=0, >24h stale, or >5m in future)
            if isinstance(att.timestamp, bool) or not isinstance(att.timestamp, int):
                continue
            if att.timestamp <= 0 or (now - att.timestamp) > 86400 or (att.timestamp - now) > 300:
                continue

            if isinstance(att.asn, bool) or not isinstance(att.asn, int) or att.asn <= 0:
                continue

            # Verify authenticated attestor ASN binding
            expected_asn = (
                self.trusted_attestor_asns[norm_pk]
                if norm_pk in self.trusted_attestor_asns
                else self._GLOBAL_ATTESTOR_ASN_REGISTRY.get(norm_pk)
            )
            if expected_asn is None or int(att.asn) != int(expected_asn):
                continue

            bound_digest = att.signing_payload(manifest_digest)
            try:
                node_vk = Ed25519PublicKey.from_public_bytes(bytes.fromhex(norm_pk))
                node_vk.verify(bytes.fromhex(norm_sig), bound_digest)
            except (InvalidSignature, ValueError):
                continue

            if self.require_pq or att.pq_signature_hex or att.pq_pubkey_hex:
                try:
                    norm_att_pq = att.pq_pubkey_hex.strip().lower()
                    node_pq_pk = bytes.fromhex(norm_att_pq)
                    node_pq_sig = bytes.fromhex(att.pq_signature_hex.strip())
                    meta_att = WOTSPlusSHAKE256.extract_signature_metadata(node_pq_pk, node_pq_sig)
                    if meta_att is None:
                        continue
                    att_root_pq_hex = meta_att["root_pq_pub_hex"]
                    if not (
                        WOTSPlusSHAKE256.is_bound_to_ed25519_pub(norm_pk, att_root_pq_hex)
                        or WOTSPlusSHAKE256.is_bound_to_ed25519_pub(norm_pk, norm_att_pq)
                    ):
                        continue
                    prev_att_msg = self._used_pq_pubkeys.get(meta_att["one_time_key_id"])
                    if prev_att_msg is not None and prev_att_msg != bound_digest:
                        continue
                    if not WOTSPlusSHAKE256.verify(node_pq_pk, bound_digest, node_pq_sig):
                        continue
                    self._used_pq_pubkeys[meta_att["one_time_key_id"]] = bound_digest
                    self._used_pq_pubkeys[norm_att_pq] = bound_digest
                except ValueError:
                    continue

            verified_attestations_count += 1
            verified_asns.add(int(att.asn))

        if verified_attestations_count < self.min_quorum:
            return False, f"insufficient_verified_attestations:{verified_attestations_count}<{self.min_quorum}"

        if len(verified_asns) < self.min_distinct_asns:
            return False, f"insufficient_distinct_asns:{len(verified_asns)}<{self.min_distinct_asns}"

        return True, "quorum_verified"


# ---------------------------------------------------------------------------
# Main Plugin Interface: ArmigerDHTDatabaseSync
# ---------------------------------------------------------------------------

class ArmigerDHTDatabaseSync:
    """End-to-end decentralized, CSAM-safe, encrypted, modular DHT database sync plugin."""

    CATEGORY_TO_CODE: dict[str, int] = {
        cat: idx for idx, cat in enumerate(ALL_CATEGORIES)
    }

    def __init__(
        self,
        subscription: SubscriptionConfig | None = None,
        master_secret: bytes | None = None,
        trusted_root_pubkeys_hex: Sequence[str] | None = None,
        db_path: str | pathlib.Path = ":memory:",
        tombstone_store: TombstoneStore | None = None,
        *,
        trusted_root_pq_pubkeys_hex: Sequence[str] | None = None,
        opt_in_sharing: bool | None = None,
        ephemeral_identity: bool = True,
        min_k_samples: int = 25,
    ) -> None:
        self.opt_in_sharing: bool = resolve_opt_in_sharing(opt_in_sharing)
        self.subscription = subscription or SubscriptionConfig(
            sync_mode=SyncMode.SUBSCRIBE_AND_SEED if self.opt_in_sharing else SyncMode.PULL_ONLY
        )
        self.master_secret = master_secret or os.urandom(32)
        self.tombstone_store = tombstone_store or TombstoneStore(
            trusted_root_pubkeys_hex,
            trusted_root_pq_pubkeys_hex=trusted_root_pq_pubkeys_hex,
        )
        self.csam_guard = CSAMGuard(self.tombstone_store)
        self.safety_gate = ArmigerSafetyGate()
        self.quorum_verifier = QuorumVerifier(
            trusted_root_pubkeys_hex=trusted_root_pubkeys_hex or [],
            min_quorum=self.subscription.min_quorum,
            min_distinct_asns=self.subscription.min_distinct_asns,
            trusted_root_pq_pubkeys_hex=trusted_root_pq_pubkeys_hex,
        )
        self.db = ShardDatabase(db_path=db_path, enable_xxx=self.subscription.enable_xxx)
        self.shard_sequences: dict[str, int] = self.db.get_shard_sequences()
        self.anonymity_guard = AnonymityGuard(
            ephemeral_identity=ephemeral_identity,
            min_k_samples=min_k_samples,
            catalog_secret=self.master_secret,
        )
        self._outbound_bundles_shared: int = 0
        self._outbound_packets_blocked_opt_out: int = 0
        self._outbound_bytes_shared: int = 0
        self._outbound_seq: int = 0
        self._catalog_sender_seqs: dict[tuple[str, str], int] = {}

    def set_sync_mode(self, mode: SyncMode) -> None:
        self.subscription.sync_mode = mode

    def ingest_candidate_record(
        self,
        infohash: str,
        title: str,
        category: str,
        language: str,
        quality_tier: str,
        epoch: str = DEFAULT_EPOCH,
        size_bytes: int = 0,
        file_count: int = 1,
        seeders: int = 0,
        leechers: int = 0,
        content_sha256: str = "",
        file_list: list[str] | None = None,
        p_adult: float = 0.0,
        p_junk: float = 0.0,
        p_excluded_scope: float = 0.0,
    ) -> tuple[bool, str, TorrentRecord | None]:
        """Validate candidate record through CSAM guard and safety gates, then insert locally."""
        if self.subscription.sync_mode == SyncMode.OFF:
            return False, "sync_mode_off", None

        # 1. CSAM & structural harmful content blocker
        files = file_list or [title]
        res = self.csam_guard.inspect_torrent(
            infohash=infohash,
            title=title,
            file_list=files,
            category=category,
        )
        if not res.is_safe:
            return False, f"csam_guard_rejection:{res.violation_type.value}:{res.reason}", None

        # 2. Armiger Conformal Safety Gate
        gate_ok, gate_reason, gate_violation = self.safety_gate.evaluate(
            p_adult=p_adult,
            p_junk=p_junk,
            p_excluded_scope=p_excluded_scope,
            target_category=category,
        )
        if not gate_ok:
            return False, f"safety_gate_rejection:{gate_reason}", None

        # 3. Construct canonical TorrentRecord
        clean_sha = content_sha256 or hashlib.sha256(title.encode()).hexdigest()
        record = TorrentRecord(
            infohash=infohash.lower().strip(),
            title=res.sanitized_title,
            category=category.lower().strip(),
            language=language.lower().strip(),
            quality_tier=quality_tier.lower().strip(),
            epoch=epoch.lower().strip(),
            size_bytes=size_bytes,
            file_count=file_count,
            seeders=seeders,
            leechers=leechers,
            content_sha256=clean_sha.lower(),
            file_list=files,
            safety_gate_passed=True,
            created_at=int(time.time()),
        )

        # 4. Insert into local partitioned database
        inserted = self.db.insert_record(record)
        if not inserted:
            return False, "record_dropped_by_subscription_policy", None

        return True, "ingested", record

    def build_and_sign_shard(
        self,
        shard_id: str,
        records: Sequence[TorrentRecord],
        root_privkey: bytes,
        seq: int,
        attestations: Sequence[QuorumAttestation] | None = None,
    ) -> tuple[DHTShardManifest, bytes]:
        """Package a list of records into an encrypted micro-shard and generate signed manifest."""
        cat, lang, qual, epoch = SubscriptionConfig.parse_shard_id(shard_id)

        # 1. Serialize records and compute Merkle tree
        compressed_records, merkle_tree = serialize_shard_records(records)
        uncompressed_size = sum(len(r.canonical_bytes()) for r in records)

        # 2. Derive ChaCha20-Poly1305 key & encrypt
        key = ShardCryptoEngine.derive_shard_key(self.master_secret, shard_id, epoch)
        ciphertext_bundle = ShardCryptoEngine.encrypt_shard(
            plaintext=compressed_records,
            key=key,
            shard_id=shard_id,
            epoch=epoch,
            seq=seq,
            merkle_root=merkle_tree.root_hex,
        )

        ciphertext_sha = hashlib.sha256(ciphertext_bundle).hexdigest()
        bep46_infohash = hashlib.sha1(ciphertext_bundle).hexdigest()

        # 3. Create manifest and sign with Ed25519 + WOTS+-SHAKE256 Merkle-authenticated leaf key
        sk = Ed25519PrivateKey.from_private_bytes(root_privkey)
        pk = sk.public_key()
        root_pub_bytes = pk.public_bytes(Encoding.Raw, PublicFormat.Raw)
        leaf_idx, pq_pub, _ = WOTSPlusSHAKE256.reserve_next_leaf(root_privkey)

        manifest_dummy = DHTShardManifest(
            seq=seq,
            shard_id=shard_id,
            epoch=epoch,
            record_count=len(records),
            compressed_size_bytes=len(ciphertext_bundle),
            uncompressed_size_bytes=uncompressed_size,
            merkle_root=merkle_tree.root_hex,
            ciphertext_sha256=ciphertext_sha,
            bep46_infohash=bep46_infohash,
            timestamp=int(time.time()),
            root_pubkey_hex=root_pub_bytes.hex(),
            ed25519_signature_hex="",
            attestations=list(attestations or []),
            root_pq_pubkey_hex=pq_pub.hex(),
        )

        payload_bytes = manifest_dummy.canonical_payload()
        sig = sk.sign(payload_bytes)
        pq_sig = WOTSPlusSHAKE256.sign_reserved_leaf(root_privkey, leaf_idx, payload_bytes)

        manifest = DHTShardManifest(
            seq=seq,
            shard_id=shard_id,
            epoch=epoch,
            record_count=len(records),
            compressed_size_bytes=len(ciphertext_bundle),
            uncompressed_size_bytes=uncompressed_size,
            merkle_root=merkle_tree.root_hex,
            ciphertext_sha256=ciphertext_sha,
            bep46_infohash=bep46_infohash,
            timestamp=manifest_dummy.timestamp,
            root_pubkey_hex=root_pub_bytes.hex(),
            ed25519_signature_hex=sig.hex(),
            attestations=list(attestations or []),
            root_pq_pubkey_hex=pq_pub.hex(),
            pq_signature_hex=pq_sig.hex(),
        )

        return manifest, ciphertext_bundle

    def receive_and_apply_shard(
        self,
        manifest: DHTShardManifest,
        ciphertext_bundle: bytes,
    ) -> tuple[bool, str, int]:
        """Validate, decrypt, Merkle-verify, and ingest an incoming decentralized shard."""
        if self.subscription.sync_mode == SyncMode.OFF:
            return False, "sync_mode_off", 0

        # 1. Check user subscription filter (zero bandwidth / zero storage for unselected shards)
        try:
            shard_cat, shard_lang, shard_qual, shard_epoch = SubscriptionConfig.parse_shard_id(manifest.shard_id)
        except ValueError:
            return False, f"malformed_shard_id:{manifest.shard_id}", 0
        if shard_epoch not in ("*", "any") and manifest.epoch.strip().lower() != shard_epoch:
            return False, f"manifest_epoch_mismatch:{manifest.epoch}!={shard_epoch}", 0
        if not self.subscription.is_subscribed(manifest.shard_id):
            return False, f"unsubscribed_shard_filtered:{manifest.shard_id}", 0

        # 2. Verify monotonic sequence & quorum
        norm_shard_id = manifest.shard_id.strip().lower()
        db_seq = self.db.get_shard_sequences().get(norm_shard_id, -1)
        cur_seq = max(self.shard_sequences.get(manifest.shard_id, -1), self.shard_sequences.get(norm_shard_id, -1), db_seq)
        q_ok, q_msg = self.quorum_verifier.verify_manifest(manifest, current_seq=cur_seq)
        if not q_ok:
            return False, f"quorum_verification_failed:{q_msg}", 0

        # 3. Content-addressed & size verification of ciphertext
        actual_sha = hashlib.sha256(ciphertext_bundle).hexdigest()
        if actual_sha.lower() != manifest.ciphertext_sha256.lower():
            return False, "ciphertext_sha256_mismatch", 0
        if len(ciphertext_bundle) != manifest.compressed_size_bytes:
            return False, f"compressed_size_mismatch:{len(ciphertext_bundle)}!={manifest.compressed_size_bytes}", 0

        # 4. Decrypt shard with AEAD
        key = ShardCryptoEngine.derive_shard_key(self.master_secret, manifest.shard_id, manifest.epoch)
        try:
            plaintext = ShardCryptoEngine.decrypt_shard(
                bundle=ciphertext_bundle,
                key=key,
                shard_id=manifest.shard_id,
                epoch=manifest.epoch,
                seq=manifest.seq,
                merkle_root=manifest.merkle_root,
            )
        except (InvalidTag, ValueError) as exc:
            return False, f"aead_decryption_failed:{exc}", 0

        # 5. Decompress records & verify Merkle root and declared counts/sizes
        try:
            records = deserialize_shard_records(plaintext)
            if len(records) != manifest.record_count:
                return False, f"record_count_mismatch:{len(records)}!={manifest.record_count}", 0
            actual_uncomp = sum(len(r.canonical_bytes()) for r in records)
            if actual_uncomp != manifest.uncompressed_size_bytes:
                return False, f"uncompressed_size_mismatch:{actual_uncomp}!={manifest.uncompressed_size_bytes}", 0
            reconstructed_leaves = [r.leaf_hash() for r in sorted(records, key=lambda r: r.infohash.lower())]
            reconstructed_tree = MerkleTree(reconstructed_leaves)
        except Exception as exc:
            return False, f"shard_deserialization_failed:{exc}", 0

        if reconstructed_tree.root_hex.lower() != manifest.merkle_root.lower():
            return False, "merkle_root_mismatch", 0

        # 6. Pre-ingest scope and safety re-inspection of decrypted records
        accepted_records: list[TorrentRecord] = []
        for rec in records:
            rec_cat = rec.category.strip().lower()
            rec_lang = rec.language.strip().lower()
            rec_qual = rec.quality_tier.strip().lower()
            rec_ep = rec.epoch.strip().lower()
            if (
                (shard_cat not in ("*", "any") and rec_cat != shard_cat)
                or (shard_lang not in ("*", "any") and rec_lang != shard_lang)
                or (shard_qual not in ("*", "any") and rec_qual not in (shard_qual, "any"))
                or (shard_epoch not in ("*", "any") and rec_ep != shard_epoch)
            ):
                return False, f"record_shard_scope_mismatch:{rec.infohash}", 0
            eff_qual = shard_qual if rec_qual == "any" else rec_qual
            if not self.subscription.is_subscribed(
                SubscriptionConfig.format_shard_id(rec_cat, rec_lang, eff_qual, rec_ep)
            ):
                return False, f"record_subscription_mismatch:{rec.infohash}", 0

            # Enforce safety gate passed flag and local conformal safety gate evaluation
            if not rec.safety_gate_passed:
                return False, f"record_failed_safety_gate:{rec.infohash}", 0
            gate_ok, gate_reason, _ = self.safety_gate.evaluate(
                p_adult=1.0 if rec_cat == CATEGORY_ADULT else 0.0,
                p_junk=0.0,
                p_excluded_scope=0.0,
                target_category=rec_cat,
            )
            if not gate_ok:
                return False, f"record_failed_safety_gate:{rec.infohash}:{gate_reason}", 0

            # Check against tombstone store
            if self.tombstone_store.is_tombstoned(rec.infohash):
                continue

            # CSAM inspection
            c_res = self.csam_guard.inspect_torrent(
                infohash=rec.infohash,
                title=rec.title,
                file_list=rec.file_list,
                category=rec.category,
            )
            if not c_res.is_safe:
                continue

            accepted_records.append(rec)

        try:
            ingested_count = self.db.insert_records_batch(
                accepted_records,
                shard_id=manifest.shard_id,
                seq=manifest.seq,
            )
        except (ValueError, OverflowError, sqlite3.Error) as exc:
            return False, f"db_commit_failed:{exc}", 0

        # 7. Update in-memory sequence cache after durable SQLite commit
        self.shard_sequences[manifest.shard_id] = manifest.seq
        self.shard_sequences[norm_shard_id] = manifest.seq
        return True, "shard_applied_successfully", ingested_count

    def get_shard_seq(self, shard_id: str) -> int:
        """Return the highest committed BEP-44 sequence number for `shard_id` (0 if unseen)."""
        norm_id = "/".join(SubscriptionConfig.parse_shard_id(shard_id))
        return max(
            int(self.shard_sequences.get(shard_id, 0)),
            int(self.shard_sequences.get(norm_id, 0)),
        )

    def apply_signed_tombstones(self, manifest: SignedTombstoneList) -> tuple[bool, str, int, int]:
        """Apply a signed tombstone list and purge local records.

        Returns (ok, msg, new_tombstones_count, purged_records_count).
        """
        ok, msg, count = self.tombstone_store.apply_signed_tombstones(manifest)
        if not ok:
            return False, msg, 0, 0

        purged = self.db.purge_blinded_hashes(self.tombstone_store)
        return True, msg, count, purged

    def sharing_status(self) -> dict[str, Any]:
        """Return runtime audit telemetry confirming opt-in status and outbound packet/bundle counters."""
        return {
            "opt_in_sharing_enabled": bool(self.opt_in_sharing),
            "sync_mode": self.subscription.sync_mode.value,
            "outbound_bundles_shared": int(self._outbound_bundles_shared),
            "outbound_packets_blocked_opt_out": int(self._outbound_packets_blocked_opt_out),
            "outbound_bytes_shared": int(self._outbound_bytes_shared),
            "ephemeral_identity_enabled": bool(self.anonymity_guard.ephemeral_identity),
            "ephemeral_pubkey_hex": self.anonymity_guard.current_pubkey_hex() if self.opt_in_sharing else None,
            "ephemeral_rotations": int(self.anonymity_guard.rotation_count),
            "blocked_anonymity_violations": int(self.anonymity_guard.blocked_violations_count),
        }

    def hash_catalog_key(self, normalized_key: str, *, reject_raw_pii: bool = True) -> str:
        """Compute strictly 16-hex-char keyed, domain-separated BLAKE2b-64(key, domain='armiger-catalog-v2' || normalized_key)."""
        return self.anonymity_guard._hash_catalog_key_with_derived_key(
            normalized_key,
            self.anonymity_guard.catalog_secret,
            reject_raw_pii=reject_raw_pii,
        )

    def _outbound_allowed(self) -> tuple[bool, str]:
        """Return whether outbound telemetry/catalog export and DHT publishing are permitted."""
        if not self.opt_in_sharing:
            return False, "opt_in_sharing_disabled"
        if self.subscription.sync_mode not in (SyncMode.SUBSCRIBE_AND_SEED, SyncMode.PUBLISHER):
            return False, f"sync_mode_disallows_outbound:{self.subscription.sync_mode.value}"
        return True, "allowed"

    def export_signed_bundle(
        self,
        records: Sequence[Mapping[str, Any] | Sequence[Any]] | None = None,
        *,
        epoch: int | float | None = None,
        seq: int | None = None,
        n_samples: int = 25,
        salt: bytes = b"armiger-catalog-v2",
        raw_wire_override: bytes | str | Mapping[str, Any] | None = None,
        rotate_key: bool | None = None,
        as_dict_records: bool = True,
    ) -> dict[str, Any] | None:
        """Export an anonymous BEP-44 catalog record bundle with 16-hex BLAKE2b keys `[h, c, r, s, a, v, n, t]`.

        Strictly returns None (0 bytes shared) when `opt_in_sharing=False` or `sync_mode` is `OFF`/`PULL_ONLY`.
        Raises `AnonymityViolationError` if any PII/canary/schema/k-anonymity invariant is violated.
        """
        allowed, _ = self._outbound_allowed()
        if not allowed:
            self._outbound_packets_blocked_opt_out += 1
            return None

        self._outbound_seq += 1
        next_seq = int(seq) if seq is not None else self._outbound_seq

        if raw_wire_override is not None:
            if isinstance(raw_wire_override, (bytes, bytearray)):
                v_bytes = bytes(raw_wire_override)
            elif isinstance(raw_wire_override, str):
                v_bytes = raw_wire_override.encode("utf-8")
            elif isinstance(raw_wire_override, Mapping):
                v_bytes = json.dumps(dict(raw_wire_override), sort_keys=True, separators=(",", ":")).encode("utf-8")
            else:
                raise AnonymityViolationError("invalid_raw_wire_override")
            signed = self.anonymity_guard.sign_wire_payload(
                v_bytes, seq=next_seq, salt=salt, rotate_key=rotate_key
            )
            self._outbound_bundles_shared += 1
            self._outbound_bytes_shared += len(v_bytes)
            return signed

        if isinstance(n_samples, bool) or not isinstance(n_samples, int) or n_samples < self.anonymity_guard.min_k_samples:
            self.anonymity_guard.blocked_violations_count += 1
            raise AnonymityViolationError(f"k_anonymity_violation:n_samples={n_samples}<{self.anonymity_guard.min_k_samples}")

        epoch_bucket = self.anonymity_guard.bucket_timestamp(epoch)
        sanitized_records: list[Any] = []
        for rec in (records or []):
            if isinstance(rec, Mapping):
                extra = set(rec.keys()) - {
                    "h", "key", "normalized_key", "c", "category", "r", "s", "a", "v", "n", "t"
                }
                if extra:
                    self.anonymity_guard.blocked_violations_count += 1
                    raise AnonymityViolationError(f"unexpected_catalog_record_keys:{sorted(extra)}")
                raw_h = rec.get("h")
                raw_key = rec.get("key") or rec.get("normalized_key")
                if raw_h is not None:
                    h_str = str(raw_h)
                    is_pre = True
                elif raw_key is not None:
                    h_str = str(raw_key)
                    is_pre = False
                else:
                    self.anonymity_guard.blocked_violations_count += 1
                    raise AnonymityViolationError("missing_catalog_key_or_hash")
                cat_in = rec.get("c", rec.get("category", 0))
                c_code = self.CATEGORY_TO_CODE.get(str(cat_in).lower(), 0) if isinstance(cat_in, str) else int(cat_in)
                r_code = int(rec.get("r", 0))
                s_code = int(rec.get("s", 0))
                a_code = 1 if rec.get("a", 0) else 0
                v_code = rec.get("v", 1)
                n_cnt = int(rec.get("n", n_samples))
                t_ts = rec.get("t", epoch_bucket)
                if isinstance(t_ts, int) and t_ts % self.anonymity_guard.bucket_seconds != 0:
                    self.anonymity_guard.blocked_violations_count += 1
                    raise AnonymityViolationError(f"record_unbucketed_timestamp:t={t_ts}")
                built = self.anonymity_guard.build_anonymous_catalog_record(
                    key_or_hash=h_str,
                    category_code=c_code,
                    res_bucket=r_code,
                    size_bucket=s_code,
                    adult_flag=a_code,
                    viability_bucket=v_code,
                    sample_count=n_cnt,
                    timestamp=t_ts,
                    is_prehashed=is_pre,
                    as_dict=as_dict_records,
                )
                sanitized_records.append(built)
            elif isinstance(rec, (list, tuple)) and len(rec) == 8:
                h_val, c_val, r_val, s_val, a_val, v_val, n_val, t_val = rec
                if isinstance(t_val, int) and t_val % self.anonymity_guard.bucket_seconds != 0:
                    self.anonymity_guard.blocked_violations_count += 1
                    raise AnonymityViolationError(f"record_unbucketed_timestamp:t={t_val}")
                built = self.anonymity_guard.build_anonymous_catalog_record(
                    key_or_hash=str(h_val),
                    category_code=int(c_val),
                    res_bucket=int(r_val),
                    size_bucket=int(s_val),
                    adult_flag=int(a_val),
                    viability_bucket=v_val,
                    sample_count=int(n_val),
                    timestamp=int(t_val),
                    is_prehashed=True,
                    as_dict=as_dict_records,
                )
                sanitized_records.append(built)
            else:
                self.anonymity_guard.blocked_violations_count += 1
                raise AnonymityViolationError(f"malformed_catalog_record:{rec!r}")

        if not sanitized_records:
            default_h = self.hash_catalog_key("default|2026")
            sanitized_records.append(
                self.anonymity_guard.build_anonymous_catalog_record(
                    key_or_hash=default_h,
                    category_code=0,
                    sample_count=n_samples,
                    timestamp=epoch_bucket,
                    is_prehashed=True,
                    as_dict=as_dict_records,
                )
            )

        wire_obj = {
            "v": 2,
            "epoch": epoch_bucket,
            "seq": next_seq,
            "n_bucket": (int(n_samples) // 25) * 25,
            "records": sanitized_records,
        }
        v_bytes = json.dumps(wire_obj, sort_keys=True, separators=(",", ":")).encode("utf-8")
        signed = self.anonymity_guard.sign_wire_payload(
            v_bytes, seq=next_seq, salt=salt, rotate_key=rotate_key
        )
        self._outbound_bundles_shared += 1
        self._outbound_bytes_shared += len(v_bytes)
        return signed

    def export_signed_delta_bundle(self, *args: Any, **kwargs: Any) -> dict[str, Any] | None:
        """Alias for `export_signed_bundle`."""
        return self.export_signed_bundle(*args, **kwargs)

    def ingest_signed_bundle(
        self,
        signed_bundle: Mapping[str, Any],
    ) -> tuple[bool, str, dict[str, Any] | None]:
        """Verify Ed25519 signature, monotonic sequence high-water mark, and AnonymityGuard schema of an inbound catalog bundle."""
        if self.subscription.sync_mode == SyncMode.OFF:
            return False, "sync_mode_off", None

        try:
            v_raw = signed_bundle["v"]
            v_bytes = v_raw if isinstance(v_raw, (bytes, bytearray)) else str(v_raw).encode("utf-8")
            raw_seq = signed_bundle["seq"]
            if isinstance(raw_seq, bool) or not isinstance(raw_seq, int) or raw_seq < 1:
                raise ValueError(f"invalid_bundle_seq:{raw_seq!r}")
            seq = int(raw_seq)
            salt = signed_bundle.get("salt", b"armiger-catalog-v2")
            if isinstance(salt, str):
                salt = salt.encode("utf-8")
            salt_bytes = bytes(salt)
            pub_raw = signed_bundle.get("k") or signed_bundle.get("pubkey_hex")
            pub_bytes = bytes.fromhex(pub_raw) if isinstance(pub_raw, str) else bytes(pub_raw)
            sig_raw = signed_bundle.get("sig") or signed_bundle.get("sig_hex")
            sig_bytes = bytes.fromhex(sig_raw) if isinstance(sig_raw, str) else bytes(sig_raw)
        except Exception as exc:
            return False, f"malformed_bundle_envelope:{exc}", None

        sender_key = (pub_bytes.hex().lower(), salt_bytes.hex().lower())
        prev_seq = self._catalog_sender_seqs.get(sender_key, 0)
        if seq <= prev_seq:
            return False, f"stale_or_replayed_seq:{seq}<={prev_seq}", None

        try:
            vk = Ed25519PublicKey.from_public_bytes(pub_bytes)
            vk.verify(sig_bytes, _bep44_canonical_sign_buffer(v=bytes(v_bytes), seq=seq, salt=salt_bytes))
        except Exception as exc:
            return False, f"invalid_ed25519_signature:{exc}", None

        try:
            parsed = self.anonymity_guard.inspect_wire_payload(bytes(v_bytes))
        except AnonymityViolationError as exc:
            return False, f"anonymity_firewall_rejected:{exc}", None

        if "seq" in parsed and int(parsed["seq"]) != seq:
            return False, f"envelope_wire_seq_mismatch:{seq}!={parsed.get('seq')}", None

        self._catalog_sender_seqs[sender_key] = seq
        return True, "verified_and_ingested", parsed

    def ingest_signed_delta_bundle(
        self,
        signed_bundle: Mapping[str, Any],
    ) -> tuple[bool, str, dict[str, Any] | None]:
        """Alias for `ingest_signed_bundle`."""
        return self.ingest_signed_bundle(signed_bundle)

    async def publish_to_dht(
        self,
        swarm: Any,
        bundle: Mapping[str, Any] | None = None,
        **kwargs: Any,
    ) -> tuple[bool, str, int]:
        """Publish an anonymous signed catalog bundle to a UDP BEP-44 swarm. Blocked when `opt_in_sharing=False` or `sync_mode` is `OFF`/`PULL_ONLY`."""
        allowed, disallow_reason = self._outbound_allowed()
        if not allowed:
            self._outbound_packets_blocked_opt_out += 1
            return False, disallow_reason, 0

        eff_bundle = dict(bundle) if bundle is not None else self.export_signed_bundle(**kwargs)
        if eff_bundle is None:
            self._outbound_packets_blocked_opt_out += 1
            return False, disallow_reason, 0

        v_bytes = eff_bundle["v"] if isinstance(eff_bundle["v"], bytes) else str(eff_bundle["v"]).encode("utf-8")
        self.anonymity_guard.inspect_wire_payload(v_bytes)

        priv_bytes = eff_bundle.get("privkey_bytes") or self.anonymity_guard.current_privkey_bytes()
        salt_bytes = eff_bundle.get("salt", b"armiger-catalog-v2")
        if isinstance(salt_bytes, str):
            salt_bytes = salt_bytes.encode("utf-8")
        responses = await swarm.put_mutable(
            privkey_bytes=priv_bytes,
            salt=salt_bytes,
            v=v_bytes,
            seq=int(eff_bundle["seq"]),
        )
        ok_count = sum(1 for r in responses if r.get(b"y") == b"r")
        return ok_count > 0, "published_anonymous_bep44", ok_count

    def close(self) -> None:
        """Close local SQLite database connection."""
        if hasattr(self, "db") and self.db is not None:
            self.db.close()

    def __del__(self) -> None:
        try:
            self.close()
        except Exception:
            pass


# ---------------------------------------------------------------------------
# Quorum Helper Factory
# ---------------------------------------------------------------------------

def create_mock_quorum_attestations(
    manifest_digest: bytes,
    distinct_asns: Sequence[int],
    timestamp: int | None = None,
) -> list[QuorumAttestation]:
    """Generate cryptographically valid Hybrid Ed25519 + WOTS+-SHAKE256 QuorumAttestations from distinct ASNs."""
    ts = int(time.time()) if timestamp is None else int(timestamp)
    attestations: list[QuorumAttestation] = []
    for idx, asn in enumerate(distinct_asns):
        priv, pub = generate_ed25519_keypair()
        QuorumVerifier.register_trusted_attestor(pub.hex(), int(asn))
        sk = Ed25519PrivateKey.from_private_bytes(priv)
        node_id = hashlib.sha1(pub + f"asn{asn}:{idx}".encode()).hexdigest()
        ip_digest = hashlib.sha256(f"192.0.2.{(asn + idx) % 250}:{idx}".encode()).hexdigest()
        temp_att = QuorumAttestation(
            node_id=node_id,
            asn=int(asn),
            ip_digest=ip_digest,
            pubkey_hex=pub.hex(),
            signature_hex="",
            timestamp=ts,
        )
        bound_msg = temp_att.signing_payload(manifest_digest)
        sig = sk.sign(bound_msg)
        pq_priv, pq_pub = WOTSPlusSHAKE256.derive_keypair_from_ed25519_priv(priv)
        pq_sig = WOTSPlusSHAKE256.sign(pq_priv, bound_msg)

        attestations.append(QuorumAttestation(
            node_id=node_id,
            asn=int(asn),
            ip_digest=ip_digest,
            pubkey_hex=pub.hex(),
            signature_hex=sig.hex(),
            timestamp=ts,
            pq_pubkey_hex=pq_pub.hex(),
            pq_signature_hex=pq_sig.hex(),
        ))
    return attestations


ArmigerDHTDBSync = ArmigerDHTDatabaseSync