youtube feeder improvement

feeder-plugin/youtube-feeder/Cargo.toml

@@ -18,9 +18,15 @@ name = "youtube-feeder"
 path = "src/main.rs"
 
 [dependencies]
-# TEMPORARY: path dep while the SDK's new rev is unpushed. Flip to:
-#   meta-feeder-sdk = { git = "https://github.com/worph/meta-feeder-sdk", rev = "<sha>" }
-meta-feeder-sdk = { path = "../../../meta-feeder-sdk" }
+# Pinned by TAG, not rev or path. The tag is the SDK's release identity, and it
+# is the one form every feeder in the fleet can be lined up on.
+#
+# ⚠ Never put this back to `path = "../../../meta-feeder-sdk"`. That resolves
+# OUTSIDE this repo, and this repo is the Docker build context — the image build
+# died at manifest resolution with a bare "No such file or directory", and the
+# crate became unconsumable as a git dependency by anyone, since a consumer's
+# checkout has no such sibling directory either.
+meta-feeder-sdk = { git = "https://github.com/worph/meta-feeder-sdk", tag = "v1.1.0" }
 
 tokio = { version = "1.40", features = ["full"] }
 axum = { version = "0.7", features = ["json"] }
@@ -37,7 +43,10 @@ redb = "2"
 base64 = "0.22"
 # The MusicBrainz client, from the crate that owns it. Only the *music*
 # path uses it; the video path never touches MusicBrainz.
-musicbrainz-feeder = { path = "../../../meta-feeder-musicbrainz/feeder-plugin/musicbrainz-feeder" }
+# v1.1.1, not v1.1.0: v1.1.0 declared the escaping path dep described above, so
+# it cannot be consumed over git at all ("no matching package named
+# `meta-feeder-sdk`"). v1.1.1 is that one-line fix and nothing else.
+musicbrainz-feeder = { git = "https://github.com/worph/meta-feeder-musicbrainz", tag = "v1.1.1" }
 
 [dev-dependencies]
 wiremock = "0.6"

feeder-plugin/youtube-feeder/src/consts.rs

@@ -55,3 +55,28 @@ pub const MAX_REFS_PER_RECORDING: usize = 3;
 
 // MusicBrainz's published rate ceiling, from the crate that owns that client.
 pub use musicbrainz_feeder::consts::{MB_BURST, MB_RATE_PER_SEC, MB_TIMEOUT_SECS};
+
+/// Ceiling on raw free-text video hits emitted for one query.
+///
+/// The music paths thin their results with a *verification* — channel identity
+/// plus duration — and refuse everything that fails it. The raw tier has no
+/// such filter by construction, so the only bound available is a count. 20 is a
+/// search page's worth: enough for the section to be useful, few enough that
+/// one query does not pull a hundred thumbnails into the gateway's blockstore
+/// (`poster_url` is fetched **and seeded** by the core, unlike a `cover`
+/// locator, which stays lazy).
+pub const YT_VIDEO_SEARCH_MAX: usize = 20;
+
+/// Left half of a raw video tier record id (`yt~<videoId>`).
+///
+/// The music tier keys a record on `<mbRecordingId>~<videoId>` because the same
+/// video can legitimately be the reference for more than one recording. The raw
+/// tier has no such second axis — a video is itself and nothing else — so the
+/// left half is a constant, which makes the id a pure function of the video id
+/// and lets the gateway dedupe the same hit across different search terms.
+///
+/// ⚠ It must stay parseable by [`crate::youtube::split_track_record_id`]: the
+/// `~` split is what keeps `compute_outcomes` a zero-I/O pure function for both
+/// tiers. A left half that could collide with a MusicBrainz id would be a bug;
+/// `yt` cannot, since MBIDs are UUIDs.
+pub const VIDEO_RECORD_NS: &str = "yt";

feeder-plugin/youtube-feeder/src/youtube.rs

@@ -1,7 +1,23 @@
 //! The `youtube` upstream — the **delegated-playback** tier.
 //!
 //! Resolves a MusicBrainz release-group to per-track YouTube references so a
