test(interop): released wire compat runs its own observer again

quest/m0/README.md

@@ -57,10 +57,8 @@ that depend on them.
 
 Interop on main (2026-10-13): the cross-client matrix fails with `unroutable`
 whenever WebSocket wins the connect race: since #5294, #4180's upgrade
-retires epochless handles before its held `Restart` reaches anyone. The
-released wire-compat step aborts on planned breaks the last release made
-stale. Both stay in m0 (maintainer, 2026-10-13): a red Interop hides real
-regressions.
+retires epochless handles before its held `Restart` reaches anyone. It
+stays in m0 (maintainer, 2026-10-13): a red Interop hides real regressions.
 
 Crashes (2026-10-13, #5407): a publisher on a host whose wall clock reads
 before 2020 panics, and `panic = "abort"` turns any library panic into a
@@ -82,7 +80,6 @@ have fixes on `main`: macOS tests (#5335), drill-sensitivity (#5338), loom
 - [Draft-22 media on 0.17](/quest/m0/release-22/README.md) - a 0.17.x with the LOCATION_FILTER and FIRST_OBJECT fixes and moq-noq 1.3.5, then draft-20+ FETCH
 - [Upgrade restarts epochless broadcasts](/quest/m0/upgrade-unroutable.md) - a WebSocket-to-QUIC upgrade retires epochless handles only once their `Restart` is delivered, and first-party consumers follow it, fixing Interop's `unroutable` rounds
 - [Panic isolation](/quest/m0/panic-isolation.md) - unwind, ignore lock poison, and supervise every task, so a panic closes only its connection or exits on a long-lived task
-- [Wire compat on main](/quest/m0/wire-compat-green.md) - the nightly released-wire step runs again: planned breaks re-measured against the new releases, and a released-API client for the released JS lanes
 - [nixpkgs ships the lld fix](/quest/m0/nixpkgs-lld-fix.md) - waiting on NixOS/nixpkgs#568680 to reach `nixpkgs-unstable`, which the release build fix bumps to
 - [Release builds link](/quest/m0/release-link.md) - the Linux zigbuild release binaries and moq-gst packages link again after the 2026-10-11 nixpkgs bump, so the next release cut can publish
 - [Fuzz builds under LTO](/quest/m0/fuzz-lto.md) - the nightly fuzz targets build again despite the release profile's fat LTO

test/interop/README.md

@@ -187,13 +187,15 @@ The relay is anonymous here, since released binaries may predate AUTH; the
 default matrix covers tokens and grants.
 Current Rust media publishers feed released Rust and JS readers, then released
 publishers feed current readers, through both relay sources. Rust exports must
-decode to a video frame through ffmpeg. The existing JS subscriber
-reconstructs the catalog and decodes the container.
+decode to a video frame through ffmpeg. The JS reader is
+`compat/client.ts`, not the interop subscriber, whose API follows the
+checkout. One copy runs against each package set, using only API both have,
+and it reconstructs the catalog and decodes the container.
 
 Each `moq-net` library states whether a draft has FETCH (lite-05 onward
 today). A draft without it in both is logged and its FETCH lanes skipped; one
 the release supports and the checkout dropped fails. Both Rust versions FETCH the group the
-publisher's own JS subscriber observed and compare every frame's exact
+publisher's own compat client observed and compare every frame's exact
 payload. JS publishers write one group after that subscriber's demand, and the
 opposite-source Rust CLI FETCHes it by the same observed ID.
 

test/interop/clients/js-native/subscribe.ts

@@ -12,7 +12,6 @@
  *
  * @module
  */
-import { writeFileSync } from "node:fs";
 import { parseArgs } from "node:util";
 import * as Catalog from "@moq/hang/catalog";
 import * as Container from "@moq/hang/container";
@@ -37,8 +36,6 @@ const { positionals, values } = parseArgs({
 		url: { type: "string" },
 		broadcast: { type: "string" },
 		timeout: { type: "string", default: "20" },
-		"track-file": { type: "string" },
-		track: { type: "string" },
 	},
 });
 
@@ -67,19 +64,14 @@ async function video(bc: Moq.Broadcast.Consumer): Promise<[string, Catalog.Video
 
 async function run(): Promise<void> {
 	const origin = new Moq.Origin.Producer();
-	// A released-compat run pins its version through its own transport and runs anonymous.
-	// Otherwise the grant arrives over AUTH, which only the work-in-progress moq-lite-07
-	// carries and no client offers by default. The WebSocket fallback cannot offer it, so it
-	// stays off.
-	const compat = process.env.INTEROP_COMPAT_TRANSPORT;
-	const connection: Moq.Connection.Established = compat
-		? await (await import(compat)).connect({ url: new URL(url as string), consume: origin })
-		: await Moq.Connection.connect({
-				url: new URL(url as string),
-				consume: origin,
-				webtransport: { protocols: ["moq-lite-07-wip"] },
-				websocket: { enabled: false },
-			});
+	// The grant arrives over AUTH, which only the work-in-progress moq-lite-07 carries and no
+	// client offers by default. The WebSocket fallback cannot offer it, so it stays off.
+	const connection = await Moq.Connection.connect({
+		url: new URL(url as string),
+		consume: origin,
+		webtransport: { protocols: ["moq-lite-07-wip"] },
+		websocket: { enabled: false },
+	});
 	// The grant the relay sent, in the Rust client's `auth granted` shape, so the harness can
 	// check it against the token this cell minted.
 	const printGrant = (grant: Moq.Auth.Grant | undefined) => {
@@ -88,11 +80,8 @@ async function run(): Promise<void> {
 		const subscribe = JSON.stringify(grant.subscribe);
 		console.error(`auth granted publish=${publish} subscribe=${subscribe}`);
 	};
-	let unwatch = () => {};
-	if (!compat) {
-		printGrant(connection.auth.grant.peek());
-		unwatch = connection.auth.grant.subscribe(printGrant);
-	}
+	printGrant(connection.auth.grant.peek());
+	const unwatch = connection.auth.grant.subscribe(printGrant);
 	let requested: Moq.Origin.Requesting | undefined;
 	try {
 		const path = Moq.Path.from(broadcast as string);
@@ -103,16 +92,6 @@ async function run(): Promise<void> {
 			bc = requested.active.peek();
 		}
 
-		if (values["track-file"]) {
-			// Name a group that live demand is filling, so a FETCH of it never races eviction.
-			// The track is `--track`, or the catalog's first video rendition.
-			const name = values.track ?? (await video(bc))[0];
-			const group = await bc.track(name).subscribe({ priority: 0 }).recvGroup();
-			if (!group) throw new Error(`${name} ended before its first group`);
-			writeFileSync(values["track-file"], `${name}\n${group.sequence}\n`);
-			return;
-		}
-
 		const [name, config] = await video(bc);
 		let format: Container.Format;
 		if (config.container.kind === "legacy") {

test/interop/compat.sh

@@ -57,9 +57,10 @@ RELEASED="$HARNESS_RUN/released/bin/moq"
 bun "$DIR/compat/resolve.ts" matrix "$HARNESS_RUN/current-help" "$HARNESS_RUN/released-help" "$DIR/compat/planned-breaks.json" "$HARNESS_RUN/shared" "$HARNESS_RUN/versions.json"
 
 # A separate package root prevents bun workspaces from replacing released deps
-# with the checkout. Copy exactly the same clients into both environments.
+# with the checkout. Copy exactly the same client into both environments; it uses
+# only API both package sets have, so it never follows the checkout's interop client.
 mkdir -p "$HARNESS_RUN/js"
-cp "$DIR/compat/client.ts" "$DIR/compat/transport.ts" "$DIR/clients/js-native/subscribe.ts" "$HARNESS_RUN/js/"
+cp "$DIR/compat/client.ts" "$DIR/compat/transport.ts" "$HARNESS_RUN/js/"
 python3 - "$HARNESS_RUN/versions.json" "$HARNESS_RUN/js/package.json" <<'PY'
 import json, sys
 versions = json.load(open(sys.argv[1]))
@@ -157,9 +158,9 @@ cell() {
 }
 
 fetch() {
-    local relay="$1" pub="$2" sub="$3" js="$4" transport="$5" version="$6"
+    local relay="$1" pub="$2" sub="$3" js="$4" version="$5"
     RELAY_BIN="$relay" MOQ_BIN="$pub" INTEROP_SUB_MOQ="$sub" INTEROP_VERSION="$version" \
-        INTEROP_NATIVE_CLIENT="$js" INTEROP_COMPAT_TRANSPORT="$transport" INTEROP_FETCH_TRACK=1 \
+        INTEROP_COMPAT_CLIENT="$js" INTEROP_FETCH_TRACK=1 \
         bash "$DIR/interop.sh" --publishers rust --subscribers rust
 }
 
@@ -186,29 +187,24 @@ while read -r version <&3; do
     for relay_source in current released; do
         if [[ "$relay_source" == current ]]; then relay="$TARGET/debug/moq-relay"; else relay="$released_relay"; fi
         for source in current released; do
-            # Each side's JS environment and the version-pinning transport beside it.
+            # Each side's compat client, with its version-pinning transport beside it.
             if [[ "$source" == current ]]; then
-                pub="$CURRENT" sub="$RELEASED" publisher_js="$DIR/compat"
-                own_js="$DIR/clients/js-native" own_transport="$DIR/compat/transport.ts"
-                other_js="$HARNESS_RUN/js" other_transport="$HARNESS_RUN/js/transport.ts"
+                pub="$CURRENT" sub="$RELEASED" own_js="$DIR/compat" other_js="$HARNESS_RUN/js"
             else
-                pub="$RELEASED" sub="$CURRENT" publisher_js="$HARNESS_RUN/js"
-                own_js="$HARNESS_RUN/js" own_transport="$HARNESS_RUN/js/transport.ts"
-                other_js="$DIR/clients/js-native" other_transport="$DIR/compat/transport.ts"
+                pub="$RELEASED" sub="$CURRENT" own_js="$HARNESS_RUN/js" other_js="$DIR/compat"
             fi
             cell media "$version" "$relay_source" "$source" env RELAY_BIN="$relay" MOQ_BIN="$pub" INTEROP_SUB_MOQ="$sub" INTEROP_VERSION="$version" \
-                INTEROP_NATIVE_CLIENT="$other_js" INTEROP_COMPAT_TRANSPORT="$other_transport" INTEROP_DECODE=1 bash "$DIR/interop.sh" --subscribers rust,js-native-node
+                INTEROP_COMPAT_CLIENT="$other_js" INTEROP_DECODE=1 bash "$DIR/interop.sh" --subscribers rust,js-native-node
             [[ "$released_fetch" == true && "$current_fetch" == true ]] || continue
             # Drive JS publication from actual subscriber demand: the publisher's own JS
             # subscribes and observes the group, which the opposite Rust CLI then
             # FETCHes by ID. Reading the newest group races its live delivery.
             cell js-publish "$version" "$relay_source" "$source" env RELAY_BIN="$relay" MOQ_BIN="$pub" INTEROP_SUB_MOQ="$sub" INTEROP_VERSION="$version" \
-                INTEROP_JS_PUBLISH_CLIENT="$publisher_js" INTEROP_NATIVE_CLIENT="$own_js" INTEROP_COMPAT_TRANSPORT="$own_transport" \
-                INTEROP_FETCH_DATA=1 bash "$DIR/interop.sh" --publishers js-native --subscribers rust
+                INTEROP_COMPAT_CLIENT="$own_js" INTEROP_FETCH_DATA=1 bash "$DIR/interop.sh" --publishers js-native --subscribers rust
             # Both Rust sources fetch the same group and compare exact payloads. The
             # publisher's own JS finds the group, so a catalog break fails only the
             # media cell.
-            cell fetch "$version" "$relay_source" "$source" fetch "$relay" "$pub" "$sub" "$own_js" "$own_transport" "$version"
+            cell fetch "$version" "$relay_source" "$source" fetch "$relay" "$pub" "$sub" "$own_js" "$version"
         done
     done
 done 3<"$HARNESS_RUN/shared"

test/interop/compat/client.ts

@@ -1,11 +1,95 @@
 import { readFileSync, writeFileSync } from "node:fs";
+import { parseArgs } from "node:util";
 import * as Auth from "@moq/auth";
 import * as Catalog from "@moq/hang/catalog";
 import * as Container from "@moq/hang/container";
+import * as Json from "@moq/json";
 import * as Moq from "@moq/net";
 import { install } from "@moq/web-transport";
 import { connect } from "./transport";
 
+// This one client runs against both the checkout's and the released packages, so it sticks to
+// API both have: the group and frame primitives and the snapshot decoder, not the consumers above
+// them, whose shapes follow the checkout.
+
+/** Find the catalog's first video rendition, reading updates until one has it. */
+async function video(bc: Moq.Broadcast.Consumer): Promise<[string, Catalog.VideoConfig]> {
+	const track = bc.track("catalog.json").subscribe({ priority: Catalog.PRIORITY.catalog });
+	const decoder = new Json.Snapshot.Decoder<Catalog.Root>({ schema: Catalog.RootSchema });
+	for (;;) {
+		const group = await track.recvGroup();
+		if (!group) throw new Error("catalog ended without a video track");
+		// Each group opens with a snapshot; the frames after it are deltas.
+		let first = true;
+		try {
+			for (let frame = await group.readFrame(); frame; frame = await group.readFrame()) {
+				if (first) decoder.snapshot(frame.payload);
+				else decoder.delta(frame.payload);
+				first = false;
+				const rendition = Object.entries(decoder.decode()?.video?.renditions ?? {})[0];
+				if (rendition) return rendition;
+			}
+		} catch (err) {
+			// The relay resets a group a newer snapshot supersedes; resync from the next.
+			if (!(err instanceof Moq.Error.Stream)) throw err;
+		}
+	}
+}
+
+/** Print `received N bytes` once a video frame decodes, or record the group live demand fills. */
+async function subscribe(url: URL, broadcast: string, trackFile?: string, trackName?: string) {
+	install();
+	const origin = new Moq.Origin.Producer();
+	const connection = await connect({ url, consume: origin, websocket: { enabled: false } });
+	const requested = origin.request(Moq.Path.from(broadcast), { announced: true });
+	try {
+		let bc = requested.active.peek();
+		while (!bc) {
+			await requested.active.changed();
+			bc = requested.active.peek();
+		}
+
+		if (trackFile) {
+			// Name a group that live demand is filling, so a FETCH of it never races eviction.
+			const name = trackName ?? (await video(bc))[0];
+			const group = await bc.track(name).subscribe({ priority: 0 }).recvGroup();
+			if (!group) throw new Error(`${name} ended before its first group`);
+			writeFileSync(trackFile, `${name}\n${group.sequence}\n`);
+			return;
+		}
+
+		const [name, config] = await video(bc);
+		if (config.container.kind !== "legacy") {
+			throw new Error(`unsupported video container: ${JSON.stringify(config.container)}`);
+		}
+		const format = new Container.Legacy.Format("video");
+		const track = bc.track(name).subscribe({ priority: 0 });
+		for (let group = await track.recvGroup(); group; group = await track.recvGroup()) {
+			try {
+				for (let frame = await group.readFrame(); frame; frame = await group.readFrame()) {
+					const bytes = format.decode(frame.payload).reduce((n, f) => n + f.payload.byteLength, 0);
+					if (bytes > 0) {
+						// The harness judges success by this marker: the @moq/web-transport addon
+						// can crash during exit teardown after a frame has arrived.
+						console.error(`received ${bytes} bytes from ${broadcast}`);
+						return;
+					}
+				}
+			} catch (err) {
+				// The relay resets a group a newer one supersedes; move on to the next.
+				if (!(err instanceof Moq.Error.Stream)) throw err;
+			}
+		}
+		throw new Error("no frame data received");
+	} finally {
+		requested.close();
+		// Wait for the close to reach the relay before exit, so it is not timed out.
+		await connection.close();
+		await connection.closed;
+		origin.close();
+	}
+}
+
 const [command, ...args] = process.argv.slice(2);
 if (command === "sign") {
 	const key = Auth.Key.parse(readFileSync(args[0], "utf8"));
@@ -67,4 +151,33 @@ if (command === "sign") {
 	});
 	group.close();
 	await connection.closed;
+} else if (command === "subscribe") {
+	// The flags of the interop subscriber this stands in for.
+	const { values } = parseArgs({
+		args,
+		options: {
+			url: { type: "string" },
+			broadcast: { type: "string" },
+			timeout: { type: "string", default: "20" },
+			"track-file": { type: "string" },
+			track: { type: "string" },
+		},
+	});
+	const timeout = Number.parseFloat(values.timeout ?? "") * 1000;
+	if (!values.url || !values.broadcast || !(timeout > 0)) {
+		throw new Error("usage: subscribe --url U --broadcast B [--timeout S] [--track-file F [--track T]]");
+	}
+	const timer = new Promise<never>((_, reject) =>
+		setTimeout(() => reject(new Error("timed out waiting for data")), timeout),
+	);
+	try {
+		await Promise.race([
+			subscribe(new URL(values.url), values.broadcast, values["track-file"], values.track),
+			timer,
+		]);
+		process.exit(0);
+	} catch (err) {
+		console.error(`error: ${err instanceof Error ? err.message : String(err)}`);
+		process.exit(1);
+	}
 } else throw new Error(`unknown compatibility client command: ${command}`);