From 0f0023efe487e2c7926362b58068a5741af34142 Mon Sep 17 00:00:00 2001 From: Luke Curley <432854+kixelated@users.noreply.github.com> Date: Mon, 19 Oct 2026 12:10:26 -0700 Subject: [PATCH 1/3] Start broadcast lifetime epoch fix Co-Authored-By: GPT-6 From 7bf3ed98828160a80c97daa53477a24f3b9c3919 Mon Sep 17 00:00:00 2001 From: Luke Curley <432854+kixelated@users.noreply.github.com> Date: Mon, 19 Oct 2026 12:28:53 -0700 Subject: [PATCH 2/3] fix(moq): mint an epoch for each published broadcast Co-Authored-By: GPT-6 --- changelog/5986.fixed.md | 2 +- src/pipecat/transports/moq/transport.py | 17 ++++++++-------- tests/test_moq_transport.py | 27 +++++++++++++++++++++++++ 3 files changed, 37 insertions(+), 9 deletions(-) diff --git a/changelog/5986.fixed.md b/changelog/5986.fixed.md index e31cb41ecea..c854a4138bc 100644 --- a/changelog/5986.fixed.md +++ b/changelog/5986.fixed.md @@ -1 +1 @@ -- Fixed `MOQTransport` ending the call when a relay cancelled the peer's audio subscription, which a relay does when the session publishing the peer's broadcast ends: the peer's relay went away, or a reconnecting peer replaced its broadcast with a fresh one. The cancel is now treated as the peer's tracks ending, so the transport subscribes again or redials and follows the peer to its new broadcast. +- Fixed `MOQTransport` ending the call when a relay cancelled the peer's audio subscription, which a relay does when the session publishing the peer's broadcast ends: the peer's relay went away, or a reconnecting peer replaced its broadcast with a fresh one. The cancel is now treated as the peer's tracks ending, so the transport subscribes again or redials and follows the peer to its new broadcast. Each rebuilt publish broadcast carries a fresh route epoch; re-announcements of that broadcast retain it. diff --git a/src/pipecat/transports/moq/transport.py b/src/pipecat/transports/moq/transport.py index d6dce3dfcbd..6c6a7a7e0a9 100644 --- a/src/pipecat/transports/moq/transport.py +++ b/src/pipecat/transports/moq/transport.py @@ -649,6 +649,7 @@ def __init__( self._publish_broadcast: moq.BroadcastProducer = self._publish_origin.create_broadcast( self._broadcast_path ) + self._publish_route = moq.Route(epoch=moq.mint_epoch()) # Lossless, ordered JSON append-log for RTVI (compression on). The # stream helper delivers every appended record in order — the # snapshot helper (publish_json_snapshot) would collapse to the @@ -661,7 +662,7 @@ def __init__( # transcript is that track, announce now. Audio on: # ``open_audio_track`` announces after adding the audio track. if not params.audio_out_enabled: - self._publish_broadcast.announce() + self._publish_broadcast.announce(self._publish_route) # Audio track is opened lazily once the pipeline's output sample # rate is known (in :meth:`open_audio_track`). We stash the rate # because ``publish_audio`` uses it to convert byte length to @@ -794,7 +795,7 @@ def open_audio_track(self, sample_rate: int): # Announce broadcast to peers now that subscriber-needed tracks exist # (catalog via audio rendition above, transcript stream). moq-rs 0.5 # requires explicit announce; until announced, broadcast is unroutable. - self._publish_broadcast.announce() + self._publish_broadcast.announce(self._publish_route) @property def _audio_track_name(self) -> str: @@ -830,11 +831,10 @@ def _publish_audio_track(self, sample_rate: int) -> "moq.AudioProducer": async def _rebuild_publish_side(self): """Replace the publish origin, broadcast and tracks before a redial. - The origin carries the publisher identity the relay sees. A redial - that re-announces under the identity of a publisher the relay still - holds a dead route for leaves the path stuck on that route, while a - new identity is parked until the dead route is reaped and then takes - the path over. The transcript log is replayed into the new stream + A rebuilt broadcast gets a fresh route epoch, so peers replace the + previous instance instead of resuming its cached groups. Re-announcing + the same broadcast keeps its epoch. The transcript log is replayed into + the new stream with its original ``seq`` and ``epoch``, so subscribers drop what they already have, and the audio track reopens at the same rate. @@ -853,6 +853,7 @@ async def _rebuild_publish_side(self): pass self._publish_origin = moq.OriginProducer() self._publish_broadcast = self._publish_origin.create_broadcast(self._broadcast_path) + self._publish_route = moq.Route(epoch=moq.mint_epoch()) self._transcript_out = self._publish_broadcast.publish_json_stream( self._params.transcript_track, compression=True ) @@ -862,7 +863,7 @@ async def _rebuild_publish_side(self): self._audio_out = self._publish_audio_track(self._audio_out_sample_rate) # Re-announce rebuilt broadcast (moq-rs 0.5 requires explicit # announce; see open_audio_track). - self._publish_broadcast.announce() + self._publish_broadcast.announce(self._publish_route) # By index against the live log: a record published while the # replay yields is only logged (see publish_transcript), and is # written from here, after the records before it. diff --git a/tests/test_moq_transport.py b/tests/test_moq_transport.py index ce4fdfc9a9d..68d95d7ca21 100644 --- a/tests/test_moq_transport.py +++ b/tests/test_moq_transport.py @@ -2461,3 +2461,30 @@ async def test_catalog_advertises_the_replacement_track(self): if __name__ == "__main__": unittest.main() + + +class TestBroadcastEpochs(unittest.IsolatedAsyncioTestCase): + """The broadcast instance, independently of its tracks, owns route identity.""" + + async def test_reannouncements_keep_identity_and_rebuild_renews_it(self): + client, _stream = _client_with_fake_moq() + client.open_audio_track(24000) + old = client._publish_broadcast + self.assertEqual(len(old.announce.call_args.args), 1) + first = old.announce.call_args.args[0] + self.assertIsNotNone(first.epoch) + client.open_audio_track(24000) + self.assertEqual(old.announce.call_args.args[0].epoch, first.epoch) + with patch.object(moq_transport.moq, "OriginProducer", side_effect=_fake_origin): + await client._rebuild_publish_side() + second = client._publish_broadcast.announce.call_args.args[0] + self.assertNotEqual(second.epoch, first.epoch) + + async def test_transcript_only_instances_are_distinct(self): + a, _ = _client_with_fake_moq(MOQParams(audio_out_enabled=False)) + b, _ = _client_with_fake_moq(MOQParams(audio_out_enabled=False)) + calls = [client._publish_broadcast.announce.call_args for client in (a, b)] + for call in calls: + self.assertEqual(len(call.args), 1) + self.assertIsNotNone(call.args[0].epoch) + self.assertNotEqual(calls[0].args[0].epoch, calls[1].args[0].epoch) From 207f86b8203d6154b03962ab9d257c992e2955fe Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 19 Oct 2026 23:41:48 -0700 Subject: [PATCH 3/3] test(moq): define the epoch tests before the main guard Run directly, unittest.main() exits before a class defined below it exists. Co-Authored-By: Claude Opus 5.5 --- tests/test_moq_transport.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/tests/test_moq_transport.py b/tests/test_moq_transport.py index 68d95d7ca21..7ec112a1f24 100644 --- a/tests/test_moq_transport.py +++ b/tests/test_moq_transport.py @@ -2459,10 +2459,6 @@ async def test_catalog_advertises_the_replacement_track(self): self.assertEqual(set(updated.audio), {"bot-audio-1"}) -if __name__ == "__main__": - unittest.main() - - class TestBroadcastEpochs(unittest.IsolatedAsyncioTestCase): """The broadcast instance, independently of its tracks, owns route identity.""" @@ -2488,3 +2484,7 @@ async def test_transcript_only_instances_are_distinct(self): self.assertEqual(len(call.args), 1) self.assertIsNotNone(call.args[0].epoch) self.assertNotEqual(calls[0].args[0].epoch, calls[1].args[0].epoch) + + +if __name__ == "__main__": + unittest.main()