diff --git a/.claude/skills/verify-subscribe/SKILL.md b/.claude/skills/verify-subscribe/SKILL.md
index 01c5a06e290..687ac14ced0 100644
--- a/.claude/skills/verify-subscribe/SKILL.md
+++ b/.claude/skills/verify-subscribe/SKILL.md
@@ -41,11 +41,12 @@ Follow Reading a failure in `rs/moq-net/verify/README.md`. Replay the path on th
 
 ## Coverage
 
-Sync commit: `81a4d971fc766431ef49361c640a36d69b7f3ac2`. Run `git log 81a4d971fc.. -- rs/moq-net/src/lite/publisher.rs rs/moq-net/src/lite/subscriber.rs rs/moq-net/src/tail.rs rs/moq-net/src/model/track.rs` to see what changed since. Follow Coverage in `rs/moq-net/verify/README.md` after a change to a covered function.
+Sync commit: `1be43c735b4ddaf5811433d82390e377cd1fe15a`. Run `git log 1be43c735b.. -- rs/moq-net/src/lite/publisher.rs rs/moq-net/src/lite/subscriber.rs rs/moq-net/src/tail.rs rs/moq-net/src/model/track.rs` to see what changed since. Follow Coverage in `rs/moq-net/verify/README.md` after a change to a covered function.
 
 | Code | Model |
 | --- | --- |
 | `rs/moq-net/src/lite/publisher.rs` `TrackRun::poll_track` (`Recv::Group`, the re-offer pop), `serve`, `mark_served` | `CursorHand`, `PubReoffer`, `Serve` |
+| `rs/moq-net/src/lite/publisher.rs` `TrackRun::give_way`, `shed` | a new group, or one shed, as `Serve`, then `StreamFail` with `old` whose header never arrives; a continuation as `StreamYield` of a stream whose header never arrives |
 | `rs/moq-net/src/lite/publisher.rs` `poll_forward_drop`, `forward_range`, `drop_range`, `replace_stale`, `sent_stale`, `hole` | `CursorGiveUp`, `ReplaceStale` |
 | `rs/moq-net/src/lite/publisher.rs` `TrackRun::poll_step` (outcomes, `redrop`) | `PubOutcomes`, `PubRedrop` |
 | `rs/moq-net/src/lite/publisher.rs` `TrackRun::update`, `owe`; `rs/moq-net/src/model/track.rs` `reoffer_stale`, `reoffer_groups` | `PubUpdate` |
diff --git a/js/net/src/integration.test.ts b/js/net/src/integration.test.ts
index 2e7016a4104..7e48c1c73f3 100644
--- a/js/net/src/integration.test.ts
+++ b/js/net/src/integration.test.ts
@@ -3010,3 +3010,59 @@ test("integration: lite-07 resumes a group reset to make room without loss", asy
 		server.abort();
 	}
 });