-//! recording with no owned bytes anywhere is still *playable*. It is the third
+//! recording with no owned bytes anywhere is still *playable*.
+//!
+//! # Two tiers, and the line between them
+//!
+//! Everything in the "resolver" section below describes the **verified** paths
+//! (`track`, `musicVideo`): id-anchored, channel- and duration-checked, and
+//! built to refuse. A third path, [`YouTubePlugin::video_search`], answers a
+//! plain `fileType:video` query with **unverified** free-text hits, because a
+//! video client asking a video question deserves an answer even when nothing
+//! can vouch for it.
+//!
+//! The two do not blur, and the reason is structural rather than editorial: raw
+//! records never carry `anchored`, and a consumer's grid drops non-anchored
+//! gateway hits unconditionally, so they can only ever land in a "raw sources"
+//! section. Read the refusal doctrine below as scoped to the verified paths —
+//! it is what keeps *those* trustworthy, and the raw tier is walled off from
+//! them by the dispatch order in [`FeederPlugin::handle_query`], not by taste. It is the third
 //! plugin in this binary, beside `musicbrainz` (identity) and
 //! `internetarchive` (free bytes), and it lives here rather than in a repo of
 //! its own because the MusicBrainz client, the rate budget, the cache and the
@@ -64,8 +80,8 @@ use tracing::{debug, warn};
 use meta_feeder_sdk::cache::MidhashCache;
 use crate::consts::{
     MAX_REFS_PER_RECORDING, MB_BURST, MB_RATE_PER_SEC, MB_TIMEOUT_SECS, PIPED_INSTANCES,
-    PIPED_TIMEOUT_SECS, USER_AGENT,
-    YT_CANDIDATES_PER_TRACK, YT_DURATION_TOLERANCE_SECS, YT_THUMB_BASE,
+    PIPED_TIMEOUT_SECS, USER_AGENT, VIDEO_RECORD_NS,
+    YT_CANDIDATES_PER_TRACK, YT_DURATION_TOLERANCE_SECS, YT_THUMB_BASE, YT_VIDEO_SEARCH_MAX,
 };
 use musicbrainz_feeder::client::{primary_artist, MusicBrainzClient};
 use meta_feeder_sdk::budget::RateBudget;
@@ -121,13 +137,27 @@ pub struct ChannelKeys {
     pub ranked: Vec<(String, String)>,
 }
 
-/// One Piped `music_songs` result, trimmed to what the filters read.
+/// `ytMatchKey` for the raw tier — the value that means "nothing verified this".
+///
+/// A *value*, not an absent field. Every record this feeder emits carries a
+/// match key, so a reader never has to distinguish "unverified" from "written
+/// by a build that predates match keys" — a distinction that is impossible to
+/// recover once records are replicated across the mesh.
+pub const RAW_MATCH_KEY: &str = "unverified:free-text";
+
+/// One Piped search result, trimmed to what the filters read.
 #[derive(Clone, Debug)]
 struct Candidate {
     video_id: String,
     title: String,
     uploader_channel: String,
     duration_secs: i64,
+    /// The channel's *display name*. The music paths never read it — they
+    /// compare channel **ids**, which is the whole point of their filter — but
+    /// the raw tier has no id to compare against and puts the uploader in the
+    /// record's description instead, so a viewer can see who they are about to
+    /// watch. Empty when Piped omits it.
+    uploader: String,
 }
 
 /// The verdict for one (track, candidate) pair.
@@ -724,6 +754,172 @@ impl YouTubePlugin {
             fields,
         })
     }
