diff --git a/doc/bin/relay/cluster.md b/doc/bin/relay/cluster.md
index d64da538e7..86178b8dc5 100644
--- a/doc/bin/relay/cluster.md
+++ b/doc/bin/relay/cluster.md
@@ -48,6 +48,57 @@ hop from the relay it connects to, fresh for each connection, followed by a 0.
 Its reconnect is therefore a new publisher downstream, a reprice on the same
 connection stays in place, and the 0 keeps it ranked as anonymous.
 
+## Edge and core roles
+
+Set `cluster.role = "edge"` for relays that serve end users, or `"core"`
+for the backbone. Leave it unset during migration to preserve the legacy mesh.
+A core accepts only sessions whose auth grant sets `peer = true`, including
+HTTP fetches and announcements. A verified certificate alone is not membership.
+Edges refuse inbound peer sessions; they dial their region's cores themselves.
+
+An edge's peer entries must declare `role = "core"` and use `tls://`.
+An edge advertises only routes ingested locally, so a route received from one
+core cannot reach another core through that edge. Cores may dial explicit cores
+and legacy peers, but refuse peers declared as edges. A role-aware relay refuses
+LAN discovery because its advertisements do not declare peer roles.
+
+```toml
+[cluster]
+role = "edge"
+connect = [
+  { url = "tls://core-a.example:4444/", role = "core", token = "CLUSTER_JWT" },
+  { url = "tls://core-b.example:4444/", role = "core", token = "CLUSTER_JWT" },
+]
+```
+
+On each core, enable encrypted qmux on its TCP listener:
+
+```toml
+[listen.tcp]
+bind = "0.0.0.0:4444"
+tls = true
+
+[listen.tls]
+cert = "core.pem"
+key = "core.key"
+
+[cluster]
+role = "core"
+```
+
+The `tls://` scheme requires an explicit port and uses `connect.tls` trust,
+certificate, and hostname settings. It speaks qmux directly over TLS on TCP.
+Its listener reports the TCP transport to the auth server, as both secure and
+plaintext WebSocket listeners report WebSocket. Use peer grants authenticated
+with cluster credentials. The TLS TCP listener refuses `listen.tls.root` because
+qmux's TLS accept API does not expose the verified client certificate to the
+relay's auth request. Use a separate QUIC listener for mTLS-authenticated peers.
+
+Generate peer lists so each edge dials all of its region's cores, cores do not
+link within a region, and inter-region cores dial only peers with lower names.
+The relay validates role compatibility and follows the configured links; it
+has no region inventory to generate those lists itself.
+
 ## Topology
 
 List the peers each relay dials. That's the whole topology: a relay dials only
diff --git a/doc/bin/relay/config.md b/doc/bin/relay/config.md
index cc16142e32..8ff3fc12a7 100644
--- a/doc/bin/relay/config.md
+++ b/doc/bin/relay/config.md
@@ -27,8 +27,9 @@ key = "key.pem"
 generate = ["localhost"]             # Or: a self-signed cert for development.
 root = ["peer-ca.pem"]               # Optional: CAs for client certs (mTLS), reported to the auth server.
 
-[listen.tcp]                         # Plaintext qmux over TCP for trusted local workers.
+[listen.tcp]                         # Qmux over TCP. Plaintext by default.
 bind = "127.0.0.1:4444"
+# tls = true                        # tls:// on this port, using listen.tls certificates.
 
 [listen.unix]                        # Plaintext qmux over a Unix socket, gated by peer credentials.
 bind = "/run/moq/internal.sock"
