fix(mux): LOC video groups end with the duration marker

doc/concept/hang.md

@@ -138,16 +138,14 @@ The `container.kind` on each rendition says how frames are framed:
 A consumer skips renditions with a kind it doesn't recognize and carries them
 through when republishing the catalog.
 
-A Legacy video publisher can close the last frame's duration with an empty
+A Legacy or LOC video publisher can close the last frame's duration with an empty
 codec payload whose timestamp is that frame's exclusive end. Consumers treat it
 as metadata and never pass it to a decoder. This lets a group close immediately
 without waiting for the next frame. Audio has codec-defined durations, and CMAF
 carries sample durations directly, so neither needs per-group duration markers.
 Audio retains its separate terminal marker before codec drain packets, allowing
-consumers to discard encoder padding beyond the source endpoint. LOC readers
-also skip empty payloads; LOC writers wait for the compatibility release before
-emitting markers. Empty payloads on data tracks remain data, including empty
-text cues.
+consumers to discard encoder padding beyond the source endpoint. Empty payloads
+on data tracks remain data, including empty text cues.
 
 ## Groups and keyframes
 

doc/lib/rs/moq-mux.md

@@ -114,7 +114,7 @@ API: [docs.rs/moq-mux](https://docs.rs/moq-mux). Real-world usage:
 Container producers and consumers take a format configured from the track's audio
 or video catalog entry (`catalog::hang::Container::try_from(&config)`). For a raw
 track, supply `container::Kind` explicitly. `cut(Some(end))` flushes and closes the
-group immediately. Legacy video writes an empty timestamped frame at that end;
+group immediately. Legacy and LOC video write an empty timestamped frame at that end;
 audio and CMAF do not. With no explicit end, the producer uses a known sample
 duration or observed cadence, independently of batching and reorder jitter.
 Streaming consumers deliver frames immediately. The live fMP4 exporter receives

quest/m1/README.md

@@ -141,7 +141,6 @@ transport, benchmark tooling); worktrees isolate commits, not semantics.
 - [#933](/quest/m1/933-video-rotation-metadata-not-propagated-from-mobile-camera.md) - the catalog rotation follows the live camera's orientation
 - [#2848](/quest/m1/2848-follow-the-bandwidth-grant-in-moq-audio-instead-of.md) - the Opus producer follows its bandwidth grant through the settled `moq_mux::rate::Control`
 - [Ladder](/quest/m1/ladder/README.md) - a transcode ladder adapts to the uplink it publishes over, instead of encoding every live rung at its ceiling
-- [LOC duration marker](/quest/m1/loc-duration-marker.md) - LOC producers write the marker once released consumers skip it
 - [#2278](/quest/m1/2278-watch-absolute-wall-clock-latency-target-for-synchronized.md) - hang: document reading the catalog-root clock and converting PTS to wall time, without synchronizing library playback to wall time
 - [Time stretch](/quest/m1/watch-audio-time-stretch.md) - js/watch: the audio ring converges by time-stretching instead of skipping or going silent
 - [Native audio quality](/quest/m1/audio-quality-native.md) - the browser lane's profiles, budgets, and metric schema run against `moq play` on a dummy device

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

@@ -1340,7 +1340,7 @@ mod tests {
 		assert_eq!(micros, vec![0, 100_000], "markers skipped, next group reached");
 	}
 
-	/// LOC consumers skip an empty payload so later producers can write the duration marker.
+	/// LOC video consumers skip an empty payload: it is the duration marker, not media.
 	#[tokio::test]
 	async fn loc_empty_payload_is_skipped() {
 		let track = track_producer("test", hang::container::track_info(hang::catalog::PRIORITY.video));

rs/moq-mux/src/container/loc/mod.rs

@@ -82,4 +82,25 @@ impl Container for Wire {
 	fn end(&self, frame: &Frame) -> Option<moq_net::Timestamp> {
 		(self.0 != Kind::Data && frame.payload.is_empty()).then_some(frame.timestamp)
 	}
+
+	fn finish_group(
+		&self,
+		group: &mut moq_net::group::Producer,
+		end: Option<moq_net::Timestamp>,
+	) -> Result<(), Self::Error> {
+		if self.0 == Kind::Video
+			&& let Some(timestamp) = end
+		{
+			self.write(
+				group,
+				&[Frame {
+					timestamp,
+					payload: bytes::Bytes::new(),
+					keyframe: false,
+					duration: None,
+				}],
+			)?;
+		}
+		Ok(())
+	}
 }

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

@@ -1160,29 +1160,80 @@ mod tests {
 		assert_eq!(collect_payloads(consumer).await, vec![vec![(0, 2), (20_000, 2)]]);
 	}
 
-	/// LOC producers do not write the marker until skipping consumers have shipped.
+	/// Drain all LOC groups, returning each group's (timestamp_micros, payload_len) pairs.
+	async fn collect_loc_payloads(mut consumer: moq_net::track::Subscriber) -> Vec<Vec<(u64, usize)>> {
+		let mut groups = Vec::new();
+		while let Some(mut group) = consumer.recv_group().await.unwrap() {
+			let mut frames = Vec::new();
+			while let Some(frame) = group.read_frame().await.unwrap() {
+				let decoded = moq_loc::decode(frame.payload).unwrap();
+				frames.push((decoded.timestamp, decoded.payload.len()));
+			}
+			groups.push(frames);
+		}
+		groups
+	}
+
+	/// A LOC video group ends with an empty frame at the next keyframe's timestamp, like Legacy.
 	#[tokio::test]
-	async fn loc_cut_writes_no_duration_marker() {
+	async fn loc_cut_writes_a_duration_marker_at_the_callers_bound() {
 		let track = track_producer("test", hang::container::track_info(hang::catalog::PRIORITY.video));
 		let consumer = track.subscribe(replay());
 		let mut producer = Producer::new(track, Container::Loc(crate::container::Kind::Video));
 
 		producer.write(frame(0, true)).unwrap();
+		producer.write(frame(10_000, false)).unwrap();
 		producer
-			.cut(Some(moq_net::Timestamp::from_micros(33_000).unwrap()))
+			.cut(Some(moq_net::Timestamp::from_micros(15_000).unwrap()))
 			.unwrap();
+		producer.write(frame(20_000, true)).unwrap();
 		producer.finish().unwrap();
 
-		let mut groups = Vec::new();
-		let mut consumer = consumer;
-		while let Some(mut group) = consumer.recv_group().await.unwrap() {
-			let mut count = 0;
-			while group.next_frame().await.unwrap().is_some() {
-				count += 1;
-			}
-			groups.push(count);
-		}
-		assert_eq!(groups, vec![1], "LOC producers do not write the marker yet");
+		let groups = collect_loc_payloads(consumer).await;
+		assert_eq!(groups[0], vec![(0, 2), (10_000, 2), (15_000, 0)]);
+		assert_eq!(groups[1][0], (20_000, 2));
+		assert_eq!(groups[1].last().unwrap().1, 0, "finish closes the last group");
+	}
+
+	/// LOC audio never writes a duration marker, even at finish.
+	#[tokio::test]
+	async fn loc_audio_writes_no_duration_marker() {
+		let track = track_producer("test", hang::container::track_info(hang::catalog::PRIORITY.audio));
+		let consumer = track.subscribe(replay());
+		let mut producer = Producer::new(track, Container::Loc(crate::container::Kind::Audio));
+
+		producer.write(frame(0, true)).unwrap();
+		producer.write(frame(20_000, false)).unwrap();
+		producer
+			.cut(Some(moq_net::Timestamp::from_micros(30_000).unwrap()))
+			.unwrap();
+		producer.write(frame(40_000, true)).unwrap();
+		producer.finish().unwrap();
+
+		assert_eq!(
+			collect_loc_payloads(consumer).await,
+			vec![vec![(0, 2), (20_000, 2)], vec![(40_000, 2)]]
+		);
+	}
+
+	/// LOC data tracks never write a duration marker, so an empty data frame stays data.
+	#[tokio::test]
+	async fn loc_data_writes_no_duration_marker() {
+		let track = track_producer("test", hang::container::track_info(hang::catalog::PRIORITY.video));
+		let consumer = track.subscribe(replay());
+		let mut producer = Producer::new(track, Container::Loc(crate::container::Kind::Data));
+
+		producer.write(frame(0, true)).unwrap();
+		producer
+			.cut(Some(moq_net::Timestamp::from_micros(15_000).unwrap()))
+			.unwrap();
+		producer.write(frame(20_000, true)).unwrap();
+		producer.finish().unwrap();
+
+		assert_eq!(
+			collect_loc_payloads(consumer).await,
+			vec![vec![(0, 2)], vec![(20_000, 2)]]
+		);
 	}
 
 	/// `cut()` flushes the current group immediately; the next write must be a keyframe.