diff --git a/quest/m1/README.md b/quest/m1/README.md
index 6b916ef069..8c48c016cb 100644
--- a/quest/m1/README.md
+++ b/quest/m1/README.md
@@ -64,7 +64,6 @@ blocks. The quests that gated m0 lines moved under them.
 - [Subscribe ranges](/quest/m1/subscribe-ranges/README.md) - a lite-07 SUBSCRIBE asks for past and live ranges in either order and replaces FETCH; relays fill misses by range, including over moq-transport
 - [Live media time](/quest/m1/subscribe-live-time.md) - re-scoped against `set_live`: a lite-07 SUBSCRIBE_OK carries the publisher's current media time only if a reader still needs it
 - [Cross-relay bursts](/quest/m1/cross-relay-bursts.md) - bursty small-group tracks cross two relays without lost groups, unanswered FETCHes, or stalls
-- [Fetched heads stay visible](/quest/m1/lite07-head-fetch-arrival.md) - on lite-07, a relay fetching the head of a group it receives mid-group still delivers that group to new subscribers
 - [Untimed lite-07](/quest/m1/lite-untimed.md) - lite-07 carries an untimed track in both languages; lite-05/06 write send time
 - [Finalize moq-lite-07](/quest/m1/lite07-finalize.md) - when the maintainer cuts it, lite-07 negotiates as `moq-lite-07` and the next release ships it
 - [Late lower groups](/quest/m1/lite-late-lower-group.md) - a moq-lite subscriber with a floor receives a group created below the first served one, as moq-transport does
diff --git a/quest/m1/ietf-rejoin-open-group.md b/quest/m1/ietf-rejoin-open-group.md
index 3ef18f8a11..e2aa3d4a98 100644
--- a/quest/m1/ietf-rejoin-open-group.md
+++ b/quest/m1/ietf-rejoin-open-group.md
@@ -21,5 +21,4 @@ Public API: none. Wire: none.
 
 ## Related
 
-- [Fetched heads stay visible](/quest/m1/lite07-head-fetch-arrival.md) - another mid-group join that loses part of a group
 - [Rejoin after the copy goes idle](/quest/m1/test-flakes-2/rejoin-idle-race.md) - a different IETF rejoin bug, serving a stale cache
diff --git a/quest/m1/lite07-finalize.md b/quest/m1/lite07-finalize.md
index 56a91b3312..14015eea2b 100644
--- a/quest/m1/lite07-finalize.md
+++ b/quest/m1/lite07-finalize.md
@@ -69,7 +69,6 @@ is published; older versions are unchanged.
 - [Live media time](/quest/m1/subscribe-live-time.md) - SUBSCRIBE_OK carries the publisher's live media time
 - [Untimed lite-07](/quest/m1/lite-untimed.md) - an untimed track crosses the wire untimed
 - [One route cost](/quest/m1/route-cost.md) - ANNOUNCE carries one cost
-- [Fetched heads stay visible](/quest/m1/lite07-head-fetch-arrival.md) - a lite-07 relay still delivers a group whose head it fetched
 
 ## Related
 
