From 9ff8594ae6596f591d1f2507634b85acc108b84b Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 2 Aug 2026 19:41:09 -0700 Subject: [PATCH 1/2] fix: wait for the bot's broadcast before subscribing to the transcript MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The transcript track subscribed as soon as the connection was established, which assumes the bot is already on the relay. It isn't always: when the bot is started in response to our own announcement, its broadcast appears strictly after ours, so the stream was reset and never retried — leaving the client connected but silent. Gate it on the announcement instead, read off `Reload` rather than the established session so the gate spans reconnects. Only this track needs it. `Watch.Broadcast` has gated its own subscribe on `conn.announced(name)` since moq-dev/moq#2199, so the catalog and audio are already covered; the transcript goes straight through `@moq/net` with no wrapper. Co-Authored-By: Claude --- transports/moq-transport/src/moqTransport.ts | 16 +++++++++++++--- 1 file changed, 13 insertions(+), 3 deletions(-) diff --git a/transports/moq-transport/src/moqTransport.ts b/transports/moq-transport/src/moqTransport.ts index f8c557e..6178d2c 100644 --- a/transports/moq-transport/src/moqTransport.ts +++ b/transports/moq-transport/src/moqTransport.ts @@ -578,12 +578,22 @@ export class MoqTransport extends Transport { ); }); - // Transcript — a lossless JSON append-stream over a single track. We - // re-subscribe on each (re)connect; @moq/json's stream Consumer yields - // every appended record in order, losslessly. + // Transcript — a lossless JSON append-stream over a single track, + // re-subscribed on every (re-)announce. @moq/json's stream Consumer + // yields every appended record in order, losslessly. + // + // Gated on the announcement rather than the connection, because + // subscribing to a path nobody publishes yet gets the stream reset: + // a bot started in response to our own announcement necessarily + // appears after us. `Watch.Broadcast` gates itself this way for the + // catalog and audio; this track goes straight through `@moq/net`, so + // it needs the gate spelled out. Reading it off `Reload` rather than + // the established session means the gate spans reconnects. this._signals.run((eff) => { const conn = eff.get(this._reload!.established); if (!conn) return; + if (!eff.get(this._reload!.announced).has(botPath)) return; + const botBroadcast = conn.consume(botPath); const track = botBroadcast.subscribe(merged.transcriptTrack, 0); const consumer = new Json.Stream.Consumer(track, { From c39688468fef60503c39e969116e6bc5e877014e Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 3 Aug 2026 19:21:39 -0700 Subject: [PATCH 2/2] feat: upgrade to @moq 0.4 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Picks up the 0.4-era libraries, which reshaped inputs/outputs onto `in` and `out` and replaced the broadcast's inline `audio` prop with a registered rendition. - Publishing: `Publish.Audio.Encoder` registers the `audio` rendition on the broadcast and fills its catalog entry, fed by the microphone's `out.source`. - The client transcript track is served off `broadcast.net` rather than the removed `publishTrack`. `net` swaps on reconnect, so the producer is rebuilt and the message log replayed into it, keeping the guarantee that a bot subscribing late still gets every message in order. - Playback: the source is built before `Sync` (it produces the jitter Sync reads), the decoder takes the sync positionally, and the graph handles move under `out`. - `Reload.announced` is a prefix stream here rather than a membership set, so the transcript gate drains it — which also picks up a same-name republish, where the membership set never changed. The characterization mocks move with the API: outputs under `out`, an `Audio.Encoder`, and a `net` on the publish broadcast. Co-Authored-By: Claude --- package-lock.json | 140 ++++++++-------- tests/src/transports/moq.spec.ts | 38 ++++- transports/moq-transport/package.json | 12 +- transports/moq-transport/src/moqTransport.ts | 160 ++++++++++++------- 4 files changed, 208 insertions(+), 142 deletions(-) diff --git a/package-lock.json b/package-lock.json index f0dee7c..8145710 100644 --- a/package-lock.json +++ b/package-lock.json @@ -625,9 +625,9 @@ "license": "LGPL-2.1" }, "node_modules/@libav.js/variant-opus-af": { - "version": "6.8.8", - "resolved": "https://registry.npmjs.org/@libav.js/variant-opus-af/-/variant-opus-af-6.8.8.tgz", - "integrity": "sha512-8KBQyA8n5goN7lyctOaPxpcx7dapOgqKh8dWW/NAcl87AgM/WoUGSex3fFc46oCtTHYrUKEm1OmZUrtkt3Q56A==", + "version": "6.9.8", + "resolved": "https://registry.npmjs.org/@libav.js/variant-opus-af/-/variant-opus-af-6.9.8.tgz", + "integrity": "sha512-4v4kBOoNnmafubEiBof+ATtr7KcU2/kL+G2hXU61orvC4Lt5OViBXnHdUHCh5ZSQQSmx9nOFEgbYGGYTrBufXQ==", "license": "LGPL-2.1" }, "node_modules/@lmdb/lmdb-darwin-arm64": { @@ -739,62 +739,62 @@ } }, "node_modules/@moq/hang": { - "version": "0.2.12", - "resolved": "https://registry.npmjs.org/@moq/hang/-/hang-0.2.12.tgz", - "integrity": "sha512-rVz98b0l9adlwmSQ4O0kUW610N7M6zljR6wL5rqJP5gbTetbjI3NgA/xakFldRJN1NQnnVGtltYqbt2OuDfpVg==", + "version": "0.3.4", + "resolved": "https://registry.npmjs.org/@moq/hang/-/hang-0.3.4.tgz", + "integrity": "sha512-1o5EgGAlk2gzJSDifC0IEY/184fTamuZgPwfauXhpuek6EyCwUTZ+NwoYvcoqjFZwV01kor+DOiIFgX0f7EvPg==", "license": "(MIT OR Apache-2.0)", "dependencies": { "@kixelated/libavjs-webcodecs-polyfill": "^0.5.5", - "@libav.js/variant-opus-af": "^6.8.8", - "@moq/json": "^0.1.1", - "@moq/loc": "^0.1.1", - "@moq/net": "^0.1.6", - "@moq/signals": "^0.1.10", - "@svta/cml-iso-bmff": "^1.0.2", + "@libav.js/variant-opus-af": "^6.9.8", + "@moq/json": "^0.3.0", + "@moq/loc": "^0.2.1", + "@moq/net": "^0.2.3", + "@moq/signals": "^0.2.1", + "@svta/cml-iso-bmff": "^1.0.3", "zod": "^4.4.3" } }, "node_modules/@moq/json": { - "version": "0.1.2", - "resolved": "https://registry.npmjs.org/@moq/json/-/json-0.1.2.tgz", - "integrity": "sha512-sQz86Iz29lzJIyzY6OtBqJHXxqtcDj2DypPNSEbHSwUQCcOxEzkMS2vhNZEK+oTRN6DpblTmelBeZJPg4RjkjA==", + "version": "0.3.1", + "resolved": "https://registry.npmjs.org/@moq/json/-/json-0.3.1.tgz", + "integrity": "sha512-KZXaXs9yao+t9uUlFd6zRowhoQgyDqrEFfKTtJM0S9IIJW7UFUf47lWBMGPjgqg24XGW5Ss8rbi6dtT5do12jA==", "license": "(MIT OR Apache-2.0)", "dependencies": { "@moq/flate": "^0.1.1", - "@moq/net": "^0.1.7", - "@moq/signals": "^0.1.10" + "@moq/net": "^0.2.3", + "@moq/signals": "^0.2.1" }, "peerDependencies": { "zod": "^4.0.0" } }, "node_modules/@moq/loc": { - "version": "0.1.1", - "resolved": "https://registry.npmjs.org/@moq/loc/-/loc-0.1.1.tgz", - "integrity": "sha512-/DUgFYRmgHauxTaskPNn6tmv1UEYhuUk3bK4y67jZzAMy2N8sMZL6dyXuZvA2+n8jd/yZqy/pB+/+Tfj35JixQ==", + "version": "0.2.1", + "resolved": "https://registry.npmjs.org/@moq/loc/-/loc-0.2.1.tgz", + "integrity": "sha512-+Fy3Nqp40asKEV77oPy0m5ZRryT/9kEYTLQZhJwnds/yDmj0yrcunr/fStjCY4Trg8v/PmeLyvYr+FRAIyx77A==", "license": "(MIT OR Apache-2.0)", "dependencies": { - "@moq/net": "^0.1.5" + "@moq/net": "^0.2.3" } }, "node_modules/@moq/msf": { - "version": "0.1.3", - "resolved": "https://registry.npmjs.org/@moq/msf/-/msf-0.1.3.tgz", - "integrity": "sha512-2Kxoniv7AgKdZfRMciUsEBELWAygzl8R89Dhn2t99qPIPwXn2yOlXEaOSoYVNvHIivbrSasIk09zRWUjSn2B8Q==", + "version": "0.2.0", + "resolved": "https://registry.npmjs.org/@moq/msf/-/msf-0.2.0.tgz", + "integrity": "sha512-gTeWd5/NpzxbYw9Cf8OfE+ltb6e4tNUXzUc+CkXgPRpT9BvRoE+4vYivhpFNd/M9+sTgqQtU2pK91VPbzbbrIg==", "license": "(MIT OR Apache-2.0)", "dependencies": { - "@moq/net": "^0.1.6", + "@moq/net": "^0.2.0", "zod": "^4.4.3" } }, "node_modules/@moq/net": { - "version": "0.1.8", - "resolved": "https://registry.npmjs.org/@moq/net/-/net-0.1.8.tgz", - "integrity": "sha512-pwx+GcqkH5M/kMk+/0SrXrwNOO8nV0B7bjkHcLD5q35c9FoQfrCwotsjodEEbND6LMDLGJiRUbWNR4gjIZjgyg==", + "version": "0.2.3", + "resolved": "https://registry.npmjs.org/@moq/net/-/net-0.2.3.tgz", + "integrity": "sha512-OQqoQRf5bJbruUlouwkgYJHg0C5SXa9Ub/GPWzn26K+AM3keh7WEr3tVRLwU3Ya3cA2cxLtNAcLPN7pOm5834g==", "license": "(MIT OR Apache-2.0)", "dependencies": { - "@moq/qmux": "^0.1.3", - "@moq/signals": "^0.1.10", + "@moq/qmux": "^0.3.2", + "@moq/signals": "^0.2.1", "async-mutex": "^0.5.0" }, "peerDependencies": { @@ -802,35 +802,34 @@ } }, "node_modules/@moq/publish": { - "version": "0.2.16", - "resolved": "https://registry.npmjs.org/@moq/publish/-/publish-0.2.16.tgz", - "integrity": "sha512-vgOCX4/VZYGzeStnwlPd6MY4hcXWAzDXSrNcjKHZdezu7hycfP++YwheVvP9c4ukfTWrqGfvdO+k8VLUTuuxSw==", + "version": "0.4.3", + "resolved": "https://registry.npmjs.org/@moq/publish/-/publish-0.4.3.tgz", + "integrity": "sha512-o9H3qTDeNc42+PET5L+7RgM+TnP/dpPmG+dnNFpnv9KoF4de77m7IY94S2M8rbmpND+bwVjpZEckoR2ixrlVFQ==", "license": "(MIT OR Apache-2.0)", "dependencies": { - "@moq/hang": "^0.2.12", - "@moq/json": "^0.1.1", - "@moq/net": "^0.1.6", - "@moq/signals": "^0.1.10" + "@moq/hang": "^0.3.4", + "@moq/net": "^0.2.3", + "@moq/signals": "^0.2.1" } }, "node_modules/@moq/qmux": { - "version": "0.1.3", - "resolved": "https://registry.npmjs.org/@moq/qmux/-/qmux-0.1.3.tgz", - "integrity": "sha512-naGmMMOxCSMeLyEATrLiY9RGyeZENC4dHFVVNWNcA17LXE9NwM3aQrQY218xXypWahYIgIkl/ZCutIP2xHibgA==", + "version": "0.3.2", + "resolved": "https://registry.npmjs.org/@moq/qmux/-/qmux-0.3.2.tgz", + "integrity": "sha512-8ymBriVxvD5ISfnTJ3k07Zzt4pBOoh/JnulGCjenY2NrUdCr5PUOy42y/9drSXA3SfCfOdaw4A+BSrWN92qVvg==", "license": "(MIT OR Apache-2.0)", "dependencies": { - "@moq/web-socket-stream": "^0.1.0" + "@moq/web-socket-stream": "^0.1.1" } }, "node_modules/@moq/signals": { - "version": "0.1.10", - "resolved": "https://registry.npmjs.org/@moq/signals/-/signals-0.1.10.tgz", - "integrity": "sha512-TSnQwcaywn/gIwC3UpwZbqYnXm2Zo8KZO6Y043jPiPnbnSlFWJVZ9TVdGAvQqrXGiaNihdcioF/0IEDABsfovw==", + "version": "0.2.1", + "resolved": "https://registry.npmjs.org/@moq/signals/-/signals-0.2.1.tgz", + "integrity": "sha512-4E7cEDw6CGdEc54RCwDUwbZf/eV4mab49hecGwoTKsvF6ZUQpIc1UtysSJT8ZZwix9ipzE0A9ap6vWqnpG7L6w==", "license": "(MIT OR Apache-2.0)", "peerDependencies": { "@types/react": "^19.2.17", - "react": "^19.0.0", - "solid-js": "^1.9.13" + "react": "^19.2.8", + "solid-js": "^1.9.14" }, "peerDependenciesMeta": { "@types/react": { @@ -845,22 +844,21 @@ } }, "node_modules/@moq/watch": { - "version": "0.2.18", - "resolved": "https://registry.npmjs.org/@moq/watch/-/watch-0.2.18.tgz", - "integrity": "sha512-6NfPr3qEfLG5Y1g60elEPpozzrda0kCHx+adoTUL7yAxYWs+MTARoSTWzdFLQCvF6nd1eVFkC0LaMKvxtyssvw==", + "version": "0.4.4", + "resolved": "https://registry.npmjs.org/@moq/watch/-/watch-0.4.4.tgz", + "integrity": "sha512-HcOB/FHEjopwM+ZlTpe01IS0giU2+PT/ipMufAv+JCVp65WwkNiHgTefzxklZj7wRIQmpds1U1izCm2l3gGxMg==", "license": "(MIT OR Apache-2.0)", "dependencies": { - "@moq/hang": "^0.2.12", - "@moq/json": "^0.1.1", - "@moq/msf": "^0.1.3", - "@moq/net": "^0.1.6", - "@moq/signals": "^0.1.10" + "@moq/hang": "^0.3.4", + "@moq/msf": "^0.2.0", + "@moq/net": "^0.2.3", + "@moq/signals": "^0.2.1" } }, "node_modules/@moq/web-socket-stream": { - "version": "0.1.0", - "resolved": "https://registry.npmjs.org/@moq/web-socket-stream/-/web-socket-stream-0.1.0.tgz", - "integrity": "sha512-VxPaJZvZ8qgFmTjG9XhTH0bQoQE9pExvlS/R0fH1aNcpqCnZdbYS2kvB2ypO5SyFTf6oH+4eWvgElDzW6yiqFQ==", + "version": "0.1.1", + "resolved": "https://registry.npmjs.org/@moq/web-socket-stream/-/web-socket-stream-0.1.1.tgz", + "integrity": "sha512-VKi/GE/CLqmbEZlgNbBLbAFk2pQTxT+KUD2DjCSoxqZxbS3KBe5BBZ0m3fila4fEGyll2JT5MmgEyVr4zkZEVA==", "license": "(MIT OR Apache-2.0)" }, "node_modules/@msgpackr-extract/msgpackr-extract-darwin-arm64": { @@ -3220,21 +3218,21 @@ "license": "MIT" }, "node_modules/@svta/cml-iso-bmff": { - "version": "1.0.2", - "resolved": "https://registry.npmjs.org/@svta/cml-iso-bmff/-/cml-iso-bmff-1.0.2.tgz", - "integrity": "sha512-c9UgY1z16zgvTEa6fS++9E9zxLuPtLJ4Dw5ebOgEDtwIw+PIhbh3URGXjv30tyDU2QQTXstI6U1q6h/Q4SkxGQ==", + "version": "1.0.4", + "resolved": "https://registry.npmjs.org/@svta/cml-iso-bmff/-/cml-iso-bmff-1.0.4.tgz", + "integrity": "sha512-ZrDOQ5Wlh6vy9RF7dzlJVP3U43DDKCitovb8Ytf6KM40UcHW2bRMgcmx1TUT4nJ7OcigEsWqrgI78hOSuzI7Hw==", "license": "Apache-2.0", "engines": { "node": ">=20" }, "peerDependencies": { - "@svta/cml-utils": "1.5.0" + "@svta/cml-utils": "1.6.0" } }, "node_modules/@svta/cml-utils": { - "version": "1.5.0", - "resolved": "https://registry.npmjs.org/@svta/cml-utils/-/cml-utils-1.5.0.tgz", - "integrity": "sha512-JMqclD7Akd+GSJiuaYNUHOP2wNtf/nauKeszlYeivSHfi0Lp3pmSW5PXDvJ2dO4aPmmSOQbF0ztTsW9Vcs2Whw==", + "version": "1.6.0", + "resolved": "https://registry.npmjs.org/@svta/cml-utils/-/cml-utils-1.6.0.tgz", + "integrity": "sha512-h8jd5Y0nrr8wHWkI7XB03Denxy6QR7dgROxchZ2S0VQ4PFINrIu6J0k9N1IGHmtJnJjG3ttszsJd4LS5V/DqcA==", "license": "Apache-2.0", "peer": true, "engines": { @@ -6569,15 +6567,15 @@ }, "transports/moq-transport": { "name": "@pipecat-ai/moq-transport", - "version": "0.0.1", + "version": "0.1.0", "license": "BSD-2-Clause", "dependencies": { - "@moq/hang": "^0.2.12", - "@moq/json": "^0.1.2", - "@moq/net": "^0.1.6", - "@moq/publish": "^0.2.16", - "@moq/signals": "^0.1.10", - "@moq/watch": "^0.2.18" + "@moq/hang": "^0.3.4", + "@moq/json": "^0.3.1", + "@moq/net": "^0.2.3", + "@moq/publish": "^0.4.3", + "@moq/signals": "^0.2.1", + "@moq/watch": "^0.4.4" }, "devDependencies": { "@pipecat-ai/client-js": "^1.13.0", diff --git a/tests/src/transports/moq.spec.ts b/tests/src/transports/moq.spec.ts index c12eecc..907b072 100644 --- a/tests/src/transports/moq.spec.ts +++ b/tests/src/transports/moq.spec.ts @@ -45,17 +45,31 @@ vi.mock("@moq/publish", () => { .catch(() => {}); }, }; - source = { - peek: () => - this._track ? { track: this._track } : undefined, + // 0.4 moved component outputs under `out`. + out = { + source: { + peek: () => (this._track ? { track: this._track } : undefined), + }, }; constructor(_opts: unknown) {} } return { Broadcast: class { + // The network broadcast, re-created per connection. Undefined here + // since these tests never connect. + net = { peek: () => undefined }; close() {} }, - Audio: { StreamTrack: class {} }, + Audio: { + StreamTrack: class {}, + // 0.4 registers the audio rendition through an encoder rather than + // an inline `audio` prop on the broadcast. + Encoder: class { + sampleRate: unknown; + constructor(_name: string, _opts: unknown) {} + close() {} + }, + }, Source: { Microphone }, }; }); @@ -65,9 +79,19 @@ vi.mock("@moq/watch", () => ({ }, Sync: class {}, Audio: { - Source: class {}, - Decoder: class {}, - Emitter: class {}, + Source: class { + out = { jitter: undefined, config: { peek: () => undefined } }; + close() {} + }, + Decoder: class { + // Read by the transport to tell the source which codecs it can play. + static supported = undefined; + out = { context: undefined, root: undefined }; + close() {} + }, + Emitter: class { + close() {} + }, }, })); vi.mock("@moq/net", () => ({ diff --git a/transports/moq-transport/package.json b/transports/moq-transport/package.json index 1c35f5f..be9706a 100644 --- a/transports/moq-transport/package.json +++ b/transports/moq-transport/package.json @@ -31,12 +31,12 @@ "@pipecat-ai/client-js": "~1.13.0" }, "dependencies": { - "@moq/hang": "^0.2.12", - "@moq/json": "^0.1.2", - "@moq/net": "^0.1.6", - "@moq/publish": "^0.2.16", - "@moq/signals": "^0.1.10", - "@moq/watch": "^0.2.18" + "@moq/hang": "^0.3.4", + "@moq/json": "^0.3.1", + "@moq/net": "^0.2.3", + "@moq/publish": "^0.4.3", + "@moq/signals": "^0.2.1", + "@moq/watch": "^0.4.4" }, "description": "Pipecat Media-over-QUIC (MoQ) Transport Package", "author": "Daily.co", diff --git a/transports/moq-transport/src/moqTransport.ts b/transports/moq-transport/src/moqTransport.ts index 6178d2c..80ebc13 100644 --- a/transports/moq-transport/src/moqTransport.ts +++ b/transports/moq-transport/src/moqTransport.ts @@ -265,6 +265,10 @@ export class MoqTransport extends Transport { // expose this as a Signal so `_connect` can update it from the resolved // options at connect time. private _audioSampleRate = new Signal(undefined); + // The mic's captured source, mirrored out of the Microphone so the + // encoder can consume it as a plain input. + private _micSource = new Signal(undefined); + private _audioEncoder: Publish.Audio.Encoder | null = null; // Watch side (bot → playback). Buffered playback via @moq/watch: // Broadcast -> Audio.Source -> Decoder -> Emitter, with `Sync.latencyMax` @@ -448,43 +452,52 @@ export class MoqTransport extends Transport { connection: this._reload.established, enabled: new Signal(true), name: new Signal(ourPath), - audio: { - source: this._microphone.source, - enabled: this._micEnabled, - sampleRate: this._audioSampleRate, - }, }); + // The mic's captured source drives the encoder, which registers the + // `audio` rendition on the broadcast and fills its catalog entry. + this._signals.run((eff) => { + this._micSource.set(eff.get(this._microphone!.out.source)); + }); + + this._audioEncoder = new Publish.Audio.Encoder("audio", { + broadcast: this._publishBroadcast, + enabled: this._micEnabled, + source: this._micSource, + }); + this._audioEncoder.sampleRate = this._audioSampleRate; + // Client-side transcript: `sendMessage` appends each RTVI message to - // a lossless JSON append-stream. `publishTrack` hands us a fresh track - // per subscription (only one, in the normal single-bot flow), so we - // spin up a `Json.Stream.Producer` per subscriber and replay the - // message log into it — a bot that subscribes after we've already sent - // messages still gets the full log, in order. The bot subscribes to - // this track by its name — same convention as the bot's own transcript - // track — so no catalog entry is needed. + // a lossless JSON append-stream, carried on a track we serve + // alongside the broadcast's own catalog/audio. The producer is + // re-created on each (re)connection — `net` swaps on reconnect — and + // the message log is replayed into it, so a bot that subscribes after + // we've already sent messages still gets the full log, in order. The + // bot subscribes by name, same convention as its own transcript + // track, so no catalog entry is needed. this._transcriptLog = []; this._transcriptOut = new Set>(); - this._publishBroadcast.publishTrack( - merged.transcriptTrack, - (track, effect) => { - const producer = new Json.Stream.Producer(track, { - compression: true, - }); - for (const msg of this._transcriptLog ?? []) producer.append(msg); - this._transcriptOut?.add(producer); - effect.cleanup(() => { - this._transcriptOut?.delete(producer); - producer.finish(); - }); - }, - ); + this._signals.run((eff) => { + const net = eff.get(this._publishBroadcast!.net); + if (!net) return; + + const track = net.createTrack(merged.transcriptTrack); + const producer = new Json.Stream.Producer(track, { + compression: true, + }); + for (const msg of this._transcriptLog ?? []) producer.append(msg); + this._transcriptOut?.add(producer); + eff.cleanup(() => { + this._transcriptOut?.delete(producer); + producer.finish(); + }); + }); // Log the mic settings the browser actually granted, so we can see // when a UA ignores the constraint (e.g. macOS often pins 48k // regardless of `sampleRate.ideal`). this._signals.run((eff) => { - const src = eff.get(this._microphone!.source); + const src = eff.get(this._microphone!.out.source); if (!src) return; const track = "track" in src ? src.track : src; const s = track.getSettings(); @@ -523,8 +536,16 @@ export class MoqTransport extends Transport { // ceiling = how much faster-than-real-time TTS the player will hold // before dropping (audioBufferMaxMs). A number-typed max opens the // buffer; "real-time" collapses to the floor (skip-ahead behavior). + // The source produces the jitter Sync reads, so it's built first to + // avoid a construction cycle. + this._audioSource = new Watch.Audio.Source({ + broadcast: this._watchBroadcast, + supported: Watch.Audio.Decoder.supported, + }); + const sync = new Watch.Sync({ connection: this._reload.established, + audio: this._audioSource.out.jitter, latency: new Signal({ min: merged.audioLatencyMs as Moq.Time.Milli, max: @@ -534,18 +555,15 @@ export class MoqTransport extends Transport { }), }); this._sync = sync; - this._audioSource = new Watch.Audio.Source(sync, { - broadcast: this._watchBroadcast, - }); - this._audioDecoder = new Watch.Audio.Decoder(this._audioSource, { + this._audioDecoder = new Watch.Audio.Decoder(this._audioSource, sync, { enabled: new Signal(true), }); this._audioEmitter = new Watch.Audio.Emitter(this._audioDecoder); // Bridge Decoder.root (AudioNode) → MediaStreamTrack for tracks().bot.audio. this._signals.run((eff) => { - const ctx = eff.get(this._audioDecoder!.context); - const root = eff.get(this._audioDecoder!.root); + const ctx = eff.get(this._audioDecoder!.out.context); + const root = eff.get(this._audioDecoder!.out.root); if (!ctx || !root) return; const dest = ctx.createMediaStreamDestination(); root.connect(dest); @@ -567,8 +585,8 @@ export class MoqTransport extends Transport { // either a parser bug or the bot is advertising a rate that doesn't // match its actual Opus stream. this._signals.run((eff) => { - const config = eff.get(this._audioSource!.config); - const ctx = eff.get(this._audioDecoder!.context); + const config = eff.get(this._audioSource!.out.config); + const ctx = eff.get(this._audioDecoder!.out.context); if (!config && !ctx) return; console.log( `[MoqTransport] consume: catalog codec=${config?.codec}, ` + @@ -592,30 +610,56 @@ export class MoqTransport extends Transport { this._signals.run((eff) => { const conn = eff.get(this._reload!.established); if (!conn) return; - if (!eff.get(this._reload!.announced).has(botPath)) return; - const botBroadcast = conn.consume(botPath); - const track = botBroadcast.subscribe(merged.transcriptTrack, 0); - const consumer = new Json.Stream.Consumer(track, { - compression: true, - }); - const ac = new AbortController(); - this._drainTranscript(consumer, ac.signal).catch((e) => { - if (!ac.signal.aborted) { - console.warn("MoqTransport bot-transcript loop:", e); - } - }); - eff.cleanup(() => { - ac.abort(); - try { - track.close(); - } catch { - // best-effort. + const announced = this._reload!.announced(botPath); + eff.cleanup(() => announced.close()); + + let live: AbortController | undefined; + eff.cleanup(() => live?.abort()); + + eff.spawn(async () => { + for (;;) { + const entry = await Promise.race([eff.cancel, announced.next()]); + if (!entry) return; + + live?.abort(); + live = undefined; + if (!entry.active) continue; + + live = this._subscribeTranscript(conn, botPath, merged.transcriptTrack); } }); }); } + /** Subscribe to the bot's transcript track; returns the handle that stops it. */ + private _subscribeTranscript( + conn: Moq.Connection.Established, + botPath: Moq.Path.Valid, + trackName: string, + ): AbortController { + const botBroadcast = conn.consume(botPath); + const track = botBroadcast.subscribe(trackName, { priority: 0 }); + const consumer = new Json.Stream.Consumer(track, { + compression: true, + }); + + const ac = new AbortController(); + this._drainTranscript(consumer, ac.signal).catch((e) => { + if (!ac.signal.aborted) { + console.warn("MoqTransport bot-transcript loop:", e); + } + }); + ac.signal.addEventListener("abort", () => { + try { + track.close(); + } catch { + // best-effort. + } + }); + return ac; + } + async _disconnect(): Promise { if (this._state === "disconnected") return; this.state = "disconnecting"; @@ -687,7 +731,7 @@ export class MoqTransport extends Transport { // -------------------------------------------------------------------- async getAllMics(): Promise { - return this._microphone?.device.available.peek() ?? []; + return this._microphone?.device.out.available.peek() ?? []; } async getAllCams(): Promise { @@ -707,10 +751,10 @@ export class MoqTransport extends Transport { updateSpeaker(_speakerId: string): void {} get selectedMic(): MediaDeviceInfo | Record { - const id = this._microphone?.device.active.peek(); + const id = this._microphone?.device.out.active.peek(); if (!id) return {}; return ( - this._microphone?.device.available.peek()?.find((d) => d.deviceId === id) ?? + this._microphone?.device.out.available.peek()?.find((d) => d.deviceId === id) ?? {} ); } @@ -736,7 +780,7 @@ export class MoqTransport extends Transport { } get isMicEnabled(): boolean { - return this._micEnabled.get(); + return this._micEnabled.peek(); } get isSharingScreen(): boolean { @@ -763,7 +807,7 @@ export class MoqTransport extends Transport { } tracks(): Tracks { - const localSource = this._microphone?.source.peek(); + const localSource = this._microphone?.out.source.peek(); const localAudio = localSource && "track" in localSource ? localSource.track : localSource; return {