diff --git a/drafts/draft-lcurley-moq-lite.md b/drafts/draft-lcurley-moq-lite.md
index 455a587917..679b2056d6 100644
--- a/drafts/draft-lcurley-moq-lite.md
+++ b/drafts/draft-lcurley-moq-lite.md
@@ -447,7 +447,7 @@ A receiver SHOULD NOT cache refusals; rate limiting is the advertiser's concern.
 A subscriber opens Subscribe Streams to request a Track.
 
 The subscriber MUST start a Subscribe Stream with a SUBSCRIBE message followed by any number of SUBSCRIBE_UPDATE messages.
-The publisher replies with a SUBSCRIBE_OK message once the start group is resolved, followed by any number of SUBSCRIBE_END and SUBSCRIBE_DROP messages.
+The publisher replies with a SUBSCRIBE_OK message once the start group is resolved, followed by a SUBSCRIBE_END message once the subscription ends.
 For a live track the publisher MAY withhold SUBSCRIBE_OK until the first matching group resolves the start; if the track has already ended with no matching groups, it sends SUBSCRIBE_END with no preceding SUBSCRIBE_OK.
 A rejection is a stream reset: a publisher that cannot serve the subscription (no such track, an ended broadcast, or any other refusal) MUST promptly reset the stream rather than leave it pending, so a subscriber distinguishes "pending" from "refused" by the reset, not by a timeout.
 A route claims capability rather than inventory, so a subscription for a covered path that names nothing is refused this way too.
@@ -455,9 +455,8 @@ A route claims capability rather than inventory, so a subscription for a covered
 The track's immutable publisher properties are not carried here; they are fetched once via a [Track Stream](#track-stream).
 The subscriber needs the track's TRACK_INFO (notably its timescale) to interpret FRAME messages, and MAY open the Track and Subscribe streams concurrently, buffering frames until it arrives.
 
