docs: add DH-Waypoint, Bitmagnet, and *Arr ecosystem integration guide and examples

README.md

@@ -62,6 +62,18 @@ print(pred["heads"].keys())     # dict_keys(['category', 'adult', 'junk', 'anime
 
 ---
 
+## Ecosystem Integrations (`DH-Waypoint`, `Bitmagnet` & `*Arr` Stack)
+
+Armiger and Armiger Ultra ship with turnkey HTTP services, Docker Compose stacks, and reference adapters in [`examples/`](examples) and [`docs/INTEGRATION_GUIDE.md`](docs/INTEGRATION_GUIDE.md):
+
+- **[`DH-Waypoint` Integration (`examples/dh-waypoint/`)](examples/dh-waypoint)** — Drop-in `processorapi/v1` service (`armiger_waypoint_processor.py`) and plugin manifest (`07-armiger-ultra.plugin.yml`) implementing batch classification (`dh-waypoint.processor.v1`), pre-ingest cohort admission (`dh-waypoint.processors.admission.v1`), and 4-tier `dh-waypoint.processors.feedback.v1` closed-loop telemetry for [`spencernorton/DH-Waypoint`](https://github.com/spencernorton/DH-Waypoint).
+- **[`Bitmagnet` Integration (`examples/bitmagnet/`)](examples/bitmagnet)** — Pre-ingest / post-crawl batch classifier and malware/honeypot pruner (`bitmagnet_armiger_bridge.py`) that upgrades upstream Bitmagnet from `35.29%` rule accuracy to **`95.49%` 11-class neural accuracy** and automatically prunes disguised `.mkv.exe`/`.scr`/`.lnk` malware traps via GraphQL `torrent.delete`.
+- **[`*Arr` Stack & `qBittorrent` Integration (`examples/arr-stack/`)](examples/arr-stack)** — Pre-grab Torznab XML filtering proxy (`armiger_torznab_proxy.py`) between `Prowlarr` / `Bitmagnet` / `DH-Waypoint` and `Sonarr`, `Radarr`, `Lidarr`, `Readarr`, and `Whisparr`, paired with a post-import webhook & `ffprobe` ground-truth receiver (`armiger_arr_webhook_receiver.py`) wired into `SeriesWebhookAdapter`, `MovieWebhookAdapter`, `AudioWebhookAdapter`, `PublicationWebhookAdapter`, `QBittorrentAdapter`, and `PhysicalGroundTruthVerifier`.
+
+See the **[Complete Ecosystem Integration Guide (`docs/INTEGRATION_GUIDE.md`)](docs/INTEGRATION_GUIDE.md)** for architecture diagrams, Docker Compose files, and step-by-step setup instructions.
+
+---
+
 ## 8 Simultaneous Classification Heads (Audited Unmutated Ground-Truth Accuracy)
 
 All metrics below are reproducibly computed using `armiger_inference.py` against the **original, unmutated ground-truth evaluation keys** (`rows.y`), separating **Standalone Neural ONNX (`armiger.onnx`)** from **Fair Blind Hybrid (`+ ZeroCueCatalogPlugin` & `AudioLanguageGuard`)**:

docs/INTEGRATION_GUIDE.md

@@ -0,0 +1,282 @@
+# Armiger & Armiger Ultra Ecosystem Integration Guide
+
+This guide provides step-by-step instructions, turnkey Docker Compose stacks, and reference HTTP adapters for integrating **Armiger** (`8.3 MB INT8 CPU`) and **Armiger Ultra** (`32.3 MB FP32 GPU/CPU` + 6 official plugins) with:
+
+1. **[Part A — Integrating with `DH-Waypoint`](#part-a--integrating-armiger-with-dh-waypoint)** (`https://github.com/spencernorton/DH-Waypoint`):
+   - Native `dh-waypoint.processor.v1` batch classification (`POST /v1/classify`)
+   - Pre-ingest `dh-waypoint.processors.admission.v1` cohort admission gate (`POST /v1/admit`)
+   - Closed-loop 4-tier `dh-waypoint.processors.feedback.v1` outcome telemetry (`POST /v1/feedback`)
+2. **[Part B — Integrating with `Bitmagnet`](#part-b--integrating-armiger-with-bitmagnet)**:
+   - Pre-ingest / post-crawl batch classifier & malware/honeypot pruner (`examples/bitmagnet/bitmagnet_armiger_bridge.py`)
+   - GraphQL `torrentDelete` / content enrichment bridge & Torznab proxy enrichment
+3. **[Part C — Integrating with the `*Arr` Stack (`Prowlarr`, `Sonarr`, `Radarr`, `Lidarr`, `Readarr`, `Whisparr`) & `qBittorrent`](#part-c--integrating-armiger-with-the-arr-stack--qbittorrent)**:
+   - Pre-grab Torznab filtering proxy (`examples/arr-stack/armiger_torznab_proxy.py`) blocking disguised `.exe`/`.scr`/`.lnk` malware, fake-seeder honeypot traps, dead swarms, and unrequested foreign-language audio tracks before grab
+   - Post-import webhook & `ffprobe` physical ground-truth receiver (`examples/arr-stack/armiger_arr_webhook_receiver.py`) wired into `SeriesWebhookAdapter`, `MovieWebhookAdapter`, `AudioWebhookAdapter`, `PublicationWebhookAdapter`, `QBittorrentAdapter`, and `PhysicalGroundTruthVerifier` for local conformal calibration and opt-in anonymous `BEP-44`/`BEP-46` RSI knowledge sharing
+
+---
+
+## Architecture Overview
+
+```mermaid
+flowchart LR
+    subgraph Indexers["DHT Indexers & Crawlers"]
+        DHW["DH-Waypoint Core<br/>(processorapi/v1)"]
+        BM["Bitmagnet<br/>(GraphQL + Torznab)"]
+        PR["Prowlarr<br/>(Indexer Manager)"]
+    end
+
+    subgraph ArmigerSuite["Armiger / Armiger Ultra"]
+        PROC["Armiger Waypoint Processor<br/>(/v1/classify, /v1/admit, /v1/feedback)"]
+        BMBR["Bitmagnet Armiger Bridge<br/>(/v1/bitmagnet/classify)"]
+        TPROXY["Armiger Torznab Proxy<br/>(Port 8097 Admission Gate)"]
+        WHOOK["Armiger Webhook & ffprobe Receiver<br/>(Port 8098 Ground Truth + BEP-44 RSI)"]
+    end
+
+    subgraph ArrStack["*Arr Media Managers & Download Client"]
+        SON["Sonarr / Radarr / Lidarr / Readarr"]
+        QBT["qBittorrent / Transmission"]
+    end
+
+    DHW <-->|"HTTP processorapi/v1"| PROC
+    BM <-->|"Batch Classify + GraphQL Prune"| BMBR
+    DHW -->|"Torznab Feed"| TPROXY
+    BM -->|"Torznab Feed"| TPROXY
+    PR -->|"Torznab Feed"| TPROXY
+    TPROXY -->|"Sanitized & Enriched XML"| SON
+    SON -->|"Grab Approved Release"| QBT
+    SON -->|"On Import Webhook"| WHOOK
+    QBT -->|"On Completion + ffprobe"| WHOOK
+    WHOOK -.->|"Local Calibration & Opt-In BEP-44 RSI"| PROC
+```
+
+---
+
+## Part A — Integrating Armiger with `DH-Waypoint`
+
+[`DH-Waypoint`](https://github.com/spencernorton/DH-Waypoint) includes a built-in external processor subsystem (`processorapi/v1`) that connects directly to [`examples/dh-waypoint/armiger_waypoint_processor.py`](../examples/dh-waypoint/armiger_waypoint_processor.py).
+
+### 1. Turnkey Docker Compose Stack
+
+Launch `DH-Waypoint`, `PostgreSQL`, and `Armiger Ultra` together using [`examples/dh-waypoint/docker-compose.yml`](../examples/dh-waypoint/docker-compose.yml):
+
+```bash
+export POSTGRES_PASSWORD="<postgres-password>"
+docker compose -f examples/dh-waypoint/docker-compose.yml up -d
+```
+
+### 2. Drop-In Plugin Manifest (`07-armiger-ultra.plugin.yml`)
+
+Mount [`examples/dh-waypoint/07-armiger-ultra.plugin.yml`](../examples/dh-waypoint/07-armiger-ultra.plugin.yml) into `DH_WAYPOINT_PROCESSORS_PLUGIN_DIR` (`/etc/dh-waypoint/plugins.d/07-armiger-ultra.plugin.yml`):
+
+```yaml
+processor:
+  name: armiger-ultra
+  endpoint: http://armiger-ultra:8095/v1/classify
+  timeout: 5s
+  max_batch: 64
+  required: false
+  forward_files: true
+  max_files: 64
+  forward_trackers: true
+  forward_swarm: true
+
+admission:
+  enabled: true
+  endpoint: http://armiger-ultra:8095/v1/admit
+  timeout: 3s
+  max_batch: 64
+  max_files: 64
+  fail_open: true
+
+learning:
+  enabled: true
+  feedback_endpoint: http://armiger-ultra:8095/v1/feedback
+  feedback_timeout: 5s
+  feedback_batch_size: 64
+  flush_interval: 15s
+  queue_size: 2048
+```
+
+### 3. Equivalent `DH_WAYPOINT_*` Environment Variables
+
+If you configure `DH-Waypoint` via environment variables instead of `plugins.d`:
+
+| Variable | Recommended Value | Purpose |
+| :--- | :--- | :--- |
+| `DH_WAYPOINT_PROCESSORS_HTTP_CHAIN` | `armiger-ultra=http://armiger-ultra:8095/v1/classify` | Ordered HTTP classification chain (`dh-waypoint.processor.v1`). |
+| `DH_WAYPOINT_PROCESSORS_HTTP_FORWARD_FILES` | `true` | Forwards physical file paths and byte sizes to Armiger's File-Tree Encoder. |
+| `DH_WAYPOINT_PROCESSORS_HTTP_FORWARD_TRACKERS` | `true` | Forwards tracker domain hints to `SwarmPriors`. |
+| `DH_WAYPOINT_PROCESSORS_HTTP_FORWARD_SWARM` | `true` | Forwards seeder/leecher counts to `SwarmSurvivalNet`. |
+| `DH_WAYPOINT_PROCESSORS_ADMISSION_ENABLED` | `true` | Enables pre-ingest cohort admission (`dh-waypoint.processors.admission.v1`). |
+| `DH_WAYPOINT_PROCESSORS_ADMISSION_ENDPOINT` | `http://armiger-ultra:8095/v1/admit` | Pre-ingest admission endpoint. |
+| `DH_WAYPOINT_LEARNING_ENABLED` | `true` | Enables the 4-tier `dh-waypoint.processors.feedback.v1` outcome loop. |
+| `DH_WAYPOINT_LEARNING_FEEDBACK_ENDPOINT` | `http://armiger-ultra:8095/v1/feedback` | Delivers `t1_discovery` through `t4_playback` feedback batches to Armiger. |
+
+---
+
+## Part B — Integrating Armiger with `Bitmagnet`
+
+Upstream `Bitmagnet` uses deterministic keyword/extension rules (`35.29%` 11-class accuracy on `N=10,262` audited torrents). Deploying [`examples/bitmagnet/bitmagnet_armiger_bridge.py`](../examples/bitmagnet/bitmagnet_armiger_bridge.py) upgrades a `Bitmagnet` deployment to **`95.49%` 11-class accuracy** (`97.55%` 6-dimension composite accuracy) and **`100%` fake-seeder malware trap recall**.
+
+### 1. Turnkey `Bitmagnet` + `Armiger` Docker Compose Stack
+
+Start the stack in [`examples/bitmagnet/docker-compose.yml`](../examples/bitmagnet/docker-compose.yml):
+
+```bash
+export POSTGRES_PASSWORD="<postgres-password>"
+docker compose -f examples/bitmagnet/docker-compose.yml up -d
+```
+
+This launches two companion services alongside `bitmagnet`:
+- **`armiger-bitmagnet-bridge` (`port 8096`)**: Evaluates batches of newly crawled Bitmagnet torrents (`POST /v1/bitmagnet/classify`), maps Armiger's 11-class predictions (`movie`, `tv_show`, `music`, `ebook`, `audiobook`, `comic`, `software`, `game`, `xxx`) and tags (`armiger:<category>`, `armiger:lang:<lang>`, `armiger:viability:<tier>`) onto Bitmagnet items, and automatically executes Bitmagnet's GraphQL `torrent.delete` mutation against disguised malware payloads (`.mkv.exe`, `.scr`, `.lnk`, `.vbs`) and fake-seeder honeypots.
+- **`armiger-torznab-proxy` (`port 8097`)**: Wraps `http://bitmagnet:3333/torznab` with real-time neural admission filtering before `Prowlarr`, `Sonarr`, or `Radarr` sees search results.
+
+### 2. Calling the Bitmagnet Classifier Bridge Directly
+
+```bash
+curl -sS -X POST http://localhost:8096/v1/bitmagnet/classify \
+  -H "Content-Type: application/json" \
+  -d '{
+    "torrents": [
+      {
+        "infoHash": "1111222233334444555566667777888899990000",
+        "name": "Sintel.2010.2160p.UHD.BluRay.REMUX.HEVC.TrueHD.7.1.Atmos-OPEN",
+        "size": 74500000000,
+        "seeders": 320,
+        "leechers": 18,
+        "files": [{"path": "Sintel.2010.2160p.mkv", "size": 74500000000}]
+      },
+      {
+        "infoHash": "aaaa22223333444455556666777788889999bbbb",
+        "name": "Big.Buck.Bunny.2008.1080p.BluRay.mkv.exe",
+        "size": 15400000,
+        "seeders": 4500,
+        "leechers": 12,
+        "files": [{"path": "Big.Buck.Bunny.2008.1080p.BluRay.mkv.exe", "size": 15400000}]
+      }
+    ]
+  }'
+```
+
+The bridge classifies the first item as `contentType: "movie"` (`viabilityTier: "alive_viable"`) and flags the second item (`Big.Buck.Bunny.2008.1080p.BluRay.mkv.exe`) for immediate deletion via `graphqlDeleteMutation`.
+
+---
+
+## Part C — Integrating Armiger with the `*Arr` Stack & `qBittorrent`
+
+Armiger integrates with `Prowlarr`, `Sonarr`, `Radarr`, `Lidarr`, `Readarr`, `Whisparr`, and `qBittorrent` at **two stages**:
+1. **Stage 1 — Pre-Grab Torznab Filtering Proxy (`port 8097`):** Blocks malware masquerades, fake-seeder honeypots, dead swarms, and unrequested foreign-language audio releases *before* any `*Arr` app grabs them.
+2. **Stage 2 — Post-Import Webhook & `ffprobe` Ground-Truth Receiver (`port 8098`):** Verifies completed downloads and `ffprobe` media streams to calibrate local Mondrian thresholds and optionally contribute differential-privacy blinded telemetry to the `BEP-44`/`BEP-46` RSI network.
+
+### 1. Stage 1: Wiring the Torznab Filtering Proxy (`Prowlarr` / `Sonarr` / `Radarr` / `Lidarr` / `Readarr`)
+
+Start [`examples/arr-stack/docker-compose.yml`](../examples/arr-stack/docker-compose.yml):
+
+```bash
+export UPSTREAM_TORZNAB_URL="http://dh-waypoint:3333/torznab"
+docker compose -f examples/arr-stack/docker-compose.yml up -d
+```
+
+#### Configuring in `Prowlarr`, `Sonarr`, `Radarr`, `Lidarr`, `Readarr`, or `Whisparr`:
+1. Open **Settings -> Indexers -> + -> Torznab (Custom)**.
+2. Set **URL** to `http://armiger-torznab-proxy:8097` (instead of pointing directly at `http://bitmagnet:3333/torznab` or `http://dh-waypoint:3333/torznab`).
+3. Every Torznab search query (`?t=search`, `?t=tvsearch`, `?t=movie`, `?t=music`, `?t=book`) is transparently forwarded upstream, batch-scored by `ArmigerUltraPipeline`, stripped of unsafe/honeypot/dead/wrong-language items, and annotated with custom `<torznab:attr>` tags:
+   - `armiger_category` (`movie`, `tv_show`, `music`, `audiobook`, `ebook`, `comic`, `software`, `game`, `xxx`)
+   - `armiger_lang` (`english`, `non_english`, `multi_or_unknown`)
+   - `armiger_viability` (`alive_strong`, `alive_viable`, `marginal`, `dead`)
+   - `armiger_survival_prob` (`0.0000`–`1.0000`)
+
+#### Torznab Proxy Environment Controls
+
+| Variable | Default | Description |
+| :--- | :---: | :--- |
+| `UPSTREAM_TORZNAB_URL` | `http://dh-waypoint:3333/torznab` | Upstream Torznab base URL (`DH-Waypoint`, `Bitmagnet`, or `Prowlarr`). |
+| `ARMIGER_VARIANT` | `int8` | `int8` (`8.3 MB` CPU fast-path, `2.44 ms` p50) or `fp32` (`32.3 MB` GPU/CPU). |
+| `ARMIGER_REJECT_DEAD_SWARMS` | `true` | Drops Torznab items whose `SwarmSurvivalNet` viability tier is `dead`. |
+| `ARMIGER_REQUIRE_ENGLISH_AUDIO` | `false` | When `true`, blocks releases classified as `non_english` audio by `AudioLanguageGuard`. |
+| `ARMIGER_ALLOW_ENGLISH_SUBS` | `true` | When `ARMIGER_REQUIRE_ENGLISH_AUDIO=true`, permits foreign-audio anime/films that carry verified English subtitles (`has_eng_subs`). |
+
+### 2. Stage 2: Wiring Post-Import Webhooks (`Sonarr`, `Radarr`, `Lidarr`, `Readarr`, `qBittorrent`)
+
+`armiger-arr-webhook-receiver` (`port 8098`) exposes dedicated endpoints backed by the adapters in [`armiger_dht_rsi.py`](../armiger_dht_rsi.py):
+
+| Application | Where to Configure in UI | Webhook URL | Underlying Armiger Adapter |
+| :--- | :--- | :--- | :--- |
+| **Sonarr** | `Settings -> Connect -> + -> Webhook` (`On Import`, `On Upgrade`) | `http://armiger-arr-webhook-receiver:8098/webhooks/sonarr` | `SeriesWebhookAdapter` |
+| **Radarr** | `Settings -> Connect -> + -> Webhook` (`On Import`, `On Upgrade`) | `http://armiger-arr-webhook-receiver:8098/webhooks/radarr` | `MovieWebhookAdapter` |
+| **Lidarr** | `Settings -> Connect -> + -> Webhook` (`On Release Import`, `On Upgrade`) | `http://armiger-arr-webhook-receiver:8098/webhooks/lidarr` | `AudioWebhookAdapter` |
+| **Readarr** | `Settings -> Connect -> + -> Webhook` (`On Release Import`, `On Upgrade`) | `http://armiger-arr-webhook-receiver:8098/webhooks/readarr` | `PublicationWebhookAdapter` |
+| **qBittorrent** | `Options -> Downloads -> Run external program` | `POST http://armiger-arr-webhook-receiver:8098/webhooks/qbittorrent` | `QBittorrentAdapter` |
+
+#### `qBittorrent` Completion Command
+
+In `qBittorrent` (`Tools -> Options -> Downloads -> Run external program on torrent completion`), configure:
+
+```bash
+curl -sS -X POST http://armiger-arr-webhook-receiver:8098/webhooks/qbittorrent -H "Content-Type: application/json" -d "{\"name\":\"%N\",\"hash\":\"%I\",\"content_path\":\"%F\",\"size\":%Z,\"progress\":1.0,\"state\":\"uploading\"}"
+```
+
+#### Programmatic Python Usage (`SeriesWebhookAdapter`, `MovieWebhookAdapter`, `QBittorrentAdapter`, `PhysicalGroundTruthVerifier`)
+
+You can also invoke the webhook adapters and `ffprobe` verifier directly in Python via `ArmigerUltraPipeline.ingest_rsi_feedback`:
+
+```python
+from armiger_ultra_pipeline import ArmigerUltraPipeline
+
+pipeline = ArmigerUltraPipeline.from_directory(".", variant="int8")
+
+# 1. Ingest a native Sonarr "Download" (Import) webhook + ffprobe output
+ok, reason, record = pipeline.ingest_rsi_feedback(
+    {
+        "eventType": "Download",
+        "series": {"title": "Cosmos Laundromat", "tvdbId": 310123},
+        "episodes": [{"seasonNumber": 1, "episodeNumber": 1}],
+        "release": {
+            "releaseTitle": "Cosmos.Laundromat.S01E01.1080p.WEB-DL.DDP5.1.H.264-OPEN",
+            "size": 1_450_000_000,
+        },
+        "episodeFile": {
+            "relativePath": "Season 01/Cosmos.Laundromat.S01E01.mkv",
+            "size": 1_450_000_000,
+            "mediaInfo": {
+                "videoCodec": "h264",
+                "audioFormat": "eac3",
+                "audioChannels": 5.1,
+                "audioLanguages": "eng",
+                "subtitles": "eng",
+            },
+        },
+    },
+    adapter="series_webhook",
+)
+assert ok and reason == "accepted_and_queued_with_poisson_jitter"
+print(record.blinded_stem.blinded_hash, record.category, record.lang)
+
+# 2. Ingest a native Radarr "Download" (Import) webhook
+ok, reason, record = pipeline.ingest_rsi_feedback(
+    {
+        "eventType": "Download",
+        "movie": {"title": "Sintel", "year": 2010, "tmdbId": 45745},
+        "release": {
+            "releaseTitle": "Sintel.2010.2160p.UHD.BluRay.REMUX.HEVC.TrueHD.7.1.Atmos-OPEN",
+            "size": 24_800_000_000,
+        },
+        "movieFile": {
+            "relativePath": "Sintel (2010)/Sintel.2010.2160p.mkv",
+            "size": 24_800_000_000,
+        },
+    },
+    adapter="movie_webhook",
+)
+assert ok and record.category == "movie"
+```
+
+### 3. `BEP-44` / `BEP-46` Privacy & RSI Participation Modes
+
+All adapters enforce `PhysicalGroundTruthVerifier` (never trusting unverified model predictions) and `TelemetryAnonymizer` (stripping tracker passkeys, local paths, and IP addresses; blinding title stems via post-quantum `HMAC-SHA256` + `SHAKE-256`; applying Laplace differential privacy $\epsilon = 1.0$; and scheduling outbound packets with Poisson timing jitter):
+
+| `ARMIGER_DHT_RSI_MODE` | Behavior |
+| :--- | :--- |
+| `off` | Disables all `BEP-44`/`BEP-46` DHT traffic; calibration and ground-truth verification stay strictly local. |
+| `pull_only` *(Default)* | Verifies and applies `M`-of-`N` hybrid post-quantum (`Ed25519` + `WOTS+-SHAKE256`) signed calibration manifests from the DHT after passing the 500-item golden canary gate; never transmits telemetry. |
+| `anonymous_contribute` | Opts in to transmitting differential-privacy blinded, hybrid post-quantum (`X25519` + `ML-KEM-768` + `ChaCha20-Poly1305`) encrypted ground-truth observations. |

examples/README.md

@@ -0,0 +1,11 @@
+# Armiger & Armiger Ultra Integration Examples
+
+Turnkey reference adapters, HTTP services, and Docker Compose stacks for integrating **Armiger** (`8.3 MB INT8`) and **Armiger Ultra** (`32.3 MB FP32` + 6 plugins) across the self-hosted indexer and media automation ecosystem:
+
+| Stack | Directory | Description |
+| :--- | :--- | :--- |
+| **[DH-Waypoint](https://github.com/spencernorton/DH-Waypoint)** | [`examples/dh-waypoint/`](dh-waypoint) | Native `processorapi/v1` HTTP processor (`dh-waypoint.processor.v1`), pre-ingest cohort admission gate (`dh-waypoint.processors.admission.v1`), and 4-tier `dh-waypoint.processors.feedback.v1` closed-loop RSI telemetry bridge. |
+| **Bitmagnet** | [`examples/bitmagnet/`](bitmagnet) | Fast pre-ingest / post-crawl batch classifier, malware/honeypot pruner, PostgreSQL/GraphQL enrichment bridge (`bitmagnet_armiger_bridge.py`), and Torznab proxy stack. |
+| **\*Arr Stack & qBittorrent** | [`examples/arr-stack/`](arr-stack) | Pre-grab Torznab filtering proxy (`armiger_torznab_proxy.py`) between `Prowlarr` / `Bitmagnet` and `Sonarr`, `Radarr`, `Lidarr`, `Readarr`, and `Whisparr`, plus a post-import webhook & `ffprobe` ground-truth receiver (`armiger_arr_webhook_receiver.py`). |
+
+See [`docs/INTEGRATION_GUIDE.md`](../docs/INTEGRATION_GUIDE.md) for the complete architecture and step-by-step setup guide.

examples/arr-stack/armiger_arr_webhook_receiver.py

@@ -0,0 +1,197 @@
+"""Armiger Post-Import Webhook & ffprobe Ground-Truth Receiver for Sonarr, Radarr, Lidarr, Readarr & qBittorrent.
+
+Wires completion and import events from the *Arr media automation stack and qBittorrent directly into
+Armiger's `SeriesWebhookAdapter`, `MovieWebhookAdapter`, `AudioWebhookAdapter`, `PublicationWebhookAdapter`,
+`QBittorrentAdapter`, and `PhysicalGroundTruthVerifier`:
+  - `POST /webhooks/sonarr`       -> `SeriesWebhookAdapter` (`adapter="series_webhook"`)
+  - `POST /webhooks/radarr`       -> `MovieWebhookAdapter` (`adapter="movie_webhook"`)
+  - `POST /webhooks/lidarr`       -> `AudioWebhookAdapter` (`adapter="audio_webhook"`)
+  - `POST /webhooks/readarr`      -> `PublicationWebhookAdapter` (`adapter="publication_webhook"`)
+  - `POST /webhooks/qbittorrent`  -> `QBittorrentAdapter` (`adapter="qbittorrent"`)
+
+When `ARMIGER_RUN_FFPROBE=true` and the imported media path is accessible on disk, the receiver runs
+`ffprobe -v quiet -print_format json -show_format -show_streams <path>` to physically verify video/audio
+codecs, channel counts, audio languages, and subtitle tracks before passing the observation to
+`ArmigerUltraPipeline.ingest_rsi_feedback(...)`.
+"""
+from __future__ import annotations
+
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+import json
+import os
+from pathlib import Path
+import subprocess
+import sys
+from typing import Any, Callable, Dict, Mapping, Optional
+
+ROOT_DIR = Path(__file__).resolve().parents[2]
+if str(ROOT_DIR) not in sys.path:
+    sys.path.insert(0, str(ROOT_DIR))
+
+from armiger_dht_rsi import DHTRSIMode
+from armiger_ultra_pipeline import ArmigerUltraPipeline
+
+ROUTE_TO_ADAPTER: Dict[str, str] = {
+    "/webhooks/sonarr": "series_webhook",
+    "/webhooks/radarr": "movie_webhook",
+    "/webhooks/lidarr": "audio_webhook",
+    "/webhooks/readarr": "publication_webhook",
+    "/webhooks/qbittorrent": "qbittorrent",
+}
+
+
+def run_ffprobe_json(media_path: str, timeout_sec: float = 10.0) -> Optional[Dict[str, Any]]:
+    """Run ffprobe on a local media file and return parsed JSON stream/format metadata."""
+    p = Path(media_path)
+    if not p.is_file():
+        return None
+    try:
+        proc = subprocess.run(
+            [
+                "ffprobe",
+                "-v",
+                "quiet",
+                "-print_format",
+                "json",
+                "-show_format",
+                "-show_streams",
+                str(p),
+            ],
+            capture_output=True,
+            text=True,
+            timeout=timeout_sec,
+            check=True,
+        )
+        return json.loads(proc.stdout)
+    except Exception:
+        return None
+
+
+class ArmigerArrWebhookService:
+    """Processes *Arr and qBittorrent completion webhooks into Armiger RSI ground-truth observations."""
+
+    def __init__(
+        self,
+        pipeline: ArmigerUltraPipeline,
+        *,
+        enable_ffprobe: bool = True,
+        ffprobe_runner: Callable[[str], Optional[Dict[str, Any]]] = run_ffprobe_json,
+    ) -> None:
+        self.pipeline = pipeline
+        self.enable_ffprobe = enable_ffprobe
+        self.ffprobe_runner = ffprobe_runner
+
+    @staticmethod
+    def _extract_media_path(payload: Mapping[str, Any]) -> Optional[str]:
+        for section_key in ("movieFile", "episodeFile"):
+            sec = payload.get(section_key)
+            if isinstance(sec, Mapping):
+                full_path = sec.get("path") or sec.get("relativePath")
+                if full_path:
+                    return str(full_path)
+        for list_key in ("trackFiles", "bookFiles"):
+            items = payload.get(list_key)
+            if isinstance(items, list) and items and isinstance(items[0], Mapping):
+                full_path = items[0].get("path")
+                if full_path:
+                    return str(full_path)
+        if payload.get("content_path") or payload.get("save_path"):
+            return str(payload.get("content_path") or payload.get("save_path"))
+        return None
+
+    def process_webhook(
+        self,
+        route_path: str,
+        payload: Mapping[str, Any],
+        *,
+        explicit_ffprobe: Optional[Dict[str, Any]] = None,
+    ) -> Dict[str, Any]:
+        """Process a webhook payload from Sonarr, Radarr, Lidarr, Readarr, or qBittorrent."""
+        adapter_name = ROUTE_TO_ADAPTER.get(route_path)
+        if not adapter_name:
+            return {"accepted": False, "reason": f"unknown_route:{route_path}"}
+
+        event_type = str(payload.get("eventType") or "")
+        if event_type.lower() == "test":
+            return {
+                "accepted": True,
+                "verified": False,
+                "adapter": adapter_name,
+                "reason": "test_webhook_acknowledged",
+            }
+
+        ffprobe_data = explicit_ffprobe
+        if ffprobe_data is None and self.enable_ffprobe:
+            media_path = self._extract_media_path(payload)
+            if media_path:
+                ffprobe_data = self.ffprobe_runner(media_path)
+
+        ok, reason, record = self.pipeline.ingest_rsi_feedback(
+            payload,
+            adapter=adapter_name,
+            ffprobe_output=ffprobe_data,
+        )
+        return {
+            "accepted": True,
+            "verified": bool(ok),
+            "adapter": adapter_name,
+            "reason": reason,
+            "blinded_hash": record.blinded_stem.blinded_hash if record is not None else None,
+            "category": record.category if record is not None else None,
+            "lang": record.lang if record is not None else None,
+            "junk": record.junk if record is not None else None,
+        }
+
+
+def create_webhook_server(host: str = "localhost", port: int = 8098) -> ThreadingHTTPServer:
+    model_dir = os.environ.get("ARMIGER_MODEL_DIR", str(ROOT_DIR))
+    variant = os.environ.get("ARMIGER_VARIANT", "int8")
+    rsi_mode_str = os.environ.get("ARMIGER_DHT_RSI_MODE", "pull_only")
+    pipeline = ArmigerUltraPipeline.from_directory(
+        model_dir,
+        variant=variant,
+        rsi_mode=DHTRSIMode(rsi_mode_str),
+        preload_catalog=False,
+    )
+    service = ArmigerArrWebhookService(
+        pipeline,
+        enable_ffprobe=os.environ.get("ARMIGER_RUN_FFPROBE", "true").lower() == "true",
+    )
+
+    class WebhookHandler(BaseHTTPRequestHandler):
+        def _send_json(self, status: int, data: Dict[str, Any]) -> None:
+            body = json.dumps(data).encode("utf-8")
+            self.send_response(status)
+            self.send_header("Content-Type", "application/json")
+            self.send_header("Content-Length", str(len(body)))
+            self.end_headers()
+            self.wfile.write(body)
+
+        def do_GET(self) -> None:
+            if self.path == "/healthz":
+                self._send_json(200, {"status": "ok", "service": "armiger-arr-webhook-receiver", "rsi_mode": rsi_mode_str})
+            else:
+                self._send_json(404, {"error": "not_found"})
+
+        def do_POST(self) -> None:
+            length = int(self.headers.get("Content-Length", "0"))
+            raw = self.rfile.read(length) if length > 0 else b"{}"
+            payload = json.loads(raw.decode("utf-8"))
+            if self.path in ROUTE_TO_ADAPTER:
+                res = service.process_webhook(self.path, payload)
+                self._send_json(200, res)
+            else:
+                self._send_json(404, {"error": "not_found"})
+
+        def log_message(self, fmt: str, *args: Any) -> None:
+            return
+
+    return ThreadingHTTPServer((host, port), WebhookHandler)
+
+
+if __name__ == "__main__":
+    listen_host = os.environ.get("ARMIGER_WEBHOOK_HOST", "localhost")
+    listen_port = int(os.environ.get("ARMIGER_WEBHOOK_PORT", "8098"))
+    srv = create_webhook_server(listen_host, listen_port)
+    print(f"Armiger *Arr & qBittorrent webhook receiver listening on {listen_host}:{listen_port}")
+    srv.serve_forever()

examples/arr-stack/armiger_torznab_proxy.py

@@ -0,0 +1,254 @@
+"""Armiger Torznab Filtering Proxy & Pre-Grab Admission Gate for Prowlarr, Bitmagnet, and the *Arr Stack.
+
+Sits transparently between upstream Torznab indexers (`Prowlarr`, `Bitmagnet`, `DH-Waypoint`, `Jackett`)
+and downstream media managers (`Sonarr`, `Radarr`, `Lidarr`, `Readarr`, `Whisparr`).
+
+Every Torznab XML `<item>` returned by an upstream search is evaluated in milliseconds by
+`ArmigerUltraPipeline.classify_batch` before `Sonarr` or `Radarr` can grab it:
+  1. Blocks disguised malware executables (`.mkv.exe`, `.scr`, `.vbs`, `.lnk`, `.ps1`, passworded archives).
+  2. Blocks fake-seeder malware honeypots (`SwarmSurvivalNet` `is_honeypot_trap == True`).
+  3. Blocks dead swarms (`viability_tier == "dead"` when `ARMIGER_REJECT_DEAD_SWARMS=true`).
+  4. Blocks unrequested foreign-language audio traps (`AudioLanguageGuard` cross-head check when
+     `ARMIGER_REQUIRE_ENGLISH_AUDIO=true`, with optional anime/subtitle allowance via `ARMIGER_ALLOW_ENGLISH_SUBS=true`).
+  5. Enriches surviving `<item>` nodes with `<torznab:attr name="armiger_category" .../>` and
+     `<torznab:attr name="armiger_viability" .../>` attributes.
+"""
+from __future__ import annotations
+
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+import json
+import os
+from pathlib import Path
+import sys
+from typing import Any, Dict, List, Optional, Tuple
+import urllib.parse
+import urllib.request
+import xml.etree.ElementTree as ET
+
+ROOT_DIR = Path(__file__).resolve().parents[2]
+if str(ROOT_DIR) not in sys.path:
+    sys.path.insert(0, str(ROOT_DIR))
+
+from armiger_dht_rsi import DHTRSIMode
+from armiger_ultra_pipeline import ArmigerUltraPipeline, ArmigerUltraPrediction
+
+TORZNAB_NS = "http://torznab.com/schemas/2015/feed"
+ET.register_namespace("torznab", TORZNAB_NS)
+
+
+class ArmigerTorznabFilter:
+    """Filters and enriches Torznab RSS/XML feeds using ArmigerUltraPipeline."""
+
+    def __init__(
+        self,
+        pipeline: ArmigerUltraPipeline,
+        *,
+        require_english_audio: bool = False,
+        allow_english_subs: bool = True,
+        reject_dead_swarms: bool = True,
+        allowed_categories: Optional[List[str]] = None,
+        use_zero_cue_catalog: bool = True,
+    ) -> None:
+        self.pipeline = pipeline
+        self.require_english_audio = require_english_audio
+        self.allow_english_subs = allow_english_subs
+        self.reject_dead_swarms = reject_dead_swarms
+        self.allowed_categories = set(allowed_categories) if allowed_categories else None
+        self.use_zero_cue_catalog = use_zero_cue_catalog
+
+    @staticmethod
+    def _extract_item_row(item_elem: ET.Element) -> Dict[str, Any]:
+        title = (item_elem.findtext("title") or "").strip()
+        size_text = item_elem.findtext("size") or "0"
+        try:
+            size_val = int(size_text)
+        except ValueError:
+            size_val = 0
+
+        attrs: Dict[str, str] = {}
+        for attr in item_elem.findall(f"{{{TORZNAB_NS}}}attr"):
+            name = (attr.attrib.get("name") or "").lower().strip()
+            val = attr.attrib.get("value") or ""
+            if name:
+                attrs[name] = val
+
+        seeders = int(attrs.get("seeders", "0") or 0)
+        peers = int(attrs.get("peers", "0") or 0)
+        leechers = max(peers - seeders, int(attrs.get("leechers", "0") or 0))
+        grabs = int(attrs["grabs"]) if "grabs" in attrs and attrs["grabs"].strip() else None
+        infohash = attrs.get("infohash") or ""
+
+        files_attr = attrs.get("files") or ""
+        if files_attr:
+            files = [{"path": p.strip(), "size": size_val} for p in files_attr.split(";") if p.strip()]
+        else:
+            tail = title.rsplit(".", 1)[-1].lower() if "." in title else ""
+            known_exts = {"mkv", "mp4", "avi", "flac", "mp3", "m4b", "epub", "pdf", "cbz", "cbr", "iso", "exe", "scr", "vbs", "bat", "cmd", "lnk", "ps1", "msi"}
+            if tail in known_exts:
+                files = [{"path": title, "size": size_val}]
+            else:
+                low = title.lower()
+                if any(k in low for k in ("flac", "24bit", "lossless")):
+                    synth_ext = ".flac"
+                elif any(k in low for k in ("m4b", "audiobook", "unabridged")):
+                    synth_ext = ".m4b"
+                elif any(k in low for k in ("epub", "ebook")):
+                    synth_ext = ".epub"
+                elif any(k in low for k in ("cbz", "cbr")):
+                    synth_ext = ".cbz"
+                else:
+                    synth_ext = ".mkv"
+                files = [{"path": f"{title}{synth_ext}", "size": size_val}]
+
+        return {
+            "infohash": infohash,
+            "name": title,
+            "size": size_val,
+            "files": files,
+            "swarm": {
+                "tracker_known": bool(seeders > 0 or leechers > 0 or "seeders" in attrs),
+                "tracker_seeders": seeders,
+                "tracker_leechers": leechers,
+                "tracker_completed": grabs,
+            },
+        }
+
+    def should_admit(self, pred: ArmigerUltraPrediction) -> Tuple[bool, str]:
+        """Evaluate whether a Torznab release should be admitted to Sonarr/Radarr/Lidarr/Readarr."""
+        if not pred.csam_safety["is_safe"]:
+            return False, str(pred.csam_safety.get("reason") or "csam_or_malware_blocked")
+        if pred.honeypot_trap["is_honeypot_trap"]:
+            return False, "honeypot_trap_blocked"
+        if pred.junk == "junk" or not pred.mondrian_gates["passed_all"]:
+            return False, str(pred.mondrian_gates.get("safety_gate_reason") or "mondrian_gate_blocked")
+
+        if self.reject_dead_swarms and pred.swarm_viability["viability_tier"] == "dead":
+            return False, "dead_swarm_blocked"
+
+        if self.allowed_categories and pred.category not in self.allowed_categories:
+            return False, f"category_mismatch:{pred.category}"
+
+        if self.require_english_audio and pred.lang == "non_english":
+            has_subs = pred.english_subs == "has_eng_subs"
+            if not (self.allow_english_subs and has_subs):
+                return False, "non_english_audio_blocked"
+
+        return True, "admitted"
+
+    def filter_torznab_xml(self, xml_bytes: bytes) -> Tuple[bytes, Dict[str, Any]]:
+        """Parse upstream Torznab XML, remove blocked `<item>` nodes, and enrich surviving items."""
+        root = ET.fromstring(xml_bytes)
+        channel = root.find("channel")
+        if channel is None:
+            return xml_bytes, {"total": 0, "admitted": 0, "blocked": 0, "blocked_reasons": {}}
+
+        items = channel.findall("item")
+        if not items:
+            return xml_bytes, {"total": 0, "admitted": 0, "blocked": 0, "blocked_reasons": {}}
+
+        rows = [self._extract_item_row(it) for it in items]
+        predictions = self.pipeline.classify_batch(
+            rows,
+            use_zero_cue_catalog=self.use_zero_cue_catalog,
+        )
+
+        admitted_count = 0
+        blocked_count = 0
+        blocked_reasons: Dict[str, int] = {}
+
+        for item_elem, pred in zip(items, predictions):
+            admit, reason = self.should_admit(pred)
+            if not admit:
+                channel.remove(item_elem)
+                blocked_count += 1
+                blocked_reasons[reason] = blocked_reasons.get(reason, 0) + 1
+                continue
+
+            admitted_count += 1
+            for attr_name, attr_val in (
+                ("armiger_category", pred.category),
+                ("armiger_lang", pred.lang),
+                ("armiger_viability", str(pred.swarm_viability["viability_tier"])),
+                ("armiger_survival_prob", f"{pred.swarm_viability['survival_net_prob']:.4f}"),
+            ):
+                attr_el = ET.SubElement(item_elem, f"{{{TORZNAB_NS}}}attr")
+                attr_el.set("name", attr_name)
+                attr_el.set("value", attr_val)
+
+        filtered_bytes = ET.tostring(root, encoding="utf-8", xml_declaration=True)
+        stats = {
+            "total": len(items),
+            "admitted": admitted_count,
+            "blocked": blocked_count,
+            "blocked_reasons": blocked_reasons,
+        }
+        return filtered_bytes, stats
+
+
+def create_proxy_server(host: str = "localhost", port: int = 8097) -> ThreadingHTTPServer:
+    model_dir = os.environ.get("ARMIGER_MODEL_DIR", str(ROOT_DIR))
+    variant = os.environ.get("ARMIGER_VARIANT", "int8")
+    upstream_url = os.environ.get("UPSTREAM_TORZNAB_URL", "http://dh-waypoint:3333/torznab").rstrip("/")
+    pipeline = ArmigerUltraPipeline.from_directory(
+        model_dir,
+        variant=variant,
+        rsi_mode=DHTRSIMode(os.environ.get("ARMIGER_DHT_RSI_MODE", "pull_only")),
+        preload_catalog=os.environ.get("ARMIGER_USE_ZERO_CUE_CATALOG", "true").lower() == "true",
+    )
+    torznab_filter = ArmigerTorznabFilter(
+        pipeline,
+        require_english_audio=os.environ.get("ARMIGER_REQUIRE_ENGLISH_AUDIO", "false").lower() == "true",
+        allow_english_subs=os.environ.get("ARMIGER_ALLOW_ENGLISH_SUBS", "true").lower() == "true",
+        reject_dead_swarms=os.environ.get("ARMIGER_REJECT_DEAD_SWARMS", "true").lower() == "true",
+    )
+
+    class ProxyHandler(BaseHTTPRequestHandler):
+        def do_GET(self) -> None:
+            if self.path == "/healthz":
+                body = json.dumps({"status": "ok", "service": "armiger-torznab-proxy", "variant": variant}).encode("utf-8")
+                self.send_response(200)
+                self.send_header("Content-Type", "application/json")
+                self.send_header("Content-Length", str(len(body)))
+                self.end_headers()
+                self.wfile.write(body)
+                return
+
+            parsed = urllib.parse.urlsplit(self.path)
+            params = urllib.parse.parse_qs(parsed.query)
+            t_param = (params.get("t") or [""])[0].lower()
+            target = f"{upstream_url}{parsed.path}"
+            if parsed.query:
+                target = f"{target}?{parsed.query}"
+
+            req = urllib.request.Request(target, method="GET")
+            with urllib.request.urlopen(req, timeout=15.0) as resp:
+                raw_xml = resp.read()
+                content_type = resp.headers.get("Content-Type", "application/rss+xml; charset=utf-8")
+
+            if t_param in ("caps", ""):
+                out_bytes = raw_xml
+                stats = {"total": 0, "admitted": 0, "blocked": 0}
+            else:
+                out_bytes, stats = torznab_filter.filter_torznab_xml(raw_xml)
+
+            self.send_response(200)
+            self.send_header("Content-Type", content_type)
+            self.send_header("X-Armiger-Total", str(stats["total"]))
+            self.send_header("X-Armiger-Admitted", str(stats["admitted"]))
+            self.send_header("X-Armiger-Blocked", str(stats["blocked"]))
+            self.send_header("Content-Length", str(len(out_bytes)))
+            self.end_headers()
+            self.wfile.write(out_bytes)
+
+        def log_message(self, fmt: str, *args: Any) -> None:
+            return
+
+    return ThreadingHTTPServer((host, port), ProxyHandler)
+
+
+if __name__ == "__main__":
+    listen_host = os.environ.get("ARMIGER_PROXY_HOST", "localhost")
+    listen_port = int(os.environ.get("ARMIGER_PROXY_PORT", "8097"))
+    srv = create_proxy_server(listen_host, listen_port)
+    print(f"Armiger Torznab proxy listening on {listen_host}:{listen_port}")
+    srv.serve_forever()

examples/arr-stack/docker-compose.yml

@@ -0,0 +1,49 @@
+services:
+  armiger-torznab-proxy:
+    image: python:3.12-slim
+    container_name: armiger-torznab-proxy
+    restart: unless-stopped
+    working_dir: /app
+    ports:
+      - "8097:8097"
+    environment:
+      ARMIGER_MODEL_DIR: /app
+      ARMIGER_VARIANT: "${ARMIGER_VARIANT:-int8}"
+      UPSTREAM_TORZNAB_URL: "${UPSTREAM_TORZNAB_URL:-http://prowlarr:9696/1/api}"
+      ARMIGER_PROXY_HOST: "localhost"
+      ARMIGER_PROXY_PORT: "8097"
+      ARMIGER_REQUIRE_ENGLISH_AUDIO: "${ARMIGER_REQUIRE_ENGLISH_AUDIO:-false}"
+      ARMIGER_ALLOW_ENGLISH_SUBS: "${ARMIGER_ALLOW_ENGLISH_SUBS:-true}"
+      ARMIGER_REJECT_DEAD_SWARMS: "${ARMIGER_REJECT_DEAD_SWARMS:-true}"
+      ARMIGER_USE_ZERO_CUE_CATALOG: "true"
+    volumes:
+      - ../..:/app:ro
+    command:
+      - sh
+      - -c
+      - "pip install --no-cache-dir numpy onnxruntime cryptography && python3 examples/arr-stack/armiger_torznab_proxy.py"
+
+  armiger-arr-webhook-receiver:
+    image: python:3.12-slim
+    container_name: armiger-arr-webhook-receiver
+    restart: unless-stopped
+    working_dir: /app
+    ports:
+      - "8098:8098"
+    environment:
+      ARMIGER_MODEL_DIR: /app
+      ARMIGER_VARIANT: "int8"
+      ARMIGER_WEBHOOK_HOST: "localhost"
+      ARMIGER_WEBHOOK_PORT: "8098"
+      ARMIGER_RUN_FFPROBE: "true"
+      ARMIGER_DHT_RSI_MODE: "${ARMIGER_DHT_RSI_MODE:-pull_only}"
+    volumes:
+      - ../..:/app:ro
+      - media_library:/data/media:ro
+    command:
+      - sh
+      - -c
+      - "apt-get update && apt-get install -y --no-install-recommends ffmpeg && pip install --no-cache-dir numpy onnxruntime cryptography && python3 examples/arr-stack/armiger_arr_webhook_receiver.py"
+
+volumes:
+  media_library:

examples/bitmagnet/bitmagnet_armiger_bridge.py

@@ -0,0 +1,266 @@
+"""Bitmagnet <-> Armiger Ultra Pre-Ingest / Post-Crawl Classifier & Admission Bridge.
+
+Provides two operational modes for upstream Bitmagnet deployments:
+  1. HTTP Classifier & Admission Webhook Bridge (`POST /v1/bitmagnet/classify`):
+     Accepts batches of newly discovered Bitmagnet torrents, runs `ArmigerUltraPipeline.classify_batch`,
+     and returns Bitmagnet-compatible content types (`movie`, `tv_show`, `music`, `ebook`, `audiobook`,
+     `comic`, `software`, `game`, `xxx`), structured tags (`armiger:category:*`, `armiger:lang:*`,
+     `armiger:viability:*`), and `delete` actions for disguised malware (`.mkv.exe`, `.scr`, `.lnk`),
+     fake-seeder honeypot traps, or CSAM/safety gate violations.
+  2. GraphQL Mutation & PostgreSQL Reconciliation Helper (`BitmagnetGraphQLSynchronizer`):
+     Generates Bitmagnet GraphQL ` torrentContentUpdate ` / ` torrentDelete ` mutations or SQL updates
+     so upstream Bitmagnet instances gain Armiger's 95.49% 11-class accuracy (vs. 35.29% built-in rules)
+     and 100% fake-seeder malware trap recall.
+"""
+from __future__ import annotations
+
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+import json
+import os
+from pathlib import Path
+import sys
+from typing import Any, Dict, List, Mapping, Optional, Sequence
+import urllib.request
+
+ROOT_DIR = Path(__file__).resolve().parents[2]
+if str(ROOT_DIR) not in sys.path:
+    sys.path.insert(0, str(ROOT_DIR))
+
+from armiger_dht_rsi import DHTRSIMode
+from armiger_ultra_pipeline import ArmigerUltraPipeline, ArmigerUltraPrediction
+
+# Map Armiger 11-class labels to Bitmagnet ContentType enum values
+BITMAGNET_CONTENT_TYPE_MAP: Dict[str, Optional[str]] = {
+    "movie": "movie",
+    "tv_show": "tv_show",
+    "music": "music",
+    "ebook": "ebook",
+    "audiobook": "audiobook",
+    "comic": "comic",
+    "software": "software",
+    "game": "game",
+    "course": "ebook",
+    "xxx": "xxx",
+    "unknown": None,
+}
+
+
+class BitmagnetArmigerBridge:
+    """Classifies, filters, and enriches Bitmagnet torrent items with Armiger Ultra."""
+
+    def __init__(
+        self,
+        pipeline: ArmigerUltraPipeline,
+        *,
+        use_zero_cue_catalog: bool = True,
+        prune_honeypots_and_malware: bool = True,
+        prune_dead_swarms: bool = False,
+        isolate_adult_namespace: bool = True,
+    ) -> None:
+        self.pipeline = pipeline
+        self.use_zero_cue_catalog = use_zero_cue_catalog
+        self.prune_honeypots_and_malware = prune_honeypots_and_malware
+        self.prune_dead_swarms = prune_dead_swarms
+        self.isolate_adult_namespace = isolate_adult_namespace
+
+    @staticmethod
+    def _bitmagnet_item_to_row(item: Mapping[str, Any]) -> Dict[str, Any]:
+        files_in = item.get("files") if isinstance(item.get("files"), list) else []
+        seeders = int(item.get("seeders", item.get("tracker_seeders", 0)) or 0)
+        leechers = int(item.get("leechers", item.get("peers", item.get("tracker_leechers", 0))) or 0)
+        return {
+            "infohash": str(item.get("infoHash") or item.get("info_hash") or item.get("infohash") or ""),
+            "name": str(item.get("name") or item.get("title") or ""),
+            "size": int(item.get("size") or 0),
+            "files": [
+                {"path": str(f.get("path") or f.get("name") or ""), "size": int(f.get("size") or 0)}
+                if isinstance(f, Mapping)
+                else {"path": str(f), "size": 0}
+                for f in files_in
+            ],
+            "file_count": int(item.get("filesCount") or item.get("file_count") or len(files_in) or 1),
+            "swarm": {
+                "tracker_known": bool(seeders > 0 or leechers > 0),
+                "tracker_seeders": seeders,
+                "tracker_leechers": leechers,
+                "tracker_completed": int(item["completed"]) if "completed" in item and item["completed"] is not None else None,
+            },
+        }
+
+    def evaluate_batch(self, items: Sequence[Mapping[str, Any]]) -> Dict[str, Any]:
+        """Run Armiger Ultra on a batch of Bitmagnet torrents and return enrichment/prune actions."""
+        rows = [self._bitmagnet_item_to_row(it) for it in items]
+        predictions: List[ArmigerUltraPrediction] = self.pipeline.classify_batch(
+            rows,
+            use_zero_cue_catalog=self.use_zero_cue_catalog,
+        )
+
+        actions: List[Dict[str, Any]] = []
+        delete_infohashes: List[str] = []
+
+        for pred in predictions:
+            is_malware_or_unsafe = not pred.csam_safety["is_safe"] or pred.junk == "junk"
+            is_honeypot = bool(pred.honeypot_trap["is_honeypot_trap"])
+            viability_tier = str(pred.swarm_viability["viability_tier"])
+            is_dead = viability_tier == "dead"
+
+            if self.prune_honeypots_and_malware and (is_malware_or_unsafe or is_honeypot):
+                reason = (
+                    pred.csam_safety.get("reason")
+                    or ("honeypot_trap" if is_honeypot else "junk_or_malware_gate")
+                )
+                delete_infohashes.append(pred.infohash)
+                actions.append({
+                    "infoHash": pred.infohash,
+                    "action": "delete",
+                    "reason": str(reason),
+                    "category": pred.category,
+                    "honeypot_trap": is_honeypot,
+                    "junk": pred.junk,
+                })
+                continue
+
+            if self.prune_dead_swarms and is_dead:
+                delete_infohashes.append(pred.infohash)
+                actions.append({
+                    "infoHash": pred.infohash,
+                    "action": "delete",
+                    "reason": "dead_swarm",
+                    "category": pred.category,
+                    "viability_tier": viability_tier,
+                })
+                continue
+
+            if self.isolate_adult_namespace and pred.adult == "adult" and pred.category != "xxx":
+                bitmagnet_type: Optional[str] = "xxx"
+            else:
+                bitmagnet_type = BITMAGNET_CONTENT_TYPE_MAP.get(pred.category)
+
+            tags = [
+                f"armiger:{pred.category}",
+                f"armiger:lang:{pred.lang}",
+                f"armiger:viability:{viability_tier}",
+            ]
+            if pred.anime == "anime":
+                tags.append("armiger:anime")
+                dub_status = pred.heads["anime"].get("anime_dub_status")
+                if dub_status:
+                    tags.append(f"armiger:dub:{dub_status}")
+            if pred.english_subs == "has_eng_subs":
+                tags.append("armiger:eng_subs")
+
+            actions.append({
+                "infoHash": pred.infohash,
+                "action": "update",
+                "contentType": bitmagnet_type,
+                "armigerCategory": pred.category,
+                "languages": ["en"] if pred.lang == "english" else [],
+                "tags": tags,
+                "viabilityTier": viability_tier,
+                "viabilityProb": float(pred.swarm_viability["viability_prob"]),
+                "mondrianPassed": bool(pred.mondrian_gates["passed_all"]),
+            })
+
+        return {
+            "total": len(actions),
+            "deletedCount": len(delete_infohashes),
+            "deleteInfoHashes": delete_infohashes,
+            "actions": actions,
+            "graphqlDeleteMutation": self.build_graphql_delete_mutation(delete_infohashes),
+        }
+
+    @staticmethod
+    def build_graphql_delete_mutation(infohashes: Sequence[str]) -> Optional[Dict[str, Any]]:
+        """Build a Bitmagnet GraphQL `torrentDelete` mutation payload for blocked infohashes."""
+        clean = [h.strip().lower() for h in infohashes if h and h.strip()]
+        if not clean:
+            return None
+        return {
+            "query": (
+                "mutation ArmigerPruneMalwareAndHoneypots($infoHashes: [Hash20!]!) { "
+                "torrent { delete(input: { infoHashes: $infoHashes }) { infoHashes } } "
+                "}"
+            ),
+            "variables": {"infoHashes": clean},
+        }
+
+    @staticmethod
+    def push_graphql_delete(
+        graphql_endpoint: str,
+        infohashes: Sequence[str],
+        timeout: float = 5.0,
+    ) -> Optional[Dict[str, Any]]:
+        """Execute the `torrentDelete` GraphQL mutation against a running Bitmagnet endpoint."""
+        mutation = BitmagnetArmigerBridge.build_graphql_delete_mutation(infohashes)
+        if mutation is None:
+            return None
+        req = urllib.request.Request(
+            graphql_endpoint,
+            data=json.dumps(mutation).encode("utf-8"),
+            headers={"Content-Type": "application/json"},
+            method="POST",
+        )
+        with urllib.request.urlopen(req, timeout=timeout) as resp:
+            return json.loads(resp.read().decode("utf-8"))
+
+
+def create_server(host: str = "localhost", port: int = 8096) -> ThreadingHTTPServer:
+    model_dir = os.environ.get("ARMIGER_MODEL_DIR", str(ROOT_DIR))
+    variant = os.environ.get("ARMIGER_VARIANT", "fp32")
+    pipeline = ArmigerUltraPipeline.from_directory(
+        model_dir,
+        variant=variant,
+        rsi_mode=DHTRSIMode(os.environ.get("ARMIGER_DHT_RSI_MODE", "pull_only")),
+        preload_catalog=os.environ.get("ARMIGER_USE_ZERO_CUE_CATALOG", "true").lower() == "true",
+    )
+    bridge = BitmagnetArmigerBridge(
+        pipeline,
+        use_zero_cue_catalog=os.environ.get("ARMIGER_USE_ZERO_CUE_CATALOG", "true").lower() == "true",
+        prune_honeypots_and_malware=os.environ.get("ARMIGER_PRUNE_MALWARE", "true").lower() == "true",
+        prune_dead_swarms=os.environ.get("ARMIGER_PRUNE_DEAD_SWARMS", "false").lower() == "true",
+    )
+
+    class Handler(BaseHTTPRequestHandler):
+        def _send_json(self, status: int, payload: Dict[str, Any]) -> None:
+            body = json.dumps(payload).encode("utf-8")
+            self.send_response(status)
+            self.send_header("Content-Type", "application/json")
+            self.send_header("Content-Length", str(len(body)))
+            self.end_headers()
+            self.wfile.write(body)
+
+        def do_GET(self) -> None:
+            if self.path == "/healthz":
+                self._send_json(200, {"status": "ok", "service": "bitmagnet-armiger-bridge", "variant": variant})
+            else:
+                self._send_json(404, {"error": "not_found"})
+
+        def do_POST(self) -> None:
+            length = int(self.headers.get("Content-Length", "0"))
+            raw = self.rfile.read(length) if length > 0 else b"{}"
+            payload = json.loads(raw.decode("utf-8"))
+            if self.path == "/v1/bitmagnet/classify":
+                items = payload.get("torrents") or payload.get("items") or []
+                result = bridge.evaluate_batch(items)
+                gql_url = os.environ.get("BITMAGNET_GRAPHQL_URL", "").strip()
+                if gql_url and result["deleteInfoHashes"]:
+                    try:
+                        bridge.push_graphql_delete(gql_url, result["deleteInfoHashes"])
+                    except Exception as exc:
+                        result["graphqlPushError"] = str(exc)
+                self._send_json(200, result)
+            else:
+                self._send_json(404, {"error": "not_found"})
+
+        def log_message(self, fmt: str, *args: Any) -> None:
+            return
+
+    return ThreadingHTTPServer((host, port), Handler)
+
+
+if __name__ == "__main__":
+    listen_host = os.environ.get("ARMIGER_BRIDGE_HOST", "localhost")
+    listen_port = int(os.environ.get("ARMIGER_BRIDGE_PORT", "8096"))
+    srv = create_server(listen_host, listen_port)
+    print(f"Bitmagnet <-> Armiger Ultra bridge listening on {listen_host}:{listen_port}")
+    srv.serve_forever()

examples/bitmagnet/docker-compose.yml

@@ -0,0 +1,90 @@
+services:
+  bitmagnet:
+    image: ghcr.io/bitmagnet-io/bitmagnet:latest
+    container_name: bitmagnet
+    restart: unless-stopped
+    depends_on:
+      postgres:
+        condition: service_healthy
+    ports:
+      - "3333:3333"
+      - "3334:3334/tcp"
+      - "3334:3334/udp"
+    environment:
+      POSTGRES_HOST: postgres
+      POSTGRES_NAME: bitmagnet
+      POSTGRES_USER: postgres
+      POSTGRES_PASSWORD: "${POSTGRES_PASSWORD:?set POSTGRES_PASSWORD in .env}"
+    command:
+      - worker
+      - run
+      - --keys=http_server
+      - --keys=queue_server
+      - --keys=dht_crawler
+
+  armiger-bitmagnet-bridge:
+    image: python:3.12-slim
+    container_name: armiger-bitmagnet-bridge
+    restart: unless-stopped
+    working_dir: /app
+    depends_on:
+      - bitmagnet
+    ports:
+      - "8096:8096"
+    environment:
+      ARMIGER_MODEL_DIR: /app
+      ARMIGER_VARIANT: "${ARMIGER_VARIANT:-fp32}"
+      ARMIGER_BRIDGE_HOST: "localhost"
+      ARMIGER_BRIDGE_PORT: "8096"
+      BITMAGNET_GRAPHQL_URL: "http://bitmagnet:3333/graphql"
+      ARMIGER_PRUNE_MALWARE: "true"
+      ARMIGER_PRUNE_DEAD_SWARMS: "false"
+    volumes:
+      - ../..:/app:ro
+    command:
+      - sh
+      - -c
+      - "pip install --no-cache-dir numpy onnxruntime cryptography && python3 examples/bitmagnet/bitmagnet_armiger_bridge.py"
+
+  armiger-torznab-proxy:
+    image: python:3.12-slim
+    container_name: armiger-torznab-proxy
+    restart: unless-stopped
+    working_dir: /app
+    depends_on:
+      - bitmagnet
+    ports:
+      - "8097:8097"
+    environment:
+      ARMIGER_MODEL_DIR: /app
+      ARMIGER_VARIANT: "${ARMIGER_VARIANT:-int8}"
+      UPSTREAM_TORZNAB_URL: "http://bitmagnet:3333/torznab"
+      ARMIGER_PROXY_PORT: "8097"
+      ARMIGER_REQUIRE_ENGLISH_AUDIO: "false"
+      ARMIGER_ALLOW_ENGLISH_SUBS: "true"
+      ARMIGER_REJECT_DEAD_SWARMS: "true"
+    volumes:
+      - ../..:/app:ro
+    command:
+      - sh
+      - -c
+      - "pip install --no-cache-dir numpy onnxruntime cryptography && python3 examples/arr-stack/armiger_torznab_proxy.py"
+
+  postgres:
+    image: postgres:16-alpine
+    container_name: bitmagnet-postgres
+    restart: unless-stopped
+    environment:
+      POSTGRES_DB: bitmagnet
+      POSTGRES_USER: postgres
+      POSTGRES_PASSWORD: "${POSTGRES_PASSWORD:?set POSTGRES_PASSWORD in .env}"
+    volumes:
+      - bitmagnet_pgdata:/var/lib/postgresql/data
+    healthcheck:
+      test: ["CMD-SHELL", "pg_isready -d bitmagnet -U postgres"]
+      interval: 10s
+      timeout: 5s
+      retries: 5
+
+volumes:
+  bitmagnet_pgdata:

examples/dh-waypoint/07-armiger-ultra.plugin.yml

@@ -0,0 +1,33 @@
+# Drop this file into DH_WAYPOINT_PROCESSORS_PLUGIN_DIR (default /etc/dh-waypoint/plugins.d).
+#
+# Full Armiger Ultra integration wiring:
+#   1. Batch classification (`POST /v1/classify`, `dh-waypoint.processor.v1`)
+#   2. Cohort admission gate (`POST /v1/admit`, `dh-waypoint.processors.admission.v1`)
+#   3. 4-tier closed-loop outcome feedback (`POST /v1/feedback`, `dh-waypoint.processors.feedback.v1`)
+
+processor:
+  name: armiger-ultra
+  endpoint: http://armiger-ultra:8095/v1/classify
+  timeout: 5s
+  max_batch: 64
+  required: false
+  forward_files: true
+  max_files: 64
+  forward_trackers: true
+  forward_swarm: true
+
+admission:
+  enabled: true
+  endpoint: http://armiger-ultra:8095/v1/admit
+  timeout: 3s
+  max_batch: 64
+  max_files: 64
+  fail_open: true
+
+learning:
+  enabled: true
+  feedback_endpoint: http://armiger-ultra:8095/v1/feedback
+  feedback_timeout: 5s
+  feedback_batch_size: 64
+  flush_interval: 15s
+  queue_size: 2048

examples/dh-waypoint/armiger_waypoint_processor.py

@@ -0,0 +1,270 @@
+"""DH-Waypoint processorapi/v1 HTTP Service for Armiger & Armiger Ultra.
+
+Implements all three DH-Waypoint external processor contracts:
+  - `POST /v1/classify` (`dh-waypoint.processor.v1`)
+  - `POST /v1/admit` (`dh-waypoint.processors.admission.v1`)
+  - `POST /v1/feedback` (`dh-waypoint.processors.feedback.v1`)
+  - `GET /healthz`
+"""
+from __future__ import annotations
+
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+import json
+import os
+from pathlib import Path
+import sys
+from typing import Any, Dict, List, Mapping, Optional
+
+ROOT_DIR = Path(__file__).resolve().parents[2]
+if str(ROOT_DIR) not in sys.path:
+    sys.path.insert(0, str(ROOT_DIR))
+
+from armiger_dht_rsi import CandidateObservation, DHTRSIMode
+from armiger_ultra_pipeline import ArmigerUltraPipeline, ArmigerUltraPrediction
+
+# Map Armiger 11-class labels to DH-Waypoint canonical content types
+WAYPOINT_CONTENT_TYPE_MAP: Dict[str, str] = {
+    "movie": "movie",
+    "tv_show": "tv_show",
+    "music": "music",
+    "ebook": "ebook",
+    "audiobook": "audiobook",
+    "comic": "comic",
+    "software": "software",
+    "game": "game",
+    "course": "ebook",
+    "xxx": "xxx",
+    "unknown": "unknown",
+}
+
+
+class ArmigerWaypointAdapter:
+    """Translates between DH-Waypoint processorapi/v1 envelopes and ArmigerUltraPipeline."""
+
+    def __init__(
+        self,
+        pipeline: ArmigerUltraPipeline,
+        *,
+        use_zero_cue_catalog: bool = True,
+        reject_dead_swarms: bool = False,
+    ) -> None:
+        self.pipeline = pipeline
+        self.use_zero_cue_catalog = use_zero_cue_catalog
+        self.reject_dead_swarms = reject_dead_swarms
+
+    @staticmethod
+    def _torrent_to_row(t: Mapping[str, Any]) -> Dict[str, Any]:
+        swarm_in = t.get("swarm") if isinstance(t.get("swarm"), Mapping) else {}
+        seeders = int(swarm_in.get("seeders", t.get("seeders", 0)) or 0)
+        leechers = int(swarm_in.get("leechers", t.get("leechers", 0)) or 0)
+        files = t.get("files") if isinstance(t.get("files"), list) else []
+        return {
+            "infohash": str(t.get("info_hash") or t.get("infohash") or ""),
+            "name": str(t.get("name") or t.get("title") or ""),
+            "size": int(t.get("size") or 0),
+            "files": files,
+            "file_count": int(t.get("files_count") or len(files) or 1),
+            "sources": t.get("sources") if isinstance(t.get("sources"), list) else [],
+            "swarm": {
+                "tracker_known": bool(seeders > 0 or leechers > 0 or swarm_in),
+                "tracker_seeders": seeders,
+                "tracker_leechers": leechers,
+                "tracker_completed": int(swarm_in["completed"]) if "completed" in swarm_in and swarm_in["completed"] is not None else None,
+            },
+        }
+
+    def handle_classify_batch(self, request_body: Mapping[str, Any]) -> Dict[str, Any]:
+        """Handle `POST /v1/classify` (`dh-waypoint.processor.v1`)."""
+        torrents = request_body.get("torrents") or []
+        rows = [self._torrent_to_row(t) for t in torrents]
+        predictions: List[ArmigerUltraPrediction] = self.pipeline.classify_batch(
+            rows,
+            use_zero_cue_catalog=self.use_zero_cue_catalog,
+        )
+
+        decisions: List[Dict[str, Any]] = []
+        for pred in predictions:
+            gates_passed = bool(pred.mondrian_gates["passed_all"])
+            viability_tier = str(pred.swarm_viability["viability_tier"])
+            if not gates_passed:
+                reason = (
+                    pred.csam_safety.get("reason")
+                    or pred.mondrian_gates.get("safety_gate_reason")
+                    or ("honeypot_trap_blocked" if pred.honeypot_trap["is_honeypot_trap"] else "safety_gate_rejected")
+                )
+                decisions.append({
+                    "info_hash": pred.infohash,
+                    "action": "delete",
+                    "reason": str(reason),
+                })
+                continue
+
+            content_type = WAYPOINT_CONTENT_TYPE_MAP.get(pred.category, "unknown")
+            tags = [
+                f"armiger:category:{pred.category}",
+                f"armiger:lang:{pred.lang}",
+                f"armiger:viability:{viability_tier}",
+            ]
+            if pred.anime == "anime":
+                tags.append("armiger:anime")
+                dub_status = pred.heads["anime"].get("anime_dub_status")
+                if dub_status:
+                    tags.append(f"armiger:anime_dub:{dub_status}")
+            if pred.english_subs == "has_eng_subs":
+                tags.append("armiger:eng_subs")
+
+            decisions.append({
+                "info_hash": pred.infohash,
+                "action": "enrich",
+                "content_type": content_type,
+                "languages": ["en"] if pred.lang == "english" else [],
+                "tags": tags,
+                "attributes": {
+                    "armiger.category": pred.category,
+                    "armiger.category_confidence": f"{pred.heads['category']['confidence']:.4f}",
+                    "armiger.lang": pred.lang,
+                    "armiger.viability_tier": viability_tier,
+                    "armiger.viability_prob": f"{pred.swarm_viability['viability_prob']:.4f}",
+                    "armiger.survival_net_prob": f"{pred.swarm_viability['survival_net_prob']:.4f}",
+                    "armiger.catalog_source": str(pred.plugin_trace["plugin_3_zero_cue_catalog"]["resolution_source"]),
+                },
+            })
+
+        return {
+            "schema": "dh-waypoint.processor.v1",
+            "decisions": decisions,
+        }
+
+    def handle_admission_batch(self, request_body: Mapping[str, Any]) -> Dict[str, Any]:
+        """Handle `POST /v1/admit` (`dh-waypoint.processors.admission.v1`)."""
+        candidates = request_body.get("candidates") or []
+        rows = [self._torrent_to_row(c) for c in candidates]
+        predictions = self.pipeline.classify_batch(
+            rows,
+            use_zero_cue_catalog=self.use_zero_cue_catalog,
+        )
+
+        decisions: List[Dict[str, Any]] = []
+        for cand, pred in zip(candidates, predictions):
+            info_hash = str(cand.get("info_hash") or pred.infohash)
+            gates_passed = bool(pred.mondrian_gates["passed_all"])
+            viability_tier = str(pred.swarm_viability["viability_tier"])
+            admit = gates_passed and (not self.reject_dead_swarms or viability_tier != "dead")
+            reason = "passed_mondrian_conformal_gates"
+            if not gates_passed:
+                reason = str(
+                    pred.csam_safety.get("reason")
+                    or pred.mondrian_gates.get("safety_gate_reason")
+                    or "honeypot_or_safety_gate_blocked"
+                )
+            elif not admit:
+                reason = "dead_swarm_rejected"
+
+            decisions.append({
+                "info_hash": info_hash,
+                "admit": admit,
+                "cohort": pred.category if admit else "rejected",
+                "score": float(pred.swarm_viability["viability_prob"] if admit else 0.0),
+                "reason": reason,
+            })
+
+        return {
+            "schema": "dh-waypoint.processors.admission.v1",
+            "decisions": decisions,
+        }
+
+    def handle_feedback_batch(self, request_body: Mapping[str, Any]) -> Dict[str, Any]:
+        """Handle `POST /v1/feedback` (`dh-waypoint.processors.feedback.v1`)."""
+        events = request_body.get("events") or []
+        accepted = 0
+        verified = 0
+        for ev in events:
+            accepted += 1
+            tier = str(ev.get("tier") or "")
+            outcome = str(ev.get("outcome") or "")
+            if tier in ("t3_files", "t4_playback") and outcome in ("verified", "imported", "completed"):
+                cand = CandidateObservation(
+                    source_adapter="ffprobe",
+                    raw_title=str(ev.get("name") or ev.get("info_hash") or "verified_item"),
+                    claimed_category=str(ev.get("content_type") or "movie"),
+                    download_completed=True,
+                    bytes_downloaded=int(ev.get("size") or 100_000_000),
+                    total_bytes=int(ev.get("size") or 100_000_000),
+                    catalog_id_type="movie_catalog" if ev.get("content_type") == "movie" else "series_catalog",
+                    catalog_id_value=str(ev.get("info_hash") or "verified"),
+                    stream_metadata={
+                        "video_codec": str((ev.get("attributes") or {}).get("video_codec", "hevc")),
+                        "audio_codec": str((ev.get("attributes") or {}).get("audio_codec", "eac3")),
+                        "audio_languages": [str((ev.get("attributes") or {}).get("audio_lang", "eng"))],
+                    },
+                )
+                ok, _, _ = self.pipeline.ingest_rsi_feedback(cand)
+                if ok:
+                    verified += 1
+
+        return {
+            "schema": "dh-waypoint.processors.feedback.v1",
+            "accepted": accepted,
+            "verified": verified,
+        }
+
+
+def create_server(host: str = "localhost", port: int = 8095) -> ThreadingHTTPServer:
+    model_dir = os.environ.get("ARMIGER_MODEL_DIR", str(ROOT_DIR))
+    variant = os.environ.get("ARMIGER_VARIANT", "fp32")
+    rsi_mode_str = os.environ.get("ARMIGER_DHT_RSI_MODE", "pull_only")
+    pipeline = ArmigerUltraPipeline.from_directory(
+        model_dir,
+        variant=variant,
+        adult_threshold=float(os.environ.get("ARMIGER_ADULT_THRESHOLD", "0.35")),
+        junk_threshold=float(os.environ.get("ARMIGER_JUNK_THRESHOLD", "0.25")),
+        scope_threshold=float(os.environ.get("ARMIGER_SCOPE_THRESHOLD", "0.35")),
+        rsi_mode=DHTRSIMode(rsi_mode_str),
+        preload_catalog=os.environ.get("ARMIGER_USE_ZERO_CUE_CATALOG", "true").lower() == "true",
+    )
+    adapter = ArmigerWaypointAdapter(
+        pipeline,
+        use_zero_cue_catalog=os.environ.get("ARMIGER_USE_ZERO_CUE_CATALOG", "true").lower() == "true",
+        reject_dead_swarms=os.environ.get("ARMIGER_REJECT_DEAD_SWARMS", "false").lower() == "true",
+    )
+
+    class Handler(BaseHTTPRequestHandler):
+        def _send_json(self, status: int, payload: Dict[str, Any]) -> None:
+            body = json.dumps(payload).encode("utf-8")
+            self.send_response(status)
+            self.send_header("Content-Type", "application/json")
+            self.send_header("Content-Length", str(len(body)))
+            self.end_headers()
+            self.wfile.write(body)
+
+        def do_GET(self) -> None:
+            if self.path == "/healthz":
+                self._send_json(200, {"status": "ok", "engine": "armiger-ultra", "variant": variant})
+            else:
+                self._send_json(404, {"error": "not_found"})
+
+        def do_POST(self) -> None:
+            length = int(self.headers.get("Content-Length", "0"))
+            raw = self.rfile.read(length) if length > 0 else b"{}"
+            payload = json.loads(raw.decode("utf-8"))
+            if self.path == "/v1/classify":
+                self._send_json(200, adapter.handle_classify_batch(payload))
+            elif self.path == "/v1/admit":
+                self._send_json(200, adapter.handle_admission_batch(payload))
+            elif self.path == "/v1/feedback":
+                self._send_json(200, adapter.handle_feedback_batch(payload))
+            else:
+                self._send_json(404, {"error": "not_found"})
+
+        def log_message(self, fmt: str, *args: Any) -> None:
+            return
+
+    return ThreadingHTTPServer((host, port), Handler)
+
+
+if __name__ == "__main__":
+    listen_host = os.environ.get("ARMIGER_HOST", "localhost")
+    listen_port = int(os.environ.get("ARMIGER_PORT", "8095"))
+    srv = create_server(listen_host, listen_port)
+    print(f"Armiger DH-Waypoint processor listening on {listen_host}:{listen_port}")
+    srv.serve_forever()

examples/dh-waypoint/docker-compose.yml

@@ -0,0 +1,78 @@
+services:
+  dh-waypoint:
+    image: ghcr.io/spencernorton/dh-waypoint:v5.6.0
+    container_name: dh-waypoint
+    restart: unless-stopped
+    depends_on:
+      postgres:
+        condition: service_healthy
+      armiger-ultra:
+        condition: service_healthy
+    ports:
+      - "3333:3333"
+      - "3334:3334/tcp"
+      - "3334:3334/udp"
+    environment:
+      POSTGRES_HOST: postgres
+      POSTGRES_PORT: "5432"
+      POSTGRES_DB: dh_waypoint
+      POSTGRES_USER: postgres
+      POSTGRES_PASSWORD: "${POSTGRES_PASSWORD:?set POSTGRES_PASSWORD in .env}"
+      DH_WAYPOINT_PROCESSORS_PLUGIN_DIR: /etc/dh-waypoint/plugins.d
+    volumes:
+      - ./07-armiger-ultra.plugin.yml:/etc/dh-waypoint/plugins.d/07-armiger-ultra.plugin.yml:ro
+    command:
+      - worker
+      - run
+      - --keys=http_server
+      - --keys=queue_server
+      - --keys=dht_crawler
+
+  armiger-ultra:
+    image: python:3.12-slim
+    container_name: armiger-ultra
+    restart: unless-stopped
+    working_dir: /app
+    ports:
+      - "8095:8095"
+    environment:
+      ARMIGER_MODEL_DIR: /app
+      ARMIGER_VARIANT: "${ARMIGER_VARIANT:-fp32}"
+      ARMIGER_PORT: "8095"
+      ARMIGER_ADULT_THRESHOLD: "0.35"
+      ARMIGER_JUNK_THRESHOLD: "0.25"
+      ARMIGER_SCOPE_THRESHOLD: "0.35"
+      ARMIGER_USE_ZERO_CUE_CATALOG: "true"
+      ARMIGER_DHT_RSI_MODE: "${ARMIGER_DHT_RSI_MODE:-pull_only}"
+    volumes:
+      - ../..:/app:ro
+      - armiger_state:/var/lib/armiger
+    command:
+      - sh
+      - -c
+      - "pip install --no-cache-dir numpy onnxruntime cryptography && python3 examples/dh-waypoint/armiger_waypoint_processor.py"
+    healthcheck:
+      test: ["CMD", "python3", "-c", "import urllib.request; urllib.request.urlopen('http://localhost:8095/healthz')"]
+      interval: 15s
+      timeout: 5s
+      retries: 5
+
+  postgres:
+    image: postgres:16-alpine
+    container_name: dh-waypoint-postgres
+    restart: unless-stopped
+    environment:
+      POSTGRES_DB: dh_waypoint
+      POSTGRES_USER: postgres
+      POSTGRES_PASSWORD: "${POSTGRES_PASSWORD:?set POSTGRES_PASSWORD in .env}"
+    volumes:
+      - dh_waypoint_pgdata:/var/lib/postgresql/data
+    healthcheck:
+      test: ["CMD-SHELL", "pg_isready -d dh_waypoint -U postgres"]
+      interval: 10s
+      timeout: 5s
+      retries: 5
+
+volumes:
+  armiger_state:
+  dh_waypoint_pgdata: