feat(studio): guests see the program and hear a mix-minus

README.md

@@ -243,7 +243,8 @@ program.
 ## Guests
 
 Guests and phones join a room the studio owns on its relay, from a web page,
-and their camera and screen appear as broadcast sources. The room follows
+and their camera and screen appear as broadcast sources. The page plays the
+show back to them, without their own voice. The room follows
 [moq-room](https://github.com/moq-dev/moq/tree/main/rs/moq-room): a guest
 named `alice` publishes `alice/camera.hang` and `alice/screen.hang` under it.
 Guests need a relay that verifies tokens, so run your own with
@@ -277,9 +278,10 @@ moq-studio show.toml --connect 'https://relay.example.com/show?jwt=...' \
   and each token the studio renews it with. While a guest is in the room, the
   studio signs them a fresh token every half of it, over a session of its own
   in the room that the room key signs for, and the join page presents it
-  without reconnecting. A guest's token may read only their own renewals, at
-  `.renew/<name>/<generation>` in the room, but anything else the relay lets subscribe
-  there can read every guest's, so grant that only to the studio. A guest who
+  without reconnecting. A guest's token may read only their own renewals and
+  return feed, at `.renew/<name>/<generation>` and `.return/<name>/<generation>`
+  in the room, and the program at `.program`, but anything else the relay lets
+  subscribe there can read every guest's, so grant that only to the studio. A guest who
   leaves needs a link, or a renewed token, that has not expired to come back.
   The Guests panel says when renewal is failing.
 - **Upgrading from a studio without renewal:** generate a new room key, drop
@@ -300,7 +302,7 @@ moq-studio show.toml --connect 'https://relay.example.com/show?jwt=...' \
 a QR code; **Copy link** puts it on the clipboard instead. The guest's token is
 in the link's fragment, which the browser never sends to a server, and lets
 them publish only under their own name, so they cannot appear as someone else,
-and read only their own renewals.
+and read only their own renewals, their return feed, and the program.
 Treat the link like a password: anyone holding it can join as that guest
 before it expires, and is then renewed for as long as they stay, until you
 remove them. A guest's broadcasts become sources named after them (`alice`,
@@ -314,13 +316,28 @@ who reconnects after you removed their source comes back under a new name
 **Remove from scenes** is your edit, so it undoes: it takes a guest, any clone
 of their sources, and any mask reading them out of every scene and the top
 layer and mutes them. A signed token cannot be taken back, so it also stops
-renewing theirs, and they are disconnected within `--guest-expiry`. That
+renewing theirs and ends their return feed, and they are disconnected within
+`--guest-expiry`. That
 holds across a restart, whichever scene file you open: removed guests are
 kept beside the room key, as `room.guests.toml` for `room.jwk`, so its
 directory must be writable. Undo puts them back in the scenes but not on
 renewal; **New link** lets them back in and renews them again, under a new
 generation, so a link from before the removal cannot read their renewals.
 
+While a guest is in the room, the join page plays their return feed: the
+program's video, and the mix on program without that guest's microphone (a
+mix-minus), so they follow the show without hearing themselves back. Ask
+guests to wear headphones, so the show does not reach their microphone. The
+studio publishes it over its session in the room, never with its own
+credentials: the program once at `.program`, and each guest's feed at
+`.return/<name>/<generation>`. The video is the program's own two sizes,
+and a guest's player picks one for their link; this session uploads the size
+a guest watches a second time, and a guest watching the half size makes it
+encode as any viewer does. Each guest in the room costs one 64 kbps Opus
+encode of their mix-minus. A guest's
+screen share stays in their feed, since only their camera's microphone is
+taken out.
+
 ## Scene file
 
 The collection is TOML, so it diffs and reviews like code. It names devices,
@@ -718,7 +735,8 @@ just join    # Serve the join page locally
 ```
 
 `rs/moq-scene` is the UI-free compositor, `rs/moq-studio` the app, and
-`js/join` the join page, built on `@moq/room` and `@moq/publish`. Like the
+`js/join` the join page, built on `@moq/room`, `@moq/publish`, and
+`@moq/watch`. Like the
 Rust crates, those come from the `moq` submodule rather than npm, so
 `bun install` needs it checked out, and Nix to build `@moq/libav`. See
 [AGENTS.md](AGENTS.md) and [CONTRIBUTING.md](CONTRIBUTING.md), and the

js/join/index.html

@@ -23,6 +23,8 @@ <h1>Join the studio</h1>
 		<section id="joined" hidden>
 			<p class="status"><span id="status-dot" class="dot"></span><span id="status">Connecting</span></p>
 			<p id="capture" class="error" hidden></p>
+			<canvas id="show"></canvas>
+			<p class="hint">The show, without your own voice. Wear headphones, so it does not reach your microphone.</p>
 			<canvas id="preview"></canvas>
 			<div class="controls">
 				<button id="camera" type="button">Camera</button>

js/join/package.json

@@ -16,7 +16,8 @@
 		"@moq/net": "workspace:*",
 		"@moq/publish": "workspace:*",
 		"@moq/room": "workspace:*",
-		"@moq/signals": "workspace:*"
+		"@moq/signals": "workspace:*",
+		"@moq/watch": "workspace:*"
 	},
 	"devDependencies": {
 		"@types/bun": "^1.4.4",

js/join/src/feed.ts

@@ -0,0 +1,17 @@
+/**
+ * The return feed: while a guest is in the room, the studio publishes the show
+ * back to them, the program with everyone's sound but their own, at a path only
+ * their own token may read.
+ *
+ * @module
+ */
+
+import * as Net from "@moq/net";
+
+/**
+ * Where the studio publishes `name`'s return feed in `generation`: `Room::feed_path` in
+ * `rs/moq-studio/src/guests.rs`.
+ */
+export function path(name: string, generation: number): Net.Path.Valid {
+	return Net.Path.from(`.return/${name}/${generation}`);
+}

js/join/src/index.css

@@ -38,6 +38,11 @@ canvas {
 	border-radius: 0.375rem;
 }
 
+.hint {
+	color: #aaa;
+	font-size: 0.875rem;
+}
+
 .controls {
 	display: flex;
 	flex-wrap: wrap;

js/join/src/index.ts

@@ -1,7 +1,8 @@
 /**
  * The join page: a guest opens the studio's link and publishes their camera,
  * microphone, and screen into the studio's room, following moq-room's
- * `{name}/camera.hang` and `{name}/screen.hang`.
+ * `{name}/camera.hang` and `{name}/screen.hang`, and plays the show the studio
+ * sends back.
  *
  * @module
  */
@@ -10,6 +11,8 @@ import * as Net from "@moq/net";
 import * as Publish from "@moq/publish";
 import { Local } from "@moq/room";
 import { Effect, Signal } from "@moq/signals";
+import * as Watch from "@moq/watch";
+import * as Feed from "./feed.ts";
 import { format, type Link, parse } from "./link.ts";
 import * as Renew from "./renew.ts";
 
@@ -67,6 +70,16 @@ function join(link: Link): () => void {
 	});
 	signals.cleanup(() => renewal.close());
 
+	// The program, with everyone's sound but the guest's own, so they hear no echo of themselves.
+	const show = new Watch.Player({
+		origin: connection.origin,
+		name: Feed.path(link.name, link.generation),
+		// The guest's receive rate, so the player drops to the half size their link can carry.
+		probe: connection.probe,
+		video: { canvas: element<HTMLCanvasElement>("show") },
+	});
+	signals.cleanup(() => show.close());
+
 	const local = new Local({
 		connection,
 		identity: Net.Path.from(link.name),

quest/m0/README.md

@@ -98,7 +98,6 @@ repository.
 - [Destinations](/quest/m0/destinations.md) - RTMP, RTMPS, and WHIP push with per-destination codecs and shared encoders, on a UI-free egress library a hosted service can reuse
 - [Multiple canvases](/quest/m0/multi-canvas.md) - landscape and portrait canvases composite at once from shared sources
 - [The join page is live](/quest/m0/join-page-live.md) - `join.studio.moq.dev` serves the join page `--join-page` defaults to
-- [Guest return feed](/quest/m0/guest-return.md) - guests see the program and hear a mix-minus
 - [Clap-to-measure sync delay](/quest/m0/sync-clap.md) - a clap sets a microphone's delay against its camera
 - [Audio filters](/quest/m0/audio-filters.md) - per-source noise suppression, gate, compressor, limiter, and plugins, configured from the studio
 - [Browser sources on macOS and Windows](/quest/m0/browser-source.md) - the out-of-process CEF pages Linux has, with shared GPU textures on the other desktops

rs/moq-studio/src/engine.rs

@@ -13,6 +13,7 @@
 //! - **audio-encode** encodes the mix on program into the program audio, and
 //!   **audio-record** the mix recordings hold into the recording broadcast's
 //!   own audio while a recording reads it.
+//! - Each guest's return feed encodes its mix-minus on the tokio runtime.
 //! - **encode**: one per program rendition, which waits for each program copy
 //!   and encodes it. A new canvas size or frame rate starts new renditions.
 //! - Recordings write their files on the tokio runtime.
@@ -557,6 +558,8 @@ pub struct Engine {
 	pub recorder: record::Recorder,
 	/// The room guests join, when configured.
 	pub room: Option<Arc<crate::guests::Room>>,
+	/// Each guest's return feed, once the studio's session in the room dials.
+	feeds: Mutex<Option<crate::feed::Feeds>>,
 	/// The program's broadcast.
 	pub program: Program,
 	/// Keeps broadcast sources' origin open when no connection holds it.
@@ -700,6 +703,7 @@ impl Engine {
 			output,
 			recorder,
 			room,
+			feeds: Mutex::default(),
 			program,
 			_origin: subscribe,
 			stopped: AtomicBool::new(false),
@@ -719,6 +723,13 @@ impl Engine {
 		// Where the program clock starts on the catalog's, so the wall time the
 		// catalog advertises for a frame is when the program drew it.
 		let epoch = Duration::from(clock.capture(start)?);
+		if let Some(room) = &engine.room
+			&& let Some(origin) = room.origin()
+		{
+			let program = engine.program.publication.broadcast().consume();
+			let feeds = crate::feed::Feeds::new(room.clone(), origin.clone(), program, clock.clone(), epoch)?;
+			*engine.feeds.lock().unwrap() = Some(feeds);
+		}
 		let program = engine.clone();
 		let thread = clocked("program".into(), async move || {
 			program.run_program(start, epoch, renditions, ladder).await
@@ -755,6 +766,18 @@ impl Engine {
 		if let Some(guests) = guests {
 			tokio::spawn(crate::guests::run(engine.clone(), guests));
 		}
+		if let Some(room) = &engine.room {
+			// Who has a return feed follows who is renewed, synced with the mixes.
+			let mut changes = room.changes();
+			// Including any made since the feeds started.
+			changes.mark_changed();
+			let follow = engine.clone();
+			tokio::spawn(async move {
+				while changes.changed().await.is_ok() {
+					follow.state.lock().unwrap().stale = true;
+				}
+			});
+		}
 
 		Ok(engine)
 	}
@@ -834,8 +857,8 @@ impl Engine {
 			.map(|monitor| (monitor.name().to_string(), monitor.failed()))
 	}
 
-	/// Point the program audio, the recorded audio, and the monitor at the
-	/// mixes `state` now has.
+	/// Point the program audio, the recorded audio, the monitor, and each
+	/// guest's return feed at the mixes `state` now has.
 	fn pick_mixes(&self, state: &State) {
 		let mixes = &state.collection.mixes;
 		if let Some(mix) = state.sources.mix(&mixes.program) {
@@ -844,6 +867,10 @@ impl Engine {
 		if let Some(mix) = state.sources.mix(mixes.recorded()) {
 			self.recorded_audio.set(mix.clone());
 		}
+		// From the mix on program, which is what each guest follows.
+		if let Some(feeds) = self.feeds.lock().unwrap().as_mut() {
+			feeds.sync(&state.collection, &state.sources);
+		}
 		if let Some(monitor) = self.monitor.lock().unwrap().as_mut() {
 			if !mixes.names.iter().any(|name| name == monitor.name()) {
 				monitor.set_name(&mixes.program);
@@ -2337,6 +2364,56 @@ mod tests {
 		.await
 	}
 
+	/// A guest's return feed lists every size of `engine`'s program that
+	/// `published` lists, read from the program as served at `.program`, which
+	/// holds the program's tracks and none of the sources'.
+	async fn returned(engine: &Engine, published: &hang::Catalog) {
+		use crate::guests::{PROGRAM, Room, tests};
+
+		let connect = url::Url::parse("https://relay.example/show").unwrap();
+		let page = url::Url::parse("https://studio.example/join/").unwrap();
+		let key = tests::key("show/guests", &["**"], &["**"]);
+		let room = Arc::new(Room::new(&connect, "guests", key, page, tests::EXPIRY).unwrap());
+		let camera = moq_net::Path::new("alice/camera.hang").unwrap();
+		room.observe(&camera, true, &Collection::new(), &History::new());
+		let origin = moq_tokio::origin::spawn();
+		let program = engine.program.publication.broadcast().consume();
+		let clock = engine.program.clock.clone();
+		let mut feeds = crate::feed::Feeds::new(room, origin.clone(), program, clock, Duration::ZERO).unwrap();
+		{
+			let state = engine.state.lock().unwrap();
+			feeds.sync(state.collection(), &state.sources);
+		}
+
+		let format = moq_mux::catalog::CatalogFormat::Hang;
+		let feed = Room::feed_path("alice", 0).unwrap();
+		let broadcast = origin.consume().follow(feed, None).unwrap().available().await.unwrap();
+		let mut listings = moq_mux::catalog::Consumer::<()>::new(&broadcast, format).await.unwrap();
+		let listed = kio::clock::timeout(TIMEOUT, async {
+			loop {
+				let catalog = listings.next().await.unwrap().expect("a catalog");
+				if !catalog.video.renditions.is_empty() {
+					return catalog;
+				}
+			}
+		})
+		.await
+		.expect("the feed lists the program's video");
+		let names = |video: &hang::catalog::Video| video.renditions.keys().cloned().collect::<Vec<_>>();
+		assert_eq!(names(&listed.video), names(&published.video), "both sizes");
+		for config in listed.video.renditions.values() {
+			assert_eq!(config.broadcast.as_ref().unwrap().as_str(), "../../.program");
+		}
+
+		let served = origin.consume().request(PROGRAM, None).await.unwrap();
+		let mut served = moq_mux::catalog::Consumer::<moq_scene::publish::Ext>::new(&served, format)
+			.await
+			.unwrap();
+		let served = served.next().await.unwrap().expect("a catalog");
+		assert_eq!(names(&served.video), names(&published.video));
+		assert!(served.ext.studio.sources.is_empty(), "a source reached the guests");
+	}
+
 	/// The program a studio publishes reaches a relay as the picture it composed,
 	/// at the canvas size and at half of it, with its audio alongside. The half
 	/// size holds an encode session only once watched.
@@ -2397,6 +2474,7 @@ mod tests {
 		let echo = catalog.echo.as_ref().expect("the catalog solicits feedback");
 		assert_eq!(echo.path.as_str(), "studio/viewers");
 		assert_eq!(engine.stats.sessions.open(), 1, "the half size is not watched yet");
+		returned(&engine, &catalog).await;
 		let mut pictures = subscribe(&broadcast, &video, config).await;
 		let shown = red(&mut pictures, moq_video::Size::new(640, 360))
 			.await