-The publisher sends SUBSCRIBE_OK once the absolute start position is resolved, and SUBSCRIBE_END once no further groups will be produced (see [SUBSCRIBE_OK](#subscribe-ok) and [SUBSCRIBE_END](#subscribe-end)).
-The publisher closes the stream (FIN) only once every group from start to end has been accounted for, either via a Group Stream (completed or reset) or a SUBSCRIBE_DROP message.
-This MAY occur after SUBSCRIBE_END, since stragglers within the range can still be dropped.
+The publisher sends SUBSCRIBE_OK once the absolute start position is resolved, and SUBSCRIBE_END once no further groups will be produced and every Group Stream it opens for the subscription has been opened (see [SUBSCRIBE_OK](#subscribe-ok) and [SUBSCRIBE_END](#subscribe-end)).
+The publisher closes the stream (FIN) after SUBSCRIBE_END, once every counted Group Stream has finished or been reset.
 Unbounded subscriptions stay open until SUBSCRIBE_END, and either endpoint MAY reset the stream at any time.
 
 ### Fetch
@@ -1116,7 +1115,7 @@ Set to 0x0 to indicate a SUBSCRIBE_OK message.
 
 **Group**:
 The absolute sequence number of the first group that will be delivered.
-It MUST be greater than or equal to the requested start group; any groups in between are unavailable and implicitly dropped, with no separate SUBSCRIBE_DROP required.
+It MUST be greater than or equal to the requested start group; any groups in between are unavailable.
 A subscriber that requested the latest group learns the resolved sequence here.
 
 There is no matching frame field, because the start frame is never in doubt: a partial group is only delivered when it was asked for, so the subscription starts either exactly where it asked or at the beginning of a later group (see [Positions](#positions)).
@@ -1136,6 +1135,7 @@ SUBSCRIBE_END Message {
   Type (i) = 0x1
   Message Length (i)
   Group (i)
+  Stream Count (i)
 }
 ~~~
 
@@ -1147,34 +1147,14 @@ The exclusive end of the range: the absolute sequence number of the first group
 A value of 0 means the track ended before producing any groups.
 The subscriber MUST NOT wait for any group at or after this sequence.
 
-SUBSCRIBE_END bounds the range but does not by itself end the stream: the publisher MAY still send SUBSCRIBE_DROP for groups below this sequence that it cannot deliver, and FINs the stream only once every group below this sequence has been accounted for.
+**Stream Count**:
+The number of Group Streams the publisher opened for this subscription, whether they finished or were reset.
+A group that was skipped, never produced, delivered only as a datagram, or given up before its stream opened is not counted.
+A relay counts the Group Streams it opened itself, never the count it received upstream.
 
-## SUBSCRIBE_DROP
-A SUBSCRIBE_DROP message is sent by the publisher on the Subscribe Stream when groups cannot be served.
-It MAY arrive at any point after the subscription is opened, including after SUBSCRIBE_END for stragglers within the resolved range (a leading range is instead dropped implicitly by SUBSCRIBE_OK).
-
-~~~
-SUBSCRIBE_DROP Message {
-  Type (i) = 0x2
-  Message Length (i)
-  Group Start (i)
-  Group End (i)
-  Error Code (i)
-}
-~~~
-
-**Type**:
-Set to 0x2 to indicate a SUBSCRIBE_DROP message.
-
-**Group Start**:
-The first absolute group sequence in the dropped range.
-
-**Group End**:
-The last absolute group sequence in the dropped range (inclusive).
-
-**Error Code**:
-An application-specific error code.
-A value of 0 indicates no error; the groups are simply unavailable.
+The publisher MUST NOT send SUBSCRIBE_END until every Group Stream it will open for the subscription has been opened, so the count is final; it does not wait for them to finish.
+The subscriber has received every Group Stream once it has read the header of `Stream Count` of them, which MAY happen after SUBSCRIBE_END or the FIN since streams are not ordered.
+A Group Stream reset before its header arrived is never seen, so a subscriber SHOULD bound how long it waits for the rest, for example by `Subscriber Max Age`.
 
 ## FETCH
 FETCH is sent by a subscriber to request a single group from a track.
@@ -1332,6 +1312,9 @@ The `Message Length` describes the payload size on the wire.
 
 - Assigned `moq-lite-07-wip` as this draft's protocol identifier until it is finalized as `moq-lite-07`.
 - Hid routes with a `.`-prefixed segment below the requested prefix from announce discovery, and added the ANNOUNCE_REQUEST `Hidden` field to opt in.
+- Added `Stream Count` to SUBSCRIBE_END: the number of Group Streams opened for the subscription. SUBSCRIBE_END is now sent once every counted Group Stream has opened, rather than as soon as the final group is known.
+- Removed SUBSCRIBE_DROP and its type 0x2; a group without a Group Stream is not counted.
+- The Subscribe Stream FIN now follows once every counted Group Stream has finished or been reset.
 
 ## moq-lite-06
 
diff --git a/js/net/src/lite/publisher.test.ts b/js/net/src/lite/publisher.test.ts
index 30b1f5bbce..bbaf0a2731 100644
--- a/js/net/src/lite/publisher.test.ts
+++ b/js/net/src/lite/publisher.test.ts
@@ -15,8 +15,8 @@ import { Group as GroupMessage } from "./group.ts";
 import { sendOrder } from "./priority.ts";
 import { Probe as ProbeMessage } from "./probe.ts";
 import { Publisher } from "./publisher.ts";
-import { decodeSubscribeResponse, Subscribe, SubscribeUpdate } from "./subscribe.ts";
-import { ALPN_05, ALPN_06, Version } from "./version.ts";
+import { decodeSubscribeResponse, Subscribe, type SubscribeEnd, SubscribeUpdate } from "./subscribe.ts";
+import { ALPN_05, ALPN_06, ALPN_07_WIP, Version } from "./version.ts";
 
 function publish(origin: OriginProducer, path: Path.Valid) {
 	const broadcast = origin.createBroadcast(path);
@@ -85,11 +85,11 @@ test.each([Version.DRAFT_01, Version.DRAFT_03, Version.DRAFT_06])(
 );
 
 // Delivers `sequences` in the given order, finishes the track, and returns the
-// SUBSCRIBE_END boundary the publisher put on the wire.
-async function subscribeEnd(sequences: number[]): Promise<number> {
-	const pair = createMockTransportPair(ALPN_05);
+// SUBSCRIBE_END the publisher put on the wire.
+async function subscribeEnd(sequences: number[], version: Version = Version.DRAFT_05): Promise<SubscribeEnd> {
+	const pair = createMockTransportPair(version === Version.DRAFT_07 ? ALPN_07_WIP : ALPN_05);
 	const origin = new OriginProducer();
-	const publisher = new Publisher(pair.server, Version.DRAFT_05, randomHop(), origin.consume());
+	const publisher = new Publisher(pair.server, version, randomHop(), origin.consume());
 
 	const broadcast = publish(origin, Path.from("test"));
 	const track = broadcast.createTrack("video");
@@ -118,8 +118,8 @@ async function subscribeEnd(sequences: number[]): Promise<number> {
 
 	try {
 		for (;;) {
-			const resp = await decodeSubscribeResponse(client.reader, Version.DRAFT_05);
-			if ("end" in resp) return resp.end.group;
+			const resp = await decodeSubscribeResponse(client.reader, version);
+			if ("end" in resp) return resp.end;
 		}
 	} finally {
 		publisher.close();
@@ -972,19 +972,153 @@ test("lite draft-05: teardown unwinds with an undelivered update queued", async
 // A Rust subscriber feeds this value straight into `track::Producer::finish_at`, which is
 // exclusive, so an inclusive bound here silently truncates the final group across languages.
 test("lite draft-05: subscribe end is the exclusive boundary", async () => {
-	expect(await subscribeEnd([0, 1, 2])).toBe(3);
+	expect((await subscribeEnd([0, 1, 2])).group).toBe(3);
 });
 
 // recvGroup is arrival-ordered, so the boundary has to clear the max sequence delivered,
 // not the last one seen. Otherwise the boundary lands on a group already on the wire.
 test("lite draft-05: subscribe end clears the max sequence when groups arrive out of order", async () => {
-	expect(await subscribeEnd([0, 2, 1])).toBe(3);
+	expect((await subscribeEnd([0, 2, 1])).group).toBe(3);
 });
 
 // 0 is the only encoding for "no groups at all"; an inclusive bound cannot express it
 // without colliding with a track whose sole group was sequence 0.
 test("lite draft-05: subscribe end is 0 when no groups were produced", async () => {
-	expect(await subscribeEnd([])).toBe(0);
+	expect((await subscribeEnd([])).group).toBe(0);
+});
+
+// The count is of group streams opened, not of groups below the end: a group the track
+// never produced has no stream and is not counted.
+test("lite draft-07: subscribe end counts the group streams opened", async () => {
+	const end = await subscribeEnd([0, 2], Version.DRAFT_07);
+	expect([end.group, end.streams]).toEqual([3, 2]);
+});
+
+test("lite draft-07: subscribe end counts zero streams when no groups were produced", async () => {
+	const end = await subscribeEnd([], Version.DRAFT_07);
+	expect([end.group, end.streams]).toEqual([0, 0]);
+});
+
+// finishAt names the end while groups below it are still being produced. The count
+// cannot include a stream that has not opened, so SUBSCRIBE_END waits for them.
+test("lite draft-07: subscribe end waits for groups below a declared finish", async () => {
+	const pair = createMockTransportPair(ALPN_07_WIP);
+	const origin = new OriginProducer();
+	const publisher = new Publisher(pair.server, Version.DRAFT_07, randomHop(), origin.consume());
+	const broadcast = publish(origin, Path.from("test"));
+	const track = broadcast.createTrack("video");
+
+	const client = await Stream.open(pair.client);
+	const server = await Stream.accept(pair.server);
+	if (!server) throw new Error("publisher never accepted the subscribe stream");
+	void publisher.runSubscribe(
+		new Subscribe({ id: 0n, broadcast: Path.from("test"), track: "video", priority: 0 }),
+		server,
+	);
+
+	try {
+		const first = new GroupProducer(0);
+		first.writeString("hello");
+		first.close();
+		track.writeGroup(first);
+		track.finishAt(2);
+
+		const start = await decodeSubscribeResponse(client.reader, Version.DRAFT_07);
+		expect("start" in start).toBe(true);
+		const pending = decodeSubscribeResponse(client.reader, Version.DRAFT_07);
+		const early = await Promise.race([pending, new Promise((resolve) => setTimeout(resolve, IDLE_MS))]);
+		expect(early).toBeUndefined();
+
+		const second = new GroupProducer(1);
+		second.writeString("hello");
+		second.close();
+		track.writeGroup(second);
+		track.close();
+
+		const resp = await pending;
+		if (!("end" in resp)) throw new Error("expected SUBSCRIBE_END");
+		expect([resp.end.group, resp.end.streams]).toEqual([2, 2]);
+	} finally {
+		publisher.close();
+		client.close();
+	}
+});
+
+// Serves one group with its stream open held until `open(ok)`, and returns the pending
+// SUBSCRIBE_END plus the call that lets the open succeed or fail.
+async function heldOpenEnd() {
+	const pair = createMockTransportPair(ALPN_07_WIP);
+	const origin = new OriginProducer();
+	const publisher = new Publisher(pair.server, Version.DRAFT_07, randomHop(), origin.consume());
+	const broadcast = publish(origin, Path.from("test"));
+	const track = broadcast.createTrack("video");
+
+	let open!: (ok: boolean) => void;
+	const opened = new Promise<boolean>((resolve) => {
+		open = resolve;
+	});
+	const createUni = pair.server.createUnidirectionalStream.bind(pair.server);
+	spyOn(pair.server, "createUnidirectionalStream").mockImplementation(async (options) => {
+		if (!(await opened)) throw new Error("no stream credit");
+		return createUni(options);
+	});
+
+	const client = await Stream.open(pair.client);
+	const server = await Stream.accept(pair.server);
+	if (!server) throw new Error("publisher never accepted the subscribe stream");
+	void publisher.runSubscribe(
+		new Subscribe({ id: 0n, broadcast: Path.from("test"), track: "video", priority: 0 }),
+		server,
+	);
+
+	const group = new GroupProducer(0);
+	group.writeString("hello");
+	group.close();
+	track.writeGroup(group);
+	track.close();
+
+	const start = await decodeSubscribeResponse(client.reader, Version.DRAFT_07);
+	expect("start" in start).toBe(true);
+	const end = decodeSubscribeResponse(client.reader, Version.DRAFT_07);
+
+	return {
+		end,
+		open,
+		close() {
+			publisher.close();
+			client.close();
+		},
+	};
+}
+
+// The count is final only once no served group is still waiting for its stream, so
+// SUBSCRIBE_END waits for the open.
+test("lite draft-07: subscribe end waits for every group stream to open", async () => {
+	const held = await heldOpenEnd();
+	try {
+		const early = await Promise.race([held.end, new Promise((resolve) => setTimeout(resolve, IDLE_MS))]);
+		expect(early).toBeUndefined();
+
+		held.open(true);
+		const resp = await held.end;
+		if (!("end" in resp)) throw new Error("expected SUBSCRIBE_END");
+		expect([resp.end.group, resp.end.streams]).toEqual([1, 1]);
+	} finally {
+		held.close();
+	}
+});
+
+// A group that never gets a stream owes the subscriber nothing, so it is not counted.
+test("lite draft-07: a group whose stream never opened is not counted", async () => {
+	const held = await heldOpenEnd();
+	try {
+		held.open(false);
+		const resp = await held.end;
+		if (!("end" in resp)) throw new Error("expected SUBSCRIBE_END");
+		expect([resp.end.group, resp.end.streams]).toEqual([1, 0]);
+	} finally {
+		held.close();
+	}
 });
 
 /** One group stream the publisher put on the wire. */
diff --git a/js/net/src/lite/publisher.ts b/js/net/src/lite/publisher.ts
index b9fb8dc098..f35fce72be 100644
--- a/js/net/src/lite/publisher.ts
+++ b/js/net/src/lite/publisher.ts
@@ -27,7 +27,15 @@ import {
 	SubscribeUpdate,
 } from "./subscribe.ts";
 import { TrackInfo as TrackInfoMessage, type Track as TrackMessage } from "./track.ts";
-import { hasAnnounceId, hasAnnounceOk, hasDatagrams, hasProbeRtt, resolvesStart, Version } from "./version.ts";
+import {
+	hasAnnounceId,
+	hasAnnounceOk,
+	hasDatagrams,
+	hasProbeRtt,
+	hasStreamCount,
+	resolvesStart,
+	Version,
+} from "./version.ts";
 
 // Where each originated route lands under the requested prefix: its suffix beneath
 // the prefix, or the empty suffix for a route above it, where the most specific
@@ -64,7 +72,7 @@ function zigzag(delta: bigint): bigint {
 	return delta >= 0n ? delta << 1n : (-delta << 1n) - 1n;
 }
 
-/** What {@link Publisher.runGroup} needs to serve one group. */
+/** What {@link Publisher.openGroup} and {@link Publisher.serveGroup} need to serve one group. */
 interface RunGroup {
 	/** The subscription ID. */
 	sub: bigint;
@@ -737,6 +745,13 @@ export class Publisher {
 		let startSent = false;
 		let endSent = false;
 
+		// Lite-07+ counts the group streams in SUBSCRIBE_END, so it goes out only once every
+		// served group has opened its stream or given up, and a cap holding groups back
+		// delays it until they are released.
+		const countStreams = hasStreamCount(this.version);
+		let streams = 0;
+		const opening = new Set<Promise<unknown>>();
+
 		// The track's exclusive final boundary. A Rust subscriber feeds SUBSCRIBE_END
 		// straight into finish_at, so it must name the track's boundary (which counts
 		// datagram sequences too), not the delivered range: a subscription cap can hold
@@ -744,17 +759,20 @@ export class Publisher {
 		// before the producer declared it.
 		const boundary = () => track.final() ?? (track.latest() ?? -1) + 1;
 
-		// SUBSCRIBE_END names that boundary, which a cap can hold groups back from, so it goes
-		// out as soon as the producer finishes and the subscription keeps serving whatever a
-		// later cap raise releases (see the Rust publisher's Recv::Boundary).
+		// Before lite-07, SUBSCRIBE_END names that boundary, which a cap can hold groups back
+		// from, so it goes out as soon as the producer finishes and the subscription keeps
+		// serving whatever a later cap raise releases (see the Rust publisher's Recv::Boundary).
 		const sendEnd = async (): Promise<boolean> => {
 			endSent = true;
-			if (emitRange) {
-				return controls.response(
-					encodeSubscribeResponse(stream, { end: new SubscribeEnd(boundary()) }, this.version),
-				);
-			}
-			return true;
+			if (!emitRange) return true;
+			return controls.response(
+				(async () => {
+					// A group that gives up before its stream opens is never counted.
+					if (countStreams) while (opening.size > 0) await Promise.all(opening);
+					const end = new SubscribeEnd(boundary(), streams);
+					await encodeSubscribeResponse(stream, { end }, this.version);
+				})(),
+			);
 		};
 
 		// One ranking for the whole subscription, shared by every group it serves.
@@ -810,9 +828,10 @@ export class Publisher {
 					case "error":
 						throw recv.error;
 					case "idle":
-						// An end declared ahead of the live edge goes out as soon as it is
-						// known, while the remaining groups are still being produced.
-						if (!endSent && track.final() !== undefined) {
+						// Before lite-07, an end declared ahead of the live edge goes out as
+						// soon as it is known, while the remaining groups are still being
+						// produced. The lite-07 count is not final until those groups open.
+						if (!endSent && !countStreams && track.final() !== undefined) {
 							if (!(await sendEnd())) return;
 							continue;
 						}
@@ -821,7 +840,7 @@ export class Publisher {
 					case "boundary":
 						// The producer finished but is still holding groups above the cap.
 						// Declare the boundary, then wait for an update to release them.
-						if (!endSent) {
+						if (!endSent && !countStreams) {
 							if (!(await sendEnd())) return;
 							continue;
 						}
@@ -868,7 +887,7 @@ export class Publisher {
 						return;
 				}
 
-				const task = this.#runGroup({
+				const options: RunGroup = {
 					sub,
 					group,
 					timescale,
@@ -876,9 +895,20 @@ export class Publisher {
 					unsubscribed,
 					start: range.start,
 					end: range.end,
+				};
+				// `opening` settles when the stream opens so the lite-07 count can be sent.
+				// `groups` covers the serve too, so the FIN still waits for every stream to
+				// finish or reset, including one that has not opened yet.
+				const opened = this.#openGroup(options);
+				const task = opened.then(async (writer) => {
+					if (!writer) return;
+					streams += 1;
+					await this.#serveGroup(writer, options);
 				});
 				groups.add(task);
 				void task.finally(() => groups.delete(task));
+				opening.add(opened);
+				void opened.finally(() => opening.delete(opened));
 			}
 		} finally {
 			if (!finished) unsubscribe();
@@ -930,7 +960,7 @@ export class Publisher {
 				// relays re-serve with the same window.
 				maxAge: info.maxAge,
 				// Lite05 mandates per-frame timestamps. Advertise the track's timescale;
-				// `#runGroup` emits each frame converted to it.
+				// `#serveGroup` emits each frame converted to it.
 				timescale: info.timescale,
 			});
 		})();
@@ -959,7 +989,7 @@ export class Publisher {
 				const datagram = await track.recvDatagram();
 				if (!datagram) return; // Track finished; #runTrack tears the subscription down.
 
-				// Convert the timestamp to the track's advertised timescale, matching #runGroup.
+				// Convert the timestamp to the track's advertised timescale, matching #serveGroup.
 				const ts = Math.round(datagram.timestamp.as(timescale));
 				const body = new DatagramMessage(sub, datagram.sequence, ts, datagram.payload).encode();
 
@@ -1009,15 +1039,13 @@ export class Publisher {
 	}
 
 	/**
-	 * Serves one group on its own unidirectional stream.
+	 * Opens the unidirectional stream for one group, or closes the group and resolves
+	 * `undefined` when it cannot get one.
 	 *
 	 * @internal
 	 */
-	async #runGroup(options: RunGroup) {
-		const { sub, group, timescale, priority, unsubscribed, start: startFrame, end: endFrame } = options;
-		// This model holds whole groups, so frame `startFrame` is always reachable unless
-		// the group ends first. Declaring it up front keeps the stream self-describing.
-		const msg = new GroupMessage({ subscribe: sub, sequence: group.sequence, frameStart: startFrame });
+	async #openGroup(options: RunGroup): Promise<Writer | undefined> {
+		const { group, priority, unsubscribed } = options;
 		try {
 			// The transport drains streams by send order, so this is what makes a high-priority
 			// track (and a newer group within it) win the link when there isn't room for both.
@@ -1031,79 +1059,88 @@ export class Publisher {
 				cancel: unsubscribed,
 				waitUntilAvailable: false,
 			});
-			if (!stream) {
-				group.close(new Error("no stream slot"));
-				return;
-			}
-
-			// Everything past this point runs inside the cleanup scope, so a failure never leaves
-			// a finished group's stream being ranked.
-			try {
-				// A SUBSCRIBE_UPDATE re-ranks the subscription, so a group already on the wire
-				// follows it too rather than keeping a stale rank until it finishes.
-				priority.add(stream, group.sequence);
+			if (!stream) group.close(new Error("no stream slot"));
+			return stream;
+		} catch (err: unknown) {
+			group.close(error(err));
+			return undefined;
+		}
+	}
 
-				await hooks.guardGroup(
-					group,
-					(async () => {
-						await stream.u53(0); // stream type
-						await msg.encode(stream, this.version);
-					})(),
-				);
+	/**
+	 * Serves one group on the stream {@link #openGroup} opened for it.
+	 *
+	 * @internal
+	 */
+	async #serveGroup(stream: Writer, options: RunGroup) {
+		const { sub, group, timescale, priority, start: startFrame, end: endFrame } = options;
+		// This model holds whole groups, so frame `startFrame` is always reachable unless
+		// the group ends first. Declaring it up front keeps the stream self-describing.
+		const msg = new GroupMessage({ subscribe: sub, sequence: group.sequence, frameStart: startFrame });
+		// Everything past this point runs inside the cleanup scope, so a failure never leaves
+		// a finished group's stream being ranked.
+		try {
+			// A SUBSCRIBE_UPDATE re-ranks the subscription, so a group already on the wire
+			// follows it too rather than keeping a stale rank until it finishes.
+			priority.add(stream, group.sequence);
+
+			await hooks.guardGroup(
+				group,
+				(async () => {
+					await stream.u53(0); // stream type
+					await msg.encode(stream, this.version);
+				})(),
+			);
 
-				// Lite05+ prefixes every frame with a zigzag-delta timestamp at the track's
-				// advertised timescale; older drafts omit it.
-				const timestamps = supportsTrackStream(this.version);
-				let prevTs = 0n;
-				// Whether the cursor ever reached the requested start, which decides how the
-				// end of the group is read below.
-				let reached = startFrame === 0;
-
-				for (;;) {
-					const read = await race([hooks.readGroupFrame(group), stream.closed]);
-					if (!read) {
-						// The group ended before the frame the subscriber asked to start
-						// at, so this publisher can't serve the range at all. FINning here
-						// would claim an empty group under that index; reset so it reads
-						// as the gap it is.
-						if (!reached) throw new Error(`group ended before frame ${startFrame}`);
-						break;
-					}
+			// Lite05+ prefixes every frame with a zigzag-delta timestamp at the track's
+			// advertised timescale; older drafts omit it.
+			const timestamps = supportsTrackStream(this.version);
+			let prevTs = 0n;
+			// Whether the cursor ever reached the requested start, which decides how the
+			// end of the group is read below.
+			let reached = startFrame === 0;
 
-					try {
-						// A group that ends exactly at the start is a valid, empty range.
-						if (read.sequence + 1 >= startFrame) reached = true;
-						// Frames below the requested start were excluded, and the receiver
-						// numbers what it gets from `startFrame`.
-						if (read.sequence < startFrame) continue;
-						if (endFrame !== undefined && read.sequence > endFrame) break;
-
-						if (timestamps) {
-							// Convert each frame to the track's advertised timescale.
-							const ts = BigInt(Math.round(read.frame.timestamp.as(timescale)));
-							await hooks.guardGroup(group, stream.u62(zigzag(ts - prevTs)));
-							prevTs = ts;
-						}
+			for (;;) {
+				const read = await race([hooks.readGroupFrame(group), stream.closed]);
+				if (!read) {
+					// The group ended before the frame the subscriber asked to start
+					// at, so this publisher can't serve the range at all. FINning here
+					// would claim an empty group under that index; reset so it reads
+					// as the gap it is.
+					if (!reached) throw new Error(`group ended before frame ${startFrame}`);
+					break;
+				}
 
-						await hooks.guardGroup(group, stream.u53(read.frame.payload.byteLength));
-						await hooks.guardGroup(group, stream.write(read.frame.payload));
-					} finally {
-						read.complete();
+				try {
+					// A group that ends exactly at the start is a valid, empty range.
+					if (read.sequence + 1 >= startFrame) reached = true;
+					// Frames below the requested start were excluded, and the receiver
+					// numbers what it gets from `startFrame`.
+					if (read.sequence < startFrame) continue;
+					if (endFrame !== undefined && read.sequence > endFrame) break;
+
+					if (timestamps) {
+						// Convert each frame to the track's advertised timescale.
+						const ts = BigInt(Math.round(read.frame.timestamp.as(timescale)));
+						await hooks.guardGroup(group, stream.u62(zigzag(ts - prevTs)));
+						prevTs = ts;
 					}
-				}
 
-				stream.close();
-				group.close();
-			} catch (err: unknown) {
-				const e = error(err);
-				stream.reset(e);
-				group.close(e);
-			} finally {
-				priority.remove(stream);
+					await hooks.guardGroup(group, stream.u53(read.frame.payload.byteLength));
+					await hooks.guardGroup(group, stream.write(read.frame.payload));
+				} finally {
+					read.complete();
+				}
 			}
+
+			stream.close();
+			group.close();
 		} catch (err: unknown) {
 			const e = error(err);
+			stream.reset(e);
 			group.close(e);
+		} finally {
+			priority.remove(stream);
 		}
 	}
 
diff --git a/js/net/src/lite/subscribe.test.ts b/js/net/src/lite/subscribe.test.ts
index 7f8f5e7fa5..afd4c7c0a6 100644
--- a/js/net/src/lite/subscribe.test.ts
+++ b/js/net/src/lite/subscribe.test.ts
@@ -152,10 +152,30 @@ test("SubscribeStart round-trips on draft-05", async () => {
 });
 
 test("SubscribeEnd round-trips on draft-05", async () => {
-	const got = await responseRoundtrip(Version.DRAFT_05, { end: new SubscribeEnd(7) });
+	// Type, length, group: no stream count before draft-07.
+	expect(await encode(Version.DRAFT_05, { end: new SubscribeEnd(7, 3) })).toEqual(new Uint8Array([1, 1, 7]));
+	const got = await responseRoundtrip(Version.DRAFT_05, { end: new SubscribeEnd(7, 3) });
 	expect("end" in got).toBe(true);
 	if (!("end" in got)) throw new Error("expected end");
-	expect(got.end.group).toBe(7);
+	expect([got.end.group, got.end.streams]).toEqual([7, 0]);
+});
+
+test("SubscribeEnd carries the stream count on draft-07", async () => {
+	expect(await encode(Version.DRAFT_07, { end: new SubscribeEnd(7, 3) })).toEqual(new Uint8Array([1, 2, 7, 3]));
+	const got = await responseRoundtrip(Version.DRAFT_07, { end: new SubscribeEnd(7, 3) });
+	if (!("end" in got)) throw new Error("expected end");
+	expect([got.end.group, got.end.streams]).toEqual([7, 3]);
+});
+
+test("SubscribeDrop is gone on draft-07", async () => {
+	const drop: SubscribeResponse = { drop: new SubscribeDrop({ start: 1, end: 3, error: 0 }) };
+	await expect(encode(Version.DRAFT_07, drop)).rejects.toThrow();
+
+	// A draft-06 DROP is an unknown response type on draft-07.
+	const wire06 = await encode(Version.DRAFT_06, drop);
+	await expect(decodeSubscribeResponse(new Reader(undefined, wire06), Version.DRAFT_07)).rejects.toThrow(
+		"unknown subscribe response type: 2",
+	);
 });
 
 test("SubscribeDrop is type 0x2 on draft-05 and 0x1 on draft-04", async () => {
diff --git a/js/net/src/lite/subscribe.ts b/js/net/src/lite/subscribe.ts
index c5a5b37b78..3d1a1f2711 100644
--- a/js/net/src/lite/subscribe.ts
+++ b/js/net/src/lite/subscribe.ts
@@ -1,7 +1,7 @@
 import * as Path from "../path.ts";
 import type { Reader, Writer } from "../stream.ts";
 import * as Message from "./message.ts";
-import { hasFrameBounds, hasGroupOrder, resolvesStart, Version } from "./version.ts";
+import { hasFrameBounds, hasGroupOrder, hasStreamCount, resolvesStart, Version } from "./version.ts";
 
 /**
  * Encode the `Group Start` field shared by SUBSCRIBE and SUBSCRIBE_UPDATE.
@@ -456,24 +456,35 @@ export class SubscribeEnd {
 	/** The exclusive final group sequence: the first sequence that will never be produced. */
 	group: number;
 
-	constructor(group: number) {
+	/**
+	 * The number of group streams the publisher opened for this subscription.
+	 * Draft-07+ only; not on the wire before, where it decodes as 0.
+	 */
+	streams: number;
+
+	constructor(group: number, streams = 0) {
 		this.group = group;
+		this.streams = streams;
 	}
 
-	async encode(w: Writer): Promise<void> {
+	async encode(w: Writer, version: Version): Promise<void> {
 		return Message.encode(w, async (w) => {
 			await w.u53(this.group);
+			if (hasStreamCount(version)) await w.u53(this.streams);
 		});
 	}
 
-	static async decode(r: Reader): Promise<SubscribeEnd> {
-		return Message.decode(r, async (r) => new SubscribeEnd(await r.u53()));
+	static async decode(r: Reader, version: Version): Promise<SubscribeEnd> {
+		return Message.decode(
+			r,
+			async (r) => new SubscribeEnd(await r.u53(), hasStreamCount(version) ? await r.u53() : 0),
+		);
 	}
 }
 
 /// Indicates that one or more groups have been dropped.
 ///
-/// Draft03+ only.
+/// Draft-03 to Draft-06 only: Draft-07 counts group streams in SUBSCRIBE_END instead.
 export class SubscribeDrop {
 	start: number;
 	end: number;
@@ -510,8 +521,9 @@ export class SubscribeDrop {
  *
  * The discriminator is version-dependent:
  * - Draft-03/04: `0x0` SUBSCRIBE_OK, `0x1` SUBSCRIBE_DROP.
- * - Draft-05+: `0x0` SUBSCRIBE_START, `0x1` SUBSCRIBE_END, `0x2` SUBSCRIBE_DROP
+ * - Draft-05/06: `0x0` SUBSCRIBE_START, `0x1` SUBSCRIBE_END, `0x2` SUBSCRIBE_DROP
  *   (SUBSCRIBE_OK was removed; acceptance is implicit).
+ * - Draft-07+: `0x0` SUBSCRIBE_START, `0x1` SUBSCRIBE_END (SUBSCRIBE_DROP was removed).
  */
 export type SubscribeResponse =
 	| { ok: SubscribeOk }
@@ -548,12 +560,12 @@ export async function encodeSubscribeResponse(w: Writer, resp: SubscribeResponse
 				await resp.start.encode(w);
 			} else if ("end" in resp) {
 				await w.u53(0x1);
-				await resp.end.encode(w);
-			} else if ("drop" in resp) {
+				await resp.end.encode(w, version);
+			} else if ("drop" in resp && !hasStreamCount(version)) {
 				await w.u53(0x2);
 				await resp.drop.encode(w);
 			} else {
-				throw new Error("SUBSCRIBE_OK not supported for this version");
+				throw new Error("subscribe response not supported for this version");
 			}
 			break;
 	}
@@ -582,8 +594,9 @@ export async function decodeSubscribeResponse(r: Reader, version: Version): Prom
 				case 0x0:
 					return { start: await SubscribeStart.decode(r) };
 				case 0x1:
-					return { end: await SubscribeEnd.decode(r) };
+					return { end: await SubscribeEnd.decode(r, version) };
 				case 0x2:
+					if (hasStreamCount(version)) throw new Error(`unknown subscribe response type: ${typ}`);
 					return { drop: await SubscribeDrop.decode(r) };
 				default:
 					throw new Error(`unknown subscribe response type: ${typ}`);
diff --git a/js/net/src/lite/version.ts b/js/net/src/lite/version.ts
index 4fda174243..28d0470dd9 100644
--- a/js/net/src/lite/version.ts
+++ b/js/net/src/lite/version.ts
@@ -230,6 +230,25 @@ export function hasHidden(version: Version): boolean {
 	}
 }
 
+/**
+ * Whether SUBSCRIBE_END carries the subscription's group stream count, sent once every
+ * counted stream is open, in place of SUBSCRIBE_DROP. Added in lite-07.
+ */
+export function hasStreamCount(version: Version): boolean {
+	// Explicitly list older versions so future versions keep the lite-07+ behavior.
+	switch (version) {
+		case Version.DRAFT_01:
+		case Version.DRAFT_02:
+		case Version.DRAFT_03:
+		case Version.DRAFT_04:
+		case Version.DRAFT_05:
+		case Version.DRAFT_06:
+			return false;
+		default:
+			return true;
+	}
+}
+
 /// The WebTransport subprotocol identifier for moq-lite.
 /// Version negotiation still happens via SETUP when this is used.
 export const ALPN = "moql";
diff --git a/quest/m1/README.md b/quest/m1/README.md
index 1f4d300698..5ef1b79dd9 100644
--- a/quest/m1/README.md
+++ b/quest/m1/README.md
@@ -19,8 +19,8 @@ transport, benchmark tooling); worktrees isolate commits, not semantics.
 
 - [Missing fetch group](/quest/m1/fetch-missing-group.md) - HTTP /fetch answers 404 and `moq fetch` fails cleanly for a group the track lacks
 - [libmoq hidden opt-in](/quest/m1/libmoq-hidden.md) - `moq_origin_announced` takes a `hidden` flag so C callers can list `.`-named broadcasts
-- [lite-07 stream count](/quest/m1/lite-stream-count.md) - moq-lite-07 replaces SUBSCRIBE_DROP with a group-stream count in SUBSCRIBE_END, like moq-transport
 - [Announce compression](/quest/m1/announce-compression.md) - a lite-07 announce reuses the path head and hop-chain tail of a live announcement on its stream instead of resending them
+- [lite-07 count settle](/quest/m1/lite-count-settle.md) - moq-lite-07 subscribers stop waiting for a subscription's tail once SUBSCRIBE_END's stream count is reached
 - [Session death error](/quest/m1/session-death-error.md) - a dying session ends its tracks with its own error in Rust and JS, never a clean end, `Dropped`, or `Cancel`
 - [JS bare FIN](/quest/m1/js-bare-fin.md) - a `@moq/net` subscriber aborts a track whose subscribe stream FINs before its declared end, like Rust
 - [Track tail interop](/quest/m1/track-tail-interop.md) - a Rust publisher ending a track with a group in flight is read to its end by the JS subscriber, and the reverse, in `just test interop`
diff --git a/quest/m1/announce-compression.md b/quest/m1/announce-compression.md
index c7c61169c1..cb0dc30bc7 100644
--- a/quest/m1/announce-compression.md
+++ b/quest/m1/announce-compression.md
@@ -102,7 +102,5 @@ Hops {
 
 ## Related
 
-- [lite-07 stream count](/quest/m1/lite-stream-count.md) - the other lite-07
-  wire change, landing in the same unreleased version
 - [Relay memory](/quest/m1/relay-memory.md) - route state per relay, which
   this leaves unchanged
diff --git a/quest/m1/lite-count-settle.md b/quest/m1/lite-count-settle.md
new file mode 100644
index 0000000000..0ff1f18a71
--- /dev/null
+++ b/quest/m1/lite-count-settle.md
@@ -0,0 +1,29 @@
+# [S] lite-07 subscribers settle on the stream count
+
+## Goal
+
+On moq-lite-07, Rust and JS subscribers stop waiting for a subscription's tail
+once they have read the headers of as many group streams as SUBSCRIBE_END
+counts, so a group the publisher skipped or never opened costs no grace. A
+late stream below the end is still accepted within the grace, which stays for
+a stream reset before its header arrived. lite-05 and -06 keep the DROP
+accounting JS already has and Rust track tail adds.
+
+## Plan
+
+- Both publishers already send the count, and both subscribers decode it
+  (`lite::SubscribeEnd::streams` in Rust, `SubscribeEnd.streams` in JS) and
+  ignore it.
+- JS track tail has landed, and its `Tail` already counts streams. Rust track
+  tail has landed (`rs/moq-net/src/tail.rs`), so lite-07's completion check
+  becomes "headers read >= Stream Count" after SUBSCRIBE_END, in place of
+  every sequence from start to end being covered. The Rust subscriber needs
+  the same count.
+- Tests in both languages: a late stream after SUBSCRIBE_END, a skipped group
+  that settles without the grace, a reset stream, and a count of zero. Add
+  the Rust-JS case to the track tail interop test.
+
+## Related
+
+- [Track tail interop](/quest/m1/track-tail-interop.md) - the Rust-JS case this adds its count check to
+- [Reliable stream reset](/quest/m1/quic/reliable-reset.md) - makes the count exact by keeping a reset stream's header
diff --git a/quest/m1/lite-stream-count.md b/quest/m1/lite-stream-count.md
deleted file mode 100644
index 053984fc56..0000000000
--- a/quest/m1/lite-stream-count.md
+++ /dev/null
@@ -1,43 +0,0 @@
-# [M] moq-lite-07 counts group streams instead of dropping groups
-
-## Goal
-
-On moq-lite-07, a subscriber knows a subscription has delivered everything
-once it has seen as many group streams as the publisher opened, the way
-moq-transport's PUBLISH_DONE Stream Count works. SUBSCRIBE_DROP is gone from
-lite-07: a group the publisher skipped or never opened is simply not counted,
-so nothing has to name it. Published versions (lite-01 to -06) keep decoding
-SUBSCRIBE_DROP unchanged.
-
-Where reliable reset is negotiated, the count is exact and the subscriber
-waits for nothing else. Where it is not (browsers today), a stream reset
-before its header arrived is still invisible, so the track-tail grace stays.
-
-## Plan
-
-- Wire: SUBSCRIBE_END gains `Stream Count`, the number of group streams the
-  publisher opened for this subscription. The publisher sends it once every
-  group stream below the end has been opened (not finished), like
-  PUBLISH_DONE, so the boundary arrives slightly later than today. Remove
-  SUBSCRIBE_DROP and its type from lite-07 and reword the Subscribe Stream
-  section: the FIN follows once every counted stream has finished or been
-  reset. lite-07 is still work-in-progress (`moq-lite-07-wip`), so this changes it in place; update
-  `drafts/draft-lcurley-moq-lite.md` and its changelog.
-- Rust and JS publishers count the streams they open per subscription and
-  send the count; a relay counts its own downstream streams, never forwarding
-  the upstream count.
-- Subscribers on lite-07 stop waiting once the count is reached, accepting a
-  late stream below the end within the grace. On lite-05 and -06, the
-  DROP accounting already in moq-net (`rs/moq-net/src/tail.rs`) and `@moq/net`
-  (`js/net/src/tail.ts`) stays as it is.
-- Tests in both languages: a late stream after SUBSCRIBE_END, a skipped group
-  that is never counted, a reset stream, and a count of zero. Add a Rust-JS
-  interop case.
-
-This lands before lite-07 is finalized. Published drafts keep SUBSCRIBE_DROP,
-which moq-net and `@moq/net` already account. lite-07 replaces it with the
-count in both. Rust still does not send SUBSCRIBE_DROP.
-
-## Related
-
-- [Reliable stream reset](/quest/m1/quic/reliable-reset.md) - makes the count exact by keeping a reset stream's header
diff --git a/rs/moq-net/src/lite/publisher.rs b/rs/moq-net/src/lite/publisher.rs
index a841eefa34..c0ec29e3c2 100644
--- a/rs/moq-net/src/lite/publisher.rs
+++ b/rs/moq-net/src/lite/publisher.rs
@@ -3,7 +3,10 @@ use crate::{SessionError, announce, frame, group, origin, track};
 use std::{
 	collections::HashMap,
 	ops::Bound,
-	sync::Arc,
+	sync::{
+		Arc,
+		atomic::{AtomicU64, Ordering},
+	},
 	task::{Context, Poll, ready},
 	time::Duration,
 };
@@ -1151,6 +1154,7 @@ impl<S: crate::transport::poll::Session> SubscribeServe<S> {
 						track_priority_seen: msg.priority,
 						version: self.shared.version,
 						timescale,
+						opens: Default::default(),
 					};
 
 					let run = TrackRun::new(sub, track, Bounds::from(&msg), track_priority_tx);
@@ -2134,6 +2138,18 @@ struct Subscription<S: crate::transport::poll::Session> {
 	/// Negotiated timestamp scale for this track. `Some(_)` on lite-05+ after
 	/// TRACK_INFO; used to validate per-frame timestamps before encoding.
 	timescale: Option<crate::Timescale>,
+	/// The group streams this subscription opened, shared by every group it serves.
+	opens: Arc<Opens>,
+}
+
+/// Counts a subscription's group streams for lite-07's SUBSCRIBE_END.
+///
+/// A group is pending from the moment it is queued until its stream opens, or it gives
+/// up first (expired, or the open failed) and is never counted.
+#[derive(Default)]
+struct Opens {
+	pending: AtomicU64,
+	opened: AtomicU64,
 }
 
 impl<S: crate::transport::poll::Session> Subscription<S> {
@@ -2216,6 +2232,9 @@ struct TrackRun<S: crate::transport::poll::Session> {
 	emit_range: bool,
 	start_sent: bool,
 	end_sent: bool,
+	// Lite07+ sends SUBSCRIBE_END with the stream count instead of as soon as the
+	// boundary is known, once every group below it has opened its stream.
+	count_streams: bool,
 	// Serve datagrams off this same subscriber, but only on lite-05+ over a
 	// datagram-capable transport (qmux/WebSocket/TCP/UDS report size 0). No group
 	// fallback: otherwise off.
@@ -2237,6 +2256,7 @@ impl<S: crate::transport::poll::Session> TrackRun<S> {
 		track.end_at(bounds.end_group.map_or(Bound::Unbounded, Bound::Included));
 
 		let emit_range = ctx.version.has_track_stream();
+		let count_streams = ctx.version.has_stream_count();
 		let datagrams = ctx.version.has_datagrams() && ctx.session.max_datagram_size() > 0;
 
 		Self {
@@ -2248,6 +2268,7 @@ impl<S: crate::transport::poll::Session> TrackRun<S> {
 			emit_range,
 			start_sent: false,
 			end_sent: false,
+			count_streams,
 			datagrams,
 			children: kio::Tasks::new(),
 		}
@@ -2300,7 +2321,7 @@ impl<S: crate::transport::poll::Session> TrackRun<S> {
 			// group and, when enabled, the next best-effort datagram. Groups are polled
 			// first so a datagram burst can't starve them; datagrams flow whenever no
 			// group is ready (including while groups are parked above the cap).
-			let emit_boundary = self.emit_range && !self.end_sent;
+			let emit_boundary = self.emit_range && !self.end_sent && !self.count_streams;
 			if let Poll::Ready(res) = poll_recv_next(&mut self.track, self.datagrams, emit_boundary, waiter) {
 				match res? {
 					Recv::Group(mut group) => {
@@ -2352,7 +2373,22 @@ impl<S: crate::transport::poll::Session> TrackRun<S> {
 						self.end_sent = true;
 						stream
 							.writer
-							.buffer(&lite::SubscribeResponse::End(lite::SubscribeEnd { group }))?;
+							.buffer(&lite::SubscribeResponse::End(lite::SubscribeEnd { group, streams: 0 }))?;
+					}
+					Recv::Finished if self.count_streams && !self.end_sent => {
+						// The count is final only once no served group is still waiting to
+						// open its stream. A group that gives up first is never counted, so
+						// the subscriber is not left waiting for it. The child's wake
+						// re-polls this loop.
+						if self.ctx.opens.pending.load(Ordering::Relaxed) > 0 {
+							return Poll::Pending;
+						}
+						let group = ready!(self.track.poll_finished(waiter))?;
+						let streams = self.ctx.opens.opened.load(Ordering::Relaxed);
+						self.end_sent = true;
+						stream
+							.writer
+							.buffer(&lite::SubscribeResponse::End(lite::SubscribeEnd { group, streams }))?;
 					}
 					Recv::Finished => return Poll::Ready(Ok(TrackEnd::Finished)),
 				}
@@ -2410,6 +2446,7 @@ impl<S: crate::transport::poll::Session> GroupServe<S> {
 		priority: PriorityHandle,
 		group: group::Consumer,
 	) -> Self {
+		ctx.opens.pending.fetch_add(1, Ordering::Relaxed);
 		Self {
 			ctx,
 			priority,
@@ -2421,6 +2458,14 @@ impl<S: crate::transport::poll::Session> GroupServe<S> {
 		}
 	}
 
+	/// Leave [`GroupState::Open`], counting the stream if it opened.
+	fn settle_open(&mut self, opened: bool) {
+		self.ctx.opens.pending.fetch_sub(1, Ordering::Relaxed);
+		if opened {
+			self.ctx.opens.opened.fetch_add(1, Ordering::Relaxed);
+		}
+	}
+
 	/// Serve the group, aborting the stream with the real reason (Old, Lagged,
 	/// Evicted, ...) on failure so the subscriber can tell a truncated group from
 	/// a routine cancel. Without this the Writer's Drop fallback would report
@@ -2430,6 +2475,7 @@ impl<S: crate::transport::poll::Session> GroupServe<S> {
 			match &mut self.state {
 				GroupState::Open => {
 					if self.group.poll_expired(waiter) {
+						self.settle_open(false);
 						self.state = GroupState::Done;
 						return Poll::Ready(Err(Error::Old));
 					}
@@ -2437,10 +2483,12 @@ impl<S: crate::transport::poll::Session> GroupServe<S> {
 					let stream = match ready!(self.ctx.session.poll_open_uni(&mut cx)) {
 						Ok(stream) => stream,
 						Err(err) => {
+							self.settle_open(false);
 							self.state = GroupState::Done;
 							return Poll::Ready(Err(Error::from_transport(err)));
 						}
 					};
+					self.settle_open(true);
 					let mut writer = Writer::new(stream, self.ctx.version);
 					writer.set_priority(self.priority.send_order());
 
@@ -2768,6 +2816,7 @@ mod serve_group_test {
 			track_priority_seen: 0,
 			version: Version::Lite06,
 			timescale: Some(crate::Timescale::default()),
+			opens: Default::default(),
 		};
 
 		let track = track::Producer::new(Arc::new(broadcast::Info::default()), "test", None);
@@ -2808,6 +2857,7 @@ mod serve_group_test {
 			track_priority_seen: 0,
 			version: Version::Lite06,
 			timescale: Some(crate::Timescale::default()),
+			opens: Default::default(),
 		};
 
 		let track = track::Producer::new(Arc::new(broadcast::Info::default()), "test", None);
@@ -2852,6 +2902,7 @@ mod serve_group_test {
 			track_priority_seen: 0,
 			version: Version::Lite06,
 			timescale: Some(crate::Timescale::default()),
+			opens: Default::default(),
 		};
 
 		let track = track::Producer::new(Arc::new(broadcast::Info::default()), "test", None);
@@ -2914,6 +2965,7 @@ mod serve_group_test {
 			track_priority_seen: 0,
 			version: Version::Lite06,
 			timescale: Some(crate::Timescale::default()),
+			opens: Default::default(),
 		};
 
 		let track = track::Producer::new(Arc::new(broadcast::Info::default()), "test", None);
@@ -2983,6 +3035,7 @@ mod serve_group_test {
 			track_priority_seen: 0,
 			version: Version::Lite06,
 			timescale: Some(crate::Timescale::default()),
+			opens: Default::default(),
 		};
 
 		let track = track::Producer::new(Arc::new(broadcast::Info::default()), "test", None);
@@ -3007,6 +3060,108 @@ mod serve_group_test {
 			"rank 0 must reach the transport as send order 255: {priorities:?}",
 		);
 	}
+
+	/// A lite-07 subscription's run loop, from group 0, and the log of its subscribe stream.
+	fn lite07_run(
+		session: SinkSession,
+		track: track::Subscriber,
+	) -> (TrackRun<SinkSession>, Stream<SinkSession, Version>, Log) {
+		let log = Log::default();
+		let stream = Stream {
+			writer: Writer::new(SinkSend::new(log.clone()), Version::Lite07),
+			reader: crate::coding::Reader::new(PendingRecv, Version::Lite07),
+		};
+		let track_priority = kio::Producer::new(0u8);
+		let ctx = Subscription {
+			session,
+			id: 0,
+			track_name: "test".into(),
+			priority: PriorityQueue::default(),
+			track_priority: track_priority.consume(),
+			track_priority_seen: 0,
+			version: Version::Lite07,
+			timescale: Some(crate::Timescale::default()),
+			opens: Default::default(),
+		};
+		let bounds = Bounds {
+			start_group: Some(0),
+			start_frame: 0,
+			end_group: None,
+			end_frame: None,
+		};
+		(TrackRun::new(ctx, track, bounds, track_priority), stream, log)
+	}
+
+	fn write_group(track: &mut track::Producer, sequence: u64, millis: u64) {
+		let mut group = track.create_group(group::Info { sequence }).unwrap();
+		group
+			.write_frame(Timestamp::from_millis(millis).unwrap(), b"x".as_slice())
+			.unwrap();
+		group.finish().unwrap();
+	}
+
+	/// SUBSCRIBE_END counts the group streams opened, not the groups below the end: a
+	/// group the track never produced has no stream and is not counted.
+	#[tokio::test]
+	async fn lite07_end_counts_the_streams_opened() {
+		let mut track = track::Producer::new(Arc::new(broadcast::Info::default()), "test", None);
+		let subscriber = track.subscribe(None);
+		let (mut run, mut stream, log) = lite07_run(SinkSession::new(Log::default()), subscriber);
+		let mut run = std::pin::pin!(kio::wait(move |waiter| run.poll(&mut stream, waiter)));
+
+		write_group(&mut track, 0, 0);
+		assert!(futures::poll!(run.as_mut()).is_pending());
+		write_group(&mut track, 2, 2);
+		track.finish().unwrap();
+		assert!(matches!(run.await.unwrap(), TrackEnd::Finished));
+
+		// SUBSCRIBE_START at 0, then SUBSCRIBE_END at 3 with 2 streams.
+		assert_eq!(*log.writes.lock().unwrap(), [0, 1, 0, 1, 2, 3, 2]);
+	}
+
+	/// A track that ends without a group still ends the subscription, with no stream owed.
+	#[tokio::test]
+	async fn lite07_end_counts_zero_streams() {
+		let track = track::Producer::new(Arc::new(broadcast::Info::default()), "test", None);
+		let subscriber = track.subscribe(None);
+		track.finish().unwrap();
+
+		let (mut run, mut stream, log) = lite07_run(SinkSession::new(Log::default()), subscriber);
+		kio::wait(|waiter| run.poll(&mut stream, waiter)).await.unwrap();
+		assert_eq!(*log.writes.lock().unwrap(), [1, 2, 0, 0]);
+	}
+
+	/// The count is sent once every served group has opened its stream or given up, so a
+	/// group still waiting for stream credit holds SUBSCRIBE_END back, and one that expires
+	/// first is never counted.
+	#[tokio::test]
+	async fn lite07_end_waits_for_every_stream_to_open() {
+		tokio::time::pause();
+
+		let gate = kio::Producer::new(false);
+		let mut track = track::Producer::new(Arc::new(broadcast::Info::default()), "test", None);
+		let subscriber = track.subscribe(None);
+		write_group(&mut track, 0, 0);
+
+		let (mut run, mut stream, log) = lite07_run(SinkSession::gated_open_uni(gate.consume()), subscriber);
+		let mut run = std::pin::pin!(kio::wait(move |waiter| run.poll(&mut stream, waiter)));
+		assert!(futures::poll!(run.as_mut()).is_pending());
+
+		// Group 1 lands a second later, expiring group 0 before it ever opened.
+		tokio::time::advance(Duration::from_secs(1)).await;
+		write_group(&mut track, 1, 1000);
+		track.finish().unwrap();
+		assert!(futures::poll!(run.as_mut()).is_pending(), "group 1 is still opening");
+		assert_eq!(*log.writes.lock().unwrap(), [0, 1, 0], "only SUBSCRIBE_START so far");
+
+		let Ok(mut open) = gate.write() else {
+			panic!("transport gate closed");
+		};
+		*open = true;
+		drop(open);
+		run.await.unwrap();
+		assert_eq!(*log.writes.lock().unwrap(), [0, 1, 0, 1, 2, 2, 1]);
+	}
 }
 
 #[cfg(test)]
diff --git a/rs/moq-net/src/lite/subscribe.rs b/rs/moq-net/src/lite/subscribe.rs
index 5de12e097f..4656e94521 100644
--- a/rs/moq-net/src/lite/subscribe.rs
+++ b/rs/moq-net/src/lite/subscribe.rs
@@ -322,6 +322,9 @@ impl Message for SubscribeStart {
 #[derive(Clone, Debug)]
 pub struct SubscribeEnd {
 	pub group: u64,
+	/// The number of group streams the publisher opened for this subscription.
+	/// Lite07+ only; not on the wire before, where it decodes as 0.
+	pub streams: u64,
 }
 
 impl Message for SubscribeEnd {
@@ -329,16 +332,23 @@ impl Message for SubscribeEnd {
 		if !version.has_track_stream() {
 			return Err(DecodeError::Version);
 		}
-		Ok(Self {
-			group: u64::decode(r, version)?,
-		})
+		let group = u64::decode(r, version)?;
+		let streams = match version.has_stream_count() {
+			true => u64::decode(r, version)?,
+			false => 0,
+		};
+		Ok(Self { group, streams })
 	}
 
 	fn encode_msg<W: bytes::BufMut>(&self, w: &mut W, version: Version) -> Result<(), EncodeError> {
 		if !version.has_track_stream() {
 			return Err(EncodeError::Version);
 		}
-		self.group.encode(w, version)
+		self.group.encode(w, version)?;
+		if version.has_stream_count() {
+			self.streams.encode(w, version)?;
+		}
+		Ok(())
 	}
 }
 
@@ -429,7 +439,7 @@ impl Message for SubscribeUpdate {
 /// The range `[start, end]` is inclusive on both ends. For example,
 /// `start = 5, end = 7` means groups 5, 6, and 7 were dropped.
 ///
-/// Lite03+ only.
+/// Lite03 to Lite06 only: Lite07 counts group streams in [`SubscribeEnd`] instead.
 #[derive(Clone, Debug)]
 pub struct SubscribeDrop {
 	/// The first absolute group sequence in the dropped range.
@@ -449,6 +459,7 @@ impl Message for SubscribeDrop {
 			Version::Lite01 | Version::Lite02 => {
 				return Err(DecodeError::Version);
 			}
+			_ if version.has_stream_count() => return Err(DecodeError::Version),
 			_ => {}
 		}
 
@@ -464,6 +475,7 @@ impl Message for SubscribeDrop {
 			Version::Lite01 | Version::Lite02 => {
 				return Err(EncodeError::Version);
 			}
+			_ if version.has_stream_count() => return Err(EncodeError::Version),
 			_ => {}
 		}
 
@@ -480,8 +492,9 @@ impl Message for SubscribeDrop {
 ///
 /// The discriminator is version-dependent:
 /// - Lite03/04: `0x0` SUBSCRIBE_OK, `0x1` SUBSCRIBE_DROP.
-/// - Lite05+: `0x0` SUBSCRIBE_START, `0x1` SUBSCRIBE_END, `0x2` SUBSCRIBE_DROP
+/// - Lite05/06: `0x0` SUBSCRIBE_START, `0x1` SUBSCRIBE_END, `0x2` SUBSCRIBE_DROP
 ///   (SUBSCRIBE_OK was removed; acceptance is implicit).
+/// - Lite07+: `0x0` SUBSCRIBE_START, `0x1` SUBSCRIBE_END (SUBSCRIBE_DROP was removed).
 #[derive(Clone, Debug)]
 pub enum SubscribeResponse {
 	Ok(SubscribeOk),
@@ -525,8 +538,8 @@ impl Encode<Version> for SubscribeResponse {
 			_ => match self {
 				Self::Start(start) => encode_typed(w, 0, start, version)?,
 				Self::End(end) => encode_typed(w, 1, end, version)?,
-				Self::Drop(drop) => encode_typed(w, 2, drop, version)?,
-				Self::Ok(_) => return Err(EncodeError::Version),
+				Self::Drop(drop) if !version.has_stream_count() => encode_typed(w, 2, drop, version)?,
+				Self::Drop(_) | Self::Ok(_) => return Err(EncodeError::Version),
 			},
 		}
 
@@ -551,7 +564,7 @@ impl Decode<Version> for SubscribeResponse {
 				match typ {
 					0 => Ok(Self::Start(SubscribeStart::decode(buf, version)?)),
 					1 => Ok(Self::End(SubscribeEnd::decode(buf, version)?)),
-					2 => Ok(Self::Drop(SubscribeDrop::decode(buf, version)?)),
+					2 if !version.has_stream_count() => Ok(Self::Drop(SubscribeDrop::decode(buf, version)?)),
 					_ => Err(DecodeError::InvalidMessage(typ)),
 				}
 			}
@@ -577,16 +590,53 @@ mod test {
 
 	#[test]
 	fn subscribe_end_roundtrips_on_lite05() {
-		let resp = SubscribeResponse::End(SubscribeEnd { group: 7 });
+		let resp = SubscribeResponse::End(SubscribeEnd { group: 7, streams: 3 });
 		let mut buf = Vec::new();
 		resp.encode(&mut buf, Version::Lite05).unwrap();
+		// Type, length, group: no stream count before lite-07.
+		assert_eq!(buf, [1, 1, 7]);
 		let mut slice = buf.as_slice();
 		match SubscribeResponse::decode(&mut slice, Version::Lite05).unwrap() {
-			SubscribeResponse::End(end) => assert_eq!(end.group, 7),
+			SubscribeResponse::End(end) => assert_eq!((end.group, end.streams), (7, 0)),
+			other => panic!("expected End, got {other:?}"),
+		}
+	}
+
+	#[test]
+	fn subscribe_end_carries_the_stream_count_on_lite07() {
+		let resp = SubscribeResponse::End(SubscribeEnd { group: 7, streams: 3 });
+		let mut buf = Vec::new();
+		resp.encode(&mut buf, Version::Lite07).unwrap();
+		assert_eq!(buf, [1, 2, 7, 3]);
+		let mut slice = buf.as_slice();
+		match SubscribeResponse::decode(&mut slice, Version::Lite07).unwrap() {
+			SubscribeResponse::End(end) => assert_eq!((end.group, end.streams), (7, 3)),
 			other => panic!("expected End, got {other:?}"),
 		}
 	}
 
+	#[test]
+	fn subscribe_drop_is_gone_on_lite07() {
+		let resp = SubscribeResponse::Drop(SubscribeDrop {
+			start: 1,
+			end: 3,
+			error: 0,
+		});
+		let mut buf = Vec::new();
+		assert!(matches!(
+			resp.encode(&mut buf, Version::Lite07),
+			Err(EncodeError::Version)
+		));
+
+		// A lite-06 DROP is an unknown response type on lite-07.
+		let mut buf = Vec::new();
+		resp.encode(&mut buf, Version::Lite06).unwrap();
+		assert!(matches!(
+			SubscribeResponse::decode(&mut buf.as_slice(), Version::Lite07),
+			Err(DecodeError::InvalidMessage(2))
+		));
+	}
+
 	#[test]
 	fn subscribe_drop_is_type_2_on_lite05() {
 		let resp = SubscribeResponse::Drop(SubscribeDrop {
diff --git a/rs/moq-net/src/lite/version.rs b/rs/moq-net/src/lite/version.rs
index dfd78d0c57..72e4ffa30b 100644
--- a/rs/moq-net/src/lite/version.rs
+++ b/rs/moq-net/src/lite/version.rs
@@ -18,13 +18,26 @@ pub enum Version {
 	/// as `moq-lite-06` and preferred by the default version sets.
 	Lite06,
 	/// Lite-07. Adds the hidden opt-in to ANNOUNCE_REQUEST: without it, a route with
-	/// a `.`-prefixed segment below the requested prefix is left out. The wire format is
-	/// still work-in-progress, so it is advertised over ALPN as `moq-lite-07-wip` and
-	/// only when explicitly requested; the default version sets leave it out.
+	/// a `.`-prefixed segment below the requested prefix is left out. SUBSCRIBE_END
+	/// carries the number of group streams opened, replacing SUBSCRIBE_DROP.
+	/// The wire format is still work-in-progress, so it is advertised over ALPN as
+	/// `moq-lite-07-wip` and only when explicitly requested; the default version sets
+	/// leave it out.
 	Lite07,
 }
 
 impl Version {
+	/// Whether SUBSCRIBE_END carries the subscription's group stream count, sent once
+	/// every counted stream is open, in place of SUBSCRIBE_DROP. Added in lite-07.
+	#[allow(clippy::match_like_matches_macro)]
+	pub fn has_stream_count(self) -> bool {
+		// Match form so future versions default forward (CLAUDE.md convention).
+		match self {
+			Self::Lite01 | Self::Lite02 | Self::Lite03 | Self::Lite04 | Self::Lite05 | Self::Lite06 => false,
+			_ => true,
+		}
+	}
+
 	/// Whether the version has lite-05's dedicated TRACK stream and related stream
 	/// layout changes.
 	///