Bind borrowed stream labels by the stream they name, not by a slot

A vendor label's `stream_number` is a slot in the one stream table its
config blob describes. The labels merged in from the playlists — to
cover streams the vendor named nothing for — carried a different number
entirely: a dense counter over every distinct stream found while
scanning the whole disc in directory order, related to no playlist's
slot numbering at all. Two coordinate systems, one field name. The
merge matched them by equality and the binder then counted streams
against the result.

Measured over the 44-image corpus: 22 discs merge such labels; of the
566 places one lands on a stream, 443 (78%) are a stream it does not
describe — the label states the PID it read itself from and it is a
different one. 142 of those are stopped by the language check 94377c7
added; 301 are applied. Those labels carry no editorial payload, so the
direct damage is confined to codec text — but the polluted list is also
what the anchor gate reads, and on 11 disc/stream-type pairs it is what
decides the anchor, which is how it reaches the vendor's forced and SDH
flags. 53 anchor facts are harvested off a merged slot. The clip-info
orphans had the same shape, numbered from `max + 1` of a list they share
no coordinate system with.

A label now either NAMES its stream — `StreamId { clip, pid }`, read out
of the very table `disc::bluray` builds the stream from — or it does
not, and only the ones that do not are ever reached by counting:

  * `label_at` returns vendor labels only, so the two numberings can no
    longer be confused by construction.
  * Playlist and clip-info labels bind by id. Exact, no anchor, no
    sequence, no ordinal.
  * A named stream outranks a guessed one, so an editorial flag reaches
    a stream only where the disc's own numbering puts it there.
  * The presence of an id is the provenance the anchor gate was missing.
    It reads the vendor's slots alone now, so a slot the vendor never
    named no longer breaks the sequence — under-yield is the normal
    shape of these blobs — and a title with fewer streams than the list
    has slots is no longer eligible to hold it. Ranking prefers the
    title that positively confirms most of the list.
  * An orphan is in no playlist, hence in no title, so it binds to
    nothing rather than to whatever counted its way.

The MPLS floor is one label per physical stream keyed by `(clip, PID)`,
not by `(type, language, codec, pid)`: the same PID in two clips is two
streams, and the old key collapsed them. Its `stream_number` is now the
entry's real slot in its own playlist's table.

41 of 44 images are byte-identical; all three that move lose a label
they should not have had, and no feature title changes on any image. A
dozen featurette playlists stop reporting a feature subtitle's SDH
marking on their own unrelated subtitle; eleven menu and bonus titles
stop advertising the feature's object-audio format on plain stereo; and
on a disc whose every title carries a single audio stream — too short a
table to anchor anything — a regional-variant tag asserted on all
seventeen titles is now asserted on none, in exchange for every title
stating the codec it actually carries, which none of them did.

Six tests had been asserting the invented numbering, including one
pinning the disc-global counter as a deliberate property.

Also fixes a defect in the same family that the corpus work surfaced:
`pgs_forced_probe::apply_verdicts` set `forced` but left `qualifier`,
so a demoted track shipped with a metadata sidecar calling it forced
next to a Matroska header saying it is not. Only a forced claim is
cleared; an SDH marking is not the probe's to touch.

src/disc/pgs_forced_probe.rs

@@ -854,6 +854,21 @@ fn apply_verdicts(title: &mut DiscTitle, verdicts: &HashMap<u16, bool>) {
             && let Some(&forced) = verdicts.get(&sub.pid)
         {
             sub.forced = forced;
+            // A demoted track must not go on describing itself as forced. The
+            // flag and the qualifier are two renderings of one fact for
+            // different consumers — the muxer writes `FlagForced` from
+            // `forced`, the JSON sidecar writes its qualifier string from
+            // `qualifier` — so leaving `Forced` behind here published a track
+            // whose sidecar said "forced" and whose Matroska header said it was
+            // not. The probe read the content; it outranks the vendor's claim.
+            //
+            // Only this direction is a contradiction. `qualifier == None` on a
+            // track the probe promoted is not one: `None` is the absence of an
+            // editorial qualifier, not an assertion that the track is ordinary,
+            // and the vendor never claimed otherwise.
+            if !forced && sub.qualifier == crate::disc::LabelQualifier::Forced {
+                sub.qualifier = crate::disc::LabelQualifier::None;
+            }
         }
     }
 }
