align with metadata standard

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/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/hash.rs

@@ -183,9 +183,9 @@ const NZB_RELEASE_MAX_DIGEST: usize = 64;
 ///
 /// 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(host: &str, release_id: &str) -> Option<String> {
+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 = host.as_bytes();
+    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());
@@ -298,6 +298,138 @@ pub fn compute_card_cid(source: &str, id: &str) -> Option<String> {
     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 the **url-locator** family: a CID whose bytes are an
+/// `http(s)` URL rather than a hash of the file (METADATA_KEYS.md §2.1), so a
+/// record references external bytes without carrying a raw URL field.
+///
+/// Copied from the git `meta-feeder-sdk` (`hash.rs`, present since v1.1.0),
+/// which this vendored copy predates. Keep the two byte-identical.
+pub const URL_LOCATOR_CODEC: u64 = 0x1006;
+
+/// Encode an `http(s)` URL as a `0x1006` identity-multihash CIDv1. The digest
+/// *is* the URL, so minting one costs no I/O.
+///
+/// `None` for anything that is not an absolute http(s) URL — a relative value
+/// would encode fine and then fail unresolvably at render time.
+pub fn compute_url_cid(url: &str) -> Option<String> {
+    let url = url.trim();
+    if !(url.starts_with("http://") || url.starts_with("https://")) {
+        return None;
+    }
+    Some(identity_locator_cid(URL_LOCATOR_CODEC, url.as_bytes()))
+}
+
 /// Custom multicodec for a **Usenet posting** identity (`nzb-posting`),
 /// minted from the article Message-IDs of a release we scanned ourselves.
 ///
@@ -535,6 +667,26 @@ pub(crate) fn base32_lower_no_padding(input: &[u8]) -> String {
 mod tests {
     use super::*;
 
+    /// Pinned against an independent encoder. The tmdb plugin's TS
+    /// `encodeUrlCid` asserts the same strings, so a drift on either side fails.
+    /// The long URL exercises the two-byte length varint (> 127 bytes).
+    #[test]
+    fn url_cid_matches_pinned_vectors() {
+        assert_eq!(
+            compute_url_cid("https://image.tmdb.org/t/p/w500/dqZENchTd7lp5zit1Q7Bkjzcxpi.jpg")
+                .as_deref(),
+            Some("bagdcaab7nb2hi4dthixs62lnmftwkltunvsgeltpojts65bpoaxxonjqgaxwi4k2ivhgg2cumq3wy4bvpjuximkrg5bgw2t2mn4ha2jonjygo")
+        );
+        let long = format!("https://image.tmdb.org/t/p/w500/{}.jpg", "a".repeat(100));
+        assert_eq!(long.len(), 136);
+        assert_eq!(
+            compute_url_cid(&long).as_deref(),
+            Some("bagdcaaeiafuhi5dqom5c6l3jnvqwozjoorwwiyron5zgol3uf5yc65zvgayc6ylbmfqwcylbmfqwcylbmfqwcylbmfqwcylbmfqwcylbmfqwcylbmfqwcylbmfqwcylbmfqwcylbmfqwcylbmfqwcylbmfqwcylbmfqwcylbmfqwcylbmfqwcylbmfqwcylbmfqwcylbmfqwcylbmfqwcylbmfqwcltkobtq")
+        );
+        assert!(compute_url_cid("/relative.jpg").is_none());
+        assert!(compute_url_cid("ftp://example.com/x.jpg").is_none());
+    }
+
     #[test]
     fn midhash256_matches_ts_fixtures() {
         assert_eq!(
@@ -672,6 +824,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!(

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

@@ -24,6 +24,7 @@ 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

@@ -74,6 +74,13 @@ pub struct HashOutcome {
 ///   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
@@ -85,6 +92,7 @@ pub enum HashKind {
     NzbRelease,
     CardLocator,
     NzbPosting,
+    YtVideo,
 }
 
 /// Startup-only configuration error returned by [`FeederPlugin::configure`].
@@ -121,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/serve.rs

@@ -57,6 +57,7 @@ pub enum HashKindDto {
     NzbRelease,
     CardLocator,
     NzbPosting,
+    YtVideo,
 }
 
 impl From<HashKind> for HashKindDto {
@@ -68,6 +69,7 @@ impl From<HashKind> for HashKindDto {
             HashKind::NzbRelease => HashKindDto::NzbRelease,
             HashKind::CardLocator => HashKindDto::CardLocator,
             HashKind::NzbPosting => HashKindDto::NzbPosting,
+            HashKind::YtVideo => HashKindDto::YtVideo,
         }
     }
 }
@@ -81,6 +83,7 @@ impl From<HashKindDto> for HashKind {
             HashKindDto::NzbRelease => HashKind::NzbRelease,
             HashKindDto::CardLocator => HashKind::CardLocator,
             HashKindDto::NzbPosting => HashKind::NzbPosting,
+            HashKindDto::YtVideo => HashKind::YtVideo,
         }
     }
 }

feeder-plugin/card-feeder/src/card.rs

@@ -17,11 +17,14 @@
 use std::collections::BTreeMap;
 use std::sync::Arc;
 
-use meta_feeder_sdk::hash::compute_card_cid;
+use meta_feeder_sdk::hash::{compute_card_cid, compute_url_cid};
 use meta_feeder_sdk::types::DiscoveryRecord;
 
 use crate::consts::MAX_CARD_SEASONS;
-use crate::tmdb_client::{TmdbClient, TmdbKind, TmdbSeasonSummary};
+use crate::tmdb_client::{
+    iso639_1_to_3, norm_tokens, normalize_title_name, title_member_key, TmdbClient, TmdbImage,
+    TmdbKind, TmdbSeasonSummary,
+};
 
 /// The metadata source that published a card. One variant per bridge; the
 /// string form is the CID's source namespace, so **changing it re-mints every
@@ -65,14 +68,17 @@ pub(crate) struct Card {
     pub(crate) title: String,
     /// Canonical + original + AKA titles. Rides along so the *content* feeders
     /// can run their resemblance guard against the card instead of re-querying
-    /// TMDB for it.
+    /// TMDB for it (phase 2, once the indexer feeder's own anchor resolution is
+    /// retired).
     ///
-    /// Not read here yet — it is consumed by phase 2, which still resolves its
-    /// own anchor until the indexer feeder's `build_anchor_record` path is
-    /// retired. Resolving it now means the card is already complete when that
-    /// switch happens.
-    #[allow(dead_code)]
+    /// Read here for one thing: picking `searchTitle` ([`search_aka`]).
     pub(crate) guard_titles: Arc<Vec<String>>,
+    /// Every name of the work as `(lang3, name)`, filed on the record as the
+    /// `titles/{lang3}/{name}` key-set (METADATA_KEYS.md). Built by
+    /// `tmdb_client::title_names`: original under its language, `title` under
+    /// `eng` when it differs, AKAs by market. Same names as `guard_titles`, with
+    /// the language each is written in.
+    pub(crate) names: Vec<(&'static str, String)>,
     pub(crate) overview: Option<String>,
     pub(crate) poster_path: Option<String>,
     /// Editorial genre names (`"Animation"`, `"Drama"`, …) — METADATA_KEYS
@@ -98,6 +104,11 @@ pub(crate) struct Card {
     /// from a mutated series card.
     #[allow(dead_code)]
     pub(crate) season_summaries: Arc<Vec<TmdbSeasonSummary>>,
+
+    /// Poster candidates (TMDB `images.posters`), filed as the
+    /// `posters/{lang3}/{cid}` key-set by [`poster_member_keys`]. Only the by-id
+    /// path has them; discovery/search cards leave this empty and file no set.
+    pub(crate) posters: Arc<Vec<TmdbImage>>,
 }
 
 impl Card {
@@ -181,8 +192,40 @@ impl Card {
         // contentKind axis stay orthogonal, so `fileType:card contentKind:series`
         // is a well-formed query and routing needs no special case.
         fields.insert("fileType".to_string(), "card".to_string());
-        fields.insert("contentKind".to_string(), self.content_kind().to_string());
+        let content_kind = self.content_kind();
+        fields.insert("contentKind".to_string(), content_kind.to_string());
+        // Third and fourth axes: which app the card is destined for, and what
+        // shape of work it is. `fileType=card` says "no bytes", `contentKind`
+        // says which work, `domain` says who wants it — meta-watch's wall is a
+        // `domain:screen` query — and `workForm` says film-or-serial, the split
+        // `domain` stopped carrying when film and tv merged (METADATA_KEYS.md
+        // §1, §14.17).
+        if let Some(domain) = meta_feeder_sdk::domain::domain_for_content_kind(content_kind) {
+            fields.insert("domain".to_string(), domain.to_string());
+        }
+        if let Some(work_form) =
+            meta_feeder_sdk::domain::work_form_for_content_kind(content_kind)
+        {
+            fields.insert("workForm".to_string(), work_form.to_string());
+        }
         fields.insert("title".to_string(), self.title.clone());
+        // The keyword a consumer should search indexers with, only when `title`
+        // itself can't be searched (`3%` → `3 percent`). meta-watch's title page
+        // has no TMDB access of its own and reads this off the card; without it
+        // its free-text query sends `3%`, which one newznab answers with nothing
+        // and another floods with every release containing a `3`.
+        if let Some(aka) = search_aka(&self.title, self.guard_titles.iter()) {
+            // Normalised like a `titles/*/*` member name: `searchTitle` is always
+            // one of the record's names, spelled the same way.
+            fields.insert("searchTitle".to_string(), normalize_title_name(aka));
+        }
+        // Every name the work is known by, as the language-nested key-set. A
+        // key-set so two peers resolving different AKAs union on key-merge.
+        for (lang3, name) in &self.names {
+            if let Some(key) = title_member_key(lang3, name) {
+                fields.insert(key, "true".to_string());
+            }
+        }
 
         // The id bag.
         if let Some(id) = self.tmdbid {
@@ -256,6 +299,15 @@ impl Card {
             "poster_url".to_string(),
             tmdb.poster_cdn_url(poster_path),
         );
+        // The alternative posters, as the language-nested key-set (METADATA_KEYS
+        // §6 `posters/{lang3}/{cid}`). Every member is a `url` locator, so listing
+        // them costs no fetch. The primary member wraps the exact `poster_url`
+        // string above: the gateway matches on it and renames the member to the
+        // content cid it seeds into `poster` — that is how "`poster` is a member"
+        // holds for a feeder card.
+        for key in poster_member_keys(tmdb, poster_path, &self.posters) {
+            fields.insert(key, "true".to_string());
+        }
 
         for (key, allowed) in query_filters {
             // `genres` joins `languages` as an echo exclusion, for the same
@@ -284,6 +336,117 @@ impl Card {
     }
 }
 
+/// Most members a record's `posters/*` key-set may carry, the primary included
+/// (METADATA_KEYS.md §6, writer rule 1).
+pub(crate) const MAX_POSTER_MEMBERS: usize = 10;
+
+/// The `posters/{lang3}/{cid}` keys for a card: the primary poster first, then
+/// the best-voted alternatives, capped at [`MAX_POSTER_MEMBERS`]. Empty when
+/// TMDB returned no image list — without one the primary's language is unknown,
+/// and a set is better absent than mislabelled.
+///
+/// Ranked by `vote_average`, then `vote_count`, then `file_path`, the last so
+/// two peers holding the same payload emit the same keys. The language is that
+/// of the text printed on the poster: `zxx` for a textless one (TMDB `null`),
+/// `und` for a code outside [`iso639_1_to_3`] or a primary TMDB did not list.
+///
+/// CROSS-BINARY CONTRACT: same ranking, cap, CDN size and language rule as the
+/// tmdb plugin's `posterMembers` (`metamesh-plugin-tmdb/src/posters.ts`).
+pub(crate) fn poster_member_keys(
+    tmdb: &TmdbClient,
+    primary_path: &str,
+    candidates: &[TmdbImage],
+) -> Vec<String> {
+    if candidates.is_empty() {
+        return Vec::new();
+    }
+    let lang3 = |img: &TmdbImage| -> &'static str {
+        match img.iso_639_1.as_deref().map(str::trim) {
+            None | Some("") => "zxx",
+            Some(code) => iso639_1_to_3(code).unwrap_or("und"),
+        }
+    };
+    let mut keys: Vec<String> = Vec::new();
+    let primary_lang = candidates
+        .iter()
+        .find(|img| img.file_path == primary_path)
+        .map_or("und", lang3);
+    if let Some(cid) = compute_url_cid(&tmdb.poster_cdn_url(primary_path)) {
+        keys.push(format!("posters/{primary_lang}/{cid}"));
+    }
+    let mut ranked: Vec<&TmdbImage> = candidates
+        .iter()
+        .filter(|img| !img.file_path.trim().is_empty() && img.file_path != primary_path)
+        .collect();
+    ranked.sort_by(|a, b| {
+        b.vote_average
+            .total_cmp(&a.vote_average)
+            .then(b.vote_count.cmp(&a.vote_count))
+            .then(a.file_path.cmp(&b.file_path))
+    });
+    for img in ranked {
+        if keys.len() >= MAX_POSTER_MEMBERS {
+            break;
+        }
+        if let Some(cid) = compute_url_cid(&tmdb.poster_cdn_url(&img.file_path)) {
+            let key = format!("posters/{}/{cid}", lang3(img));
+            if !keys.contains(&key) {
+                keys.push(key);
+            }
+        }
+    }
+    keys
+}
+
+/// Punctuation keyword searches fold to a space. Only used to *exclude* these
+/// from [`is_search_hostile_symbol`] — a folded character searches fine.
+///
+/// CROSS-BINARY CONTRACT: identical to the indexer feeder's
+/// `torznab/mod.rs::KEYWORD_PUNCTUATION`, and to meta-watch's
+/// `clean_title` + `sanitize_free_text` sets combined.
+const KEYWORD_PUNCTUATION: [char; 11] = ['.', '_', ':', ';', ',', '-', '/', '(', ')', '=', '!'];
+
+/// An ASCII symbol an indexer's search box can neither carry nor safely lose
+/// (`%` in `3%`): punctuation outside [`KEYWORD_PUNCTUATION`], apostrophe
+/// excepted. See the indexer feeder's `is_search_hostile_symbol` for the
+/// measurements.
+///
+/// CROSS-BINARY CONTRACT: same rule as the indexer feeder's
+/// `torznab/mod.rs::is_search_hostile_symbol` and meta-watch's
+/// `catalog::query::has_search_hostile_symbol`.
+fn is_search_hostile_symbol(c: char) -> bool {
+    c.is_ascii_punctuation() && c != '\'' && !KEYWORD_PUNCTUATION.contains(&c)
+}
+
+/// The AKA to search with in place of `title` — the card's `searchTitle`.
+/// `None` unless `title` carries a [search-hostile symbol](is_search_hostile_symbol)
+/// **and** an AKA is symbol-free and strictly more specific (all of the title's
+/// word tokens, plus at least one): `3 percent` for `3%`, never the bare `3`.
+/// First qualifying AKA wins; `TmdbAltTitles::all` orders English markets first.
+///
+/// CROSS-BINARY CONTRACT: mirrors the indexer feeder's
+/// `torznab/mod.rs::search_aka`, which picks the anchored jobs' `q=` the same
+/// way. Change both.
+pub(crate) fn search_aka<'a>(
+    title: &str,
+    akas: impl IntoIterator<Item = &'a String>,
+) -> Option<&'a str> {
+    if !title.chars().any(is_search_hostile_symbol) {
+        return None;
+    }
+    let want = norm_tokens(title);
+    if want.is_empty() {
+        return None;
+    }
+    akas.into_iter()
+        .filter(|aka| !aka.chars().any(is_search_hostile_symbol))
+        .find(|aka| {
+            let have = norm_tokens(aka);
+            have.len() > want.len() && want.iter().all(|w| have.contains(w))
+        })
+        .map(String::as_str)
+}
+
 /// Split a card `record_id` (`"tmdb:tv:95479"`) back into `(source, source_id)`
 /// — the two halves of the CID preimage. `None` when the id carries no source
 /// prefix.
@@ -305,12 +468,18 @@ mod tests {
             imdb_id: Some("tt22248376".to_string()),
             title: "Frieren: Beyond Journey's End".to_string(),
             guard_titles: Arc::new(vec!["Sousou no Frieren".to_string()]),
+            names: vec![
+                ("jpn", "葬送のフリーレン".to_string()),
+                ("eng", "Frieren: Beyond Journey's End".to_string()),
+                ("jpn", "Sousou no Frieren".to_string()),
+            ],
             overview: Some("The story follows the elf mage Frieren.".to_string()),
             poster_path: Some("/dqZENchTd7lp5zit1Q7Bkjzcxpi.jpg".to_string()),
             genres: vec!["Animation".to_string(), "Sci-Fi & Fantasy".to_string()],
             year: Some(2023),
             seasons: 1,
             season_summaries: Arc::new(Vec::new()),
+            posters: Arc::new(Vec::new()),
         }
     }
 
@@ -379,6 +548,102 @@ mod tests {
         assert!(no_overview.to_record(&tmdb, &BTreeMap::new()).is_none());
     }
 
+    /// `3%` (TMDB tv 68467): the card spells the symbol out for searchers, and
+    /// keeps the real title for display.
+    #[test]
+    fn symbol_title_card_carries_a_search_title() {
+        let tmdb = TmdbClient::new("token".to_string());
+        let mut card = tv_card();
+        card.title = "3%".to_string();
+        card.guard_titles = Arc::new(
+            ["3%", "3 %", "3", "Three Percent", "3 percent", "3 Pourcent"]
+                .iter()
+                .map(|s| s.to_string())
+                .collect(),
+        );
+        let rec = card.to_record(&tmdb, &BTreeMap::new()).expect("record");
+        assert_eq!(rec.fields["title"], "3%");
+        assert_eq!(rec.fields.get("searchTitle").map(String::as_str), Some("3 percent"));
+        // A searchable title gets no searchTitle — Frieren's apostrophe and
+        // colon are not hostile.
+        let rec = tv_card().to_record(&tmdb, &BTreeMap::new()).expect("record");
+        assert!(!rec.fields.contains_key("searchTitle"));
+    }
+
+    #[test]
+    fn record_files_every_name_as_a_language_nested_key_set() {
+        let tmdb = TmdbClient::new("token".to_string());
+        let rec = tv_card().to_record(&tmdb, &BTreeMap::new()).expect("record");
+        for key in [
+            "titles/jpn/葬送のフリーレン",
+            "titles/eng/Frieren: Beyond Journey's End",
+            "titles/jpn/Sousou no Frieren",
+        ] {
+            assert_eq!(rec.fields.get(key).map(String::as_str), Some("true"), "{key}");
+        }
+        // Leaves only: no scalar at the language level.
+        assert!(!rec.fields.keys().any(|k| k.starts_with("titles/") && k.matches('/').count() == 1));
+    }
+
+    /// The METADATA_KEYS.md example, end to end from TMDB's data.
+    #[test]
+    fn title_names_match_the_three_percent_example() {
+        use crate::tmdb_client::{title_names, TmdbAltTitle};
+        let aka = |m: &str, t: &str| TmdbAltTitle { title: t.to_string(), iso_3166_1: m.to_string() };
+        let names = title_names(
+            "3%",
+            Some("3%"),
+            Some("pt"),
+            &[aka("US", "3 Percent"), aka("FR", "3 Pourcent"), aka("BR", "3%"), aka("CA", "3 Por  Cento ")],
+        );
+        assert_eq!(
+            names,
+            vec![
+                ("por", "3%".to_string()),
+                ("eng", "3 Percent".to_string()),
+                ("fra", "3 Pourcent".to_string()),
+                ("und", "3 Por Cento".to_string()),
+            ],
+            "title == original is filed once; BR `3%` repeats por/3%; CA is multilingual → und"
+        );
+        // A translated title is filed under eng beside the original.
+        let names = title_names("Naruto", Some("ナルト"), Some("ja"), &[]);
+        assert_eq!(names, vec![("jpn", "ナルト".to_string()), ("eng", "Naruto".to_string())]);
+        // Unknown original language → und; TMDB's placeholder is not a name.
+        assert_eq!(title_names("(untitled)", Some("X"), Some("xx"), &[]), vec![("und", "X".to_string())]);
+    }
+
+    #[test]
+    fn title_member_key_normalises_like_two_peers_must() {
+        assert_eq!(title_member_key("jpn", "Fate/Zero").as_deref(), Some("titles/jpn/Fate\u{2215}Zero"));
+        assert_eq!(title_member_key("eng", "  3   Percent ").as_deref(), Some("titles/eng/3 Percent"));
+        assert_eq!(title_member_key("eng", "Pokémon").as_deref(), Some("titles/eng/Pokémon"));
+        assert_eq!(title_member_key("eng", "   "), None);
+    }
+
+    #[test]
+    fn search_aka_never_picks_something_weaker() {
+        let akas = |v: &[&str]| v.iter().map(|s| s.to_string()).collect::<Vec<_>>();
+        assert_eq!(search_aka("M*A*S*H", akas(&["MASH"]).iter()), None);
+        assert_eq!(search_aka("3%", akas(&["3", "3 %"]).iter()), None);
+        assert_eq!(search_aka("Law & Order", akas(&["Law and Order"]).iter()), Some("Law and Order"));
+        assert_eq!(search_aka("Naruto", akas(&["Naruto Shippuden"]).iter()), None);
+    }
+
+    #[test]
+    fn tmdb_alt_titles_put_english_markets_first() {
+        let alt: crate::tmdb_client::TmdbAltTitles = serde_json::from_str(
+            r#"{"results":[{"iso_3166_1":"FR","title":"3 Pourcent"},{"iso_3166_1":"US","title":"3 percent"},{"title":"no market"}]}"#,
+        )
+        .expect("decode");
+        assert_eq!(alt.all(), vec!["3 percent", "3 Pourcent", "no market"]);
+        assert!(alt.has_markets());
+        let legacy: crate::tmdb_client::TmdbAltTitles =
+            serde_json::from_str(r#"{"results":[{"title":"3 Pourcent"},{"title":"3 percent"}]}"#).expect("decode");
+        assert!(!legacy.has_markets(), "a pre-market cache entry must refetch");
+        assert!(crate::tmdb_client::TmdbAltTitles::default().has_markets(), "no AKAs is complete");
+    }
+
     #[test]
     fn movie_card_uses_the_movie_content_kind() {
         let mut card = tv_card();
@@ -416,4 +681,100 @@ mod tests {
         assert_eq!(rec.fields["genres/Animation"], "true");
         assert_eq!(rec.fields["genres/Sci-Fi & Fantasy"], "true");
     }
+
+    fn img(path: &str, lang: Option<&str>, avg: f64, count: u32) -> TmdbImage {
+        TmdbImage {
+            file_path: path.to_string(),
+            iso_639_1: lang.map(str::to_string),
+            vote_average: avg,
+            vote_count: count,
+        }
+    }
+
+    /// METADATA_KEYS §6 `posters/{lang3}/{cid}`: the primary is a member, filed
+    /// under its own language, as the locator of the exact `poster_url` the
+    /// gateway seeds — that equality is what lets the gateway rename it.
+    #[test]
+    fn record_files_the_primary_poster_as_a_locator_member() {
+        let tmdb = TmdbClient::new("token".to_string());
+        let mut card = tv_card();
+        let primary = card.poster_path.clone().expect("fixture poster");
+        card.posters = Arc::new(vec![
+            img("/alt-fr.jpg", Some("fr"), 5.5, 10),
+            img(&primary, Some("ja"), 5.0, 3),
+            img("/textless.jpg", None, 6.0, 1),
+        ]);
+        let rec = card.to_record(&tmdb, &BTreeMap::new()).expect("record");
+
+        let primary_cid = compute_url_cid(&rec.fields["poster_url"]).expect("locator");
+        assert_eq!(
+            rec.fields.get(&format!("posters/jpn/{primary_cid}")).map(String::as_str),
+            Some("true")
+        );
+        let fr = compute_url_cid(&tmdb.poster_cdn_url("/alt-fr.jpg")).expect("locator");
+        let textless = compute_url_cid(&tmdb.poster_cdn_url("/textless.jpg")).expect("locator");
+        assert!(rec.fields.contains_key(&format!("posters/fra/{fr}")));
+        assert!(rec.fields.contains_key(&format!("posters/zxx/{textless}")));
+        let members = rec.fields.keys().filter(|k| k.starts_with("posters/")).count();
+        assert_eq!(members, 3, "the primary is listed once, not twice");
+        // Leaves only, and never a 639-2/B code.
+        assert!(!rec.fields.keys().any(|k| k.starts_with("posters/") && k.matches('/').count() != 2));
+        assert!(!rec.fields.keys().any(|k| k.starts_with("posters/fre/")));
+    }
+
+    #[test]
+    fn poster_members_rank_by_votes_and_cap_at_ten() {
+        let tmdb = TmdbClient::new("token".to_string());
+        let candidates: Vec<TmdbImage> = (0..15)
+            .map(|i| img(&format!("/p{i:02}.jpg"), Some("en"), f64::from(i), 1))
+            .collect();
+        let keys = poster_member_keys(&tmdb, "/primary.jpg", &candidates);
+        assert_eq!(keys.len(), MAX_POSTER_MEMBERS);
+        // Primary first; TMDB didn't list it among its images, so `und`.
+        let primary = compute_url_cid(&tmdb.poster_cdn_url("/primary.jpg")).expect("locator");
+        assert_eq!(keys[0], format!("posters/und/{primary}"));
+        // Then the best vote_average: p14, p13, … p06 — p05 is cut.
+        let best = compute_url_cid(&tmdb.poster_cdn_url("/p14.jpg")).expect("locator");
+        assert_eq!(keys[1], format!("posters/eng/{best}"));
+        let cut = compute_url_cid(&tmdb.poster_cdn_url("/p05.jpg")).expect("locator");
+        assert!(!keys.contains(&format!("posters/eng/{cut}")));
+        // vote_count breaks a vote_average tie.
+        let tie = poster_member_keys(
+            &tmdb,
+            "/primary.jpg",
+            &[img("/few.jpg", None, 5.0, 1), img("/many.jpg", None, 5.0, 9)],
+        );
+        let many = compute_url_cid(&tmdb.poster_cdn_url("/many.jpg")).expect("locator");
+        assert_eq!(tie[1], format!("posters/zxx/{many}"));
+        // An unmapped language is `und`, never a 2-letter key.
+        let odd = poster_member_keys(&tmdb, "/primary.jpg", &[img("/x.jpg", Some("xx"), 1.0, 1)]);
+        assert!(odd[1].starts_with("posters/und/"), "{odd:?}");
+    }
+
+    /// No image list (a discovery hit, a pre-append cache entry): no set at all,
+    /// rather than a primary filed under a guessed `und`.
+    #[test]
+    fn card_without_an_image_list_files_no_poster_set() {
+        let tmdb = TmdbClient::new("token".to_string());
+        let rec = tv_card().to_record(&tmdb, &BTreeMap::new()).expect("record");
+        assert!(rec.fields.contains_key("poster_url"), "poster itself is unaffected");
+        assert!(!rec.fields.keys().any(|k| k.starts_with("posters/")));
+    }
+
+    #[test]
+    fn tmdb_images_append_decodes() {
+        let d: crate::tmdb_client::TmdbTvDetails = serde_json::from_str(
+            r#"{"name":"X","images":{"posters":[{"file_path":"/a.jpg","iso_639_1":null,"vote_average":5.3,"vote_count":4},{"file_path":"/b.jpg","iso_639_1":"fr","vote_average":0}]}}"#,
+        )
+        .expect("decode");
+        let posters = d.images.expect("images").posters;
+        assert_eq!(posters.len(), 2);
+        assert_eq!(posters[0].iso_639_1, None);
+        assert_eq!(posters[1].iso_639_1.as_deref(), Some("fr"));
+        assert_eq!(posters[1].vote_count, 0);
+        // A pre-append cache entry still decodes, with no image list.
+        let old: crate::tmdb_client::TmdbTvDetails =
+            serde_json::from_str(r#"{"name":"X"}"#).expect("decode");
+        assert!(old.images.is_none());
+    }
 }

feeder-plugin/card-feeder/src/discovery.rs

@@ -60,7 +60,7 @@ use tracing::debug;
 use crate::card::{Card, CardSource};
 use crate::consts::{DISCOVERY_EXCLUDED_KEYWORDS, DISCOVERY_MAX_PAGES, DISCOVERY_MIN_VOTES};
 use crate::resolve::{dedup_titles, Resolver};
-use crate::tmdb_client::{TmdbHit, TmdbKind};
+use crate::tmdb_client::{title_names, TmdbHit, TmdbKind};
 
 /// TMDB keyword id for "anime". `/discover` is the only catalog endpoint that
 /// takes a keyword filter, which is why `anime:true` switches endpoints rather
@@ -300,6 +300,14 @@ fn card_from_hit(hit: &TmdbHit, kind: TmdbKind, genres: &HashMap<u32, String>) -
         imdb_id: None,
         title: hit.title.clone(),
         guard_titles: Arc::new(dedup_titles(titles)),
+        // A list hit carries no AKAs, so this is title + original only; the
+        // click-time `card_by_id` card adds the AKAs under the same CID.
+        names: title_names(
+            &hit.title,
+            hit.original_title.as_deref(),
+            hit.original_language.as_deref(),
+            &hit.akas,
+        ),
         overview: hit.overview.clone().filter(|s| !s.trim().is_empty()),
         poster_path: hit.poster_path.clone().filter(|s| !s.trim().is_empty()),
         // A list hit carries bare genre ids; the table names them.
@@ -309,6 +317,10 @@ fn card_from_hit(hit: &TmdbHit, kind: TmdbKind, genres: &HashMap<u32, String>) -
         // omitted from the record (`Card::to_record` skips it at `seasons == 0`).
         seasons: 0,
         season_summaries: Arc::new(Vec::new()),
+        // A list response carries no `images` append, so a discovery card files
+        // no `posters/*` set; the click-time `card_by_id` card adds it under the
+        // same CID.
+        posters: Arc::new(Vec::new()),
     }
 }
 
@@ -592,6 +604,7 @@ mod tests {
 
     fn hit() -> TmdbHit {
         TmdbHit {
+            akas: Vec::new(),
             tmdbid: 95479,
             title: "Frieren: Beyond Journey's End".to_string(),
             original_title: Some("葬送のフリーレン".to_string()),
@@ -601,6 +614,7 @@ mod tests {
             poster_path: Some("/poster.jpg".to_string()),
             genre_ids: vec![16],
             alt_titles: Vec::new(),
+            posters: Vec::new(),
         }
     }
 

feeder-plugin/card-feeder/src/resolve.rs

@@ -22,7 +22,8 @@ use crate::consts::{
 };
 use crate::tmdb_budget::{Lease, TmdbBudget};
 use crate::tmdb_client::{
-    principal_top_n, TmdbCall, TmdbClient, TmdbExternalIds, TmdbHit, TmdbKind, TmdbTvDetails,
+    principal_top_n, title_names, TmdbAltTitles, TmdbCall, TmdbClient, TmdbExternalIds, TmdbHit,
+    TmdbKind, TmdbTvDetails,
 };
 
 /// TMDB resolution with a persistent cache in front and a token budget behind.
@@ -149,7 +150,11 @@ impl Resolver {
         if let Ok(Some(json)) = self.cache.get_tmdb_moviedetails(&key) {
             if json != "null" {
                 if let Ok(hit) = serde_json::from_str::<TmdbHit>(&json) {
-                    return Some(hit);
+                    // Self-heal: AKAs cached without their markets can't be
+                    // filed under `titles/{lang3}/{name}` — refetch once.
+                    if hit.alt_titles.is_empty() || !hit.akas.is_empty() {
+                        return Some(hit);
+                    }
                 }
             }
         }
@@ -265,7 +270,18 @@ impl Resolver {
                     .as_deref()
                     .and_then(|d| d.get(0..4))
                     .and_then(|y| y.parse::<u16>().ok());
+                let names = title_names(
+                    &details.name,
+                    details.original_name.as_deref(),
+                    details.original_language.as_deref(),
+                    &details
+                        .alternative_titles
+                        .as_ref()
+                        .map(TmdbAltTitles::entries)
+                        .unwrap_or_default(),
+                );
                 Some(Card {
+                    names,
                     source: CardSource::Tmdb,
                     kind,
                     source_id: format!("tv:{tmdbid}"),
@@ -280,6 +296,13 @@ impl Resolver {
                     year,
                     seasons: Card::clamp_seasons(details.number_of_seasons),
                     season_summaries: Arc::new(details.seasons.clone()),
+                    posters: Arc::new(
+                        details
+                            .images
+                            .as_ref()
+                            .map(|i| i.posters.clone())
+                            .unwrap_or_default(),
+                    ),
                 })
             }
             TmdbKind::Movie => {
@@ -298,7 +321,14 @@ impl Resolver {
                     titles.push(o.clone());
                 }
                 titles.extend(hit.alt_titles.iter().cloned());
+                let names = title_names(
+                    &hit.title,
+                    hit.original_title.as_deref(),
+                    hit.original_language.as_deref(),
+                    &hit.akas,
+                );
                 Some(Card {
+                    names,
                     source: CardSource::Tmdb,
                     kind,
                     source_id: format!("movie:{tmdbid}"),
@@ -313,6 +343,7 @@ impl Resolver {
                     year: hit.year,
                     seasons: 0,
                     season_summaries: Arc::new(Vec::new()),
+                    posters: Arc::new(hit.posters),
                 })
             }
         }

feeder-plugin/card-feeder/src/tmdb_client.rs

@@ -190,7 +190,7 @@ impl TmdbClient {
     /// any failure returns `None` so validation degrades to "accept".
     pub(crate) async fn tv_details(&self, tmdbid: u64) -> TmdbCall<TmdbTvDetails> {
         let url = format!(
-            "{}/tv/{}?append_to_response=alternative_titles",
+            "{}/tv/{}?append_to_response=alternative_titles,images",
             self.api_base.trim_end_matches('/'),
             tmdbid,
         );
@@ -253,7 +253,7 @@ impl TmdbClient {
     /// [`tv_details`].
     pub(crate) async fn movie_details(&self, tmdbid: u64) -> TmdbCall<TmdbMovieDetails> {
         let url = format!(
-            "{}/movie/{}?append_to_response=alternative_titles",
+            "{}/movie/{}?append_to_response=alternative_titles,images",
             self.api_base.trim_end_matches('/'),
             tmdbid
         );
@@ -457,8 +457,8 @@ pub(crate) struct TmdbSearchItem {
     #[serde(default)]
     pub(crate) genre_ids: Vec<u32>,
     /// ISO 639-1 (2-letter) language the title was originally produced in.
-    /// Mapped to `lang3` and used to file `original_title` under
-    /// `titles/{lang3}` (METADATA_KEYS.md §3).
+    /// Mapped to `lang3` and used to file `original_title` as a
+    /// `titles/{lang3}/{name}` member (METADATA_KEYS.md §3).
     #[serde(default)]
     pub(crate) original_language: Option<String>,
 }
@@ -493,6 +493,8 @@ impl TmdbSearchItem {
             genre_ids: self.genre_ids,
             // search-list items carry no AKAs (a details/append fetch does)
             alt_titles: Vec::new(),
+            akas: Vec::new(),
+            posters: Vec::new(),
         }
     }
 }
@@ -500,8 +502,8 @@ impl TmdbSearchItem {
 /// Map an ISO 639-1 (2-letter) code to its ISO 639-3 (`lang3`) equivalent,
 /// matching the store's convention (`eng`, `jpn`, `fra`; the 639-2/T variant
 /// where B/T differ, e.g. `deu` not `ger`). Covers the languages TMDB
-/// commonly returns; unknown codes return `None` so the caller skips the
-/// `titles/{lang3}` write rather than persisting a non-`lang3` key.
+/// commonly returns; unknown codes return `None` so the caller files the name under `und`
+/// rather than persisting a non-`lang3` key.
 pub(crate) fn iso639_1_to_3(code: &str) -> Option<&'static str> {
     Some(match code.to_ascii_lowercase().as_str() {
         "en" => "eng",
@@ -541,6 +543,113 @@ pub(crate) fn iso639_1_to_3(code: &str) -> Option<&'static str> {
     })
 }
 
+/// The language TMDB's `name`/`title` is written in. Every request goes out
+/// without a `language` param, so TMDB answers in its default, en-US.
+pub(crate) const METADATA_LANG3: &str = "eng";
+
+/// A title as a `titles/{lang3}/{name}` member name (METADATA_KEYS.md): trimmed,
+/// internal whitespace collapsed to one space, case/diacritics/`%` verbatim.
+/// The `/` → `∕` mapping is the key's job ([`title_member_key`]), not the name's.
+pub(crate) fn normalize_title_name(name: &str) -> String {
+    name.split_whitespace().collect::<Vec<_>>().join(" ")
+}
+
+/// The `titles/{lang3}/{name}` key-set member for `name`, or `None` when the
+/// name normalises to empty. `/` is the key-set separator, so a `/` inside a
+/// name (`Fate/Zero`) is written as U+2215 `∕` (METADATA_KEYS.md).
+///
+/// CROSS-BINARY CONTRACT: the indexer feeder's `tmdb.rs::title_member_key`
+/// writes the same shape; two peers must produce the same key for one name.
+pub(crate) fn title_member_key(lang3: &str, name: &str) -> Option<String> {
+    let name = normalize_title_name(name);
+    if name.is_empty() {
+        return None;
+    }
+    Some(format!("titles/{lang3}/{}", name.replace('/', "\u{2215}")))
+}
+
+/// The language a TMDB AKA market (`iso_3166_1`) clearly implies, or `und` for
+/// a multilingual or unknown market (METADATA_KEYS.md `titles/{lang3}/{name}`:
+/// map what is unambiguous, never guess). `CA`, `CH`, `BE`, `IN`, `HK`, `SG`,
+/// empty, … are `und` on purpose.
+pub(crate) fn market_lang3(market: &str) -> &'static str {
+    match market.trim().to_ascii_uppercase().as_str() {
+        "US" | "GB" | "AU" | "NZ" | "IE" => "eng",
+        "FR" => "fra",
+        "DE" | "AT" => "deu",
+        "ES" | "MX" | "AR" | "CO" | "CL" | "PE" | "VE" | "UY" => "spa",
+        "BR" | "PT" => "por",
+        "IT" => "ita",
+        "JP" => "jpn",
+        "KR" => "kor",
+        "CN" | "TW" => "zho",
+        "RU" => "rus",
+        "UA" => "ukr",
+        "PL" => "pol",
+        "NL" => "nld",
+        "SE" => "swe",
+        "NO" => "nor",
+        "DK" => "dan",
+        "FI" => "fin",
+        "TR" => "tur",
+        "GR" => "ell",
+        "HU" => "hun",
+        "CZ" => "ces",
+        "SK" => "slk",
+        "RO" => "ron",
+        "BG" => "bul",
+        "RS" => "srp",
+        "HR" => "hrv",
+        "IL" => "heb",
+        "IR" => "fas",
+        "TH" => "tha",
+        "VN" => "vie",
+        "ID" => "ind",
+        _ => "und",
+    }
+}
+
+/// Every name of a work as `(lang3, name)` — what the card files under
+/// `titles/{lang3}/{name}`:
+///
+/// - `original` under its `original_language` (`und` when unmapped/absent);
+/// - `title` under [`METADATA_LANG3`], **unless it equals the original** — TMDB
+///   falls back to the original name when it has no en-US translation, so `3%`
+///   is filed once, as `titles/por/3%`, not also as an English name;
+/// - every AKA under the language its market implies ([`market_lang3`]).
+///
+/// Names are normalised ([`normalize_title_name`]); blanks, TMDB's
+/// `(untitled)` placeholder and exact `(lang3, name)` repeats are dropped.
+pub(crate) fn title_names(
+    title: &str,
+    original: Option<&str>,
+    original_language: Option<&str>,
+    akas: &[TmdbAltTitle],
+) -> Vec<(&'static str, String)> {
+    let mut out: Vec<(&'static str, String)> = Vec::new();
+    let mut push = |lang3: &'static str, name: &str| {
+        let name = normalize_title_name(name);
+        if !name.is_empty() && !out.iter().any(|(l, n)| *l == lang3 && *n == name) {
+            out.push((lang3, name));
+        }
+    };
+    let original = original.map(normalize_title_name).filter(|o| !o.is_empty());
+    if let Some(o) = &original {
+        let lang3 = original_language
+            .and_then(|l| iso639_1_to_3(l.trim()))
+            .unwrap_or("und");
+        push(lang3, o);
+    }
+    let title = normalize_title_name(title);
+    if title != "(untitled)" && original.as_deref() != Some(title.as_str()) {
+        push(METADATA_LANG3, &title);
+    }
+    for aka in akas {
+        push(market_lang3(&aka.iso_3166_1), &aka.title);
+    }
+    out
+}
+
 // `Clone` is required so the single-flight `Shared` future (whose `Output`
 // must be `Clone`) can hand the same hit to every coalesced caller; serde so
 // hits persist in the redb TMDB-search cache.
@@ -566,17 +675,53 @@ pub(crate) struct TmdbHit {
     /// Empty for hits built from a search list (which carries no AKAs).
     #[serde(default)]
     pub(crate) alt_titles: Vec<String>,
+    /// The same AKAs **with their market**, which the card needs to file each
+    /// under `titles/{lang3}/{name}` ([`title_names`]). Empty on a movie-details
+    /// cache entry written before it existed — `Resolver::movie_hit` refetches
+    /// those once.
+    #[serde(default)]
+    pub(crate) akas: Vec<TmdbAltTitle>,
+    /// Poster candidates from the details `images` append, filed by the card as
+    /// `posters/{lang3}/{cid}` (METADATA_KEYS.md §6). Empty for list hits and for
+    /// a movie-details cache entry written before it existed — deliberately NOT a
+    /// self-heal trigger in `Resolver::movie_hit`: no backfill, no TMDB budget.
+    #[serde(default)]
+    pub(crate) posters: Vec<TmdbImage>,
 }
 
 impl TmdbHit {
     /// TMDB `original_language` (ISO 639-1) mapped to the store's `lang3`
     /// (ISO 639-3), or `None` when the language is outside the common set or
-    /// absent. Used to file `original_title` under `titles/{lang3}` (§3).
+    /// absent. Used to file `original_title` under `titles/{lang3}/{name}` (§3).
     pub(crate) fn original_lang3(&self) -> Option<&'static str> {
         iso639_1_to_3(self.original_language.as_deref()?.trim())
     }
 }
 
+/// One poster from a details payload's `images` append — a candidate for the
+/// card's `posters/{lang3}/{cid}` key-set ([`crate::card::poster_member_keys`]).
+#[derive(Debug, Clone, Default, PartialEq, serde::Serialize, serde::Deserialize)]
+pub(crate) struct TmdbImage {
+    #[serde(default)]
+    pub(crate) file_path: String,
+    /// Language of the text printed on the poster (ISO 639-1); `None` = textless.
+    #[serde(default)]
+    pub(crate) iso_639_1: Option<String>,
+    #[serde(default)]
+    pub(crate) vote_average: f64,
+    #[serde(default)]
+    pub(crate) vote_count: u32,
+}
+
+/// The `images` object `?append_to_response=images` adds to `GET /3/{tv,movie}/{id}`.
+/// Only `posters` is decoded. No `language` param is sent, so TMDB returns the
+/// posters of every language.
+#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)]
+pub(crate) struct TmdbImages {
+    #[serde(default)]
+    pub(crate) posters: Vec<TmdbImage>,
+}
+
 /// Authoritative TV structure from `GET /3/tv/{id}`, used to bounds-check
 /// a title-parsed season/episode. Only the two structural fields are
 /// decoded; everything else in the (large) details payload is ignored.
@@ -617,6 +762,10 @@ pub(crate) struct TmdbTvDetails {
     /// fetch self-heals with one refetch (same pattern as the display fields).
     #[serde(default)]
     pub(crate) alternative_titles: Option<TmdbAltTitles>,
+    /// Poster candidates from `?append_to_response=images`. `None` on a cache
+    /// entry written before the append — NOT a self-heal trigger (no backfill).
+    #[serde(default)]
+    pub(crate) images: Option<TmdbImages>,
 }
 
 impl TmdbTvDetails {
@@ -630,8 +779,15 @@ impl TmdbTvDetails {
     /// True when the AKA field was populated by the current (append-enabled)
     /// decoder. A `false` on a cache hit means a pre-AKA entry → the cached
     /// fetch refetches once to upgrade it (mirrors [`has_display`]).
+    ///
+    /// Also `false` for an AKA list decoded before `iso_3166_1` was: the
+    /// search-term pick (`card::search_aka`) prefers English-market AKAs, and a
+    /// market-less list would silently keep TMDB's order (`3 Pourcent` over
+    /// `3 percent`) forever. One refetch per such entry upgrades it.
     pub(crate) fn has_akas(&self) -> bool {
-        self.alternative_titles.is_some()
+        self.alternative_titles
+            .as_ref()
+            .is_some_and(TmdbAltTitles::has_markets)
     }
 
     /// The show's canonical + original + AKA titles, for the id-based-job
@@ -672,6 +828,16 @@ impl TmdbTvDetails {
                 .as_ref()
                 .map(TmdbAltTitles::all)
                 .unwrap_or_default(),
+            akas: self
+                .alternative_titles
+                .as_ref()
+                .map(TmdbAltTitles::entries)
+                .unwrap_or_default(),
+            posters: self
+                .images
+                .as_ref()
+                .map(|i| i.posters.clone())
+                .unwrap_or_default(),
         })
     }
 }
@@ -699,17 +865,42 @@ pub(crate) struct TmdbAltTitles {
 pub(crate) struct TmdbAltTitle {
     #[serde(default)]
     pub(crate) title: String,
+    /// The market the AKA is used in (`US`, `FR`, …). Empty on a cache entry
+    /// written before it was decoded — ordering then falls back to TMDB's.
+    #[serde(default)]
+    pub(crate) iso_3166_1: String,
 }
 
 impl TmdbAltTitles {
+    /// Was this list decoded with markets? TMDB stamps `iso_3166_1` on every
+    /// AKA, so a non-empty list where none carries one is a pre-market cache
+    /// entry. An empty list is complete ("fetched, no AKAs").
+    pub(crate) fn has_markets(&self) -> bool {
+        let mut titles = self.results.iter().chain(self.titles.iter()).peekable();
+        titles.peek().is_none() || titles.any(|t| !t.iso_3166_1.is_empty())
+    }
+
     /// The non-empty AKA titles across both the TV (`results`) and movie
-    /// (`titles`) shapes.
+    /// (`titles`) shapes, **English-market (`US`, `GB`) titles first**, TMDB's
+    /// order otherwise. The relevance guard only asks "any of these?", so the
+    /// order is free for it; it matters to the search-term AKA pick
+    /// (`card::search_aka`), where `3 percent` (US) must beat `3 Pourcent` (FR).
     pub(crate) fn all(&self) -> Vec<String> {
-        self.results
-            .iter()
-            .chain(self.titles.iter())
-            .map(|t| t.title.trim().to_string())
-            .filter(|t| !t.is_empty())
+        self.entries().into_iter().map(|t| t.title).collect()
+    }
+
+    /// [`Self::all`] with each AKA's market kept — same order, same filtering
+    /// (titles trimmed, blanks dropped).
+    pub(crate) fn entries(&self) -> Vec<TmdbAltTitle> {
+        let mut titles: Vec<&TmdbAltTitle> = self.results.iter().chain(self.titles.iter()).collect();
+        titles.sort_by_key(|t| !matches!(t.iso_3166_1.as_str(), "US" | "GB"));
+        titles
+            .into_iter()
+            .map(|t| TmdbAltTitle {
+                title: t.title.trim().to_string(),
+                iso_3166_1: t.iso_3166_1.trim().to_string(),
+            })
+            .filter(|t| !t.title.is_empty())
             .collect()
     }
 }
@@ -748,6 +939,9 @@ pub(crate) struct TmdbMovieDetails {
     pub(crate) poster_path: Option<String>,
     #[serde(default)]
     pub(crate) alternative_titles: Option<TmdbAltTitles>,
+    /// See [`TmdbTvDetails::images`].
+    #[serde(default)]
+    pub(crate) images: Option<TmdbImages>,
 }
 
 impl TmdbMovieDetails {
@@ -762,6 +956,12 @@ impl TmdbMovieDetails {
             .as_ref()
             .map(TmdbAltTitles::all)
             .unwrap_or_default();
+        let akas = self
+            .alternative_titles
+            .as_ref()
+            .map(TmdbAltTitles::entries)
+            .unwrap_or_default();
+        let posters = self.images.map(|i| i.posters).unwrap_or_default();
         TmdbHit {
             tmdbid: self.id,
             title: self.title.unwrap_or_else(|| "(untitled)".to_string()),
@@ -772,6 +972,8 @@ impl TmdbMovieDetails {
             poster_path: self.poster_path,
             genre_ids: Vec::new(),
             alt_titles,
+            akas,
+            posters,
         }
     }
 }

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

@@ -64,7 +64,7 @@ fn tv_details_json() -> serde_json::Value {
         "poster_path": "/poster.jpg",
         "number_of_seasons": 1,
         "seasons": [{"season_number": 1, "episode_count": 28}],
-        "alternative_titles": { "results": [{"title": "Sousou no Frieren"}] }
+        "alternative_titles": { "results": [{"iso_3166_1": "JP", "title": "Sousou no Frieren"}] }
     })
 }
 
@@ -148,6 +148,15 @@ async fn query_returns_a_card_with_both_axes_and_the_id_bag() {
     assert_eq!(f["fileType"], "card");
     assert_eq!(f["contentKind"], "series");
     assert_eq!(f["title"], "Frieren: Beyond Journey's End");
+    // Every name as the language-nested key-set: original (ja), the en-US
+    // title, and the JP-market AKA.
+    for key in [
+        "titles/jpn/\u{845}\u{9001}\u{306e}\u{30d5}\u{30ea}\u{30fc}\u{30ec}\u{30f3}",
+        "titles/eng/Frieren: Beyond Journey's End",
+        "titles/jpn/Sousou no Frieren",
+    ] {
+        assert_eq!(f.get(key).map(String::as_str), Some("true"), "{key}");
+    }
     // The id bag — phase 2 selects from this rather than assuming `tmdbid`.
     assert_eq!(f["tmdbid"], "95479");
     assert_eq!(f["tvdbid"], "367189");