clippy: a char array for the subtitle separators in main_title (manual_pattern_char_comparison)

The test job's linker was killed with signal 9 twice (run 705): cargo test links
several test binaries at once, each with full debug info. CARGO_PROFILE_DEV_DEBUG=0
and CARGO_BUILD_JOBS=2 keep the link within the job's memory; the tests are the
same.

.gitattributes

@@ -0,0 +1,6 @@
+# FIX CRLF always when developer has not set
+# Linux/Mac: git config --global core.autocrlf input
+# Windows: git config --global core.autocrlf true
+# Auto detect text files and perform LF normalization
+* text=auto
+* eol=lf

.gitea/workflows/publish.yml

@@ -0,0 +1,85 @@
+# Gitea twin of .github/workflows/publish.yml. Gitea runs only .gitea/workflows/ once this
+# folder exists, and GitHub never reads it.
+#
+# Publishes to IPCR on watch.nsl.sh instead of ghcr.io:
+#   push  ipcr-push-85-17-246-67.nip.io/metadec.eth/<repo>:<tag>   docker login as metadec.eth with the
+#         IPCR_PUSH_TOKEN secret (the metadec.eth namespace token from the IPCR dashboard)
+#   pull  ipcr-watch.nsl.sh/ipns/metadec.eth/metadec.eth/<repo>:<tag>   anonymous
+# The push goes to ipcr-push-85-17-246-67.nip.io, the box's own address, not ipcr-push-watch.nsl.sh: Cloudflare fronts
+# *.watch.nsl.sh and refuses any request over 100 MB (413), which a large image layer is.
+# That host's certificate is signed by the box's private Mesh Router CA, given to buildkit from
+# the IPCR_CA_PEM repo variable. docker login would make the job's Docker daemon check the
+# certificate without that CA, so the login is written straight into ~/.docker/config.json.
+# The runner's job image (node:22-bookworm) has no docker CLI, so the static one is
+# installed first; buildx installs itself. No `type=gha` cache: the runner has none.
+
+name: Publish image
+
+# Build and push the feeder image to the IPCR (ipcr-push-watch.nsl.sh):
+#   * every push to main  -> the moving `main` tag
+#   * every `v*` tag       -> semver tags (e.g. v1.0.0 -> 1.0.0, 1.0, 1)
+# The build context is the repo root (the workspace); `meta-feeder-sdk` is a
+# tag-pinned git dependency fetched during the build. The Dockerfile lives under
+# feeder-plugin/book-feeder/.
+on:
+  push:
+    branches: [main]
+    tags: ['v*']
+
+env:
+  REGISTRY: ipcr-push-85-17-246-67.nip.io
+  IMAGE_NAME: metadec.eth/${{ github.event.repository.name }}
+
+jobs:
+  publish:
+    runs-on: ubuntu-latest
+    permissions:
+      contents: read
+      packages: write
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+
+      - name: Install Docker CLI
+        run: curl -fsSL https://download.docker.com/linux/static/stable/x86_64/docker-28.5.1.tgz | tar -xz -C /usr/local/bin --strip-components=1 docker/docker
+
+      - name: Trust the Mesh Router CA
+        run: printf '%s\n' "$IPCR_CA_PEM" > /tmp/ipcr-ca.pem
+        env:
+          IPCR_CA_PEM: ${{ vars.IPCR_CA_PEM }}
+
+      - name: Set up Docker Buildx
+        uses: docker/setup-buildx-action@v3
+        with:
+          version: latest
+          buildkitd-config-inline: |
+            [registry."${{ env.REGISTRY }}"]
+              ca = ["/tmp/ipcr-ca.pem"]
+
+      - name: Log in to IPCR
+        env:
+          IPCR_PUSH_TOKEN: ${{ secrets.IPCR_PUSH_TOKEN }}
+        run: |
+          mkdir -p ~/.docker
+          printf '{"auths":{"%s":{"auth":"%s"}}}\n' "$REGISTRY" "$(printf 'metadec.eth:%s' "$IPCR_PUSH_TOKEN" | base64 -w0)" > ~/.docker/config.json
+
+      - name: Extract image metadata
+        id: meta
+        uses: docker/metadata-action@v5
+        with:
+          images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}
+          tags: |
+            type=ref,event=branch
+            type=semver,pattern={{version}}
+            type=semver,pattern={{major}}.{{minor}}
+            type=semver,pattern={{major}}
+            type=sha,format=short
+
+      - name: Build and push
+        uses: docker/build-push-action@v6
+        with:
+          context: .
+          file: ./feeder-plugin/book-feeder/Dockerfile
+          push: true
+          tags: ${{ steps.meta.outputs.tags }}
+          labels: ${{ steps.meta.outputs.labels }}

.gitea/workflows/test.yml

@@ -0,0 +1,31 @@
+# Runs the feeder's tests on every push. The repo had only the image workflow,
+# so nothing checked a change before it shipped.
+
+name: Test
+
+on:
+  push:
+    branches: ["**"]
+
+jobs:
+  rust:
+    name: cargo test
+    runs-on: ubuntu-latest
+    # The job ran out of memory linking the test binaries (ld killed with
+    # signal 9, run 705 twice): no debug info, and at most two links at once.
+    # What is tested does not change.
+    env:
+      CARGO_PROFILE_DEV_DEBUG: "0"
+      CARGO_BUILD_JOBS: "2"
+    steps:
+      - uses: actions/checkout@v4
+      # rust-toolchain.toml pins the channel (with clippy); rustup installs it on
+      # first use, so the action only has to provide rustup itself.
+      - uses: dtolnay/rust-toolchain@stable
+        with:
+          components: clippy
+      - uses: Swatinem/rust-cache@v2
+      - name: cargo test
+        run: cargo test --locked
+      - name: cargo clippy
+        run: cargo clippy --all-targets -- -D warnings

.github/workflows/publish.yml

@@ -3,8 +3,8 @@ name: Publish image
 # Build and push the feeder image to the GitHub Container Registry (ghcr.io):
 #   * every push to main  -> the moving `main` tag
 #   * every `v*` tag       -> semver tags (e.g. v1.0.0 -> 1.0.0, 1.0, 1)
-# The build context is the repo root (the workspace) so the vendored
-# `meta-feeder-sdk` path dependency resolves; the Dockerfile lives under
+# The build context is the repo root (the workspace); `meta-feeder-sdk` is a
+# tag-pinned git dependency fetched during the build. The Dockerfile lives under
 # feeder-plugin/book-feeder/.
 on:
   push:

Cargo.toml

@@ -1,6 +1,10 @@
 [workspace]
 resolver = "2"
-members = ["crates/*", "feeder-plugin/*"]
+members = ["feeder-plugin/*"]
+# The vendored SDK copy is no longer built: book-feeder pins meta-feeder-sdk by
+# git tag. Excluded (not deleted) so a stray `cargo` run inside it does not try
+# to join this workspace.
+exclude = ["crates/meta-feeder-sdk"]
 
 [workspace.package]
 edition = "2021"

README.md

