Only send modern transcriptions to client sdks which can auto back convert modern -> legacy transcriptions internally

livekit-agents/livekit/agents/types.py

@@ -68,6 +68,12 @@ def recording_enabled(options: Mapping[str, object]) -> bool:
 TOPIC_CHAT = "lk.chat"
 TOPIC_TRANSCRIPTION = "lk.transcription"
 
+CLIENT_PROTOCOL_TRANSCRIPTION_STREAMS = 3
+"""Minimum ``ParticipantInfo.client_protocol`` of a client that rebuilds transcription
+events from ``lk.transcription`` text streams and ignores the deprecated ``rtc.Transcription``
+data packet. While any considered participant is below this, the legacy packet is still
+published. The value is defined by the client SDKs (``client-sdk-js`` ``src/version.ts``)."""
+
 USERDATA_TIMED_TRANSCRIPT = "lk.timed_transcripts"
 """
 The key for the timed transcripts in the audio frame userdata.

livekit-agents/livekit/agents/voice/room_io/_output.py

@@ -23,6 +23,7 @@
     ATTRIBUTE_TRANSCRIPTION_FINAL,
     ATTRIBUTE_TRANSCRIPTION_SEGMENT_ID,
     ATTRIBUTE_TRANSCRIPTION_TRACK_ID,
+    CLIENT_PROTOCOL_TRANSCRIPTION_STREAMS,
     TOPIC_TRANSCRIPTION,
     TimedString,
 )
@@ -264,6 +265,46 @@ async def _forward_audio(self) -> None:
                 self._forwarding_idle.set()
 
 
+def _legacy_transcription_needed(room: rtc.Room) -> bool:
+    """True while some remote participant may still rely on the deprecated
+    ``rtc.Transcription`` data packet.
+
+    ``publish_transcription`` has no destination parameter, so this is all-or-nothing for the
+    room: the packet is dropped only once every considered participant advertises
+    ``client_protocol >= CLIENT_PROTOCOL_TRANSCRIPTION_STREAMS``, meaning it rebuilds
+    transcription events from the ``lk.transcription`` stream channel instead.
+
+    The client protocol is read from the private ``_info``, because livekit-rtc exposes no
+    public property for it. There is no fallback: the field is ``required`` in the FFI
+    protobuf, and ``livekit`` is pinned to an exact version, so it is always present. A
+    rename must fail loudly here rather than report 0 for every participant, which would
+    silently keep legacy publishing on for good.
+
+    Only STANDARD participants -- user-created client SDK instances -- are considered. SIP,
+    INGRESS, AGENT (including avatar workers), CONNECTOR and BRIDGE participants never render
+    legacy transcripts, so their client protocol tells us nothing about whether the legacy
+    packet is still necessary. Counting them would also keep legacy publishing alive in every
+    telephony room for good: the Go SDK does not send a client protocol at all, so SIP and
+    INGRESS participants report 0 permanently. EGRESS participants join hidden and never
+    reach ``remote_participants``.
+    """
+    local_identity = room.local_participant.identity
+    for p in room.remote_participants.values():
+        if p.kind != rtc.ParticipantKind.PARTICIPANT_KIND_STANDARD:
+            continue
+
+        # an out-of-repo avatar worker that joined as STANDARD rather than AGENT. Note this
+        # must not be `_is_local_proxy_participant`, which also matches the participant the
+        # output is attributed to -- for the user output that is the user themselves.
+        if p.attributes.get(ATTRIBUTE_PUBLISH_ON_BEHALF) == local_identity:
+            continue
+
+        if p._info.client_protocol < CLIENT_PROTOCOL_TRANSCRIPTION_STREAMS:
+            return True
+
+    return False
+
+
 class _ParticipantLegacyTranscriptionOutput:
     def __init__(
         self,
@@ -273,6 +314,9 @@ def __init__(
         participant: rtc.Participant | str | None = None,
     ):
         self._room, self._is_delta_stream = room, is_delta_stream
+        # the last status written to the log, so only transitions are logged. This never
+        # takes part in the decision itself.
+        self._legacy_status_logged: bool | None = None
         self._track_id: str | None = None
         self._participant_identity: str | None = None
 
@@ -366,10 +410,30 @@ async def aclose(self) -> None:
         if self._flush_task:
             await self._flush_task
 
+    def _should_publish(self) -> bool:
+        needed = _legacy_transcription_needed(self._room)
+        if needed != self._legacy_status_logged:
+            self._legacy_status_logged = needed
+            logger.debug(
+                "legacy transcription publishing %s",
+                "enabled" if needed else "disabled",
+                extra={"participant": self._participant_identity},
+            )
+
+        return needed
+
     async def _publish_transcription(self, id: str, text: str, final: bool) -> None:
         if self._participant_identity is None or self._track_id is None:
             return
 
+        # Gate here, not in capture_text: every legacy publish carries the whole accumulated
+        # segment under a stable id, so _pushed_text/_current_id must stay warm. A legacy
+        # client that joins mid-segment then gets the complete segment on the very next
+        # publish, and a client that only ever sees the final=True packet still gets a
+        # complete, correctly-closed segment.
+        if not self._should_publish():
+            return
+
         transcription = rtc.Transcription(
             participant_identity=self._represented_by or self._participant_identity,
             track_sid=self._track_id,

tests/test_room_io.py

@@ -16,6 +16,7 @@
 )
 from livekit.agents.voice.room_io._output import (
     _ParticipantAudioOutput,
+    _ParticipantLegacyTranscriptionOutput,
     _ParticipantStreamTranscriptionOutput,
     _ParticipantTranscriptionOutput,
 )
@@ -127,6 +128,50 @@ async def aclose(self, attributes: dict[str, str] | None = None) -> None:
         self.close_calls += 1
 
 
+def _fake_remote(
+    identity: str,
+    *,
+    client_protocol: int = 0,
+    kind: rtc.ParticipantKind.ValueType = rtc.ParticipantKind.PARTICIPANT_KIND_STANDARD,
+    on_behalf: str | None = None,
+) -> SimpleNamespace:
+    """A remote participant stand-in for the legacy-transcription gate.
+
+    `attributes` must be a real dict -- the gate calls `.get` on it -- and
+    `client_protocol` is read off `_info`, matching livekit-rtc's private shape.
+    """
+    attributes: dict[str, str] = {}
+    if on_behalf is not None:
+        attributes["lk.publish_on_behalf"] = on_behalf
+    return SimpleNamespace(
+        identity=identity,
+        kind=kind,
+        attributes=attributes,
+        _info=SimpleNamespace(client_protocol=client_protocol),
+    )
+
+
+def _make_legacy_output(
+    room: _FakeRoom,
+    *,
+    participant_identity: str = "agent",
+) -> _ParticipantLegacyTranscriptionOutput:
+    """A legacy sink wired up without going through `set_participant`, which would walk
+    `track_publications` that the fakes do not have."""
+    output = _ParticipantLegacyTranscriptionOutput(room=room, participant=None)
+    output._participant_identity = participant_identity
+    output._represented_by = participant_identity
+    output._track_id = "TR_legacy"
+    return output
+
+
+async def _capture_and_flush(output: _ParticipantLegacyTranscriptionOutput, text: str) -> None:
+    await output.capture_text(text)
+    output.flush()
+    if output._flush_task is not None:
+        await output._flush_task
+
+
 def _make_track_available_args(
     identity: str = "test-user", sid: str = "TR_123"
 ) -> tuple[MagicMock, MagicMock, MagicMock]:
@@ -680,3 +725,189 @@ async def test_audio_output_waits_for_active_submission_and_source_playout() ->
 
     assert not finished.interrupted
     assert finished.playback_position == pytest.approx(frame.duration)
+
+
+# -- legacy transcription gate ------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_legacy_transcription_skipped_when_all_clients_are_modern() -> None:
+    room = _FakeRoom()
+    room.local_participant.publish_transcription = AsyncMock()
+    room.remote_participants = {
+        "a": _fake_remote("a", client_protocol=3),
+        "b": _fake_remote("b", client_protocol=3),
+    }
+
+    await _capture_and_flush(_make_legacy_output(room), "hello")
+
+    assert room.local_participant.publish_transcription.await_count == 0
+
+
+@pytest.mark.asyncio
+async def test_legacy_transcription_published_when_a_client_is_legacy() -> None:
+    room = _FakeRoom()
+    room.local_participant.publish_transcription = AsyncMock()
+    room.remote_participants = {
+        "modern": _fake_remote("modern", client_protocol=3),
+        "legacy": _fake_remote("legacy", client_protocol=0),
+    }
+
+    await _capture_and_flush(_make_legacy_output(room), "hello")
+
+    calls = room.local_participant.publish_transcription.await_args_list
+    assert len(calls) == 2
+
+    partial = calls[0].args[0].segments[0]
+    assert partial.text == "hello"
+    assert partial.final is False
+
+    final = calls[1].args[0].segments[0]
+    assert final.text == "hello"
+    assert final.final is True
+    assert final.id == partial.id
+
+
+@pytest.mark.asyncio
+async def test_legacy_transcription_skipped_with_no_standard_participants() -> None:
+    """An empty considered set means skip: a SIP-only room has nobody who renders text."""
+    for remotes in (
+        {},
+        {"sip": _fake_remote("sip", kind=rtc.ParticipantKind.PARTICIPANT_KIND_SIP)},
+    ):
+        room = _FakeRoom()
+        room.local_participant.publish_transcription = AsyncMock()
+        room.remote_participants = dict(remotes)
+
+        await _capture_and_flush(_make_legacy_output(room), "hello")
+
+        assert room.local_participant.publish_transcription.await_count == 0
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "kind",
+    [
+        rtc.ParticipantKind.PARTICIPANT_KIND_SIP,
+        rtc.ParticipantKind.PARTICIPANT_KIND_INGRESS,
+        rtc.ParticipantKind.PARTICIPANT_KIND_AGENT,
+        rtc.ParticipantKind.PARTICIPANT_KIND_CONNECTOR,
+    ],
+)
+async def test_legacy_transcription_ignores_non_standard_kinds(kind) -> None:
+    """Only user-created client SDK instances gate the legacy packet."""
+    room = _FakeRoom()
+    room.local_participant.publish_transcription = AsyncMock()
+    room.remote_participants = {
+        "user": _fake_remote("user", client_protocol=3),
+        "service": _fake_remote("service", client_protocol=0, kind=kind),
+    }
+
+    await _capture_and_flush(_make_legacy_output(room), "hello")
+
+    assert room.local_participant.publish_transcription.await_count == 0
+
+
+@pytest.mark.asyncio
+async def test_legacy_transcription_published_for_legacy_standard_participant() -> None:
+    room = _FakeRoom()
+    room.local_participant.publish_transcription = AsyncMock()
+    room.remote_participants = {"user": _fake_remote("user", client_protocol=0)}
+
+    await _capture_and_flush(_make_legacy_output(room), "hello")
+
+    assert room.local_participant.publish_transcription.await_count == 2
+
+
+@pytest.mark.asyncio
+async def test_legacy_transcription_excludes_our_own_avatar_worker() -> None:
+    room = _FakeRoom()
+    room.local_participant.publish_transcription = AsyncMock()
+    room.remote_participants = {
+        "user": _fake_remote("user", client_protocol=3),
+        "avatar": _fake_remote("avatar", client_protocol=0, on_behalf="local"),
+    }
+
+    await _capture_and_flush(_make_legacy_output(room), "hello")
+
+    assert room.local_participant.publish_transcription.await_count == 0
+
+
+@pytest.mark.asyncio
+async def test_legacy_transcription_counts_another_agents_avatar_worker() -> None:
+    room = _FakeRoom()
+    room.local_participant.publish_transcription = AsyncMock()
+    room.remote_participants = {
+        "user": _fake_remote("user", client_protocol=3),
+        "avatar": _fake_remote("avatar", client_protocol=0, on_behalf="other-agent"),
+    }
+
+    await _capture_and_flush(_make_legacy_output(room), "hello")
+
+    assert room.local_participant.publish_transcription.await_count == 2
+
+
+@pytest.mark.asyncio
+async def test_legacy_transcription_never_mistakes_the_user_for_a_proxy() -> None:
+    """Regression guard: the exclusion must not reuse `_is_local_proxy_participant`, which
+    also matches the participant the output is attributed to."""
+    room = _FakeRoom()
+    room.local_participant.publish_transcription = AsyncMock()
+    room.remote_participants = {"test-user": _fake_remote("test-user", client_protocol=0)}
+
+    output = _make_legacy_output(room, participant_identity="test-user")
+    await _capture_and_flush(output, "hello")
+
+    assert room.local_participant.publish_transcription.await_count == 2
+
+
+@pytest.mark.asyncio
+async def test_legacy_transcription_gate_is_dynamic_and_keeps_state_warm() -> None:
+    """A legacy client joining mid-segment gets the whole accumulated segment, which only
+    holds while the gate sits at the publish site rather than in capture_text."""
+    room = _FakeRoom()
+    room.local_participant.publish_transcription = AsyncMock()
+    room.remote_participants = {"modern": _fake_remote("modern", client_protocol=3)}
+
+    output = _make_legacy_output(room)
+
+    await output.capture_text("hello ")
+    assert room.local_participant.publish_transcription.await_count == 0
+
+    room.remote_participants["legacy"] = _fake_remote("legacy", client_protocol=0)
+
+    await output.capture_text("world")
+    assert room.local_participant.publish_transcription.await_count == 1
+    partial = room.local_participant.publish_transcription.await_args_list[0].args[0].segments[0]
+    assert partial.text == "hello world"
+    assert partial.final is False
+
+    output.flush()
+    assert output._flush_task is not None
+    await output._flush_task
+
+    final = room.local_participant.publish_transcription.await_args_list[1].args[0].segments[0]
+    assert final.text == "hello world"
+    assert final.final is True
+    assert final.id == partial.id
+
+
+@pytest.mark.asyncio
+async def test_modern_stream_still_published_when_legacy_is_skipped() -> None:
+    room = _FakeRoom()
+    room.local_participant.publish_transcription = AsyncMock()
+    writer = _FakeWriter()
+    room.local_participant.stream_text = AsyncMock(return_value=writer)
+    room.remote_participants = {"modern": _fake_remote("modern", client_protocol=3)}
+
+    output = _ParticipantTranscriptionOutput(room=room, participant="agent")
+    legacy_output, _ = output._ParticipantTranscriptionOutput__outputs
+    legacy_output._track_id = "TR_legacy"
+
+    await output.capture_text("hello")
+    output.flush()
+    if legacy_output._flush_task is not None:
+        await legacy_output._flush_task
+
+    assert "".join(writer.chunks) == "hello"
+    assert room.local_participant.publish_transcription.await_count == 0