feat(cli): publish stdin imports on the broadcast clock

doc/bin/cli.md

@@ -53,6 +53,13 @@ moq --connect https://relay.example.com/anon --broadcast my-stream.hang export t
 moq --connect "https://relay.example.com/rooms/1?jwt=$TOKEN" --broadcast alice.hang import ts
 ```
 
+The `ts`, `fmp4`, and `flv` imports publish on the broadcast clock the catalog
+advertises, not the input's own timestamps. The first frame is stamped when it
+arrives, every track keeps its offset from the others, and an input that
+restarts its timestamps, such as a restarted encoder, continues forward
+after the real gap rather than rewinding. So a feed whose PTS starts hours in,
+or whose first frame arrives late, still names the right wall time.
+
 MPEG-TS import carries H.264/H.265 and AAC/MP2/AC-3/E-AC-3, passes SCTE-35 and
 subtitle PIDs through as tracks, and round-trips the service tables. A
 `discontinuity_indicator` on the program's PCR PID is a system time-base reset,

doc/lib/rs/moq-mux.md

@@ -80,6 +80,14 @@ The producer sets the entry's `mode` and encodes the track with its
 `compression`. Read it back from `Catalog<Ext>` and subscribe with
 `catalog::Entry::new(name, &entry.binary)`.
 
+The fMP4, MPEG-TS, and FLV importers publish the source's own timestamps unless
+built with `live()`, which translates them onto the catalog's broadcast clock:
+the first frame is live on arrival, every track of the input shares that one
+mapping, and a source that restarts its timestamps continues forward after the
+real idle gap. fMP4 passthrough rewrites each fragment's `tfdt` to match. Use
+it for a live feed with its own zero; publish verbatim only when the catalog's
+clock (`Config::with_clock`) already names the source's zero.
+
 ```bash
 cargo add moq-mux
 ```

quest/m1/2278-watch-absolute-wall-clock-latency-target-for-synchronized.md

@@ -28,10 +28,6 @@ application knows whether remote clocks are synchronized.
 Verify application access using the built-in publisher integration, including
 a live-only broadcast with no archive timeline.
 
-## Required
-
-- [CLI import clock](/quest/m1/cli-import-clock.md) - built-in publishers populate the mapping applications read
-
 ## Closes
 
 - [#2278](https://github.com/moq-dev/moq/issues/2278) - close this issue when the quest finishes

quest/m1/README.md

@@ -33,7 +33,6 @@ transport, benchmark tooling); worktrees isolate commits, not semantics.
 - [Track demand](/quest/m1/track-demand.md) - Rust and JS watch a track's subscribers through `demand()` alone
 - [Broadcast close](/quest/m1/broadcast-close/README.md) - `close()` is the one way to end a broadcast in every language, a permanent retraction that leaves in-flight tracks alone
 - [Relay peer set](/quest/m1/relay-peer-set.md) - a wire consumer tells a client hop from a peer hop, and every mesh credential can mark a peer
-- [CLI import clock](/quest/m1/cli-import-clock.md) - fMP4, TS, and FLV imports publish on the shared broadcast clock across restarts
 - [CLI inspection](/quest/m1/cli-inspect/README.md) - `moq ls` lists what is live and `moq fetch` reads a group over MoQ, and a guide shows how to inspect a relay
 - [JS caught up](/quest/m1/js-announce-caught-up.md) - @moq/net's announce consumer says when the initial set has landed, like Rust
 - [Bindings caught up](/quest/m1/announce-live-bindings.md) - moq-ffi, libmoq, and every wrapper yield the same flat announce event, `Live` included

quest/m2/teleop/correlation.md

@@ -30,4 +30,3 @@ the same property that makes an MCAP recording valuable.
 ## Required
 
 - [Robot teleoperation primitive](/quest/m2/teleop/robot.md)
-- [CLI import clock](/quest/m1/cli-import-clock.md) - publishers populate the fixed broadcast mapping used to join tracks

rs/moq-cli/src/publish.rs

@@ -255,6 +255,9 @@ impl Publish {
 	/// `broadcast`. Announce the broadcast afterwards: this constructor creates
 	/// the catalog tracks, so announcing after it lands the advertisement with
 	/// the tracks already in place.
+	///
+	/// Stdin is a live feed with its own zero, so the container importers translate its
+	/// timestamps onto the broadcast clock the catalog advertises (`live`).
 	pub fn new(
 		mut broadcast: moq_net::broadcast::Producer,
 		format: &PublishFormat,
@@ -267,7 +270,7 @@ impl Publish {
 		if let PublishFormat::Ts = format {
 			let config = config.with_catalog(moq_mux::catalog::hang::Catalog::<ts::Ext>::default());
 			let catalog = moq_mux::catalog::Producer::new(&mut broadcast, config)?;
-			let ts = ts::Import::new(broadcast.clone(), catalog.reserve());
+			let ts = ts::Import::new(broadcast.clone(), catalog.reserve()).live();
 			return Ok(Self {
 				source: Source::Stream(PublishDecoder::Ts(Box::new(ts))),
 				broadcast,
@@ -286,12 +289,12 @@ impl Publish {
 				})
 			}
 			PublishFormat::Fmp4 => {
-				let fmp4 = fmp4::Import::new(broadcast.clone(), catalog.reserve());
+				let fmp4 = fmp4::Import::new(broadcast.clone(), catalog.reserve()).live();
 				Source::Stream(PublishDecoder::Fmp4(Box::new(fmp4)))
 			}
 			PublishFormat::Ts => unreachable!("TS is handled above with the mpegts catalog extension"),
 			PublishFormat::Flv => {
-				let flv = flv::Import::new(broadcast.clone(), catalog.reserve());
+				let flv = flv::Import::new(broadcast.clone(), catalog.reserve()).live();
 				Source::Stream(PublishDecoder::Flv(Box::new(flv)))
 			}
 		};
@@ -757,6 +760,54 @@ mod tests {
 		);
 	}
 
+	/// `moq import ts` publishes on the broadcast clock it advertises: a feed arriving a minute
+	/// after the broadcast began is live on arrival, not stamped with its own PTS (1.4s into bbb).
+	#[tokio::test]
+	async fn ts_import_publishes_on_the_broadcast_clock() {
+		let ago = Duration::from_secs(60);
+		let clock = moq_mux::Clock::at(std::time::Instant::now() - ago, std::time::SystemTime::now() - ago).unwrap();
+		let broadcast = moq_net::broadcast::Info::new().produce();
+		let consumer = broadcast.consume();
+		let config = moq_mux::catalog::Config::default().with_clock(clock);
+		let mut publish = Publish::new(broadcast, &PublishFormat::Ts, config).unwrap();
+		#[allow(irrefutable_let_patterns)]
+		let Source::Stream(decoder) = &mut publish.source else {
+			panic!("expected a stream source");
+		};
+		let before = clock.now();
+		decoder.decode_chunk(BBB).unwrap();
+		let after = clock.now();
+		decoder.finish().unwrap();
+
+		let catalog = hang::catalog::Catalog::<()>::subscribe(&consumer)
+			.await
+			.unwrap()
+			.next()
+			.await
+			.unwrap()
+			.expect("a catalog");
+		assert_eq!(
+			catalog.clock,
+			Some(clock.wall()),
+			"the advertised clock is the one stamped on"
+		);
+		let (name, config) = catalog.video.renditions.iter().next().expect("a video rendition");
+		let track = consumer.track(name).unwrap().subscribe(None).await.unwrap();
+		let container = moq_mux::catalog::hang::Container::try_from(config).unwrap();
+		let first = Consumer::new(track, container)
+			.read()
+			.await
+			.unwrap()
+			.expect("a video frame")
+			.timestamp;
+		// The PES that anchors the mapping need not be this frame: the mux spaces them apart.
+		let skew = Duration::from_secs(2).as_micros();
+		assert!(
+			before.as_micros() - skew <= first.as_micros() && first.as_micros() <= after.as_micros() + skew,
+			"the first frame is live on arrival: {first:?} not in {before:?}..={after:?}"
+		);
+	}
+
 	/// Read the first frame of a verbatim track back as raw bytes.
 	async fn read_frame(consumer: &moq_net::broadcast::Consumer, name: &str) -> Vec<u8> {
 		let track = consumer.track(name).unwrap().subscribe(None).await.unwrap();