@@ -1 +1,96 @@
-# meta-feeder-book
+# meta-feeder-book
+
+Book **feeder sidecar** for [MetaMesh](https://github.com/worph/meta-gateway).
+One binary, `book-feeder`, hosts two upstreams:
+
+| Upstream | Serves | What it is |
+|----------|--------|------------|
+| `gutenberg` | `document` × `book` | Public-domain **editions** from Project Gutenberg (via the [Gutendex](https://gutendex.com) API), with an epub url-locator per record |
+| `openlibrary` | `card` × `book` | **Work cards** from [Open Library](https://openlibrary.org) — the book identity tier (`TMDB : video :: AniList : comics :: Open Library : books`) |
+
+The two meet through `openlibraryid`. A Gutenberg edition carries it only when
+Open Library itself records the link: exactly one Open Library work lists the
+etext's Project Gutenberg id (`search.json?q=id_project_gutenberg:<id>`). The
+edition then also carries the transient trust claim `anchored=true`,
+`anchorSource=openlibrary`, which the gateway stores as
+`anchoredBy/openlibrary:OL…W`. There is no title or author matching.
+
+An Open Library card is addressed `compute_card_cid("openlibrary", "OL…W")`,
+the bare work id, byte-identical to meta-read's `card_cid::openlibrary`. A work
+with no cover or no description gets no card.
+
+## Role in MetaMesh
+
+A feeder is a stateless HTTP sidecar. It **finds records and fetches bytes**; it
+does *not* talk to meta-core or the libp2p blockstore. The gateway core that
+calls it owns the meta-core store-back and the blockstore seeding. A gateway
+registers this feeder as a `RemoteFeederPlugin` pointing at its `/` and then
+drives the contract:
+
+| Endpoint | Purpose |
+|----------|---------|
+| `GET /manifest` | feeder identity + capabilities |
+| `GET /health` | liveness |
+| `POST /query`, `POST /query_stream` | structured search against the upstream |
+| `POST /compute` | enrichment / outcome compute |
+| `GET /fetch/:upstream_id/:record_id` | fetch a record's bytes |
+| `GET /blob/:upstream_id/:cid` | fetch a content-addressed blob |
+| `GET /config`, `GET /config/schema`, `GET\|PUT /config/values` | runtime config UI + API |
+
+## Configuration
+
+| Env var | Default | Notes |
+|---------|---------|-------|
+| `META_FEEDER_HTTP_LISTEN` | `0.0.0.0:8080` | HTTP listen address |
+| `META_FEEDER_STATE_DIR` | `/data/meta-feeder` | redb cache + state |
+| `OPENLIBRARY_CONTACT` | — | e-mail or URL sent in the Open Library User-Agent. First-boot seed for the `contact` config field; a value saved in the config UI wins |
+| `OPENLIBRARY_TOP_N` | `12` | cards per search |
+| `OPENLIBRARY_DISCOVERY_N` | `20` | cards per browse row (`popular:true` → Open Library trending, `fresh:true` → recently added) |
+| `RUST_LOG` | `info` | tracing filter |
+
+No upstream API key is required. Open Library asks API clients to identify
+themselves, so set a contact in production: an identified client gets a larger
+request budget. Both upstreams share one Open Library rate budget. The config
+page edits the `gutenberg` upstream, and the contact saved there is read by
+both.
+
+## Image
+
+```
+ghcr.io/worph/meta-feeder-book
+```
+
+Exposes `8080`. Built and pushed by CI on every push to `main` (the `main`
+tag) and on `v*` tags (semver tags).
+
+## Build locally
+
+The build context is the **repo root** (the Cargo workspace).
+`meta-feeder-sdk` is a git dependency pinned by tag, so the builder needs
+network access to github.com:
+
+```bash
+docker build -f feeder-plugin/book-feeder/Dockerfile -t ghcr.io/worph/meta-feeder-book:dev .
+```
+
+Tests (`cargo` is not needed on the host; wiremock stands in for every upstream):
+
+```bash
+docker run --rm -v "$PWD":/src -w /src -e CARGO_TARGET_DIR=/target rust:1.89 \
+  cargo test -p book-feeder
+```
+
+## Repo layout
+
+```
+Cargo.toml                      # workspace: members = feeder-plugin/*
+feeder-plugin/book-feeder/      # this feeder's crate + Dockerfile
+  src/gutenberg.rs              #   `gutenberg` upstream
+  src/openlibrary.rs            #   `openlibrary` upstream (cards)
+  src/openlibrary_client.rs     #   Open Library client (budget, cache, binding lookup)
+crates/meta-feeder-sdk/         # old vendored SDK copy — no longer built (excluded)
+```
+
+`meta-feeder-sdk` comes from [`worph/meta-feeder-sdk`](https://github.com/worph/meta-feeder-sdk)
+at the tag in `feeder-plugin/book-feeder/Cargo.toml`. An SDK change reaches
+this feeder only when it is tagged and the pin is bumped.

crates/meta-feeder-sdk/src/cache.rs

@@ -53,6 +53,13 @@ const SUBTITLES_TABLE: TableDefinition<'_, &str, &str> = TableDefinition::new("s
 /// OpenSubtitles search results keyed by `"<tmdb_id>\x01<lang3>"`. Value is the
 /// resolved subtitle's cid, or the literal `"null"` negative-cache sentinel.
 const OPENSUBTITLES_TABLE: TableDefinition<'_, &str, &str> = TableDefinition::new("opensubtitles");
+/// Small durable plugin scratch values that must survive a restart but aren't
+/// content-derived — today, the torznab plugin's Prowlarr indexer-set
+/// fingerprint, so a set that changed *while the feeder was down* is still
+/// detected as a change on the next boot. Keys are plugin-namespaced
+/// (`prowlarr:indexer-set`). redb creates the table on open, so an older
+/// `.redb` file gains it transparently.
+const MISC_TABLE: TableDefinition<'_, &str, &str> = TableDefinition::new("misc");
 
 /// Per-plugin midhash cache. Cheap to clone (the inner `Database` is shared
 /// via `Arc`).
@@ -105,6 +112,7 @@ impl MidhashCache {
             tx.open_table(TMDB_PRINCIPAL_TOPN_TABLE)?;
             tx.open_table(SUBTITLES_TABLE)?;
             tx.open_table(OPENSUBTITLES_TABLE)?;
+            tx.open_table(MISC_TABLE)?;
             tx.commit()?;
         }
         Ok(Self { db: Arc::new(db) })
@@ -112,6 +120,14 @@ impl MidhashCache {
 
     str_table_accessors!(get_midhash, put_midhash, MIDHASH_TABLE);
 
+    str_table_accessors!(
+        /// Read a durable plugin scratch value (see [`MISC_TABLE`]).
+        get_misc,
+        /// Write a durable plugin scratch value (see [`MISC_TABLE`]).
+        put_misc,
+        MISC_TABLE,
+    );
+
     pub fn entry_count(&self) -> Result<u64, redb::Error> {
         let tx = self.db.begin_read()?;
         let table = tx.open_table(MIDHASH_TABLE)?;

crates/meta-feeder-sdk/src/config_page.html

@@ -17,6 +17,7 @@
   .field { margin: 0 0 16px; }
   .field > label { display: block; font-weight: 600; margin: 0 0 4px; }
   .help { color: #8a8a96; font-size: 12px; margin: 3px 0 0; }
+  .req { color: #f0a0a0; margin-left: 3px; font-weight: 700; }
   input[type=text], input[type=password], input[type=number], textarea {
     width: 100%; padding: 8px 10px; border-radius: 8px;
     border: 1px solid #2a2a35; background: #15151c; color: #e6e6ea; font: inherit;
@@ -46,7 +47,7 @@
 </head>
 <body>
   <h1 id="title">Plugin configuration</h1>
-  <p class="sub">Secrets are write-only — set values are shown as “set”, never echoed back. Saving requires a feeder restart to take effect.</p>
+  <p class="sub">Secrets are write-only — set values are shown as “set”, never echoed back. Saving requires a feeder restart to take effect. Fields marked <span class="req">*</span> are required.</p>
   <div id="banner" class="banner"></div>
   <form id="form"></form>
   <div class="bar" id="bar" style="display:none">
@@ -102,15 +103,24 @@ function linkify(text) {
 
 function helpEl(f) { return f.help ? el("p", { class: "help" }, ...linkify(f.help)) : null; }
 
+// A "*" marker for required fields (null for optional ones).
+function reqStar(f) { return f.required ? el("span", { class: "req", text: "*", title: "required" }) : null; }
+
 // Render one scalar field. `vals` is the object holding the value/`_set` flag.
+// Every returned widget exposes `_collect()` and `_key` so collect() can read it.
 function fieldInput(f, vals) {
   if (f.type === "bool") {
     const cb = el("input", { type: "checkbox" });
     cb.checked = vals[f.key] === true;
-    cb._get = () => cb.checked;
-    return el("label", {}, cb, " ", el("span", { text: " " + f.label }));
+    const w = el("div", { class: "field" },
+      el("label", {}, cb, " ", el("span", { text: " " + f.label }), reqStar(f)),
+      helpEl(f));
+    w._collect = () => cb.checked;
+    w._key = f.key;
+    return w;
   }
   const label = el("label", { text: f.label });
+  const star = reqStar(f); if (star) label.append(star);
   let input;
   if (f.type === "secret") {
     const isSet = vals[f.key + "_set"] === true;
@@ -134,14 +144,6 @@ function fieldInput(f, vals) {
   const wrap = el("div", { class: "field" }, label, input, helpEl(f));
   wrap._collect = () => input._get();
   wrap._key = f.key;
-  // bool path returns a <label>; normalize collect/key for it
-  if (f.type === "bool") {
-    const w = el("div", { class: "field" }, label, helpEl(f));
-    const cb = label.querySelector("input");
-    w._collect = () => cb.checked;
-    w._key = f.key;
-    return w;
-  }
   return wrap;
 }
 
@@ -175,8 +177,9 @@ function recordArray(f, vals) {
   const add = el("button", { type: "button", class: "add", text: "+ add " + f.label.toLowerCase().replace(/s$/, "") });
   add.onclick = () => addRecord({});
 
-  const wrap = el("div", { class: "field" },
-    el("label", { text: f.label }), helpEl(f), list, add);
+  const lbl = el("label", { text: f.label });
+  const star = reqStar(f); if (star) lbl.append(star);
+  const wrap = el("div", { class: "field" }, lbl, helpEl(f), list, add);
   wrap._key = f.key;
   wrap._collect = () => Array.from(list.children).map((c) => c._collect());
   return wrap;
@@ -215,14 +218,34 @@ async function save() {
       headers: { "Content-Type": "application/json" },
       body: JSON.stringify(collect()),
     });
-    if (!res.ok) throw new Error((await res.text()) || res.status);
+    if (!res.ok) {
+      const detail = (await res.text().catch(() => "")).trim();
+      throw new Error(detail ? `the feeder rejected it (HTTP ${res.status}): ${detail}`
+                             : `the feeder returned HTTP ${res.status}`);
+    }
     const j = await res.json().catch(() => ({}));
-    if (j.restart_required) banner("warn", "Saved. Restart the feeder container for changes to take effect.");
-    else banner("ok", "Saved.");
-    // Re-fetch so secret pills reflect the new stored state.
-    await load();
+    if (j.reloading) {
+      // The feeder is restarting itself (~2-3s) to apply the change. Poll it
+      // back up, then refresh the form so secret pills reflect stored state.
+      banner("warn", "Saved — reloading the feeder to apply (~3 s)…");
+      let tries = 6;
+      const poll = async () => {
+        try { await load(); banner("ok", "Reloaded — new config applied."); }
+        catch { if (--tries > 0) setTimeout(poll, 1500); else banner("warn", "Saved & reloading — reopen this panel shortly to confirm."); }
+      };
+      setTimeout(poll, 3000);
+    } else {
+      if (j.restart_required) banner("warn", "Saved. Restart the feeder container for changes to take effect.");
+      else banner("ok", "Saved.");
+      // Re-fetch so secret pills reflect the new stored state.
+      await load();
+    }
   } catch (e) {
-    banner("err", "Save failed: " + e.message);
+    const msg = (e && e.message) ? e.message : String(e);
+    const friendly = /Failed to fetch|NetworkError|load failed/i.test(msg)
+      ? "couldn't reach the feeder — it may be restarting; wait a moment and try again."
+      : msg;
+    banner("err", "Could not save: " + friendly);
   } finally {
     $("save").disabled = false;
   }

crates/meta-feeder-sdk/src/domain.rs

@@ -0,0 +1,549 @@
+//! `contentKind` → `domain` / `workForm` — the routing tables from
+//! METADATA_KEYS.md §1.
+//!
+//! `domain` answers **"which application is this record destined for?"**. It
+//! is the coarsest facet in the registry and the key every client filters its
+//! catalogue on — **one value per app**, no alternation:
+//!
+//! ```text
+//! meta-watch   →  domain:screen
+//! meta-listen  →  domain:music
+//! meta-read    →  domain:literature
+//! ```
+//!
+//! `workForm` answers the *other* question the old `film`/`tv` split was
+//! secretly encoding: **is this work complete in itself, or one instalment of
+//! a run?** (`standalone` | `serial`). It is cross-domain — a comic issue is
+//! `serial` in `literature` exactly as an episode is `serial` in `screen` —
+//! which is what lets a client ask for it independently of the wall.
+//!
+//! # Why they are stored and not derived on read
+//!
+//! `contentKind`'s vocabulary is deliberately open — a new feeder may mint a
+//! kind and ship without a fleet-wide update. That only stays safe because the
+//! domain travels *with* the record: a peer that has never heard of
+//! `podcastEpisode` still routes it correctly, because it never has to
+//! interpret the token. Deriving on read would silently drop every record
+//! whose kind the reader predates. The same argument applies to `workForm`,
+//! plus a second one: the two kinds that most need it (`pack` and `card`) are
+//! exactly the two the kind alone cannot decide.
+//!
+//! # ⚠ The binding is permanent and one-way
+//!
+//! Add kinds freely; **never re-map an existing kind to a different domain**.
+//! Because the value is stored, two peers on different versions that disagree
+//! about where `audiobook` belongs produce a split-brain that redeploying
+//! cannot heal — the wrong domain is already written into records across the
+//! mesh. If a kind genuinely has to move, mint a new kind.
+//!
+//! The `film`+`tv` → `screen` merge (METADATA_KEYS.md §14.17) is the one
+//! deliberate exception, taken once and early and paid for with a one-shot
+//! meta-core corpus sweep. It is not a precedent.
+//!
+//! # Why `domain` and not one kind-key per medium
+//!
+//! `record_matches` is AND-across-keys / OR-within-key, and a non-matching key
+//! fails the record outright. A per-medium split (`videoKind` / `audioKind` /
+//! `documentKind`) would force meta-listen to express "music video **or**
+//! audio track" as an OR across two *different* keys — which the query
+//! language cannot represent at all. One domain key, OR-ed within itself,
+//! always can. `workForm` is one key everywhere for the same reason.
+
+/// The `contentKind` → `domain` table. A kind absent from this list has no
+/// domain and is **not independently routable** — correct for sidecars
+/// (subtitles, artwork) and for un-anchored hits, a producer bug for anything
+/// else.
+///
+/// ⚠ `pack` is deliberately absent. A season pack and an album release are the
+/// same structural thing, so the kind alone cannot tell `screen` from `music`;
+/// the writer stamps `pack`'s domain from its own context (the same reason
+/// `derive_file_type` cannot have a second `pack` arm).
+///
+/// ⚠ `podcast` / `podcastEpisode` are absent too: the `spoken` domain was
+/// retired unused (METADATA_KEYS.md §14.17), and the PR that first *writes*
+/// those kinds picks their domain.
+const DOMAIN_BY_CONTENT_KIND: &[(&str, &str)] = &[
+    // screen — everything meta-watch consumes, films and serials alike. The
+    // standalone/serial split lives in `workForm`, not here.
+    ("movie", "screen"),
+    ("series", "screen"),
+    ("episode", "screen"),
+    // music
+    ("track", "music"),
+    ("album", "music"),
+    ("artist", "music"),
+    ("musicVideo", "music"),
+    ("djMix", "music"),
+    ("liveSet", "music"),
+    // literature — an audiobook's identity is the book (same author, same ISBN
+    // family, same series as the ebook), so meta-read groups the two editions
+    // on one page. The player does not decide the domain.
+    ("book", "literature"),
+    ("audiobook", "literature"),
+    ("comic", "literature"),
+    ("manga", "literature"),
+    ("magazine", "literature"),
+    // science
+    ("paper", "science"),
+];
+
+/// The `contentKind` → `workForm` table (METADATA_KEYS.md §1).
+///
+/// `standalone` = a work complete in itself. `serial` = an instalment of an
+/// ongoing run, **or the container or headline for one** — which is why a
+/// `series` card and an `episode` are both `serial`.
+///
+/// ⚠ `pack` is absent for the same reason it is absent above: a season pack is
+/// `serial`, an album release is `standalone`, and only the writer knows which.
+const WORK_FORM_BY_CONTENT_KIND: &[(&str, &str)] = &[
+    // screen
+    ("movie", "standalone"),
+    ("series", "serial"),
+    ("episode", "serial"),
+    // music — an album is a closed work, not an ongoing run, so every rung of
+    // the music ladder is standalone.
+    ("track", "standalone"),
+    ("album", "standalone"),
+    ("artist", "standalone"),
+    ("musicVideo", "standalone"),
+    ("djMix", "standalone"),
+    ("liveSet", "standalone"),
+    // literature — the split that proves `workForm` is not a screen-only axis.
+    ("book", "standalone"),
+    ("audiobook", "standalone"),
+    ("comic", "serial"),
+    ("manga", "serial"),
+    ("magazine", "serial"),
+    // science
+    ("paper", "standalone"),
+];
+
+/// Resolve the domain a `contentKind` belongs to.
+///
+/// Returns `None` for `pack` (not derivable from the kind — the writer supplies
+/// it), for format/sidecar kinds that route nowhere, and for any kind this
+/// build has never heard of. A caller that already knows the domain from its
+/// own context must prefer that over this table.
+pub fn domain_for_content_kind(content_kind: &str) -> Option<&'static str> {
+    lookup(DOMAIN_BY_CONTENT_KIND, content_kind)
+}
+
+/// Resolve the work form a `contentKind` sits on. Same contract as
+/// [`domain_for_content_kind`]: `None` means "not derivable", never a guess.
+pub fn work_form_for_content_kind(content_kind: &str) -> Option<&'static str> {
+    lookup(WORK_FORM_BY_CONTENT_KIND, content_kind)
+}
+
+fn lookup(table: &'static [(&'static str, &'static str)], key: &str) -> Option<&'static str> {
+    let k = key.trim();
+    table
+        .iter()
+        .find(|(kind, _)| kind.eq_ignore_ascii_case(k))
+        .map(|(_, value)| *value)
+}
+
+/// Kinds whose `domain` the kind→domain table deliberately cannot decide, so
+/// the **writer** supplies it (METADATA_KEYS.md §1). Today that is `pack`
+/// alone: a season pack and an album release are the same structural thing, and
+/// only the producer knows which — indexer-feeder's `bt.rs` reads the domain off
+/// the per-file kind *before* overwriting it with `pack` precisely so the value
+/// survives.
+///
+/// They must ride along in every `domain:` → `contentKind:` expansion. The
+/// expansion is only ever a **pre-filter** — the real `domain:` term is enforced
+/// afterwards by [`crate::query_eval::record_matches`] — so widening it is free,
+/// while omitting these kinds would drop exactly the records whose `domain` is
+/// the most authoritative one on the wire.
+pub const WRITER_SUPPLIED_DOMAIN_KINDS: &[&str] = &["pack"];
+
+/// The `workForm` twin of [`WRITER_SUPPLIED_DOMAIN_KINDS`], and for the same
+/// reason: a season pack is `serial`, an album release is `standalone`.
+pub const WRITER_SUPPLIED_WORK_FORM_KINDS: &[&str] = &["pack"];
+
+/// The inverse: every `contentKind` that belongs to `domain`.
+///
+/// Used at the gateway-routing boundary. A gateway declares the *content kinds*
+/// it serves (`served_content_kinds`), not domains, so a `domain:` filter is
+/// matched by expanding it here rather than by adding a `served_domains` field
+/// to the heartbeat — which would be a wire-protocol change requiring every
+/// feeder manifest and every peer to move at once.
+///
+/// ⚠ Since the `film`+`tv` merge this returns **several** kinds for `screen`
+/// (`movie`, `series`, `episode`) where the two old values were disjoint. A
+/// caller that reads only the first element silently means "movie" — scan the
+/// whole list, or narrow with a `workForm:` term first.
+pub fn content_kinds_for_domain(domain: &str) -> Vec<&'static str> {
+    inverse(DOMAIN_BY_CONTENT_KIND, domain)
+}
+
+/// The inverse of the `workForm` table. Same routing-boundary role, same
+/// scan-the-whole-list warning as [`content_kinds_for_domain`].
+pub fn content_kinds_for_work_form(work_form: &str) -> Vec<&'static str> {
+    inverse(WORK_FORM_BY_CONTENT_KIND, work_form)
+}
+
+fn inverse(table: &'static [(&'static str, &'static str)], value: &str) -> Vec<&'static str> {
+    let v = value.trim();
+    table
+        .iter()
+        .filter(|(_, val)| val.eq_ignore_ascii_case(v))
+        .map(|(kind, _)| *kind)
+        .collect()
+}
+
+/// Rewrite a query so a `domain:` / `workForm:` filter can reach a plugin that
+/// has never heard of either key.
+///
+/// Returns `None` when the query carries neither — the common path, and cheaper
+/// than cloning.
+///
+/// # Why translate instead of teaching every plugin
+///
+/// `record_matches` fails a record outright when a filter key is missing from
+/// its fields, and `domain` is missing from every record a not-yet-updated
+/// feeder emits (`workForm` from every record written before it existed).
+/// Most feeders pin the SDK by git tag, so they cannot pick up a new key on our
+/// schedule. Translating `domain:literature` into the `contentKind:`
+/// alternation it covers means the wire query a plugin sees never mentions
+/// either key — old plugins keep working unchanged, while the dispatcher still
+/// enforces the real filters afterwards, against records it has just stamped.
+///
+/// # The two axes intersect
+///
+/// `domain:screen workForm:serial` covers `series` and `episode` — the
+/// **intersection**, not the union. Widening to the union would hand a serial
+/// query the whole film corpus as a pre-filter; the post-stamp check would then
+/// throw it away, having already paid for it upstream.
+///
+/// An explicit `contentKind:` filter already in the query is left alone: it is
+/// the narrower of the three, and the `domain`/`workForm` terms are still
+/// enforced post-stamp, so the AND semantics hold either way.
+pub fn expand_routing_filters(
+    query: &crate::query::GatewayQuery,
+) -> Option<crate::query::GatewayQuery> {
+    let domains = query.filters.get("domain");
+    let work_forms = query.filters.get("workForm");
+    if domains.is_none() && work_forms.is_none() {
+        return None;
+    }
+
+    let by_domain = domains.map(|vs| expand_values(vs, content_kinds_for_domain));
+    let by_work_form = work_forms.map(|vs| expand_values(vs, content_kinds_for_work_form));
+
+    let mut out = query.clone();
+    out.filters.remove("domain");
+    out.filters.remove("workForm");
+    if out.filters.contains_key("contentKind") {
+        return Some(out);
+    }
+
+    let mut kinds: Vec<&'static str> = match (by_domain, by_work_form) {
+        (Some(d), Some(w)) => d.into_iter().filter(|k| w.contains(k)).collect(),
+        (Some(d), None) => d,
+        (None, Some(w)) => w,
+        (None, None) => unreachable!("guarded above"),
+    };
+    // Domain-ambiguous kinds come along whenever the expansion is non-empty
+    // — see [`WRITER_SUPPLIED_DOMAIN_KINDS`]. Skipped when nothing mapped,
+    // so an unknown value still expands to nothing and leaves the filter
+    // out entirely (below) rather than narrowing to `contentKind:pack`.
+    if !kinds.is_empty() {
+        for k in WRITER_SUPPLIED_DOMAIN_KINDS
+            .iter()
+            .chain(WRITER_SUPPLIED_WORK_FORM_KINDS.iter())
+        {
+            if !kinds.contains(k) {
+                kinds.push(k);
+            }
+        }
+    }
+    // An unknown value — or two axes with no kind in common — expands to
+    // nothing. Inserting an empty alternation would drop every record at the
+    // plugin; leaving the filter out lets the plugin answer and the post-stamp
+    // check do the (correct, empty) filtering.
+    if !kinds.is_empty() {
+        out.filters.insert(
+            "contentKind".to_string(),
+            kinds.into_iter().map(|k| k.to_string()).collect(),
+        );
+    }
+    Some(out)
+}
+
+/// Deprecated alias for [`expand_routing_filters`], kept so a caller pinned to
+/// an older SDK keeps compiling. It expands `workForm:` too — the name is the
+/// only thing that is stale.
+#[deprecated(note = "renamed to expand_routing_filters; it now expands workForm: as well")]
+pub fn expand_domain_filter(
+    query: &crate::query::GatewayQuery,
+) -> Option<crate::query::GatewayQuery> {
+    expand_routing_filters(query)
+}
+
+fn expand_values(
+    values: &[String],
+    expand: fn(&str) -> Vec<&'static str>,
+) -> Vec<&'static str> {
+    let mut out: Vec<&'static str> = Vec::new();
+    for v in values {
+        for k in expand(v) {
+            if !out.contains(&k) {
+                out.push(k);
+            }
+        }
+    }
+    out
+}
+
+/// Stamp `domain` on a field map that already carries a `contentKind`.
+///
+/// The convention from METADATA_KEYS.md §1: **whoever writes `contentKind`
+/// writes `domain` in the same breath**. An existing `domain` is never
+/// overwritten — a writer that knows better than the table (a `pack`, say) has
+/// already set it.
+pub fn stamp_domain(fields: &mut std::collections::BTreeMap<String, String>) {
+    stamp(fields, "domain", domain_for_content_kind);
+}
+
+/// Stamp `workForm` the same way, from the same `contentKind`. Call it
+/// wherever [`stamp_domain`] is called — the registry treats the two as one
+/// breath.
+pub fn stamp_work_form(fields: &mut std::collections::BTreeMap<String, String>) {
+    stamp(fields, "workForm", work_form_for_content_kind);
+}
+
+fn stamp(
+    fields: &mut std::collections::BTreeMap<String, String>,
+    key: &str,
+    derive: fn(&str) -> Option<&'static str>,
+) {
+    if fields.contains_key(key) {
+        return;
+    }
+    let Some(kind) = fields.get("contentKind").cloned() else {
+        return;
+    };
+    if let Some(value) = derive(&kind) {
+        fields.insert(key.to_string(), value.to_string());
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::collections::BTreeMap;
+
+    #[test]
+    fn kinds_map_to_their_domain() {
+        assert_eq!(domain_for_content_kind("movie"), Some("screen"));
+        assert_eq!(domain_for_content_kind("episode"), Some("screen"));
+        assert_eq!(domain_for_content_kind("track"), Some("music"));
+        // The case that proves medium and domain are different axes.
+        assert_eq!(domain_for_content_kind("musicVideo"), Some("music"));
+        // The case that proves the player does not decide the domain.
+        assert_eq!(domain_for_content_kind("audiobook"), Some("literature"));
+    }
+
+    #[test]
+    fn film_and_tv_are_one_domain_now() {
+        // The merge (METADATA_KEYS.md §14.17): meta-watch's wall is one term.
+        assert_eq!(
+            domain_for_content_kind("movie"),
+            domain_for_content_kind("series")
+        );
+        // ...and the split it used to carry lives here instead.
+        assert_eq!(work_form_for_content_kind("movie"), Some("standalone"));
+        assert_eq!(work_form_for_content_kind("series"), Some("serial"));
+        assert_eq!(work_form_for_content_kind("episode"), Some("serial"));
+    }
+
+    #[test]
+    fn work_form_is_not_a_screen_only_axis() {
+        // The point of the key: it means the same thing in every domain.
+        assert_eq!(work_form_for_content_kind("comic"), Some("serial"));
+        assert_eq!(work_form_for_content_kind("book"), Some("standalone"));
+        assert_eq!(work_form_for_content_kind("album"), Some("standalone"));
+    }
+
+    #[test]
+    fn the_spoken_domain_is_gone() {
+        // Retired unused. The kinds stay reserved with no domain until the PR
+        // that writes them picks one.
+        assert_eq!(domain_for_content_kind("podcast"), None);
+        assert_eq!(domain_for_content_kind("podcastEpisode"), None);
+        assert!(content_kinds_for_domain("spoken").is_empty());
+    }
+
+    #[test]
+    fn pack_has_no_derivable_domain_or_work_form() {
+        assert_eq!(domain_for_content_kind("pack"), None);
+        assert_eq!(work_form_for_content_kind("pack"), None);
+    }
+
+    #[test]
+    fn unknown_kind_is_none_not_a_guess() {
+        assert_eq!(domain_for_content_kind("hologram"), None);
+        assert_eq!(work_form_for_content_kind("hologram"), None);
+    }
+
+    #[test]
+    fn expansion_is_the_inverse() {
+        let music = content_kinds_for_domain("music");
+        assert!(music.contains(&"track"));
+        assert!(music.contains(&"musicVideo"));
+        assert!(!music.contains(&"audiobook"));
+        for kind in content_kinds_for_domain("literature") {
+            assert_eq!(domain_for_content_kind(kind), Some("literature"));
+        }
+        for kind in content_kinds_for_work_form("serial") {
+            assert_eq!(work_form_for_content_kind(kind), Some("serial"));
+        }
+    }
+
+    #[test]
+    fn screen_expands_to_every_rung_not_just_the_first() {
+        // The regression the merge creates for any caller reading `.first()`.
+        let screen = content_kinds_for_domain("screen");
+        assert!(screen.contains(&"movie"));
+        assert!(screen.contains(&"series"));
+        assert!(screen.contains(&"episode"));
+    }
+
+    fn q(filters: &[(&str, &[&str])]) -> crate::query::GatewayQuery {
+        crate::query::GatewayQuery {
+            raw_text: String::new(),
+            free_text: String::new(),
+            filters: filters
+                .iter()
+                .map(|(k, vs)| {
+                    (
+                        k.to_string(),
+                        vs.iter().map(|v| v.to_string()).collect::<Vec<_>>(),
+                    )
+                })
+                .collect(),
+            ranges: Vec::new(),
+            negations: Vec::new(),
+        }
+    }
+
+    #[test]
+    fn no_routing_filter_is_left_untouched() {
+        assert!(expand_routing_filters(&q(&[("fileType", &["video"])])).is_none());
+    }
+
+    #[test]
+    fn domain_becomes_the_content_kinds_it_covers() {
+        let out = expand_routing_filters(&q(&[("domain", &["literature"])])).unwrap();
+        // The plugin never sees `domain:` — that is the whole point.
+        assert!(!out.filters.contains_key("domain"));
+        let kinds = out.filters.get("contentKind").unwrap();
+        assert!(kinds.contains(&"book".to_string()));
+        assert!(kinds.contains(&"audiobook".to_string()));
+        assert!(!kinds.contains(&"track".to_string()));
+    }
+
+    #[test]
+    fn work_form_alone_also_expands() {
+        let out = expand_routing_filters(&q(&[("workForm", &["serial"])])).unwrap();
+        assert!(!out.filters.contains_key("workForm"));
+        let kinds = out.filters.get("contentKind").unwrap();
+        assert!(kinds.contains(&"episode".to_string()));
+        assert!(kinds.contains(&"comic".to_string()));
+        assert!(!kinds.contains(&"movie".to_string()));
+    }
+
+    #[test]
+    fn the_two_axes_intersect_they_do_not_union() {
+        let out = expand_routing_filters(&q(&[
+            ("domain", &["screen"][..]),
+            ("workForm", &["serial"][..]),
+        ]))
+        .unwrap();
+        let kinds = out.filters.get("contentKind").unwrap();
+        assert!(kinds.contains(&"series".to_string()));
+        assert!(kinds.contains(&"episode".to_string()));
+        // The union would have dragged the whole film corpus in.
+        assert!(!kinds.contains(&"movie".to_string()));
+        // A `serial` comic is in `literature`, not on meta-watch's wall.
+        assert!(!kinds.contains(&"comic".to_string()));
+    }
+
+    #[test]
+    fn an_explicit_content_kind_filter_wins() {
+        let out = expand_routing_filters(&q(&[
+            ("domain", &["music"][..]),
+            ("contentKind", &["track"][..]),
+        ]))
+        .unwrap();
+        assert_eq!(
+            out.filters.get("contentKind").map(Vec::as_slice),
+            Some(&["track".to_string()][..])
+        );
+    }
+
+    #[test]
+    fn expansion_carries_the_writer_supplied_kinds() {
+        // `pack` has no domain in the table (a season pack and an album are the
+        // same structural thing), so the writer stamps one. Leaving it out of
+        // the alternation would make a `domain:screen` pre-filter drop the whole
+        // torrent-pack corpus, which `record_matches` would then never see.
+        let out = expand_routing_filters(&q(&[("domain", &["screen"])])).unwrap();
+        let kinds = out.filters.get("contentKind").unwrap();
+        assert!(kinds.contains(&"episode".to_string()));
+        assert!(kinds.contains(&"pack".to_string()));
+        // ...and it rides along on the `workForm` axis too, exactly once.
+        let out = expand_routing_filters(&q(&[("workForm", &["serial"])])).unwrap();
+        let kinds = out.filters.get("contentKind").unwrap();
+        assert_eq!(kinds.iter().filter(|k| *k == "pack").count(), 1);
+    }
+
+    #[test]
+    fn unknown_value_adds_no_alternation() {
+        // An empty alternation would drop every record at the plugin; the
+        // post-stamp check does the (correct, empty) filtering.
+        let out = expand_routing_filters(&q(&[("domain", &["hologram"])])).unwrap();
+        // Not even `pack`: a lone `contentKind:pack` would be a NARROWING of a
+        // filter that should widen to nothing.
+        assert!(!out.filters.contains_key("contentKind"));
+    }
+
+    #[test]
+    fn two_axes_with_nothing_in_common_widen_rather_than_narrow() {
+        // No music kind is `serial`. The pre-filter drops out; the post-stamp
+        // check still enforces both terms.
+        let out = expand_routing_filters(&q(&[
+            ("domain", &["music"][..]),
+            ("workForm", &["serial"][..]),
+        ]))
+        .unwrap();
+        assert!(!out.filters.contains_key("contentKind"));
+    }
+
+    #[test]
+    fn stamp_fills_in_and_never_overwrites() {
+        let mut f: BTreeMap<String, String> = BTreeMap::new();
+        f.insert("contentKind".into(), "episode".into());
+        stamp_domain(&mut f);
+        stamp_work_form(&mut f);
+        assert_eq!(f.get("domain").map(String::as_str), Some("screen"));
+        assert_eq!(f.get("workForm").map(String::as_str), Some("serial"));
+
+        // A pack writer that already knows its context keeps it.
+        let mut p: BTreeMap<String, String> = BTreeMap::new();
+        p.insert("contentKind".into(), "pack".into());
+        p.insert("domain".into(), "music".into());
+        p.insert("workForm".into(), "standalone".into());
+        stamp_domain(&mut p);
+        stamp_work_form(&mut p);
+        assert_eq!(p.get("domain").map(String::as_str), Some("music"));
+        assert_eq!(p.get("workForm").map(String::as_str), Some("standalone"));
+
+        // No contentKind → nothing invented.
+        let mut n: BTreeMap<String, String> = BTreeMap::new();
+        stamp_domain(&mut n);
+        stamp_work_form(&mut n);
+        assert!(n.get("domain").is_none());
+        assert!(n.get("workForm").is_none());
+    }
+}

crates/meta-feeder-sdk/src/enrich.rs

@@ -73,16 +73,40 @@ fn env_nonempty(key: &str) -> Option<String> {
         .filter(|s| !s.is_empty())
 }
 
+/// Derive a co-bundled sidecar plugin URL from the feeder's own hostname, by the
+/// app-compose convention `<prefix>-feeder` → `<prefix>-<plugin>:8080`
+/// (e.g. `metafeedertorznab-feeder` → `http://metafeedertorznab-tmdb:8080`). Lets
+/// a feeder reach its bundled filename-parser / tmdb sidecars with **no explicit
+/// `*_PLUGIN_URL` env** — config lives in the UI, wiring is derived. Returns
+/// `None` when `HOSTNAME` is unset or not in `<prefix>-feeder` form, in which case
+/// the caller falls back to the env seed (then soft-skips that plugin).
+fn derive_sidecar_url(plugin: &str) -> Option<String> {
+    let host = env_nonempty("HOSTNAME")?;
+    let prefix = host.strip_suffix("-feeder")?;
+    Some(format!("http://{prefix}-{plugin}:8080"))
+}
+
 impl EnrichmentConfig {
     /// Parse from the feeder env. Returns `None` when `META_CORE_URL` is unset —
-    /// the feeder then falls back to the legacy "return record, gateway stores"
-    /// path (backward compatible; used by tests and gateway-stores deployments).
+    /// the **bundled-feeder** path (records returned over HTTP, the gateway
+    /// dispatcher stores them). Self-publishing feeders that source meta-core from
+    /// their own dashboard config use [`from_meta_core`](Self::from_meta_core).
     pub fn from_env() -> Option<Self> {
-        let meta_core_url = env_nonempty("META_CORE_URL")?;
-        Some(EnrichmentConfig {
+        Some(Self::from_meta_core(env_nonempty("META_CORE_URL")?))
+    }
+
+    /// Build with an explicit meta-core root — sourced from the feeder's own
+    /// `config.json` (dashboard) when set, else the `META_CORE_URL` env seed. The
+    /// sidecar URLs come from their `*_PLUGIN_URL` env, falling back to the
+    /// hostname-derived co-bundled sidecar (see [`derive_sidecar_url`]); the rest
+    /// keep their env seeds / defaults. Secrets (tmdb token, language) are
+    /// overlaid by the caller from the per-feeder config.
+    pub fn from_meta_core(meta_core_url: String) -> Self {
+        EnrichmentConfig {
             meta_core_url,
-            filename_parser_url: env_nonempty("FILENAME_PARSER_URL"),
-            tmdb_url: env_nonempty("TMDB_PLUGIN_URL"),
+            filename_parser_url: env_nonempty("FILENAME_PARSER_URL")
+                .or_else(|| derive_sidecar_url("filename-parser")),
+            tmdb_url: env_nonempty("TMDB_PLUGIN_URL").or_else(|| derive_sidecar_url("tmdb")),
             tmdb_token: env_nonempty("TMDB_TOKEN").or_else(|| env_nonempty("TORZNAB_TMDB_TOKEN")),
             tmdb_language: env_nonempty("TMDB_LANGUAGE").unwrap_or_else(|| "en-US".to_string()),
             gateway_peer: env_nonempty("META_GATEWAY_PEER_ID")
@@ -96,7 +120,7 @@ impl EnrichmentConfig {
             opensubtitles_password: env_nonempty("OPENSUBTITLES_PASSWORD"),
             opensubtitles_languages: env_nonempty("OPENSUBTITLES_LANGUAGES")
                 .unwrap_or_else(|| "en".to_string()),
-        })
+        }
     }
 }
 

crates/meta-feeder-sdk/src/filename_meta.rs

@@ -57,9 +57,9 @@ pub struct SeasonEpisode {
     /// is deliberately **not** given a single `episode`.
     pub is_pack: bool,
     /// True when [`season`] came from an explicit season-bearing token
-    /// (`S01E05`, `Season 2`, `S03`), as opposed to the "episode with no
-    /// season → assume 1" default. Lets a per-file reparse know whether to
-    /// borrow the season from the pack/parent title (see
+    /// (`S01E05`, `Season 2`, `S03`). An episode with no such token is left
+    /// season-less (no "assume season 1" default), so a per-file reparse uses
+    /// this to know it should borrow the season from the pack/parent title (see
     /// [`extract_season_episode_with_context`]).
     pub season_explicit: bool,
 }
@@ -550,21 +550,22 @@ pub fn extract_season_episode(title: &str) -> SeasonEpisode {
 
     let mut out = SeasonEpisode::default();
     // season/episode 0 are valid; only < 0 means "not found".
-    let mut sea_eff = sea;
     if sea >= 0.0 {
         out.season = Some(fmt_num(sea));
         out.season_explicit = true;
     }
     if ep >= 0.0 {
         out.episode = Some(fmt_num(ep));
-        if sea_eff < 0.0 {
-            // An episode with no season — assume season 1 (matches TS). The
-            // season is *not* explicit, so a per-file reparse may override it
-            // from the parent (pack) title.
-            sea_eff = 1.0;
-            out.season = Some("1".to_string());
+        // An episode with no season is left season-LESS on purpose: a season is
+        // only asserted when explicitly hinted — an S-token in the title here,
+        // or (via extract_season_episode_with_context) the parent/pack title.
+        // Absolute-numbered releases ("Show - 28") thus carry an episode and no
+        // season, so the UI buckets them under "Episodes" rather than a
+        // fabricated "Season 1". `increment` encodes a season, so it is only
+        // emitted when the season is known.
+        if sea >= 0.0 {
+            out.increment = Some(fmt_num(sea * 10000.0 + ep));
         }
-        out.increment = Some(fmt_num(sea_eff * 10000.0 + ep));
     } else if sea >= 0.0 {
         // Season present, no episode → a whole-season pack (`Season 3`, `S03`).
         out.is_pack = true;
@@ -583,8 +584,8 @@ pub fn extract_season_episode(title: &str) -> SeasonEpisode {
 pub fn extract_season_episode_with_context(parent: &str, filename: &str) -> SeasonEpisode {
     let mut file_se = extract_season_episode(filename);
     // Only borrow the parent's season for a concrete single episode whose own
-    // filename carries no *explicit* season token (so its season is the assumed
-    // `1` default). A per-file record is an episode, not a pack, so we don't
+    // filename carries no *explicit* season token (so it is otherwise
+    // season-less). A per-file record is an episode, not a pack, so we don't
     // propagate the parent's pack/range shape here.
     if file_se.episode.is_some() && !file_se.season_explicit {
         if let Some(s) = find_pack_season(&preclean(parent)) {
@@ -1004,12 +1005,13 @@ mod tests {
     }
 
     #[test]
-    fn anime_trailing_number_assumes_season_one() {
-        // No season marker → episode via rightmost-number fallback,
-        // season assumed 1. SFV id stripped, brackets ignored.
+    fn anime_trailing_number_is_season_less() {
+        // No season marker → episode via rightmost-number fallback, and NO
+        // season is fabricated (absolute numbering). SFV id stripped, brackets
+        // ignored. The UI buckets a season-less episode under "Episodes".
         assert_eq!(
             se("[SubsPlease] Some Anime - 117 [FC412C51]"),
-            (Some("1".into()), Some("117".into()))
+            (None, Some("117".into()))
         );
     }
 
@@ -1184,8 +1186,8 @@ mod tests {
     #[test]
     fn per_file_context_recovers_season_from_parent() {
         // Opened pack: the inner file carries the episode, the pack title the
-        // season. `Frieren - 03.mkv` alone assumes S1; with the S2 pack parent
-        // it resolves to S2E3.
+        // season. `Frieren - 03.mkv` alone is season-less; with the S2 pack
+        // parent it resolves to S2E3 (the parent is the "season hint").
         let r = extract_season_episode_with_context("Frieren S2 (01-10) [Batch]", "Frieren - 03.mkv");
         assert_eq!(r.season.as_deref(), Some("2"));
         assert_eq!(r.episode.as_deref(), Some("3"));
@@ -1198,12 +1200,14 @@ mod tests {
     }
 
     #[test]
-    fn absolute_numbered_anime_still_assumes_season_one() {
-        // No season keyword → the rightmost-number fallback still fires, season
-        // assumed 1 (Phase 3 remaps this against TMDB episode counts later).
+    fn absolute_numbered_anime_is_season_less() {
+        // No season keyword → the rightmost-number fallback still finds the
+        // episode, but NO season is fabricated. Season is only ever asserted
+        // from an explicit token (title or parent), never inferred — so an
+        // absolute-numbered release lands in the UI's "Episodes" bucket.
         assert_eq!(
             se("[Naruto-Kun.Hu] Spy x Family - 37 [1080p].mkv"),
-            (Some("1".into()), Some("37".into()))
+            (None, Some("37".into()))
         );
     }
 

crates/meta-feeder-sdk/src/hash.rs

@@ -160,6 +160,366 @@ fn bt_file_cid(codec: u64, infohash: &[u8], file_index: u64) -> String {
     format!("b{}", base32_lower_no_padding(&wire))
 }
 
+/// Custom multicodec for a self-describing **Newznab release** locator
+/// (`nzb-release`). MetaMesh-private, adjacent to the torrent-file codecs in the
+/// `0x10xx` range. Unlike a content hash, the cid *embeds* the indexer host +
+/// release id in an identity multihash, so the credentialed meta-share peer
+/// decodes `{host,id}` straight from the cid and grabs the `.nzb` via `t=get`
+/// only at playback — no side-table, no `.nzb` download at search time. (The
+/// `0x1004` `sha256(host‖id)` opaque hash and the `0x1003` `nzb-posting`
+/// variant were both removed; only this self-describing form is emitted.)
+pub const NZB_RELEASE_CODEC: u64 = 0x1005;
+
+/// Digest byte ceiling for an `nzb-release` cid, matching meta-share's
+/// `MAX_MULTIHASH_SIZE` — its `CidGeneric<64>` rejects a longer multihash.
+const NZB_RELEASE_MAX_DIGEST: usize = 64;
+
+/// Encode a self-describing `nzb-release` CID from a Newznab indexer's host +
+/// bare release id (the hex `<guid>` id). The multihash is *identity* (`0x00`)
+/// and its digest embeds the locator as `varint(host_len) ‖ host ‖ id_bytes`,
+/// where `id_bytes` is the hex-decoded id. meta-share's `decode_nzb_release_cid`
+/// is the exact inverse. Host-namespaced: the same release on two indexers
+/// yields different cids.
+///
+/// Returns `None` when the id isn't hex or the digest would overflow the
+/// multihash budget (an oversized host) — the caller drops that row.
+pub fn compute_nzb_release_cid(api_base: &str, release_id: &str) -> Option<String> {
+    let id_bytes = decode_hex_id(release_id)?;
+    let host_b = api_base.as_bytes();
+
+    // identity-multihash digest = varint(host_len) ‖ host ‖ id_bytes
+    let mut digest = Vec::with_capacity(2 + host_b.len() + id_bytes.len());
+    write_pb_varint(host_b.len() as u64, &mut digest);
+    digest.extend_from_slice(host_b);
+    digest.extend_from_slice(&id_bytes);
+    if digest.len() > NZB_RELEASE_MAX_DIGEST {
+        return None;
+    }
+
+    // CIDv1: [version=0x01][codec varint][mh code=0x00 identity][len varint][digest]
+    let mut wire = Vec::with_capacity(1 + 3 + 2 + digest.len());
+    wire.push(0x01);
+    write_pb_varint(NZB_RELEASE_CODEC, &mut wire);
+    wire.push(0x00); // multihash code: identity
+    write_pb_varint(digest.len() as u64, &mut wire);
+    wire.extend_from_slice(&digest);
+    Some(format!("b{}", base32_lower_no_padding(&wire)))
+}
+
+/// Hex-decode a Newznab release id. The guid parser only keeps hex runs, so the
+/// id is all-hex by construction; odd-length ids are left-padded with a `0`
+/// nibble. Returns `None` on any non-hex byte (defensive).
+fn decode_hex_id(id: &str) -> Option<Vec<u8>> {
+    let padded;
+    let s: &str = if id.len() % 2 == 1 {
+        padded = format!("0{id}");
+        &padded
+    } else {
+        id
+    };
+    let bytes = s.as_bytes();
+    let mut out = Vec::with_capacity(bytes.len() / 2);
+    for pair in bytes.chunks(2) {
+        let hi = (pair[0] as char).to_digit(16)?;
+        let lo = (pair[1] as char).to_digit(16)?;
+        out.push(((hi << 4) | lo) as u8);
+    }
+    Some(out)
+}
+
+/// Custom multicodec for a **card locator** — the identity of a *work* (a
+/// series, a film) as published by a metadata bridge, rather than of any
+/// bytes. MetaMesh-private, the next free slot after `0x1006` url. (`0x1004`
+/// nzb-release-hash stays **retired — do not reuse**. `0x1003` was retired too,
+/// but has since been **revived** for the self-scanned Usenet posting identity —
+/// see [`NZB_POSTING_CODEC`] and `METADATA_KEYS.md` §2.)
+///
+/// This is the first CID family in the platform where **no bytes exist
+/// anywhere, ever** — `0x1005`/`0x1006` are locators too, but both eventually
+/// resolve to bytes (NNTP, HTTP), and `0x1000` midhash256 is a digest over
+/// real file bytes. A card is pure identity: the record *is* the payload, and
+/// it lives in meta-core. Consequences, all already handled:
+///
+/// * Bitswap seeding is gated on `HashKind::Sha2_256` (gateway invariant 6),
+///   so a card is never offered to the swarm — nothing to seed.
+/// * meta-share's byte path 404s for it, which is correct.
+/// * It ranks `RANK_LOCATOR` (5), so it can never outrank a real digest for
+///   `canonical_cid` — which is what makes the future "one card, two CIDs"
+///   merge (a TMDB and a MyAnimeList locator on one meta-core record) safe.
+/// The `url` locator codec (`0x1006`).
+///
+/// ⚠ Back-ported into this vendored SDK copy because the shared crate has it and
+/// this one predates it. Keep it byte-identical to
+/// `meta-feeder-sdk::hash::compute_url_cid`: it is how a record gets an address
+/// without the feeder ever fetching bytes, and two encoders disagreeing would
+/// mint two cids for one URL.
+pub const URL_LOCATOR_CODEC: u64 = 0x1006;
+
+/// Encode an `http(s)` URL as a `0x1006` identity-multihash CIDv1.
+///
+/// The digest IS the URL bytes, so this costs no I/O and no lookup — which is
+/// what makes it usable at search time, on every record, before anything has
+/// been fetched. Resolution happens later, on whichever peer wants the bytes.
+///
+/// `None` for anything that is not an absolute http(s) URL: a relative or
+/// malformed value would encode fine and then fail unresolvably at read time,
+/// which is a far more expensive way to find out.
+pub fn compute_url_cid(url: &str) -> Option<String> {
+    let u = url.trim();
+    if !(u.starts_with("http://") || u.starts_with("https://")) {
+        return None;
+    }
+    let digest = u.as_bytes();
+    let mut wire = Vec::with_capacity(1 + 3 + 2 + digest.len());
+    wire.push(0x01); // CIDv1
+    write_pb_varint(URL_LOCATOR_CODEC, &mut wire);
+    wire.push(0x00); // multihash code: identity
+    write_pb_varint(digest.len() as u64, &mut wire);
+    wire.extend_from_slice(digest);
+    Some(format!("b{}", base32_lower_no_padding(&wire)))
+}
+
+pub const CARD_LOCATOR_CODEC: u64 = 0x1007;
+
+/// Digest byte ceiling for a card locator, matching meta-share's
+/// `MAX_MULTIHASH_SIZE` — its `CidGeneric<64>` rejects a longer multihash.
+const CARD_LOCATOR_MAX_DIGEST: usize = 64;
+
+/// Encode a self-describing **card locator** CID from a metadata source and
+/// that source's own id for the work (`("tmdb", "tv:95479")`,
+/// `("mal", "52991")`). The multihash is *identity* (`0x00`) and its digest
+/// embeds the locator as `varint(source_len) ‖ source ‖ id`, both UTF-8 —
+/// exactly the shape [`compute_nzb_release_cid`] uses, minus the hex decode
+/// (a card id is arbitrary text, not a hex release id).
+///
+/// **The CID is a pure function of `(source, id)`**, which is the whole point:
+/// any peer that knows a tmdb id can derive the card's address offline and
+/// look it up locally, with no discovery round-trip. It also means the same
+/// card discovered by two different peers converges on one meta-core record
+/// for free, via the `cids/<bareCid>` reverse index.
+///
+/// Source-namespaced: the same work on TMDB and on MyAnimeList yields
+/// different cids. Merging those two into one record is a meta-core alias
+/// operation driven by an explicit cross-source id mapping — never inferred
+/// here (see `meta-gateway/docs/others/card-tier-search.md` §9).
+///
+/// Returns `None` when `source` is empty or the digest would overflow the
+/// multihash budget — the caller drops that card.
+pub fn compute_card_cid(source: &str, id: &str) -> Option<String> {
+    if source.is_empty() || id.is_empty() {
+        return None;
+    }
+    let source_b = source.as_bytes();
+    let id_b = id.as_bytes();
+
+    // identity-multihash digest = varint(source_len) ‖ source ‖ id
+    let mut digest = Vec::with_capacity(2 + source_b.len() + id_b.len());
+    write_pb_varint(source_b.len() as u64, &mut digest);
+    digest.extend_from_slice(source_b);
+    digest.extend_from_slice(id_b);
+    if digest.len() > CARD_LOCATOR_MAX_DIGEST {
+        return None;
+    }
+
+    // CIDv1: [version=0x01][codec varint][mh code=0x00 identity][len varint][digest]
+    let mut wire = Vec::with_capacity(1 + 3 + 2 + digest.len());
+    wire.push(0x01);
+    write_pb_varint(CARD_LOCATOR_CODEC, &mut wire);
+    wire.push(0x00); // multihash code: identity
+    write_pb_varint(digest.len() as u64, &mut wire);
+    wire.extend_from_slice(&digest);
+    Some(format!("b{}", base32_lower_no_padding(&wire)))
+}
+
+// ---------------------------------------------------------------------------
+// Delegated-playback locators — `yt-video` (0x1008) and `ext-play` (0x1009)
+// ---------------------------------------------------------------------------
+//
+// These two are the limit case *beyond* `card` (0x1007). `card` addresses a
+// work with no bytes anywhere; these address a **rendition whose bytes exist
+// but are permanently someone else's**. An external player renders the media,
+// nothing is transported, nothing is content-addressed, nothing is re-seedable.
+//
+// See `docs/cid-formats.md` §7 and
+// `docs/study/listenbrainz-youtube-playback-tier-2026-08-21.md`.
+
+/// Custom multicodec for a **YouTube rendition** — delegated playback.
+///
+/// Its own codec rather than a `url` (`0x1006`) for three reasons, in
+/// descending order of how much trouble the alternative causes:
+///
+/// * **`0x1006` means "fetch these bytes once and seed them".** Point a `url`
+///   locator at a `watch` page and meta-share does exactly that: fetches the
+///   HTML, chunks it, seeds it, and links the resulting content CID back onto
+///   the record *as the file*. That is the MSR1 poisoning shape, except
+///   permanent and replicated across peers. A resolved `googlevideo` URL is no
+///   better — time-limited and IP-bound, so it is not a locator at all.
+/// * **The id is canonical, a URL is not.** `youtube.com` / `music.youtube.com`
+///   / `youtu.be` all address the same video, and query parameters are tracking
+///   noise, so the same rendition would mint several different CIDs. Being a
+///   pure function of `(kind, id)`, this codec makes two peers that resolve the
+///   same track converge on one meta-core record for free, via the
+///   `cids/<bareCid>` reverse index — the property that makes [`compute_card_cid`]
+///   useful, for the same reason.
+/// * **It is tiny** — 11 bytes of id plus a short kind prefix.
+///
+/// ⚠ **An immutable id is not durable playback.** Takedowns, geo-blocks and
+/// embed-disabled uploads all kill a reference whose CID stays valid forever —
+/// the same rot class as [`NZB_RELEASE_CODEC`]. The mitigation is at the
+/// *record* level (carry several ranked references per recording so a client
+/// can fall through), never at the CID level.
+pub const YT_VIDEO_CODEC: u64 = 0x1008;
+
+/// Custom multicodec for an **external page to open, never fetch**.
+///
+/// Wire-identical to `url` (`0x1006`) apart from the codec, and that difference
+/// is the entire contract: `0x1006` resolves cache-through and lands in the
+/// blockstore, `0x1009` **must be refused by the byte path** and only ever
+/// surfaces as an "Open on …↗" affordance. Use it for anything worth linking
+/// but not worth a per-provider codec (Bandcamp, Jamendo, an arbitrary stream
+/// page).
+pub const EXT_PLAY_CODEC: u64 = 0x1009;
+
+/// Encode a [`YT_VIDEO_CODEC`] locator from a `kind` and a bare YouTube id.
+///
+/// `kind` is `"video"` | `"playlist"` | `"channel"`; `id` is the bare
+/// identifier (`4D7u5KF7SP8`, `OLAK5uy_…`, `UCRr1xG_2WIDs18a6cIiCxeA`) — **not**
+/// a URL, and not a `watch?v=` fragment. The framing is byte-identical to
+/// [`compute_card_cid`] on purpose, so each of the seven rank/decode
+/// implementations extends a parser it already wrote rather than learning a
+/// novel shape.
+///
+/// Returns `None` on an empty field or a digest past the multihash budget —
+/// the caller drops that reference.
+pub fn compute_yt_video_cid(kind: &str, id: &str) -> Option<String> {
+    kinded_locator_cid(YT_VIDEO_CODEC, kind, id)
+}
+
+/// Encode an [`EXT_PLAY_CODEC`] locator from an `http(s)` URL.
+///
+/// The scheme check is not cosmetic: it is what keeps a `javascript:` or
+/// `data:` payload from ever reaching a client that will hand this string to
+/// an anchor or a new tab.
+pub fn compute_ext_play_cid(url: &str) -> Option<String> {
+    if !(url.starts_with("http://") || url.starts_with("https://")) {
+        return None;
+    }
+    let digest = url.as_bytes();
+    if digest.is_empty() || digest.len() > CARD_LOCATOR_MAX_DIGEST {
+        return None;
+    }
+    Some(identity_locator_cid(EXT_PLAY_CODEC, digest))
+}
+
+/// Shared encoder for the `varint(len(kind)) ‖ kind ‖ id` locator framing used
+/// by [`compute_card_cid`] and [`compute_yt_video_cid`].
+fn kinded_locator_cid(codec: u64, kind: &str, id: &str) -> Option<String> {
+    if kind.is_empty() || id.is_empty() {
+        return None;
+    }
+    let kind_b = kind.as_bytes();
+    let id_b = id.as_bytes();
+
+    let mut digest = Vec::with_capacity(2 + kind_b.len() + id_b.len());
+    write_pb_varint(kind_b.len() as u64, &mut digest);
+    digest.extend_from_slice(kind_b);
+    digest.extend_from_slice(id_b);
+    if digest.len() > CARD_LOCATOR_MAX_DIGEST {
+        return None;
+    }
+    Some(identity_locator_cid(codec, &digest))
+}
+
+/// CIDv1 with an *identity* multihash:
+/// `[version=0x01][codec varint][mh=0x00][len varint][digest]`, base32lower.
+fn identity_locator_cid(codec: u64, digest: &[u8]) -> String {
+    let mut wire = Vec::with_capacity(1 + 3 + 2 + digest.len());
+    wire.push(0x01);
+    write_pb_varint(codec, &mut wire);
+    wire.push(0x00); // multihash code: identity
+    write_pb_varint(digest.len() as u64, &mut wire);
+    wire.extend_from_slice(digest);
+    format!("b{}", base32_lower_no_padding(&wire))
+}
+
+/// Custom multicodec for a **Usenet posting** identity (`nzb-posting`),
+/// minted from the article Message-IDs of a release we scanned ourselves.
+///
+/// Unlike [`NZB_RELEASE_CODEC`] (`0x1005`), this is **not** a locator: it
+/// embeds no indexer host, so any peer with a plain NNTP provider can fetch the
+/// articles — no `IndexerCred`, no `t=get` grab. That portability is the whole
+/// point of running our own header scanner.
+///
+/// It is also **not** fetchable by cid: the digest is over the Message-ID set,
+/// not over any block's bytes, so bitswap can never satisfy a want for it (the
+/// receiver derives a block's cid by hashing the block — see
+/// `beetswap::incoming_stream`). The `.nzb` manifest travels separately, as an
+/// ordinary sha2-256 IPFS cid in the record's `manifest` field. See
+/// `meta-gateway/docs/others/self-hosted-usenet-indexer-study.md` §5.
+pub const NZB_POSTING_CODEC: u64 = 0x1003;
+
+/// Mint an `nzb-posting` cid (`0x1003`) from a release's article Message-IDs.
+///
+/// # The normalisation rule IS the dedup contract
+///
+/// Mirrored verbatim in meta-share (`nzb/manifest.rs`), which re-derives this
+/// digest from the fetched manifest and rejects a mismatch. **Any change here
+/// must land on both sides together** — a drift silently makes every posting
+/// unplayable. The rule:
+///
+/// 1. Trim each id, and strip a surrounding `<…>` pair if present (NZB
+///    `<segment>` bodies carry the bare form, but be defensive — NNTP `BODY`
+///    re-adds the brackets).
+/// 2. Drop empties.
+/// 3. Deduplicate, then **sort** lexicographically by bytes.
+/// 4. Join with `\n` and SHA-256 the result.
+///
+/// Case is **preserved**: a Message-ID's local part is case-sensitive per RFC
+/// 5322, so lowercasing could collide two genuinely distinct articles.
+///
+/// Sorting is what makes the cid survive segment/file reordering between two
+/// NZBs of the same posting — the reason we hash the *id set* and never the raw
+/// `.nzb` bytes (the retired `0x1004` did the latter and broke dedup; see
+/// meta-share `nzb/mod.rs`).
+///
+/// **v1 scope:** every segment counts, par2 volumes included. Two NZBs for one
+/// posting that disagree about whether to bundle par2 therefore mint different
+/// cids. Acceptable while we are the only producer of these manifests; if a
+/// second producer ever appears, the fix is to filter par2 subjects here *and*
+/// in meta-share's mirror — do not do it on one side only.
+pub fn compute_nzb_posting_cid<S: AsRef<str>>(message_ids: &[S]) -> String {
+    let mut ids: Vec<&str> = message_ids
+        .iter()
+        .map(|s| {
+            let t = s.as_ref().trim();
+            t.strip_prefix('<')
+                .and_then(|r| r.strip_suffix('>'))
+                .unwrap_or(t)
+        })
+        .filter(|s| !s.is_empty())
+        .collect();
+    ids.sort_unstable();
+    ids.dedup();
+
+    let mut hasher = Sha256::new();
+    for (i, id) in ids.iter().enumerate() {
+        if i > 0 {
+            hasher.update(b"\n");
+        }
+        hasher.update(id.as_bytes());
+    }
+    let digest: [u8; 32] = hasher.finalize().into();
+
+    // CIDv1: [version=0x01][codec varint][mh code=0x12 sha2-256][len=32][digest]
+    let mut wire = Vec::with_capacity(1 + 3 + 2 + digest.len());
+    wire.push(0x01);
+    write_pb_varint(NZB_POSTING_CODEC, &mut wire);
+    wire.push(0x12); // multihash code: sha2-256
+    write_pb_varint(digest.len() as u64, &mut wire);
+    wire.extend_from_slice(&digest);
+    format!("b{}", base32_lower_no_padding(&wire))
+}
+
 /// Compute a standard IPFS CIDv1 over `bytes`. Output matches kubo's
 /// `ipfs add --cid-version=1 --raw-leaves=true --chunker=size-262144
 /// --hash=sha2-256 <file>`.
@@ -456,6 +816,82 @@ mod tests {
         assert_eq!(base32_lower_no_padding(b"foobar"), "mzxw6ytboi");
     }
 
+    #[test]
+    fn yt_video_cid_matches_golden_vector() {
+        // Pinned by `/cid-rank-vectors.json`'s `yt-video-locator` entry. The id
+        // is the real resolver output for Daft Punk / "Get Lucky" measured in
+        // the delegated-playback study §4.3.
+        assert_eq!(
+            compute_yt_video_cid("video", "4D7u5KF7SP8").unwrap(),
+            "bagecaaarav3gszdfn42ein3vgvfumn2tka4a"
+        );
+    }
+
+    #[test]
+    fn yt_video_cid_framing_matches_card_locator() {
+        // The two codecs share the `varint(len(kind)) || kind || id` framing on
+        // purpose, so every decoder extends a parser it already wrote. If this
+        // ever fails, the two shapes have drifted and §7.2 of docs/cid-formats.md
+        // is no longer true.
+        let yt = compute_yt_video_cid("tmdb", "tv:95479").unwrap();
+        let card = compute_card_cid("tmdb", "tv:95479").unwrap();
+        // Same digest, different codec => same length, differing only in the
+        // codec varint region.
+        assert_ne!(yt, card);
+        assert_eq!(yt.len(), card.len());
+    }
+
+    #[test]
+    fn yt_video_cid_is_a_pure_function_of_kind_and_id() {
+        // Convergence is the reason this is a codec and not a URL: two peers
+        // that resolve the same video must derive the same address offline.
+        assert_eq!(
+            compute_yt_video_cid("video", "4D7u5KF7SP8"),
+            compute_yt_video_cid("video", "4D7u5KF7SP8")
+        );
+        // ...and the kind is part of the identity, not decoration.
+        assert_ne!(
+            compute_yt_video_cid("video", "OLAK5uy_kZ8Xq"),
+            compute_yt_video_cid("playlist", "OLAK5uy_kZ8Xq")
+        );
+    }
+
+    #[test]
+    fn yt_video_cid_rejects_empty_fields() {
+        assert!(compute_yt_video_cid("", "4D7u5KF7SP8").is_none());
+        assert!(compute_yt_video_cid("video", "").is_none());
+    }
+
+    #[test]
+    fn ext_play_cid_matches_golden_vector() {
+        assert_eq!(
+            compute_ext_play_cid("https://example.bandcamp.com/album/x").unwrap(),
+            "bagesaabenb2hi4dthixs6zlymfwxa3dffzrgc3temnqw24bomnxw2l3bnrrhk3jppa"
+        );
+    }
+
+    #[test]
+    fn ext_play_cid_rejects_non_http_schemes() {
+        // Load-bearing: a client hands this string to an anchor or a new tab,
+        // so a `javascript:` or `data:` payload must never mint a locator.
+        assert!(compute_ext_play_cid("javascript:alert(1)").is_none());
+        assert!(compute_ext_play_cid("data:text/html,<script>").is_none());
+        assert!(compute_ext_play_cid("ftp://example.com/x").is_none());
+        assert!(compute_ext_play_cid("").is_none());
+    }
+
+    #[test]
+    fn ext_play_cid_differs_from_the_url_locator_for_the_same_url() {
+        // The whole contract lives in the codec slot: `0x1006` means "fetch and
+        // seed these bytes", `0x1009` means "open this, there are no bytes for
+        // us". Wire-identical otherwise — which is exactly why a decoder that
+        // matched on *shape* would seed a Bandcamp page as the file.
+        let url = "https://example.bandcamp.com/album/x";
+        let ext = compute_ext_play_cid(url).unwrap();
+        let card = compute_card_cid("x", url).unwrap();
+        assert_ne!(ext, card);
+    }
+
     #[test]
     fn ipfs_cid_single_leaf_matches_kubo() {
         assert_eq!(
@@ -518,4 +954,156 @@ mod tests {
         assert_eq!(out.blocks[0].1.len(), 0);
         assert_eq!(out.blocks[0].0, out.root);
     }
+
+    // ---- card locator (0x1007) --------------------------------------------
+
+    /// Inverse of [`base32_lower_no_padding`], test-only. Lets the card tests
+    /// assert the actual CIDv1 bytes (codec, multihash code, digest) rather
+    /// than just round-tripping the encoder against itself.
+    fn base32_lower_decode(s: &str) -> Vec<u8> {
+        const ALPHABET: &[u8] = b"abcdefghijklmnopqrstuvwxyz234567";
+        let mut out = Vec::new();
+        let (mut acc, mut bits) = (0u32, 0u32);
+        for c in s.bytes() {
+            let v = ALPHABET.iter().position(|&a| a == c).expect("base32 char") as u32;
+            acc = (acc << 5) | v;
+            bits += 5;
+            if bits >= 8 {
+                bits -= 8;
+                out.push((acc >> bits) as u8);
+                acc &= (1 << bits) - 1;
+            }
+        }
+        out
+    }
+
+    /// The property the whole card design leans on: the CID is a pure function
+    /// of `(source, id)`, so any peer can derive a card's address offline. If
+    /// this literal ever changes, every already-published card silently forks
+    /// into a second identity — treat a failure here as a wire break.
+    #[test]
+    fn card_cid_is_deterministic() {
+        let cid = compute_card_cid("tmdb", "tv:95479").expect("card cid");
+        assert_eq!(cid, compute_card_cid("tmdb", "tv:95479").unwrap());
+        assert_eq!(cid, "bagdsaaanar2g2zdcor3duojvgq3ts");
+    }
+
+    /// Source- and kind-namespaced: the same numeric id on a different bridge,
+    /// or the same id under a different media kind, must not collide.
+    #[test]
+    fn card_cid_is_source_and_kind_namespaced() {
+        let tmdb = compute_card_cid("tmdb", "tv:95479").unwrap();
+        let mal = compute_card_cid("mal", "95479").unwrap();
+        let movie = compute_card_cid("tmdb", "movie:95479").unwrap();
+        assert_ne!(tmdb, mal);
+        assert_ne!(tmdb, movie);
+    }
+
+    /// Pins the on-the-wire CIDv1 layout: v1, codec `0x1007` (varint
+    /// `[0x87,0x20]`), **identity** multihash `0x00` (this family has no
+    /// digest — the locator itself is the payload), then
+    /// `varint(source_len) ‖ source ‖ id`.
+    #[test]
+    fn card_cid_wire_shape() {
+        let cid = compute_card_cid("tmdb", "tv:95479").unwrap();
+        assert!(cid.starts_with('b'), "multibase base32-lower prefix");
+        let wire = base32_lower_decode(&cid[1..]);
+
+        assert_eq!(wire[0], 0x01, "CIDv1");
+        assert_eq!(&wire[1..3], &[0x87, 0x20], "codec varint for 0x1007");
+        assert_eq!(wire[3], 0x00, "identity multihash — no digest");
+
+        let digest_len = wire[4] as usize;
+        let digest = &wire[5..5 + digest_len];
+        assert_eq!(digest[0], 4, "varint(len(\"tmdb\"))");
+        assert_eq!(&digest[1..5], b"tmdb");
+        assert_eq!(&digest[5..], b"tv:95479");
+    }
+
+    /// Empty inputs are rejected rather than encoded as an ambiguous locator,
+    /// and an id past the 64-byte multihash budget (meta-share's
+    /// `CidGeneric<64>`) returns `None` so the caller drops the card instead of
+    /// emitting a cid that peer would refuse to parse.
+    #[test]
+    fn card_cid_rejects_empty_and_oversized() {
+        assert!(compute_card_cid("", "tv:1").is_none());
+        assert!(compute_card_cid("tmdb", "").is_none());
+        let huge = "x".repeat(CARD_LOCATOR_MAX_DIGEST);
+        assert!(compute_card_cid("tmdb", &huge).is_none());
+        // Exactly at the ceiling still encodes.
+        let fits = "x".repeat(CARD_LOCATOR_MAX_DIGEST - 5); // varint(4) + "tmdb"
+        assert!(compute_card_cid("tmdb", &fits).is_some());
+    }
+
+    // ---- nzb-posting (0x1003) — the dedup contract -----------------------
+    //
+    // These pin the normalisation rule documented on `compute_nzb_posting_cid`.
+    // meta-share re-derives the same digest from the fetched manifest and
+    // rejects a mismatch, so a change that breaks one of these silently makes
+    // every self-scanned posting unplayable. Mirror any edit on both sides.
+
+    /// Segment/file **order must not matter** — that is the whole reason we
+    /// hash a sorted id set instead of the raw `.nzb` bytes (the retired
+    /// `0x1004` did the latter and broke dedup across reposts).
+    #[test]
+    fn nzb_posting_cid_is_order_independent() {
+        let a = ["[email protected]", "[email protected]", "[email protected]"];
+        let b = ["[email protected]", "[email protected]", "[email protected]"];
+        let c = ["[email protected]", "[email protected]", "[email protected]"];
+        assert_eq!(compute_nzb_posting_cid(&a), compute_nzb_posting_cid(&b));
+        assert_eq!(compute_nzb_posting_cid(&a), compute_nzb_posting_cid(&c));
+    }
+
+    /// Angle brackets, surrounding whitespace, empty entries and duplicates are
+    /// all normalised away. NZB `<segment>` bodies carry the bare form but NNTP
+    /// `BODY` re-adds the brackets, so both spellings must converge.
+    #[test]
+    fn nzb_posting_cid_normalises_brackets_blanks_and_dupes() {
+        let bare = ["[email protected]", "[email protected]"];
+        let noisy = [" <[email protected]> ", "[email protected]", "", "  ", "[email protected]"];
+        assert_eq!(
+            compute_nzb_posting_cid(&bare),
+            compute_nzb_posting_cid(&noisy)
+        );
+    }
+
+    /// Case is preserved: a Message-ID's local part is case-sensitive per RFC
+    /// 5322, so lowercasing could collide two genuinely distinct articles.
+    #[test]
+    fn nzb_posting_cid_is_case_sensitive() {
+        assert_ne!(
+            compute_nzb_posting_cid(&["[email protected]"]),
+            compute_nzb_posting_cid(&["[email protected]"])
+        );
+    }
+
+    /// A different article set is a different posting.
+    #[test]
+    fn nzb_posting_cid_differs_on_different_ids() {
+        assert_ne!(
+            compute_nzb_posting_cid(&["[email protected]", "[email protected]"]),
+            compute_nzb_posting_cid(&["[email protected]", "[email protected]"])
+        );
+    }
+
+    /// Wire shape: CIDv1, codec `0x1003` (varint `[0x83, 0x20]`), multihash
+    /// sha2-256 (`0x12`), 32-byte digest. Pinned because meta-share matches on
+    /// the **codec** slot — a drift here makes every posting undecodable there.
+    #[test]
+    fn nzb_posting_cid_wire_shape() {
+        let cid = compute_nzb_posting_cid(&["[email protected]"]);
+        assert!(cid.starts_with('b'), "multibase base32-lower prefix");
+
+        // Round-trip the base32 body back to bytes to assert the header.
+        let decoded = base32_lower_decode(&cid[1..]);
+        assert_eq!(decoded[0], 0x01, "CIDv1");
+        assert_eq!(&decoded[1..3], &[0x83, 0x20], "varint(0x1003)");
+        assert_eq!(decoded[3], 0x12, "multihash sha2-256");
+        assert_eq!(decoded[4], 0x20, "digest length 32");
+        assert_eq!(decoded.len(), 5 + 32);
+
+        // The digest is sha256 over the newline-joined sorted set.
+        let expect: [u8; 32] = Sha256::digest(b"[email protected]").into();
+        assert_eq!(&decoded[5..], &expect);
+    }
 }

crates/meta-feeder-sdk/src/lib.rs

@@ -14,9 +14,17 @@
 // pure noise for a single-process store, so silence the lint crate-wide.
 #![allow(clippy::result_large_err)]
 
+/// pcs-convention default meta-core root a self-publishing feeder targets when
+/// neither its dashboard config nor the `META_CORE_URL` env seed sets one. Keeps
+/// the modular storage seam out of compose: a standard install just works; a
+/// non-standard meta-core is set in the feeder's config UI. Matches the
+/// `metacore-app` service the AppStore feeder apps ship on the shared network.
+pub const DEFAULT_META_CORE_URL: &str = "http://metacore-app:9000";
+
 pub mod cache;
 pub mod common;
 pub mod config;
+pub mod domain;
 pub mod enrich;
 pub mod filename_meta;
 pub mod hash;

crates/meta-feeder-sdk/src/meta_core.rs

@@ -184,6 +184,48 @@ impl FeederStore {
         &self.meta_core_url
     }
 
+    /// Raw keys containing `contains` (`GET /api/kv/search`). Returns
+    /// `(keys, truncated)`; `truncated` means the server capped the result and
+    /// there may be more.
+    ///
+    /// This is a keyspace scan, so it belongs on rare control-plane paths only —
+    /// today, invalidating the gateway's search-coverage markers when a feeder's
+    /// upstream source set changes. Never call it per search.
+    pub async fn search_keys(&self, contains: &str, limit: usize) -> anyhow::Result<(Vec<String>, bool)> {
+        let url = format!("{}/api/kv/search", self.meta_core_url);
+        let resp = self
+            .http
+            .get(&url)
+            .query(&[("contains", contains), ("limit", &limit.to_string())])
+            .send()
+            .await?;
+        let status = resp.status();
+        if !status.is_success() {
+            anyhow::bail!("meta-core KV search `{contains}` returned {status}");
+        }
+        #[derive(Deserialize)]
+        struct KvSearchResponse {
+            #[serde(default)]
+            keys: Vec<String>,
+            #[serde(default)]
+            truncated: bool,
+        }
+        let body: KvSearchResponse = resp.json().await?;
+        Ok((body.keys, body.truncated))
+    }
+
+    /// Delete one raw KV key (`DELETE /api/kv/value?key=`). A missing key is not
+    /// an error.
+    pub async fn delete_kv(&self, key: &str) -> anyhow::Result<()> {
+        let url = format!("{}/api/kv/value", self.meta_core_url);
+        let resp = self.http.delete(&url).query(&[("key", key)]).send().await?;
+        let status = resp.status();
+        if status.is_success() || status == reqwest::StatusCode::NOT_FOUND {
+            return Ok(());
+        }
+        anyhow::bail!("meta-core KV DELETE {key} returned {status}");
+    }
+
     /// Self-publish a **metadata-only** outcome to meta-core, returning the
     /// **sparse** outcome (`record: None`) so the gateway's `(_, None)` branch
     /// skips it — no double write. Byte outcomes and already-sparse outcomes are

crates/meta-feeder-sdk/src/plugin.rs

@@ -57,6 +57,30 @@ pub struct HashOutcome {
 ///   today emits.
 /// - `BtV1File` — a single file inside a BitTorrent v1 torrent, custom
 ///   multicodec `0x1001`. Locator, not content hash: opaque to bitswap.
+/// - `NzbRelease` — a Usenet release (a Newznab listing), custom multicodec
+///   `nzb-release` `0x1005`. Self-describing locator: the cid embeds `{host,
+///   id}` in an identity multihash, opaque to bitswap and redeemed to bytes
+///   only by a credentialed meta-share peer, which decodes the locator from the
+///   cid and grabs the `.nzb` via `t=get` — no KV side-table.
+/// - `CardLocator` — the identity of a *work* (a series, a film) as published by
+///   a metadata bridge, custom multicodec `0x1007`. The only family with **no
+///   bytes behind it at all**: the record IS the payload. Always paired with
+///   `bytes: None`, so it lands in the `(None, Some)` metadata-only branch
+///   above. Opaque to bitswap, and deliberately so — there is nothing to seed.
+/// - `NzbPosting` — a Usenet posting we **scanned ourselves**, custom multicodec
+///   `nzb-posting` `0x1003`. A content-derived digest over the article
+///   Message-ID set (not a locator: no host is embedded), so any peer with a
+///   plain NNTP provider can fetch it — no `IndexerCred`. Opaque to bitswap: the
+///   digest is over the id set, not over any block, so the `.nzb` manifest
+///   travels separately as an ordinary sha2-256 cid in the record's `manifest`
+///   field. Produced by `hash::compute_nzb_posting_cid`.
+/// - `YtVideo` — a **delegated-playback** reference, custom multicodec
+///   `yt-video` `0x1008`. Like `CardLocator` it is always paired with
+///   `bytes: None` and lands in the `(None, Some)` metadata-only branch, but it
+///   means something stronger than "no bytes yet": the bytes **exist and are
+///   permanently someone else's**. Nothing is fetched, seeded or
+///   content-addressed, and meta-share's byte path REFUSES the cid with a `400`
+///   rather than 404ing. Produced by `hash::compute_yt_video_cid`.
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum HashKind {
     /// Reserved: no feeder currently emits midhash256 outcomes, but the family
@@ -65,6 +89,10 @@ pub enum HashKind {
     Midhash256,
     Sha2_256,
     BtV1File,
+    NzbRelease,
+    CardLocator,
+    NzbPosting,
+    YtVideo,
 }
 
 /// Startup-only configuration error returned by [`FeederPlugin::configure`].
@@ -101,6 +129,33 @@ pub fn upstream_id_field(upstream_id: &str) -> String {
     format!("{upstream_id}id")
 }
 
+/// Key-set prefix for the provenance `source/<label>` field (METADATA_KEYS §5).
+/// Each member is its own hash field (`source/gateway:nyaa.si = "true"`) so two
+/// producers that reach the same content by different routes union without a
+/// last-writer-wins clobber. The `<label>` uses `:` internally (not `/`, which
+/// is the key-set path separator) — e.g. `gateway:nyaa.si`, `gateway:tribler`.
+pub const SOURCE_KEYSET_PREFIX: &str = "source/";
+
+/// Stamp the SDK's default provenance member — `source/gateway:<upstream_id>` —
+/// onto a record's fields, but **only if it carries no `source/*` member yet**.
+/// This gives every gateway hit a provenance label for free (so meta-watch never
+/// falls back to "unknown" for a feeder record), while a plugin that knows a
+/// finer origin (torznab → the specific Prowlarr indexer) stamps its own flat
+/// `source/*` facets first (`source/gateway`, `source/<protocol>`,
+/// `source/<indexer-slug>`) and this default then no-ops.
+pub fn stamp_default_source(
+    fields: &mut std::collections::BTreeMap<String, String>,
+    upstream_id: &str,
+) {
+    if fields.keys().any(|k| k.starts_with(SOURCE_KEYSET_PREFIX)) {
+        return;
+    }
+    fields.insert(
+        format!("{SOURCE_KEYSET_PREFIX}gateway:{upstream_id}"),
+        "true".to_string(),
+    );
+}
+
 /// Static plugin contract. Each enabled `upstream_id` is implemented by
 /// exactly one `Box<dyn FeederPlugin>`. Lives behind `&self` in steady state —
 /// called concurrently, owns any interior mutability for caches.

crates/meta-feeder-sdk/src/query_eval.rs

@@ -27,6 +27,16 @@ pub fn record_matches(fields: &BTreeMap<String, String>, query: &GatewayQuery) -
             }
             continue;
         }
+        // `genres` is a key-set too (`genres/<Name> = "true"`, METADATA_KEYS
+        // §`genres/{name}`) on everything the card tier writes. Without this arm
+        // a `genres:` filter looks up a flat `genres` field, finds nothing on a
+        // card, and drops it — so a genre-scoped row would return empty.
+        if key == "genres" {
+            if !genres_filter_matches(fields, allowed) {
+                return false;
+            }
+            continue;
+        }
         let Some(field_value) = fields.get(key) else {
             return false;
         };
@@ -155,6 +165,54 @@ fn record_has_concrete_language(fields: &BTreeMap<String, String>) -> bool {
 
 /// Key-set members are written as the literal `"true"`; tolerate a couple of
 /// obvious truthy spellings and treat anything else as absent.
+/// Fold a genre label to its comparison form: lowercase, alphanumerics only.
+///
+/// Genre display names carry spaces and ampersands (`"Action & Adventure"`,
+/// `"Sci-Fi & Fantasy"`, `"Science Fiction"`), and **space is the query DSL's
+/// token separator** — so a filter can never spell a genre the way the record
+/// stores it. Callers pass a slug (`genres:action-adventure`); folding both
+/// sides makes them one value without forcing the stored name to be
+/// machine-shaped.
+fn genre_fold(s: &str) -> String {
+    s.chars()
+        .filter(|c| c.is_alphanumeric())
+        .flat_map(|c| c.to_lowercase())
+        .collect()
+}
+
+/// True iff the record carries one of the requested genres, in **either**
+/// encoding: a truthy `genres/<Name>` key-set member (the card tier's shape) or
+/// the legacy flat comma-joined `genres` field. OR across the requested genres,
+/// both sides folded — see [`genre_fold`].
+///
+/// **Fails closed, unlike [`languages_filter_matches`].** A language filter is a
+/// viewer *preference* narrowing a set, so an unknown-language release is kept.
+/// A genre filter *defines what a row is*: a record with no genre information is
+/// not known to be Action, so putting it in an Action row states something
+/// untrue. Upstream has already filtered by genre, so this is a re-validation.
+fn genres_filter_matches(fields: &BTreeMap<String, String>, allowed: &[String]) -> bool {
+    let wanted: Vec<String> = allowed.iter().map(|g| genre_fold(g)).collect();
+    if wanted.iter().all(|g| g.is_empty()) {
+        return true; // a filter with no usable value constrains nothing
+    }
+    for (key, value) in fields {
+        let Some(name) = key.strip_prefix("genres/") else {
+            continue;
+        };
+        if is_truthy_member(value) && wanted.contains(&genre_fold(name)) {
+            return true;
+        }
+    }
+    if let Some(flat) = fields.get("genres") {
+        for member in flat.split(|c| c == ',' || c == '|') {
+            if wanted.contains(&genre_fold(member)) {
+                return true;
+            }
+        }
+    }
+    false
+}
+
 fn is_truthy_member(v: &str) -> bool {
     matches!(v.trim().to_ascii_lowercase().as_str(), "true" | "1" | "yes")
 }

crates/meta-feeder-sdk/src/serve.rs

@@ -54,6 +54,10 @@ pub enum HashKindDto {
     Midhash256,
     Sha2_256,
     BtV1File,
+    NzbRelease,
+    CardLocator,
+    NzbPosting,
+    YtVideo,
 }
 
 impl From<HashKind> for HashKindDto {
@@ -62,6 +66,10 @@ impl From<HashKind> for HashKindDto {
             HashKind::Midhash256 => HashKindDto::Midhash256,
             HashKind::Sha2_256 => HashKindDto::Sha2_256,
             HashKind::BtV1File => HashKindDto::BtV1File,
+            HashKind::NzbRelease => HashKindDto::NzbRelease,
+            HashKind::CardLocator => HashKindDto::CardLocator,
+            HashKind::NzbPosting => HashKindDto::NzbPosting,
+            HashKind::YtVideo => HashKindDto::YtVideo,
         }
     }
 }
@@ -72,6 +80,10 @@ impl From<HashKindDto> for HashKind {
             HashKindDto::Midhash256 => HashKind::Midhash256,
             HashKindDto::Sha2_256 => HashKind::Sha2_256,
             HashKindDto::BtV1File => HashKind::BtV1File,
+            HashKindDto::NzbRelease => HashKind::NzbRelease,
+            HashKindDto::CardLocator => HashKind::CardLocator,
+            HashKindDto::NzbPosting => HashKind::NzbPosting,
+            HashKindDto::YtVideo => HashKind::YtVideo,
         }
     }
 }
@@ -424,8 +436,20 @@ async fn config_values_put(
     std::fs::write(&tmp, body)
         .and_then(|_| std::fs::rename(&tmp, &path))
         .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, format!("write config: {e}")).into_response())?;
-    info!(target: "meta-feeder", upstream_id = id, path = %path.display(), "feeder config saved (restart to apply)");
-    Ok(Json(serde_json::json!({ "saved": true, "restart_required": true })))
+    info!(target: "meta-feeder", upstream_id = id, path = %path.display(), "feeder config saved; scheduling self-restart to apply");
+    // Apply the new config by restarting the process. We can't hot-swap config
+    // in-place: plugins hold an exclusive redb cache handle for their whole life,
+    // so a fresh instance can't reopen the cache while the old one is alive. A
+    // clean process exit is the reliable apply path — Docker's restart policy
+    // (`unless-stopped`) brings the feeder back in ~2-3s and `configure()`
+    // re-reads config.json at startup. Delay briefly so THIS response flushes to
+    // the client before we exit.
+    tokio::spawn(async {
+        tokio::time::sleep(std::time::Duration::from_millis(800)).await;
+        info!(target: "meta-feeder", "restarting now to apply saved config");
+        std::process::exit(0);
+    });
+    Ok(Json(serde_json::json!({ "saved": true, "reloading": true })))
 }
 
 /// `POST /enrich/callback` — sink for the enrichment plugins' completion
@@ -478,10 +502,13 @@ pub fn router(
 /// Configure each plugin (cache dir under `state_dir/gateway/<id>`), then serve
 /// the feeder HTTP contract on `listen` until shutdown.
 ///
-/// A plugin whose `configure()` returns [`ConfigError::MissingConfig`] is
-/// **soft-skipped** (warn + dropped) — mirroring the gateway registry's
-/// behaviour, so a misconfigured plugin doesn't take the whole feeder down.
-/// Any other `ConfigError` aborts startup.
+/// The feeder process never aborts on a config problem: a plugin whose
+/// `configure()` returns [`ConfigError::MissingConfig`] is **soft-skipped**
+/// (warn + dropped — an optional source the operator didn't ask for), and any
+/// other failure (bad/empty key, cache-dir setup, …) registers a
+/// [`DegradedPlugin`] that stays up and reports the reason via `/health` and
+/// `/manifest`. The container keeps running so the error is visible in the UX
+/// instead of crash-looping.
 pub async fn serve_feeders(
     plugins: Vec<Box<dyn FeederPlugin>>,
     state_dir: impl Into<PathBuf>,
@@ -503,9 +530,72 @@ pub async fn serve_feeders(
         .map_err(anyhow::Error::from)
 }
 
-/// Run each plugin's `configure()` against its per-plugin cache dir, applying
-/// the soft-skip rule, and return the surviving plugins keyed by `upstream_id`.
-/// Public so feeder binaries / tests can configure without serving.
+/// A plugin whose `configure()` (or cache-dir setup) failed. It stays
+/// **registered** so the failure is visible at `GET /health` (as
+/// [`PluginHealth::Degraded`]) and `GET /manifest`; any query/compute against it
+/// returns the captured reason as a permanent error. This is what keeps one
+/// misconfigured feeder from crash-looping the whole container — the operator
+/// sees the reason in the UX and fixes the config, no restart-loop required.
+struct DegradedPlugin {
+    id: &'static str,
+    reason: String,
+    file_types: &'static [&'static str],
+    content_kinds: &'static [&'static str],
+    /// Preserved from the wrapped plugin so a degraded/unconfigured feeder still
+    /// serves its config form — otherwise the dashboard shows "no configuration
+    /// required" and the operator can never fix the very config that's missing.
+    schema: crate::config::ConfigSchema,
+}
+
+#[async_trait::async_trait]
+impl FeederPlugin for DegradedPlugin {
+    fn upstream_id(&self) -> &'static str {
+        self.id
+    }
+
+    fn configure(&mut self, _cache_dir: &FsPath) -> Result<(), ConfigError> {
+        Ok(())
+    }
+
+    async fn handle_query(
+        &self,
+        _query: &GatewayQuery,
+        _max_results: usize,
+    ) -> Result<Vec<DiscoveryRecord>, GatewayError> {
+        Err(GatewayError::Permanent(self.reason.clone()))
+    }
+
+    async fn compute_outcomes(
+        &self,
+        _record_id: &str,
+    ) -> Result<Vec<HashOutcome>, GatewayError> {
+        Err(GatewayError::Permanent(self.reason.clone()))
+    }
+
+    fn health(&self) -> PluginHealth {
+        PluginHealth::Degraded {
+            reason: self.reason.clone(),
+        }
+    }
+
+    fn served_file_types(&self) -> &'static [&'static str] {
+        self.file_types
+    }
+
+    fn served_content_kinds(&self) -> &'static [&'static str] {
+        self.content_kinds
+    }
+
+    fn config_schema(&self) -> crate::config::ConfigSchema {
+        self.schema.clone()
+    }
+}
+
+/// Run each plugin's `configure()` against its per-plugin cache dir and return
+/// the plugins keyed by `upstream_id`. Never errors on a per-plugin problem:
+/// `MissingConfig` is soft-skipped, every other failure becomes a
+/// [`DegradedPlugin`] (see [`serve_feeders`]). Public so feeder binaries /
+/// tests can configure without serving.
 pub fn configure_plugins(
     plugins: Vec<Box<dyn FeederPlugin>>,
     state_dir: &FsPath,
@@ -513,25 +603,63 @@ pub fn configure_plugins(
     let mut out: HashMap<String, Arc<dyn FeederPlugin>> = HashMap::new();
     for mut plugin in plugins {
         let id = plugin.upstream_id();
+        // Captured up front (both are `'static`, so this borrow of `plugin`
+        // ends immediately) — lets us build a DegradedPlugin without holding a
+        // borrow across `configure(&mut self)` / the move into `out`.
+        let file_types = plugin.served_file_types();
+        let content_kinds = plugin.served_content_kinds();
+        // Preserve the config surface so a degraded stand-in still serves the
+        // real config form (see DegradedPlugin.schema).
+        let schema = plugin.config_schema();
         let cache_dir = state_dir.join("gateway").join(id);
-        std::fs::create_dir_all(&cache_dir).map_err(|e| {
-            anyhow::anyhow!("create cache dir {} for {id}: {e}", cache_dir.display())
-        })?;
+
+        // Build (don't insert) a degraded stand-in: keeps the container up and
+        // surfaces the reason via /health instead of crash-looping.
+        let make_degraded = |reason: String| -> Arc<dyn FeederPlugin> {
+            warn!(
+                target: "meta-feeder",
+                upstream_id = id,
+                %reason,
+                "feeder plugin degraded: container stays up; error surfaced via /health"
+            );
+            Arc::new(DegradedPlugin {
+                id,
+                reason,
+                file_types,
+                content_kinds,
+                schema: schema.clone(),
+            })
+        };
+
+        if let Err(e) = std::fs::create_dir_all(&cache_dir) {
+            let reason = format!("cache dir {} setup failed: {e}", cache_dir.display());
+            out.insert(id.to_string(), make_degraded(reason));
+            continue;
+        }
+
         match plugin.configure(&cache_dir) {
             Ok(()) => {
-                let id = id.to_string();
-                out.insert(id, Arc::from(plugin));
+                out.insert(id.to_string(), Arc::from(plugin));
             }
             Err(ConfigError::MissingConfig { plugin: p, what }) => {
+                // Keep a degraded stand-in (don't drop it) so its config form
+                // stays reachable from the dashboard and the operator can supply
+                // the missing config — then restart the feeder to apply it.
                 warn!(
                     target: "meta-feeder",
-                    upstream_id = p,
-                    missing = what,
-                    "feeder plugin skipped: required config not supplied; \
-                     other plugins continue"
+                    upstream_id = %p,
+                    missing = %what,
+                    "feeder plugin degraded: required config not supplied; \
+                     config form still served so it can be fixed from the dashboard"
+                );
+                out.insert(
+                    id.to_string(),
+                    make_degraded(format!("required config not supplied: {what}")),
                 );
             }
-            Err(other) => return Err(anyhow::anyhow!(other)),
+            Err(other) => {
+                out.insert(id.to_string(), make_degraded(other.to_string()));
+            }
         }
     }
     Ok(out)

feeder-plugin/book-feeder/Cargo.toml

@@ -6,9 +6,10 @@ rust-version.workspace = true
 license.workspace = true
 publish = false
 
-# Book feeder sidecar: Project Gutenberg (via Gutendex). The proof-of-concept
-# feeder — single upstream, no preview-seeding, no BitTorrent — that validates
-# the meta-feeder-sdk + feeder HTTP contract end to end.
+# Book feeder sidecar, two upstreams:
+# - `gutenberg` — Project Gutenberg editions (via Gutendex), the byte tier.
+# - `openlibrary` — Open Library work cards, the book identity tier
+#   (`TMDB : video :: AniList : comics :: Open Library : books`).
 
 [lib]
 name = "book_feeder"
@@ -19,7 +20,11 @@ name = "book-feeder"
 path = "src/main.rs"
 
 [dependencies]
-meta-feeder-sdk = { path = "../../crates/meta-feeder-sdk" }
+# Pinned by TAG, never by path. A path dep resolves OUTSIDE the crate, and the
+# vendored copy this used to point at had already drifted behind the shared SDK
+# (no `RateBudget`, a capped card cid). The rest of the fleet pins the same
+# way; see meta-feeder-anilist's manifest.
+meta-feeder-sdk = { git = "https://gitea-watch.nsl.sh/metastar/meta-feeder-sdk", tag = "v1.7.0" }
 
 tokio = { version = "1.40", features = ["full"] }
 axum = { version = "0.7", features = ["json"] }

feeder-plugin/book-feeder/Dockerfile

@@ -1,12 +1,13 @@
 # syntax=docker/dockerfile:1
 # =============================================================================
-# book-feeder — Project Gutenberg feeder sidecar. Multi-stage build.
+# book-feeder — Project Gutenberg + Open Library feeder sidecar. Multi-stage
+# build; build context is this repo's root (the Cargo workspace).
 #
-# The build context must be `packages/meta-gateway/` (the workspace root) so
-# the `meta-feeder-sdk` path dependency (../../crates/meta-feeder-sdk) resolves.
-# Same BuildKit cache-mount strategy as the gateway Dockerfile: the cargo
-# registry/git and `target/` persist across builds. The release binary is
-# copied OUT of the target cache mount within the same RUN.
+# meta-feeder-sdk arrives as a git dependency (tag-pinned), so the vendored
+# `crates/` copy is not copied in — `git` must be present in the builder for
+# cargo to fetch it. Same BuildKit cache-mount strategy as the gateway
+# Dockerfile: the cargo registry/git and `target/` persist across builds. The
+# release binary is copied OUT of the target cache mount within the same RUN.
 # =============================================================================
 
 FROM rust:1.89-slim-bookworm AS builder
@@ -20,7 +21,6 @@ RUN apt-get update && apt-get install -y --no-install-recommends \
     && rm -rf /var/lib/apt/lists/*
 
 COPY Cargo.toml Cargo.lock rust-toolchain.toml ./
-COPY crates/ crates/
 COPY feeder-plugin/ feeder-plugin/
 
 RUN --mount=type=cache,id=cargo-registry,target=/usr/local/cargo/registry \

feeder-plugin/book-feeder/src/lib.rs

@@ -1,5 +1,13 @@
-//! `book-feeder` library surface — exposes the Gutenberg plugin so integration
-//! tests (and, later, other binaries) can construct it. The binary
-//! (`main.rs`) is a thin wrapper over [`meta_feeder_sdk::serve_feeders`].
+//! `book-feeder` library surface — exposes the plugins so integration tests
+//! (and, later, other binaries) can construct them. The binary (`main.rs`) is a
+//! thin wrapper over [`meta_feeder_sdk::serve_feeders`].
+//!
+//! Two upstreams:
+//! - [`gutenberg`] — Project Gutenberg editions, the byte tier (`document` × `book`);
+//! - [`openlibrary`] — Open Library work cards, the identity tier (`card` × `book`),
+//!   over the client in [`openlibrary_client`], which `gutenberg` also uses to
+//!   bind an edition to its work.
 
 pub mod gutenberg;
+pub mod openlibrary;
+pub mod openlibrary_client;

feeder-plugin/book-feeder/src/main.rs

@@ -1,17 +1,27 @@
-//! `book-feeder` — Project Gutenberg feeder sidecar.
+//! `book-feeder` — the book tier's feeder sidecar.
 //!
-//! Hosts the [`gutenberg::GutenbergPlugin`] over the feeder HTTP contract via
-//! [`meta_feeder_sdk::serve_feeders`]. The gateway core reaches it server-to-
-//! server on the internal docker network.
+//! Hosts two plugins over the feeder HTTP contract via
+//! [`meta_feeder_sdk::serve_feeders`]:
+//! - [`GutenbergPlugin`] — Project Gutenberg editions (`document` × `book`);
+//! - [`OpenLibraryPlugin`] — Open Library work cards (`card` × `book`).
+//!
+//! The gateway core reaches it server-to-server on the internal docker network.
 //!
 //! Env:
 //! - `META_FEEDER_HTTP_LISTEN` — listen addr (default `0.0.0.0:8080`)
 //! - `META_FEEDER_STATE_DIR`   — per-plugin cache root (default `/data/meta-feeder`)
+//! - `OPENLIBRARY_CONTACT`     — e-mail/URL sent in the Open Library User-Agent;
+//!   first-boot seed for the `contact` config field (the saved field wins)
+//! - `OPENLIBRARY_TOP_N`, `OPENLIBRARY_DISCOVERY_N` — optional card result sizing
 //! - `RUST_LOG`                — tracing filter (default `info`)
 
 use std::net::SocketAddr;
+use std::path::Path;
 
 use book_feeder::gutenberg::GutenbergPlugin;
+use book_feeder::openlibrary::OpenLibraryPlugin;
+use book_feeder::openlibrary_client;
+use meta_feeder_sdk::plugin::FeederPlugin;
 use meta_feeder_sdk::serve_feeders;
 use tracing_subscriber::EnvFilter;
 
@@ -27,5 +37,16 @@ async fn main() -> anyhow::Result<()> {
     let state_dir =
         std::env::var("META_FEEDER_STATE_DIR").unwrap_or_else(|_| "/data/meta-feeder".to_string());
 
-    serve_feeders(vec![Box::new(GutenbergPlugin::new())], state_dir, listen).await
+    // ONE budget for both plugins: the Gutenberg → Open Library binding and the
+    // card plugin call the same upstream, which rate-limits per client, not per
+    // plugin. Two buckets would each think they had the whole allowance.
+    let contact =
+        openlibrary_client::load_contact(&Path::new(&state_dir).join("gateway").join("openlibrary"));
+    let budget = openlibrary_client::budget(contact.is_some());
+
+    let plugins: Vec<Box<dyn FeederPlugin>> = vec![
+        Box::new(GutenbergPlugin::new().with_openlibrary_budget(budget.clone())),
+        Box::new(OpenLibraryPlugin::new().with_budget(budget)),
+    ];
+    serve_feeders(plugins, state_dir, listen).await
 }

feeder-plugin/book-feeder/tests/contract.rs

@@ -1,24 +1,36 @@
 //! End-to-end contract test for the feeder HTTP surface: boots the real
-//! `serve_feeders` router (book-feeder hosting GutenbergPlugin) on an ephemeral
-//! port, pointed at a wiremock Gutendex, and drives `/manifest`, `/query`, and
-//! `/compute` over HTTP. Deterministic — no live Gutendex.
+//! `serve_feeders` router (book-feeder hosting GutenbergPlugin AND
+//! OpenLibraryPlugin, as `main.rs` does) on an ephemeral port, pointed at one
+//! wiremock server that plays Gutendex, the Gutenberg RDF host and Open Library,
+//! and drives `/manifest`, `/query`, and `/compute` over HTTP. Deterministic —
+//! no live upstream.
 
 use book_feeder::gutenberg::GutenbergPlugin;
+use book_feeder::openlibrary::OpenLibraryPlugin;
+use book_feeder::openlibrary_client;
 use meta_feeder_sdk::{
-    configure_plugins, hash, router, ComputeRequest, ComputeResponse, GatewayQuery, HashKindDto,
-    ManifestResponse, QueryRequest, QueryResponse,
+    configure_plugins, hash, router, ComputeRequest, ComputeResponse, FeederPlugin, GatewayQuery,
+    HashKindDto, ManifestResponse, QueryRequest, QueryResponse,
 };
-use wiremock::matchers::{method, path};
+use wiremock::matchers::{method, path, query_param};
 use wiremock::{Mock, MockServer, ResponseTemplate};
 
-/// Spawn the feeder router (Gutenberg pointed at `upstream`) on an ephemeral
-/// port; return its base URL. The TempDir is returned so the per-plugin cache
-/// outlives the test.
+/// Spawn the feeder router (every upstream pointed at `upstream`) on an
+/// ephemeral port; return its base URL. The TempDir is returned so the
+/// per-plugin cache outlives the test.
 async fn spawn_feeder(upstream: &MockServer) -> (String, tempfile::TempDir) {
     let dir = tempfile::tempdir().expect("tempdir");
-    let plugin = GutenbergPlugin::with_base_url(upstream.uri());
-    let plugins =
-        configure_plugins(vec![Box::new(plugin)], dir.path()).expect("configure plugins");
+    let budget = openlibrary_client::budget(false);
+    let plugins: Vec<Box<dyn FeederPlugin>> = vec![
+        Box::new(
+            GutenbergPlugin::with_base_url(upstream.uri())
+                .with_rdf_base_url(upstream.uri())
+                .with_openlibrary_base_url(upstream.uri())
+                .with_openlibrary_budget(budget.clone()),
+        ),
+        Box::new(OpenLibraryPlugin::with_api_base(upstream.uri()).with_budget(budget)),
+    ];
+    let plugins = configure_plugins(plugins, dir.path()).expect("configure plugins");
     let app = router(plugins, "book-feeder-test".to_string(), dir.path());
 
     let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
@@ -44,7 +56,7 @@ fn books_json() -> serde_json::Value {
 }
 
 #[tokio::test]
-async fn manifest_lists_book_capabilities() {
+async fn manifest_lists_both_upstreams() {
     let upstream = MockServer::start().await;
     let (base, _dir) = spawn_feeder(&upstream).await;
 
@@ -57,21 +69,36 @@ async fn manifest_lists_book_capabilities() {
         .await
         .expect("decode manifest");
 
-    assert_eq!(manifest.plugins.len(), 1);
-    let p = &manifest.plugins[0];
-    assert_eq!(p.id, "gutenberg");
-    assert_eq!(p.served_file_types, vec!["document".to_string()]);
-    assert_eq!(p.served_content_kinds, vec!["book".to_string()]);
+    assert_eq!(manifest.plugins.len(), 2);
+    let gutenberg = manifest.plugins.iter().find(|p| p.id == "gutenberg").expect("gutenberg");
+    assert_eq!(gutenberg.served_file_types, vec!["document".to_string()]);
+    assert_eq!(gutenberg.served_content_kinds, vec!["book".to_string()]);
+    let openlibrary = manifest
+        .plugins
+        .iter()
+        .find(|p| p.id == "openlibrary")
+        .expect("openlibrary");
+    assert_eq!(openlibrary.served_file_types, vec!["card".to_string()]);
+    assert_eq!(openlibrary.served_content_kinds, vec!["book".to_string()]);
 }
 
 #[tokio::test]
-async fn query_returns_gutenberg_records() {
+async fn query_returns_gutenberg_records_with_their_open_library_claim() {
     let upstream = MockServer::start().await;
     Mock::given(method("GET"))
-        .and(path("/books"))
+        .and(path("/books/"))
         .respond_with(ResponseTemplate::new(200).set_body_json(books_json()))
         .mount(&upstream)
         .await;
+    Mock::given(method("GET"))
+        .and(path("/search.json"))
+        .and(query_param("q", "id_project_gutenberg:1342"))
+        .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
+            "numFound": 1,
+            "docs": [{ "key": "/works/OL66554W", "id_project_gutenberg": ["1342", "42671"] }]
+        })))
+        .mount(&upstream)
+        .await;
     let (base, _dir) = spawn_feeder(&upstream).await;
 
     let resp: QueryResponse = reqwest::Client::new()
@@ -90,9 +117,75 @@ async fn query_returns_gutenberg_records() {
 
     assert_eq!(resp.records.len(), 1);
     let r = &resp.records[0];
+    let f = |k: &str| r.fields.get(k).map(String::as_str);
     assert_eq!(r.upstream_id, "gutenberg");
-    assert_eq!(r.fields.get("gutenbergid").map(String::as_str), Some("1342"));
-    assert_eq!(r.fields.get("contentKind").map(String::as_str), Some("book"));
+    assert_eq!(f("gutenbergid"), Some("1342"));
+    assert_eq!(f("contentKind"), Some("book"));
+    assert_eq!(f("authors/Jane Austen"), Some("true"));
+    assert_eq!(f("languages/eng"), Some("true"));
+    // The transient claim the gateway transcribes into
+    // `anchoredBy/openlibrary:OL66554W`.
+    assert_eq!(f("openlibraryid"), Some("OL66554W"));
+    assert_eq!(f("anchored"), Some("true"));
+    assert_eq!(f("anchorSource"), Some("openlibrary"));
+    assert_eq!(f("anchorMethod"), None, "an identifier link is not a title match");
+}
+
+#[tokio::test]
+async fn query_returns_an_open_library_card() {
+    let upstream = MockServer::start().await;
+    Mock::given(method("GET"))
+        .and(path("/search.json"))
+        .and(query_param("q", "pride"))
+        .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
+            "numFound": 1,
+            "docs": [{
+                "key": "/works/OL66554W",
+                "title": "Pride and Prejudice",
+                "author_name": ["Jane Austen"],
+                "cover_i": 14348537,
+                "first_publish_year": 1813
+            }]
+        })))
+        .mount(&upstream)
+        .await;
+    Mock::given(method("GET"))
+        .and(path("/works/OL66554W.json"))
+        .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
+            "key": "/works/OL66554W",
+            "description": "A novel of manners."
+        })))
+        .mount(&upstream)
+        .await;
+    let (base, _dir) = spawn_feeder(&upstream).await;
+
+    let mut query = GatewayQuery::from_free_text("pride");
+    query.filters.insert("fileType".into(), vec!["card".into()]);
+    let resp: QueryResponse = reqwest::Client::new()
+        .post(format!("{base}/query"))
+        .json(&QueryRequest {
+            upstream_id: "openlibrary".to_string(),
+            query,
+            max_results: 10,
+        })
+        .send()
+        .await
+        .expect("POST /query")
+        .json()
+        .await
+        .expect("decode query response");
+
+    assert_eq!(resp.records.len(), 1);
+    let r = &resp.records[0];
+    assert_eq!(r.upstream_id, "openlibrary");
+    assert_eq!(r.record_id, "OL66554W");
+    assert_eq!(r.fields.get("fileType").map(String::as_str), Some("card"));
+    assert_eq!(
+        r.fields
+            .get("cids/bagdsaaaubnxxazlonruwe4tboj4u6tbwgy2tkncx")
+            .map(String::as_str),
+        Some("true")
+    );
 }
 
 #[tokio::test]
@@ -151,6 +244,57 @@ async fn compute_returns_outcome_with_inline_epub_bytes() {
     assert_eq!(hash::compute_ipfs_cid(&decoded), expected_cid);
 }
 
+#[tokio::test]
+async fn compute_for_an_open_library_card_is_a_byteless_card_locator() {
+    let upstream = MockServer::start().await;
+    Mock::given(method("GET"))
+        .and(path("/search.json"))
+        .and(query_param("q", "key:/works/OL45804W"))
+        .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
+            "numFound": 1,
+            "docs": [{
+                "key": "/works/OL45804W",
+                "title": "Fantastic Mr Fox",
+                "author_name": ["Roald Dahl"],
+                "cover_i": 6498519,
+                "first_publish_year": 1970
+            }]
+        })))
+        .mount(&upstream)
+        .await;
+    Mock::given(method("GET"))
+        .and(path("/works/OL45804W.json"))
+        .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
+            "key": "/works/OL45804W",
+            "description": "A clever fox outwits three farmers."
+        })))
+        .mount(&upstream)
+        .await;
+    let (base, _dir) = spawn_feeder(&upstream).await;
+
+    let resp: ComputeResponse = reqwest::Client::new()
+        .post(format!("{base}/compute"))
+        .json(&ComputeRequest {
+            upstream_id: "openlibrary".to_string(),
+            record_id: "OL45804W".to_string(),
+        })
+        .send()
+        .await
+        .expect("POST /compute")
+        .json()
+        .await
+        .expect("decode compute response");
+
+    assert_eq!(resp.outcomes.len(), 1);
+    let o = &resp.outcomes[0];
+    assert_eq!(o.hash, "bagdsaaaubnxxazlonruwe4tboj4u6tbugu4dancx");
+    assert_eq!(o.hash_kind, HashKindDto::CardLocator);
+    assert!(o.bytes_b64.is_none());
+    assert!(o.file_extension.is_none());
+    let record = o.record.as_ref().expect("the card record rides the outcome");
+    assert_eq!(record.fields.get("title").map(String::as_str), Some("Fantastic Mr Fox"));
+}
+
 #[tokio::test]
 async fn query_unknown_upstream_is_404() {
     let upstream = MockServer::start().await;