improvement

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "meta-feeder-sdk"
-version = "0.1.0"
+version = "1.3.0"
 edition = "2021"
 rust-version = "1.85"
 license = "MIT"

README.md

@@ -17,7 +17,7 @@ bytes; it never joins the swarm.
 
 ```toml
 [dependencies]
-meta-feeder-sdk = { git = "https://github.com/worph/meta-feeder-sdk", tag = "v0.1.0" }
+meta-feeder-sdk = { git = "https://github.com/worph/meta-feeder-sdk", tag = "v1.3.0" }
 ```
 
 Not on crates.io yet — the contract is still moving, so consumers pin a tag.
@@ -47,8 +47,32 @@ Four required methods; everything else has a default:
 | `compute_outcomes` | resolve a `record_id` to `HashOutcome`s | `bytes: None` for metadata-only sources |
 
 Defaults you can override: `handle_query_stream`, `handle_fetch`, `health`,
-`served_file_types`, `served_content_kinds`, `config_schema`, `config_values`,
-`get_blob`.
+`served_file_types`, `served_content_kinds`, `package`, `redeems`,
+`config_schema`, `config_values`, `get_blob`.
+
+### Redeeming locators (1.3.0)
+
+A plugin can **redeem** a locator family — turn a locator cid into the bytes it
+names, usually spending a metered quota (a Newznab `.nzb` grab, an OpenSubtitles
+download):
+
+1. Return a `RedeemClaim` from `redeems()` — `RedeemClaim::nzb_release(hosts)` or
+   `RedeemClaim::provider_file(sources)` — reflecting the *current* config (no
+   credential → no claim). Served per plugin in `GET /manifest` and, for polling,
+   at `GET /redeems` → `{"redeems": {"<upstream_id>": [claim…]}}`.
+2. Name a stable `package()` — the `/files/plugin/<package>/` folder the gateway
+   stores the bytes under.
+3. In `compute_outcomes`, accept the locator cid as `record_id` and return ONE
+   `Sha2_256` outcome: `hash = hash::compute_ipfs_cid(&bytes)`, `bytes`,
+   `file_extension`, and optionally a `record` whose fields the gateway merges
+   onto the locator record. Return `NotFound` when the cid is not yours (e.g. no
+   key for that host) so the gateway can try the next claimer; `RateLimited` when
+   the quota is spent.
+
+The gateway calls this only on a real play, one claimer at a time, and writes
+the bytes' cid onto the locator record under the claim's `field` (`manifest` /
+`file`), so each file's quota is spent once. Decode helpers:
+`hash::decode_nzb_release_cid`, `hash::decode_provider_file_cid`, `hash::codec_of`.
 
 ### The routing gate
 
@@ -101,6 +125,9 @@ into the bitswap blockstore.
 The last two carry an identity multihash. Anything that dispatches on the
 multihash instead of the **codec** will silently mis-rank them.
 
+`hash` also encodes/decodes the redeemable locators: `nzb-release` (`0x1005`)
+and `provider-file` (`0x100A`, `(source, id)` — `docs/cid-formats.md` §8).
+
 ## Layout
 
 | module | what |
@@ -121,13 +148,19 @@ multihash instead of the **codec** will silently mis-rank them.
 cargo build && cargo test
 ```
 
-To work on the SDK from inside a feeder checkout without pushing a tag, add a
-gitignored `.cargo/config.toml` in the feeder:
+To test a feeder against an SDK change **before its tag is pushed**: cargo will
+not resolve an unpushed tag, not even under a `[patch]` override ("failed to
+find tag"). Build a scratch copy of the feeder (outside its git tree) with the
+dependency swapped for a path, and run `cargo test` there:
 
-```toml
-paths = ["../meta-feeder-sdk"]
+```bash
+mkdir -p /tmp/feeder && (cd meta-feeder-x && tar cf - --exclude=./target --exclude=./.git .) | (cd /tmp/feeder && tar xf -)
+sed -i 's#meta-feeder-sdk = { git = .*#meta-feeder-sdk = { path = "/abs/path/to/meta-feeder-sdk" }#' /tmp/feeder/feeder-plugin/*/Cargo.toml
+(cd /tmp/feeder && cargo test)
 ```
 
+Never commit the path form: each feeder repo is its own Docker build context.
+
 ## Context
 
 Design rationale, the feeder roster, and the naming rules live in the meta-root:

src/domain.rs

@@ -1,23 +1,32 @@
-//! `contentKind` → `domain` — the routing table from METADATA_KEYS.md §1.
+//! `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:
+//! catalogue on — **one value per app**, no alternation:
 //!
 //! ```text
-//! meta-watch   →  domain:film|tv
+//! meta-watch   →  domain:screen
 //! meta-listen  →  domain:music
 //! meta-read    →  domain:literature
 //! ```
 //!
-//! # Why it is stored and not derived on read
+//! `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.
+//! 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
 //!
@@ -27,6 +36,10 @@
 //! 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
@@ -34,7 +47,7 @@
 //! `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.
+//! 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
@@ -42,14 +55,19 @@
 /// else.
 ///
 /// ⚠ `pack` is deliberately absent. A season pack and an album release are the