@@ -2045,6 +2060,52 @@ mod tests {
         })
     }
 
+    /// Spec: a verdict that DEMOTES a track clears a `Forced` qualifier with
+    /// it. The two fields are one fact rendered for two consumers — the muxer
+    /// writes Matroska `FlagForced` from `forced`, the JSON metadata sidecar
+    /// writes its qualifier string from `qualifier` — so leaving the qualifier
+    /// behind publishes a track that calls itself forced next to a header that
+    /// says it is not.
+    ///
+    /// Mutation: delete the qualifier assignment in `apply_verdicts` — the
+    /// track comes out `forced == false, qualifier == Forced`.
+    #[test]
+    fn a_demoted_track_stops_calling_itself_forced() {
+        let mut title = pgs_title(0x1200, true);
+        if let Stream::Subtitle(sub) = &mut title.streams[0] {
+            sub.qualifier = LabelQualifier::Forced;
+        }
+        apply_verdicts(&mut title, &HashMap::from([(0x1200u16, false)]));
+        let Stream::Subtitle(sub) = &title.streams[0] else {
+            unreachable!()
+        };
+        assert!(!sub.forced);
+        assert_eq!(
+            sub.qualifier,
+            LabelQualifier::None,
+            "the content outranks the vendor's claim, and both renderings of it move together"
+        );
+    }
+
+    /// Spec: a qualifier that is not a forced claim is not the probe's to
+    /// touch. `Sdh` says something about the track's content that a
+    /// forced-narrative verdict neither confirms nor refutes.
+    ///
+    /// Mutation: clear the qualifier unconditionally on demotion — the SDH
+    /// marking is lost.
+    #[test]
+    fn a_demoted_track_keeps_a_qualifier_that_is_not_a_forced_claim() {
+        let mut title = pgs_title(0x1200, true);
+        if let Stream::Subtitle(sub) = &mut title.streams[0] {
+            sub.qualifier = LabelQualifier::Sdh;
+        }
+        apply_verdicts(&mut title, &HashMap::from([(0x1200u16, false)]));
+        let Stream::Subtitle(sub) = &title.streams[0] else {
+            unreachable!()
+        };
+        assert_eq!(sub.qualifier, LabelQualifier::Sdh);
+    }
+
     fn forced_flag(title: &DiscTitle, pid: u16) -> bool {
         title
             .streams

src/labels/criterion.rs

@@ -47,6 +47,7 @@ pub fn parse(reader: &mut dyn SectorSource, udf: &UdfFs) -> Option<ParseResult>
     let mut labels = Vec::new();
     for (info, &stream_num) in stream_infos.iter().zip(stream_nums.iter()) {
         labels.push(StreamLabel {
+            stream_id: None,
             stream_number: stream_num,
             stream_type: info.stream_type,
             language: info.language.clone(),

src/labels/ctrm.rs

@@ -207,6 +207,7 @@ fn parse_language_streams_text(text: &str) -> Vec<StreamLabel> {
         }
 
         labels.push(StreamLabel {
+            stream_id: None,
             stream_number: stream_num,
             stream_type,
             language,
@@ -401,6 +402,7 @@ mod tests {
 
     fn lbl(t: StreamLabelType, n: u16, name: &str) -> StreamLabel {
         StreamLabel {
+            stream_id: None,
             stream_number: n,
             stream_type: t,
             language: String::new(),
@@ -910,6 +912,7 @@ fn parse_menu_base_text(text: &str) -> Vec<StreamLabel> {
             .unwrap_or_default();
 
         labels.push(StreamLabel {
+            stream_id: None,
             stream_number: stream_num,
             stream_type,
             language,

src/labels/dbp.rs

@@ -171,6 +171,7 @@ fn make_label(num: u16, label: String, stream_type: StreamLabelType) -> StreamLa
     let qualifier = vocab::qualifier(&label);
     let purpose = vocab::purpose(&label);
     StreamLabel {
+        stream_id: None,
         stream_number: num,
         stream_type,
         language,

src/labels/deluxe.rs

@@ -1066,6 +1066,7 @@ fn interpret_streams(constructions: &[Construction], master: &MasterEnumTable) -
         }
 
         out.push(StreamLabel {
+            stream_id: None,
             stream_number,
             stream_type,
             language,

src/labels/mpls_universal.rs

@@ -86,44 +86,50 @@ pub fn parse(reader: &mut dyn SectorSource, udf: &UdfFs) -> Option<ParseResult>
     Some(ParseResult::low(labels))
 }
 
-/// Convert every stream entry across `playlists` into deduped
-/// [`StreamLabel`]s. Factored out of [`parse`] so unit tests can drive
-/// the actual conversion logic (stream-type mapping, dedup key, dense
-/// global counters) directly from already-parsed [`crate::mpls::Playlist`]
-/// values, without needing a synthetic on-disc UDF image.
+/// Convert every stream entry across `playlists` into one [`StreamLabel`] per
+/// physical stream. Factored out of [`parse`] so unit tests can drive the
+/// actual conversion logic (stream-type mapping, identity, slot numbering)
+/// directly from already-parsed [`crate::mpls::Playlist`] values, without
+/// needing a synthetic on-disc UDF image.
+///
+/// Identity is `(clip, PID)` — what the STN entry states — and it is both the
+/// dedup key and the label's [`StreamId`]. A stream twenty playlists list is
+/// one label; two clips that both open their first audio at 0x1100 are two.
+/// This replaced a disc-global dense counter that numbered surviving entries
+/// 1, 2, 3, … in playlist-directory order: that number was not an STN slot in
+/// anything, but it was handed to a binder that reads `stream_number` as one.
 fn build_labels(playlists: &[crate::mpls::Playlist]) -> Vec<StreamLabel> {
+    use std::collections::HashSet;
     let mut labels: Vec<StreamLabel> = Vec::new();
-    // (stream_type_tag, language, codec_hint, pid) — PID is the
-    // canonical "same physical stream" key; type+lang+codec round
-    // out the rare case where two distinct logical streams happen
-    // to share a PID across playlists with different metadata.
-    let mut seen: Vec<(StreamLabelType, String, String, u16)> = Vec::new();
-
-    // Global 1-based counters keyed by StreamLabelType. Incremented
-    // only when an entry survives dedup, so stream_numbers are dense
-    // (1, 2, 3, ...) per type across the whole disc — not reset per
-    // playlist. A disc with 2 MPLS files that each list the same
-    // 8 audio streams ends up with audio_1..audio_8, not audio_1..
-    // audio_16 or audio_1..audio_8 with audio_1 duplicated.
-    let mut audio_idx: u16 = 0;
-    let mut sub_idx: u16 = 0;
+    let mut seen: HashSet<super::StreamId> = HashSet::new();
 
     for playlist in playlists {
+        // `Playlist::streams` is the FIRST play item's STN table, so every
+        // entry here is a stream of that play item's clip — the same clip
+        // `disc::bluray` records as the title's `clips[0]`. That pairing is
+        // what makes the PID an identity rather than a 16-bit number.
+        //
+        // Streams cannot be non-empty without a play item to have read them
+        // from, so the empty case is unreachable on a real disc; entries we
+        // cannot identify are skipped rather than emitted as unbindable
+        // labels.
+        let Some(clip_id) = playlist.play_items.first().map(|pi| pi.clip_id.clone()) else {
+            continue;
+        };
+
+        // 1-based STN slot within THIS playlist's table, per type — the
+        // `stream_number` field's documented meaning, counted the same way
+        // `disc::bluray` counts the stream list it builds from these entries.
+        // Nothing binds through it (these labels bind by id); it is stated
+        // truthfully rather than invented so that a reader of the label list
+        // sees where on its own playlist each stream sits.
+        let mut audio_idx: u16 = 0;
+        let mut sub_idx: u16 = 0;
+
         for entry in &playlist.streams {
             let Some(label_type) = label_type_for(entry) else {
                 continue;
             };
-
-            let language = normalize_language(&entry.language);
-            let name = language_display_name(&language);
-            let codec_hint = build_codec_hint(label_type, entry);
-
-            let key = (label_type, language.clone(), codec_hint.clone(), entry.pid);
-            if seen.contains(&key) {
-                continue;
-            }
-            seen.push(key);
-
             let stream_number = match label_type {
                 StreamLabelType::Audio => {
                     audio_idx += 1;
@@ -135,7 +141,20 @@ fn build_labels(playlists: &[crate::mpls::Playlist]) -> Vec<StreamLabel> {
                 }
             };
 
+            let stream_id = super::StreamId {
+                clip_id: clip_id.clone(),
+                pid: entry.pid,
+            };
+            if !seen.insert(stream_id.clone()) {
+                continue;
+            }
+
+            let language = normalize_language(&entry.language);
+            let name = language_display_name(&language);
+            let codec_hint = build_codec_hint(label_type, entry);
+
             labels.push(StreamLabel {
+                stream_id: Some(stream_id),
                 stream_number,
                 stream_type: label_type,
                 language,
@@ -365,10 +384,24 @@ mod tests {
         }
     }
 
+    /// A playlist over clip "00001". `Playlist::streams` is read out of the
+    /// first play item's STN table, so a playlist that has streams always has
+    /// a play item to have read them from — the fixture carries one so tests
+    /// exercise the shape production sees, and so each label gets the
+    /// `(clip, PID)` identity it is bound by.
     fn playlist_with(streams: Vec<StreamEntry>) -> Playlist {
+        playlist_on("00001", streams)
+    }
+
+    fn playlist_on(clip_id: &str, streams: Vec<StreamEntry>) -> Playlist {
         Playlist {
             version: "0200".to_string(),
-            play_items: Vec::new(),
+            play_items: vec![crate::mpls::PlayItem {
+                clip_id: clip_id.to_string(),
+                in_time: 0,
+                out_time: 0,
+                connection_condition: 1,
+            }],
             streams,
             marks: Vec::new(),
         }
@@ -503,31 +536,48 @@ mod tests {
         );
     }
 
+    /// Two playlists over the SAME clip that both list PID 0x1100: one
+    /// physical stream, so one label. Identity is `(clip, PID)`, and each
+    /// label states the STN slot it holds in its own playlist.
     #[test]
-    fn dedup_streams_across_playlists() {
-        // Two playlists, same English TrueHD 7.1 PID 0x1100 in both.
-        // Expect one Audio label, not two.
-        let pl1 = playlist_with(vec![
-            audio_entry(0x1100, 0x83, 12, 1, "eng"),
-            audio_entry(0x1101, 0x81, 6, 1, "fra"),
-        ]);
-        let pl2 = playlist_with(vec![
-            audio_entry(0x1100, 0x83, 12, 1, "eng"), // duplicate
-            audio_entry(0x1102, 0x82, 6, 1, "deu"),  // new
-        ]);
+    fn one_label_per_stream_across_playlists_on_the_same_clip() {
+        let pl1 = playlist_on(
+            "00001",
+            vec![
+                audio_entry(0x1100, 0x83, 12, 1, "eng"),
+                audio_entry(0x1101, 0x81, 6, 1, "fra"),
+            ],
+        );
+        let pl2 = playlist_on(
+            "00001",
+            vec![
+                audio_entry(0x1100, 0x83, 12, 1, "eng"), // same stream
+                audio_entry(0x1102, 0x82, 6, 1, "deu"),  // new
+            ],
+        );
         let labels = labels_from_playlists(&[pl1, pl2]);
-        // Expected: eng@0x1100, fra@0x1101, deu@0x1102 — three uniques.
-        assert_eq!(labels.len(), 3);
-        // PID isn't stored on StreamLabel, so assert on the surviving
-        // language set instead.
-        let mut langs: Vec<String> = labels.iter().map(|l| l.language.clone()).collect();
-        langs.sort();
-        assert_eq!(langs, vec!["deu", "eng", "fra"]);
+        assert_eq!(
+            labels.len(),
+            3,
+            "eng/fra/deu — the duplicate eng is one stream"
+        );
 
-        // Stream numbers must be DENSE and GLOBAL across playlists, not
-        // reset per playlist. eng (pl1) = 1, fra (pl1) = 2, the duplicate
-        // eng in pl2 is deduped (no number consumed), and deu (pl2) = 3.
-        // Regression guard for the per-playlist counter-reset divergence.
+        let id = |lang: &str| {
+            labels
+                .iter()
+                .find(|l| l.language == lang)
+                .and_then(|l| l.stream_id.clone())
+                .map(|i| (i.clip_id, i.pid))
+        };
+        assert_eq!(id("eng"), Some(("00001".into(), 0x1100)));
+        assert_eq!(id("fra"), Some(("00001".into(), 0x1101)));
+        assert_eq!(id("deu"), Some(("00001".into(), 0x1102)));
+
+        // `stream_number` is the entry's slot in ITS OWN playlist's STN table
+        // — deu is pl2's second audio, so 2, not "the third distinct stream
+        // seen while scanning the disc". It used to be the latter: a dense
+        // disc-global counter that named no table anyone could count against,
+        // handed to a binder that reads the field as an STN slot.
         let num = |lang: &str| {
             labels
                 .iter()
@@ -536,7 +586,25 @@ mod tests {
         };
         assert_eq!(num("eng"), Some(1));
         assert_eq!(num("fra"), Some(2));
-        assert_eq!(num("deu"), Some(3));
+        assert_eq!(num("deu"), Some(2), "pl2's second audio slot");
+    }
+
+    /// The same PID in two DIFFERENT clips is two different streams — a PID is
+    /// only unique within one clip. Deduping on the PID alone (as the old
+    /// key's `(type, language, codec_hint, pid)` did across clips) collapses
+    /// them into one label, and the second clip's stream is then described by
+    /// the first clip's.
+    #[test]
+    fn same_pid_in_two_clips_is_two_streams() {
+        let pl1 = playlist_on("00001", vec![audio_entry(0x1100, 0x83, 12, 1, "eng")]);
+        let pl2 = playlist_on("00002", vec![audio_entry(0x1100, 0x83, 12, 1, "eng")]);
+        let labels = labels_from_playlists(&[pl1, pl2]);
+        assert_eq!(labels.len(), 2, "different clips: two distinct streams");
+        let clips: Vec<String> = labels
+            .iter()
+            .filter_map(|l| l.stream_id.as_ref().map(|i| i.clip_id.clone()))
+            .collect();
+        assert_eq!(clips, vec!["00001", "00002"]);
     }
 
     #[test]

src/labels/paramount.rs

@@ -149,6 +149,7 @@ fn labels_from_feature(feature: &str) -> Vec<StreamLabel> {
                 LabelPurpose::Normal
             };
             labels.push(StreamLabel {
+                stream_id: None,
                 stream_number,
                 stream_type: StreamLabelType::Audio,
                 language: lang.to_string(),
@@ -202,6 +203,7 @@ fn labels_from_feature(feature: &str) -> Vec<StreamLabel> {
             };
 
             labels.push(StreamLabel {
+                stream_id: None,
                 stream_number,
                 stream_type: StreamLabelType::Subtitle,
                 language: lang.to_string(),

src/labels/pixelogic.rs

@@ -272,6 +272,7 @@ fn assign_labels(strings: &[String], unknown: &mut UnknownParts) -> Vec<StreamLa
                     }
                     audio_num += 1;
                     labels.push(StreamLabel {
+                        stream_id: None,
                         stream_number: audio_num,
                         ..label
                     });
@@ -282,6 +283,7 @@ fn assign_labels(strings: &[String], unknown: &mut UnknownParts) -> Vec<StreamLa
                     }
                     sub_num += 1;
                     labels.push(StreamLabel {
+                        stream_id: None,
                         stream_number: sub_num,
                         ..label
                     });
@@ -464,6 +466,7 @@ fn parse_token_inner(s: &str, mut unknown: Option<&mut UnknownParts>) -> Option<
     };
 
     Some(StreamLabel {
+        stream_id: None,
         stream_number: 0,
         stream_type,
         language: lang.to_string(),

src/labels/png_filenames.rs

@@ -50,6 +50,7 @@ fn labels_from_filenames(names: &[String]) -> Vec<StreamLabel> {
     seen.into_iter()
         .enumerate()
         .map(|(i, code)| StreamLabel {
+            stream_id: None,
             stream_number: (i as u16).saturating_add(1),
             stream_type: StreamLabelType::Audio,
             language: code.to_string(),