diff --git a/quest/m1/lite07-head-fetch-arrival.md b/quest/m1/lite07-head-fetch-arrival.md
deleted file mode 100644
index 5c434c1eca..0000000000
--- a/quest/m1/lite07-head-fetch-arrival.md
+++ /dev/null
@@ -1,66 +0,0 @@
-# [S] A fetched group head keeps the live group visible on lite-07
-
-## Goal
-
-On moq-lite-07, a relay that fetches the head of a group it is receiving
-mid-group still delivers that group to new arrival-order subscribers, and
-frames the publisher writes into it afterward still reach them. Today a fresh
-subscriber with no start receives nothing, even after the publisher writes more
-frames into the group. lite-05, lite-06, and moq-transport are unaffected.
-
-A long-lived open group (a catalog: snapshot at frame 0, deltas after) is the
-case that hurts: a viewer joining the relay never gets the catalog.
-
-## Plan
-
-Scenario, reproduced on `main` 2704e10e2 for `moq-lite-07-wip` only:
-
-1. Publisher P holds open group G with a snapshot and two deltas. Relay R pulls
-   P; mesh peer M pulls R.
-2. M subscribes to `catalog.json` with a start of `(G, 1)`, then reads its own
-   copy from frame 0, so it fetches G's head through R.
-3. A fresh reader on R subscribes with no start: it receives no group within
-   5 s of virtual time, and a frame P writes into G after 1 s doesn't help.
-
-Test shape (`rs/moq-net/tests`, the mock harness, paused time; run it for every
-version and expect group G starting with `snapshot`):
-
-```rust
-let start = track::Position { group: sequence, frame: 1 };
-let mut sub = peer_remote.track("catalog.json").unwrap()
-	.subscribe(track::Subscription::default().with_start(start)).await.unwrap();
-let mut group = sub.recv_group().await.unwrap().unwrap();
-// The peer reads its headless copy from frame 0: a fetch of G's head through R.
-let _ = tokio::time::timeout(Duration::from_millis(500), group.read_frame()).await;
-
-// A fresh reader on R wants the latest group from its snapshot.
-let mut sub = remote.track("catalog.json").unwrap().subscribe(None).await.unwrap();
-let mut group = tokio::time::timeout(Duration::from_secs(5), sub.recv_group())
-	.await.expect("no group").unwrap().unwrap();
-assert_eq!(group.sequence, sequence);
-assert_eq!(group.read_frame().await.unwrap().unwrap().payload, b"snapshot".as_ref());
-```
-
-Inferred cause, unverified: a lite-07 SUBSCRIBE_OK carries the publisher's
-largest position, so `TrackServe::widen_frame_bounds`
-(`rs/moq-net/src/lite/subscriber.rs`) trusts it and keeps M's mid-group start:
-R's upstream subscription starts G at frame 1. M's fetch reaches
-`TrackState::insert_group_request` (`rs/moq-net/src/model/track.rs`), whose
-`claim_sequence` drops that headless live slot because it can't answer from
-frame 0, then commits the fetched group invisible. G's arrival entry keeps the
-old stamp, so `poll_recv_group` skips it, and the live copy R keeps filling
-from P is no longer in its cache. Settle whether a fetch may replace a live,
-visible slot at all, or must keep it arrival-visible and still writable.
-
-Only bites once lite-07 ships. lite-06 has frame bounds too, but no largest
-(`Version::has_largest`), so `widen_frame_bounds` rounds the start to the head
-of the group and R never holds a headless G. Widening on lite-07 too would hide
-the bug at the cost of what the largest position is for; the fix belongs in the
-cache.
-
-## Related
-
-- [#4829](https://github.com/moq-dev/moq/pull/4829) - `release`'s backport of the lite-06 head widening
-- [The moq.pro mesh runs lite-07](/quest/m3/lite07-mesh.md) - when this starts to bite
-- [Subscribe ranges](/quest/m1/subscribe-ranges/README.md) - lite-07 relays fill misses by range, the same cache seam
-- [Late lower groups](/quest/m1/lite-late-lower-group.md) - another lite group a subscriber silently never sees
diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs
index 95d0acae27..eacd270e3c 100644
--- a/rs/moq-net/src/model/track.rs
+++ b/rs/moq-net/src/model/track.rs
@@ -186,7 +186,8 @@ pub(crate) struct TrackState {
 	// Publisher-produced groups in arrival order as (sequence, stamp), walked by
 	// subscriptions; an entry only resolves while its stamp matches the slot's.
 	// Fetched backfill (`insert_group_request`) is deliberately absent: it is
-	// served by sequence, never replayed to arrival-order subscribers.
+	// served by sequence, never replayed to arrival-order subscribers. A fetched
+	// head of a live group is the exception, served through that group's entry.
 	arrival: VecDeque<(u64, u32)>,
 
 	// Eviction order under memory pressure as (sequence, stamp): every cached
@@ -308,6 +309,83 @@ struct Slot {
 	// Whether this incarnation came from the live publisher and can replace older
 	// subscription content. Fetch-only backfill stays cached but never anchors drift.
 	visible: bool,
+
+	// A fetched copy of a live `group` that the feed started past the frames a fetch
+	// asked for. It runs to the end of the group too, so it is served in the live
+	// group's place while it lasts; `group` stays the slot, still written by the feed,
+	// so an abandoned fetch leaves the group servable rather than gone. Boxed: rare.
+	head: Option<Box<group::Producer>>,
+}
+
+impl Slot {
+	/// Whether the live feed's group still holds this slot.
+	fn is_live(&self) -> bool {
+		self.visible && !self.group.is_aborted()
+	}
+
+	/// The copy to hand readers: the fetched head while it lasts, else the live group.
+	fn serving(&self) -> &group::Producer {
+		match &self.head {
+			Some(head) if !head.is_aborted() => head,
+			_ => &self.group,
+		}
+	}
+
+	/// Every copy this slot holds, the live group first.
+	fn copies(&self) -> impl Iterator<Item = &group::Producer> {
+		std::iter::once(&self.group).chain(self.head.as_deref())
+	}
+
+	/// Whether every copy is gone, so the slot has nothing left to serve.
+	fn is_aborted(&self) -> bool {
+		self.serving().is_aborted()
+	}
+
+	/// The mean access over copies still held, weighed the way each one samples the
+	/// pool's average: a maximum would sit above the mean forever, so a lone slot with
+	/// two copies could never be evicted.
+	fn cache_accessed(&self) -> u64 {
+		let (sum, count) = self
+			.copies()
+			.filter(|group| !group.is_aborted())
+			.fold((0u128, 0u128), |(sum, count), group| {
+				(sum + group.cache_accessed() as u128, count + 1)
+			});
+		(sum / count.max(1)) as u64
+	}
+
+	/// The bytes every copy holds, all freed together.
+	fn cache_size(&self) -> u64 {
+		self.copies().map(group::Producer::cache_size).sum()
+	}
+
+	/// Whether every copy still held sat idle past the scan's expiry window. Asks
+	/// each one, so a cleanup scan dates them all.
+	fn is_expired(&self, scan: &ExpiryScan) -> bool {
+		self.copies()
+			.filter(|group| !group.is_aborted())
+			.fold(true, |expired, group| {
+				let idle = group
+					.cache_accessed_tick(scan.gc.then_some(scan.now))
+					.is_some_and(|tick| scan.now.saturating_sub(tick) > scan.max_ticks);
+				expired && idle
+			})
+	}
+
+	/// Enter every copy into the evictable population, so the pool's access average
+	/// samples whichever one readers use.
+	fn cache_demote(&self) {
+		for group in self.copies() {
+			group.cache_demote();
+		}
+	}
+
+	/// Abort every copy this slot holds.
+	fn abort(&self, err: Error) {
+		for group in self.copies() {
+			let _ = group.clone().abort(err.clone());
+		}
+	}
 }
 
 /// Heap the track keeps per cached group, excluding the group itself
@@ -390,9 +468,9 @@ impl TrackState {
 			if *sequence >= min_sequence
 				&& let Some(slot) = self.lookup.get(sequence)
 				&& slot.stamp == *stamp
-				&& !slot.group.is_aborted()
+				&& !slot.is_aborted()
 			{
-				return Poll::Ready(Ok(Some((slot.group.clone(), self.offset + i))));
+				return Poll::Ready(Ok(Some((slot.serving().clone(), self.offset + i))));
 			}
 		}
 
@@ -480,7 +558,7 @@ impl TrackState {
 			.lookup
 			.range(next_sequence.max(self.live_floor.unwrap_or(0))..)
 			.filter(|_| self.readable())
-			.map(|(_, slot)| &slot.group)
+			.map(|(_, slot)| slot.serving())
 			.take_while(|group| super::subscription::before_end(group.sequence, end_sequence))
 			.find(|group| !group.is_aborted());
 
@@ -512,9 +590,11 @@ impl TrackState {
 	/// it to someone who asked for the whole group would silently hand back a tail. It
 	/// is a miss instead, so the fetch goes upstream for the frames that are missing.
 	fn covering_group(&self, sequence: u64, frame_start: u64) -> Option<&group::Producer> {
-		let slot = self.lookup.get(&sequence)?;
-		let first = slot.group.live_first_frame()?;
-		(first as u64 <= frame_start).then_some(&slot.group)
+		self.lookup.get(&sequence)?.copies().find(|group| {
+			group
+				.live_first_frame()
+				.is_some_and(|first| first as u64 <= frame_start)
+		})
 	}
 
 	/// The local retention bound, or `None` when unknown or unlimited.
@@ -574,7 +654,7 @@ impl TrackState {
 	fn holds(&self, sequence: u64, stamp: u32) -> bool {
 		self.lookup
 			.get(&sequence)
-			.is_some_and(|slot| slot.stamp == stamp && !slot.group.is_aborted())
+			.is_some_and(|slot| slot.stamp == stamp && !slot.is_aborted())
 	}
 
 	/// The furthest presentation time the group at `sequence` could still reach: where
@@ -741,13 +821,7 @@ impl TrackState {
 				if slot.stamp != stamp {
 					continue;
 				}
-				if slot.group.is_aborted()
-					|| (!self.protects(sequence)
-						&& slot
-							.group
-							.cache_accessed_tick(scan.gc.then_some(scan.now))
-							.is_some_and(|tick| scan.now.saturating_sub(tick) > scan.max_ticks))
-				{
+				if slot.is_aborted() || (!self.protects(sequence) && slot.is_expired(&scan)) {
 					return true;
 				}
 				retained += 1;
@@ -775,14 +849,8 @@ impl TrackState {
 			let Some(slot) = self.lookup.get(sequence) else {
 				continue;
 			};
-			if slot.stamp == *stamp
-				&& !slot.group.is_aborted()
-				&& slot
-					.group
-					.cache_accessed_tick(scan.gc.then_some(scan.now))
-					.is_some_and(|tick| scan.now.saturating_sub(tick) > scan.max_ticks)
-			{
-				let _ = slot.group.clone().abort(Error::Old);
+			if slot.stamp == *stamp && !slot.is_aborted() && slot.is_expired(&scan) {
+				slot.abort(Error::Old);
 			}
 		}
 	}
@@ -805,16 +873,11 @@ impl TrackState {
 				}
 				// Already aborted: the frames are gone, reclaim the slot so a
 				// later fetch can serve the sequence again.
-				if slot.group.is_aborted() {
+				if slot.is_aborted() {
 					self.lookup.remove(&sequence);
 					continue;
 				}
-				if self.protects(sequence)
-					|| slot
-						.group
-						.cache_accessed_tick(scan.gc.then_some(scan.now))
-						.is_none_or(|tick| scan.now.saturating_sub(tick) <= scan.max_ticks)
-				{
+				if self.protects(sequence) || !slot.is_expired(&scan) {
 					// Writes keep their scan bounded. Cleanup visits the entire
 					// queue to date pending accesses and find idle entries behind them.
 					retained += 1;
@@ -826,8 +889,7 @@ impl TrackState {
 				// Take the group out of the cache and abort it, so any consumer
 				// still reading surfaces `Error::Old` instead of blocking forever
 				// on a frame that will never arrive.
-				let slot = self.lookup.remove(&sequence).unwrap();
-				let _ = slot.group.abort(Error::Old);
+				self.lookup.remove(&sequence).unwrap().abort(Error::Old);
 			}
 		}
 
@@ -879,7 +941,7 @@ impl TrackState {
 		if let Some(latest) = self.latest_group
 			&& let Some(slot) = self.lookup.get(&latest)
 		{
-			slot.group.cache_demote();
+			slot.cache_demote();
 			self.evict.push_back((latest, slot.stamp));
 		}
 	}
@@ -888,7 +950,8 @@ impl TrackState {
 	/// keep the finished ones for readers still draining. A consumer that already
 	/// pulled an open group keeps its own handle and ends with it.
 	fn drop_open_groups(&mut self) {
-		self.lookup.retain(|_, slot| slot.group.is_finished());
+		self.lookup
+			.retain(|_, slot| slot.copies().any(group::Producer::is_finished));
 	}
 
 	/// Attach the publisher's immutable metadata without replacing it with local cache policy.
@@ -922,19 +985,11 @@ impl TrackState {
 	/// request from there, so the wider producer takes the slot. Readers already
 	/// draining the old one keep their own handle.
 	fn claim_sequence(&mut self, sequence: u64, frame_start: u64) -> Result<()> {
-		if let Some(slot) = self.lookup.get(&sequence) {
-			// The same question `covering_group` asks: can this slot still answer from
-			// `frame_start`? If it can, the sequence is taken; if it can't, it is dead
-			// and the caller gets to replace it.
-			if slot
-				.group
-				.live_first_frame()
-				.is_some_and(|first| first as u64 <= frame_start)
-			{
-				return Err(Error::Duplicate);
-			}
-			self.lookup.remove(&sequence);
+		// Either copy that can still answer from `frame_start` takes the sequence.
+		if self.covering_group(sequence, frame_start).is_some() {
+			return Err(Error::Duplicate);
 		}
+		self.lookup.remove(&sequence);
 		Ok(())
 	}
 
@@ -959,7 +1014,7 @@ impl TrackState {
 				&& sequence > latest
 				&& let Some(prev) = self.lookup.get(&latest)
 			{
-				prev.group.cache_demote();
+				prev.cache_demote();
 				self.evict.push_back((latest, prev.stamp));
 			}
 			self.latest_group = Some(sequence);
@@ -976,6 +1031,7 @@ impl TrackState {
 				group: group.clone(),
 				stamp,
 				visible,
+				head: None,
 			},
 		);
 		if visible {
@@ -1038,7 +1094,7 @@ impl TrackState {
 		let Some(slot) = self.lookup.get(sequence) else {
 			return false;
 		};
-		slot.stamp == *stamp && !slot.group.is_aborted() && slot.group.cache_accessed() <= average
+		slot.stamp == *stamp && !slot.is_aborted() && slot.cache_accessed() <= average
 	}
 
 	/// Abort this track's stalest groups until the outstanding debt is paid, or
@@ -1072,7 +1128,7 @@ impl TrackState {
 				// A historical hint; the live entry is elsewhere in the queue.
 				continue;
 			}
-			if slot.group.is_aborted() {
+			if slot.is_aborted() {
 				// Aborted upstream: the frames are already gone, reclaim the slot.
 				self.lookup.remove(&sequence);
 				continue;
@@ -1087,13 +1143,13 @@ impl TrackState {
 			// Protected: accessed more recently than the average (a fresh insert,
 			// an active reader, or a FETCH hit, which also covers a backfill still
 			// being filled). Rotate to the back.
-			if slot.group.cache_accessed() > average {
+			if slot.cache_accessed() > average {
 				self.evict.push_back((sequence, stamp));
 				continue;
 			}
 			// The full footprint including overhead, so even empty groups repay
 			// their share of the budget when evicted.
-			let size = slot.group.cache_size();
+			let size = slot.cache_size();
 			if size > self.debt {
 				self.evict.push_front((sequence, stamp));
 				return;
@@ -1101,8 +1157,7 @@ impl TrackState {
 
 			self.debt -= size;
 			paid = paid.saturating_add(size);
-			let slot = self.lookup.remove(&sequence).unwrap();
-			let _ = slot.group.abort(Error::Evicted);
+			self.lookup.remove(&sequence).unwrap().abort(Error::Evicted);
 		}
 	}
 
@@ -1278,8 +1333,17 @@ impl TrackState {
 		}
 		let info = self.info.clone().unwrap();
 
-		// An evicted sequence can be re-fetched; a live one is a duplicate.
-		self.claim_sequence(sequence, frame_start)?;
+		// A live group the feed started past `frame_start` keeps its slot: subscribers
+		// walk it in arrival order and the feed keeps writing it. The fetch fills its
+		// head instead. Anything else that can't answer from `frame_start` (evicted,
+		// aborted, or narrower backfill) is replaced; a group that can is a duplicate.
+		if self.lookup.get(&sequence).is_some_and(Slot::is_live) {
+			if self.covering_group(sequence, frame_start).is_some() {
+				return Err(Error::Duplicate);
+			}
+		} else {
+			self.claim_sequence(sequence, frame_start)?;
+		}
 
 		let mut group = group::Producer::new(group::Info { sequence }, info, self.cache.clone());
 		// Start where the request did before the group is visible: a fetch looking it up
@@ -1288,10 +1352,24 @@ impl TrackState {
 		group.start_at(frame_start)?;
 		// A backfill exists because someone is fetching it right now: stamp that
 		// access so the eviction walk can't kill it before the fetch resolves.
-		// It is also invisible to arrival-order subscribers: fetched on demand,
-		// not produced live by the publisher.
 		group.cache_refresh();
-		self.commit_group(&group, false);
+
+		// Settled first, like `commit_group`, which may evict the live slot itself.
+		self.charge_debt();
+		let protected = self.protects(sequence);
+		match self.lookup.get_mut(&sequence).filter(|slot| slot.is_live()) {
+			Some(slot) => {
+				// Joins the slot's standing: demoted already unless it is the live edge.
+				if !protected {
+					group.cache_demote();
+				}
+				slot.head = Some(Box::new(group.clone()));
+			}
+			// Invisible to arrival-order subscribers: fetched on demand, not produced
+			// live by the publisher.
+			None => self.insert_group(&group, false),
+		}
+		self.evict_expired();
 		Ok(group)
 	}
 }
@@ -2542,7 +2620,7 @@ impl Consumer {
 			.lookup
 			.range(..sequence)
 			.rev()
-			.map(|(_, slot)| &slot.group)
+			.map(|(_, slot)| slot.serving())
 			.find(|group| !group.is_aborted())
 			.map(|group| group.consume())
 	}
@@ -2553,11 +2631,11 @@ impl Consumer {
 	/// miss.
 	pub(crate) fn peek_group(&self, sequence: u64) -> Option<group::Consumer> {
 		let state = self.state.read();
-		let slot = state.lookup.get(&sequence)?;
-		if slot.group.is_aborted() {
+		let group = state.lookup.get(&sequence)?.serving();
+		if group.is_aborted() {
 			return None;
 		}
-		Some(slot.group.consume())
+		Some(group.consume())
 	}
 
 	/// Poll for group `sequence` the way the live feed delivers it: `Some` once it is
@@ -2566,9 +2644,9 @@ impl Consumer {
 	pub(crate) fn poll_group(&self, sequence: u64, waiter: &kio::Waiter) -> Poll<Option<group::Consumer>> {
 		let res = self.state.poll(waiter, |state| {
 			if let Some(slot) = state.lookup.get(&sequence)
-				&& !slot.group.is_aborted()
+				&& !slot.is_aborted()
 			{
-				return Poll::Ready(Some(slot.group.consume()));
+				return Poll::Ready(Some(slot.serving().consume()));
 			}
 			let passed = state.max_sequence.is_some_and(|newest| newest > sequence)
 				|| state.start_sequence.is_some_and(|start| start > sequence)
@@ -3725,7 +3803,7 @@ impl Subscriber {
 							&& state
 								.lookup
 								.get(sequence)
-								.is_some_and(|slot| slot.stamp == *stamp && !slot.group.is_aborted())
+								.is_some_and(|slot| slot.stamp == *stamp && !slot.is_aborted())
 					}))
 	}
 
@@ -7997,6 +8075,154 @@ mod test {
 		);
 	}
 
+	/// A fetch for the head of a live group the feed started partway into keeps that group
+	/// in arrival order: subscribers get the fetched copy while it lasts, and the live
+	/// group again once the fetch is gone.
+	#[tokio::test]
+	async fn fetch_fills_the_head_of_a_live_group() {
+		let producer = track_producer("test", None);
+		let dynamic = producer.dynamic();
+		let consumer = producer.consume();
+
+		let mut live = producer.create_group(group::Info { sequence: 0 }).unwrap();
+		live.start_at(1).unwrap();
+		live.write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"delta"))
+			.unwrap();
+
+		let fetch = consumer.fetch_group(0, None);
+		let request = dynamic.requested_group().await.unwrap();
+		let mut head = request.accept(None).unwrap();
+		for payload in [&b"snapshot"[..], b"delta"] {
+			head.write_frame(Timestamp::ZERO, bytes::Bytes::from_static(payload))
+				.unwrap();
+		}
+		assert_eq!(fetch.await.unwrap().index(), 0);
+
+		// Arrival order hands out the fetched copy, from its first frame.
+		let mut sub = producer.subscribe(None);
+		let mut group = sub.recv_group().now_or_never().unwrap().unwrap().unwrap();
+		assert_eq!(
+			group.read_frame().await.unwrap().unwrap().payload,
+			bytes::Bytes::from_static(b"snapshot")
+		);
+
+		// An abandoned fetch leaves the live group in its slot, still written by the feed.
+		head.abort(Error::Cancel).unwrap();
+		live.write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"delta2"))
+			.unwrap();
+		let mut sub = producer.subscribe(None);
+		let mut group = sub.recv_group().now_or_never().unwrap().unwrap().unwrap();
+		// Positioning at the start clamps up to the live group's first frame.
+		group.start_at(0);
+		assert_eq!(group.index(), 1);
+		assert_eq!(
+			group.read_frame().await.unwrap().unwrap().payload,
+			bytes::Bytes::from_static(b"delta")
+		);
+	}
+
+	/// A live group with a fetched head: group 0, fed from frame 1, plus a two-frame head.
+	async fn live_group_with_head(producer: &Producer) -> (group::Producer, group::Producer) {
+		let dynamic = producer.dynamic();
+		let consumer = producer.consume();
+
+		let mut live = producer.create_group(group::Info { sequence: 0 }).unwrap();
+		live.start_at(1).unwrap();
+		live.write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"delta"))
+			.unwrap();
+
+		let fetch = consumer.fetch_group(0, None);
+		let request = dynamic.requested_group().await.unwrap();
+		let mut head = request.accept(None).unwrap();
+		for payload in [&b"snapshot"[..], b"delta"] {
+			head.write_frame(Timestamp::ZERO, bytes::Bytes::from_static(payload))
+				.unwrap();
+		}
+		head.finish().unwrap();
+		assert_eq!(fetch.await.unwrap().index(), 0);
+		(live, head)
+	}
+
+	/// The live copy ending early (its subscription went away) leaves a fetched head
+	/// that still holds the whole group servable, instead of reclaiming the slot.
+	#[tokio::test]
+	async fn aborted_live_group_keeps_its_fetched_head() {
+		let producer = track_producer("test", None);
+		let (live, _head) = live_group_with_head(&producer).await;
+
+		live.abort(Error::Cancel).unwrap();
+		// The next group demotes 0 into the eviction order and runs an expiry scan.
+		producer.create_group(group::Info { sequence: 1 }).unwrap();
+
+		let mut group = producer.consume().peek_group(0).expect("the head still serves group 0");
+		assert_eq!(
+			group.read_frame().await.unwrap().unwrap().payload,
+			bytes::Bytes::from_static(b"snapshot")
+		);
+	}
+
+	/// Reads land on the fetched head, so they keep the slot from expiring even though
+	/// the live copy beside it sits idle.
+	#[tokio::test]
+	async fn reading_the_head_keeps_the_slot_from_expiring() {
+		let producer = track_producer("test", None);
+		let (live, _head) = live_group_with_head(&producer).await;
+		live.finish().unwrap();
+		producer.create_group(group::Info { sequence: 1 }).unwrap();
+
+		crate::model::clock::advance(cache::DEFAULT_EXPIRY + Duration::from_secs(1));
+		// A fetch hit on the head, past the window the live copy was last written in.
+		producer.consume().fetch_group(0, None).await.unwrap();
+		producer.create_group(group::Info { sequence: 2 }).unwrap();
+
+		assert!(producer.consume().peek_group(0).is_some(), "a read head keeps its slot");
+	}
+
+	/// Fill group 0 as a live copy plus a fetched head written after it, optionally
+	/// abort the live copy, then write past capacity into a protected latest group, so
+	/// slot 0 is the only evictable content. Returns whether pressure evicted it.
+	async fn pressure_evicts_a_headed_slot(abort_live: bool) -> bool {
+		let (producer, _pool) = pooled_producer(10_000);
+		let dynamic = producer.dynamic();
+		let consumer = producer.consume();
+
+		let mut live = producer.create_group(group::Info { sequence: 0 }).unwrap();
+		live.start_at(1).unwrap();
+		live.write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"delta"))
+			.unwrap();
+		live.finish().unwrap();
+		let fetch = consumer.fetch_group(0, None);
+		let mut head = dynamic.requested_group().await.unwrap().accept(None).unwrap();
+		head.write_frame(Timestamp::ZERO, bytes::Bytes::from(vec![0u8; 10_000]))
+			.unwrap();
+		head.finish().unwrap();
+		drop(fetch.await.unwrap());
+		if abort_live {
+			live.abort(Error::Cancel).unwrap();
+		}
+
+		let mut next = producer.create_group(group::Info { sequence: 1 }).unwrap();
+		for _ in 0..30 {
+			next.write_frame(Timestamp::ZERO, bytes::Bytes::from(vec![0u8; 10_000]))
+				.unwrap();
+		}
+		consumer.peek_group(0).is_none()
+	}
+
+	/// A head left alone in its slot still joins the pool's access average when the
+	/// slot is demoted, so memory pressure can evict it.
+	#[tokio::test]
+	async fn a_head_only_slot_yields_to_memory_pressure() {
+		assert!(pressure_evicts_a_headed_slot(true).await);
+	}
+
+	/// A slot weighs its copies the way the pool's average samples them, so two copies
+	/// with different access stamps can't keep it above the average on their own.
+	#[tokio::test]
+	async fn a_two_copy_slot_yields_to_memory_pressure() {
+		assert!(pressure_evicts_a_headed_slot(false).await);
+	}
+
 	/// Joining a queued fetch widens its range, and a caller that arrives once the range
 	/// is already on the wire fails cleanly rather than being handed the narrower group.
 	#[tokio::test]
diff --git a/rs/moq-net/tests/head_fetch.rs b/rs/moq-net/tests/head_fetch.rs
new file mode 100644
index 0000000000..54c1b19897
--- /dev/null
+++ b/rs/moq-net/tests/head_fetch.rs
@@ -0,0 +1,162 @@
+//! A relay that fetches the head of a group it is receiving mid-group keeps delivering
+//! that group to arrival-order subscribers, along with the frames written into it after.
+//!
+//! A mesh peer subscribing from partway into an open group (a catalog: snapshot, then
+//! deltas) has its relay pull the group headless. Reading the peer's copy from frame 0
+//! then fetches the head through the relay, whose cache must not trade the live group
+//! for the fetched copy.
+
+mod support;
+
+use std::time::Duration;
+
+use moq_net::track::{Position, Subscription};
+use moq_net::{Hop, Timestamp, Version};
+use support::harness::{MockConnectOptions, connect_mock};
+
+const TIMEOUT: Duration = Duration::from_secs(5);
+
+const VERSIONS: &[&str] = &[
+	"moq-lite-03",
+	"moq-lite-05",
+	"moq-lite-06",
+	"moq-lite-07-wip",
+	"moq-transport-14",
+	"moq-transport-17",
+	"moq-transport-22",
+];
+
+fn produce_origin(hop: u64) -> moq_net::origin::Producer {
+	let (producer, driver) = moq_net::origin::Producer::new(moq_net::origin::Config::new(Hop::new(hop).unwrap()));
+	tokio::spawn(support::harness::run(driver));
+	producer
+}
+
+async fn resolve(origin: &moq_net::origin::Producer) -> moq_net::broadcast::Consumer {
+	let consumer = origin.consume();
+	tokio::time::timeout(TIMEOUT, consumer.routed("bcast"))
+		.await
+		.expect("announce timeout")
+		.expect("routed");
+	tokio::time::timeout(TIMEOUT, consumer.request_broadcast("bcast"))
+		.await
+		.expect("resolve timeout")
+		.expect("broadcast resolves")
+}
+
+async fn read(group: &mut moq_net::group::Consumer) -> Vec<u8> {
+	tokio::time::timeout(TIMEOUT, group.read_frame())
+		.await
+		.expect("no frame")
+		.expect("read_frame")
+		.expect("group ended")
+		.payload
+		.to_vec()
+}
+
+/// Publisher P holds an open group, relay R pulls P, peer M pulls R from the group's
+/// second frame and then reads its copy from the first. A fresh subscriber on R must
+/// still get the whole group, and a frame P writes afterward.
+async fn round(version: &str) {
+	let version: Version = version.parse().unwrap();
+	let publisher = produce_origin(1);
+	let broadcast = publisher.create_broadcast("bcast").unwrap();
+	let track = broadcast.create_track("catalog.json", None).unwrap();
+	broadcast.announce(Default::default()).unwrap();
+
+	let mut group = track.append_group().unwrap();
+	let sequence = group.sequence;
+	for payload in [&b"snapshot"[..], b"delta1", b"delta2"] {
+		group.write_frame(Timestamp::now(), payload).unwrap();
+	}
+
+	let relay = produce_origin(2);
+	let mut options = MockConnectOptions::new(version);
+	options.server_publish = Some(publisher.consume());
+	options.client_subscribe = Some(relay.clone());
+	let upstream = connect_mock(options).await;
+
+	let peer = produce_origin(3);
+	let mut options = MockConnectOptions::new(version);
+	options.server_publish = Some(relay.consume());
+	options.client_subscribe = Some(peer.clone());
+	let downstream = connect_mock(options).await;
+
+	let remote = resolve(&relay).await;
+	let peer_remote = resolve(&peer).await;
+
+	let start = Position {
+		group: sequence,
+		frame: 1,
+	};
+	let mut peer_sub = peer_remote
+		.track("catalog.json")
+		.unwrap()
+		.subscribe(Subscription::default().with_start(start))
+		.await
+		.expect("peer subscribe");
+	let mut peer_group = tokio::time::timeout(TIMEOUT, peer_sub.recv_group())
+		.await
+		.expect("peer got no group")
+		.unwrap()
+		.unwrap();
+	// The peer reads its headless copy from frame 0: a fetch of the head through R.
+	let _ = tokio::time::timeout(Duration::from_millis(500), peer_group.read_frame()).await;
+
+	// A fresh reader on R wants the latest group from its snapshot.
+	let mut sub = remote.track("catalog.json").unwrap().subscribe(None).await.unwrap();
+	let mut got = tokio::time::timeout(TIMEOUT, sub.recv_group())
+		.await
+		.unwrap_or_else(|_| panic!("{version}: no group"))
+		.unwrap()
+		.unwrap();
+	assert_eq!(got.sequence, sequence, "{version}");
+	assert_eq!(read(&mut got).await, b"snapshot", "{version}");
+	assert_eq!(read(&mut got).await, b"delta1", "{version}");
+	assert_eq!(read(&mut got).await, b"delta2", "{version}");
+
+	// The publisher keeps writing into the group: the reader on R hears it.
+	tokio::time::sleep(Duration::from_secs(1)).await;
+	group.write_frame(Timestamp::now(), &b"delta3"[..]).unwrap();
+	assert_eq!(
+		read(&mut got).await,
+		b"delta3",
+		"{version}: frame written after the fetch"
+	);
+
+	// Every reader of the fetched copy leaves, so its fetch is abandoned, while both
+	// subscriptions keep the live feed running. The live group still holds R's slot: a
+	// later reader gets the whole group again.
+	drop((got, peer_group));
+	tokio::time::sleep(Duration::from_secs(1)).await;
+	let mut late_sub = remote.track("catalog.json").unwrap().subscribe(None).await.unwrap();
+	let mut late = tokio::time::timeout(TIMEOUT, late_sub.recv_group())
+		.await
+		.unwrap_or_else(|_| panic!("{version}: no group after the fetch was abandoned"))
+		.unwrap()
+		.unwrap();
+	assert_eq!(late.sequence, sequence, "{version}");
+	for payload in [&b"snapshot"[..], b"delta1", b"delta2", b"delta3"] {
+		assert_eq!(
+			read(&mut late).await,
+			payload,
+			"{version}: after the fetch was abandoned"
+		);
+	}
+
+	drop((
+		late, late_sub, sub, peer_sub, group, track, broadcast, downstream, upstream, publisher, relay, peer,
+	));
+}
+
+#[tokio::test]
+async fn a_fetched_head_keeps_the_live_group_visible() {
+	tokio::time::pause();
+	let mut failures = Vec::new();
+	for version in VERSIONS {
+		if tokio::spawn(round(version)).await.is_err() {
+			failures.push(*version);
+		}
+	}
+	assert!(failures.is_empty(), "failed: {failures:?}");
+}