@@ -139,7 +140,8 @@ See [Authentication](/bin/relay/auth).
 
 ```toml
 [cluster]
-connect = ["https://us-east.example.com/?cost=10"]   # Peers to dial. ?cost prices the link, or use {url, cost, egress, token} objects.
+# role = "edge"                                    # Or "core"; unset keeps legacy mesh behavior.
+connect = ["https://us-east.example.com/?cost=10"]   # Peers to dial. ?cost prices the link, or use {url, role, cost, egress, token} objects.
 node = "https://us-west.example.com/"                 # This relay's own URL.
 connect_api = "https://api.example.com/peers"        # Or fetch the peer list (JSON array of URLs and/or objects) live.
 token = "cluster.jwt"                                 # JWT for dials without an inline ?jwt=.
diff --git a/rs/moq-net/benches/origin.rs b/rs/moq-net/benches/origin.rs
index a07846379e..c132abc502 100644
--- a/rs/moq-net/benches/origin.rs
+++ b/rs/moq-net/benches/origin.rs
@@ -477,9 +477,48 @@ fn bench_handoff(c: &mut Criterion) {
 	group.finish();
 }
 
+/// An edge's core-facing announce cursors export local ingress and exclude
+/// core ingress. Sweep route count and core-facing cursors independently.
+fn bench_edge_split_horizon(c: &mut Criterion) {
+	let mut group = c.benchmark_group("origin/edge_split_horizon");
+	for (publishers, cores) in [(100, 2), (1_000, 2), (1_000, 20)] {
+		for peer in [false, true] {
+			let label = if peer { "core_ingress" } else { "local_ingress" };
+			let id = BenchmarkId::new(label, format!("{publishers}p_{cores}c"));
+			group.bench_function(id, |b| {
+				let (producer, _driver) = origin::Producer::new(origin::Config::default());
+				let ingress = if peer {
+					producer.clone().peer()
+				} else {
+					producer.clone()
+				};
+				let _routes: Vec<_> = (0..publishers)
+					.map(|i| ingress.publish(format!("room/{i}"), origin::Route::default()).unwrap())
+					.collect();
+				let mut cursors: Vec<_> = (0..cores).map(|_| producer.consume().local().announced()).collect();
+				for cursor in &mut cursors {
+					while cursor.next().now_or_never().flatten().is_some() {}
+				}
+				b.iter(|| {
+					let broadcast = ingress.publish("room/incoming", origin::Route::default()).unwrap();
+					for cursor in &mut cursors {
+						assert_eq!(cursor.next().now_or_never().flatten().is_some(), !peer);
+					}
+					drop(broadcast);
+					for cursor in &mut cursors {
+						assert_eq!(cursor.next().now_or_never().flatten().is_some(), !peer);
+					}
+				});
+			});
+		}
+	}
+	group.finish();
+}
+
 criterion_group!(
 	benches,
 	bench_announce,
+	bench_edge_split_horizon,
 	bench_announce_mounted,
 	bench_announce_fleet,
 	bench_announce_duplicate,
diff --git a/rs/moq-relay/src/cluster.rs b/rs/moq-relay/src/cluster.rs
index bdfa5d9149..e752eae53d 100644
--- a/rs/moq-relay/src/cluster.rs
+++ b/rs/moq-relay/src/cluster.rs
@@ -29,11 +29,34 @@ pub(crate) const CLUSTER_PATH: &str = "/.cluster";
 /// ticks turn into real conditional GETs.
 const CONNECT_API_POLL_INTERVAL: Duration = Duration::from_secs(30);
 
+/// The relay's place in the cluster topology.
+#[derive(Clone, Copy, Debug, PartialEq, Eq, usage::ValueEnum, serde::Serialize, serde::Deserialize)]
+#[serde(rename_all = "lowercase")]
+#[non_exhaustive]
+pub enum Role {
+	/// Serve end users and dial this region's cores.
+	Edge,
+	/// Serve cluster peers and dial other regions' cores.
+	Core,
+}
+
+impl std::str::FromStr for Role {
+	type Err = anyhow::Error;
+
+	fn from_str(value: &str) -> Result<Self, Self::Err> {
+		match value {
+			"edge" => Ok(Self::Edge),
+			"core" => Ok(Self::Core),
+			_ => anyhow::bail!("cluster role must be edge or core"),
+		}
+	}
+}
+
 /// One cluster peer to dial, as listed in [`Config::connect`] or returned
 /// by a `connect_api` endpoint.
 ///
 /// Accepts a URL string or an object with `url` plus
-/// policy: `cost` prices the link, `egress` declares this relay's own price to
+/// policy: `role` declares the peer's place in the topology, `cost` prices the link, `egress` declares this relay's own price to
 /// the peer, and `token` carries the peer's credential. `egress` defaults to
 /// `cost`; a different effective value is rejected until asymmetric routing
 /// lands. An object `token` behaves exactly like an inline `?jwt=` and is
@@ -46,6 +69,7 @@ pub struct Peer {
 	cost: Option<u64>,
 	egress: Option<u64>,
 	token: Option<String>,
+	role: Option<Role>,
 }
 
 impl std::fmt::Debug for Peer {
@@ -56,6 +80,7 @@ impl std::fmt::Debug for Peer {
 			.field("url", &self.url)
 			.field("cost", &self.cost)
 			.field("egress", &self.egress)
+			.field("role", &self.role)
 			.field("token", &self.token.as_ref().map(|_| "..."))
 			.finish()
 	}
@@ -69,6 +94,7 @@ impl Peer {
 			cost: None,
 			egress: None,
 			token: None,
+			role: None,
 		}
 	}
 
@@ -94,6 +120,17 @@ impl Peer {
 		self.token.as_deref()
 	}
 
+	/// Declare the peer's configured cluster role; an absent role is a legacy peer.
+	pub fn with_role(mut self, role: Role) -> Self {
+		self.role = Some(role);
+		self
+	}
+
+	/// The peer's configured cluster role, or None for a legacy peer.
+	pub fn role(&self) -> Option<Role> {
+		self.role
+	}
+
 	/// Price the link this peer is dialed on.
 	pub fn with_cost(mut self, cost: u64) -> Self {
 		self.cost = Some(cost);
@@ -134,6 +171,8 @@ struct PeerObject {
 	egress: Option<u64>,
 	#[serde(default)]
 	token: Option<String>,
+	#[serde(default)]
+	role: Option<Role>,
 }
 
 /// A bare string is the URL form; a map is the object form. Dispatching on the
@@ -150,7 +189,7 @@ impl<'de> serde::Deserialize<'de> for Peer {
 			type Value = Peer;
 
 			fn expecting(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-				f.write_str("a peer URL string or an object with url, cost, egress, token")
+				f.write_str("a peer URL string or an object with url, role, cost, egress, token")
 			}
 
 			fn visit_str<E: serde::de::Error>(self, url: &str) -> Result<Peer, E> {
@@ -165,6 +204,7 @@ impl<'de> serde::Deserialize<'de> for Peer {
 					cost: object.cost,
 					egress: object.egress,
 					token: object.token,
+					role: object.role,
 				})
 			}
 		}
@@ -178,7 +218,7 @@ impl serde::Serialize for Peer {
 	where
 		S: serde::Serializer,
 	{
-		if self.cost.is_none() && self.egress.is_none() && self.token.is_none() {
+		if self.cost.is_none() && self.egress.is_none() && self.token.is_none() && self.role.is_none() {
 			return serializer.serialize_str(&self.url);
 		}
 		use serde::ser::SerializeStruct as _;
@@ -192,6 +232,9 @@ impl serde::Serialize for Peer {
 		if self.token.is_some() {
 			len += 1;
 		}
+		if self.role.is_some() {
+			len += 1;
+		}
 		let mut state = serializer.serialize_struct("Peer", len)?;
 		state.serialize_field("url", &self.url)?;
 		if let Some(cost) = &self.cost {
@@ -203,6 +246,9 @@ impl serde::Serialize for Peer {
 		if let Some(token) = &self.token {
 			state.serialize_field("token", token)?;
 		}
+		if let Some(role) = self.role {
+			state.serialize_field("role", &role)?;
+		}
 		state.end()
 	}
 }
@@ -238,6 +284,12 @@ impl DialTarget {
 	/// defaults to `cost`; anything else is rejected, never ignored.
 	fn from_peer(peer: &Peer) -> anyhow::Result<Self> {
 		let mut url = peer_url(&peer.url)?;
+		if url.scheme() == "tls" {
+			anyhow::ensure!(
+				url.host().is_some() && url.port().is_some(),
+				"a tls:// cluster peer needs a host and explicit port"
+			);
+		}
 		let has_cost = url.query_pairs().any(|(key, _)| key == "cost");
 		let has_jwt = url.query_pairs().any(|(key, value)| key == "jwt" && !value.is_empty());
 		if peer.cost.is_some() || peer.egress.is_some() {
@@ -522,6 +574,23 @@ impl DialMap {
 	}
 }
 
+fn validate_links(role: Option<Role>, peers: &[Peer]) -> anyhow::Result<()> {
+	for peer in peers {
+		match role {
+			Some(Role::Edge) => {
+				anyhow::ensure!(peer.role == Some(Role::Core), "an edge must dial explicit cores only");
+				anyhow::ensure!(
+					peer_url(&peer.url)?.scheme() == "tls",
+					"an edge must dial cores over tls:// qmux"
+				);
+			}
+			Some(Role::Core) => anyhow::ensure!(peer.role != Some(Role::Edge), "a core must never dial an edge"),
+			None => {}
+		}
+	}
+	Ok(())
+}
+
 /// Configuration for relay clustering.
 ///
 /// [`Self::connect`] / [`Self::connect_api`] list peers to dial, and
@@ -537,6 +606,10 @@ impl DialMap {
 #[serde(default, deny_unknown_fields)]
 #[non_exhaustive]
 pub struct Config {
+	/// Run as an edge or core; unset preserves the legacy mesh.
+	#[usage(long = "cluster-role", env = "MOQ_CLUSTER_ROLE", setting = "cluster.role")]
+	pub role: Option<Role>,
+
 	/// Fixed origin (hop) id for this relay, identifying it in the hop chains
 	/// carried on each broadcast for loop detection and shortest-path routing.
 	///
@@ -956,6 +1029,18 @@ pub struct Cluster {
 }
 
 impl Cluster {
+	pub(crate) fn authorize_peer(&self, peer: bool) -> anyhow::Result<()> {
+		anyhow::ensure!(
+			self.config.role != Some(Role::Core) || peer,
+			"core relays serve cluster peers only"
+		);
+		anyhow::ensure!(
+			self.config.role != Some(Role::Edge) || !peer,
+			"edges do not accept inbound cluster links"
+		);
+		Ok(())
+	}
+
 	/// The origin ID used by this relay on the wire.
 	pub fn id(&self) -> u64 {
 		self.origin.hop().id()
@@ -990,6 +1075,7 @@ impl Cluster {
 		// Reject a repeated static identity with conflicting policy up front,
 		// matching the `--cluster-connect-api` validation, instead of silently
 		// keeping the first entry when dials spawn.
+		validate_links(config.role, &config.connect)?;
 		parse_peer_list(config.connect.clone(), None).context("invalid --cluster-connect peer list")?;
 		let mut origin_config = origin::Config::new(id);
 		if let Some(cache) = cache {
@@ -1213,6 +1299,10 @@ impl Cluster {
 		let lan = self.lan();
 		#[cfg(feature = "cluster-lan")]
 		self.config.lan.validate()?;
+		anyhow::ensure!(
+			!lan || self.config.role.is_none(),
+			"cluster LAN discovery requires legacy mesh mode; role-aware peer lists must be configured explicitly"
+		);
 		#[cfg(feature = "cluster-lan")]
 		if lan {
 			let advertise = self.advertise.as_ref();
@@ -1538,13 +1628,14 @@ impl Cluster {
 		// Dedupe against the shared dial map (and filter out self) on stable identity,
 		// while retaining the full dial configuration so a cost or credential update
 		// replaces the existing session.
-		let desired = match parse_peer_list(list, node.as_deref()) {
-			Ok(desired) => desired,
-			Err(err) => {
-				tracing::warn!(%err, "invalid cluster.connect_api peer list; keeping current peers");
-				return;
-			}
-		};
+		let desired =
+			match validate_links(self.config.role, &list).and_then(|()| parse_peer_list(list, node.as_deref())) {
+				Ok(desired) => desired,
+				Err(err) => {
+					tracing::warn!(%err, "invalid cluster.connect_api peer list; keeping current peers");
+					return;
+				}
+			};
 
 		dialed.reconcile_api(&desired, |target| {
 			tracing::info!(peer = %target.key, "cluster.connect_api peer; dialing");
@@ -1653,8 +1744,12 @@ impl Cluster {
 		// The peer's routes entered the cluster elsewhere. A peer that predates the
 		// hidden opt-in still discovers our hidden routes; see `connection::authorize`.
 		let origin = self.origin.clone().peer();
+		let mut publish = origin.consume().with_hidden(true);
+		if self.config.role == Some(Role::Edge) {
+			publish = publish.local();
+		}
 		let mut client = client
-			.with_publisher(origin.consume().with_hidden(true))
+			.with_publisher(publish)
 			.with_subscriber(origin)
 			.with_stats(self.stats.tier(self.cluster_tier()).session(""));
 		if let Some(cost) = cost {
@@ -2054,7 +2149,123 @@ mod tests {
 		assert_eq!(cluster.cluster_tier(), Tier::new("region/sjc"));
 	}
 
-	/// Stand-in dial task: never makes progress, exposes an AbortHandle.
+	#[test]
+	fn role_links_refuse_incompatible_topologies() {
+		let core = Peer::new("tls://core.example:4443").with_role(Role::Core);
+		let edge = Peer::new("tls://edge.example:4443").with_role(Role::Edge);
+		let legacy = Peer::new("https://legacy.example");
+		assert!(validate_links(Some(Role::Edge), std::slice::from_ref(&core)).is_ok());
+		assert!(validate_links(Some(Role::Edge), std::slice::from_ref(&edge)).is_err());
+		assert!(validate_links(Some(Role::Edge), std::slice::from_ref(&legacy)).is_err());
+		assert!(
+			validate_links(
+				Some(Role::Edge),
+				&[Peer::new("https://core.example").with_role(Role::Core)]
+			)
+			.is_err()
+		);
+		assert!(validate_links(Some(Role::Core), std::slice::from_ref(&edge)).is_err());
+		assert!(validate_links(Some(Role::Core), &[core, legacy]).is_ok());
+		assert!(validate_links(None, &[edge]).is_ok());
+	}
+
+	#[test]
+	fn peer_role_survives_serialization() {
+		let peer = Peer::new("tls://core.example:4443").with_role(Role::Core);
+		let json = serde_json::to_string(&peer).unwrap();
+		assert_eq!(serde_json::from_str::<Peer>(&json).unwrap(), peer);
+		assert_eq!(
+			serde_json::from_str::<Peer>(r#""https://legacy.example""#)
+				.unwrap()
+				.role(),
+			None
+		);
+		assert!(serde_json::from_str::<Peer>(r#"{"url":"tls://core.example:4443","role":"corre"}"#).is_err());
+	}
+
+	#[tokio::test]
+	async fn role_admission_requires_a_peer_grant_on_cores() {
+		let mut config = Config {
+			role: Some(Role::Core),
+			..Default::default()
+		};
+		let core = new_cluster(config.clone()).unwrap();
+		let patterns: moq_auth::Patterns = [moq_auth::Pattern::all()].into_iter().collect();
+		let mut grant = moq_auth::Grant::new(patterns.clone(), patterns);
+		let token = auth::Token::new("/", &grant);
+		assert!(
+			crate::connection::authorize(&core, &token, None, true, &"tls").is_err(),
+			"a certificate alone must not grant cluster membership"
+		);
+		grant.peer = true;
+		let token = auth::Token::new("/", &grant);
+		assert!(crate::connection::authorize(&core, &token, None, false, &"tls").is_ok());
+		config.role = Some(Role::Edge);
+		let edge = new_cluster(config).unwrap();
+		assert!(crate::connection::authorize(&edge, &token, None, true, &"tls").is_err());
+	}
+
+	#[tokio::test]
+	async fn core_loss_moves_only_its_paths() {
+		tokio::time::pause();
+		let config = Config {
+			role: Some(Role::Edge),
+			..Default::default()
+		};
+		let first = new_cluster(config.clone()).unwrap();
+		let second = new_cluster(config).unwrap();
+		let populate = |edge: &Cluster, core: u64| {
+			(0..64)
+				.map(|path| {
+					let mut route = origin::Route::default();
+					route.hops = vec![Hop::new(7).unwrap(), Hop::new(core).unwrap()].try_into().unwrap();
+					edge.origin
+						.clone()
+						.peer()
+						.publish(format!("room/{path}"), route)
+						.unwrap()
+				})
+				.collect::<Vec<_>>()
+		};
+		let lost_first = populate(&first, 10);
+		let _survivor_first = populate(&first, 11);
+		let lost_second = populate(&second, 10);
+		let _survivor_second = populate(&second, 11);
+		let mut first = first.origin.consume().announced();
+		let mut second = second.origin.consume().announced();
+		let snapshot = |cursor: &mut moq_net::announce::Consumer| {
+			std::iter::from_fn(|| cursor.try_next())
+				.map(|update| {
+					(
+						update.prefix.to_string(),
+						update.route.hops.as_slice().last().unwrap().id(),
+					)
+				})
+				.collect::<HashMap<_, _>>()
+		};
+		let before = snapshot(&mut first);
+		assert_eq!(
+			before,
+			snapshot(&mut second),
+			"both edges must agree on the selected core"
+		);
+		let moved = before.values().filter(|&&core| core == 10).count();
+		assert!(
+			moved > 0 && moved < 64,
+			"the pool must distribute paths across both cores"
+		);
+		drop(lost_first);
+		drop(lost_second);
+		for cursor in [&mut first, &mut second] {
+			for _ in 0..moved {
+				let update = cursor.next().await.unwrap();
+				assert_eq!(before.get(update.prefix.as_str()), Some(&10));
+				assert_eq!(update.route.hops.as_slice().last().unwrap().id(), 11);
+			}
+			assert!(cursor.try_next().is_none(), "paths on the surviving core must not move");
+		}
+	}
+
 	fn placeholder_handle() -> AbortHandle {
 		tokio::spawn(std::future::pending::<()>()).abort_handle()
 	}
diff --git a/rs/moq-relay/src/connection.rs b/rs/moq-relay/src/connection.rs
index 96eb782d95..f92acb5e65 100644
--- a/rs/moq-relay/src/connection.rs
+++ b/rs/moq-relay/src/connection.rs
@@ -209,6 +209,7 @@ pub(crate) fn authorize(
 	cluster_peer: bool,
 	transport: &dyn std::fmt::Display,
 ) -> anyhow::Result<Grants> {
+	cluster.authorize_peer(token.peer)?;
 	let publish = cluster.publisher(token);
 	let subscribe = cluster.subscriber(token);
 
diff --git a/rs/moq-relay/src/settings.rs b/rs/moq-relay/src/settings.rs
index ad212f27c0..96d6b7df6a 100644
--- a/rs/moq-relay/src/settings.rs
+++ b/rs/moq-relay/src/settings.rs
@@ -66,6 +66,8 @@ struct Runtime {
 #[derive(usage::Config)]
 #[usage(prefix = "cluster")]
 struct Cluster {
+	#[usage(env = "MOQ_CLUSTER_ROLE", cli("--cluster-role"))]
+	role: Option<String>,
 	#[usage(env = "MOQ_CLUSTER_ID", cli("--cluster-id"))]
 	id: Option<u64>,
 
diff --git a/rs/moq-relay/src/web.rs b/rs/moq-relay/src/web.rs
index 8c8b6dd8db..85b4faa1bd 100644
--- a/rs/moq-relay/src/web.rs
+++ b/rs/moq-relay/src/web.rs
@@ -870,7 +870,12 @@ async fn admit_http(
 	request.server_name = request_host(uri, headers);
 	request.remote = Some(remote.0);
 	request.tls = mtls.and_then(|Extension(MtlsPeer(identity))| auth::peer(&identity));
-	state.auth.admit(request).await
+	let lease = state.auth.admit(request).await?;
+	state
+		.cluster
+		.authorize_peer(lease.token().peer)
+		.map_err(|_| auth::Error::Refused)?;
+	Ok(lease)
 }
 
 /// Serve the announced broadcasts for a given prefix.
diff --git a/rs/moq-relay/tests/cluster_roles.rs b/rs/moq-relay/tests/cluster_roles.rs
new file mode 100644
index 0000000000..33d4715a7f
--- /dev/null
+++ b/rs/moq-relay/tests/cluster_roles.rs
@@ -0,0 +1,208 @@
+//! Edge/core admission and split horizon over encrypted qmux links.
+
+use std::time::Duration;
+
+use moq_net::{origin, stats};
+use moq_relay::{
+	Config, Relay,
+	cluster::{Peer, Role},
+};
+
+const TIMEOUT: Duration = Duration::from_secs(15);
+
+struct Node {
+	url: url::Url,
+	origin: origin::Producer,
+	stats: stats::Registry,
+	trigger: moq_relay::shutdown::Trigger,
+	run: tokio::task::JoinHandle<anyhow::Result<()>>,
+	auth: tokio::task::JoinHandle<()>,
+}
+
+impl Node {
+	async fn start(id: u64, role: Role, peers: &[&Node]) -> Self {
+		let mut config = Config::default();
+		config.listen.tcp.bind = Some("127.0.0.1:0".parse().unwrap());
+		config.listen.tcp.tls = Some(true);
+		config.listen.tls.generate = vec!["localhost".into()];
+		config.connect.tls.insecure = Some(true);
+		config.cluster.id = Some(id);
+		config.cluster.role = Some(role);
+		config.cluster.connect = peers
+			.iter()
+			.map(|peer| Peer::new(peer.url.as_str()).with_role(Role::Core).with_token("cluster"))
+			.collect();
+		config.stats.enabled = true;
+		config.stats.node = Some(id.to_string());
+		config.drain_timeout = Duration::ZERO;
+		let mut relay = Relay::load(config).await.unwrap();
+		let addr = relay.tcp_addr().unwrap();
+		let url = format!("tls://localhost:{}/", addr.port()).parse().unwrap();
+		let origin = relay.cluster().origin.clone();
+		let stats = relay.cluster().stats.clone();
+		let trigger = relay.shutdown_trigger().clone();
+		let ready = relay.ready();
+		let mut admissions = relay.admissions().unwrap();
+		let auth = tokio::spawn(async move {
+			while let Some(admission) = admissions.next().await {
+				let patterns = [moq_auth::Pattern::all()].into_iter().collect::<moq_auth::Patterns>();
+				let mut grant = moq_auth::Grant::new(patterns.clone(), patterns);
+				grant.peer = admission.request.query.as_deref().is_some_and(|query| {
+					url::form_urlencoded::parse(query.as_bytes()).any(|(key, value)| key == "jwt" && value == "cluster")
+				});
+				admission.grant(moq_auth::lease::Consumer::fixed(grant));
+			}
+		});
+		let run = tokio::spawn(relay.run());
+		ready.wait().await.unwrap();
+		Self {
+			url,
+			origin,
+			stats,
+			trigger,
+			run,
+			auth,
+		}
+	}
+
+	async fn stop(self) {
+		self.trigger.start();
+		self.run.await.unwrap().unwrap();
+		self.auth.abort();
+	}
+}
+
+fn client() -> moq_tokio::Client {
+	let mut config = moq_tokio::connect::Config::default();
+	config.tls.insecure = Some(true);
+	config.once = Some(true);
+	config.init(Default::default()).unwrap().with_reconnect(false)
+}
+
+async fn subscriber(node: &Node) -> (moq_tokio::Connection, origin::Consumer) {
+	let origin = moq_tokio::origin::spawn();
+	let consumer = origin.consume();
+	let connection = client().with_subscriber(origin).connect(node.url.clone());
+	let connection = connection.established().await.unwrap();
+	(connection, consumer)
+}
+
+// The native multi-transport client makes these futures large, as in hidden_cluster.
+fn run<F: Future<Output = ()> + Send + 'static>(test: F) {
+	std::thread::Builder::new()
+		.stack_size(32 * 1024 * 1024)
+		.spawn(move || {
+			let _ = rustls::crypto::aws_lc_rs::default_provider().install_default();
+			tokio::runtime::Builder::new_current_thread()
+				.enable_all()
+				.build()
+				.unwrap()
+				.block_on(async {
+					tokio::time::timeout(TIMEOUT, test)
+						.await
+						.expect("cluster role test timed out");
+				});
+		})
+		.unwrap()
+		.join()
+		.unwrap();
+}
+
+#[test]
+fn core_refuses_end_users() {
+	run(async {
+		let core = Node::start(1, Role::Core, &[]).await;
+		let connection = client().connect(core.url.clone());
+		assert!(connection.closed().await.is_err());
+		core.stop().await;
+	});
+}
+
+#[test]
+fn edge_never_connects_cores_through_itself() {
+	run(async {
+		let first = Node::start(1, Role::Core, &[]).await;
+		let second = Node::start(2, Role::Core, &[]).await;
+		let edge = Node::start(3, Role::Edge, &[&first, &second]).await;
+		let broadcast = first.origin.create_broadcast("private").unwrap();
+		broadcast.announce(Default::default()).unwrap();
+		let marker = second.origin.create_broadcast("second-ready").unwrap();
+		marker.announce(Default::default()).unwrap();
+		let consumer = edge.origin.consume();
+		consumer.routed("private").await.unwrap();
+		consumer.routed("second-ready").await.unwrap();
+		// Both core links have delivered a route. On real sockets a negative
+		// assertion needs a bounded observation window rather than mocked time.
+		assert!(
+			tokio::time::timeout(Duration::from_secs(1), second.origin.consume().routed("private"))
+				.await
+				.is_err()
+		);
+		drop(broadcast);
+		drop(marker);
+		edge.stop().await;
+		second.stop().await;
+		first.stop().await;
+	});
+}
+
+#[test]
+fn two_edges_share_one_cross_region_subscription() {
+	run(async {
+		let far = Node::start(1, Role::Core, &[]).await;
+		let first = Node::start(2, Role::Core, &[&far]).await;
+		let second = Node::start(3, Role::Core, &[&far]).await;
+		let edge_a = Node::start(4, Role::Edge, &[&first, &second]).await;
+		let edge_b = Node::start(5, Role::Edge, &[&first, &second]).await;
+		let broadcast = far.origin.create_broadcast("shared").unwrap();
+		broadcast.announce(Default::default()).unwrap();
+		let track = broadcast.create_track("video", None).unwrap();
+		first.origin.consume().routed("shared").await.unwrap();
+		second.origin.consume().routed("shared").await.unwrap();
+		let (_a, a) = subscriber(&edge_a).await;
+		let (_b, b) = subscriber(&edge_b).await;
+		let a = a.routed_broadcast("shared").await.unwrap();
+		let b = b.routed_broadcast("shared").await.unwrap();
+		let mut a = a.track("video").unwrap().subscribe(None).await.unwrap();
+		let mut b = b.track("video").unwrap().subscribe(None).await.unwrap();
+		let mut group = track.append_group().unwrap();
+		group
+			.write_frame(moq_net::Timestamp::ZERO, b"payload".as_ref())
+			.unwrap();
+		group.finish().unwrap();
+		a.recv_group()
+			.await
+			.unwrap()
+			.unwrap()
+			.read_frame()
+			.await
+			.unwrap()
+			.unwrap();
+		b.recv_group()
+			.await
+			.unwrap()
+			.unwrap()
+			.read_frame()
+			.await
+			.unwrap()
+			.unwrap();
+		let mut report = stats::Report::default();
+		far.stats.report(&mut report);
+		let shared = report
+			.traffic
+			.iter()
+			.find(|entry| entry.path.as_str() == "shared")
+			.unwrap();
+		assert_eq!(
+			shared.publisher.subscriptions_started, 1,
+			"both edges must choose the same regional core"
+		);
+		drop(a);
+		drop(b);
+		edge_b.stop().await;
+		edge_a.stop().await;
+		second.stop().await;
+		first.stop().await;
+		far.stop().await;
+	});
+}
diff --git a/rs/moq-tokio/src/client.rs b/rs/moq-tokio/src/client.rs
index 686fd6c6df..8b93783879 100644
--- a/rs/moq-tokio/src/client.rs
+++ b/rs/moq-tokio/src/client.rs
@@ -62,12 +62,14 @@ pub struct Client {
 	pub(crate) reconnect: bool,
 	pub(crate) backoff: Backoff,
 	pub(crate) goaway: Goaway,
-	/// The resolved Happy Eyeballs timings, used by the `tcp://` dial here; the
+	/// The resolved Happy Eyeballs timings, used by TCP dials here; the
 	/// QUIC backend captures its own copy from the config.
 	#[cfg(feature = "tcp")]
 	failover_delay: std::time::Duration,
 	#[cfg(feature = "tcp")]
 	resolution_delay: std::time::Duration,
