test(mux): skip truncated spliced groups as aborted

quest/m1/README.md

@@ -30,7 +30,6 @@ QUIC studies there on that rule.
 - [One route cost](/quest/m1/route-cost.md) - Warm and Cold collapse to one static route cost
 - [Delete removed cluster flags](/quest/m1/cluster-shims.md) - once a release has carried their refusals, `mesh` and `linger` leave `cluster::Config`
 - [Track tail interop](/quest/m1/track-tail-interop.md) - a Rust publisher ending a track with a group in flight is read to its end by the JS subscriber, and the reverse, in `just test interop`
-- [moq-mux truncated group](/quest/m1/mux-truncated-group.md) - a test pins that moq-mux skips a truncated spliced group as aborted
 - [Demand everywhere](/quest/m1/demand-everywhere.md) - requests and producers at every level watch subscribers through `demand()` alone, ahead of the bindings exposing it
 - [Merge queue](/quest/m1/merge-queue.md) - the required checks run on `merge_group`, so a stale green check can no longer break main
 - [Binding audio delay](/quest/m1/binding-surface.md) - moq-ffi and every wrapper configure and observe audio playout delay

rs/moq-mux/src/container/consumer.rs

@@ -1643,6 +1643,91 @@ mod tests {
 		assert!(frames.len() >= 4, "Expected >= 4 frames, got {}", frames.len());
 	}
 
+	#[tokio::test(start_paused = true)]
+	async fn truncated_spliced_group_skips_to_the_next_clean_group() {
+		let origin = crate::source::produce_origin();
+		let hops = moq_net::Hops::try_from(vec![moq_net::Hop::new(10).unwrap()]).unwrap();
+		let first_route = origin
+			.dynamic(
+				"live",
+				moq_net::origin::Route::default().with_hops(hops.clone()).with_cost(5),
+			)
+			.unwrap();
+		let pending = origin.consume().request_broadcast("live");
+		let first = moq_net::broadcast::Info::new().produce();
+		let info = hang::container::track_info(hang::catalog::PRIORITY.video);
+		let first_track = first.create_track("video", info.clone()).unwrap();
+		first_route.requested_broadcast().await.unwrap().accept(&first);
+		let broadcast = pending.await.unwrap();
+		let track = broadcast
+			.track("video")
+			.unwrap()
+			.subscribe(moq_net::track::Subscription::default().with_max_age(Duration::from_secs(2)))
+			.await
+			.unwrap();
+		let format = Container::Legacy(crate::container::Kind::Data);
+		let mut consumer = Consumer::new(track, Container::Legacy(crate::container::Kind::Data));
+
+		let mut open = first_track.create_group(0u64.into()).unwrap();
+		format
+			.write(
+				&mut open,
+				&[Frame {
+					timestamp: ts(0),
+					payload: Bytes::from_static(&[0xDE, 0xAD]),
+					keyframe: true,
+					duration: None,
+				}],
+			)
+			.unwrap();
+
+		// Each cheaper route advances beyond the seam without serving its continuation.
+		// Enough takeovers prune every route that could still cover group 0.
+		let mut routes = Vec::new();
+		let mut sources = Vec::new();
+		for sequence in 1..=4 {
+			let route = origin
+				.dynamic(
+					"live",
+					moq_net::origin::Route::default()
+						.with_hops(hops.clone())
+						.with_cost(5 - sequence),
+				)
+				.unwrap();
+			let source = moq_net::broadcast::Info::new().produce();
+			let mut track = source.create_track("video", info.clone()).unwrap();
+			route.requested_broadcast().await.unwrap().accept(&source);
+			track.demand().used().await.unwrap();
+			if sequence == 1 {
+				assert_eq!(consumer.read().await.unwrap().unwrap().timestamp, ts(0));
+			}
+			write_group(&mut track, sequence, &[ts(sequence * 100_000)]);
+			routes.push(route);
+			sources.push((source, track));
+		}
+
+		let waiter = kio::Waiter::noop();
+
+		// Group 1 was never handed out before its segment was pruned. An abort
+		// skips straight to the retained successor; a short clean group emits GroupEnd.
+		let event = consumer.poll_event(&waiter);
+		let Poll::Ready(Ok(Some(Event::Frame(frame)))) = event else {
+			panic!(
+				"expected a successor frame, current={}; pending={}",
+				consumer.current,
+				event.is_pending()
+			);
+		};
+		assert_eq!(frame.timestamp, ts(200_000));
+		assert!(frame.keyframe);
+		assert_eq!(consumer.current, 2);
+		assert!(
+			matches!(consumer.poll_event(&waiter), Poll::Ready(Ok(Some(Event::GroupEnd)))),
+			"the complete successor still emits its clean boundary"
+		);
+		assert_eq!(consumer.read().await.unwrap().unwrap().timestamp, ts(300_000));
+	}
+
 	// ---- Eviction recovery (pause/resume) ----
 
 	/// A group that aged out of the relay cache (aborted with `Error::Old`) while the