+
+// Out of credit, a late group that ranks below the stream holding it waits for credit without
+// holding back a newer group that outranks that stream. Mirrors the Rust
+// `a_waiting_late_group_lets_a_newer_group_through`.
+for (const alpn of [Lite.ALPN_06, Lite.ALPN_07_WIP]) {
+	test(`integration: ${alpn} lets a newer group past a late one waiting for credit`, async () => {
+		using clock = manual();
+		const settled = async <T>(work: Promise<T>, what: string): Promise<T> => {
+			for (let i = 0; i < 100; i++) if (await settlesWithin(work, clock, Duration.millis(10))) return work;
+			throw new Error(`${what} never settled`);
+		};
+		const pair = qmuxPair(alpn, 1n);
+		await Promise.all([pair.client.ready, pair.server.ready]);
+		const origin = new OriginProducer();
+		const [client, server] = await Promise.all([
+			connect(url, { transport: pair.client }),
+			accept(pair.server, url, { publish: origin.consume() }),
+		]);
+		const broadcast = publish(origin, Path.from("live"));
+		const producer = broadcast.createTrack("0", { timescale: Timescale.MILLI, maxAge: null });
+		const remote = wireOf(client).consume(Path.from("live"));
+		const subscriber = remote
+			.track("0")
+			.subscribe({ order: "descending", delay: { max: Duration.millis(30_000) } });
+		try {
+			// Unfinished, so its stream holds the only credit.
+			// Each group stamped in sequence order, since the track checks a late group's start.
+			const at = (sequence: bigint) => Timestamp.fromMillis(1_000 + Number(sequence));
+			const held = producer.createGroup(2n);
+			held.writeFrame(textFrame("held", at(2n)));
+			expect((await settled(subscriber.recvGroup(), "the held group"))?.sequence).toBe(2n);
+
+			// Late and below the held stream: it waits for credit it cannot make room for.
+			const late = producer.createGroup(0n);
+			late.writeFrame(textFrame("late", at(0n)));
+			late.finish();
+
+			const newer = producer.createGroup(3n);
+			newer.writeFrame(textFrame("new", at(3n)));
+			newer.finish();
+			// lite-07 may fill the late group by a fetch first; only group 3 must not wait behind it.
+			for (;;) {
+				const group = await settled(subscriber.recvGroup(), "the newer group past the late one");
+				if (!group) throw new Error("the subscription ended");
+				if (group.sequence === 3n) break;
+				expect(group.sequence).toBe(0n);
+			}
+		} finally {
+			subscriber.close();
+			remote.close();
+			broadcast.close();
+			client.abort();
+			server.abort();
+		}
+	});
+}
diff --git a/quest/m1/stream-credit/rank-opening.md b/quest/m1/stream-credit/rank-opening.md
deleted file mode 100644
index 32ac4beb2e9..00000000000
--- a/quest/m1/stream-credit/rank-opening.md
+++ /dev/null
@@ -1,43 +0,0 @@
-# [M] A lower-ranked group waiting for stream credit holds back higher-ranked ones
-
-## Goal
-
-Out of stream credit, a Rust publisher's subscription never holds a
-higher-ranked group back behind a lower-ranked one waiting for credit: a
-newer group that would reset a lower stream reaches the open line even while
-another group of the subscription waits there.
-
-Today `TrackRun` (`rs/moq-net/src/lite/publisher.rs`) keeps one group waiting
-for credit in `opening`, and its cursor hands out nothing while one waits.
-A group that ranks below every stream the subscription holds can never make
-room, so it blocks the cursor until credit returns or a stream finishes. With
-one credit and descending order, unfinished group 2 holds the stream, late
-group 0 waits in `opening`, and newer group 3, which would reset group 2,
-never reaches the line. This reproduces on lite-06 at `4265dc5b57`, before
-any lite-07 YIELDED: a test that publishes groups 2 (unfinished), 0, then 3
-times out waiting for group 3.
-[moq-dev/moq#8545](https://github.com/moq-dev/moq/pull/8545) handles one case,
-a lite-07 continuation waiting in `opening`, by letting the cursor hand out
-one more group (`TrackRun::stashed`); a late lower-ranked group stashed there
-blocks the cursor again.
-
-## Plan
-
-Recommended: while a group waits in `opening`, keep polling the cursor, and
-keep the higher-ranked of the waiting group and the one it hands out
-(`Paused::outranks`). The other leaves the line: a lite-07 continuation goes
-back to `paused`, and a new group is given up as a group expired under the
-subscription's budget is (a lite-07 stale drop, which the next
-SUBSCRIBE_UPDATE may re-offer). This keeps one waiting group per
-subscription plus the paused continuations stream credit bounds, and folds
-in #8545's `stashed`. Check whether `SubscribeAccounting.tla`'s accounting
-rules still hold, since a handed-out group may now end in a stale drop
-without a stream.
-
-JS (`js/net/src/lite/publisher.ts`) opens every group concurrently, ranked by
-send order, so it has no counterpart; confirm with the same test there.
-
-Tests: the repro above, on lite-06 and lite-07, in
-`rs/moq-net/tests/integration/stream_credit.rs`.
-
-Public API: none. Wire: none.
diff --git a/rs/kio/src/task.rs b/rs/kio/src/task.rs
index 838da944839..2d02270ac2c 100644
--- a/rs/kio/src/task.rs
+++ b/rs/kio/src/task.rs
@@ -330,6 +330,11 @@ impl<T> Tasks<T> {
 		self.len == 0
 	}
 
+	/// The tasks still running, in no particular order.
+	pub fn iter(&self) -> impl Iterator<Item = &T> {
+		self.children.iter().flatten().map(|occupant| &occupant.task)
+	}
+
 	/// Add a task, queued for its first poll.
 	///
 	/// Wakes the owner, so a task pushed between polls (from another arm of the