+	#[cfg(feature = "tcp")]
+	tcp_tls: crate::tls::Connect,
 	#[cfg(feature = "websocket")]
 	websocket: crate::websocket::Config,
 	/// The TLS server name override used by the WebSocket fallback.
@@ -141,6 +143,8 @@ impl Client {
 				feature = "uds"
 			))]
 			versions,
+			#[cfg(feature = "tcp")]
+			tcp_tls: config.tls.clone(),
 			connect: config.url,
 			timeout,
 			reconnect: !config.once.unwrap_or(false),
@@ -384,6 +388,19 @@ impl Client {
 			return Ok(connect_session(&moq, crate::transport::Session::new(session)).await?);
 		}
 
+		#[cfg(feature = "tcp")]
+		if url.scheme() == "tls" {
+			let session = crate::tcp::connect_tls(
+				url,
+				&self.versions.alpns(),
+				&self.tcp_tls,
+				self.failover_delay,
+				self.resolution_delay,
+			)
+			.await?;
+			return Ok(connect_session(&moq, crate::transport::Session::new(session)).await?);
+		}
+
 		// Unix domain socket (qmux, no TLS). Same-host only; the server can
 		// authenticate us by uid/gid via SO_PEERCRED.
 		#[cfg(all(feature = "uds", unix))]