+
+    /// The raw tier's resolver: a free-text YouTube search, unverified.
+    ///
+    /// Deliberately thin. There is no channel to compare against and no
+    /// duration to check, so there is nothing to *decide* — the ranking is
+    /// YouTube's own, the only judgement applied is a ceiling. Anything more
+    /// (a fuzzy title score, a "looks official" heuristic) would manufacture
+    /// confidence this tier has no basis for and cannot label.
+    async fn video_search(
+        &self,
+        query: &GatewayQuery,
+        max_results: usize,
+    ) -> Result<Vec<DiscoveryRecord>, GatewayError> {
+        // No term, no answer. A `tmdbid` alone cannot drive this path: turning
+        // an id into a searchable title needs TMDB, which this feeder has no
+        // client for — and inventing a term from the id would search for the
+        // literal number. The anchored form is a different plugin's job (see
+        // the module TODO); this one refuses rather than guessing.
+        let term = query.free_text.trim().to_string();
+        if term.is_empty() {
+            debug!(
+                target: "meta-music",
+                "youtube: video query carries no free text; nothing to search on"
+            );
+            return Ok(Vec::new());
+        }
+        let cap = max_results.min(YT_VIDEO_SEARCH_MAX);
+        if cap == 0 {
+            return Ok(Vec::new());
+        }
+
+        let mut out = Vec::new();
+        let mut seen: std::collections::HashSet<String> = std::collections::HashSet::new();
+        for c in self.search_videos(&term, cap).await {
+            if out.len() >= cap {
+                break;
+            }
+            // Piped can repeat a video across result shapes; a duplicate record
+            // id would collapse downstream anyway, but emitting it wastes a row
+            // of the ceiling.
+            if !seen.insert(c.video_id.clone()) {
+                continue;
+            }
+            if let Some(r) = self.video_record(&c, &query.filters) {
+                out.push(r);
+            }
+        }
+        debug!(
+            target: "meta-music",
+            term = %term,
+            hits = out.len(),
+            "youtube: raw video search"
+        );
+        Ok(out)
+    }
+
+    /// `filter=videos` — the general corpus, not the Music one.
+    ///
+    /// The mirror of [`Self::search_track`]'s `filter=music_songs`: that filter
+    /// exists to bias towards official audio, which is the right corpus for a
+    /// recording and the wrong one for "a video of this thing".
+    ///
+    /// Cached under its **own** namespace. Sharing `search_key("yt", …)` with
+    /// the track path would let a `music_songs` response answer a `videos`
+    /// query whenever the two terms happened to coincide — a silent corpus swap
+    /// that would look like a ranking quirk.
+    async fn search_videos(&self, term: &str, take: usize) -> Vec<Candidate> {
+        let key = meta_feeder_sdk::cache::search_key("yt-video", term);
+        if let Some(c) = self.cache.as_ref() {
+            if let Ok(Some(hit)) = c.get_yt_search(&key) {
+                if let Ok(v) = serde_json::from_str::<Value>(&hit) {
+                    return parse_candidates_n(&v, take);
+                }
+            }
+        }
+        let Some(v) = self
+            .piped_get(&format!(
+                "search?q={}&filter=videos",
+                urlencoding_encode(term)
+            ))
+            .await
+        else {
+            // Same rule as the track path: a failed search is NOT an empty one,
+            // and nothing is cached, so the next attempt retries. Caching a
+            // Piped outage as "no such video" is the partial-answer trap that
+            // pinned an artist to the wrong channel key for good.
+            return Vec::new();
+        };
+        if let (Some(c), Ok(json)) = (self.cache.as_ref(), serde_json::to_string(&v)) {
+            let _ = c.put_yt_search(&key, &json);
+        }
+        parse_candidates_n(&v, take)
+    }
+
+    /// One raw, unverified video record.
+    ///
+    /// Returns `None` if the `yt-video` cid cannot be minted — a record with no
+    /// `cids/*` member is unplayable and would only add a dead row.
+    fn video_record(
+        &self,
+        c: &Candidate,
+        query_filters: &BTreeMap<String, Vec<String>>,
+    ) -> Option<DiscoveryRecord> {
+        let yt_cid = meta_feeder_sdk::hash::compute_yt_video_cid("video", &c.video_id)?;
+
+        let mut fields: BTreeMap<String, String> = BTreeMap::new();
+        fields.insert("fileType".to_string(), "video".to_string());
+        // **No `contentKind`.** The music paths know what they resolved; this
+        // one genuinely does not, and a guess would be worse than a gap:
+        // `contentKind:pack` in particular is dropped outright by meta-watch's
+        // raw projection, and any wrong value narrows the record out of filters
+        // it should have matched. An absent axis is accepted everywhere.
+        fields.insert("title".to_string(), c.title.clone());
+        fields.insert(format!("cids/{yt_cid}"), "true".to_string());
+        fields.insert("youtubeid".to_string(), c.video_id.clone());
+        fields.insert("source/gateway:youtube".to_string(), "true".to_string());
+        if c.duration_secs > 0 {
+            fields.insert("fileinfo/duration".to_string(), c.duration_secs.to_string());
+        }
+        // The confidence signal, stated rather than implied. Every other record
+        // this feeder emits carries the channel key that verified it; this one
+        // carries the fact that nothing did. A consumer that later wants to
+        // rank or hide the raw tier has the field to do it with, and no reader
+        // has to infer "unverified" from the *absence* of a key.
+        fields.insert("ytMatchKey".to_string(), RAW_MATCH_KEY.to_string());
+        // Artwork + synopsis. Neither is needed for the raw bucket — meta-watch
+        // renders those rows without a poster — but a record outlives the query
+        // that produced it, and a consumer applying a poster+description gate
+        // (meta-listen's, or meta-watch's enriched grid should this hit ever be
+        // anchored by something else) would otherwise drop it silently. That
+        // silence is the expensive failure mode here, so pay two fields now.
+        //
+        // `poster_url`, not `cover`: `cover` is the music tier's sleeve
+        // vocabulary and this is a video record. The core fetches and seeds a
+        // `*_url` field (`remote_feeder::SEEDABLE_FIELDS`), which is why the
+        // per-query ceiling exists.
+        fields.insert(
+            "poster_url".to_string(),
+            format!("{YT_THUMB_BASE}/{}/hqdefault.jpg", c.video_id),
+        );
+        let uploader = c.uploader.trim();
+        fields.insert(
+            "description/eng".to_string(),
+            if uploader.is_empty() {
+                "Delegated playback — YouTube. Unverified search result.".to_string()
+            } else {
+                format!("Delegated playback — YouTube, {uploader}. Unverified search result.")
+            },
+        );
+        if !uploader.is_empty() {
+            fields.insert("uploader".to_string(), uploader.to_string());
+        }
+
+        // Echoes the query's own filters onto the record — which is how a
+        // `tmdbid:`-carrying query gets that id stamped on the hit, tying it to
+        // the work the consumer asked about. Note what this is NOT: echoing an
+        // id is not *verifying* one, and it deliberately does not set
+        // `anchored`. A record that claimed the anchor it merely inherited is
+        // the leak that once made a whole indexer's results look trusted.
+        echo_filters(&mut fields, query_filters);
+        Some(DiscoveryRecord {
+            upstream_id: "youtube".to_string(),
+            record_id: video_record_id(&c.video_id),
+            fields,
+        })
+    }
 }
 
 /// Rank order for [`ChannelKey`] as a tie-break: authoritative first.
