diff --git a/doc/bin/cli.md b/doc/bin/cli.md
index d3b3b27388e..c8dd21d2694 100644
--- a/doc/bin/cli.md
+++ b/doc/bin/cli.md
@@ -118,7 +118,7 @@ service; `export ts --passthrough` writes it back out. FLV accepts the same
 legacy and enhanced-RTMP codecs as the [RTMP gateway](/bin/rtmp).
 
 fMP4 import splits each fragment into one frame per sample, so players need no
-MP4 parser. `import fmp4 --passthrough` publishes each fragment whole as
+MP4 parser. `import fmp4 --cmaf` publishes each fragment whole as
 [`cmaf`](/concept/hang#container) instead, keeping the source's exact sample
 timing.
 
@@ -418,7 +418,7 @@ ending with `EXT-X-ENDLIST`. A dropped connection or an error closes the
 broadcast at once, and a reconnect, which publishes a newer
 [epoch](/concept/moq-lite#publisher-epochs) at the same path, ends the wait.
 
-`export ts`, `export flv`, and `export mkv` take `--delay` instead, like an SRT
+`export ts`, `export flv`, `export rtmp`, and `export mkv` take `--delay` instead, like an SRT
 receiver's latency, all tracks in decode order, so two exporters of one broadcast
 emit the same order. FLV and MKV mux each frame that long after its decode time
 (default 500 ms). TS output runs that far behind the source (default 1 s): the
@@ -429,7 +429,7 @@ top. A frame arriving after its deadline is dropped, and video resumes at its
 next keyframe. The delay also bounds how stale a group may get, so `--delay 0`
 writes each frame as soon as it is read and drops none as late, but still skips
 a stalled group once a newer one is ready.
-FLV and MKV reject their former `--max-delay` spelling with a migration
+FLV, MKV and RTMP refuse the released `--max-age` spelling with a migration
 message. The output follows the source's clock within what ISO/IEC 13818-1
 allows a TS PCR. A source further off fails the export, naming its rate, once
 it has used half the delay; a brief network stall only drops what it made late.
diff --git a/doc/bin/gstreamer.md b/doc/bin/gstreamer.md
index b77e58889fb..d03f1b35345 100644
--- a/doc/bin/gstreamer.md
+++ b/doc/bin/gstreamer.md
@@ -61,11 +61,15 @@ Each `sink_%u` request pad is one track. Pad properties: `track` names it
 [LOC](/concept/standard#loc) instead of the legacy hang container,
 `encoder=true` marks it as fed by a local encoder, and
 `track-status`/`track-error` report its lifecycle. Element properties:
-`url`, `broadcast`, `tls-disable-verify`, `quic-idle-timeout`, `quic-keylog`,
-`quic-keep-alive`, `linger`, and read-only `status`, `connected`, `moq-version`, and
+`url`, `broadcast`, `connect-version`, `tls-disable-verify`, `quic-idle-timeout`, `quic-keylog`,
+`quic-keep-alive`, and read-only `status`, `connected`, `moq-version`, and
 `estimated-send-rate`, `estimated-recv-rate`, `connection-stats`, and
 `sessions`. The sink reconnects for as long as the pipeline runs and only
 reports `failed` on an answer redialing can't change, such as a rejected token.
+Both elements accept the same connection properties. `connect-version` is a
+comma-separated list, matching the CLI's repeatable `--connect-version`;
+unknown versions fail startup. QUIC durations are milliseconds, with zero
+disabling the local idle timeout or keep-alive.
 Each run from `READY` publishes under a fresh
 [publisher epoch](/concept/moq-lite#publisher-epochs), so on moq-lite 07 (opt-in)
 or moq-transport draft 17 and later, viewers switch to a restarted pipeline at once instead of stalling on the old
@@ -77,12 +81,7 @@ it is released, reaches EOS, or fails; a flush or new stream after that EOS
 holds it back again. Opaque data pads are not listed. Later
 catalog updates do not wait for newly requested pads.
 
-Once every pad reaches EOS, the sink finishes the broadcast and posts EOS.
-With `linger=true` it first keeps serving the finished broadcast until its
-media is stale (30 seconds for audio and video), so a viewer still catching up
-gets the tail, and posts EOS only then, which is when `gst-launch-1.0 -e`
-exits. Stopping the pipeline, such as a second Ctrl-C under `-e`, closes the
-broadcast at once.
+Once every pad reaches EOS, the sink closes the broadcast and posts EOS.
 
 All pads share one media timeline on one wall clock. The first timed buffer places
 it using UNIX or NTP reference timestamp metadata; other clock domains use the
@@ -143,9 +142,9 @@ on its own. Losing the relay connection also holds the pads: `moqsrc` redials
 without a retry deadline and resumes when the broadcast returns. A refused
 connection, refused path or catalog, or malformed protocol input or catalog posts
 an error. Properties: `url`,
-`broadcast`, `versions` (comma-separated protocol versions to offer, most
+`broadcast`, `connect-version` (comma-separated protocol versions to offer, most
 preferred first; every stable version by default), `tls-disable-verify`,
-`quic-keylog`.
+`quic-idle-timeout`, `quic-keep-alive`, and `quic-keylog`.
 
 Debug with `GST_DEBUG=*:4` for GStreamer and `RUST_LOG=debug` for the plugin.
 To decrypt a capture, set `quic-keylog=true` on either element along with
diff --git a/doc/bin/hls.md b/doc/bin/hls.md
index 490a0323af0..96f9976a01e 100644
--- a/doc/bin/hls.md
+++ b/doc/bin/hls.md
@@ -67,13 +67,14 @@ returns HTTP 400. A live-only timeline or range past a finished recording
 returns HTTP 416 instead of a partial clip, and a failed timeline returns 500.
 
 `--listen-tls-cert`/`--listen-tls-key` or `--listen-tls-generate` serve HTTPS,
-and `--cors-origin` opens it to browsers. See `moq export hls --help`.
+and `--cors-origin` opens it to browsers. HLS TLS reads `MOQ_HLS_TLS_*`;
+`MOQ_LISTEN_TLS_*` configures only the MoQ listener. See `moq export hls --help`.
 
 ## Import
 
 Import handles classic HLS; LL-HLS parts are not implemented. It publishes the
 playlist's own media times, and anchors the catalog clock on the first imported
-segment. Each fMP4 segment is split into one frame per sample; `--passthrough`
+segment. Each fMP4 segment is split into one frame per sample; `--cmaf`
 publishes each fragment whole as CMAF instead, keeping its exact sample timing.
 
 The library, including the recorder cursors for mirroring a broadcast to
diff --git a/doc/bin/rtmp.md b/doc/bin/rtmp.md
index df5ff22bcb4..1ec618cc968 100644
--- a/doc/bin/rtmp.md
+++ b/doc/bin/rtmp.md
@@ -38,7 +38,7 @@ and anchors the catalog clock on the first frame, so it names the wall time the
 push arrived. A group starting before the previous group's start, such as an
 encoder restarting its timestamps mid-push, ends that push with an error.
 
-Egress uses `--max-delay` (default 500 ms) for both the FLV release delay and
+Egress uses `--delay` (default 500 ms) for both the FLV release delay and
 the stalled-group budget. Frames go out in decode order; a late video frame
 is dropped and video resumes at the next keyframe. Zero releases available
 frames immediately without deadline drops.
diff --git a/doc/lib/js/publish.md b/doc/lib/js/publish.md
index a9cb0263b66..70a92171d94 100644
--- a/doc/lib/js/publish.md
+++ b/doc/lib/js/publish.md
@@ -34,7 +34,6 @@ WebCodecs, writes the catalog, and publishes a hang broadcast.
 | `preview` | What the nested element shows: the raw `source` (default), a decoded copy of the `encoded` stream to see what viewers get, or `none`. |
 | `backend` | How a `<canvas>` preview is drawn; see [@moq/video](/lib/js/video). `auto` (default), `webgpu`, or `2d`. |
 | `announce` | When to advertise: once the `source` is live (default), `always`, or `never`. Nobody can see or subscribe to the broadcast until it is announced. |
-| `linger` | Finish the broadcast cleanly when advertising stops, such as a `source` that ends, instead of unannouncing it at once. It stays up while its media is fresh so viewers drain the tail, and the next announce publishes a new broadcast. Removing the element ends it at once. Off by default. |
 
 Every attribute is also a reactive property. The
 [README](https://www.npmjs.com/package/@moq/publish) lists types and defaults.
diff --git a/js/publish/README.md b/js/publish/README.md
index bfd1d907f34..f8a152cf403 100644
--- a/js/publish/README.md
+++ b/js/publish/README.md
@@ -72,7 +72,6 @@ The simplest way to publish a stream:
 | `preview`   | string  | `"source"` | What the preview renders: `"source"`, `"encoded"`, `"none"` |
 | `backend`   | string  | `"auto"` | Graphics API for a `<canvas>` preview: `"auto"`, `"webgpu"`, `"2d"` |
 | `announce`  | string  | `"source"` | When to advertise: `"always"`, `"never"`, `"source"` (once media is captured). A camera source waits for every enabled track, and `"source"` waits until each captured track's config resolves or fails, so the first catalog lists every rendition. The broadcast is created while connected either way. |
-| `linger`    | boolean | false    | When advertising stops, finish the broadcast cleanly instead of unannouncing it: it stays up while its media is fresh so viewers drain the tail, and the next announce publishes a new one. |
 
 A nested `<video>` shows the raw capture; a `<canvas>` is drawn by the element.
 Camera and microphone failures are observable through `el.sources.video`
diff --git a/js/publish/src/broadcast.test.ts b/js/publish/src/broadcast.test.ts
index e1b885461ea..9631e92b857 100644
--- a/js/publish/src/broadcast.test.ts
+++ b/js/publish/src/broadcast.test.ts
@@ -2,7 +2,7 @@ import { expect, test } from "bun:test";
 import { Manual, Milli, Nano } from "@moq/clock";
 import * as Catalog from "@moq/hang/catalog";
 import * as Json from "@moq/json";
-import { Origin, Path, Time, Track } from "@moq/net";
+import { Origin, Path, Track } from "@moq/net";
 import { Effect, Signal } from "@moq/signals";
 import { Broadcast } from "./broadcast.ts";
 
@@ -228,68 +228,9 @@ test("serves no catalog until announced, so the first snapshot is the one at ann
 	broadcast.close();
 });
 
-// With `linger`, a source that stops ends the publish cleanly: the finished broadcast stays announced
-// while its media is fresh, so a viewer still catching up drains the tail, then it closes.
-test("stopping the announce with linger serves the finished broadcast until its media is stale", async () => {
-	const clock = new Manual();
-	const installed = clock.install();
-	const origin = new Origin.Producer();
-	try {
-		const announce = new Signal(true);
-		const broadcast = new Broadcast({
-			enabled: true,
-			origin,
-			name: Path.from("test.hang"),
-			announce,
-			linger: true,
-		});
-		const video = broadcast.video("video");
-		video.config.set(videoConfig);
-		await settle();
-
-		const net = broadcast.net.peek();
-		if (!net) throw new Error("expected a network producer once connected");
-		const subscriber = net.track("video").subscribe();
-		await settle();
-		const track = video.track.peek();
-		if (!track) throw new Error("expected the rendition track once subscribed");
-		const group = track.appendGroup();
-		group.writeFrame({ payload: new Uint8Array([1]), timestamp: Time.Timestamp.now() });
-		group.close();
-
-		announce.set(false);
-		await settle();
-		expect(broadcast.net.peek()).toBeUndefined();
-		expect(net.route).toBeDefined();
-		const late = net.track("video").subscribe().ordered();
-		expect((await late.readFrame())?.payload).toEqual(new Uint8Array([1]));
-		const catalog = net.track(Broadcast.CATALOG_TRACK).subscribe();
-		expect(
-			(await new Json.Snapshot.Consumer<Catalog.Root>({ track: catalog }).latest())?.value.video,
-		).toBeDefined();
-
-		await clock.advance(Nano.fromMilli(Milli(30_000)));
-		expect(net.closed.peek()).toBeNull();
-		expect(net.route).toBeUndefined();
-		// A fresh broadcast waits unannounced for the next announce.
-		const next = broadcast.net.peek();
-		expect(next).toBeDefined();
-		expect(next).not.toBe(net);
-		expect(next?.route).toBeUndefined();
-
-		late.close();
-		catalog.close();
-		subscriber.close();
-		broadcast.close();
-	} finally {
-		origin.close();
-		installed.close();
-	}
-});
-
 // Regression: a re-announce re-created the catalog track, restarting its group sequence under the
-// same epoch, so a resumed viewer mistook the new catalog for the old one. Without `linger`, turning
-// the announce off unannounces at once and keeps the broadcast for the next announce.
+// same epoch, so a resumed viewer mistook the new catalog for the old one. Turning the announce off
+// unannounces at once and keeps the broadcast for the next announce.
 test("a re-announce continues the catalog's group sequence", async () => {
 	const announce = new Signal(true);
 	const broadcast = new Broadcast({
@@ -325,123 +266,6 @@ test("a re-announce continues the catalog's group sequence", async () => {
 	broadcast.close();
 });
 
-// With `linger`, a re-announce publishes a new broadcast under a new epoch, replacing the finished
-// one at once.
-test("announcing again replaces a finished broadcast with a new one", async () => {
-	const announce = new Signal(true);
-	const broadcast = new Broadcast({
-		enabled: true,
-		origin: new Origin.Producer(),
-		name: Path.from("test.hang"),
-		announce,
-		linger: true,
-	});
-	broadcast.video("video").config.set(videoConfig);
-	await settle();
-
-	const net = broadcast.net.peek();
-	if (!net) throw new Error("expected a network producer once connected");
-	const epoch = net.route?.epoch;
-	expect(epoch).toBeDefined();
-
-	announce.set(false);
-	await settle();
-	broadcast.audio("audio").config.set(audioConfig);
-	await settle();
-	expect(net.closed.peek()).toBeUndefined();
-	announce.set(true);
-	await settle();
-
-	expect(net.closed.peek()).toBeNull();
-	const next = broadcast.net.peek();
-	if (!next) throw new Error("expected a new network producer");
-	expect(next).not.toBe(net);
-	expect(next.route?.epoch).toBeDefined();
-	expect(next.route?.epoch).not.toBe(epoch);
-	const group = await next.track(Broadcast.CATALOG_TRACK).subscribe().ordered().nextGroup();
-	const catalog = (await group?.readJson()) as Catalog.Root;
-	expect(Object.keys(catalog.audio?.renditions ?? {})).toEqual(["audio"]);
-
-	broadcast.close();
-});
-
-// Disabling, renaming, or closing the publisher is abrupt: it ends a finished broadcast's linger at once.
-test("disabling, renaming, or closing ends a finished broadcast at once", async () => {
-	const enabled = new Signal(true);
-	const announce = new Signal(true);
-	const name = new Signal(Path.from("test.hang"));
-	const origin = new Origin.Producer();
-	const broadcast = new Broadcast({ enabled, origin, name, announce, linger: true });
-	broadcast.video("video").config.set(videoConfig);
-	await settle();
-
-	const first = broadcast.net.peek();
-	if (!first) throw new Error("expected a network producer once connected");
-	announce.set(false);
-	await settle();
-	expect(first.closed.peek()).toBeUndefined();
-	enabled.set(false);
-	await settle();
-	expect(first.closed.peek()).toBeNull();
-
-	enabled.set(true);
-	announce.set(true);
-	await settle();
-	const second = broadcast.net.peek();
-	if (!second) throw new Error("expected a network producer once re-enabled");
-	announce.set(false);
-	await settle();
-	expect(second.closed.peek()).toBeUndefined();
-	name.set(Path.from("renamed.hang"));
-	await settle();
-	expect(second.closed.peek()).toBeNull();
-
-	announce.set(true);
-	await settle();
-	const third = broadcast.net.peek();
-	if (!third) throw new Error("expected a network producer once announced again");
-	announce.set(false);
-	await settle();
-	expect(third.closed.peek()).toBeUndefined();
-	broadcast.close();
-	expect(third.closed.peek()).toBeNull();
-	origin.close();
-});
-
-// Regression: a rename or maxAge change in the same tick the announce stopped coalesced with it, so
-// the old broadcast finished and kept its path instead of ending at once.
-test("a rename as the announce stops ends the old broadcast at once", async () => {
-	const announce = new Signal(true);
-	const name = new Signal(Path.from("test.hang"));
-	const maxAge = new Signal<Milli | undefined>(undefined);
-	const origin = new Origin.Producer();
-	const broadcast = new Broadcast({ enabled: true, origin, name, announce, maxAge, linger: true });
-	broadcast.video("video").config.set(videoConfig);
-	await settle();
-
-	const first = broadcast.net.peek();
-	if (!first) throw new Error("expected a network producer once connected");
-	announce.set(false);
-	name.set(Path.from("renamed.hang"));
-	await settle();
-	expect(first.closed.peek()).toBeNull();
-	const second = broadcast.net.peek();
-	if (!second) throw new Error("expected a network producer under the new name");
-	expect(second).not.toBe(first);
-
-	announce.set(true);
-	await settle();
-	announce.set(false);
-	maxAge.set(Milli(1_000));
-	await settle();
-	expect(second.closed.peek()).toBeNull();
-	expect(broadcast.net.peek()).toBeDefined();
-	expect(broadcast.net.peek()).not.toBe(second);
-
-	broadcast.close();
-	origin.close();
-});
-
 // Regression: any rendition change re-created every rendition's track, so the next append restarted
 // a live track's group sequence.
 test("a rendition keeps its track across rendition changes", async () => {
diff --git a/js/publish/src/broadcast.ts b/js/publish/src/broadcast.ts
index 80496f9c8c0..63169486c96 100644
--- a/js/publish/src/broadcast.ts
+++ b/js/publish/src/broadcast.ts
@@ -30,15 +30,9 @@ export type BroadcastInput = {
 	// see or subscribe to it. The flip rather than a gate on creating it: tracks can be
 	// populated while this is false, then announced once ready. The catalog is served from the
 	// first announce, so its first snapshot is the one current at announce time.
-	// Turning it off unannounces the broadcast, unless `linger` finishes it.
+	// Turning it off unannounces the broadcast.
 	announce: Getter<boolean>;
 
-	// Whether turning `announce` off after an announce finishes the broadcast cleanly instead of
-	// unannouncing it. Defaults to false. A finished broadcast stays announced while its media is
-	// fresh, so a viewer still catching up drains the tail, then closes, and announcing again
-	// publishes a new broadcast that replaces it at once. Read when announcing stops.
-	linger: Getter<boolean>;
-
 	// The broadcast name.
 	name: Getter<Moq.Path.Valid>;
 
@@ -135,12 +129,6 @@ export class Broadcast {
 	// The identity of each track name in the latest catalog fold, recorded when the catalog is first served.
 	#listed = new Map<string, Identity>();
 
-	// The last network broadcast if it ended cleanly, serving its tail until its media is stale,
-	// with the identity it was published under.
-	#finished:
-		| { finish: Moq.Broadcast.Finish; origin: Moq.Origin.Table; name: Moq.Path.Valid; maxAge: Milli | undefined }
-		| undefined;
-
 	#signals = new Effect();
 
 	constructor(props?: Inputs<BroadcastInput>) {
@@ -148,7 +136,6 @@ export class Broadcast {
 			origin: getter(props?.origin),
 			enabled: getter(props?.enabled ?? true),
 			announce: getter(props?.announce ?? true),
-			linger: getter(props?.linger ?? false),
 			name: getter(props?.name ?? Moq.Path.empty()),
 			display: getter(props?.display),
 			flip: getter(props?.flip ?? false),
@@ -343,27 +330,6 @@ export class Broadcast {
 
 	#run(effect: Effect) {
 		const values = effect.getAll([this.in.enabled, this.in.origin]);
-
-		// A finished broadcast keeps its path until its linger ends. Announcing again replaces it,
-		// and anything else that reruns this (disabling, a new origin or name) ends it at once. The
-		// identity is compared, not just watched, since it may change in the same tick it finished.
-		const finished = this.#finished;
-		if (finished && finished.finish.closed.peek() === undefined) {
-			if (
-				values &&
-				!effect.get(this.in.announce) &&
-				values[1] === finished.origin &&
-				effect.get(this.in.name) === finished.name &&
-				effect.get(this.in.maxAge) === finished.maxAge
-			) {
-				effect.get(finished.finish.closed);
-				effect.cleanup(() => finished.finish.close());
-				return;
-			}
-			finished.finish.close();
-		}
-		this.#finished = undefined;
-
 		if (!values) return;
 		const [_enabled, origin] = values;
 
@@ -378,16 +344,7 @@ export class Broadcast {
 		// Creating into the origin outlives any single session: a reconnect re-announces the
 		// broadcast and new subscriptions land on the same producer.
 		const broadcast = origin.createBroadcast(name);
-		// Set once announcing stops after serving with `linger`, which ends this broadcast cleanly.
-		const ended = new Signal(false);
-		effect.get(ended);
-		effect.cleanup(() => {
-			if (!ended.peek()) return broadcast.close();
-			// Cleanups run last-in first-out, so every track already closed cleanly.
-			const finish = broadcast.finish();
-			this.#finished = { finish, origin, name, maxAge };
-			void finish.wait();
-		});
+		effect.cleanup(() => broadcast.close());
 		// One publisher instance per broadcast, kept across unannounce and announce, so a
 		// restart is a new broadcast rather than a resume into the old one.
 		const epoch = Moq.Epoch.mint();
@@ -416,9 +373,7 @@ export class Broadcast {
 		let serving = false;
 		effect.run((inner) => {
 			if (!inner.get(this.in.announce)) {
-				// Before the first announce there is nothing to finish.
-				if (serving && this.in.linger.peek()) ended.set(true);
-				else broadcast.unannounce();
+				broadcast.unannounce();
 				return;
 			}
 
@@ -506,7 +461,6 @@ export class Broadcast {
 	/** Stop and release everything this holds. */
 	close() {
 		this.#signals.close();
-		this.#finished?.finish.close();
 	}
 }
 
diff --git a/js/publish/src/element.ts b/js/publish/src/element.ts
index 25340772a79..356e6f5930c 100644
--- a/js/publish/src/element.ts
+++ b/js/publish/src/element.ts
@@ -35,18 +35,7 @@ function parseVersions(value: string | null): string[] | undefined {
 	return versions?.length ? versions : undefined;
 }
 
-const OBSERVED = [
-	"url",
-	"versions",
-	"name",
-	"muted",
-	"invisible",
-	"source",
-	"preview",
-	"backend",
-	"announce",
-	"linger",
-] as const;
+const OBSERVED = ["url", "versions", "name", "muted", "invisible", "source", "preview", "backend", "announce"] as const;
 type Observed = (typeof OBSERVED)[number];
 
 /** The built-in capture sources selectable via the `source` attribute. */
@@ -128,8 +117,6 @@ export default class MoqPublish extends HTMLElement {
 		backend: new Signal<Preview.Backend>("auto"),
 		// When to advertise the broadcast: always, never, or only once a source is live.
 		announce: new Signal<AnnounceMode>("source"),
-		// Whether advertising stopping finishes the broadcast, serving its tail until stale.
-		linger: new Signal(false),
 	};
 
 	/**
@@ -233,7 +220,6 @@ export default class MoqPublish extends HTMLElement {
 			origin: this.connection.origin,
 			enabled: this.#enabled,
 			announce: this.#announcing,
-			linger: this.controls.linger,
 			name: this.#name,
 			display: this.#capture.out.display,
 			flip: this.#flip,
@@ -361,8 +347,6 @@ export default class MoqPublish extends HTMLElement {
 			this.controls.source.set(parseSource(newValue));
 		} else if (name === "announce") {
 			this.controls.announce.set(parseAnnounce(newValue));
-		} else if (name === "linger") {
-			this.controls.linger.set(parseBoolean(newValue, false));
 		} else if (name === "muted") {
 			this.controls.muted.set(parseBoolean(newValue, false));
 		} else if (name === "invisible") {
@@ -544,15 +528,6 @@ export default class MoqPublish extends HTMLElement {
 	set announce(value: AnnounceMode) {
 		this.controls.announce.set(value);
 	}
-
-	/** Whether advertising stopping finishes the broadcast, keeping it up while its media is fresh. */
-	get linger(): boolean {
-		return this.controls.linger.peek();
-	}
-
-	set linger(value: boolean) {
-		this.controls.linger.set(value);
-	}
 }
 
 customElements.define("moq-publish", MoqPublish);
diff --git a/quest/m1/api-audit/README.md b/quest/m1/api-audit/README.md
index ce991f34eee..b5296a8dbcf 100644
--- a/quest/m1/api-audit/README.md
+++ b/quest/m1/api-audit/README.md
@@ -79,7 +79,7 @@ Rejected in planning:
 - [Bindings read gaps and ends](/quest/m1/api-audit/decode-read-bindings.md) - moq-ffi and every wrapper surface `Gap` and `End` instead of discarding them
 - [Media surface trim](/quest/m1/api-audit/media-trim.md) - media crate items with no consumer go private, and Rust/JS media names agree
 - [One frame rate name](/quest/m1/api-audit/frame-rate.md) - core `moq_video::Rate` becomes `FrameRate`, as the bindings name it
-- [CLI and gst flags](/quest/m1/api-audit/cli-flags.md) - one flag name per meaning across CLI subcommands and gst elements, and `linger` means one thing
+- [Pinned consumers adopt the CLI flag names](/quest/m1/api-audit/cli-flags-consumers.md) - pinned downstream deployments move past the flag rename and use its spellings
 - [Transport config](/quest/m1/api-audit/transport-config.md) - moq-tokio and moq-quic configs have one meaning for unset, enums for modes, no order-dependent setters, and no leftover features
 - [Session-only items stay private](/quest/m1/api-audit/model-private.md) - model items only the built-in sessions use are private
 - [Only sessions accept a request](/quest/m1/api-audit/claim-request-type.md) - accepting an application claim's request fails to compile, and the peer-route hatches are crate-private
diff --git a/quest/m1/api-audit/cli-flags-consumers.md b/quest/m1/api-audit/cli-flags-consumers.md
new file mode 100644
index 00000000000..cde8e5f6793
--- /dev/null
+++ b/quest/m1/api-audit/cli-flags-consumers.md
@@ -0,0 +1,32 @@
+# [S] Pinned consumers adopt the CLI flag names
+
+## Goal
+
+Downstream deployments that pin moq at a commit before
+[moq-dev/moq#6522](https://github.com/moq-dev/moq/pull/6522) move their pin to
+a commit that includes it and use its spellings, so the next release cut does
+not break their invocations. Done when no pinned consumer passes a spelling
+that #6522 renamed or removed.
+
+## Plan
+
+Decided 2026-10-23: the upstream rename landed on its own, and this migration
+was split out because it can only follow a pin bump.
+
+Move each pin and rewrite its invocations in the same change. The older pins
+only understand the released retry spellings, so renaming first breaks the
+binary they run. Upstream keeps no compatibility aliases; released spellings
+fail at startup and name their replacement.
+
+The `## Unreleased` entries in the `MIGRATE.md` of moq-cli and moq-gst are the
+full list, plus the two unreleased `linger` opt-ins below. Search each consumer for:
+
+- `--backoff-*`, `--goaway-*`, `MOQ_BACKOFF_*`, `MOQ_GOAWAY_*`: add the
+  `connect-` / `CONNECT_` prefix.
+- `MOQ_LISTEN_TLS_*` on `moq export hls`: `MOQ_HLS_TLS_*`.
+- `--passthrough` on `import fmp4` or `import hls`: `--cmaf`. TS keeps
+  `--passthrough`.
+- `--max-delay` on `export rtmp`: `--delay`.
+- `moqsrc versions=`: `connect-version=`.
+- `moqsink linger` and the `<moq-publish>` `linger` attribute: remove them;
+  both close at the end, as they did without the opt-in.
diff --git a/quest/m1/api-audit/cli-flags.md b/quest/m1/api-audit/cli-flags.md
deleted file mode 100644
index 88fccdfa8cd..00000000000
--- a/quest/m1/api-audit/cli-flags.md
+++ /dev/null
@@ -1,39 +0,0 @@
-# [S] CLI and gst flags
-
-## Goal
-
-One flag or property name means one thing across moq-cli subcommands and
-the gst elements, and the same knob has the same name everywhere.
-
-## Plan
-
-Decided 2026-10-22 (API audit):
-
-- `export rtmp --max-delay` and `export flv --delay` name the same paced
-  release; use `--delay` on both (neither spelling was released). Delete the
-  hidden refusal of `--max-delay` on flv/mkv (`args.rs`): only released
-  spellings get refusals, and release had `--max-age` there.
-- `--passthrough` means CMAF fragments on `import fmp4|hls` (#6250) and the
-  raw TS multiplex on `import|export ts`. Rename the fmp4/hls one to
-  `--container cmaf` or another distinct name, consistent with
-  [container formats](/quest/m1/api-audit/container-format.md).
-- `linger`: drop moqsink's new boolean `linger` property and
-  `<moq-publish>`'s `linger` attribute, keeping today's close at the end.
-  Revised 2026-10-23 (maintainer): don't make every publisher finish through
-  moq-net's `Finish`; [broadcast lifecycle](/quest/m1/broadcast-lifecycle/README.md)
-  deletes it and owns completion and retention.
-- gst: moqsrc's `versions` (comma list) versus the CLI's repeatable
-  `--connect-version`; mirror one name, give moqsink the same, and give both
-  elements the same connection properties (moqsrc lacks
-  `quic-idle-timeout` and `quic-keep-alive`). moq-tokio has both
-  `client-version` and `connect-version`; keep one.
-- `auth serve --key`/`--key-dir`/`--key-set` conflicts become clap
-  conflicts, not a runtime check.
-- `--backoff-*` and `--goaway-*` take the `connect-` prefix their TOML keys
-  have; `export hls --listen-tls-*` stops sharing `MOQ_LISTEN_TLS_*` with the
-  MoQ listener; keep rustdoc links out of `--help`.
-
-Update `doc/bin/cli.md`, `doc/bin/gstreamer.md`, and every invocation in the
-repo and moq.pro.
-
-Public API: CLI and gst breaking. Wire: none.
diff --git a/quest/m1/broadcast-lifecycle/README.md b/quest/m1/broadcast-lifecycle/README.md
index 909b355a5d4..6179c67f015 100644
--- a/quest/m1/broadcast-lifecycle/README.md
+++ b/quest/m1/broadcast-lifecycle/README.md
@@ -55,5 +55,5 @@ timer on main at 1582fdc07d:
 ## Related
 
 - [moq-transport finish interop](/quest/m2/ietf-finish.md) - the interop `finish` case, whose late readers rely on retention
-- [CLI and gst flags](/quest/m1/api-audit/cli-flags.md) - removes the opt-in `linger` booleans, leaving completion to this epic
+- [CLI and gst flags](https://github.com/moq-dev/moq/pull/6522) - removed the opt-in `linger` booleans, leaving completion to this epic
 - [Publication owner](/quest/m1/api-audit/announce-owner.md) - adds `origin::Publication`; it must not forward a `finish` this epic deletes
diff --git a/rs/moq-cli/src/args.rs b/rs/moq-cli/src/args.rs
index 9cc63db9f92..5cbc04a7324 100644
--- a/rs/moq-cli/src/args.rs
+++ b/rs/moq-cli/src/args.rs
@@ -849,7 +849,7 @@ impl ImportSource {
 pub struct Fmp4Import {
 	/// Publish each fragment whole as CMAF instead of splitting it into one frame per sample.
 	/// Keeps the source's exact sample timing, but every player has to parse ISO BMFF.
-	#[usage(long)]
+	#[usage(long = "cmaf")]
 	pub passthrough: bool,
 }
 
@@ -977,8 +977,8 @@ impl Export {
 			ExportSink::Ts(args) => found.extend(args.deprecated()),
 			ExportSink::Flv(args) => found.extend(args.deprecated()),
 			ExportSink::H264(args) | ExportSink::H265(args) => found.extend(args.deprecated()),
-			ExportSink::Hls(hls) => found.extend(hls.tls.deprecated()),
 			ExportSink::Rtmp(rtmp) => found.extend(rtmp.deprecated()),
+			ExportSink::Hls(hls) => found.extend(hls.tls.deprecated()),
 			_ => {}
 		}
 		found
@@ -1156,9 +1156,6 @@ pub struct Buffered {
 	/// Accepted only to refuse a nonzero value: only TS can mark a broadcast restart.
 	#[usage(long, default = "0s", hide = true)]
 	pub linger: moq_tokio::cli::Duration,
-	/// The retired staleness-only setting.
-	#[usage(long, hide = true)]
-	max_delay: Option<moq_tokio::cli::Duration>,
 	/// An older spelling of the staleness setting.
 	#[usage(long = "max-age", hide = true)]
 	max_age: Option<moq_tokio::cli::Duration>,
@@ -1171,7 +1168,6 @@ impl Buffered {
 	fn deprecated(&self) -> moq_tokio::cli::Deprecated {
 		let mut found = moq_tokio::cli::Deprecated::default();
 		for (present, flag) in [
-			(self.max_delay.is_some(), "--max-delay"),
 			(self.max_age.is_some(), "--max-age"),
 			(self.latency_max.is_some(), "--latency-max"),
 		] {
@@ -1329,6 +1325,164 @@ mod tests {
 		}
 	}
 
+	#[test]
+	fn hls_tls_environment_is_independent_of_the_moq_listener() {
+		if !test_env::Env::default()
+			.set("MOQ_LISTEN_TLS_CERT", "moq.pem")
+			.set("MOQ_HLS_TLS_CERT", "hls.pem")
+			.set("MOQ_HLS_TLS_KEY", "hls.key")
+			.set("MOQ_HLS_TLS_ROOT", "root.pem")
+			.isolate()
+		{
+			return;
+		}
+		let cli = Invocation::try_parse_from(["moq", "export", "hls"]).unwrap();
+		assert_eq!(cli.moq.server.tls.cert, [std::path::PathBuf::from("moq.pem")]);
+		let Command::Export(export) = &cli.stages[0] else {
+			panic!("expected export")
+		};
+		let ExportSink::Hls(hls) = &export.sink else {
+			panic!("expected HLS")
+		};
+		assert_eq!(hls.tls.cert, [std::path::PathBuf::from("hls.pem")]);
+		assert_eq!(hls.tls.key, [std::path::PathBuf::from("hls.key")]);
+		assert_eq!(hls.tls.root, [std::path::PathBuf::from("root.pem")]);
+	}
+
+	macro_rules! released_environment {
+		($test:ident, $old:literal, $replacement:literal, $value:literal) => {
+			#[test]
+			fn $test() {
+				if !test_env::Env::default().set($old, $value).isolate() {
+					return;
+				}
+				let err = Invocation::try_parse_from(["moq", "import", "ts"])
+					.err()
+					.expect("released environment must not be ignored");
+				assert!(
+					err.to_string().contains($old) && err.to_string().contains($replacement),
+					"{err}"
+				);
+			}
+		};
+	}
+
+	released_environment!(
+		released_backoff_initial_env,
+		"MOQ_BACKOFF_INITIAL",
+		"MOQ_CONNECT_BACKOFF_INITIAL",
+		"2s"
+	);
+	released_environment!(
+		released_backoff_multiplier_env,
+		"MOQ_BACKOFF_MULTIPLIER",
+		"MOQ_CONNECT_BACKOFF_MULTIPLIER",
+		"3"
+	);
+	released_environment!(
+		released_backoff_max_env,
+		"MOQ_BACKOFF_MAX",
+		"MOQ_CONNECT_BACKOFF_MAX",
+		"2s"
+	);
+	released_environment!(
+		released_backoff_timeout_env,
+		"MOQ_BACKOFF_TIMEOUT",
+		"MOQ_CONNECT_BACKOFF_TIMEOUT",
+		"2s"
+	);
+	released_environment!(
+		released_goaway_redirect_env,
+		"MOQ_GOAWAY_REDIRECT",
+		"MOQ_CONNECT_GOAWAY_REDIRECT",
+		"ignore"
+	);
+	released_environment!(
+		released_goaway_handover_env,
+		"MOQ_GOAWAY_HANDOVER",
+		"MOQ_CONNECT_GOAWAY_HANDOVER",
+		"2s"
+	);
+
+	#[test]
+	fn connection_prefixes_refuse_released_spellings() {
+		for (family, names) in [
+			("backoff", &["initial", "multiplier", "max", "timeout"][..]),
+			("goaway", &["redirect", "handover"][..]),
+		] {
+			for name in names {
+				let value = match *name {
+					"multiplier" => "3",
+					"redirect" => "ignore",
+					_ => "2s",
+				};
+				let current = format!("--connect-{family}-{name}");
+				Invocation::try_parse_from(["moq", current.as_str(), value, "import", "ts"]).unwrap();
+				let old = format!("--{family}-{name}");
+				let err = Invocation::try_parse_from(["moq", old.as_str(), value, "import", "ts"])
+					.err()
+					.expect("released spelling refused");
+				assert!(
+					err.to_string().contains(&old) && err.to_string().contains(&current),
+					"{err}"
+				);
+			}
+		}
+	}
+
+	#[test]
+	fn audited_flags_parse_the_consumer_policy() {
+		for format in ["fmp4", "hls"] {
+			let mut argv = vec!["moq", "import", format];
+			if format == "hls" {
+				argv.push("input.m3u8");
+			}
+			argv.push("--cmaf");
+			let cli = Invocation::try_parse_from(argv).expect("explicit CMAF publishing");
+			let Command::Import(import) = &cli.stages[0] else {
+				panic!("expected import")
+			};
+			match &import.source {
+				ImportSource::Fmp4(args) => assert!(args.passthrough),
+				ImportSource::Hls(args) => assert!(args.passthrough),
+				_ => panic!("expected fragment input"),
+			}
+		}
+		Invocation::try_parse_from([
+			"moq",
+			"export",
+			"rtmp",
+			"--connect",
+			"rtmp://localhost/live/key",
+			"--delay",
+			"2s",
+		])
+		.expect("RTMP paced release");
+	}
+
+	#[test]
+	fn unpublished_spellings_have_no_migration_shim() {
+		for format in ["fmp4", "hls"] {
+			let mut argv = vec!["moq", "import", format];
+			if format == "hls" {
+				argv.push("input.m3u8");
+			}
+			argv.push("--passthrough");
+			let err = Invocation::try_parse_from(argv).err().expect("CMAF is explicit");
+			assert_eq!(err.kind(), ParseErrorKind::UnknownArgument);
+		}
+		for format in ["flv", "mkv", "rtmp"] {
+			let mut argv = vec!["moq", "export", format, "--max-delay", "2s"];
+			if format == "rtmp" {
+				argv.extend(["--connect", "rtmp://localhost/live/key"]);
+			}
+			let err = Invocation::try_parse_from(argv)
+				.err()
+				.expect("unreleased spelling is unknown");
+			assert_eq!(err.kind(), ParseErrorKind::UnknownArgument);
+		}
+	}
+
 	/// `export ts --passthrough` writes the source's own packets, so it neither switches
 	/// programs, waits for a return, nor pads to a rate.
 	#[test]
@@ -1670,7 +1824,7 @@ mod tests {
 
 	/// A stage carries config of its own, and the check has to reach it.
 	///
-	/// `export hls` flattens `tls::Listen`. Its listener decides whether to serve
+	/// `export hls` adapts its TLS arguments. Its listener decides whether to serve
 	/// TLS at all from the canonical `cert`/`generate` fields, so a released
 	/// `--tls-cert` left it serving plaintext HTTP without ever reaching the builder
 	/// that refuses: the certificate and the mTLS roots both silently gone.
@@ -2163,7 +2317,7 @@ mod tests {
 		for format in ["flv", "mkv"] {
 			assert_eq!(stdout(&[format]), Duration::from_millis(500));
 			assert_eq!(stdout(&[format, "--delay", "2s"]), Duration::from_secs(2));
-			for old in ["--max-delay", "--max-age", "--latency-max"] {
+			for old in ["--max-age", "--latency-max"] {
 				let Err(err) = export(&[format, old, "2s"]) else {
 					panic!("{old} must be refused")
 				};
@@ -2178,18 +2332,18 @@ mod tests {
 		assert!(export(&["ts", "--max-delay", "2s"]).is_err(), "ts spells it --delay");
 
 		let rtmp = |flag: &str| export(&["rtmp", "--connect", "rtmp://example.com/live/key", flag, "2s"]);
-		let cli = rtmp("--max-delay").unwrap();
+		let cli = rtmp("--delay").unwrap();
 		let Command::Export(export_rtmp) = &cli.stages[0] else {
 			panic!("expected export")
 		};
 		let ExportSink::Rtmp(args) = &export_rtmp.sink else {
 			panic!("expected rtmp")
 		};
-		assert_eq!(Duration::from(args.max_delay), Duration::from_secs(2));
+		assert_eq!(Duration::from(args.delay), Duration::from_secs(2));
 		let Err(err) = rtmp("--max-age") else {
 			panic!("rtmp: --max-age must not start a run");
 		};
-		assert!(err.to_string().contains("--max-age -> --max-delay"), "rtmp: {err}");
+		assert!(err.to_string().contains("--max-age -> --delay"), "rtmp: {err}");
 
 		assert!(
 			Invocation::try_parse_from(["moq", "import", "--max-delay", "5s", "ts"]).is_err(),
@@ -2330,7 +2484,7 @@ mod tests {
 		let dial: &[&[&str]] = &[
 			&["--connect", "https://relay.example"],
 			&["--connect-tls-insecure"],
-			&["--backoff-initial", "2s"],
+			&["--connect-backoff-initial", "2s"],
 			&["--quic-idle-timeout", "10s"],
 			#[cfg(feature = "iroh")]
 			&["--iroh-enabled"],
@@ -2676,6 +2830,27 @@ mod tests {
 		assert!(play.validate().is_ok());
 	}
 
+	#[test]
+	fn flag_help_uses_plain_names() {
+		for command in [
+			vec![],
+			vec!["import", "fmp4"],
+			vec!["import", "hls"],
+			vec!["export", "rtmp"],
+			vec!["export", "hls"],
+			vec!["auth", "serve"],
+		] {
+			let mut argv = vec!["moq"];
+			argv.extend(command);
+			argv.push("--help");
+			let help = Invocation::try_parse_from(argv)
+				.err()
+				.expect("help request")
+				.to_string();
+			assert!(!help.contains("[`"), "{help}");
+		}
+	}
+
 	/// Help and version are answered with their actual page, not an empty string.
 	///
 	/// Usage renders those variants as nothing through `render_failure`, because the
diff --git a/rs/moq-cli/src/auth.rs b/rs/moq-cli/src/auth.rs
index 67acd137418..625c79aa1d3 100644
--- a/rs/moq-cli/src/auth.rs
+++ b/rs/moq-cli/src/auth.rs
@@ -370,15 +370,15 @@ pub struct Serve {
 	listen_public: bool,
 
 	/// The key file a `jwt` is verified against.
-	#[usage(long, conflicts = "--key-dir", value_hint = usage_rs::ValueHint::FilePath, extensions("jwk", "json"))]
+	#[usage(long, conflicts("--key-dir", "--key-set"), value_hint = usage_rs::ValueHint::FilePath, extensions("jwk", "json"))]
 	key: Option<PathBuf>,
 
 	/// A directory of `{kid}.jwk` files, selected by the token's `kid`.
-	#[usage(long, conflicts = "--key", value_hint = usage_rs::ValueHint::DirPath)]
+	#[usage(long, conflicts("--key", "--key-set"), value_hint = usage_rs::ValueHint::DirPath)]
 	key_dir: Option<PathBuf>,
 
 	/// A JWK Set file, selecting one verification key by the token's `kid`.
-	#[usage(long, value_hint = usage_rs::ValueHint::FilePath, extensions("json", "jwks"))]
+	#[usage(long, conflicts("--key", "--key-dir"), value_hint = usage_rs::ValueHint::FilePath, extensions("json", "jwks"))]
 	key_set: Option<PathBuf>,
 
 	/// Patterns an anonymous session may publish, rooted at `/` (repeatable); `foo/**` for a subtree.
@@ -476,9 +476,6 @@ impl Serve {
 				subscribe.iter().cloned().collect::<moq_auth::Patterns>(),
 			)
 		};
-		if self.key_set.is_some() && (self.key.is_some() || self.key_dir.is_some()) {
-			anyhow::bail!("--key-set conflicts with --key and --key-dir");
-		}
 		let mut policy = Policy::default();
 		policy.keys = match (&self.key, &self.key_dir, &self.key_set) {
 			(Some(key), _, _) => Some(Keys::File(key.clone())),
@@ -643,6 +640,25 @@ mod tests {
 		}
 	}
 
+	#[test]
+	fn verification_key_sources_conflict_in_the_parser() {
+		let flags = ["--key", "--key-dir", "--key-set"];
+		for first in flags {
+			assert!(parse(&["moq", "auth", "serve", first, "unused"]).is_ok());
+			for second in flags {
+				if first == second {
+					continue;
+				}
+				let err = parse(&["moq", "auth", "serve", first, "unused", second, "unused"])
+					.expect_err("key sources must conflict before loading files");
+				assert!(
+					err.to_string().contains(first) && err.to_string().contains(second),
+					"{err}"
+				);
+			}
+		}
+	}
+
 	#[test]
 	fn serve_refuses_a_public_bind_without_saying_so() {
 		let err = serve(&["moq", "auth", "serve", "--listen", "0.0.0.0:4440"])
@@ -715,11 +731,7 @@ mod tests {
 			.policy()
 			.unwrap();
 		assert!(matches!(&set.keys, Some(Keys::Set(path)) if path == std::path::Path::new("/keys.jwks")));
-		assert!(
-			serve(&["moq", "auth", "serve", "--key", "/key.jwk", "--key-set", "/keys.jwks"])
-				.policy()
-				.is_err()
-		);
+		assert!(parse(&["moq", "auth", "serve", "--key", "/key.jwk", "--key-set", "/keys.jwks"]).is_err());
 
 		// Nothing configured is a server that refuses everyone, and like 0.14, never
 		// re-checks or closes a session on its own.
diff --git a/rs/moq-cli/src/hls.rs b/rs/moq-cli/src/hls.rs
index 17055f57b17..4690c27fc43 100644
--- a/rs/moq-cli/src/hls.rs
+++ b/rs/moq-cli/src/hls.rs
@@ -22,7 +22,7 @@ pub struct ImportArgs {
 
 	/// Publish each fMP4 fragment whole as CMAF instead of splitting it into one frame per
 	/// sample. Keeps the source's exact sample timing, but every player has to parse ISO BMFF.
-	#[usage(long)]
+	#[usage(long = "cmaf")]
 	pub passthrough: bool,
 }
 
@@ -36,7 +36,7 @@ pub struct ExportArgs {
 
 	/// TLS certificates, keys, self-signed generation, and optional mTLS roots.
 	#[usage(flatten)]
-	pub tls: moq_tokio::tls::Listen,
+	pub tls: Tls,
 
 	/// Minimum media listed in each rendition's playlist window, at most 256 segments.
 	/// Keep it within the relay's group-cache retention, since segments are fetched from
@@ -49,6 +49,106 @@ pub struct ExportArgs {
 	pub cors: crate::web::Cors,
 }
 
+/// TLS for the HLS HTTP listener, independent of the MoQ listener's environment.
+#[derive(usage_rs::Args, Clone)]
+#[usage(unknown_flags = "error", args_override_self = false)]
+pub struct Tls {
+	/// Load the HLS listener's PEM certificate.
+	#[usage(long = "listen-tls-cert", env = "MOQ_HLS_TLS_CERT", value_hint = usage_rs::ValueHint::FilePath)]
+	pub cert: Vec<PathBuf>,
+	/// Load the HLS listener's PEM private key.
+	#[usage(long = "listen-tls-key", env = "MOQ_HLS_TLS_KEY", value_hint = usage_rs::ValueHint::FilePath)]
+	pub key: Vec<PathBuf>,
+	/// Generate the HLS listener's self-signed certificate for these hostnames.
+	#[usage(long = "listen-tls-generate", env = "MOQ_HLS_TLS_GENERATE", delimiter = ',')]
+	pub generate: Vec<String>,
+	/// Trust these PEM roots for optional HLS client certificates.
+	#[usage(long = "listen-tls-root", env = "MOQ_HLS_TLS_ROOT", delimiter = ',', value_hint = usage_rs::ValueHint::FilePath)]
+	pub root: Vec<PathBuf>,
+	#[usage(
+		name = "hls-legacy-cert",
+		long = "tls-cert",
+		alias = "server-tls-cert",
+		env = "MOQ_SERVER_TLS_CERT",
+		hide = true
+	)]
+	legacy_cert: Vec<PathBuf>,
+
+	#[usage(
+		name = "hls-legacy-key",
+		long = "tls-key",
+		alias = "server-tls-key",
+		env = "MOQ_SERVER_TLS_KEY",
+		hide = true
+	)]
+	legacy_key: Vec<PathBuf>,
+
+	#[usage(
+		name = "hls-legacy-generate",
+		long = "tls-generate",
+		alias = "server-tls-generate",
+		env = "MOQ_SERVER_TLS_GENERATE",
+		delimiter = ',',
+		hide = true
+	)]
+	legacy_generate: Vec<String>,
+
+	#[usage(
+		name = "hls-legacy-root",
+		long = "server-tls-root",
+		env = "MOQ_SERVER_TLS_ROOT",
+		delimiter = ',',
+		hide = true
+	)]
+	legacy_root: Vec<PathBuf>,
+}
+
+impl Tls {
+	pub(super) fn deprecated(&self) -> moq_tokio::cli::Deprecated {
+		let mut found = moq_tokio::cli::Deprecated::default();
+		if !self.legacy_cert.is_empty() {
+			found.flag(
+				"--tls-cert",
+				Some("MOQ_SERVER_TLS_CERT"),
+				"--listen-tls-cert / MOQ_HLS_TLS_CERT",
+			);
+		}
+		if !self.legacy_key.is_empty() {
+			found.flag(
+				"--tls-key",
+				Some("MOQ_SERVER_TLS_KEY"),
+				"--listen-tls-key / MOQ_HLS_TLS_KEY",
+			);
+		}
+		if !self.legacy_generate.is_empty() {
+			found.flag(
+				"--tls-generate",
+				Some("MOQ_SERVER_TLS_GENERATE"),
+				"--listen-tls-generate / MOQ_HLS_TLS_GENERATE",
+			);
+		}
+		if !self.legacy_root.is_empty() {
+			found.flag(
+				"--server-tls-root",
+				Some("MOQ_SERVER_TLS_ROOT"),
+				"--listen-tls-root / MOQ_HLS_TLS_ROOT",
+			);
+		}
+		found
+	}
+
+	fn into_config(self) -> anyhow::Result<moq_tokio::tls::Listen> {
+		let deprecated = self.deprecated();
+		anyhow::ensure!(deprecated.is_empty(), "{deprecated}");
+		let mut config = moq_tokio::tls::Listen::default();
+		config.cert = self.cert;
+		config.key = self.key;
+		config.generate = self.generate;
+		config.root = self.root;
+		Ok(config)
+	}
+}
+
 /// Pull a remote HLS/LL-HLS playlist (URL or file path) into the Origin under `target.name`,
 /// announced under `epoch`. Once the playlist ends, serve the finished broadcast until its
 /// media is stale.
@@ -159,11 +259,12 @@ pub async fn export(origin: moq_net::origin::Consumer, args: ExportArgs, name: S
 	let server = moq_hls::Server::new(scoped, config);
 	let app = server.router().layer(args.cors.layer([Method::GET])?);
 
-	let tls = if args.tls.cert.is_empty() && args.tls.generate.is_empty() {
+	let tls = args.tls.into_config()?;
+	let tls = if tls.cert.is_empty() && tls.generate.is_empty() {
 		None
 	} else {
 		let alpn = vec![b"h2".to_vec(), b"http/1.1".to_vec()];
-		Some(args.tls.server_config(alpn)?)
+		Some(tls.server_config(alpn)?)
 	};
 
 	let listener = moq_tokio::bind::tcp(args.listen)?;
diff --git a/rs/moq-cli/src/main.rs b/rs/moq-cli/src/main.rs
index 2c8b934ab1d..f62a8158034 100644
--- a/rs/moq-cli/src/main.rs
+++ b/rs/moq-cli/src/main.rs
@@ -840,7 +840,7 @@ fn spawn_export(
 				tasks.spawn(hls::export(origin.consume(), args, name));
 			}
 			ExportSink::Rtmp(rtmp) => {
-				let max_delay = rtmp.max_delay.into();
+				let max_delay = rtmp.delay.into();
 				if let Some(addr) = rtmp.endpoint.listen {
 					let name = require_broadcast(name, "export rtmp --listen")?;
 					tasks.spawn(rtmp::listen_export(origin.consume(), addr, name, max_delay));
diff --git a/rs/moq-cli/src/rtmp.rs b/rs/moq-cli/src/rtmp.rs
index d1655edd89d..47096634c1f 100644
--- a/rs/moq-cli/src/rtmp.rs
+++ b/rs/moq-cli/src/rtmp.rs
@@ -40,13 +40,13 @@ pub struct ExportArgs {
 	/// FLV frame release delay and stalled-group budget. Late video resumes at
 	/// its next keyframe; zero releases available frames without deadline drops.
 	#[usage(long, default = "500ms")]
-	pub max_delay: moq_tokio::cli::Duration,
+	pub delay: moq_tokio::cli::Duration,
 
-	/// The released spelling of [`Self::max_delay`].
+	/// The released spelling of `--delay`.
 	#[usage(long = "max-age", hide = true)]
 	max_age: Option<moq_tokio::cli::Duration>,
 
-	/// The released spelling of [`Self::max_delay`], before `--max-age`.
+	/// The released spelling of `--delay`, before `--max-age`.
 	#[usage(long = "latency-max", hide = true)]
 	latency_max: Option<moq_tokio::cli::Duration>,
 }
@@ -55,10 +55,10 @@ impl ExportArgs {
 	pub(crate) fn deprecated(&self) -> moq_tokio::cli::Deprecated {
 		let mut found = moq_tokio::cli::Deprecated::default();
 		if self.max_age.is_some() {
-			found.flag("--max-age", None, "--max-delay");
+			found.flag("--max-age", None, "--delay");
 		}
 		if self.latency_max.is_some() {
-			found.flag("--latency-max", None, "--max-delay");
+			found.flag("--latency-max", None, "--delay");
 		}
 		found
 	}
diff --git a/rs/moq-gst/src/connection.rs b/rs/moq-gst/src/connection.rs
new file mode 100644
index 00000000000..f612fbdccb8
--- /dev/null
+++ b/rs/moq-gst/src/connection.rs
@@ -0,0 +1,163 @@
+//! Connection properties shared by publishers and subscribers.
+
+use gst::glib;
+use gst::prelude::*;
+use std::time::Duration;
+
+#[derive(Debug, Clone, Default)]
+pub(crate) struct Settings {
+	version: Option<String>,
+	tls_disable_verify: bool,
+	quic_idle_timeout: Option<Duration>,
+	quic_keep_alive: Option<Duration>,
+	quic_keylog: bool,
+}
+
+pub(crate) fn duration_millis(duration: Duration) -> u64 {
+	duration.as_millis().try_into().unwrap_or(u64::MAX)
+}
+
+pub(crate) fn properties() -> Vec<glib::ParamSpec> {
+	let quic = moq_tokio::quic::Resolved::default();
+	vec![
+		glib::ParamSpecString::builder("connect-version")
+			.nick("Connect version")
+			.blurb(
+				"Comma-separated protocol versions to offer, most preferred first; unset offers every stable version",
+			)
+			.mutable_ready()
+			.build(),
+		glib::ParamSpecBoolean::builder("tls-disable-verify")
+			.nick("TLS disable verify")
+			.blurb("Disable TLS certificate verification")
+			.mutable_ready()
+			.build(),
+		glib::ParamSpecUInt64::builder("quic-idle-timeout")
+			.nick("QUIC idle timeout")
+			.blurb("QUIC idle timeout in milliseconds, 0 to disable locally")
+			.default_value(duration_millis(quic.idle_timeout))
+			.mutable_ready()
+			.build(),
+		glib::ParamSpecUInt64::builder("quic-keep-alive")
+			.nick("QUIC keep-alive")
+			.blurb("QUIC keep-alive interval in milliseconds, 0 to disable; iroh still pings at least every 5s")
+			.default_value(quic.keep_alive.map(duration_millis).unwrap_or(0))
+			.mutable_ready()
+			.build(),
+		glib::ParamSpecBoolean::builder("quic-keylog")
+			.nick("QUIC TLS key log")
+			.blurb("Danger: write TLS session secrets to the file SSLKEYLOGFILE names; without it the start fails")
+			.mutable_ready()
+			.build(),
+	]
+}
+
+impl Settings {
+	pub(crate) fn set(&mut self, name: &str, value: &glib::Value) {
+		match name {
+			"connect-version" => self.version = value.get().unwrap(),
+			"tls-disable-verify" => self.tls_disable_verify = value.get().unwrap(),
+			"quic-idle-timeout" => self.quic_idle_timeout = Some(Duration::from_millis(value.get().unwrap())),
+			"quic-keep-alive" => self.quic_keep_alive = Some(Duration::from_millis(value.get().unwrap())),
+			"quic-keylog" => self.quic_keylog = value.get().unwrap(),
+			_ => unreachable!(),
+		}
+	}
+
+	pub(crate) fn get(&self, name: &str) -> glib::Value {
+		match name {
+			"connect-version" => self.version.to_value(),
+			"tls-disable-verify" => self.tls_disable_verify.to_value(),
+			"quic-idle-timeout" => duration_millis(
+				self.quic_idle_timeout
+					.unwrap_or_else(|| moq_tokio::quic::Resolved::default().idle_timeout),
+			)
+			.to_value(),
+			"quic-keep-alive" => self
+				.quic_keep_alive
+				.or_else(|| moq_tokio::quic::Resolved::default().keep_alive)
+				.map(duration_millis)
+				.unwrap_or(0)
+				.to_value(),
+			"quic-keylog" => self.quic_keylog.to_value(),
+			_ => unreachable!(),
+		}
+	}
+
+	pub(crate) fn resolve(&self) -> anyhow::Result<(moq_tokio::connect::Config, moq_tokio::quic::Config)> {
+		let mut connect = moq_tokio::connect::Config::default();
+		connect.version = self
+			.version
+			.iter()
+			.flat_map(|versions| versions.split(','))
+			.map(str::trim)
+			.filter(|version| !version.is_empty())
+			.map(|version| version.parse().map_err(anyhow::Error::msg))
+			.collect::<anyhow::Result<_>>()?;
+		connect.tls.insecure = Some(self.tls_disable_verify);
+		connect.backoff.timeout = Some(Duration::ZERO);
+		let mut quic = moq_tokio::quic::Config::default();
+		quic.idle_timeout = self.quic_idle_timeout;
+		quic.keep_alive = self.quic_keep_alive;
+		quic.keylog = Some(self.quic_keylog);
+		Ok((connect, quic))
+	}
+}
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+
+	#[test]
+	fn both_elements_parse_the_same_connection_properties() {
+		gst::init().unwrap();
+		let elements: [gst::Element; 2] = [
+			glib::Object::new::<crate::sink::MoqSink>().upcast(),
+			glib::Object::new::<crate::source::MoqSrc>().upcast(),
+		];
+		for element in elements {
+			assert!(element.find_property("versions").is_none());
+			for prop in properties() {
+				let actual = element.find_property(prop.name()).unwrap();
+				assert_eq!(actual.value_type(), prop.value_type());
+				assert!(actual.flags().contains(gst::PARAM_FLAG_MUTABLE_READY));
+			}
+			for (name, value) in [("quic-idle-timeout", 15_000u64), ("quic-keep-alive", 0u64)] {
+				element.set_property(name, value);
+				assert_eq!(element.property::<u64>(name), value);
+			}
+			element.set_property("connect-version", " moq-transport-14 , moq-lite-06 ");
+			assert_eq!(
+				element.property::<String>("connect-version"),
+				" moq-transport-14 , moq-lite-06 "
+			);
+		}
+	}
+
+	#[test]
+	fn properties_reach_versions_tls_and_quic_config() {
+		let mut settings = Settings::default();
+		for (name, value) in [
+			("connect-version", " moq-transport-14 , moq-lite-06 ".to_value()),
+			("tls-disable-verify", true.to_value()),
+			("quic-idle-timeout", 15_000u64.to_value()),
+			("quic-keep-alive", 0u64.to_value()),
+			("quic-keylog", true.to_value()),
+		] {
+			settings.set(name, &value);
+		}
+		let (connect, quic) = settings.resolve().unwrap();
+		assert_eq!(
+			connect.version,
+			["moq-transport-14".parse().unwrap(), "moq-lite-06".parse().unwrap()]
+		);
+		assert_eq!(connect.tls.insecure, Some(true));
+		assert_eq!(connect.backoff.timeout, Some(Duration::ZERO));
+		let quic = quic.resolve();
+		assert_eq!(quic.idle_timeout, Duration::from_secs(15));
+		assert_eq!(quic.keep_alive, None);
+		assert!(quic.keylog);
+		settings.set("connect-version", &"moq-lite-99".to_value());
+		assert!(settings.resolve().unwrap_err().to_string().contains("moq-lite-99"));
+	}
+}
diff --git a/rs/moq-gst/src/lib.rs b/rs/moq-gst/src/lib.rs
index 547187b7adb..a44b7082595 100644
--- a/rs/moq-gst/src/lib.rs
+++ b/rs/moq-gst/src/lib.rs
@@ -4,6 +4,7 @@
 use gst::glib;
 
 mod block;
+mod connection;
 mod sink;
 mod source;
 
diff --git a/rs/moq-gst/src/sink/imp.rs b/rs/moq-gst/src/sink/imp.rs
index 7e876a8436d..784a3a7a57c 100644
--- a/rs/moq-gst/src/sink/imp.rs
+++ b/rs/moq-gst/src/sink/imp.rs
@@ -11,6 +11,7 @@
 
 use std::collections::HashSet;
 use std::sync::{LazyLock, Mutex};
+#[cfg(test)]
 use std::time::Duration;
 
 use anyhow::{Context, Result};
@@ -31,21 +32,14 @@ use super::session::{
 struct Settings {
 	url: Option<String>,
 	broadcast: Option<String>,
-	tls_disable_verify: bool,
-	quic_idle_timeout: Option<Duration>,
-	quic_keep_alive: Option<Duration>,
-	quic_keylog: bool,
-	linger: bool,
-}
-
-fn duration_millis(duration: Duration) -> u64 {
-	duration.as_millis().try_into().unwrap_or(u64::MAX)
+	connection: crate::connection::Settings,
 }
 
 impl TryFrom<Settings> for ResolvedSettings {
 	type Error = anyhow::Error;
 
 	fn try_from(value: Settings) -> Result<Self> {
+		let (connect, quic) = value.connection.resolve()?;
 		Ok(Self {
 			url: url::Url::parse(value.url.as_ref().context("url property is required")?)?,
 			broadcast: value
@@ -53,10 +47,8 @@ impl TryFrom<Settings> for ResolvedSettings {
 				.as_ref()
 				.context("broadcast property is required")?
 				.clone(),
-			tls_disable_verify: value.tls_disable_verify,
-			quic_idle_timeout: value.quic_idle_timeout,
-			quic_keep_alive: value.quic_keep_alive,
-			quic_keylog: value.quic_keylog,
+			connect,
+			quic,
 		})
 	}
 }
@@ -67,10 +59,6 @@ struct State {
 	broadcast: moq_net::broadcast::Producer,
 	catalog: Option<moq_mux::catalog::Producer>,
 	wall: std::sync::Arc<super::wall::Wall>,
-	/// Whether a clean end keeps serving the finished broadcast until its media is stale.
-	linger: bool,
-	/// Whether the finished broadcast stopped serving its tail, so the EOS it owes may go out.
-	drained: bool,
 	/// Whether this publication's single EOS message already went out. One per publication, not per
 	/// entry to PLAYING: a finished publication is terminal here, so there is nothing to announce twice.
 	eos_delivered: bool,
@@ -97,8 +85,8 @@ struct Control {
 
 /// Claim this publication's EOS, if it is owed and may go out now.
 ///
-/// One claim per publication, so whichever arrives first (the transition into PLAYING, or the end of
-/// the linger) is the only one that posts. Called under `control`, which is what serializes the two.
+/// One claim per publication, so whichever arrives first (the transition into PLAYING, or the pad that
+/// ended it) is the only one that posts. Called under `control`, which is what serializes the two.
 fn claim_eos(control: &mut Control) -> Posted {
 	if !control.playing {
 		return Posted::Nothing;
@@ -106,8 +94,7 @@ fn claim_eos(control: &mut Control) -> Posted {
 	let Some(state) = control.live.as_mut() else {
 		return Posted::Nothing;
 	};
-	// Only a clean finish drains, so a drained publication ended with EOS.
-	if state.eos_delivered || !state.drained {
+	if state.eos_delivered || state.session.completion().get() != Completion::Eos {
 		return Posted::Nothing;
 	}
 	state.eos_delivered = true;
@@ -138,7 +125,7 @@ enum Posted {
 
 /// The element-wide outcome a publishing session earned.
 enum MessageKind {
-	/// Every pad ended and the finished broadcast closed.
+	/// Every pad ended and the broadcast closed.
 	Eos,
 	/// A producer failed to close, already formatted for the bus.
 	FinalizeError(String),
@@ -177,7 +164,6 @@ impl ObjectSubclass for MoqSink {
 impl ObjectImpl for MoqSink {
 	fn properties() -> &'static [glib::ParamSpec] {
 		static PROPS: LazyLock<Vec<glib::ParamSpec>> = LazyLock::new(|| {
-			let quic = moq_tokio::quic::Resolved::default();
 			vec![
 				glib::ParamSpecString::builder("url")
 					.nick("Destination URL")
@@ -189,38 +175,6 @@ impl ObjectImpl for MoqSink {
 					.blurb("The name of the broadcast to publish")
 					.mutable_ready()
 					.build(),
-				glib::ParamSpecBoolean::builder("tls-disable-verify")
-					.nick("TLS disable verify")
-					.blurb("Disable TLS verification")
-					.default_value(false)
-					.mutable_ready()
-					.build(),
-				glib::ParamSpecUInt64::builder("quic-idle-timeout")
-					.nick("QUIC idle timeout")
-					.blurb("QUIC idle timeout in milliseconds, 0 to disable locally")
-					.default_value(duration_millis(quic.idle_timeout))
-					.mutable_ready()
-					.build(),
-				glib::ParamSpecUInt64::builder("quic-keep-alive")
-					.nick("QUIC keep-alive")
-					.blurb("QUIC keep-alive interval in milliseconds, 0 to disable; iroh still pings at least every 5s")
-					.default_value(quic.keep_alive.map(duration_millis).unwrap_or(0))
-					.mutable_ready()
-					.build(),
-				glib::ParamSpecBoolean::builder("quic-keylog")
-					.nick("QUIC TLS key log")
-					.blurb(
-						"Danger: write TLS session secrets to the file SSLKEYLOGFILE names; without it the start fails",
-					)
-					.default_value(false)
-					.mutable_ready()
-					.build(),
-				glib::ParamSpecBoolean::builder("linger")
-					.nick("Linger")
-					.blurb("At end of stream, serve the finished broadcast until its media is stale, then post EOS")
-					.default_value(false)
-					.mutable_ready()
-					.build(),
 				// Read-only, served from the live session's status. Each notifies on change.
 				glib::ParamSpecEnum::builder::<ConnectionStatus>("status")
 					.nick("Connection status")
@@ -258,6 +212,9 @@ impl ObjectImpl for MoqSink {
 					.read_only()
 					.build(),
 			]
+			.into_iter()
+			.chain(crate::connection::properties())
+			.collect()
 		});
 		PROPS.as_ref()
 	}
@@ -285,12 +242,7 @@ impl ObjectImpl for MoqSink {
 		match pspec.name() {
 			"url" => settings.url = value.get().unwrap(),
 			"broadcast" => settings.broadcast = value.get().unwrap(),
-			"tls-disable-verify" => settings.tls_disable_verify = value.get().unwrap(),
-			"quic-idle-timeout" => settings.quic_idle_timeout = Some(Duration::from_millis(value.get().unwrap())),
-			"quic-keep-alive" => settings.quic_keep_alive = Some(Duration::from_millis(value.get().unwrap())),
-			"quic-keylog" => settings.quic_keylog = value.get().unwrap(),
-			"linger" => settings.linger = value.get().unwrap(),
-			_ => unreachable!(),
+			name => settings.connection.set(name, value),
 		}
 	}
 
@@ -321,22 +273,7 @@ impl ObjectImpl for MoqSink {
 				match name {
 					"url" => settings.url.to_value(),
 					"broadcast" => settings.broadcast.to_value(),
-					"tls-disable-verify" => settings.tls_disable_verify.to_value(),
-					"quic-idle-timeout" => duration_millis(
-						settings
-							.quic_idle_timeout
-							.unwrap_or_else(|| moq_tokio::quic::Resolved::default().idle_timeout),
-					)
-					.to_value(),
-					"quic-keep-alive" => settings
-						.quic_keep_alive
-						.or_else(|| moq_tokio::quic::Resolved::default().keep_alive)
-						.map(duration_millis)
-						.unwrap_or(0)
-						.to_value(),
-					"quic-keylog" => settings.quic_keylog.to_value(),
-					"linger" => settings.linger.to_value(),
-					_ => unreachable!(),
+					name => settings.connection.get(name),
 				}
 			}
 		}
@@ -600,7 +537,6 @@ impl MoqSink {
 	/// Create the session and producers before any buffer flows.
 	fn start_session(&self) -> Result<SessionRegistration, gst::StateChangeError> {
 		let settings = self.settings.lock().unwrap().clone();
-		let linger = settings.linger;
 		let settings = ResolvedSettings::try_from(settings).map_err(|err| {
 			gst::error!(CAT, obj = self.obj(), "invalid settings: {err:#}");
 			gst::StateChangeError
@@ -619,8 +555,6 @@ impl MoqSink {
 				broadcast,
 				catalog: Some(catalog),
 				wall,
-				linger,
-				drained: false,
 				eos_delivered: false,
 			});
 			let pads = self.pads_locked(&control);
@@ -701,7 +635,7 @@ impl MoqSink {
 		if let Some(err) = failure {
 			gst::warning!(CAT, "finalize on stop: {err:?}");
 		}
-		// Close the broadcast before reaping the session task. This also ends a linger early.
+		// Close the broadcast before reaping the session task.
 		state.broadcast.close();
 		drop(closed);
 		state.session.stop();
@@ -1082,59 +1016,22 @@ impl MoqSink {
 				false => Posted::Nothing,
 			},
 			None if completion.finish() => {
-				// Taken while the producers are alive, so it holds the tracks they finished.
-				let finish = state.broadcast.clone().finish();
+				state.broadcast.close();
 				drop((closed, catalog));
-				self.linger_locked(control, finish)
+				claim_eos(control)
 			}
 			None => Posted::Nothing,
 		};
 		Finished { updates, message }
 	}
 
-	/// With `linger` set, serve the finished broadcast so subscribers still catching up drain its tail
-	/// until the media is stale, and owe the EOS once it closes. Otherwise, or with nothing to drain,
-	/// close it and owe the EOS now.
-	///
-	/// The EOS waits for the linger because `gst-launch` exits on it, and NULL closes the broadcast.
-	/// Stopping the pipeline is the interrupt: it closes the broadcast, which ends the wait.
-	fn linger_locked(&self, control: &mut Control, finish: moq_net::broadcast::Finish) -> Posted {
-		let state = control.live.as_mut().expect("a finished publication is live");
-		let linger = finish.linger();
-		if !state.linger || linger.is_zero() {
-			drop(finish);
-			state.drained = true;
-			return claim_eos(control);
-		}
-		gst::info!(
-			CAT,
-			obj = self.obj(),
-			"all pads ended, serving the broadcast for {linger:?}"
-		);
-		RUNTIME.spawn(linger_task(finish, self.obj().downgrade(), state.session.completion()));
-		Posted::Nothing
-	}
-
-	/// Owe the EOS now that the finished broadcast closed, unless a newer run replaced `session`.
-	fn drained(&self, session: &CompletionHandle) {
-		let message = {
-			let mut control = self.control.lock().unwrap();
-			match control.live.as_mut() {
-				Some(state) if std::sync::Arc::ptr_eq(&state.session.completion(), session) => state.drained = true,
-				_ => return,
-			}
-			claim_eos(&mut control)
-		};
-		self.post_current(message);
-	}
-
 	fn notify_updates(&self, updates: Vec<PadUpdate>) {
 		for update in updates {
 			update.pad.notify_changes(update.changes);
 		}
 	}
 
-	/// Open the EOS gate for this run and post the message if the finished broadcast already closed.
+	/// Open the EOS gate for this run and post the message if the publication already ended.
 	fn enter_playing(&self) {
 		let message = {
 			let mut control = self.control.lock().unwrap();
@@ -1214,18 +1111,6 @@ impl MoqSink {
 	}
 }
 
-/// Serve a finished broadcast for its linger, then release the EOS its publication owes.
-async fn linger_task(
-	finish: moq_net::broadcast::Finish,
-	element: glib::WeakRef<super::MoqSink>,
-	session: CompletionHandle,
-) {
-	finish.wait().await;
-	if let Some(element) = element.upgrade() {
-		element.imp().drained(&session);
-	}
-}
-
 /// Build a pad's producer for `caps` in the open publication, returning its status changes and whether
 /// the caps were accepted.
 fn observe_caps(
@@ -1252,11 +1137,6 @@ fn observe_caps(
 }
 
 #[cfg(test)]
-#[expect(
-	clippy::disallowed_methods,
-	clippy::disallowed_types,
-	reason = "tests read the host clock directly"
-)]
 mod tests {
 	use std::sync::Arc;
 
@@ -2117,59 +1997,6 @@ mod tests {
 		let _ = element.set_state(gst::State::Null);
 	}
 
-	// A clean end keeps the broadcast up for subscribers still catching up, and the EOS waits for it:
-	// `gst-launch` exits on that message, which would close the broadcast under them. The real sink
-	// lingers on its own runtime, so this drives the same task on a paused clock over a stand-in track.
-	#[tokio::test(start_paused = true)]
-	async fn the_eos_waits_for_the_finished_broadcast_to_go_stale() {
-		gst::init().unwrap();
-		let sink = sink();
-		sink.set_property("url", "https://127.0.0.1:1");
-		sink.set_property("broadcast", "linger-test");
-		let element = sink.clone().upcast::<gst::Element>();
-		let bus = gst::Bus::new();
-		element.set_bus(Some(&bus));
-		element.set_state(gst::State::Playing).expect("start playing");
-		let (broadcast, session) = {
-			let control = sink.imp().control.lock().unwrap();
-			let state = control.live.as_ref().expect("live session");
-			(state.broadcast.clone(), state.session.completion())
-		};
-		let consumer = broadcast.consume();
-		let max_age = Duration::from_secs(30);
-		let track = broadcast
-			.clone()
-			.create_track("video", moq_net::track::Info::default().with_max_age(max_age))
-			.unwrap();
-		track.append_group().unwrap().finish().unwrap();
-		track.finish().unwrap();
-		assert!(session.finish());
-		while bus.pop().is_some() {}
-
-		let finish = broadcast.finish();
-		assert_eq!(finish.linger(), max_age);
-		let start = tokio::time::Instant::now();
-		let linger = tokio::spawn(linger_task(finish, sink.downgrade(), session));
-		drop(track);
-
-		tokio::time::sleep(max_age - Duration::from_secs(1)).await;
-		let mut late = consumer.track("video").unwrap().subscribe(None).await.unwrap();
-		assert!(
-			late.recv_group().await.unwrap().is_some(),
-			"a late subscriber drains the tail"
-		);
-		assert!(bus.pop_filtered(&[gst::MessageType::Eos]).is_none(), "the EOS waits");
-
-		linger.await.unwrap();
-		assert_eq!(start.elapsed(), max_age);
-		assert!(consumer.is_closed());
-		assert!(
-			bus.pop_filtered(&[gst::MessageType::Eos]).is_some(),
-			"the EOS follows the linger"
-		);
-		let _ = element.set_state(gst::State::Null);
-	}
-
 	#[test]
 	fn errors_are_scoped_to_the_session_that_produced_them() {
 		gst::init().unwrap();
@@ -2244,7 +2071,6 @@ mod tests {
 			"quic-idle-timeout",
 			"quic-keep-alive",
 			"quic-keylog",
-			"linger",
 		] {
 			assert!(
 				spec(&sink, name).flags().contains(gst::PARAM_FLAG_MUTABLE_READY),
@@ -2277,7 +2103,7 @@ mod tests {
 
 	#[test]
 	fn duration_millis_saturates() {
-		assert_eq!(duration_millis(Duration::MAX), u64::MAX);
+		assert_eq!(crate::connection::duration_millis(Duration::MAX), u64::MAX);
 	}
 
 	#[test]
diff --git a/rs/moq-gst/src/sink/session.rs b/rs/moq-gst/src/sink/session.rs
index 7b07596b63c..a69fbae8435 100644
--- a/rs/moq-gst/src/sink/session.rs
+++ b/rs/moq-gst/src/sink/session.rs
@@ -137,37 +137,20 @@ pub struct ResolvedSettings {
 	pub url: url::Url,
 	/// Name to publish the broadcast under.
 	pub broadcast: String,
-	/// Disable TLS certificate verification (local/dev use).
-	pub tls_disable_verify: bool,
-	/// QUIC idle timeout override.
-	pub quic_idle_timeout: Option<std::time::Duration>,
-	/// QUIC keep-alive override, including zero to disable it.
-	pub quic_keep_alive: Option<std::time::Duration>,
-	/// Write TLS session secrets to the file `SSLKEYLOGFILE` names.
-	pub quic_keylog: bool,
+	/// The shared connection properties resolved before startup.
+	pub connect: moq_tokio::connect::Config,
+	/// The shared QUIC properties resolved before startup.
+	pub quic: moq_tokio::quic::Config,
 }
 
 /// Builds the connect configuration with the sink's TLS and backoff overrides.
 pub(super) fn connect_config(settings: &ResolvedSettings) -> moq_tokio::connect::Config {
-	let mut config = moq_tokio::connect::Config::default();
-	config.tls.insecure = Some(settings.tls_disable_verify);
-	config.backoff.timeout = Some(std::time::Duration::ZERO);
-	config
+	settings.connect.clone()
 }
 
 /// The QUIC transport overrides the sink exposes as properties.
 pub(super) fn quic_config(settings: &ResolvedSettings) -> moq_tokio::quic::Config {
-	let mut config = moq_tokio::quic::Config::default();
-	// The properties are optional and the config fields are not: an unset property
-	// leaves the library default rather than overriding it with one of its own.
-	if let Some(idle_timeout) = settings.quic_idle_timeout {
-		config.idle_timeout = Some(idle_timeout);
-	}
-	if let Some(keep_alive) = settings.quic_keep_alive {
-		config.keep_alive = Some(keep_alive);
-	}
-	config.keylog = Some(settings.quic_keylog);
-	config
+	settings.quic.clone()
 }
 
 /// Whether the publication is still open, and if not, how it ended.
@@ -555,10 +538,8 @@ mod tests {
 		let settings = ResolvedSettings {
 			url: "https://127.0.0.1:1".parse().unwrap(),
 			broadcast: "test".into(),
-			tls_disable_verify: false,
-			quic_idle_timeout: None,
-			quic_keep_alive: None,
-			quic_keylog: false,
+			connect: crate::connection::Settings::default().resolve().unwrap().0,
+			quic: Default::default(),
 		};
 		let (session, registration, broadcast, _) = Session::start(settings, glib::WeakRef::new()).unwrap();
 		registration.mark_registered();
diff --git a/rs/moq-gst/src/source/imp.rs b/rs/moq-gst/src/source/imp.rs
index 0650391255c..1a6a60e8a1e 100644
--- a/rs/moq-gst/src/source/imp.rs
+++ b/rs/moq-gst/src/source/imp.rs
@@ -40,25 +40,22 @@ static NEXT_AUDIO_PAD_ID: AtomicU64 = AtomicU64::new(0);
 struct Settings {
 	url: Option<String>,
 	broadcast: Option<String>,
-	versions: Option<String>,
-	tls_disable_verify: bool,
-	quic_keylog: bool,
+	connection: crate::connection::Settings,
 }
 
 #[derive(Debug, Clone)]
 struct ResolvedSettings {
 	url: url::Url,
 	broadcast: String,
-	/// Empty offers every supported version.
-	versions: Vec<moq_net::Version>,
-	tls_disable_verify: bool,
-	quic_keylog: bool,
+	connect: moq_tokio::connect::Config,
+	quic: moq_tokio::quic::Config,
 }
 
 impl TryFrom<Settings> for ResolvedSettings {
 	type Error = anyhow::Error;
 
 	fn try_from(value: Settings) -> Result<Self> {
+		let (connect, quic) = value.connection.resolve()?;
 		Ok(Self {
 			url: url::Url::parse(value.url.as_ref().context("url property is required")?)?,
 			broadcast: value
@@ -66,16 +63,8 @@ impl TryFrom<Settings> for ResolvedSettings {
 				.as_ref()
 				.context("broadcast property is required")?
 				.clone(),
-			versions: value
-				.versions
-				.iter()
-				.flat_map(|versions| versions.split(','))
-				.map(str::trim)
-				.filter(|version| !version.is_empty())
-				.map(|version| version.parse().map_err(anyhow::Error::msg))
-				.collect::<Result<_>>()?,
-			tls_disable_verify: value.tls_disable_verify,
-			quic_keylog: value.quic_keylog,
+			connect,
+			quic,
 		})
 	}
 }
@@ -218,11 +207,7 @@ impl SessionController {
 
 /// The dial's configuration: versions and TLS as set, retrying without a budget.
 fn connect_config(settings: &ResolvedSettings) -> moq_tokio::connect::Config {
-	let mut config = moq_tokio::connect::Config::default();
-	config.version = settings.versions.clone();
-	config.tls.insecure = Some(settings.tls_disable_verify);
-	config.backoff.timeout = Some(std::time::Duration::ZERO);
-	config
+	settings.connect.clone()
 }
 
 /// Start the reconnecting dial and the origin its announcements land in.
@@ -235,8 +220,7 @@ fn connect(
 	let _rt = RUNTIME.enter();
 	let origin = moq_tokio::origin::spawn();
 	let consumer = origin.consume();
-	let mut quic = moq_tokio::quic::Config::default();
-	quic.keylog = Some(settings.quic_keylog);
+	let quic = settings.quic.clone();
 	let client = config.init(quic)?.with_subscriber(origin);
 	let connection = client.connect(settings.url.clone());
 	Ok((client, connection, consumer))
@@ -273,26 +257,10 @@ impl ObjectImpl for MoqSrc {
 					.blurb("The broadcast name to subscribe to")
 					.mutable_ready()
 					.build(),
-				glib::ParamSpecString::builder("versions")
-					.nick("Versions")
-					.blurb("Comma-separated protocol versions to offer, most preferred first, like moq-lite-06; unset offers every stable version")
-					.mutable_ready()
-					.build(),
-				glib::ParamSpecBoolean::builder("tls-disable-verify")
-					.nick("TLS Disable Verify")
-					.blurb("Disable TLS certificate verification")
-					.default_value(false)
-					.mutable_ready()
-					.build(),
-				glib::ParamSpecBoolean::builder("quic-keylog")
-					.nick("QUIC TLS key log")
-					.blurb(
-						"Danger: write TLS session secrets to the file SSLKEYLOGFILE names; without it the start fails",
-					)
-					.default_value(false)
-					.mutable_ready()
-					.build(),
 			]
+			.into_iter()
+			.chain(crate::connection::properties())
+			.collect()
 		});
 		PROPS.as_ref()
 	}
@@ -320,10 +288,7 @@ impl ObjectImpl for MoqSrc {
 		match pspec.name() {
 			"url" => settings.url = value.get().unwrap(),
 			"broadcast" => settings.broadcast = value.get().unwrap(),
-			"versions" => settings.versions = value.get().unwrap(),
-			"tls-disable-verify" => settings.tls_disable_verify = value.get().unwrap(),
-			"quic-keylog" => settings.quic_keylog = value.get().unwrap(),
-			_ => unreachable!(),
+			name => settings.connection.set(name, value),
 		}
 	}
 
@@ -332,10 +297,7 @@ impl ObjectImpl for MoqSrc {
 		match pspec.name() {
 			"url" => settings.url.to_value(),
 			"broadcast" => settings.broadcast.to_value(),
-			"versions" => settings.versions.to_value(),
-			"tls-disable-verify" => settings.tls_disable_verify.to_value(),
-			"quic-keylog" => settings.quic_keylog.to_value(),
-			_ => unreachable!(),
+			name => settings.connection.get(name),
 		}
 	}
 }
@@ -1976,9 +1938,7 @@ mod session_tests {
 	use hang::catalog::{AudioCodec, AudioConfig, Container, H264, VideoCodec, VideoConfig};
 	use tokio::sync::watch;
 
-	use super::{
-		NEXT_VIDEO_PAD_ID, Pads, ResolvedSettings, SessionController, Settings, TrackKind, follow_catalog, follow_path,
-	};
+	use super::{NEXT_VIDEO_PAD_ID, Pads, ResolvedSettings, SessionController, TrackKind, follow_catalog, follow_path};
 
 	/// The pad-id counters are process-global, so a test reading one has to be the only test
 	/// allocating while it runs. `cargo test` shares a process across tests (nextest doesn't),
@@ -2942,31 +2902,11 @@ mod session_tests {
 		ResolvedSettings {
 			url: "https://127.0.0.1:1".parse().unwrap(),
 			broadcast: "test".into(),
-			versions: Vec::new(),
-			tls_disable_verify: false,
-			quic_keylog: false,
+			connect: crate::connection::Settings::default().resolve().unwrap().0,
+			quic: Default::default(),
 		}
 	}
 
-	/// `versions` reaches the dial in order, and an unknown name fails the start instead of
-	/// offering the defaults.
-	#[test]
-	fn versions_reach_the_dial() {
-		let settings = |versions: &str| Settings {
-			url: Some("https://127.0.0.1:1".into()),
-			broadcast: Some("test".into()),
-			versions: Some(versions.into()),
-			..Default::default()
-		};
-		let resolved = ResolvedSettings::try_from(settings(" moq-transport-14 , moq-lite-06 ")).unwrap();
-		let config = super::connect_config(&resolved);
-		let expected: Vec<moq_net::Version> = vec!["moq-transport-14".parse().unwrap(), "moq-lite-06".parse().unwrap()];
-		assert_eq!(config.version, expected);
-
-		let err = ResolvedSettings::try_from(settings("moq-lite-06,moq-lite-99")).unwrap_err();
-		assert!(err.to_string().contains("moq-lite-99"), "{err}");
-	}
-
 	/// The source retries without a budget, so an outage of any length is ridden out
 	/// (`relay_loss_reconnects_after_refused_redials` covers the redial itself).
 	#[test]
@@ -2981,10 +2921,8 @@ mod session_tests {
 		if !test_env::Env::without(&["SSLKEYLOGFILE"]).isolate() {
 			return;
 		}
-		let keylog = ResolvedSettings {
-			quic_keylog: true,
-			..unreachable()
-		};
+		let mut keylog = unreachable();
+		keylog.quic.keylog = Some(true);
 		let err = super::connect(&keylog)
 			.err()
 			.expect("a keylog without SSLKEYLOGFILE must be refused");
diff --git a/rs/moq-gst/tests/element.rs b/rs/moq-gst/tests/element.rs
index 6e884dbd528..de6a72a8dbf 100644
--- a/rs/moq-gst/tests/element.rs
+++ b/rs/moq-gst/tests/element.rs
@@ -226,13 +226,12 @@ fn play_one_frame(sink: &gst::Element) -> gst::Pad {
 	pad
 }
 
-// Lingering is opt-in: by default the end of stream closes the broadcast and posts EOS at once, even
-// with media a late subscriber could still have drained.
+// The end of stream closes the broadcast and posts EOS at once, even with media a late subscriber
+// could still have drained.
 #[test]
-fn published_media_posts_the_eos_at_once_by_default() {
+fn published_media_posts_the_eos_at_once() {
 	init();
 	let sink = publisher();
-	assert!(!sink.property::<bool>("linger"));
 	let seen = recording_bus(&sink);
 	let pad = play_one_frame(&sink);
 
@@ -241,25 +240,6 @@ fn published_media_posts_the_eos_at_once_by_default() {
 	let _ = sink.set_state(gst::State::Null);
 }
 
-// With `linger`, published media is served until it is stale, so the EOS that ends `gst-launch` waits
-// for it. Stopping the pipeline is the interrupt: it closes the broadcast at once, and the run it ended
-// owes no EOS.
-#[test]
-fn a_lingering_sink_holds_the_eos_until_the_pipeline_stops() {
-	init();
-	let sink = publisher();
-	sink.set_property("linger", true);
-	let seen = recording_bus(&sink);
-	let pad = play_one_frame(&sink);
-
-	assert!(pad.send_event(gst::event::Eos::new()));
-	assert_eq!(status_of(&sink, "sink_0"), "ended");
-	assert_eq!(posted_eos(&seen), 0, "the finished broadcast is still serving its tail");
-
-	assert_eq!(sink.set_state(gst::State::Null), Ok(gst::StateChangeSuccess::Success));
-	assert_eq!(posted_eos(&seen), 0, "the stopped run owes no EOS");
-}
-
 // After the publication ends, a buffer that got past the pad's own EOS flag (a flush clears it) must
 // report the end rather than be dropped as if it had been published.
 #[test]
@@ -1568,3 +1548,18 @@ fn connect_failure_retries_without_erroring() {
 	assert_eq!(send_bitrate, 0);
 	assert_eq!(recv_bitrate, 0);
 }
+
+// Both element factories expose the same connection syntax, and refuse unknown versions
+// before opening a transport rather than silently offering the defaults.
+#[test]
+fn parsed_connection_properties_are_shared_and_validated() {
+	init();
+	for factory in ["moqsink", "moqsrc"] {
+		let element = gst::parse::launch(&format!("{factory} url=https://127.0.0.1:1 broadcast=test connect-version=moq-lite-99 quic-idle-timeout=15000 quic-keep-alive=0"))
+			.expect("parse the shared property syntax");
+		assert_eq!(element.property::<u64>("quic-idle-timeout"), 15_000);
+		assert_eq!(element.property::<u64>("quic-keep-alive"), 0);
+		assert_eq!(element.set_state(gst::State::Paused), Err(gst::StateChangeError));
+		element.set_state(gst::State::Null).unwrap();
+	}
+}
diff --git a/rs/moq-relay/src/cluster.rs b/rs/moq-relay/src/cluster.rs
index 3c24c5291d2..c555b048421 100644
--- a/rs/moq-relay/src/cluster.rs
+++ b/rs/moq-relay/src/cluster.rs
@@ -750,7 +750,7 @@ pub struct Config {
 
 	/// Connect to one or more other cluster nodes. Each entry is a full URL, e.g.
 	/// `https://host/?jwt=TOKEN`, or an object with `url`, `cost`, `egress`,
-	/// `token`, and `upstream`; see [`Peer`]. A bare host or `host:port` is refused; pass the
+	/// `token`, and `upstream`. A bare host or `host:port` is refused; pass the
 	/// full URL instead. Accepts a comma-separated list on the CLI or repeat the
 	/// flag; in config files use a TOML array of URLs and/or objects.
 	///
@@ -793,7 +793,7 @@ pub struct Config {
 	pub(crate) removed_api_root: Option<serde::de::IgnoredAny>,
 
 	/// This relay's own externally-reachable URL (identity). Sent to
-	/// [`Self::api`] as a `?node=` query param so the endpoint can return
+	/// `--cluster-api-url` as a `?node=` query param so the endpoint can return
 	/// this node's peers, and advertised over mDNS when LAN discovery is on. On
 	/// its own it neither opens nor accepts a connection.
 	#[usage(
@@ -886,7 +886,7 @@ impl Config {
 pub struct LanConfig {
 	/// Enable mDNS discovery. Boolean flag: pass `--cluster-lan` (or `=true` /
 	/// `=false`). Advertises the listener fingerprint when the certificate was
-	/// generated or supplied in-memory, the [`Config::node`] URL when one
+	/// generated or supplied in-memory, the `--cluster-node` URL when one
 	/// is configured, and needs at least one of them.
 	#[usage(
 		name = "cluster-lan",
diff --git a/rs/moq-relay/src/cluster/api.rs b/rs/moq-relay/src/cluster/api.rs
index 3714285b10b..2278557d0d3 100644
--- a/rs/moq-relay/src/cluster/api.rs
+++ b/rs/moq-relay/src/cluster/api.rs
@@ -11,16 +11,16 @@ pub struct Config {
 	/// Fetch the list of peers to dial from an HTTP(S) URL or a local file,
 	/// reloading at runtime without a restart. The source returns a JSON array
 	/// of peers: bare URL strings and/or objects with `url`, `cost`, `egress`,
-	/// `token`, and `upstream`, exactly as [`super::Config::connect`] accepts, e.g.
+	/// `token`, and `upstream`, exactly as `--cluster-connect` accepts, e.g.
 	/// `["https://a.pop.example/?cost=1", {"url": "https://b.pop.example/", "cost": 2}]`.
 	/// An http(s) URL is re-checked on a fixed cadence, with caching, conditional revalidation
 	/// (`ETag` / `Last-Modified`), and stale-if-error handled by the shared HTTP
 	/// cache client, so the response's `Cache-Control` controls how often a real
 	/// fetch hits the endpoint; a local path is watched via OS filesystem
 	/// notifications (with a periodic re-check fallback). This relay's own
-	/// [`super::Config::node`] value, when set, is sent as a `?node=` query param so the
+	/// `--cluster-node` value, when set, is sent as a `?node=` query param so the
 	/// server can return this node's peers. The relay keeps the last good list if
-	/// a fetch fails. Composes with [`super::Config::connect`].
+	/// a fetch fails. Composes with `--cluster-connect`.
 
 	#[usage(long = "cluster-api-url", env = "MOQ_CLUSTER_API_URL", setting = "cluster.api.url")]
 	pub url: Option<String>,
diff --git a/rs/moq-relay/src/stats.rs b/rs/moq-relay/src/stats.rs
index a41bf8ba93a..e31011274d5 100644
--- a/rs/moq-relay/src/stats.rs
+++ b/rs/moq-relay/src/stats.rs
@@ -72,7 +72,7 @@ pub struct Config {
 	/// single `<prefix>/node/<node>` broadcast for the whole node. Set to 1 to
 	/// publish a per-first-segment broadcast (e.g. per tenant), so a consumer can
 	/// announce-scope to just that group rather than slurping every node's full
-	/// stats. See [`moq_stats::produce::Config::depth`].
+	/// stats.
 	#[usage(
 		long = "stats-depth",
 		env = "MOQ_STATS_DEPTH",
diff --git a/rs/moq-relay/tests/integration/goaway_cluster.rs b/rs/moq-relay/tests/integration/goaway_cluster.rs
index 8aa8c448d59..9df99a48fb1 100644
--- a/rs/moq-relay/tests/integration/goaway_cluster.rs
+++ b/rs/moq-relay/tests/integration/goaway_cluster.rs
@@ -26,7 +26,7 @@ const TEST_TIMEOUT: Duration = Duration::from_secs(15);
 /// the session as a failed attempt, so the redial waits this long first.
 const BACKOFF: Duration = Duration::from_millis(300);
 
-/// The cluster's `--goaway-handover`: short, so a test observes the old session
+/// The cluster's `--connect-goaway-handover`: short, so a test observes the old session
 /// close quickly, but past [`BACKOFF`] plus the redial, so the draining leg
 /// serves until the replacement takes over.
 const HANDOVER: Duration = Duration::from_secs(1);
@@ -822,7 +822,7 @@ async fn goaway_handover_is_enforced_while_the_replacement_dial_hangs_inner() {
 	tokio::time::sleep(Duration::from_millis(120)).await;
 
 	// Send the client somewhere that will never finish handshaking. No wire timeout:
-	// our own `--goaway-handover` is then the only thing bounding the old session.
+	// our own `--connect-goaway-handover` is then the only thing bounding the old session.
 	server_session
 		.drain()
 		.send(moq_net::goaway::Goaway::redirect(format!(
diff --git a/rs/moq-tokio/src/connect.rs b/rs/moq-tokio/src/connect.rs
index d2cdfc30a8c..b73e2bb4061 100644
--- a/rs/moq-tokio/src/connect.rs
+++ b/rs/moq-tokio/src/connect.rs
@@ -641,8 +641,8 @@ pub struct Config {
 
 	/// Dial once instead of redialing (with backoff) whenever the session drops.
 	///
-	/// A [`crate::Connection`] reconnects by default. With this set, the session's
-	/// close surfaces through [`crate::Connection::closed`] instead.
+	/// The connection reconnects by default. With this set, the session's
+	/// close surfaces through its closed event instead.
 	#[serde(skip_serializing_if = "Option::is_none")]
 	#[usage(
 		name = "connect-once",
@@ -665,12 +665,12 @@ pub struct Config {
 	#[usage(skip)]
 	pub(crate) reconnect: Option<bool>,
 
-	/// Retry pacing for [`crate::Client::connect`] (`--backoff-*`).
+	/// Retry pacing for [`crate::Client::connect`] (`--connect-backoff-*`).
 	#[usage(flatten)]
 	#[serde(default)]
 	pub backoff: Backoff,
 
-	/// How [`crate::Client::connect`] reacts to a peer's GOAWAY (`--goaway-*`).
+	/// How [`crate::Client::connect`] reacts to a peer's GOAWAY (`--connect-goaway-*`).
 	#[usage(flatten)]
 	#[serde(default)]
 	pub goaway: Goaway,
@@ -736,6 +736,8 @@ impl Config {
 	/// skipped the check can't reach a dial that quietly ignored half of it.
 	pub fn deprecated(&self) -> crate::cli::Deprecated {
 		let mut found = self.legacy.deprecated();
+		found.extend(self.backoff.deprecated());
+		found.extend(self.goaway.deprecated());
 		if self.connect.is_some() {
 			found.toml("connect", "url", None);
 		}
diff --git a/rs/moq-tokio/src/connection.rs b/rs/moq-tokio/src/connection.rs
index 31c87b02d70..6ceffff789c 100644
--- a/rs/moq-tokio/src/connection.rs
+++ b/rs/moq-tokio/src/connection.rs
@@ -80,9 +80,9 @@ pub struct Backoff {
 
 	/// Initial delay before the first reconnect attempt. Defaults to 1s, floored at 50ms.
 	#[usage(
-		name = "backoff-initial",
+		name = "connect-backoff-initial",
 		long,
-		env = "MOQ_BACKOFF_INITIAL",
+		env = "MOQ_CONNECT_BACKOFF_INITIAL",
 		setting = "connect.backoff.initial"
 	)]
 	#[serde(default, rename = "__cli_initial", skip_serializing_if = "Option::is_none")]
@@ -90,9 +90,9 @@ pub struct Backoff {
 
 	/// Multiplier applied to delay after each failure. Defaults to 2.
 	#[usage(
-		name = "backoff-multiplier",
+		name = "connect-backoff-multiplier",
 		long,
-		env = "MOQ_BACKOFF_MULTIPLIER",
+		env = "MOQ_CONNECT_BACKOFF_MULTIPLIER",
 		default = "2",
 		setting = "connect.backoff.multiplier"
 	)]
@@ -105,7 +105,12 @@ pub struct Backoff {
 	pub max: Option<Duration>,
 
 	/// Maximum delay between reconnect attempts. Defaults to 5s.
-	#[usage(name = "backoff-max", long, env = "MOQ_BACKOFF_MAX", setting = "connect.backoff.max")]
+	#[usage(
+		name = "connect-backoff-max",
+		long,
+		env = "MOQ_CONNECT_BACKOFF_MAX",
+		setting = "connect.backoff.max"
+	)]
 	#[serde(default, rename = "__cli_max", skip_serializing_if = "Option::is_none")]
 	max_arg: Option<CliDuration>,
 
@@ -119,7 +124,7 @@ pub struct Backoff {
 	/// Resets after a stable connection (one that outlives the initial backoff) outside a redirect
 	/// loop, so a flapping session that reconnects then immediately drops still counts toward the
 	/// timeout. Set to `Some(Duration::ZERO)` for unlimited retries. `None` takes the default. When this config was
-	/// parsed from the command line, a `--backoff-timeout` given there wins over this field.
+	/// parsed from the command line, a `--connect-backoff-timeout` given there wins over this field.
 	///
 	/// An `Option` because zero is a real setting: a command-line parse leaves this field at its
 	/// type's default, and only `None` tells "unset" apart from "retry forever".
@@ -131,13 +136,43 @@ pub struct Backoff {
 	// window set in code. The default lives in [`Backoff::window`] instead.
 	/// Maximum time to spend retrying before giving up. Defaults to 10s; 0s retries forever.
 	#[usage(
-		name = "backoff-timeout",
+		name = "connect-backoff-timeout",
 		long,
-		env = "MOQ_BACKOFF_TIMEOUT",
+		env = "MOQ_CONNECT_BACKOFF_TIMEOUT",
 		setting = "connect.backoff.timeout"
 	)]
 	#[serde(default, rename = "__cli_timeout", skip_serializing_if = "Option::is_none")]
 	timeout_arg: Option<CliDuration>,
+	#[usage(
+		name = "backoff-initial",
+		long = "backoff-initial",
+		env = "MOQ_BACKOFF_INITIAL",
+		hide = true
+	)]
+	#[serde(skip)]
+	legacy_initial: Option<CliDuration>,
+
+	#[usage(
+		name = "backoff-multiplier",
+		long = "backoff-multiplier",
+		env = "MOQ_BACKOFF_MULTIPLIER",
+		hide = true
+	)]
+	#[serde(skip)]
+	legacy_multiplier: Option<u32>,
+
+	#[usage(name = "backoff-max", long = "backoff-max", env = "MOQ_BACKOFF_MAX", hide = true)]
+	#[serde(skip)]
+	legacy_max: Option<CliDuration>,
+
+	#[usage(
+		name = "backoff-timeout",
+		long = "backoff-timeout",
+		env = "MOQ_BACKOFF_TIMEOUT",
+		hide = true
+	)]
+	#[serde(skip)]
+	legacy_timeout: Option<CliDuration>,
 }
 
 impl Default for Backoff {
@@ -150,11 +185,48 @@ impl Default for Backoff {
 			max_arg: None,
 			timeout: Some(DEFAULT_TIMEOUT),
 			timeout_arg: None,
+			legacy_initial: None,
+			legacy_multiplier: None,
+			legacy_max: None,
+			legacy_timeout: None,
 		}
 	}
 }
 
 impl Backoff {
+	pub(crate) fn deprecated(&self) -> crate::cli::Deprecated {
+		let mut found = crate::cli::Deprecated::default();
+		if self.legacy_initial.is_some() {
+			found.flag(
+				"--backoff-initial",
+				Some("MOQ_BACKOFF_INITIAL"),
+				"--connect-backoff-initial / MOQ_CONNECT_BACKOFF_INITIAL",
+			);
+		}
+		if self.legacy_multiplier.is_some() {
+			found.flag(
+				"--backoff-multiplier",
+				Some("MOQ_BACKOFF_MULTIPLIER"),
+				"--connect-backoff-multiplier / MOQ_CONNECT_BACKOFF_MULTIPLIER",
+			);
+		}
+		if self.legacy_max.is_some() {
+			found.flag(
+				"--backoff-max",
+				Some("MOQ_BACKOFF_MAX"),
+				"--connect-backoff-max / MOQ_CONNECT_BACKOFF_MAX",
+			);
+		}
+		if self.legacy_timeout.is_some() {
+			found.flag(
+				"--backoff-timeout",
+				Some("MOQ_BACKOFF_TIMEOUT"),
+				"--connect-backoff-timeout / MOQ_CONNECT_BACKOFF_TIMEOUT",
+			);
+		}
+		found
+	}
+
 	/// The retry window in effect, where zero means unlimited: the flag, then the typed field, then
 	/// the default.
 	fn window(&self) -> Duration {
@@ -341,9 +413,9 @@ pub struct Goaway {
 	/// chose, `follow` also lets it name the host, and `ignore` redials the
 	/// current address list.
 	#[usage(
-		name = "goaway-redirect",
+		name = "connect-goaway-redirect",
 		long,
-		env = "MOQ_GOAWAY_REDIRECT",
+		env = "MOQ_CONNECT_GOAWAY_REDIRECT",
 		value_enum,
 		default = "same-host",
 		setting = "connect.goaway.redirect"
@@ -361,13 +433,31 @@ pub struct Goaway {
 
 	/// How long the old session keeps serving after its replacement connects. Defaults to 10s.
 	#[usage(
-		name = "goaway-handover",
+		name = "connect-goaway-handover",
 		long,
-		env = "MOQ_GOAWAY_HANDOVER",
+		env = "MOQ_CONNECT_GOAWAY_HANDOVER",
 		setting = "connect.goaway.handover"
 	)]
 	#[serde(default, rename = "__cli_handover", skip_serializing_if = "Option::is_none")]
 	handover_arg: Option<CliDuration>,
+	#[usage(
+		name = "goaway-redirect",
+		long = "goaway-redirect",
+		env = "MOQ_GOAWAY_REDIRECT",
+		value_enum,
+		hide = true
+	)]
+	#[serde(skip)]
+	legacy_redirect: Option<Redirect>,
+
+	#[usage(
+		name = "goaway-handover",
+		long = "goaway-handover",
+		env = "MOQ_GOAWAY_HANDOVER",
+		hide = true
+	)]
+	#[serde(skip)]
+	legacy_handover: Option<CliDuration>,
 }
 
 impl Default for Goaway {
@@ -376,6 +466,8 @@ impl Default for Goaway {
 			redirect: Redirect::SameHost,
 			handover: Some(DEFAULT_HANDOVER),
 			handover_arg: None,
+			legacy_redirect: None,
+			legacy_handover: None,
 		}
 	}
 }
@@ -437,6 +529,25 @@ impl Pacing {
 const DEFAULT_HANDOVER: Duration = Duration::from_secs(10);
 
 impl Goaway {
+	pub(crate) fn deprecated(&self) -> crate::cli::Deprecated {
+		let mut found = crate::cli::Deprecated::default();
+		if self.legacy_redirect.is_some() {
+			found.flag(
+				"--goaway-redirect",
+				Some("MOQ_GOAWAY_REDIRECT"),
+				"--connect-goaway-redirect / MOQ_CONNECT_GOAWAY_REDIRECT",
+			);
+		}
+		if self.legacy_handover.is_some() {
+			found.flag(
+				"--goaway-handover",
+				Some("MOQ_GOAWAY_HANDOVER"),
+				"--connect-goaway-handover / MOQ_CONNECT_GOAWAY_HANDOVER",
+			);
+		}
+		found
+	}
+
 	fn resolve(&self) -> Resolved {
 		Resolved {
 			redirect: self.redirect,
@@ -1187,7 +1298,7 @@ impl Connection {
 			// replacement dial, so a predecessor can still be draining when that dial
 			// fails and lands here. A plain sleep would stop enforcing its handover
 			// until some later dial succeeded, holding an upstream that asked to drain
-			// open for the whole retry window, or forever with `--backoff-timeout=0`.
+			// open for the whole retry window, or forever with `--connect-backoff-timeout=0`.
 			sleep_draining(wait, &mut draining, shared).await;
 			retry.delay = pacing.next(retry.delay);
 		}
@@ -1817,8 +1928,11 @@ mod tests {
 		assert_eq!(parsed.backoff.window(), DEFAULT_TIMEOUT);
 
 		// And a flag still lands where the merge can see it.
-		let parsed =
-			Wrapper::parse_from(&[std::ffi::OsStr::new("--backoff-initial"), std::ffi::OsStr::new("3s")]).unwrap();
+		let parsed = Wrapper::parse_from(&[
+			std::ffi::OsStr::new("--connect-backoff-initial"),
+			std::ffi::OsStr::new("3s"),
+		])
+		.unwrap();
 		assert_eq!(Pacing::new(&parsed.backoff).initial, Duration::from_secs(3));
 	}
 
@@ -1842,9 +1956,9 @@ mod tests {
 
 		// Flags passed: they land where the merge can see them.
 		let parsed = Wrapper::parse_from(&[
-			std::ffi::OsStr::new("--goaway-redirect"),
+			std::ffi::OsStr::new("--connect-goaway-redirect"),
 			std::ffi::OsStr::new("ignore"),
-			std::ffi::OsStr::new("--goaway-handover"),
+			std::ffi::OsStr::new("--connect-goaway-handover"),
 			std::ffi::OsStr::new("3s"),
 		])
 		.unwrap();
@@ -3212,7 +3326,7 @@ mod duration_tests {
 		initial_set_in_code_beats_parser_default,
 		Backoff,
 		initial,
-		"--backoff-initial",
+		"--connect-backoff-initial",
 		|c: &Backoff| Pacing::new(c).initial,
 		|d: Duration| d.max(MIN_BACKOFF)
 	);
@@ -3220,7 +3334,7 @@ mod duration_tests {
 		max_set_in_code_beats_parser_default,
 		Backoff,
 		max,
-		"--backoff-max",
+		"--connect-backoff-max",
 		|c: &Backoff| Pacing::new(c).max,
 		|d: Duration| d.max(DEFAULT_INITIAL)
 	);
@@ -3228,7 +3342,7 @@ mod duration_tests {
 		timeout_set_in_code_beats_parser_default,
 		Backoff,
 		timeout,
-		"--backoff-timeout",
+		"--connect-backoff-timeout",
 		|c: &Backoff| c.window(),
 		|d| d
 	);
@@ -3236,7 +3350,7 @@ mod duration_tests {
 		handover_set_in_code_beats_parser_default,
 		Goaway,
 		handover,
-		"--goaway-handover",
+		"--connect-goaway-handover",
 		|c: &Goaway| c.resolve().handover,
 		|d| d
 	);
diff --git a/rs/moq-tokio/src/listen.rs b/rs/moq-tokio/src/listen.rs
index ce57c681422..e2a6afa1a6b 100644
--- a/rs/moq-tokio/src/listen.rs
+++ b/rs/moq-tokio/src/listen.rs
@@ -188,7 +188,7 @@ pub struct Config {
 	/// anycast IP, steady-state on this host's unicast IP.
 	///
 	/// Honored by the QUIC backend. Accept-only, which is why it lives
-	/// here rather than in the shared [`crate::quic::Config`].
+	/// here rather than in the shared QUIC settings.
 	#[usage(
 		name = "listen-preferred-v4",
 		long = "listen-preferred-v4",
@@ -198,7 +198,7 @@ pub struct Config {
 	#[serde(default, skip_serializing_if = "Option::is_none")]
 	pub preferred_v4: Option<std::net::SocketAddrV4>,
 
-	/// IPv6 address advertised as the QUIC preferred_address. See [`Self::preferred_v4`].
+	/// IPv6 address advertised as the QUIC preferred_address. Like `--listen-preferred-v4`, for IPv6.
 	#[usage(
 		name = "listen-preferred-v6",
 		long = "listen-preferred-v6",
diff --git a/rs/moq-tokio/src/settings.rs b/rs/moq-tokio/src/settings.rs
index efd8521bd33..61d820479ea 100644
--- a/rs/moq-tokio/src/settings.rs
+++ b/rs/moq-tokio/src/settings.rs
@@ -200,31 +200,31 @@ struct ConnectTls {
 	host_name: Option<String>,
 }
 
-/// `[connect.backoff]` / `--backoff-*`.
+/// `[connect.backoff]` / `--connect-backoff-*`.
 #[derive(usage_rs::Config)]
 #[usage(prefix = "connect.backoff")]
 struct ConnectBackoff {
-	#[usage(env = "MOQ_BACKOFF_INITIAL", cli("--backoff-initial"))]
+	#[usage(env = "MOQ_CONNECT_BACKOFF_INITIAL", cli("--connect-backoff-initial"))]
 	initial: Option<String>,
 
-	#[usage(env = "MOQ_BACKOFF_MULTIPLIER", cli("--backoff-multiplier"))]
+	#[usage(env = "MOQ_CONNECT_BACKOFF_MULTIPLIER", cli("--connect-backoff-multiplier"))]
 	multiplier: Option<u32>,
 
-	#[usage(env = "MOQ_BACKOFF_MAX", cli("--backoff-max"))]
+	#[usage(env = "MOQ_CONNECT_BACKOFF_MAX", cli("--connect-backoff-max"))]
 	max: Option<String>,
 
-	#[usage(env = "MOQ_BACKOFF_TIMEOUT", cli("--backoff-timeout"))]
+	#[usage(env = "MOQ_CONNECT_BACKOFF_TIMEOUT", cli("--connect-backoff-timeout"))]
 	timeout: Option<String>,
 }
 
-/// `[connect.goaway]` / `--goaway-*`.
+/// `[connect.goaway]` / `--connect-goaway-*`.
 #[derive(usage_rs::Config)]
 #[usage(prefix = "connect.goaway")]
 struct ConnectGoaway {
-	#[usage(env = "MOQ_GOAWAY_REDIRECT", cli("--goaway-redirect"))]
+	#[usage(env = "MOQ_CONNECT_GOAWAY_REDIRECT", cli("--connect-goaway-redirect"))]
 	redirect: Option<String>,
 
-	#[usage(env = "MOQ_GOAWAY_HANDOVER", cli("--goaway-handover"))]
+	#[usage(env = "MOQ_CONNECT_GOAWAY_HANDOVER", cli("--connect-goaway-handover"))]
 	handover: Option<String>,
 }
 
diff --git a/rs/moq-tokio/src/tcp.rs b/rs/moq-tokio/src/tcp.rs
index b2591d78df2..92c15552915 100644
--- a/rs/moq-tokio/src/tcp.rs
+++ b/rs/moq-tokio/src/tcp.rs
@@ -30,7 +30,7 @@ const WIRE_VERSION: qmux::Version = qmux::Version::QMux01;
 #[serde(deny_unknown_fields, default)]
 #[non_exhaustive]
 pub struct Config {
-	/// Bind a qmux TCP listener on this address, plaintext unless [`Self::tls`].
+	/// Bind a qmux TCP listener on this address, plaintext unless `--listen-tcp-tls` is set.
 	#[usage(
 		long = "listen-tcp-bind",
 		name = "listen-tcp-bind",
diff --git a/rs/moq-tokio/src/tls.rs b/rs/moq-tokio/src/tls.rs
index 976b53b58e8..2fca3f373ae 100644
--- a/rs/moq-tokio/src/tls.rs
+++ b/rs/moq-tokio/src/tls.rs
@@ -1287,11 +1287,11 @@ pub struct Listen {
 	/// PEM file(s) of root CAs for validating optional client certificates (mTLS).
 	///
 	/// When set, clients *may* present a certificate during the TLS handshake.
-	/// Valid presentations are reported via [`crate::server::Request::peer_identity`]
+	/// Valid presentations are reported as the accepted request's peer identity
 	/// and can be used by the application to grant elevated access. Clients that
 	/// do not present a certificate are unaffected.
 	///
-	/// Plain-TLS listeners built via [`Self::server_config`] also use these roots
+	/// Plain-TLS listeners also use these roots
 	/// for optional mTLS. Root files are hot reloaded for new handshakes.
 	#[usage(
 		long = "listen-tls-root",
@@ -1408,8 +1408,7 @@ impl Listen {
 	/// The released spellings in use, each paired with what replaced it.
 	///
 	/// Public for the same reason as [`Connect::deprecated`]: this type is flattened
-	/// into parsers of its own (moq-cli's `export hls` is one), not only through
-	/// [`crate::listen::Config`]. The methods that build a server config refuse
+	/// into custom listener parsers, not only through [`crate::listen::Config`]. The methods that build a server config refuse
 	/// anyway, since a dropped `--server-tls-root` would take the mTLS client CAs
 	/// with it and leave the listener accepting unauthenticated peers.
 	pub fn deprecated(&self) -> crate::cli::Deprecated {
diff --git a/rs/moq-tokio/tests/integration/reconnect.rs b/rs/moq-tokio/tests/integration/reconnect.rs
index e7e61f2242b..86ddb905e76 100644
--- a/rs/moq-tokio/tests/integration/reconnect.rs
+++ b/rs/moq-tokio/tests/integration/reconnect.rs
@@ -267,7 +267,7 @@ async fn a_redirect_to_another_host_is_refused_by_default() {
 	);
 }
 
-/// `--goaway-redirect follow` is the opt-in that hands the peer the host, so the
+/// `--connect-goaway-redirect follow` is the opt-in that hands the peer the host, so the
 /// same redirect is followed. Without this the default above would be
 /// indistinguishable from ignoring the URI outright.
 #[tokio::test]
diff --git a/test/interop/interop.sh b/test/interop/interop.sh
index 279e26bcd63..09c2bb161dc 100755
--- a/test/interop/interop.sh
+++ b/test/interop/interop.sh
@@ -1117,7 +1117,7 @@ run_gst_audio() {
     done
     GST_PLUGIN_PATH_1_0="$GST_PLUGIN_DIR" GST_REGISTRY_1_0="$HARNESS_RUN/gst-run-registry.bin" \
         GST_TRACERS=log GST_DEBUG=GST_BUFFER:7,GST_EVENT:7 GST_DEBUG_NO_COLOR=1 GST_DEBUG_FILE="$out.trace" \
-        gst-launch-1.0 moqsrc name=s url="$url" broadcast="$broadcast" ${version:+versions="$version"} \
+        gst-launch-1.0 moqsrc name=s url="$url" broadcast="$broadcast" ${version:+connect-version="$version"} \
         s.video_0 ! filesink location="$out.video" buffer-mode=2 "${pads[@]}" >"$out.gst" 2>&1 &
     pid=$!
     while :; do
@@ -1330,7 +1330,7 @@ PYCODE
                 trap '' PIPE
                 GST_PLUGIN_PATH_1_0="$GST_PLUGIN_DIR" GST_REGISTRY_1_0="$HARNESS_RUN/gst-run-registry.bin" \
                     timeout -k 3 "$TIMEOUT" gst-launch-1.0 -q \
-                    moqsrc name=s url="$url" broadcast="$broadcast" ${version:+versions="$version"} \
+                    moqsrc name=s url="$url" broadcast="$broadcast" ${version:+connect-version="$version"} \
                     s.video_0 ! filesink location=/dev/stdout buffer-mode=2 \
                     2>/dev/null | head -c 1 | wc -c | tr -d ' ' || true
             )
@@ -1552,7 +1552,7 @@ run_finish_gst() {
     deadline=$((SECONDS + ${TIMEOUT%.*} * 2))
     [[ "$TIMEOUT" != *.* ]] || deadline=$((deadline + 2))
     GST_PLUGIN_PATH_1_0="$GST_PLUGIN_DIR" GST_REGISTRY_1_0="$HARNESS_RUN/gst-run-registry.bin" \
-        gst-launch-1.0 -v moqsrc name=s url="$url" broadcast="$broadcast" ${version:+versions="$version"} \
+        gst-launch-1.0 -v moqsrc name=s url="$url" broadcast="$broadcast" ${version:+connect-version="$version"} \
         s.video_0 ! identity name=frames silent=false ! fakesink >"$out" 2>&1 &
     pid=$!
     count=0