diff --git a/rs/moq-net/src/lite/publisher.rs b/rs/moq-net/src/lite/publisher.rs
index 5887a4ffeca..a078a9f40fa 100644
--- a/rs/moq-net/src/lite/publisher.rs
+++ b/rs/moq-net/src/lite/publisher.rs
@@ -3557,7 +3557,7 @@ mod test {
 				.finish_at(track::Position::group(7), Some(Error::App(7)))
 				.unwrap();
 			let next = |subscriber: &mut track::Subscriber| {
-				kio::wait(|waiter| poll_recv_next(subscriber, true, true, end_error, waiter)).now_or_never()
+				kio::wait(|waiter| poll_recv_next(subscriber, Some(ALL), true, end_error, waiter)).now_or_never()
 			};
 			match (end_error || capped, next(&mut subscriber)) {
 				(true, Some(Ok(Recv::Boundary(end)))) => {
@@ -4687,6 +4687,9 @@ fn final_sequence(end: track::Position) -> u64 {
 	end.final_sequence().unwrap_or(u64::MAX)
 }
 
+/// Every group sequence, for [`poll_recv_next`] to hand out any.
+const ALL: (Bound<u64>, Bound<u64>) = (Bound::Unbounded, Bound::Unbounded);
+
 /// Poll a single [`track::Subscriber`] for the next group (cap-aware, in arrival order).
 ///
 /// Groups are served in arrival order (`poll_recv_group`), not sequence order: on a relay, a
@@ -4694,8 +4697,9 @@ fn final_sequence(end: track::Position) -> u64 {
 /// permanently skip the older group even though it is cached and in demand. Staleness is
 /// governed by max age and the subscriber's max delay, not arrival raciness.
 ///
-/// Without `groups` the cursor hands out none, leaving them in the cache, so it never reports
-/// [`Recv::Finished`] either.
+/// The cursor hands out the next group only if its sequence lies in `groups`, leaving it in
+/// the cache otherwise, and without `groups` none at all. Either way it then never reports
+/// [`Recv::Finished`].
 ///
 /// When `emit_boundary` is set, a declared-but-not-yet-reached final sequence surfaces as
 /// [`Recv::Boundary`] in an idle moment (after groups), so the caller can send
@@ -4706,7 +4710,7 @@ fn final_sequence(end: track::Position) -> u64 {
 /// that end finishes like a clean one. Otherwise reaching it is the error, which resets.
 fn poll_recv_next(
 	track: &mut track::Subscriber,
-	groups: bool,
+	groups: Option<(Bound<u64>, Bound<u64>)>,
 	emit_boundary: bool,
 	end_error: bool,
 	waiter: &kio::Waiter,
@@ -4714,8 +4718,8 @@ fn poll_recv_next(
 	{
 		let mut groups_finished = false;
 		let next = match groups {
-			true => track.poll_recv_group(waiter),
-			false => Poll::Pending,
+			Some(admit) => track.poll_recv_group_in(admit, waiter),
+			None => Poll::Pending,
 		};
 		match next {
 			Poll::Ready(Ok(Some(group))) => return Poll::Ready(Ok(Recv::Group(group))),
@@ -4755,7 +4759,7 @@ fn poll_recv_next(
 /// The async form of [`poll_recv_next`], for callers with nothing else to poll.
 #[cfg(test)]
 async fn recv_next(track: &mut track::Subscriber, emit_boundary: bool) -> Result<Recv, Error> {
-	kio::wait(|waiter| poll_recv_next(track, true, emit_boundary, true, waiter)).await
+	kio::wait(|waiter| poll_recv_next(track, Some(ALL), emit_boundary, true, waiter)).await
 }
 
 /// Bound `group` to what this subscription asked for, or `None` when it cannot be served
@@ -5033,11 +5037,9 @@ struct Paused {
 }
 
 impl Paused {
-	/// Whether `a` ranks above `b` for a subscription's stream credit, by sequence in its
-	/// `order`, as [`crate::priority::in_group`] ranks them.
-	fn outranks(order: crate::track::Order, a: &Self, b: &Self) -> bool {
-		crate::priority::in_group(a.group.sequence, a.frame_start, order)
-			> crate::priority::in_group(b.group.sequence, b.frame_start, order)
+	/// Its rank for the subscription's stream credit in `order`, higher first.
+	fn rank(&self, order: crate::track::Order) -> i64 {
+		crate::priority::in_group(self.group.sequence, self.frame_start, order)
 	}
 }
 
@@ -5133,9 +5135,6 @@ struct TrackRun {
 	paused: std::collections::VecDeque<Paused>,
 	// Lite-07: the group `opening` holds continues a paused one.
 	continuing: bool,
-	// Lite-07: a group the cursor handed out while a continuation waited for credit, ranked
-	// below it. One at most: the cursor hands out no other until it opens.
-	stashed: Option<Paused>,
 	end_sent: bool,
 	// Lite-07 with every range bounded: once every group in its ranges is given up, it is
 	// answered without a group to resolve its start, rather than left unanswered.
@@ -5143,7 +5142,8 @@ struct TrackRun {
 	/// The one group waiting for stream credit. The rest of the backlog waits in the cache,
 	/// which bills, ages, and evicts it, and the cursor's budget skips whatever went stale
 	/// meanwhile: a machine per parked group would hold state for as long as the peer
-	/// withholds credit.
+	/// withholds credit. One that cannot make room gives way to a group that outranks it
+	/// (see [`Self::stuck`]).
 	opening: Option<GroupServe>,
 	/// The groups with a stream, serving or awaiting the acknowledgement of their FIN.
 	children: kio::Tasks<GroupServe>,
@@ -5202,7 +5202,6 @@ impl TrackRun {
 			reoffered: Default::default(),
 			paused: Default::default(),
 			continuing: false,
-			stashed: None,
 			end_sent: false,
 			opening: None,
 			children: kio::Tasks::new(),
@@ -5605,27 +5604,16 @@ impl TrackRun {
 
 		// A paused group takes the next stream credit, ahead of re-offers and new groups, the
 		// highest-ranked first. It is owed already, and the subscriber holds its group open until
-		// it continues. A group stashed behind a continuation goes first if it outranks them all.
-		let best = match self.opening.is_none() {
-			true => self.best_paused(),
-			false => None,
-		};
-		if self.opening.is_none()
-			&& let Some(stashed) = self.stashed.take()
-		{
-			let order = self.ctx.steer_current().order;
-			match best.map(|best| &self.paused[best]) {
-				Some(paused) if !Paused::outranks(order, &stashed, paused) => self.stashed = Some(stashed),
-				_ => {
-					self.open(stashed.group, stashed.frame_start);
-					return Poll::Ready(Ok(ControlFlow::Continue(())));
-				}
-			}
-		}
-		if self.opening.is_none()
-			&& let Some(Paused { group, frame_start }) = best.and_then(|best| self.paused.remove(best))
+		// it continues. A group waiting for credit it cannot make room for gives way to one that
+		// outranks it.
+		let order = self.ctx.steer_current().order;
+		let stuck = self.stuck();
+		if let Some(best) = self.best_paused()
+			&& (self.opening.is_none() || stuck.is_some_and(|waiting| self.paused[best].rank(order) > waiting))
 		{
+			let Paused { group, frame_start } = self.paused.remove(best).expect("best is in range");
 			tracing::debug!(subscribe = self.ctx.id, track = %self.ctx.track_name, sequence = group.sequence, frame_start, "continuing paused group");
+			self.give_way();
 			self.open(group, frame_start);
 			self.continuing = true;
 			return Poll::Ready(Ok(ControlFlow::Continue(())));
@@ -5657,11 +5645,21 @@ impl TrackRun {
 
 		// One cursor drives the whole subscription: poll the cap-aware arrival-order group.
 		let emit_boundary = self.emit_range && !self.end_sent;
-		// A continuation waiting for credit ranks below every stream it could take, so it must
-		// not hold back a newer group that outranks one of them: the cursor hands one more out.
-		let groups = self.opening.is_none() || (self.continuing && self.stashed.is_none());
+		// A group waiting for credit it cannot make room for must not hold back one that could:
+		// a group that outranks it takes its place. Newest first sheds the backlog, so the cursor
+		// hands out every group and one ranked below the waiting one is given up. Oldest first
+		// drains it in order, so the cursor hands out only a group that outranks the waiting
+		// one, and the rest waits in the cache. Behind a group that can make room, the backlog
+		// waits in the cache either way.
+		let admit = match &self.opening {
+			None => Some(ALL),
+			Some(waiting) => stuck.map(|_| match order {
+				Order::Descending => ALL,
+				Order::Ascending => (Bound::Unbounded, Bound::Excluded(waiting.sequence)),
+			}),
+		};
 		let end_error = self.ctx.version.has_end_error();
-		if let Poll::Ready(res) = poll_recv_next(&mut self.track, groups, emit_boundary, end_error, waiter) {
+		if let Poll::Ready(res) = poll_recv_next(&mut self.track, admit, emit_boundary, end_error, waiter) {
 			match res? {
 				Recv::Group(group) => {
 					// Never a second stream for a sequence this subscription already accounted
@@ -5692,6 +5690,15 @@ impl TrackRun {
 						}
 						return Poll::Ready(Ok(ControlFlow::Continue(())));
 					};
+					if let Some(waiting) = &self.opening
+						&& crate::priority::in_group(sequence, 0, order)
+							< crate::priority::in_group(waiting.sequence, 0, order)
+					{
+						tracing::debug!(subscribe = self.ctx.id, track = %self.ctx.track_name, sequence, "shedding a group ranked below one waiting for credit");
+						self.mark_served(sequence);
+						self.shed(sequence);
+						return Poll::Ready(Ok(ControlFlow::Continue(())));
+					}
 					match self.emit_range && self.start.is_none() {
 						true => self.first = Some(group),
 						false => self.serve(group),
@@ -6053,32 +6060,63 @@ impl TrackRun {
 		Ok(())
 	}
 
-	/// Open a group machine for `group`, which waits for stream credit before the cursor
-	/// hands out another.
+	/// Open a group machine for `group`, which waits for stream credit. The cursor hands it
+	/// out past a group already waiting only if it outranks that one, which gives way.
 	fn serve(&mut self, group: group::Consumer) {
 		let sequence = group.sequence;
 		let frame_start = group.index();
 		self.mark_served(sequence);
 		tracing::debug!(subscribe = self.ctx.id, track = %self.ctx.track_name, sequence, "serving group");
-		let Some(waiting) = self.opening.take() else {
-			return self.open(group, frame_start);
-		};
-		// Lite-07: handed out while a continuation waits for credit. The higher-ranked of the two
-		// waits for it, and the other right behind; a continuation goes back among the paused.
-		debug_assert!(self.continuing, "only a continuation shares the wait");
-		let serving = Paused { group, frame_start };
-		let continuation = Paused {
-			group: waiting.group.clone(),
-			frame_start: waiting.frame_start,
+		self.give_way();
+		self.open(group, frame_start);
+	}
+
+	/// The rank of the group waiting for stream credit, if it cannot make room: it ranks below
+	/// every stream this subscription could take over, so it waits for credit to return, while
+	/// a group that outranks one of those streams would take it over now. A finished stream
+	/// waiting for its acknowledgement can't be taken over, so it isn't ranked.
+	fn stuck(&mut self) -> Option<i64> {
+		let waiting = self.opening.as_ref()?;
+		let order = self.ctx.steer_current().order;
+		let rank = |serve: &GroupServe| crate::priority::in_group(serve.sequence, serve.frame_start, order);
+		let waiting = rank(waiting);
+		let mut held = self
+			.children
+			.iter()
+			.filter(|serve| serve.serving())
+			.map(rank)
+			.peekable();
+		(held.peek().is_some() && held.all(|held| held > waiting)).then_some(waiting)
+	}
+
+	/// Take the group waiting for stream credit out of line, for one that outranks it: a
+	/// continuation goes back among the paused, and a new group is given up as one expired
+	/// under the subscription's budget is, which a lite-07 update may re-offer. Either way at
+	/// most one group per subscription waits for credit.
+	fn give_way(&mut self) {
+		let Some(mut waiting) = self.opening.take() else {
+			return;
 		};
-		if Paused::outranks(self.ctx.steer_current().order, &serving, &continuation) {
-			drop(waiting);
-			self.continuing = false;
-			self.paused.push_back(continuation);
-			self.open(serving.group, serving.frame_start);
-		} else {
-			self.opening = Some(waiting);
-			self.stashed = Some(serving);
+		tracing::debug!(subscribe = self.ctx.id, track = %self.ctx.track_name, sequence = waiting.sequence, "giving way to a higher-ranked group");
+		match std::mem::take(&mut self.continuing) {
+			true => self.paused.push_back(Paused {
+				group: waiting.group.clone(),
+				frame_start: waiting.frame_start,
+			}),
+			false => waiting.settle_failed(&Error::Old),
+		}
+	}
+
+	/// Give up a group the cursor handed out without a stream, as one expired under the
+	/// subscription's budget is.
+	fn shed(&self, sequence: u64) {
+		if self.ctx.version.has_track_stream() {
+			self.ctx
+				.outcomes
+				.dropped
+				.lock()
+				.unwrap()
+				.push((sequence, StreamError::Old));
 		}
 	}
 
@@ -6087,7 +6125,7 @@ impl TrackRun {
 	fn best_paused(&mut self) -> Option<usize> {
 		let order = self.ctx.steer_current().order;
 		(0..self.paused.len()).reduce(|best, next| {
-			match Paused::outranks(order, &self.paused[next], &self.paused[best]) {
+			match self.paused[next].rank(order) > self.paused[best].rank(order) {
 				true => next,
 				false => best,
 			}
@@ -6317,6 +6355,11 @@ enum GroupState {
 }
 
 impl GroupServe {
+	/// Whether it holds a stream still sending, which a higher-ranked group may take over.
+	fn serving(&self) -> bool {
+		matches!(self.state, GroupState::Serve { .. })
+	}
+
 	fn new(
 		ctx: GroupContext,
 		sequence: u64,
diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs
index a93dce1acb1..8f8645fa559 100644
--- a/rs/moq-net/src/model/track.rs
+++ b/rs/moq-net/src/model/track.rs
@@ -9610,10 +9610,31 @@ impl Subscriber {
 		self.poll_group(false, waiter)
 	}
 
+	/// [`Self::poll_recv_group`], handing out the next group only if its sequence lies in
+	/// `admit`: one outside stays next, and the read is `Pending` without parking on it.
+	#[cfg(feature = "async")]
+	pub(crate) fn poll_recv_group_in(
+		&mut self,
+		admit: (ops::Bound<u64>, ops::Bound<u64>),
+		waiter: &kio::Waiter,
+	) -> Poll<Result<Option<group::Consumer>>> {
+		self.poll_group_in(false, admit, waiter)
+	}
+
 	/// The next group in arrival order, or in sequence order for `ordered`, skipping any
 	/// already handed out.
 	fn poll_group(&mut self, ordered: bool, waiter: &kio::Waiter) -> Poll<Result<Option<group::Consumer>>> {
-		let res = self.poll_group_unsettled(ordered, waiter);
+		self.poll_group_in(ordered, (ops::Bound::Unbounded, ops::Bound::Unbounded), waiter)
+	}
+
+	/// [`Self::poll_group`], refusing a next group whose sequence lies outside `admit`.
+	fn poll_group_in(
+		&mut self,
+		ordered: bool,
+		admit: (ops::Bound<u64>, ops::Bound<u64>),
+		waiter: &kio::Waiter,
+	) -> Poll<Result<Option<group::Consumer>>> {
+		let res = self.poll_group_unsettled(ordered, admit, waiter);
 		// What the read settled is billed now, and pays what it can of its readers' debt. The
 		// group it hands out stays above the fold until its reader accounts for it.
 		let handed = match &res {
@@ -9626,7 +9647,12 @@ impl Subscriber {
 		}
 	}
 
-	fn poll_group_unsettled(&mut self, ordered: bool, waiter: &kio::Waiter) -> Poll<Result<Option<group::Consumer>>> {
+	fn poll_group_unsettled(
+		&mut self,
+		ordered: bool,
+		admit: (ops::Bound<u64>, ops::Bound<u64>),
+		waiter: &kio::Waiter,
+	) -> Poll<Result<Option<group::Consumer>>> {
 		if self.record().failed() {
 			return Poll::Ready(Err(Error::Lagged));
 		}
@@ -9644,12 +9670,16 @@ impl Subscriber {
 			_ => {}
 		}
 		match self.poll_peek_group(ordered, waiter) {
-			Poll::Ready(Ok(Some(reserved))) => Poll::Ready(self.take(reserved).map(Some)),
+			Poll::Ready(Ok(Some(reserved))) if admit.contains(&reserved.sequence()) => {
+				Poll::Ready(self.take(reserved).map(Some))
+			}
 			res => {
 				// What the peek skipped for age counts even when nothing is handed out.
 				self.count_stale(&self.stats.meter());
 				match res {
 					Poll::Ready(Ok(None)) => Poll::Ready(self.ended()),
+					// Refused: it stays next, so nothing behind it is either.
+					Poll::Ready(Ok(Some(_))) => Poll::Pending,
 					res => res.map(|res| res.map(|_| None)),
 				}
 			}
diff --git a/rs/moq-net/tests/integration/stream_credit.rs b/rs/moq-net/tests/integration/stream_credit.rs
index d261c4af0c1..1efb881393a 100644
--- a/rs/moq-net/tests/integration/stream_credit.rs
+++ b/rs/moq-net/tests/integration/stream_credit.rs
@@ -505,3 +505,287 @@ async fn an_abandoned_setup_returns_credit_with_the_publisher_still_live() {
 		drop(first);
 	}
 }
+
+/// Out of credit, a late group that ranks below the stream holding it waits for credit
+/// without holding back a newer group that outranks that stream: the newer group still
+/// resets it, on each protocol.
+#[kio::test]
+async fn a_waiting_late_group_lets_a_newer_group_through() {
+	for version in ["moq-lite-06", "moq-lite-07-wip"] {
+		let version: Version = version.parse().unwrap();
+		let publisher = origin(1);
+		let viewer = origin(2);
+		let broadcast = publisher
+			.route("live", moq_net::origin::RouteOptions::hidden())
+			.unwrap();
+		let info = track::Info::default()
+			.with_timescale(moq_net::Timescale::MILLI)
+			.with_max_age(None);
+		let track = broadcast.broadcast().create_track("0", info).unwrap();
+		broadcast.announce().unwrap();
+
+		let mut options = MockConnectOptions::new(version);
+		options.transport.max_concurrent_uni_streams = Some(1);
+		options.server_publish = Some(publisher.consume());
+		options.client_subscribe = Some(viewer.clone());
+		let _pair = connect_mock(options).await;
+
+		let remote = viewer
+			.consume()
+			.follow("live", None)
+			.unwrap()
+			.available()
+			.await
+			.unwrap()
+			.unwrap();
+		let prefs = track::Subscription::default()
+			.with_order(track::Order::Descending)
+			.with_delay(track::Delay::default().with_max(Duration::from_secs(30)));
+		let mut subscriber = remote.track("0").unwrap().subscribe(prefs).await.unwrap();
+		macro_rules! next {
+			($what:literal) => {
+				kio::clock::timeout(Duration::from_secs(5), subscriber.recv_group())
+					.await
+					.unwrap_or_else(|_| panic!("{version}: {}", $what))
+					.unwrap()
+					.unwrap()
+			};
+		}
+
+		// Unfinished, so its stream holds the only credit.
+		let mut held = track.create_group(moq_net::group::Info { sequence: 2 }).unwrap();
+		held.write_frame(Timestamp::ZERO, b"held".as_slice()).unwrap();
+		assert_eq!(next!("the held group arrives").sequence, 2, "{version}");
+
+		// Late and below the held stream: it waits for credit it cannot make room for.
+		let mut late = track.create_group(moq_net::group::Info { sequence: 0 }).unwrap();
+		late.write_frame(Timestamp::ZERO, b"late".as_slice()).unwrap();
+		late.finish().unwrap();
+
+		let mut newer = track.create_group(moq_net::group::Info { sequence: 3 }).unwrap();
+		newer.write_frame(Timestamp::ZERO, b"new".as_slice()).unwrap();
+		newer.finish().unwrap();
+		assert_eq!(
+			next!("the newer group waits behind the late one").sequence,
+			3,
+			"{version}"
+		);
+		drop(held);
+	}
+}
+
+/// Newest first sheds the backlog: under a late group waiting for credit, a later late group
+/// ranked below it is given up rather than holding back a newer group behind it, on each
+/// protocol.
+#[kio::test]
+async fn a_shed_late_group_lets_a_newer_group_through() {
+	for version in ["moq-lite-06", "moq-lite-07-wip"] {
+		let version: Version = version.parse().unwrap();
+		let publisher = origin(1);
+		let viewer = origin(2);
+		let broadcast = publisher
+			.route("live", moq_net::origin::RouteOptions::hidden())
+			.unwrap();
+		let info = track::Info::default()
+			.with_timescale(moq_net::Timescale::MILLI)
+			.with_max_age(None);
+		let track = broadcast.broadcast().create_track("0", info).unwrap();
+		broadcast.announce().unwrap();
+
+		let mut options = MockConnectOptions::new(version);
+		options.transport.max_concurrent_uni_streams = Some(1);
+		options.server_publish = Some(publisher.consume());
+		options.client_subscribe = Some(viewer.clone());
+		let _pair = connect_mock(options).await;
+
+		let remote = viewer
+			.consume()
+			.follow("live", None)
+			.unwrap()
+			.available()
+			.await
+			.unwrap()
+			.unwrap();
+		let prefs = track::Subscription::default()
+			.with_order(track::Order::Descending)
+			.with_delay(track::Delay::default().with_max(Duration::from_secs(30)));
+		let mut subscriber = remote.track("0").unwrap().subscribe(prefs).await.unwrap();
+		macro_rules! next {
+			($what:literal) => {
+				kio::clock::timeout(Duration::from_secs(5), subscriber.recv_group())
+					.await
+					.unwrap_or_else(|_| panic!("{version}: {}", $what))
+					.unwrap()
+					.unwrap()
+			};
+		}
+
+		// Unfinished, so its stream holds the only credit.
+		let mut held = track.create_group(moq_net::group::Info { sequence: 3 }).unwrap();
+		held.write_frame(Timestamp::ZERO, b"held".as_slice()).unwrap();
+		assert_eq!(next!("the held group arrives").sequence, 3, "{version}");
+
+		// Group 1 waits for credit it cannot make room for, and group 0 ranks below it.
+		for sequence in [1, 0] {
+			let mut late = track.create_group(moq_net::group::Info { sequence }).unwrap();
+			late.write_frame(Timestamp::ZERO, b"late".as_slice()).unwrap();
+			late.finish().unwrap();
+		}
+
+		let mut newer = track.create_group(moq_net::group::Info { sequence: 4 }).unwrap();
+		newer.write_frame(Timestamp::ZERO, b"new".as_slice()).unwrap();
+		newer.finish().unwrap();
+		assert_eq!(
+			next!("the newer group waits behind the late ones").sequence,
+			4,
+			"{version}"
+		);
+		drop(held);
+	}
+}
+
+/// Oldest first, a backlog behind a group waiting for credit stays in the cache rather than
+/// giving it up: a newer group ranks below the waiting one, so every group still arrives once
+/// the stream holding the credit finishes, on each protocol.
+#[kio::test]
+async fn an_ascending_backlog_waits_behind_a_waiting_group() {
+	for version in ["moq-lite-06", "moq-lite-07-wip"] {
+		let version: Version = version.parse().unwrap();
+		let publisher = origin(1);
+		let viewer = origin(2);
+		let broadcast = publisher
+			.route("live", moq_net::origin::RouteOptions::hidden())
+			.unwrap();
+		let info = track::Info::default()
+			.with_timescale(moq_net::Timescale::MILLI)
+			.with_max_age(None);
+		let track = broadcast.broadcast().create_track("0", info).unwrap();
+		broadcast.announce().unwrap();
+
+		let mut options = MockConnectOptions::new(version);
+		options.transport.max_concurrent_uni_streams = Some(1);
+		options.server_publish = Some(publisher.consume());
+		options.client_subscribe = Some(viewer.clone());
+		let _pair = connect_mock(options).await;
+
+		let remote = viewer
+			.consume()
+			.follow("live", None)
+			.unwrap()
+			.available()
+			.await
+			.unwrap()
+			.unwrap();
+		let prefs = track::Subscription::default()
+			.with_order(track::Order::Ascending)
+			.with_delay(track::Delay::default().with_max(Duration::from_secs(30)));
+		let mut subscriber = remote.track("0").unwrap().subscribe(prefs).await.unwrap();
+		macro_rules! next {
+			($what:literal) => {
+				kio::clock::timeout(Duration::from_secs(5), subscriber.recv_group())
+					.await
+					.unwrap_or_else(|_| panic!("{version}: {}", $what))
+					.unwrap()
+					.unwrap()
+			};
+		}
+
+		// Unfinished, so its stream holds the only credit, and outranks every newer group.
+		let mut held = track.append_group().unwrap();
+		held.write_frame(Timestamp::ZERO, b"held".as_slice()).unwrap();
+		assert_eq!(next!("the held group arrives").sequence, 0, "{version}");
+
+		for _ in 0..3 {
+			let mut group = track.append_group().unwrap();
+			group.write_frame(Timestamp::ZERO, b"x".as_slice()).unwrap();
+			group.finish().unwrap();
+		}
+		// Let the publisher take what it would before credit returns.
+		kio::clock::sleep(Duration::from_millis(100)).await;
+		held.finish().unwrap();
+
+		let mut received = Vec::new();
+		while received.len() < 3 {
+			received.push(next!("the backlog arrives").sequence);
+		}
+		assert_eq!(received, [1, 2, 3], "{version}");
+	}
+}
+
+/// A finished stream still waiting for its acknowledgement holds credit that no group can
+/// take over, so it doesn't count when ranking a waiting group: a late group ranked above it
+/// but below the stream still serving waits, and a newer group resets that stream rather than
+/// waiting behind the late one, on each protocol.
+#[kio::test]
+async fn a_closing_stream_does_not_rank_a_waiting_group() {
+	const LATENCY: Duration = Duration::from_millis(100);
+	const SETTLE: Duration = Duration::from_millis(10);
+
+	for version in ["moq-lite-06", "moq-lite-07-wip"] {
+		let version: Version = version.parse().unwrap();
+		let publisher = origin(1);
+		let viewer = origin(2);
+		let broadcast = publisher
+			.route("live", moq_net::origin::RouteOptions::hidden())
+			.unwrap();
+		let info = track::Info::default()
+			.with_timescale(moq_net::Timescale::MILLI)
+			.with_max_age(None);
+		let track = broadcast.broadcast().create_track("0", info).unwrap();
+		broadcast.announce().unwrap();
+
+		let mut options = MockConnectOptions::new(version);
+		options.transport.max_concurrent_uni_streams = Some(2);
+		options.latency = LATENCY;
+		options.server_publish = Some(publisher.consume());
+		options.client_subscribe = Some(viewer.clone());
+		let _pair = connect_mock(options).await;
+
+		let remote = viewer
+			.consume()
+			.follow("live", None)
+			.unwrap()
+			.available()
+			.await
+			.unwrap()
+			.unwrap();
+		let prefs = track::Subscription::default()
+			.with_order(track::Order::Descending)
+			.with_delay(track::Delay::default().with_max(Duration::from_secs(30)));
+		let mut subscriber = remote.track("0").unwrap().subscribe(prefs).await.unwrap();
+
+		// Group 7 finishes, so its stream waits for the acknowledgement; group 10 holds the other.
+		let mut closing = track.create_group(moq_net::group::Info { sequence: 7 }).unwrap();
+		closing.write_frame(Timestamp::ZERO, b"closing".as_slice()).unwrap();
+		closing.finish().unwrap();
+		let mut held = track.create_group(moq_net::group::Info { sequence: 10 }).unwrap();
+		held.write_frame(Timestamp::ZERO, b"held".as_slice()).unwrap();
+		kio::clock::sleep(SETTLE).await;
+
+		// Ranked above group 7 but below group 10: it can only wait for group 7's credit.
+		let mut late = track.create_group(moq_net::group::Info { sequence: 8 }).unwrap();
+		late.write_frame(Timestamp::ZERO, b"late".as_slice()).unwrap();
+		late.finish().unwrap();
+		kio::clock::sleep(SETTLE).await;
+
+		let mut newer = track.create_group(moq_net::group::Info { sequence: 11 }).unwrap();
+		newer.write_frame(Timestamp::ZERO, b"new".as_slice()).unwrap();
+		newer.finish().unwrap();
+
+		loop {
+			let group = kio::clock::timeout(Duration::from_secs(5), subscriber.recv_group())
+				.await
+				.unwrap_or_else(|_| panic!("{version}: the newer group never arrives"))
+				.unwrap()
+				.unwrap();
+			assert_ne!(
+				group.sequence, 8,
+				"{version}: the newer group waits behind the late one"
+			);
+			if group.sequence == 11 {
+				break;
+			}
+		}
+		drop(held);
+	}
+}