-/// same structural thing, so the kind alone cannot tell `tv` from `music`; the
-/// writer stamps `pack`'s domain from its own context (the same reason
+/// 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)] = &[
-    // film / tv
-    ("movie", "film"),
-    ("series", "tv"),
-    ("episode", "tv"),
+    // 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"),
@@ -57,9 +75,6 @@ const DOMAIN_BY_CONTENT_KIND: &[(&str, &str)] = &[
     ("musicVideo", "music"),
     ("djMix", "music"),
     ("liveSet", "music"),
-    // spoken
-    ("podcast", "spoken"),
-    ("podcastEpisode", "spoken"),
     // 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.
@@ -72,6 +87,37 @@ const DOMAIN_BY_CONTENT_KIND: &[(&str, &str)] = &[
     ("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
@@ -79,13 +125,41 @@ const DOMAIN_BY_CONTENT_KIND: &[(&str, &str)] = &[
 /// 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> {
-    let k = content_kind.trim();
-    DOMAIN_BY_CONTENT_KIND
+    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(|(_, domain)| *domain)
+        .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*
@@ -93,56 +167,134 @@ pub fn domain_for_content_kind(content_kind: &str) -> Option<&'static str> {
 /// 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> {
-    let d = domain.trim();
-    DOMAIN_BY_CONTENT_KIND
+    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(|(_, dom)| dom.eq_ignore_ascii_case(d))
+        .filter(|(_, val)| val.eq_ignore_ascii_case(v))
         .map(|(kind, _)| *kind)
         .collect()
 }
 
-/// Rewrite a query so a `domain:` filter can reach a plugin that has never
-/// heard of the key.
+/// 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 no `domain:` filter — the common
-/// path, and cheaper than cloning.
+/// 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. Nine of the fifteen 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 `domain` at all, and old plugins keep working unchanged
-/// — while the dispatcher still enforces the real `domain:` filter afterwards,
-/// against records it has just stamped.
+/// 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 two, and the `domain` term is still enforced
-/// post-stamp, so the AND semantics hold either way.
-pub fn expand_domain_filter(query: &crate::query::GatewayQuery) -> Option<crate::query::GatewayQuery> {
-    let domains = query.filters.get("domain")?;
+/// 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");
-    if !out.filters.contains_key("contentKind") {
-        let kinds: Vec<String> = domains
+    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()
-            .flat_map(|d| content_kinds_for_domain(d))
-            .map(|k| k.to_string())
-            .collect();
-        // An unknown domain 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 `domain:`
+            .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);
-        }
+        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`
@@ -150,14 +302,29 @@ pub fn expand_domain_filter(query: &crate::query::GatewayQuery) -> Option<crate:
 /// 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>) {
-    if fields.contains_key("domain") {
+    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(domain) = domain_for_content_kind(&kind) {
-        fields.insert("domain".to_string(), domain.to_string());
+    if let Some(value) = derive(&kind) {
+        fields.insert(key.to_string(), value.to_string());
     }
 }
 
@@ -168,8 +335,8 @@ mod tests {
 
     #[test]
     fn kinds_map_to_their_domain() {
-        assert_eq!(domain_for_content_kind("movie"), Some("film"));
-        assert_eq!(domain_for_content_kind("episode"), Some("tv"));
+        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"));
@@ -178,13 +345,45 @@ mod tests {
     }
 
     #[test]
-    fn pack_has_no_derivable_domain() {
+    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]
@@ -196,6 +395,18 @@ mod tests {
         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 {
@@ -217,13 +428,13 @@ mod tests {
     }
 
     #[test]
-    fn no_domain_filter_is_left_untouched() {
-        assert!(expand_domain_filter(&q(&[("fileType", &["video"])])).is_none());
+    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_domain_filter(&q(&[("domain", &["literature"])])).unwrap();
+        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();
@@ -232,9 +443,35 @@ mod tests {
         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_domain_filter(&q(&[
+        let out = expand_routing_filters(&q(&[
             ("domain", &["music"][..]),
             ("contentKind", &["track"][..]),
         ]))
@@ -246,10 +483,40 @@ mod tests {
     }
 
     #[test]
-    fn unknown_domain_adds_no_alternation() {
+    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 `domain:` check does the (correct, empty) filtering.
-        let out = expand_domain_filter(&q(&[("domain", &["hologram"])])).unwrap();
+        // 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"));
     }
 
@@ -258,18 +525,25 @@ mod tests {
         let mut f: BTreeMap<String, String> = BTreeMap::new();
         f.insert("contentKind".into(), "episode".into());
         stamp_domain(&mut f);
-        assert_eq!(f.get("domain").map(String::as_str), Some("tv"));
+        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 → no domain invented.
+        // 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());
     }
 }

src/hash.rs

@@ -206,6 +206,76 @@ pub fn compute_nzb_release_cid(api_base: &str, release_id: &str) -> Option<Strin
     Some(format!("b{}", base32_lower_no_padding(&wire)))
 }
 
+/// Decode a self-describing `nzb-release` cid into `(api_base, id)` — the exact
+/// inverse of [`compute_nzb_release_cid`], string-level (no `cid` crate).
+///
+/// `api_base` is scheme-less authority + optional path (`api.nzbgeek.info/api`);
+/// match credentials on its authority (split at the first `/`), grab against
+/// `https://{api_base}`. `id` is lowercase hex, re-encoded from the digest — so
+/// an **odd-length** id comes back left-padded with `0` (the encoder pads it),
+/// which Newznab `t=get` accepts as the same release.
+///
+/// `None` on a wrong codec, a non-identity multihash, or a malformed digest.
+pub fn decode_nzb_release_cid(cid: &str) -> Option<(String, String)> {
+    let digest = identity_locator_digest(cid, NZB_RELEASE_CODEC)?;
+    let mut pos = 0usize;
+    let base_len = read_varint(&digest, &mut pos)? as usize;
+    let rest = digest.get(pos..)?;
+    if rest.len() <= base_len {
+        return None; // need the base AND at least one id byte
+    }
+    let (base, id) = rest.split_at(base_len);
+    let api_base = std::str::from_utf8(base).ok()?.to_string();
+    let id_hex: String = id.iter().map(|b| format!("{b:02x}")).collect();
+    Some((api_base, id_hex))
+}
+
+/// The multicodec of a base32 (`b…`) CIDv1, or `None` for anything else
+/// (CIDv0 `Qm…`, other multibases, garbage). Dispatch on this — never on the
+/// multihash, which is identity (`0x00`) for every locator family.
+pub fn codec_of(cid: &str) -> Option<u64> {
+    let wire = base32_lower_decode(cid.strip_prefix('b')?)?;
+    let mut pos = 0usize;
+    if read_varint(&wire, &mut pos)? != 1 {
+        return None;
+    }
+    read_varint(&wire, &mut pos)
+}
+
+/// The digest of a well-formed identity-multihash CIDv1 carrying `codec`, or
+/// `None`. Shared by the locator decoders.
+fn identity_locator_digest(cid: &str, codec: u64) -> Option<Vec<u8>> {
+    let wire = base32_lower_decode(cid.strip_prefix('b')?)?;
+    let mut pos = 0usize;
+    if read_varint(&wire, &mut pos)? != 1
+        || read_varint(&wire, &mut pos)? != codec
+        || read_varint(&wire, &mut pos)? != 0x00
+    {
+        return None;
+    }
+    let len = read_varint(&wire, &mut pos)? as usize;
+    let digest = wire.get(pos..)?;
+    (digest.len() == len).then(|| digest.to_vec())
+}
+
+/// Unsigned LEB128 varint at `*pos`, advancing it.
+fn read_varint(bytes: &[u8], pos: &mut usize) -> Option<u64> {
+    let mut value = 0u64;
+    let mut shift = 0u32;
+    loop {
+        let byte = *bytes.get(*pos)?;
+        *pos += 1;
+        value |= u64::from(byte & 0x7f) << shift;
+        if byte & 0x80 == 0 {
+            return Some(value);
+        }
+        shift += 7;
+        if shift >= 64 {
+            return None;
+        }
+    }
+}
+
 /// 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).
@@ -248,8 +318,12 @@ fn decode_hex_id(id: &str) -> Option<Vec<u8>> {
 ///   merge (a TMDB and a MyAnimeList locator on one meta-core record) safe.
 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.
+/// Digest byte ceiling for the locators that still carry one — `yt-video`
+/// (via [`kinded_locator_cid`]) and `ext-play`. Matches meta-share's
+/// `MAX_MULTIHASH_SIZE`, whose `CidGeneric<64>` rejects a longer multihash.
+///
+/// ⚠ **[`compute_card_cid`] is deliberately NOT subject to this** — see its
+/// doc. The name is kept because both remaining users were written against it.
 const CARD_LOCATOR_MAX_DIGEST: usize = 64;
 
 /// Encode a self-describing **card locator** CID from a metadata source and
@@ -270,8 +344,28 @@ const CARD_LOCATOR_MAX_DIGEST: usize = 64;
 /// 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.
+/// # No length ceiling, deliberately
+///
+/// A CIDv1 digest length is a varint, so the wire format imposes no limit; the
+/// 64-byte figure elsewhere in this file is Rust's fixed-size `CidGeneric<64>`,
+/// which meta-share needs only because beetswap is
+/// `Behaviour<const MAX_MULTIHASH_SIZE, B>`. A card is never bitswap content
+/// (seeding is gated on sha2-256), so that ceiling does not apply — exactly the
+/// reasoning `METADATA_KEYS.md` §2.1 already records for the `url` locator,
+/// which ships URLs past it today. meta-core's Go `Decode` hand-parses varints
+/// with no cap and reads these correctly.
+///
+/// This is load-bearing for URL-shaped ids: a card keyed on a work's own
+/// address (`("weebcentral.com", "/series/…/some-long-slug")`) runs to ~88
+/// bytes, and capping would silently drop the card rather than mint it.
+///
+/// ⚠ The one consumer that still cares is meta-share's `cid_rank`, which parses
+/// through `MsCid::from_str`. A locator it cannot parse scores `RANK_UNKNOWN`
+/// instead of `RANK_LOCATOR`; harmless for a card (its record carries one CID,
+/// so it wins the canonical election at any rank) but pinned by a golden vector
+/// in `cid-rank-vectors.json` so the two implementations cannot drift unseen.
+///
+/// Returns `None` only when `source` or `id` is empty.
 pub fn compute_card_cid(source: &str, id: &str) -> Option<String> {
     if source.is_empty() || id.is_empty() {
         return None;
@@ -284,9 +378,6 @@ pub fn compute_card_cid(source: &str, id: &str) -> Option<String> {
     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());
@@ -378,8 +469,39 @@ pub fn compute_ext_play_cid(url: &str) -> Option<String> {
     Some(identity_locator_cid(EXT_PLAY_CODEC, digest))
 }
 
+/// Custom multicodec for a fetchable **`provider-file`** locator —
+/// `docs/cid-formats.md` §8. One codec for every provider whose bytes are
+/// fetched the same way; the provider lives inside the digest:
+/// `varint(len(source)) ‖ source ‖ id`. Framing byte-identical to `card`
+/// (`0x1007`) — only the codec differs — but the byte path *redeems* it (through
+/// the claiming feeder's `/compute`) instead of 404ing.
+pub const PROVIDER_FILE_CODEC: u64 = 0x100A;
+
+/// Encode a `provider-file` CID from a registered source token
+/// (`opensubtitles`) and that source's id (`file:7061834`). `None` on an empty
+/// part or a digest past 64 bytes — the caller drops that row rather than emit
+/// an address no peer can parse.
+pub fn compute_provider_file_cid(source: &str, id: &str) -> Option<String> {
+    kinded_locator_cid(PROVIDER_FILE_CODEC, source, id)
+}
+
+/// Inverse of [`compute_provider_file_cid`]: `(source, id)`. Matches the
+/// **codec slot** — a `card` with identical framing does not decode here.
+pub fn decode_provider_file_cid(cid: &str) -> Option<(String, String)> {
+    let digest = identity_locator_digest(cid, PROVIDER_FILE_CODEC)?;
+    let mut pos = 0usize;
+    let source_len = read_varint(&digest, &mut pos)? as usize;
+    let source = digest.get(pos..pos.checked_add(source_len)?)?;
+    let id = digest.get(pos + source_len..)?;
+    Some((
+        String::from_utf8(source.to_vec()).ok()?,
+        String::from_utf8(id.to_vec()).ok()?,
+    ))
+}
+
 /// Shared encoder for the `varint(len(kind)) ‖ kind ‖ id` locator framing used
-/// by [`compute_card_cid`] and [`compute_yt_video_cid`].
+/// by [`compute_card_cid`], [`compute_yt_video_cid`] and
+/// [`compute_provider_file_cid`].
 fn kinded_locator_cid(codec: u64, kind: &str, id: &str) -> Option<String> {
     if kind.is_empty() || id.is_empty() {
         return None;
@@ -1067,19 +1189,88 @@ mod tests {
         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.
+    /// Empty inputs are rejected rather than encoded as an ambiguous locator.
     #[test]
-    fn card_cid_rejects_empty_and_oversized() {
+    fn card_cid_rejects_empty_halves() {
         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());
+    }
+
+    /// ⚠ A card locator has **no length ceiling**, and this is the test that
+    /// says so on purpose — it used to assert the opposite.
+    ///
+    /// The 64-byte figure is meta-share's `CidGeneric<64>` bitswap alias, and a
+    /// card is never bitswap content. Capping here would silently drop exactly
+    /// the cards a URL-shaped identity exists to mint: a `weebcentral.com` work
+    /// path digests to ~88 bytes. Same reasoning `METADATA_KEYS.md` §2.1
+    /// already records for the `url` locator.
+    #[test]
+    fn card_cid_has_no_length_ceiling() {
+        let long_id = "/series/01J76XYGF1RKSPWA4TQ6GGSVFV/\
+                       Ive-Been-Killing-Slimes-for-300-Years-The-Red-Dragon-Academy-for-Girls";
+        let cid = compute_card_cid("weebcentral.com", long_id).expect("no ceiling");
+        let (source, id) = split_card_digest(&cid);
+        assert!(
+            source.len() + id.len() > CARD_LOCATOR_MAX_DIGEST,
+            "fixture must actually exceed the old cap, got {}",
+            source.len() + id.len()
+        );
+        assert_eq!(source, "weebcentral.com");
+        assert_eq!(id, long_id);
+    }
+
+    /// The host-form identity: the source slot names the site that *owns* the
+    /// work, never the tool that reached it. Pinned as a literal because
+    /// `meta-read`'s `card_cid::encode` carries the mirrored assertion — the two
+    /// addressing one work differently is the failure the whole scheme cannot
+    /// survive.
+    #[test]
+    fn host_form_card_cid_golden_vector() {
+        let cid = compute_card_cid("mangadex.org", "/manga/b0b721ff-c388-4486-aa0f-c2b0bb321512")
+            .expect("card cid");
+        assert_eq!(
+            cid,
+            "bagdsaabybrwwc3thmfsgk6bon5zgol3nmfxgoyjpmiygenzsgftgmlldgm4dqljugq4dmllbmeygmlldgjrdaytcgmzdcnjrgi"
+        );
+    }
+
+    /// `varint(source_len) ‖ source ‖ id` back out of a card locator, so the
+    /// tests above can assert on the halves without a decoder in this crate.
+    fn split_card_digest(cid: &str) -> (String, String) {
+        let raw = base32_lower_decode(&cid[1..]);
+        let mut i = 0usize;
+        let mut rd = || {
+            let (mut n, mut sh) = (0u64, 0u32);
+            loop {
+                let b = raw[i];
+                i += 1;
+                n |= u64::from(b & 0x7f) << sh;
+                if b & 0x80 == 0 {
+                    return n;
+                }
+                sh += 7;
+            }
+        };
+        assert_eq!(rd(), 1, "CIDv1");
+        assert_eq!(rd(), CARD_LOCATOR_CODEC);
+        assert_eq!(rd(), 0, "identity multihash");
+        let len = rd() as usize;
+        let digest = &raw[i..i + len];
+        let mut j = 0usize;
+        let (mut sl, mut sh) = (0usize, 0u32);
+        loop {
+            let b = digest[j];
+            j += 1;
+            sl |= usize::from(b & 0x7f) << sh;
+            if b & 0x80 == 0 {
+                break;
+            }
+            sh += 7;
+        }
+        (
+            String::from_utf8(digest[j..j + sl].to_vec()).unwrap(),
+            String::from_utf8(digest[j + sl..].to_vec()).unwrap(),
+        )
     }
 
     // ---- nzb-posting (0x1003) — the dedup contract -----------------------
@@ -1299,3 +1490,84 @@ mod roundtrip_tests {
         assert_eq!(decode_url_cid(&card), None);
     }
 }
+
+#[cfg(test)]
+mod redeem_locator_tests {
+    use super::*;
+
+    /// `docs/cid-formats.md` §8.2 golden vector — moved here from
+    /// meta-feeder-opensubtitles' local `locator.rs`; the vector proves the
+    /// move changed no bytes.
+    #[test]
+    fn provider_file_matches_the_documented_vector() {
+        assert_eq!(
+            compute_provider_file_cid("opensubtitles", "file:7061834").as_deref(),
+            Some("bagfcaaa2bvxxazloon2we5djorwgk43gnfwgkorxga3dcobtgq")
+        );
+    }
+
+    #[test]
+    fn provider_file_round_trips_and_refuses_other_codecs() {
+        let cid = compute_provider_file_cid("opensubtitles", "file:12658754").unwrap();
+        assert_eq!(
+            decode_provider_file_cid(&cid),
+            Some(("opensubtitles".to_string(), "file:12658754".to_string()))
+        );
+        // Identical framing, different codec: a card must not decode as a file.
+        let card = compute_card_cid("tmdb", "tv:95479").unwrap();
+        assert_eq!(card, "bagdsaaanar2g2zdcor3duojvgq3ts");
+        assert_eq!(decode_provider_file_cid(&card), None);
+        assert_eq!(decode_provider_file_cid("not-a-cid"), None);
+        assert_eq!(decode_provider_file_cid(""), None);
+    }
+
+    #[test]
+    fn provider_file_digest_ceiling_is_64_bytes() {
+        // varint(13) + "opensubtitles" = 14 bytes, so the id may use 50.
+        assert!(compute_provider_file_cid("opensubtitles", &"x".repeat(50)).is_some());
+        assert!(compute_provider_file_cid("opensubtitles", &"x".repeat(51)).is_none());
+    }
+
+    #[test]
+    fn nzb_release_round_trips() {
+        let cid = compute_nzb_release_cid("api.nzbgeek.info/api", "0123abcdef").unwrap();
+        assert_eq!(
+            decode_nzb_release_cid(&cid),
+            Some(("api.nzbgeek.info/api".to_string(), "0123abcdef".to_string()))
+        );
+        // A host:port authority survives too.
+        let cid = compute_nzb_release_cid("127.0.0.1:8080", "ff00").unwrap();
+        assert_eq!(
+            decode_nzb_release_cid(&cid),
+            Some(("127.0.0.1:8080".to_string(), "ff00".to_string()))
+        );
+    }
+
+    /// The encoder left-pads an odd-length id, so the decode is the padded form.
+    #[test]
+    fn nzb_release_odd_length_id_comes_back_padded() {
+        let cid = compute_nzb_release_cid("api.nzb.life", "abc").unwrap();
+        assert_eq!(
+            decode_nzb_release_cid(&cid),
+            Some(("api.nzb.life".to_string(), "0abc".to_string()))
+        );
+    }
+
+    #[test]
+    fn nzb_release_refuses_other_codecs() {
+        let card = compute_card_cid("tmdb", "tv:95479").unwrap();
+        assert_eq!(decode_nzb_release_cid(&card), None);
+        assert_eq!(decode_nzb_release_cid("bafkreigarbage"), None);
+    }
+
+    #[test]
+    fn codec_of_reads_the_codec_slot() {
+        let nzb = compute_nzb_release_cid("api.nzb.life", "ab").unwrap();
+        assert_eq!(codec_of(&nzb), Some(NZB_RELEASE_CODEC));
+        let file = compute_provider_file_cid("opensubtitles", "file:1").unwrap();
+        assert_eq!(codec_of(&file), Some(PROVIDER_FILE_CODEC));
+        assert_eq!(codec_of(&compute_ipfs_cid(b"hello")), Some(0x55));
+        assert_eq!(codec_of("QmYwAPJzv5CZsnA625s3Xf2nemtYgPpHdWEz79ojWnPbdG"), None);
+        assert_eq!(codec_of(""), None);
+    }
+}

src/lib.rs

@@ -45,10 +45,12 @@ pub use enrich::{EnrichTarget, EnrichmentConfig, Enricher};
 pub use meta_core::FeederStore;
 pub use plugin::{
     upstream_id_field, ConfigError, FeederPlugin, HashKind, HashOutcome, PluginRegistry,
+    RedeemClaim,
 };
 pub use query::{GatewayQuery, GatewaySearchEvent, GatewayWireError, Negation, RangeFilter};
 pub use serve::{
     configure_plugins, router, serve_feeders, ComputeRequest, ComputeResponse, HashKindDto,
     HealthResponse, ManifestResponse, OutcomeDto, PluginManifest, QueryRequest, QueryResponse,
+    RedeemsResponse,
 };
 pub use types::{ByteStream, DiscoveryId, DiscoveryRecord, GatewayError, Hash, PluginHealth};

src/plugin.rs

@@ -17,6 +17,7 @@ use std::path::Path;
 
 use async_trait::async_trait;
 use futures::stream::BoxStream;
+use serde::{Deserialize, Serialize};
 
 // Re-exported so plugins can `use meta_feeder_sdk::plugin::GatewayQuery` (the
 // gateway's original `plugin.rs` re-exported it the same way).
@@ -95,6 +96,66 @@ pub enum HashKind {
     YtVideo,
 }
 
+/// `RedeemClaim::codec` for a Newznab release locator (`0x1005`).
+pub const REDEEM_CODEC_NZB_RELEASE: &str = "nzb-release";
+/// `RedeemClaim::codec` for a `provider-file` locator (`0x100A`).
+pub const REDEEM_CODEC_PROVIDER_FILE: &str = "provider-file";
+/// `RedeemClaim::field` naming the `.nzb` manifest pointer on a release record.
+pub const POINTER_FIELD_MANIFEST: &str = "manifest";
+/// `RedeemClaim::field` naming the fetched-file pointer on a locator record.
+pub const POINTER_FIELD_FILE: &str = "file";
+
+/// A plugin's claim that it can **redeem** a locator family: turn a locator cid
+/// into the bytes it names, through `POST /compute` with `record_id` = the
+/// locator cid.
+///
+/// Redeeming usually spends a metered upstream quota (a Newznab `t=get` grab,
+/// an OpenSubtitles `/download`), so the gateway routes a redeem only to the
+/// plugins claiming the cid's codec, one at a time, and only on a real play. It
+/// stores the returned bytes and writes their cid onto the locator record under
+/// [`field`](Self::field), so the quota is spent once per file.
+///
+/// Served at `GET /manifest` (per plugin) and `GET /redeems`. Reflects the
+/// plugin's *current* config: a plugin with no credential for a family claims
+/// nothing for it.
+#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
+pub struct RedeemClaim {
+    /// Locator family: [`REDEEM_CODEC_NZB_RELEASE`] or [`REDEEM_CODEC_PROVIDER_FILE`].
+    pub codec: String,
+    /// Pointer field the gateway writes the redeemed bytes' cid under:
+    /// [`POINTER_FIELD_MANIFEST`] or [`POINTER_FIELD_FILE`].
+    pub field: String,
+    /// Indexer hosts this plugin holds a key for (`nzb-release`) — bare
+    /// authority, lowercase, never the key itself.
+    #[serde(default)]
+    pub hosts: Vec<String>,
+    /// Provider sources this plugin redeems (`provider-file`), e.g. `opensubtitles`.
+    #[serde(default)]
+    pub sources: Vec<String>,
+}
+
+impl RedeemClaim {
+    /// Claim `nzb-release` locators for these indexer hosts.
+    pub fn nzb_release(hosts: Vec<String>) -> Self {
+        Self {
+            codec: REDEEM_CODEC_NZB_RELEASE.to_string(),
+            field: POINTER_FIELD_MANIFEST.to_string(),
+            hosts,
+            sources: Vec::new(),
+        }
+    }
+
+    /// Claim `provider-file` locators for these sources.
+    pub fn provider_file(sources: Vec<String>) -> Self {
+        Self {
+            codec: REDEEM_CODEC_PROVIDER_FILE.to_string(),
+            field: POINTER_FIELD_FILE.to_string(),
+            hosts: Vec::new(),
+            sources,
+        }
+    }
+}
+
 /// Startup-only configuration error returned by [`FeederPlugin::configure`].
 #[derive(Debug, thiserror::Error)]
 pub enum ConfigError {
@@ -215,6 +276,20 @@ pub trait FeederPlugin: Send + Sync + 'static {
         &[]
     }
 
+    /// Folder under meta-core's `/files/plugin/` where the gateway stores bytes
+    /// this plugin redeems (e.g. `meta-feeder-usenet`). Stable across
+    /// deployments — unlike the feeder's compose service name. Default: none.
+    fn package(&self) -> Option<&'static str> {
+        None
+    }
+
+    /// Locator families this plugin can redeem through
+    /// [`compute_outcomes`](FeederPlugin::compute_outcomes) — see
+    /// [`RedeemClaim`]. Computed from the current config. Default: none.
+    fn redeems(&self) -> Vec<RedeemClaim> {
+        Vec::new()
+    }
+
     /// Self-describing config schema. The feeder SDK serves this at
     /// `GET /config/schema` and renders a generic form from it — the gateway and
     /// UI carry no per-plugin field knowledge. Default: no configuration.
@@ -255,4 +330,23 @@ mod tests {
         assert!(msg.contains("scihub"));
         assert!(msg.contains("mirrors"));
     }
+
+    /// The JSON keys are the gateway/meta-share contract — pin them.
+    #[test]
+    fn redeem_claim_wire_shape() {
+        let claim = RedeemClaim::nzb_release(vec!["api.nzb.life".into()]);
+        assert_eq!(
+            serde_json::to_value(&claim).unwrap(),
+            serde_json::json!({
+                "codec": "nzb-release",
+                "field": "manifest",
+                "hosts": ["api.nzb.life"],
+                "sources": []
+            })
+        );
+        // hosts/sources may be omitted on the wire.
+        let parsed: RedeemClaim =
+            serde_json::from_str(r#"{"codec":"provider-file","field":"file"}"#).unwrap();
+        assert_eq!(parsed, RedeemClaim::provider_file(vec![]));
+    }
 }

src/serve.rs

@@ -10,6 +10,17 @@
 //! - `POST /compute`                   → [`ComputeRequest`] → [`ComputeResponse`]
 //! - `GET  /fetch/{upstream_id}/{record_id}` → streamed bytes (or 404)
 //! - `GET  /blob/{upstream_id}/{cid}`  → bytes (or 404)
+//! - `GET  /redeems`                   → [`RedeemsResponse`] (current redeem claims)
+//!
+//! ## Redeeming a locator
+//!
+//! A plugin that advertises a [`RedeemClaim`] answers `POST /compute` with
+//! `record_id` = a locator cid of that family (a Newznab `nzb-release`, an
+//! OpenSubtitles `provider-file`) by fetching the bytes it names — usually
+//! spending a metered quota — and returning ONE `Sha2_256` outcome carrying
+//! them. `NotFound` means "not mine" (e.g. no key for that indexer host), so the
+//! gateway tries the next claimer. The claims are re-read from `GET /redeems`
+//! so a config change shows up without restarting the gateway.
 //!
 //! ## Bytes & the v1 simplification
 //!
@@ -21,7 +32,7 @@
 //! documented follow-up — those outcomes are metadata-only anyway
 //! (`bytes: None`), so they never ride this path.
 
-use std::collections::HashMap;
+use std::collections::{BTreeMap, HashMap};
 use std::net::SocketAddr;
 use std::path::{Path as FsPath, PathBuf};
 use std::sync::Arc;
@@ -41,7 +52,7 @@ use serde_json::Value;
 use tracing::{info, warn};
 
 use crate::config::{self, CONFIG_PAGE_HTML};
-use crate::plugin::{ConfigError, FeederPlugin, HashKind, HashOutcome};
+use crate::plugin::{ConfigError, FeederPlugin, HashKind, HashOutcome, RedeemClaim};
 use crate::query::{GatewayQuery, GatewaySearchEvent, GatewayWireError};
 use crate::types::{DiscoveryRecord, GatewayError, PluginHealth};
 
@@ -94,6 +105,22 @@ pub struct PluginManifest {
     pub id: String,
     pub served_file_types: Vec<String>,
     pub served_content_kinds: Vec<String>,
+    /// [`FeederPlugin::package`] — the `/files/plugin/<package>/` folder redeemed
+    /// bytes are stored under. `null` when the plugin redeems nothing.
+    #[serde(default)]
+    pub package: Option<String>,
+    /// [`FeederPlugin::redeems`] at the time the manifest was served.
+    #[serde(default)]
+    pub redeems: Vec<RedeemClaim>,
+}
+
+/// `GET /redeems` — every hosted plugin's current [`RedeemClaim`]s, keyed by
+/// `upstream_id` (an empty list for a plugin that claims nothing). Polled by
+/// the gateway, so a credential added on the config page is advertised without
+/// a gateway restart.
+#[derive(Serialize, Deserialize, Clone, Debug, Default)]
+pub struct RedeemsResponse {
+    pub redeems: BTreeMap<String, Vec<RedeemClaim>>,
 }
 
 /// `GET /manifest` — the feeder's hosted plugins + their served types. The
@@ -224,6 +251,17 @@ impl AppState {
 /// status semantics the gateway's `RemoteFeederPlugin` will translate back
 /// into a `GatewayError` on the core side.
 fn gateway_error_response(e: GatewayError) -> Response {
+    // A 429 carries the wait as a `Retry-After` header too, so the gateway can
+    // forward it to the client (a redeem blocked on a spent quota) instead of
+    // parsing the message.
+    if let GatewayError::RateLimited { retry_after_s } = &e {
+        return (
+            StatusCode::TOO_MANY_REQUESTS,
+            [(axum::http::header::RETRY_AFTER, retry_after_s.to_string())],
+            format!("rate limited; retry in {retry_after_s}s"),
+        )
+            .into_response();
+    }
     let (status, msg) = match e {
         GatewayError::NotFound => (StatusCode::NOT_FOUND, "not found".to_string()),
         GatewayError::RateLimited { retry_after_s } => (
@@ -251,6 +289,8 @@ async fn manifest(State(state): State<AppState>) -> Json<ManifestResponse> {
                 .iter()
                 .map(|s| s.to_string())
                 .collect(),
+            package: p.package().map(str::to_string),
+            redeems: p.redeems(),
         })
         .collect();
     plugins.sort_by(|a, b| a.id.cmp(&b.id));
@@ -260,6 +300,16 @@ async fn manifest(State(state): State<AppState>) -> Json<ManifestResponse> {
     })
 }
 
+async fn redeems(State(state): State<AppState>) -> Json<RedeemsResponse> {
+    Json(RedeemsResponse {
+        redeems: state
+            .plugins
+            .values()
+            .map(|p| (p.upstream_id().to_string(), p.redeems()))
+            .collect(),
+    })
+}
+
 async fn health(State(state): State<AppState>) -> Json<HealthResponse> {
     let mut entries: Vec<PluginHealthEntry> = state
         .plugins
@@ -492,6 +542,7 @@ pub fn router(
         .route("/compute", post(compute))
         .route("/fetch/:upstream_id/:record_id", get(fetch))
         .route("/blob/:upstream_id/:cid", get(blob))
+        .route("/redeems", get(redeems))
         .route("/config", get(config_page))
         .route("/config/schema", get(config_schema))
         .route("/config/values", get(config_values_get).put(config_values_put))
@@ -541,6 +592,9 @@ struct DegradedPlugin {
     reason: String,
     file_types: &'static [&'static str],
     content_kinds: &'static [&'static str],
+    /// Preserved so the manifest keeps naming the plugin's folder. A degraded
+    /// plugin claims no redeems (it cannot redeem anything).
+    package: Option<&'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.
@@ -586,6 +640,10 @@ impl FeederPlugin for DegradedPlugin {
         self.content_kinds
     }
 
+    fn package(&self) -> Option<&'static str> {
+        self.package
+    }
+
     fn config_schema(&self) -> crate::config::ConfigSchema {
         self.schema.clone()
     }
@@ -608,6 +666,7 @@ pub fn configure_plugins(
         // borrow across `configure(&mut self)` / the move into `out`.
         let file_types = plugin.served_file_types();
         let content_kinds = plugin.served_content_kinds();
+        let package = plugin.package();
         // Preserve the config surface so a degraded stand-in still serves the
         // real config form (see DegradedPlugin.schema).
         let schema = plugin.config_schema();
@@ -627,6 +686,7 @@ pub fn configure_plugins(
                 reason,
                 file_types,
                 content_kinds,
+                package,
                 schema: schema.clone(),
             })
         };
@@ -759,4 +819,118 @@ mod tests {
         assert!(matches!(events[1], GatewaySearchEvent::EnrichPatch { .. }));
         assert!(matches!(events[2], GatewaySearchEvent::Done));
     }
+
+    /// A plugin that redeems `nzb-release` locators for one host.
+    struct StubRedeemer;
+
+    #[async_trait]
+    impl FeederPlugin for StubRedeemer {
+        fn upstream_id(&self) -> &'static str {
+            "redeemer"
+        }
+        fn configure(&mut self, _cache_dir: &FsPath) -> Result<(), ConfigError> {
+            Ok(())
+        }
+        async fn handle_query(
+            &self,
+            _q: &GatewayQuery,
+            _n: usize,
+        ) -> Result<Vec<DiscoveryRecord>, GatewayError> {
+            Ok(vec![])
+        }
+        async fn compute_outcomes(
+            &self,
+            _record_id: &str,
+        ) -> Result<Vec<HashOutcome>, GatewayError> {
+            Ok(vec![])
+        }
+        fn package(&self) -> Option<&'static str> {
+            Some("meta-feeder-stub")
+        }
+        fn redeems(&self) -> Vec<RedeemClaim> {
+            vec![RedeemClaim::nzb_release(vec!["api.example".into()])]
+        }
+    }
+
+    async fn serve_stubs() -> String {
+        let mut plugins: HashMap<String, Arc<dyn FeederPlugin>> = HashMap::new();
+        plugins.insert("redeemer".to_string(), Arc::new(StubRedeemer));
+        plugins.insert("stub".to_string(), Arc::new(StubStreaming));
+        let app = router(plugins, "test".to_string(), std::env::temp_dir());
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let addr = listener.local_addr().unwrap();
+        tokio::spawn(async move {
+            axum::serve(listener, app).await.unwrap();
+        });
+        format!("http://{addr}")
+    }
+
+    /// The manifest carries `package` + `redeems` per plugin; a plugin that
+    /// claims nothing serializes `null` + `[]`.
+    #[tokio::test]
+    async fn manifest_carries_package_and_redeems() {
+        let base = serve_stubs().await;
+        let m: serde_json::Value = reqwest::get(format!("{base}/manifest"))
+            .await
+            .unwrap()
+            .json()
+            .await
+            .unwrap();
+        let redeemer = &m["plugins"][0];
+        assert_eq!(redeemer["id"], "redeemer");
+        assert_eq!(redeemer["package"], "meta-feeder-stub");
+        assert_eq!(
+            redeemer["redeems"],
+            serde_json::json!([{
+                "codec": "nzb-release", "field": "manifest",
+                "hosts": ["api.example"], "sources": []
+            }])
+        );
+        let stub = &m["plugins"][1];
+        assert_eq!(stub["package"], serde_json::Value::Null);
+        assert_eq!(stub["redeems"], serde_json::json!([]));
+    }
+
+    #[tokio::test]
+    async fn redeems_route_keys_claims_by_upstream_id() {
+        let base = serve_stubs().await;
+        let r: RedeemsResponse = reqwest::get(format!("{base}/redeems"))
+            .await
+            .unwrap()
+            .json()
+            .await
+            .unwrap();
+        assert_eq!(
+            r.redeems.get("redeemer"),
+            Some(&vec![RedeemClaim::nzb_release(vec!["api.example".into()])])
+        );
+        assert_eq!(r.redeems.get("stub"), Some(&vec![]));
+    }
+
+    /// The gateway forwards a feeder 429's `Retry-After` — pin that it is set.
+    #[test]
+    fn rate_limited_sets_retry_after() {
+        let resp = gateway_error_response(GatewayError::RateLimited { retry_after_s: 3600 });
+        assert_eq!(resp.status(), StatusCode::TOO_MANY_REQUESTS);
+        assert_eq!(
+            resp.headers()
+                .get(axum::http::header::RETRY_AFTER)
+                .and_then(|v| v.to_str().ok()),
+            Some("3600")
+        );
+        // Other errors carry no Retry-After.
+        let resp = gateway_error_response(GatewayError::NotFound);
+        assert_eq!(resp.status(), StatusCode::NOT_FOUND);
+        assert!(resp.headers().get(axum::http::header::RETRY_AFTER).is_none());
+    }
+
+    /// A manifest from a pre-1.3 feeder (no `package` / `redeems`) still parses.
+    #[test]
+    fn old_manifest_without_redeem_fields_parses() {
+        let m: PluginManifest = serde_json::from_str(
+            r#"{"id":"x","served_file_types":[],"served_content_kinds":[]}"#,
+        )
+        .unwrap();
+        assert!(m.package.is_none() && m.redeems.is_empty());
+    }
 }