@@ -547,7 +564,7 @@ fn setup_path(url: &Url) -> Option<String> {
 			.filter(|path| !path.is_empty()),
 		// Raw QUIC and qmux over TCP negotiate an ALPN and nothing else, so the whole
 		// request target travels in the SETUP.
-		"moqt" | "moql" | "tcp" => request_target(url),
+		"moqt" | "moql" | "tcp" | "tls" => request_target(url),
 		_ => None,
 	}
 }
@@ -785,6 +802,8 @@ mod tests {
 			("tcp://localhost:4443/room", Some("/room")),
 			("tcp://localhost:4443/room?jwt=abc", Some("/room?jwt=abc")),
 			("tcp://localhost:4443", None),
+			("tls://localhost:4443/room?jwt=abc", Some("/room?jwt=abc")),
+			("tls://localhost:4443", None),
 			// Raw QUIC: the URL is ours alone, so the path and query have to ride the
 			// SETUP or the server never sees them.
 			("moqt://relay.example.com/anon", Some("/anon")),
diff --git a/rs/moq-tokio/src/connect.rs b/rs/moq-tokio/src/connect.rs
index 3dbbe6a889..1b6fdf1fbe 100644
--- a/rs/moq-tokio/src/connect.rs
+++ b/rs/moq-tokio/src/connect.rs
@@ -507,7 +507,7 @@ pub struct Config {
 	/// When DNS returns multiple addresses, attempts alternate between IPv6 and
 	/// IPv4, each starting this long after the previous one (or immediately when
 	/// it fails), and the first connection to complete wins. `0s` dials every
-	/// address at once. Defaults to 250ms. Applies to the QUIC and `tcp://` dials.
+	/// address at once. Defaults to 250ms. Applies to QUIC, `tcp://`, and `tls://` dials.
 	///
 	/// This staggers the attempts within one [`crate::Client::connect`]; [`Self::timeout`]
 	/// bounds that call as a whole.
diff --git a/rs/moq-tokio/src/connection.rs b/rs/moq-tokio/src/connection.rs
index 1a5698049c..b85a73d821 100644
--- a/rs/moq-tokio/src/connection.rs
+++ b/rs/moq-tokio/src/connection.rs
@@ -262,7 +262,7 @@ impl Redirect {
 /// Unknown schemes rank lowest, so a forgotten classification is refused.
 fn scheme_tier(scheme: &str) -> u8 {
 	match scheme {
-		"https" | "moqt" | "moql" | "wss" | "iroh" => 2,
+		"https" | "moqt" | "moql" | "wss" | "iroh" | "tls" => 2,
 		"tcp" | "ws" | "http" => 1,
 		// `unix` lands here deliberately: local IPC is not an upgrade over a
 		// network transport, it is a different reachability class (see `is_local`).
diff --git a/rs/moq-tokio/src/error.rs b/rs/moq-tokio/src/error.rs
index 209306d96a..482d178cff 100644
--- a/rs/moq-tokio/src/error.rs
+++ b/rs/moq-tokio/src/error.rs
@@ -133,8 +133,8 @@ pub enum Error {
 	#[error("Iroh support is not enabled")]
 	IrohDisabled,
 
-	/// A client certificate was configured, but this QUIC backend can't do mTLS.
-	#[error("tls.root (mTLS) is not supported by the selected QUIC backend")]
+	/// Client certificate roots were configured, but this transport cannot expose the peer identity.
+	#[error("tls.root (mTLS) is not supported by the selected transport")]
 	MtlsUnsupported,
 
 	/// A worker group was asked for more members than the connection ID's one-byte
diff --git a/rs/moq-tokio/src/lib.rs b/rs/moq-tokio/src/lib.rs
index f520b86565..1d44ceea78 100644
--- a/rs/moq-tokio/src/lib.rs
+++ b/rs/moq-tokio/src/lib.rs
@@ -4,7 +4,7 @@
 //! - WebTransport (HTTP/3)
 //! - Raw QUIC (with ALPN negotiation)
 //! - WebSocket (fallback via [web-transport-ws](https://crates.io/crates/web-transport-ws))
-//! - Plain TCP via the `tcp://` scheme (qmux, no TLS; requires `tcp` feature)
+//! - TCP via the `tcp://` and `tls://` schemes (qmux; requires `tcp` feature)
 //! - Unix domain socket via the `unix://` scheme (qmux, peer-credential aware; requires `uds` feature, unix-only)
 //! - Iroh P2P (requires `iroh` feature)
 //!
diff --git a/rs/moq-tokio/src/listen.rs b/rs/moq-tokio/src/listen.rs
index cbd88f64bd..c19f97de80 100644
--- a/rs/moq-tokio/src/listen.rs
+++ b/rs/moq-tokio/src/listen.rs
@@ -100,7 +100,7 @@ pub struct Config {
 	#[usage(skip)]
 	pub(crate) listen: Option<String>,
 
-	/// Plaintext qmux TCP listener (`--listen-tcp-bind`, no TLS). Requires the
+	/// Qmux TCP listener (`--listen-tcp-bind`, optional `--listen-tcp-tls`). Requires the
 	/// `tcp` feature.
 	#[cfg(feature = "tcp")]
 	#[usage(flatten)]
diff --git a/rs/moq-tokio/src/server.rs b/rs/moq-tokio/src/server.rs
index 07e5977533..3bb4b3b429 100644
--- a/rs/moq-tokio/src/server.rs
+++ b/rs/moq-tokio/src/server.rs
@@ -280,6 +280,17 @@ impl Server {
 		// Resolve here rather than in `init`, so a caller that builds the config by hand
 		// gets the released spellings folded in too.
 		config.validate()?;
+		#[cfg(feature = "tcp")]
+		if config.tcp.tls.unwrap_or(false) && config.tcp.bind.is_none() {
+			return Err(Error::NoBackend("--listen-tcp-tls requires --listen-tcp-bind"));
+		}
+
+		#[cfg(feature = "tcp")]
+		if config.tcp.tls.unwrap_or(false) && !config.tls.root.is_empty() {
+			// Qmux verifies the certificate but its TLS accept API discards the
+			// identity, so an auth server cannot authorize that mTLS credential.
+			return Err(Error::MtlsUnsupported);
+		}
 
 		let versions = config.versions();
 		let timeout = config.resolved_timeout();
@@ -327,9 +338,34 @@ impl Server {
 		// Collect the configured stream listeners (at most one TCP, one Unix).
 		#[cfg(any(feature = "tcp", all(feature = "uds", unix)))]
 		let mut stream_binds = Vec::new();
+		#[cfg(all(feature = "tcp", feature = "watch", feature = "_certs"))]
+		let mut tcp_reload = None;
 		#[cfg(feature = "tcp")]
 		if let Some(addr) = config.tcp.bind.filter(|_| build_streams) {
-			stream_binds.push(StreamBind::Tcp(addr));
+			let tls = if config.tcp.tls.unwrap_or(false) {
+				#[cfg(feature = "_certs")]
+				{
+					let alpn = stream_versions(&versions)
+						.alpns()
+						.iter()
+						.map(|alpn| format!("qmux-01.{alpn}").into_bytes())
+						.collect();
+					#[cfg(feature = "watch")]
+					{
+						let reload = config.tls.server_config_reloading(alpn)?;
+						let tls = reload.config();
+						tcp_reload = Some(reload);
+						Some(tls)
+					}
+					#[cfg(not(feature = "watch"))]
+					Some(config.tls.server_config(alpn)?)
+				}
+				#[cfg(not(feature = "_certs"))]
+				return Err(Error::NoBackend("--listen-tcp-tls requires a crypto provider"));
+			} else {
+				None
+			};
+			stream_binds.push(StreamBind::Tcp(addr, tls));
 		}
 		#[cfg(all(feature = "uds", unix))]
 		if let Some(path) = config.unix.bind.clone().filter(|_| build_streams) {
@@ -347,6 +383,13 @@ impl Server {
 			unix_allow,
 		);
 
+		#[cfg(all(feature = "tcp", feature = "watch", feature = "_certs"))]
+		let streams = {
+			let mut streams = streams;
+			streams._tcp_reload = tcp_reload;
+			streams
+		};
+
 		// The endpoint was bound before this server's versions were known, so it
 		// accepts nothing until they name its ALPNs.
 		#[cfg(feature = "iroh")]
@@ -989,7 +1032,7 @@ fn stream_versions(base: &moq_net::Versions) -> moq_net::Versions {
 #[cfg(any(feature = "tcp", all(feature = "uds", unix)))]
 enum StreamBind {
 	#[cfg(feature = "tcp")]
-	Tcp(net::SocketAddr),
+	Tcp(net::SocketAddr, Option<std::sync::Arc<rustls::ServerConfig>>),
 	#[cfg(all(feature = "uds", unix))]
 	Unix(PathBuf),
 }
@@ -1009,7 +1052,9 @@ impl StreamBind {
 	fn name(&self) -> &'static str {
 		match self {
 			#[cfg(feature = "tcp")]
-			Self::Tcp(_) => "tcp",
+			Self::Tcp(_, Some(_)) => "tls",
+			#[cfg(feature = "tcp")]
+			Self::Tcp(_, None) => "tcp",
 			#[cfg(all(feature = "uds", unix))]
 			Self::Unix(_) => "unix",
 		}
@@ -1025,6 +1070,8 @@ impl StreamBind {
 /// stopped when the [`Listener`] closes or drops, so no socket lingers.
 #[cfg(any(feature = "tcp", all(feature = "uds", unix)))]
 struct StreamListeners {
+	#[cfg(all(feature = "tcp", feature = "watch", feature = "_certs"))]
+	_tcp_reload: Option<crate::tls::ReloadingServerConfig>,
 	binds: Vec<StreamBind>,
 	/// One per entry in `binds`, in the same order, and created up front rather than
 	/// with the listener: an owner registering these with a metrics endpoint does so
@@ -1057,6 +1104,8 @@ impl StreamListeners {
 			.map(|bind| crate::accept::Health::new(bind.name()))
 			.collect();
 		Self {
+			#[cfg(all(feature = "tcp", feature = "watch", feature = "_certs"))]
+			_tcp_reload: None,
 			binds,
 			health,
 			versions,
@@ -1086,13 +1135,14 @@ impl StreamListeners {
 			let alpns = self.versions.alpns();
 			match bind {
 				#[cfg(feature = "tcp")]
-				StreamBind::Tcp(addr) => {
-					if !addr.ip().is_loopback() {
+				StreamBind::Tcp(addr, tls) => {
+					if tls.is_none() && !addr.ip().is_loopback() {
 						tracing::warn!(%addr, "tcp listener bound to a non-loopback address; qmux is UNENCRYPTED, ensure the network is trusted");
 					}
 					let listener = crate::tcp::Listener::bind(addr)
 						.await?
 						.with_protocols(alpns)
+						.with_tls(tls)
 						.with_accept_health(health);
 					let local = listener.local_addr()?;
 					tracing::info!(addr = %local, "listening (tcp)");
@@ -1338,7 +1388,7 @@ pub enum Transport {
 	Iroh,
 	/// A WebSocket connection using qmux framing.
 	WebSocket,
-	/// A plaintext TCP connection using qmux framing.
+	/// A TCP connection using qmux framing, with optional TLS.
 	Tcp,
 	/// A Unix domain socket using qmux framing.
 	Unix,
diff --git a/rs/moq-tokio/src/settings.rs b/rs/moq-tokio/src/settings.rs
index bec2ac4d22..fbf0ef3ffc 100644
--- a/rs/moq-tokio/src/settings.rs
+++ b/rs/moq-tokio/src/settings.rs
@@ -70,6 +70,8 @@ struct ListenTls {
 #[derive(usage::Config)]
 #[usage(prefix = "listen.tcp")]
 struct ListenTcp {
+	#[usage(env = "MOQ_LISTEN_TCP_TLS", cli("--listen-tcp-tls"))]
+	tls: Option<bool>,
 	#[usage(env = "MOQ_LISTEN_TCP_BIND", cli("--listen-tcp-bind"))]
 	bind: Option<String>,
 }
diff --git a/rs/moq-tokio/src/tcp.rs b/rs/moq-tokio/src/tcp.rs
index b275388633..e79d8b0656 100644
--- a/rs/moq-tokio/src/tcp.rs
+++ b/rs/moq-tokio/src/tcp.rs
@@ -1,12 +1,9 @@
-//! Plain-TCP qmux transport, reachable via the `tcp://` URL scheme.
+//! Qmux over TCP, with `tcp://` for plaintext and `tls://` for TLS.
 //!
-//! Runs the QMux wire format directly over TCP with no TLS or WebSocket
-//! framing. There is no transport encryption and no authentication, so only
-//! use this on a trusted network (loopback, a private VPC interface, etc.).
-//!
-//! TCP has no TLS handshake, so the application protocol (the moq ALPN) is
-//! negotiated in-band: pass the offered/supported protocols and the resulting
-//! `qmux::Session::protocol()` is populated before connect/accept returns.
+//! Both schemes speak qmux directly without WebSocket framing. Plaintext
+//! `tcp://` is for trusted networks only; `tls://` authenticates and encrypts
+//! with the same trust and certificate settings as the other TLS transports.
+//! Plaintext negotiates the MoQ ALPN in-band; TLS negotiates it in its handshake.
 
 use std::net;
 use url::Url;
@@ -15,11 +12,11 @@ use url::Url;
 /// negotiated) since there's no TLS ALPN to carry it.
 const WIRE_VERSION: qmux::Version = qmux::Version::QMux01;
 
-/// Plaintext-TCP qmux listener settings (no TLS, no UDP).
+/// Qmux TCP listener settings, with optional TLS and no UDP.
 ///
-/// Flattened onto [`crate::listen::Config::tcp`]. TCP carries no peer identity, so
-/// the listener must only be reachable from trusted clients. Bind it to loopback
-/// or a private interface; a non-loopback bind logs a warning but is allowed.
+/// Flattened onto [`crate::listen::Config::tcp`]. Plaintext connections need a
+/// trusted network; a non-loopback plaintext bind logs a warning. Enable `tls`
+/// to encrypt this port using the listen TLS certificate.
 // The derived arg group is named after the struct, so it needs an explicit id to
 // stay unique across the flattened sections.
 #[derive(usage::Args, Clone, Debug, Default, serde::Serialize, serde::Deserialize)]
@@ -27,7 +24,7 @@ const WIRE_VERSION: qmux::Version = qmux::Version::QMux01;
 #[serde(deny_unknown_fields, default)]
 #[non_exhaustive]
 pub struct Config {
-	/// Bind a plaintext qmux TCP listener on this address.
+	/// Bind a qmux TCP listener on this address.
 	#[usage(
 		long = "listen-tcp-bind",
 		name = "listen-tcp-bind",
@@ -37,6 +34,11 @@ pub struct Config {
 	#[serde(default, skip_serializing_if = "Option::is_none")]
 	pub bind: Option<net::SocketAddr>,
 
+	/// Encrypt this TCP listener with the configured listen TLS certificate.
+	#[usage(long = "listen-tcp-tls", env = "MOQ_LISTEN_TCP_TLS", setting = "listen.tcp.tls", default_missing = "true", num_args = 0..=1, require_equals = true)]
+	#[serde(skip_serializing_if = "Option::is_none")]
+	pub tls: Option<bool>,
+
 	/// The released `--server-tcp-bind` spelling and its env var, folded in by
 	/// [`Config::resolved`].
 	#[usage(flatten)]
@@ -76,7 +78,7 @@ impl Config {
 	}
 }
 
-/// Errors specific to the plain-TCP qmux transport.
+/// Errors specific to the qmux TCP transport.
 #[derive(Debug, thiserror::Error)]
 #[non_exhaustive]
 pub enum Error {
@@ -86,11 +88,11 @@ pub enum Error {
 	#[error(transparent)]
 	Io(#[from] std::io::Error),
 
-	/// The `tcp://` URL had no host.
+	/// The qmux TCP URL had no host.
 	#[error("missing hostname")]
 	MissingHostname,
 
-	/// The `tcp://` URL had no port. Unlike `https`, there is no default.
+	/// The qmux TCP URL had no port. Unlike `https`, there is no default.
 	#[error("missing port")]
 	MissingPort,
 
@@ -152,6 +154,43 @@ pub(crate) async fn connect(
 	connect_addrs(candidates, protocols, failover_delay).await
 }
 
+/// Dial encrypted qmux over TCP, racing resolved addresses through the TLS handshake.
+pub(crate) async fn connect_tls(
+	url: Url,
+	protocols: &[&str],
+	config: &crate::tls::Connect,
+	failover_delay: std::time::Duration,
+	resolution_delay: std::time::Duration,
+) -> crate::Result<qmux::Session> {
+	#[cfg(not(any(feature = "aws-lc-rs", feature = "ring")))]
+	if rustls::crypto::CryptoProvider::get_default().is_none() {
+		return Err(crate::Error::NoBackend("tls:// requires a crypto provider"));
+	}
+
+	let host = url.host().ok_or(Error::MissingHostname)?;
+	let port = url.port().ok_or(Error::MissingPort)?;
+	let name = config.host_name.clone().unwrap_or_else(|| match host {
+		url::Host::Domain(name) => name.to_owned(),
+		url::Host::Ipv4(ip) => ip.to_string(),
+		url::Host::Ipv6(ip) => ip.to_string(),
+	});
+	let client = qmux::tls::Client::new(std::sync::Arc::new(config.build()?))
+		.with_protocols(protocols.iter().map(|&alpn| (alpn, &[WIRE_VERSION][..])))
+		.require_protocol();
+	let candidates = crate::resolve::Candidates::resolve(host, port, resolution_delay);
+	Ok(crate::failover::race(candidates, failover_delay, |addr| {
+		let client = client.clone();
+		let name = name.clone();
+		async move {
+			client
+				.connect(addr, &name)
+				.await
+				.map_err(|err| Error::Connect(crate::error::message(err)))
+		}
+	})
+	.await?)
+}
+
 /// Dial `candidates` in Happy Eyeballs order, performing the qmux handshake on
 /// each attempt; the first session to complete wins.
 async fn connect_addrs(
@@ -172,11 +211,12 @@ async fn connect_addrs(
 	.await
 }
 
-/// Listens for incoming plain-TCP qmux connections on a TCP port.
+/// Listen for qmux connections on a TCP port.
 pub struct Listener {
 	listener: tokio::net::TcpListener,
 	protocols: Vec<String>,
 	health: crate::accept::Health,
+	tls: Option<qmux::tls::Server>,
 }
 
 impl Listener {
@@ -187,9 +227,15 @@ impl Listener {
 			listener,
 			protocols: Vec::new(),
 			health: crate::accept::Health::new("tcp"),
+			tls: None,
 		})
 	}
 
+	pub(crate) fn with_tls(mut self, tls: Option<std::sync::Arc<rustls::ServerConfig>>) -> Self {
+		self.tls = tls.map(qmux::tls::Server::new);
+		self
+	}
+
 	/// A live handle to this listener's accept-loop health, for an embedder that
 	/// publishes it (see [`crate::accept`]).
 	pub fn accept_health(&self) -> crate::accept::Health {
@@ -243,11 +289,13 @@ impl Listener {
 		let (stream, addr) = self.accept_socket().await;
 		tracing::debug!(%addr, "accepted TCP connection");
 		let config = qmux::tcp::Config::new(WIRE_VERSION).protocols(self.protocols.iter().map(String::as_str));
+		let tls = self.tls.clone();
 		async move {
-			let session = config
-				.accept(stream)
-				.await
-				.map_err(|err| Error::Accept(crate::error::message(err)))?;
+			let session = match tls {
+				Some(tls) => tls.accept(stream).await,
+				None => config.accept(stream).await,
+			}
+			.map_err(|err| Error::Accept(crate::error::message(err)))?;
 			Ok((session, addr))
 		}
 	}
@@ -276,6 +324,71 @@ mod tests {
 	use std::time::Duration;
 	use web_transport_trait::Session as _;
 
+	#[test]
+	fn tls_listener_refuses_missing_bind_and_mtls_roots() {
+		let mut listen = crate::listen::Config::default();
+		listen.tcp.tls = Some(true);
+		assert!(matches!(
+			listen.clone().init(Default::default()),
+			Err(crate::Error::NoBackend(_))
+		));
+		listen.tcp.bind = Some("127.0.0.1:0".parse().unwrap());
+		listen.tls.root = vec!["client-ca.pem".into()];
+		assert!(matches!(
+			listen.init(Default::default()),
+			Err(crate::Error::MtlsUnsupported)
+		));
+	}
+
+	#[cfg(feature = "_certs")]
+	#[tokio::test]
+	async fn tls_negotiates_qmux_and_refuses_untrusted_certificates() {
+		let _ = rustls::crypto::aws_lc_rs::default_provider().install_default();
+		let mut listen = crate::listen::Config::default();
+		listen.tcp.bind = Some("127.0.0.1:0".parse().unwrap());
+		listen.tcp.tls = Some(true);
+		listen.tls.generate = vec!["localhost".into()];
+		let mut server = listen.init(Default::default()).unwrap().listen().await.unwrap();
+		let addr = server.tcp_local_addr().unwrap();
+		let url: Url = format!("tls://localhost:{}/room?jwt=credential", addr.port())
+			.parse()
+			.unwrap();
+		let accept = tokio::spawn(async move {
+			let request = server.accept().await.unwrap();
+			assert_eq!(request.transport(), crate::server::Transport::Tcp);
+			assert_eq!(request.path(), "/room");
+			assert_eq!(request.query(), Some("jwt=credential"));
+			let session = request.ok().await.unwrap();
+			(server, session)
+		});
+		let untrusted = crate::connect::Config {
+			once: Some(true),
+			..Default::default()
+		};
+		assert!(
+			untrusted
+				.init(Default::default())
+				.unwrap()
+				.with_reconnect(false)
+				.connect(url.clone())
+				.established()
+				.await
+				.is_err()
+		);
+		let mut trusted = crate::connect::Config {
+			once: Some(true),
+			..Default::default()
+		};
+		trusted.tls.insecure = Some(true);
+		let connection = trusted
+			.init(Default::default())
+			.unwrap()
+			.with_reconnect(false)
+			.connect(url);
+		let _connection = connection.established().await.unwrap();
+		let (_server, _session) = accept.await.unwrap();
+	}
+
 	/// End-to-end failover: the preferred candidate blackholes (TEST-NET-1 never
 	/// answers, or is unroutable outright in a sandbox), so the race must fall
 	/// through to the loopback listener within the stagger delay.