@@ -811,12 +1007,24 @@ fn video_id_from_url(url: &str) -> Option<String> {
 }
 
 fn parse_candidates(v: &Value) -> Vec<Candidate> {
+    parse_candidates_n(v, YT_CANDIDATES_PER_TRACK)
+}
+
+/// [`parse_candidates`] with an explicit ceiling.
+///
+/// The music paths want [`YT_CANDIDATES_PER_TRACK`] — a *verification* budget,
+/// sized by how deep into the ranking a correct match plausibly sits. The raw
+/// tier's ceiling is a different quantity entirely (how many rows to show), so
+/// it passes its own rather than inheriting a constant that means something
+/// else. Same parse either way: `filter=videos` and `filter=music_*` return the
+/// same item shape.
+fn parse_candidates_n(v: &Value, take: usize) -> Vec<Candidate> {
     let Some(items) = v.get("items").and_then(|i| i.as_array()) else {
         return Vec::new();
     };
     items
         .iter()
-        .take(YT_CANDIDATES_PER_TRACK)
+        .take(take)
         .filter_map(|it| {
             Some(Candidate {
                 video_id: video_id_from_url(it.get("url")?.as_str()?)?,
@@ -827,6 +1035,11 @@ fn parse_candidates(v: &Value) -> Vec<Candidate> {
                     .and_then(channel_id_from_url)
                     .unwrap_or_default(),
                 duration_secs: it.get("duration").and_then(|d| d.as_i64()).unwrap_or(-1),
+                uploader: it
+                    .get("uploaderName")
+                    .and_then(|u| u.as_str())
+                    .unwrap_or_default()
+                    .to_string(),
             })
         })
         .collect()
@@ -873,6 +1086,48 @@ pub fn split_track_record_id(record_id: &str) -> Option<(&str, &str)> {
     (!rec.is_empty() && !vid.is_empty()).then_some((rec, vid))
 }
 
+/// `yt~<videoId>` — the record id the **raw** tier mints.
+///
+/// Shares [`split_track_record_id`]'s shape on purpose, so `compute_outcomes`
+/// stays one zero-I/O function serving both tiers: it only ever needs the right
+/// half. The left half is [`VIDEO_RECORD_NS`] rather than a second MBID because
+/// a raw hit has no recording behind it — and a constant makes the id a pure
+/// function of the video, so the same video found via two different search
+/// terms dedupes to one record instead of two.
+pub fn video_record_id(video_id: &str) -> String {
+    format!("{VIDEO_RECORD_NS}~{video_id}")
+}
+
+/// Did the query **explicitly** ask for this file type?
+///
+/// Explicit, unlike [`wants_content_kind`]: an absent axis means "no
+/// preference", and treating that as a video request would let the raw tier
+/// answer questions nobody pointed at it. The music paths can afford the
+/// permissive reading because an id anchor follows; this one cannot.
+fn requests_file_type(query: &GatewayQuery, file_type: &str) -> bool {
+    query
+        .filters
+        .get("fileType")
+        .is_some_and(|v| v.iter().any(|x| x.trim().eq_ignore_ascii_case(file_type)))
+}
+
+/// Does the query carry a MusicBrainz anchor of any kind?
+///
+/// The guard that keeps the raw tier out of the verified tiers' way. If a
+/// caller handed us an id, there is a path that can *verify* an answer, and
+/// serving a free-text guess instead would be a silent downgrade — the kind
+/// that shows up as "it used to be right".
+fn has_musicbrainz_anchor(query: &GatewayQuery) -> bool {
+    ["mbReleaseGroupId", "mbArtistId", "mbRecordingId"]
+        .iter()
+        .any(|k| {
+            query
+                .filters
+                .get(*k)
+                .is_some_and(|v| v.iter().any(|s| !s.trim().is_empty()))
+        })
+}
+
 #[async_trait]
 impl FeederPlugin for YouTubePlugin {
     fn upstream_id(&self) -> &'static str {
@@ -961,6 +1216,39 @@ impl FeederPlugin for YouTubePlugin {
         if wants_content_kind(query, "musicVideo") && !wants_content_kind(query, "track") {
             return self.music_videos(query, max_results).await;
         }
+        // **The raw video tier.** Everything above this line is id-anchored and
+        // refuses far more than it emits; this branch is the deliberate
+        // exception, and the reason it is safe is *structural*, not a matter of
+        // resolver discipline.
+        //
+        // A consumer's untyped free-text search never reaches a gateway at all
+        // (meta-search's load gate), so the query that lands here has already
+        // said `fileType:video` — a video client asking a video question. What
+        // it cannot say is *which* work it means, and there is no anchor to
+        // verify a match against. So this path emits **unverified** hits, which
+        // every other path in this module exists to prevent.
+        //
+        // What makes that honest rather than a regression: these records carry
+        // no `anchored` flag, and meta-watch's `release_from_record` drops every
+        // non-anchored gateway hit from the enriched grid *unconditionally* —
+        // poster and synopsis do not buy a way in. They can therefore only
+        // surface in the "raw sources" section, which is precisely the UI
+        // affordance for "found it, can't vouch for it". The tier is
+        // second-class by construction, not by convention.
+        //
+        // Ordering matters here and is load-bearing:
+        //
+        // * `musicVideo` is matched **above**, so meta-listen's artist row
+        //   (`mbArtistId:… fileType:video contentKind:musicVideo`) still takes
+        //   the verified path and never falls through to this one.
+        // * The `mbReleaseGroupId` track path below is `fileType:audio`, so it
+        //   cannot reach this branch either.
+        // * `has_musicbrainz_anchor` is belt-and-braces for both: an id-anchored
+        //   query has something better than a free-text guess available, and
+        //   answering it from here would quietly downgrade a verified tier.
+        if requests_file_type(query, "video") && !has_musicbrainz_anchor(query) {
+            return self.video_search(query, max_results).await;
+        }
         if !wants_content_kind(query, "track") {
             return Ok(Vec::new());
         }
@@ -1096,8 +1384,12 @@ impl FeederPlugin for YouTubePlugin {
     /// record and stops. There is no blockstore seed (gated on `Sha2_256`), no
     /// WebDAV blob, and no `file_extension` (which MUST be `None` when `bytes`
     /// is).
+    /// Serves **both** tiers: the left half of the id is whatever minted it (a
+    /// MusicBrainz recording, or [`VIDEO_RECORD_NS`] for a raw hit) and is
+    /// deliberately unread — the cid is a pure function of the video id alone,
+    /// so there is nothing here to branch on.
     async fn compute_outcomes(&self, record_id: &str) -> Result<Vec<HashOutcome>, GatewayError> {
-        let (_recording_mbid, video_id) = split_track_record_id(record_id).ok_or_else(|| {
+        let (_minted_by, video_id) = split_track_record_id(record_id).ok_or_else(|| {
             GatewayError::Permanent(format!("malformed youtube record_id '{record_id}'"))
         })?;
         let hash = meta_feeder_sdk::hash::compute_yt_video_cid("video", video_id).ok_or_else(
@@ -1235,6 +1527,7 @@ mod tests {
             title: title.to_string(),
             uploader_channel: channel.to_string(),
             duration_secs: secs,
+            uploader: String::new(),
         }
     }
 
@@ -1393,6 +1686,267 @@ mod tests {
         assert!(p.handle_query(&q, 20).await.expect("ok").is_empty());
     }
 
+    // --- the raw video tier -------------------------------------------------
+
+    fn vq(free_text: &str, filters: &[(&str, &[&str])]) -> GatewayQuery {
+        GatewayQuery {
+            raw_text: free_text.to_string(),
+            free_text: free_text.to_string(),
+            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 file_type_request_is_explicit_not_permissive() {
+        // The asymmetry with `wants_content_kind` is the point: an absent axis
+        // means "no preference", and the raw tier must not read that as an
+        // invitation.
+        assert!(requests_file_type(
+            &vq("x", &[("fileType", &["video"])]),
+            "video"
+        ));
+        assert!(requests_file_type(
+            &vq("x", &[("fileType", &["audio", "video"])]),
+            "video"
+        ));
+        assert!(!requests_file_type(&vq("x", &[]), "video"));
+        assert!(!requests_file_type(
+            &vq("x", &[("fileType", &["audio"])]),
+            "video"
+        ));
+    }
+
+    #[test]
+    fn musicbrainz_anchor_is_detected_on_every_id_axis() {
+        for k in ["mbReleaseGroupId", "mbArtistId", "mbRecordingId"] {
+            assert!(
+                has_musicbrainz_anchor(&vq("x", &[(k, &["8bfac288-ccc5-445d-9a7e-1a1"])])),
+                "{k} should count as an anchor"
+            );
+        }
+        assert!(!has_musicbrainz_anchor(&vq("x", &[("tmdbid", &["95479"])])));
+        // Present but empty is not an anchor — it cannot verify anything.
+        assert!(!has_musicbrainz_anchor(&vq("x", &[("mbArtistId", &[""])])));
+    }
+
+    #[test]
+    fn video_record_id_round_trips_through_the_shared_splitter() {
+        // `compute_outcomes` serves both tiers off this one parse; a raw id it
+        // cannot split would be a permanent error at hash time, long after the
+        // record looked fine.
+        let id = video_record_id("4D7u5KF7SP8");
+        let (ns, vid) = split_track_record_id(&id).expect("splits");
+        assert_eq!(ns, VIDEO_RECORD_NS);
+        assert_eq!(vid, "4D7u5KF7SP8");
+    }
+
+    #[tokio::test]
+    async fn compute_outcomes_serves_a_raw_record_id() {
+        let p = YouTubePlugin::default();
+        let out = p
+            .compute_outcomes(&video_record_id("4D7u5KF7SP8"))
+            .await
+            .expect("ok");
+        assert_eq!(out.len(), 1);
+        assert_eq!(
+            out[0].hash.0,
+            meta_feeder_sdk::hash::compute_yt_video_cid("video", "4D7u5KF7SP8").unwrap()
+        );
+        assert!(out[0].bytes.is_none(), "the tier has no byte path");
+    }
+
+    #[test]
+    fn raw_record_carries_the_fields_a_consumer_gates_on() {
+        let p = YouTubePlugin::default();
+        let c = Candidate {
+            video_id: "4D7u5KF7SP8".to_string(),
+            title: "Three-Body EP01".to_string(),
+            uploader_channel: "UCabc".to_string(),
+            duration_secs: 2712,
+            uploader: "Tencent Video".to_string(),
+        };
+        let filters: BTreeMap<String, Vec<String>> = [
+            ("fileType".to_string(), vec!["video".to_string()]),
+            ("tmdbid".to_string(), vec!["95479".to_string()]),
+        ]
+        .into_iter()
+        .collect();
+        let r = p.video_record(&c, &filters).expect("record");
+        let f = &r.fields;
+
+        assert_eq!(f.get("fileType").map(String::as_str), Some("video"));
+        assert_eq!(f.get("youtubeid").map(String::as_str), Some("4D7u5KF7SP8"));
+        assert_eq!(f.get("ytMatchKey").map(String::as_str), Some(RAW_MATCH_KEY));
+        assert_eq!(f.get("fileinfo/duration").map(String::as_str), Some("2712"));
+        assert!(
+            f.contains_key("source/gateway:youtube"),
+            "raw bucket needs it"
+        );
+        assert!(f.contains_key("poster_url") && f.contains_key("description/eng"));
+        let cid = meta_feeder_sdk::hash::compute_yt_video_cid("video", "4D7u5KF7SP8").unwrap();
+        assert_eq!(
+            f.get(&format!("cids/{cid}")).map(String::as_str),
+            Some("true")
+        );
+
+        // The query's anchor is echoed…
+        assert_eq!(f.get("tmdbid").map(String::as_str), Some("95479"));
+        // …but never promoted into a trust claim. A raw hit that stamped
+        // `anchored` would leak into the enriched grid as if it had been
+        // verified, which is the whole failure this tier is shaped to avoid.
+        assert!(!f.contains_key("anchored"));
+        // No guessed contentKind — `pack` in particular would be dropped
+        // outright by meta-watch's raw projection.
+        assert!(!f.contains_key("contentKind"));
+    }
+
+    #[tokio::test]
+    async fn a_music_video_query_never_reaches_the_raw_path() {
+        // meta-listen's artist row is `mbArtistId:<uuid> fileType:video
+        // contentKind:musicVideo`. It requests video and would satisfy the raw
+        // branch's file-type test, so the ONLY things keeping it on the
+        // verified path are the dispatch order and the anchor guard. With no
+        // MusicBrainz client configured the verified path yields nothing —
+        // which is exactly how we can tell the two apart: a raw answer here
+        // would be non-empty.
+        let p = YouTubePlugin::default();
+        let q = vq(
+            "daft punk",
+            &[
+                ("mbArtistId", &["056e4f3e-d505-4dad-8ec1-d04f521cbb56"]),
+                ("fileType", &["video"]),
+                ("contentKind", &["musicVideo"]),
+            ],
+        );
+        assert!(p.handle_query(&q, 20).await.expect("ok").is_empty());
+    }
+
+    #[tokio::test]
+    async fn an_album_query_never_reaches_the_raw_path() {
+        // meta-listen's album query is `fileType:audio` — it must keep falling
+        // through to the MusicBrainz track path.
+        let p = YouTubePlugin::default();
+        let q = vq(
+            "daft punk discovery",
+            &[
+                (
+                    "mbReleaseGroupId",
+                    &["aa997ea0-2936-40bd-884d-3af8a0e064dc"],
+                ),
+                ("fileType", &["audio"]),
+                ("contentKind", &["track"]),
+            ],
+        );
+        assert!(p.handle_query(&q, 20).await.expect("ok").is_empty());
+    }
+
+    #[tokio::test]
+    async fn a_typed_video_query_with_no_free_text_returns_nothing() {
+        // A `tmdbid` alone cannot drive the raw path: there is no TMDB client
+        // here to turn the id into a searchable title, and searching for the
+        // literal number would be worse than refusing.
+        let p = YouTubePlugin::default();
+        let q = vq("", &[("fileType", &["video"]), ("tmdbid", &["95479"])]);
+        assert!(p.handle_query(&q, 20).await.expect("ok").is_empty());
+    }
+
+    #[tokio::test]
+    async fn a_typed_video_query_answers_from_a_free_text_search() {
+        use wiremock::matchers::{method, path, query_param};
+        use wiremock::{Mock, MockServer, ResponseTemplate};
+
+        let server = MockServer::start().await;
+        Mock::given(method("GET"))
+            .and(path("/search"))
+            .and(query_param("filter", "videos"))
+            .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
+                "items": [
+                    {
+                        "url": "/watch?v=4D7u5KF7SP8",
+                        "title": "Three-Body EP01",
+                        "uploaderUrl": "/channel/UCabc",
+                        "uploaderName": "Tencent Video",
+                        "duration": 2712
+                    },
+                    {
+                        "url": "/watch?v=OLAK5uy_kZ8X",
+                        "title": "Three-Body EP02",
+                        "uploaderUrl": "/channel/UCabc",
+                        "uploaderName": "Tencent Video",
+                        "duration": 2680
+                    }
+                ]
+            })))
+            .mount(&server)
+            .await;
+
+        let p = YouTubePlugin::with_bases(server.uri(), server.uri());
+        let q = vq("three-body", &[("fileType", &["video"])]);
+        let out = p.handle_query(&q, 20).await.expect("ok");
+
+        assert_eq!(out.len(), 2, "both hits emitted, unverified");
+        assert_eq!(out[0].upstream_id, "youtube");
+        assert_eq!(out[0].record_id, video_record_id("4D7u5KF7SP8"));
+        assert_eq!(
+            out[0].fields.get("ytMatchKey").map(String::as_str),
+            Some(RAW_MATCH_KEY)
+        );
+        // No channel/duration filter ran — this is the point of the tier, and a
+        // regression that reintroduced one would silently empty it.
+        assert_eq!(
+            out[1].fields.get("title").map(String::as_str),
+            Some("Three-Body EP02")
+        );
+    }
+
+    #[tokio::test]
+    async fn the_raw_ceiling_bounds_one_query() {
+        use wiremock::matchers::{method, path};
+        use wiremock::{Mock, MockServer, ResponseTemplate};
+
+        // 30 distinct hits, an 11-char id each.
+        let items: Vec<_> = (0..30)
+            .map(|i| {
+                serde_json::json!({
+                    "url": format!("/watch?v=vid{:08}", i),
+                    "title": format!("hit {i}"),
+                    "uploaderUrl": "/channel/UCabc",
+                    "uploaderName": "Someone",
+                    "duration": 100
+                })
+            })
+            .collect();
+        let server = MockServer::start().await;
+        Mock::given(method("GET"))
+            .and(path("/search"))
+            .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
+                "items": items
+            })))
+            .mount(&server)
+            .await;
+
+        let p = YouTubePlugin::with_bases(server.uri(), server.uri());
+        let q = vq("anything", &[("fileType", &["video"])]);
+        // `max_results` above the ceiling: the ceiling wins, because every hit
+        // costs the core a thumbnail fetch.
+        let out = p.handle_query(&q, 500).await.expect("ok");
+        assert_eq!(out.len(), YT_VIDEO_SEARCH_MAX);
+
+        // …and below it, the caller's own budget wins.
+        let out = p.handle_query(&q, 3).await.expect("ok");
+        assert_eq!(out.len(), 3);
+    }
+
     fn tref(video: &str, delta: i64, key: &str) -> TrackRef {
         TrackRef {
             video_id: video.to_string(),