Generalize catalog/rule descriptions and add Intended Use & Legal Compliance section

README.md

@@ -22,7 +22,7 @@ pipeline_tag: other
 
 <p align="center">
   <strong>Decentralized <code>BEP-44</code> / <code>BEP-46</code> database synchronization engine featuring multi-layer CSAM &amp; abuse blocking, <code>Ed25519</code>-signed blinded <code>SHA-256</code> tombstone purging, modular fine-grained category &amp; language subscriptions with isolated <code>xxx</code> containment, <code>X25519</code> + <code>ChaCha20-Poly1305</code> AEAD encryption, and $K$-of-$N$ distinct-ASN Sybil quorum verification.</strong><br>
-  Designed for decentralized indexers (<code>Bitmagnet</code>, and <code>*Arr</code> fleets) powered by <a href="https://huggingface.co/NorviTech/armiger-ultra"><code>NorviTech/armiger-ultra</code></a>.
+  Designed for decentralized indexers (<code>Bitmagnet</code> and self-hosted library fleets) powered by <a href="https://huggingface.co/NorviTech/armiger-ultra"><code>NorviTech/armiger-ultra</code></a>.
 </p>
 
 <p align="center">
@@ -97,7 +97,7 @@ sync_engine = ArmigerDHTDatabaseSync(
 # 2. Ingest candidate record with CSAM & safety checks
 ok, msg, record = sync_engine.ingest_candidate_record(
     infohash="0123456789abcdef0123456789abcdef01234567",
-    title="Steins.Gate.Complete.1080p.BluRay.x265",
+    title="Spring.Complete.1080p.BluRay.x265",
     category="anime",
     language="ja",
     quality_tier="hd_1080p",
@@ -109,7 +109,7 @@ ok, msg, record = sync_engine.ingest_candidate_record(
 print("Ingestion result:", ok, msg)
 
 # 3. Search local database
-results = sync_engine.db.search(query="Steins.Gate", category="anime")
+results = sync_engine.db.search(query="Spring", category="anime")
 print("Found torrents:", len(results))
 
 # 4. Apply Ed25519-signed blinded tombstone purge
@@ -173,13 +173,24 @@ Ran 27 tests in 0.011s - OK
 | :--- | :---: | :---: | :--- |
 | **[`NorviTech/armiger-swarm-survival`](https://huggingface.co/NorviTech/armiger-swarm-survival)** | **Yes** (`swarm_adapter.pt`, `swarm_survival.py`) | `163 KB` | **SwarmSurvivalNet v2 & Honeypot Residual Adapter:** **`96.83%`** Swarm Liveliness AUROC (vs. `53.04%` raw seeders), **`100.0%`** fake-seeder malware trap recall, **`0%`** content-category flip guarantee. |
 | **[`NorviTech/armiger-swarm-priors`](https://huggingface.co/NorviTech/armiger-swarm-priors)** | **Yes** (`swarm_priors.py`) | `29.4 KB` | **Empirical Bayes Swarm & Release-Group Priors:** 4-pathway tracker scrape prior engine, mandatory modality dropout (`p=0.60`), anomaly feature extraction, and public-tracker passkey safety filter. |
-| **[`NorviTech/armiger-zero-cue-catalog`](https://huggingface.co/NorviTech/armiger-zero-cue-catalog)** | **Yes** (`zero_cue_plugin.py`, `validated_zero_cue_catalog.json`) | `36.0 MB` | **Zero-Cue Catalog & Structural Disambiguator:** Offline 929,400-key normalized `TMDB`/`TVDB`/`AniDB`/`Manami` lookup + structural pack/ROM/VN/course rules; lifts Movie-vs-TV accuracy to **`99.46%`** and 11-class accuracy to **`98.48%`** (`98.93%` Hard Edge-Cases). |
+| **[`NorviTech/armiger-zero-cue-catalog`](https://huggingface.co/NorviTech/armiger-zero-cue-catalog)** | **Yes** (`zero_cue_plugin.py`, `validated_zero_cue_catalog.json`) | `36.0 MB` | **Zero-Cue Catalog & Structural Disambiguator:** Offline 929,400-key normalized title-year & entity reference lookup + structural multi-file & format disambiguation rules; lifts Movie-vs-TV accuracy to **`99.46%`** and 11-class accuracy to **`98.48%`** (`98.93%` Hard Edge-Cases). |
 | **[`NorviTech/armiger-audio-guard`](https://huggingface.co/NorviTech/armiger-audio-guard)** | **Yes** (`audio_language_guard.py`) | `4.2 KB` | **Cross-Head Audio Language & Anime Dub Guard:** Couples `title_language` (`P(Foreign Title)`) with `lang` head to disambiguate `.en.` prepositions, `<Foreign>.ENG.Sub`, regional dubs, and `MULTi (FR EN)` tags; reaches **`98.72%`** Gold Standard and **`99.88%`** Hard Edge-Case audio accuracy. |
-| **[`NorviTech/armiger-calibration-dht`](https://huggingface.co/NorviTech/armiger-calibration-dht)** | **Yes** (`calibration.json`, `calibration_and_dht.py`, `armiger_dht_rsi.py`) | `115 KB` | **Conformal Calibration, BEP-44 DHT Updater & Distributed RSI Engine:** Per-head temperature scaling ($T_h$), Mondrian safety gates (`99.95%` junk balanced accuracy), `Ed25519` threshold-signed `BEP-44`/`BEP-46` DHT hot-reload, universal `*Arr`/qBittorrent/Transmission/ffprobe ground-truth RSI, `X25519`+`ChaCha20-Poly1305` AEAD, distinct-ASN quorum, and 500-canary regression gate (`36/36` tests). |
+| **[`NorviTech/armiger-calibration-dht`](https://huggingface.co/NorviTech/armiger-calibration-dht)** | **Yes** (`calibration.json`, `calibration_and_dht.py`, `armiger_dht_rsi.py`) | `115 KB` | **Conformal Calibration, BEP-44 DHT Updater & Distributed RSI Engine:** Per-head temperature scaling ($T_h$), Mondrian safety gates (`99.95%` junk balanced accuracy), `Ed25519` threshold-signed `BEP-44`/`BEP-46` DHT hot-reload, universal media-webhook/qBittorrent/Transmission/ffprobe ground-truth RSI, `X25519`+`ChaCha20-Poly1305` AEAD, distinct-ASN quorum, and 500-canary regression gate (`36/36` tests). |
 | **[`NorviTech/armiger-dht-db-sync`](https://huggingface.co/NorviTech/armiger-dht-db-sync)** | **Yes** (`csam_guard.py`, `armiger_dht_db_sync.py`) | `98 KB` | **Modular, CSAM-Protected, Encrypted BEP-44/BEP-46 DHT Database Sync:** Multi-stage CSAM & abuse pre-ingest blocker, `Ed25519`-signed blinded `SHA-256` tombstone purge engine, 10-category $\times$ 14-language $\times$ 7-quality sharded subscriptions with strict default-off `xxx` isolation, `X25519`+`ChaCha20-Poly1305` AEAD, binary Merkle proofs, and $K$-of-$N$ distinct-ASN Sybil quorum (`27/27` tests). |
 
 ---
 
+## Intended Use & Legal Compliance
+
+`Armiger`, `Armiger Ultra`, and all bundled plugins are open-source machine learning and cryptographic metadata-classification utilities designed strictly for **lawful use cases**, including:
+- Organizing and indexing licensed, public-domain, Creative Commons, and open-source digital media, software distributions (e.g., Linux ISOs, open-source games), and educational archives (e.g., OpenCourseWare, LibriVox, Project Gutenberg).
+- Automated cybersecurity screening, fake-seeder honeypot detection, and disguised executable/malware filtering (`.mkv.exe`, `.scr`, `.vbs`) in peer-to-peer and decentralized storage research.
+- Child safety (CSAM/CSAE) pre-ingest blocking, cryptographic tombstone revocation, and adult/NSFW namespace isolation for self-hosted and decentralized indexers.
+
+**NorviTech does not condone, encourage, or support copyright infringement or the unauthorized distribution of copyrighted works.** Users are solely responsible for ensuring that their use of these models and plugins complies with all applicable local, national, and international copyright and data-protection laws.
+
+---
+
 <p align="center">
   <a href="https://norvitech.com"><img alt="Part of the NorviTech Suite — open-source apps for the Linux desktop and the self-hosted stack" src="https://norvitech.com/assets/banner.svg" width="640"></a>
 </p>

config.json

@@ -62,16 +62,16 @@
   },
   "artifacts": {
     "csam_guard.py": {
-      "size_bytes": 20561,
-      "sha256": "295f2b89c695549d012e00584339c7319a573060eeab0dab9448a15b46d0a468"
+      "size_bytes": 23295,
+      "sha256": "3749e36abf4e121de9eea655a5ad9055d0946a1dc8b9753b656bbadaabfde36b"
     },
     "armiger_dht_db_sync.py": {
-      "size_bytes": 41809,
-      "sha256": "802319de50b14f98730d7db2ea7fc66bb05216929a66fc43d6af91e362108984"
+      "size_bytes": 44070,
+      "sha256": "fd10adfc4ac9e82b093bcbb1f94b7f00c92184b45e0ad18640ed375509ae9415"
     },
     "test_dht_db_sync.py": {
-      "size_bytes": 34296,
-      "sha256": "fa4e5929943a0bc1061b7be7b2fab4d47ad1e086b0c9263efb5f6e805e66fb75"
+      "size_bytes": 59514,
+      "sha256": "8986cf54165e6e70c8260bc0c4304f3bd36fef3b9c98403794d3fe4bf07ed681"
     }
   }
 }

test_dht_db_sync.py

@@ -94,11 +94,11 @@ class TestCSAMGuardAndSafetyGates(unittest.TestCase):
         self.assertEqual(normalize_text(raw3), "teen porn")
 
     def test_private_tracker_passkey_sanitization(self) -> None:
-        raw_title = "Inception.2010.1080p.BluRay[passkey=abc123secretkey]?passkey=xyz789"
+        raw_title = "Big.Buck.Bunny.2008.1080p.BluRay[passkey=abc123secretkey]?passkey=xyz789"
         sanitized = sanitize_title(raw_title)
         self.assertNotIn("passkey=", sanitized)
         self.assertNotIn("abc123secretkey", sanitized)
-        self.assertEqual(sanitized, "Inception.2010.1080p.BluRay")
+        self.assertEqual(sanitized, "Big.Buck.Bunny.2008.1080p.BluRay")
 
     def test_direct_csam_detection(self) -> None:
         forbidden_samples = [
@@ -131,10 +131,10 @@ class TestCSAMGuardAndSafetyGates(unittest.TestCase):
 
     def test_disguised_malware_dropper_in_media_torrents(self) -> None:
         # Dangerous executable disguised as media in a movie torrent
-        malicious_files = ["Inception.2010.1080p.mkv", "codec_update.mp4.exe", "instructions.txt"]
+        malicious_files = ["Big.Buck.Bunny.2008.1080p.mkv", "codec_update.mp4.exe", "instructions.txt"]
         res = self.guard.inspect_torrent(
             infohash="d" * 40,
-            title="Inception.2010.1080p.BluRay.x264",
+            title="Big.Buck.Bunny.2008.1080p.BluRay.x264",
             file_list=malicious_files,
             category="movie",
         )
@@ -253,7 +253,7 @@ class TestModularShardedSubscription(unittest.TestCase):
         # Ingest mainstream record
         rec_main = TorrentRecord(
             infohash="1111222233334444555566667777888899990000",
-            title="Dune.Part.Two.2024.1080p",
+            title="Tears.of.Steel.2012.1080p",
             category="movie",
             language="en",
             quality_tier="hd_1080p",
@@ -263,7 +263,7 @@ class TestModularShardedSubscription(unittest.TestCase):
             seeders=150,
             leechers=10,
             content_sha256="abcd" * 16,
-            file_list=["Dune.Part.Two.2024.1080p.mkv"],
+            file_list=["Tears.of.Steel.2012.1080p.mkv"],
             safety_gate_passed=True,
             created_at=100,
         )
@@ -289,7 +289,7 @@ class TestModularShardedSubscription(unittest.TestCase):
         self.assertFalse(sync_engine.db.insert_record(rec_adult))
 
         # Verify search only sees mainstream
-        results = sync_engine.db.search("Dune")
+        results = sync_engine.db.search("Tears.of.Steel")
         self.assertEqual(len(results), 1)
         self.assertEqual(results[0].category, "movie")
 
@@ -567,7 +567,7 @@ class TestEndToEndDHTSyncEngine(unittest.TestCase):
         for i in range(1, 4):
             ok, msg, rec = self.publisher.ingest_candidate_record(
                 infohash=f"abcdef{i:034x}",
-                title=f"Attack on Titan Final S04E{i:02d} 1080p",
+                title=f"Wing It Open Series S04E{i:02d} 1080p",
                 category="anime",
                 language="ja",
                 quality_tier="hd_1080p",
@@ -604,7 +604,7 @@ class TestEndToEndDHTSyncEngine(unittest.TestCase):
         self.assertEqual(count, 3)
 
         # Verify records exist in subscriber database
-        sub_records = self.subscriber.db.search("Attack on Titan")
+        sub_records = self.subscriber.db.search("Wing It Open Series")
         self.assertEqual(len(sub_records), 3)
 
         # Test tombstone purge on subscriber
@@ -619,7 +619,7 @@ class TestEndToEndDHTSyncEngine(unittest.TestCase):
         self.assertEqual(purged, 1)
 
         # Database should now have only 2 records
-        remaining = self.subscriber.db.search("Attack on Titan")
+        remaining = self.subscriber.db.search("Wing It Open Series")
         self.assertEqual(len(remaining), 2)
         self.assertNotIn(records_to_shard[0].infohash, [r.infohash for r in remaining])
 
@@ -846,5 +846,505 @@ class TestEndToEndDHTSyncEngine(unittest.TestCase):
         self.assertEqual(target, manifest_dummy.bep44_target())
 
 
+class TestPhase2CatalogRedTeamAndBenchmarks(unittest.TestCase):
+    """Test Suite 6 (Phase 2): 10,000-Item Catalog Sharding, 120+ Adversarial Red-Team Traps & Benchmarks."""
+
+    def setUp(self) -> None:
+        self.root_priv, self.root_pub = generate_ed25519_keypair()
+        self.master_secret = os.urandom(32)
+        self.tombstone_store = TombstoneStore(trusted_root_pubkeys_hex=[self.root_pub.hex()])
+        self.guard = CSAMGuard(tombstone_store=self.tombstone_store)
+        self.safety_gate = ArmigerSafetyGate()
+
+    def test_10000_item_catalog_sharding_selective_sync_and_merkle_proofs(self) -> None:
+        """Verify 10,000-item multi-category/multi-language sharding, selective sync, 0% xxx leakage, and 10k Merkle proofs."""
+        categories = list(ALL_CATEGORIES)  # 10 categories (9 mainstream + 'xxx')
+        languages = ["en", "ja", "es", "fr", "de"]  # 5 languages
+        qualities = ["uhd_4k", "hd_1080p"]  # 2 quality tiers
+        epoch = "2026_q4"
+
+        publisher = ArmigerDHTDatabaseSync(
+            subscription=SubscriptionConfig(
+                categories=set(categories),
+                languages=set(languages),
+                quality_tiers=set(qualities),
+                epochs={epoch},
+                enable_xxx=True,
+                sync_mode=SyncMode.PUBLISHER,
+            ),
+            master_secret=self.master_secret,
+            trusted_root_pubkeys_hex=[self.root_pub.hex()],
+        )
+
+        subscriber = ArmigerDHTDatabaseSync(
+            subscription=SubscriptionConfig(
+                categories={"movie", "tv_show", "anime"},
+                languages={"en", "ja"},
+                quality_tiers={"uhd_4k", "hd_1080p"},
+                epochs={epoch},
+                enable_xxx=False,
+                sync_mode=SyncMode.PULL_ONLY,
+                min_quorum=5,
+                min_distinct_asns=5,
+            ),
+            master_secret=self.master_secret,
+            trusted_root_pubkeys_hex=[self.root_pub.hex()],
+        )
+
+        try:
+            # 1. Generate 10,000 catalog items partitioned into 100 shards (100 items per shard)
+            shards_map: dict[str, list[TorrentRecord]] = {}
+            total_generated = 0
+
+            for c_idx, cat in enumerate(categories):
+                for l_idx, lang in enumerate(languages):
+                    for q_idx, qual in enumerate(qualities):
+                        shard_id = SubscriptionConfig.format_shard_id(cat, lang, qual, epoch)
+                        shard_records: list[TorrentRecord] = []
+                        for item_idx in range(100):
+                            global_id = (c_idx * 1000) + (l_idx * 200) + (q_idx * 100) + item_idx + 1
+                            infohash = f"{global_id:040x}"
+                            rec = TorrentRecord(
+                                infohash=infohash,
+                                title=f"Catalog_{cat}_{lang}_{qual}_Item_{global_id:05d}",
+                                category=cat,
+                                language=lang,
+                                quality_tier=qual,
+                                epoch=epoch,
+                                size_bytes=1_000_000 * (item_idx + 1),
+                                file_count=1,
+                                seeders=50 + item_idx,
+                                leechers=5,
+                                content_sha256=f"{global_id:064x}",
+                                file_list=[f"item_{global_id:05d}.mkv"],
+                                safety_gate_passed=True,
+                                created_at=1700000000 + global_id,
+                            )
+                            shard_records.append(rec)
+                            total_generated += 1
+                        shards_map[shard_id] = shard_records
+
+            self.assertEqual(total_generated, 10_000)
+            self.assertEqual(len(shards_map), 100)
+
+            # 2. Build & sign all 100 shards, verify 10,000 Merkle inclusion proofs, and feed to subscriber
+            distinct_asns = [15169, 13335, 16509, 7922, 3356]
+            reusable_node_keys = [generate_ed25519_keypair() for _ in distinct_asns]
+
+            merkle_proofs_verified = 0
+            subscribed_shards_applied = 0
+            unsubscribed_shards_skipped = 0
+            records_ingested = 0
+
+            for shard_id, recs in shards_map.items():
+                # Verify binary Merkle tree inclusion proof for every record in the shard
+                sorted_recs = sorted(recs, key=lambda r: r.infohash.lower())
+                leaves = [r.leaf_hash() for r in sorted_recs]
+                tree = MerkleTree(leaves)
+                for idx, r in enumerate(sorted_recs):
+                    proof = tree.get_proof(idx)
+                    self.assertTrue(MerkleTree.verify_proof(r.leaf_hash(), proof, tree.root_hex))
+                    merkle_proofs_verified += 1
+
+                manifest, bundle = publisher.build_and_sign_shard(
+                    shard_id=shard_id,
+                    records=recs,
+                    root_privkey=self.root_priv,
+                    seq=1,
+                )
+                digest = manifest.canonical_digest()
+                attestations = []
+                for asn, (n_priv, n_pub) in zip(distinct_asns, reusable_node_keys):
+                    sk = Ed25519PrivateKey.from_private_bytes(n_priv)
+                    sig = sk.sign(digest)
+                    attestations.append(
+                        QuorumAttestation(
+                            node_id=hashlib.sha1(n_pub + f"asn{asn}".encode()).hexdigest(),
+                            asn=asn,
+                            ip_digest=hashlib.sha256(f"198.51.100.{asn % 250}".encode()).hexdigest(),
+                            pubkey_hex=n_pub.hex(),
+                            signature_hex=sig.hex(),
+                            timestamp=manifest.timestamp,
+                        )
+                    )
+                manifest_quorum = manifest.with_attestations(attestations)
+
+                ok, msg, count = subscriber.receive_and_apply_shard(manifest_quorum, bundle)
+                cat, lang, qual, _ = SubscriptionConfig.parse_shard_id(shard_id)
+                expected_match = (cat in {"movie", "tv_show", "anime"}) and (lang in {"en", "ja"})
+                if expected_match:
+                    self.assertTrue(ok, f"Expected shard {shard_id} to be applied, got: {msg}")
+                    self.assertEqual(count, 100)
+                    subscribed_shards_applied += 1
+                    records_ingested += count
+                else:
+                    self.assertFalse(ok)
+                    self.assertIn("unsubscribed_shard_filtered", msg)
+                    self.assertEqual(count, 0)
+                    unsubscribed_shards_skipped += 1
+
+            self.assertEqual(merkle_proofs_verified, 10_000)
+            self.assertEqual(subscribed_shards_applied, 12)  # 3 categories * 2 languages * 2 qualities
+            self.assertEqual(unsubscribed_shards_skipped, 88)
+            self.assertEqual(records_ingested, 1_200)
+
+            # 3. Verify SQLite state & 0% 'xxx' leakage
+            self.assertEqual(subscriber.db.count_records(), 1_200)
+            self.assertEqual(subscriber.db.count_records(category="movie"), 400)
+            self.assertEqual(subscriber.db.count_records(category="tv_show"), 400)
+            self.assertEqual(subscriber.db.count_records(category="anime"), 400)
+            self.assertEqual(subscriber.db.count_records(category="xxx"), 0)
+
+            cur = subscriber.db.conn.cursor()
+            cur.execute("SELECT COUNT(*) FROM mainstream_torrents WHERE category = 'xxx'")
+            self.assertEqual(cur.fetchone()[0], 0)
+            cur.execute("SELECT DISTINCT category, language FROM mainstream_torrents")
+            pairs = {(r[0], r[1]) for r in cur.fetchall()}
+            self.assertEqual(
+                pairs,
+                {
+                    ("movie", "en"), ("movie", "ja"),
+                    ("tv_show", "en"), ("tv_show", "ja"),
+                    ("anime", "en"), ("anime", "ja"),
+                },
+            )
+        finally:
+            publisher.close()
+            subscriber.close()
+
+    def test_adversarial_csam_nonconsensual_malware_redteam_100_plus_traps(self) -> None:
+        """Red-team 120+ obfuscated CSAM, non-consensual, extreme violence, malware, conformal, and tombstone traps."""
+        total_traps_verified = 0
+
+        # Surface A: 25 Unicode NFKD, Fullwidth, Mathematical & Cyrillic/Greek Homoglyph Traps
+        homoglyph_traps = [
+            "\uff23\uff33\uff21\uff2d Archive 2026",                         # Fullwidth CSAM
+            "\uff23\uff33\uff21\uff25 Collection",                            # Fullwidth CSAE
+            "\uff30\uff45\uff44\uff4f\uff50\uff48\uff49\uff4c\uff45",    # Fullwidth Pedophile
+            "\U0001d429\U0001d41e\U0001d41d\U0001d428\U0001d429\U0001d421\U0001d422\U0001d425\U0001d41e",  # Math Bold
+            "\U0001d450\U0001d455\U0001d456\U0001d459\U0001d451 \U0001d45d\U0001d45c\U0001d45f\U0001d45b",  # Math Italic child porn
+            "\U0001d52d\U0001d52f\U0001d522\U0001d531\U0001d522\U0001d522\U0001d52b \U0001d530\U0001d522\U0001d535",  # Math Fraktur preteen sex
+            "p\u00e9d\u00f6ph\u00efl\u00eb hidden pack",                      # Diacritic storm
+            "pr\u00ebt\u00e9\u00e8n s\u00eax leak",                           # Diacritics preteen sex
+            "l\u00f6l\u00eft\u00e4 h\u00e4rdc\u00f6r\u00eb",                # Diacritics lolita hardcore
+            "\u0440\u0435d\u043e\u0440hil\u0435 underground",                # Cyrillic р, е, о
+            "\u0441hild \u0440\u043ern video",                                 # Cyrillic с, р, о
+            "l\u043elit\u0430 h\u0430rd\u0441\u043er\u0435",                # Cyrillic о, а, с, е
+            "\u0455h\u043et\u0430\u0441\u043en collection",                  # Cyrillic ѕ, о, а, с
+            "t\u043eddler \u0440\u043ern archive",                             # Cyrillic о, р
+            "kinderg\u0430rten \u0455\u0435\u0445",                           # Cyrillic а, ѕ, е, х
+            "inf\u0430nt \u0435r\u043eti\u0441",                              # Cyrillic а, е, о, с
+            "und\u0435r\u0430ge \u0440\u043ern",                              # Cyrillic е, а, р, о
+            "j\u0430ilb\u0430it pack",                                          # Cyrillic а
+            "in\u0441\u0435\u0455t \u0441hild",                               # Cyrillic с, е, ѕ
+            "r\u0430\u0440\u0435 vid\u0435\u043e",                           # Cyrillic а, р, е, о
+            "r\u0435v\u0435ng\u0435 \u0440\u043ern",                         # Cyrillic е, р, о
+            "\u0455\u0440\u0443\u0441\u0430m u\u0440\u0455kirt",           # Cyrillic ѕ, р, у, с, а
+            "\u0455nuff film uncensored",                                        # Cyrillic ѕ
+            "r\u0435\u0430l b\u0435h\u0435\u0430ding",                       # Cyrillic е, а
+            "\u03c1\u03b5d\u03bf\u03c1hil\u03b5 archive",                    # Greek ρ, ε, ο
+        ]
+        self.assertEqual(len(homoglyph_traps), 25)
+        for sample in homoglyph_traps:
+            res = self.guard.inspect_torrent("a" * 40, sample, category="movie")
+            self.assertFalse(res.is_safe, f"Homoglyph trap bypassed guard: {sample!r}")
+            total_traps_verified += 1
+
+        # Surface B: 25 Zero-Width Joiner, Soft Hyphen & Bidirectional Override Traps
+        zw_bidi_traps = [
+            "c\u200bs\u200ca\u200dm leak",
+            "c\u2060s\u2066a\u2069e dump",
+            "p\u200be\u200cd\u200do\u200ep\u200fh\u202ai\u202el\u202ce",
+            "ch\u200bild p\u200dorn",
+            "cp\u200b v\u200cids",
+            "cp\u2060 l\u2060inks",
+            "cp\ufeff m\ufeffegas",
+            "ch\u00adild er\u00adotic",
+            "under\u200bage p\u200dorn",
+            "pre\u200bteen s\u200dex",
+            "pre\u202eteen n\u202cude",
+            "loli\u200bta hard\u200ccore",
+            "loli\u200dta hen\u200btai",
+            "sho\u200btacon archive",
+            "todd\u200bler p\u200dorn",
+            "kinder\u200bgarten s\u200cex",
+            "in\u200bfant n\u200dude",
+            "elemen\u200btary er\u200cotic",
+            "jail\u200bbait collection",
+            "in\u200bcest mi\u200dnor",
+            "non\u200bconsensual leak",
+            "rape\u200b v\u200dideo",
+            "revenge\u200b p\u200corn",
+            "snuff\u200b f\u200dilm",
+            "real\u200b ex\u202eecution",
+        ]
+        self.assertEqual(len(zw_bidi_traps), 25)
+        for sample in zw_bidi_traps:
+            res = self.guard.inspect_torrent("b" * 40, sample, category="movie")
+            self.assertFalse(res.is_safe, f"Zero-width/Bidi trap bypassed guard: {sample!r}")
+            total_traps_verified += 1
+
+        # Surface C: 25 Heavy Leetspeak, Character Repetition & Delimiter Spacing Traps
+        leet_traps = [
+            "P3d0ph1l14_v1ds",
+            "c_h_1_l_d__p_0_r_n",
+            "pr3333t333n s333x",
+            "l0l1t@ h@rdc0r3",
+            "j@1lb@1t_2026",
+            "r3v3ng3 p0rn",
+            "5nuff f1lm",
+            "b1@ckm@1l s3x",
+            "5pyc@m uP5k1rt",
+            "g@ng r@p3",
+            "f0rc3d s3x",
+            "c.s.a.m.pack",
+            "c-s-a-e-collection",
+            "k1nd3rg@rt3n p0rn",
+            "t0ddl3r s3x",
+            "1nf@nt p0rn",
+            "3l3m3nt@ry 3r0t1c",
+            "53xt0rt10n_leaks",
+            "r3@l b3h3@d1ng",
+            "r3@l 3x3cut10n",
+            "l1v3 5u1c1d3 v1d30",
+            "m@55@cr3 l1v35tr3@m",
+            "r3@l t0rtur3 f1lm",
+            "cru5h f3t15h k1ll",
+            "1nc35t ch1ld",
+        ]
+        self.assertEqual(len(leet_traps), 25)
+        for sample in leet_traps:
+            res = self.guard.inspect_torrent("c" * 40, sample, category="movie")
+            self.assertFalse(res.is_safe, f"Leetspeak trap bypassed guard: {sample!r}")
+            total_traps_verified += 1
+
+        # Surface D: 25 Nested Archive & Double-Extension Disguised Executable Malware Traps
+        malware_file_traps = [
+            ("movie", ["Movie.2026.1080p.mp4.exe"]),
+            ("tv_show", ["Series.S01E01.mkv.scr"]),
+            ("anime", ["Episode_01.avi.bat"]),
+            ("book", ["Complete_Guide.pdf.vbs"]),
+            ("music", ["01_Lossless_Track.flac.ps1"]),
+            ("music", ["Album_Track_02.mp3.cmd"]),
+            ("book", ["Novel_Edition.epub.hta"]),
+            ("comic", ["Issue_001.cbz.jar"]),
+            ("movie", ["Trailer_4K.mov.msi"]),
+            ("comic", ["Cover_Art.jpg.pif"]),
+            ("tv_show", ["Subtitles_EN.srt.wsf"]),
+            ("music", ["Studio_Master.wav.vbe"]),
+            ("music", ["Podcast_Ep1.m4a.jse"]),
+            ("movie", ["WebRip_Stream.webm.com"]),
+            ("tv_show", ["Remux_2160p.mkv.cpl"]),
+            ("anime", ["OVA_Special.avi.wsh"]),
+            ("movie", ["Feature_Film.mp4.psm1"]),
+            ("book", ["Reference_Manual.pdf.application"]),
+            ("music", ["Bonus_Track.flac.gadget"]),
+            ("tv_show", ["Finale_1080p.mkv.msp"]),
+            ("Archive.zip/Nested/Video.mp4.exe", ["Archive.zip/Nested/Video.mp4.exe"]),
+            ("Season1.rar/S01E02.mkv.scr", ["Season1.rar/S01E02.mkv.scr"]),
+            ("Pack.7z/Clip.avi.bat", ["Pack.7z/Clip.avi.bat"]),
+            ("Library.tar.gz/Book.pdf.vbs", ["Library.tar.gz/Book.pdf.vbs"]),
+            ("Disc.iso/Audio/01.flac.ps1", ["Disc.iso/Audio/01.flac.ps1"]),
+        ]
+        self.assertEqual(len(malware_file_traps), 25)
+        for cat_or_path, flist in malware_file_traps:
+            cat = cat_or_path if cat_or_path in ALL_CATEGORIES else None
+            res = self.guard.inspect_torrent(
+                infohash="d" * 40,
+                title="Clean_Release_Title_2026",
+                file_list=flist,
+                category=cat,
+            )
+            self.assertFalse(res.is_safe, f"Malware dropper bypassed guard: {flist}")
+            self.assertEqual(res.violation_type, CSAMViolationType.MALWARE_PAYLOAD)
+            total_traps_verified += 1
+
+        # Surface E: 20 Mondrian Conformal Safety Gate Boundary Traps
+        # E1: Safe just below thresholds
+        ok_safe, _, _ = self.safety_gate.evaluate(p_adult=0.5450, p_junk=0.1009, p_excluded_scope=0.4769, target_category="movie")
+        self.assertTrue(ok_safe)
+
+        conformal_traps = [
+            # 5 junk boundary & high-risk cases (threshold = 0.1010)
+            (0.01, 0.1010, 0.01, "movie", CSAMViolationType.SAFETY_GATE_JUNK),
+            (0.01, 0.1011, 0.01, "tv_show", CSAMViolationType.SAFETY_GATE_JUNK),
+            (0.01, 0.1500, 0.01, "anime", CSAMViolationType.SAFETY_GATE_JUNK),
+            (0.01, 0.5000, 0.01, "software", CSAMViolationType.SAFETY_GATE_JUNK),
+            (0.01, 0.9900, 0.01, "game", CSAMViolationType.SAFETY_GATE_JUNK),
+            # 6 scope boundary & high-risk cases (threshold = 0.4770)
+            (0.01, 0.01, 0.4770, "movie", CSAMViolationType.SAFETY_GATE_OUT_OF_SCOPE),
+            (0.01, 0.01, 0.4771, "tv_show", CSAMViolationType.SAFETY_GATE_OUT_OF_SCOPE),
+            (0.01, 0.01, 0.5500, "music", CSAMViolationType.SAFETY_GATE_OUT_OF_SCOPE),
+            (0.01, 0.01, 0.7000, "book", CSAMViolationType.SAFETY_GATE_OUT_OF_SCOPE),
+            (0.01, 0.01, 0.8500, "comic", CSAMViolationType.SAFETY_GATE_OUT_OF_SCOPE),
+            (0.01, 0.01, 0.9900, "course", CSAMViolationType.SAFETY_GATE_OUT_OF_SCOPE),
+            # 9 unisolated adult boundary cases across all 9 mainstream categories (threshold = 0.5451)
+            (0.5451, 0.01, 0.01, "movie", CSAMViolationType.UNISOLATED_ADULT),
+            (0.5451, 0.01, 0.01, "tv_show", CSAMViolationType.UNISOLATED_ADULT),
+            (0.5451, 0.01, 0.01, "music", CSAMViolationType.UNISOLATED_ADULT),
+            (0.5451, 0.01, 0.01, "game", CSAMViolationType.UNISOLATED_ADULT),
+            (0.5451, 0.01, 0.01, "software", CSAMViolationType.UNISOLATED_ADULT),
+            (0.5451, 0.01, 0.01, "anime", CSAMViolationType.UNISOLATED_ADULT),
+            (0.5451, 0.01, 0.01, "book", CSAMViolationType.UNISOLATED_ADULT),
+            (0.5451, 0.01, 0.01, "comic", CSAMViolationType.UNISOLATED_ADULT),
+            (0.5451, 0.01, 0.01, "course", CSAMViolationType.UNISOLATED_ADULT),
+        ]
+        self.assertEqual(len(conformal_traps), 20)
+        for p_a, p_j, p_s, cat, expected_vtype in conformal_traps:
+            ok, reason, vtype = self.safety_gate.evaluate(p_a, p_j, p_s, cat)
+            self.assertFalse(ok, f"Conformal trap bypassed gate: {(p_a, p_j, p_s, cat)}")
+            self.assertEqual(vtype, expected_vtype)
+            total_traps_verified += 1
+
+        # Verify total adversarial traps >= 120
+        self.assertGreaterEqual(total_traps_verified, 120)
+
+        # Surface F: Live Ed25519-Signed Blinded SHA-256 Tombstone Purge Propagation & Permanent Block
+        engine = ArmigerDHTDatabaseSync(
+            subscription=SubscriptionConfig(
+                categories={"movie"},
+                languages={"en"},
+                quality_tiers={"hd_1080p"},
+                epochs={"2026_q4"},
+                enable_xxx=False,
+                sync_mode=SyncMode.PUBLISHER,
+            ),
+            master_secret=self.master_secret,
+            trusted_root_pubkeys_hex=[self.root_pub.hex()],
+        )
+        try:
+            all_hashes = [f"{i:040x}" for i in range(500, 550)]
+            for idx, h in enumerate(all_hashes):
+                ok, _, _ = engine.ingest_candidate_record(
+                    infohash=h,
+                    title=f"Verified_Mainstream_Film_{idx}",
+                    category="movie",
+                    language="en",
+                    quality_tier="hd_1080p",
+                    epoch="2026_q4",
+                )
+                self.assertTrue(ok)
+            self.assertEqual(engine.db.count_records(), 50)
+
+            revoked_hashes = all_hashes[:15]
+            tomb_manifest = sign_tombstone_list(
+                seq=10,
+                infohashes_or_blinded=revoked_hashes,
+                private_key_bytes=self.root_priv,
+            )
+            ok_t, msg_t, new_t, purged_t = engine.apply_signed_tombstones(tomb_manifest)
+            self.assertTrue(ok_t)
+            self.assertEqual(new_t, 15)
+            self.assertEqual(purged_t, 15)
+            self.assertEqual(engine.db.count_records(), 35)
+
+            # Confirm permanent re-ingest block on all 15 tombstoned infohashes
+            for rh in revoked_hashes:
+                ok_re, msg_re, _ = engine.ingest_candidate_record(
+                    infohash=rh,
+                    title="Innocent_Renamed_Film",
+                    category="movie",
+                    language="en",
+                    quality_tier="hd_1080p",
+                    epoch="2026_q4",
+                )
+                self.assertFalse(ok_re)
+                self.assertIn("tombstoned_hash", msg_re)
+        finally:
+            engine.close()
+
+    def test_throughput_and_latency_benchmarks(self) -> None:
+        """Measure CSAM Guard items/sec, AEAD encryption/decryption MB/s, and SQLite batch ingest speed."""
+        import time
+
+        # 1. CSAM Guard inspection throughput (5,000 items)
+        n_inspect = 5_000
+        sample_titles = [
+            ("Clean.Movie.2026.2160p.WEB-DL.DDP5.1.Atmos.H.265", ["Movie.2026.2160p.mkv", "subs.srt"], True),
+            ("Ch1ld_p0rn_hidden_archive_2026", ["archive.zip"], False),
+            ("Disguised.Release.1080p", ["Release.1080p.mp4.exe"], False),
+            ("Anime.Series.S02E12.1080p.HEVC", ["Anime.S02E12.mkv"], True),
+            ("Secret_Bathroom_Spycam_Upskirt", ["clip.mp4"], False),
+        ]
+        t0 = time.perf_counter()
+        for i in range(n_inspect):
+            title, flist, expected_safe = sample_titles[i % len(sample_titles)]
+            res = self.guard.inspect_torrent(f"{i + 10000:040x}", title, file_list=flist, category="movie")
+            self.assertEqual(res.is_safe, expected_safe)
+        t_inspect = time.perf_counter() - t0
+        inspect_rate = n_inspect / max(t_inspect, 1e-9)
+
+        # 2. AEAD Encryption & Decryption Throughput (4 MB payload x 5 iterations = 20 MB)
+        shard_id = "movie/en/uhd_4k/2026_q4"
+        epoch = "2026_q4"
+        merkle_root = "a" * 64
+        key = ShardCryptoEngine.derive_shard_key(self.master_secret, shard_id, epoch)
+        payload_4mb = os.urandom(4 * 1024 * 1024)
+        iters = 5
+        total_mb = (len(payload_4mb) * iters) / (1024 * 1024)
+
+        t_enc0 = time.perf_counter()
+        bundles = [
+            ShardCryptoEngine.encrypt_shard(payload_4mb, key, shard_id, epoch, seq=i + 1, merkle_root=merkle_root)
+            for i in range(iters)
+        ]
+        t_enc = time.perf_counter() - t_enc0
+        enc_mbps = total_mb / max(t_enc, 1e-9)
+
+        t_dec0 = time.perf_counter()
+        for i, b in enumerate(bundles):
+            pt = ShardCryptoEngine.decrypt_shard(b, key, shard_id, epoch, seq=i + 1, merkle_root=merkle_root)
+            self.assertEqual(len(pt), len(payload_4mb))
+        t_dec = time.perf_counter() - t_dec0
+        dec_mbps = total_mb / max(t_dec, 1e-9)
+
+        # 3. SQLite Batch Ingest Throughput (10,000 records)
+        n_sqlite = 10_000
+        batch_records = [
+            TorrentRecord(
+                infohash=f"{i + 100000:040x}",
+                title=f"Benchmark_Record_{i:05d}",
+                category="movie",
+                language="en",
+                quality_tier="hd_1080p",
+                epoch="2026_q4",
+                size_bytes=2_000_000_000,
+                file_count=2,
+                seeders=100,
+                leechers=10,
+                content_sha256=f"{i + 100000:064x}",
+                file_list=[f"movie_{i:05d}.mkv", f"movie_{i:05d}.srt"],
+                safety_gate_passed=True,
+                created_at=1700000000 + i,
+            )
+            for i in range(n_sqlite)
+        ]
+        engine = ArmigerDHTDatabaseSync(
+            subscription=SubscriptionConfig(enable_xxx=False),
+            master_secret=self.master_secret,
+            trusted_root_pubkeys_hex=[self.root_pub.hex()],
+        )
+        try:
+            t_db0 = time.perf_counter()
+            inserted = engine.db.insert_records_batch(batch_records)
+            t_db = time.perf_counter() - t_db0
+            sqlite_rate = inserted / max(t_db, 1e-9)
+            self.assertEqual(inserted, n_sqlite)
+            self.assertEqual(engine.db.count_records(), n_sqlite)
+        finally:
+            engine.close()
+
+        # Assert conservative floors to catch performance regressions across hosts
+        self.assertGreater(inspect_rate, 5_000.0)
+        self.assertGreater(enc_mbps, 50.0)
+        self.assertGreater(dec_mbps, 50.0)
+        self.assertGreater(sqlite_rate, 20_000.0)
+        print(
+            f"\n[BENCHMARK] CSAMGuard={inspect_rate:,.0f} items/s ({1e6/inspect_rate:.1f} us/item) | "
+            f"AEAD_Enc={enc_mbps:,.1f} MB/s | AEAD_Dec={dec_mbps:,.1f} MB/s | "
+            f"SQLite_Batch={sqlite_rate:,.0f} records/s"
+        )
+
+
 if __name__ == "__main__":
     unittest.main()