fix(expressive): correctness and pacing fixes for transcript markup stripping

Addresses the issues found reviewing this branch:

- room output: rotate segments by opening the replacement writer before
  closing the current one, so a failed stream_text no longer strands
  self._writer on a closed writer and drops the rest of the turn.
- room output: record _latest_text before the isconnected() check, so a
  brief disconnect no longer finalizes the turn with an earlier chunk.
- room output: lk.expression rides the opening header only, matching the
  ATTRIBUTE_TRANSCRIPTION_EXPRESSION docstring (drops the closing-header
  fallback that contradicted it).
- markup: vanish_trail now sees the character preceding a tag in the
  OUTPUT (new scan_and_replace helper plus a prev_char threaded through
  split_all_markup/_split_expr/extract_and_strip). Fixes words gluing
  across chunk boundaries (Ready.Go) and stacked markers leaving a
  doubled space.
- markup: _emit strips only spaces/tabs, so newlines and indentation
  straddling a chunk boundary survive in every delta transcript.
- markup: the stripper holds only the partial tag, not the text in front
  of it, so words spoken before a marker stay in the current segment and
  a tag-shaped angle bracket no longer stalls the turn until flush.
- markup: strip_all_markup takes an opt-in drop_open_tail. It was applied
  to every input, truncating complete text in final transcripts and in
  stored chat history.
- synchronizer: _visible_hyphens advances a cursor instead of re-stripping
  and re-hyphenating the whole prefix twice per word. An 800-word turn
  drops from 4.40s to 0.0145s of blocking CPU with identical counts, and
  no longer needs prefix truncation.

Adds 19 tests, including a property test asserting streaming output
matches whole-text stripping at every 2- and 3-way split.

Author: tinalenguyen

livekit-agents/livekit/agents/tts/_provider_format.py

@@ -21,7 +21,12 @@
 from typing import TYPE_CHECKING, TypedDict
 
 from ..types import ATTRIBUTE_TRANSCRIPTION_EXPRESSION, TimedString
-from .markup_utils import convert_expression_tags, extract_and_strip, vanish_trail
+from .markup_utils import (
+    convert_expression_tags,
+    extract_and_strip,
+    scan_and_replace,
+    vanish_trail,
+)
 
 
 class ExpressiveTag(TypedDict):
@@ -752,7 +757,7 @@ def _expr_attrs(attrs: str) -> dict[str, str]:
     return dict(_EXPR_ATTR_RE.findall(attrs))
 
 
-def _split_expr(text: str) -> tuple[str, list[ExpressiveTag]]:
+def _split_expr(text: str, *, prev_char: str = "") -> tuple[str, list[ExpressiveTag]]:
     """Strip expr markers and collect (type, label) pairs, in document order.
 
     The generic ``extract_and_strip`` pass can't produce the right ExpressiveTag for
@@ -766,13 +771,18 @@ def _split_expr(text: str) -> tuple[str, list[ExpressiveTag]]:
 
     tags: list[ExpressiveTag] = []
 
-    def _repl(m: re.Match[str]) -> str:
+    def _repl(m: re.Match[str], before: str) -> str:
         attrs = _expr_attrs(m.group(1))
         tags.append({"type": attrs.get("type", ""), "value": attrs.get("label", "")})
-        return vanish_trail(m, m.group("trail"))
-
-    clean = _EXPR_OPEN_STRIP_RE.sub(_repl, text)
-    clean = _EXPR_CLOSE_STRIP_RE.sub(lambda m: vanish_trail(m, m.group("trail")), clean)
+        return vanish_trail(before, m.group("trail"))
+
+    clean = scan_and_replace(_EXPR_OPEN_STRIP_RE, text, _repl, prev_char=prev_char)
+    clean = scan_and_replace(
+        _EXPR_CLOSE_STRIP_RE,
+        clean,
+        lambda m, before: vanish_trail(before, m.group("trail")),
+        prev_char=prev_char,
+    )
     return clean, tags
 
 
@@ -891,7 +901,7 @@ def llm_instructions(provider: str, steering: SpeechSteeringOptions | None = Non
 _ALL_MARKUP_TAGS: list[str] = sorted({tag for tags in _PROVIDER_MARKUP.values() for tag in tags})
 
 
-def split_all_markup(text: str) -> tuple[str, list[ExpressiveTag]]:
+def split_all_markup(text: str, *, prev_char: str = "") -> tuple[str, list[ExpressiveTag]]:
     """Strip the union of every provider's expressive XML markup (provider-agnostic).
 
     The transcript sinks strip downstream, where the originating TTS/provider is no
@@ -909,8 +919,8 @@ def split_all_markup(text: str) -> tuple[str, list[ExpressiveTag]]:
     if "<" not in text:
         return text, []
 
-    text, expr_tags = _split_expr(text)
-    clean, raw_tags = extract_and_strip(text, xml_tags=_ALL_MARKUP_TAGS)
+    text, expr_tags = _split_expr(text, prev_char=prev_char)
+    clean, raw_tags = extract_and_strip(text, xml_tags=_ALL_MARKUP_TAGS, prev_char=prev_char)
     return clean, expr_tags + [{"type": tag, "value": value} for tag, value in raw_tags]
 
 
@@ -942,13 +952,19 @@ def _drop_open_tail(text: str) -> str:
     return text
 
 
-def strip_all_markup(text: str) -> str:
+def strip_all_markup(text: str, *, drop_open_tail: bool = False) -> str:
     """:func:`split_all_markup` returning only the clean text (tags discarded).
 
-    Also drops a trailing unterminated tag, so callers slicing text at arbitrary
-    character offsets that may fall inside a tag never see a partial tag.
+    Args:
+        text: The text to strip.
+        drop_open_tail: Also drop a trailing unterminated tag. For prefixes and
+            mid-stream accumulations only — it is lossy on complete text
+            (``"the <emotion I felt was strong"`` -> ``"the "``), so leave it off
+            for final transcripts and stored chat history.
     """
-    return split_all_markup(_drop_open_tail(text))[0]
+    if drop_open_tail:
+        text = _drop_open_tail(text)
+    return split_all_markup(text)[0]
 
 
 def strip_expr_markup(text: str) -> str:
@@ -1001,10 +1017,9 @@ class TranscriptMarkupStripper:
     def __init__(self) -> None:
         self._buf = ""
         self._tags: list[ExpressiveTag] = []
-        # last emitted char was whitespace (or nothing emitted yet): a stripped tag
-        # often leaves its separating space in the next chunk, so lstrip the next
-        # emission rather than surface a doubled/leading space
-        self._tail_ws = True
+        # last character emitted ("" until the first emission), so a tag opening a chunk
+        # keeps the space separating it from the previous one instead of gluing words
+        self._last_char = ""
 
     def _has_open_tag(self) -> bool:
         # hold a trailing "<" that could still be a known tag (so "3 < 5" or "i<n then"
@@ -1013,30 +1028,37 @@ def _has_open_tag(self) -> bool:
         return _open_tag_fragment(self._buf)
 
     def _emit(self, clean: str) -> str:
-        if self._tail_ws:
-            clean = clean.lstrip()
+        # a stripped tag leaves its separator at the head of the next chunk. Spaces/tabs
+        # only: newlines and indentation are transcript content, not leftover separators
+        if not self._last_char or self._last_char in " \t":
+            clean = clean.lstrip(" \t")
         if clean:
-            self._tail_ws = clean[-1].isspace()
+            self._last_char = clean[-1]
         return clean
 
+    def _consume(self, upto: int | None = None) -> str:
+        head = self._buf if upto is None else self._buf[:upto]
+        self._buf = "" if upto is None else self._buf[upto:]
+        if not head:
+            return ""
+        clean, tags = split_all_markup(head, prev_char=self._last_char)
+        self._tags.extend(tags)
+        return self._emit(clean)
+
     def push(self, text: str) -> str:
         """Feed a chunk; return the clean text ready to emit (may be empty)."""
         self._buf += text
         if self._has_open_tag():
-            return ""
-        clean, tags = split_all_markup(self._buf)
-        self._buf = ""
-        self._tags.extend(tags)
-        return self._emit(clean)
+            # hold only the partial tag: the text in front of it was spoken earlier, so it
+            # belongs to the current wire segment and must not stall until the tag closes
+            return self._consume(self._buf.rfind("<"))
+        return self._consume()
 
     def flush(self) -> str:
         """Drain any buffered text at segment end; return the remaining clean text."""
         if not self._buf:
             return ""
-        clean, tags = split_all_markup(self._buf)
-        self._buf = ""
-        self._tags.extend(tags)
-        return self._emit(clean)
+        return self._consume()
 
     @property
     def tags(self) -> list[ExpressiveTag]:

livekit-agents/livekit/agents/tts/markup_utils.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import re
+from collections.abc import Callable
 
 _EXPRESSION_RE = re.compile(r'<expression\s+value="([^"]*)"(?:\s*/>|>(?:.*?)</expression>)')
 _SOUND_RE = re.compile(r'<sound\s+value="([^"]*)"(?:\s*/>|>(?:.*?)</sound>)')
@@ -16,19 +17,55 @@ def convert_expression_tags(text: str) -> str:
 _VALUE_ATTR_RE = re.compile(r'\b[\w-]+\s*=\s*"([^"]*)"')
 
 
-def vanish_trail(m: re.Match[str], trail: str) -> str:
+def scan_and_replace(
+    pattern: re.Pattern[str],
+    text: str,
+    replace: Callable[[re.Match[str], str], str],
+    *,
+    prev_char: str = "",
+) -> str:
+    """``pattern.sub`` that also tells ``replace`` what character precedes each match.
+
+    The character is the last of the *output* so far, so back-to-back tags all see the
+    one before the first of them. ``prev_char`` seeds it for text continuing an earlier
+    chunk (``""`` means start of stream).
+    """
+    out: list[str] = []
+    last = prev_char
+    pos = 0
+    for m in pattern.finditer(text):
+        gap = text[pos : m.start()]
+        if gap:
+            out.append(gap)
+            last = gap[-1]
+        rep = replace(m, last)
+        if rep:
+            out.append(rep)
+            last = rep[-1]
+        pos = m.end()
+    out.append(text[pos:])
+    return "".join(out)
+
+
+def vanish_trail(before: str, trail: str) -> str:
     """Trailing spaces of a fully removed tag.
 
-    A tag with nothing visible before it on the line takes its trailing spaces with
-    it (``"a <t/> b"`` -> ``"a b"``, ``"<t/> b"`` -> ``"b"``); one glued to the
-    preceding word keeps them as the word separator (``"a.<t/> b"`` -> ``"a. b"``).
+    A tag with nothing visible before it takes its trailing spaces with it
+    (``"a <t/> b"`` -> ``"a b"``, ``"<t/> b"`` -> ``"b"``); one glued to the preceding
+    word keeps them as the word separator (``"a.<t/> b"`` -> ``"a. b"``).
+
+    ``before`` is the preceding character in the *output* (see :func:`scan_and_replace`),
+    so stacked markers collapse to one separator and a tag opening a chunk still sees the
+    previous chunk.
     """
-    if m.start() == 0 or m.string[m.start() - 1].isspace():
+    if not before or before.isspace():
         return ""
     return trail
 
 
-def extract_and_strip(text: str, *, xml_tags: list[str]) -> tuple[str, list[tuple[str, str]]]:
+def extract_and_strip(
+    text: str, *, xml_tags: list[str], prev_char: str = ""
+) -> tuple[str, list[tuple[str, str]]]:
     """Strip XML markup tags and collect the stripped tags in a single pass.
 
     One regex scan both removes the markup and records each removed tag, so
@@ -46,14 +83,15 @@ def extract_and_strip(text: str, *, xml_tags: list[str]) -> tuple[str, list[tupl
       ``"A7X9"``), else its first quoted attribute value
       (``<emotion value="happy"/>`` -> ``"happy"``), falling back to ``""``.
 
-    Wrapping tags keep their inner content in ``clean_text`` (only the delimiters
-    are removed); self-closing and lone tags are removed entirely, along with their
-    trailing spaces where those were only separating the tag from its neighbours
-    (see :func:`vanish_trail`).
+    Wrapping tags keep their inner content in ``clean_text`` (only the delimiters are
+    removed); self-closing and lone tags are removed entirely, along with any trailing
+    spaces that were only separating them (see :func:`vanish_trail`).
 
     Args:
         text: The text containing markup.
         xml_tags: XML tag names to handle (e.g. ``["emotion", "sound"]``).
+        prev_char: Character preceding ``text`` in the stream (``""`` = start of
+            stream), so a tag opening a chunk keeps its separator from the previous one.
     """
     if not xml_tags:
         return text, []
@@ -71,7 +109,7 @@ def extract_and_strip(text: str, *, xml_tags: list[str]) -> tuple[str, list[tupl
     )
     tags: list[tuple[str, str]] = []
 
-    def _repl(m: re.Match[str]) -> str:
+    def _repl(m: re.Match[str], before: str) -> str:
         groups = m.groupdict()
         trail = groups.get("trail") or ""
         tag = groups.get("tag")
@@ -84,9 +122,9 @@ def _repl(m: re.Match[str]) -> str:
                 value = attr_match.group(1) if attr_match else ""
             tags.append((tag, value))
             # wrapping tags keep their inner content; self-closing/lone tags vanish
-            return inner + trail if inner is not None else vanish_trail(m, trail)
+            return inner + trail if inner is not None else vanish_trail(before, trail)
 
-        return vanish_trail(m, trail)  # lone closing tag
+        return vanish_trail(before, trail)  # lone closing tag
 
     # iterate to a fixed point so nested wrapping tags are fully removed: a single pass
     # strips only the outer tag (e.g. <excited><loud>hi</loud></excited> -> keeps the
@@ -96,5 +134,5 @@ def _repl(m: re.Match[str]) -> str:
     prev = None
     while clean != prev:
         prev = clean
-        clean = pattern.sub(_repl, clean)
+        clean = scan_and_replace(pattern, clean, _repl, prev_char=prev_char)
     return clean, tags

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

@@ -279,7 +279,7 @@ async def capture_text(self, text: str) -> None:
         # Stripping the whole accumulation each time avoids partial-tag edge cases; the
         # expression is dropped here — the deprecated rtc Transcription API has no
         # attribute channel (the stream-based output carries lk.expression instead).
-        clean_text = strip_all_markup(self._pushed_text)
+        clean_text = strip_all_markup(self._pushed_text, drop_open_tail=True)
         await self._publish_transcription(self._current_id, clean_text, final=False)
 
     @utils.log_exceptions(logger=logger)
@@ -481,18 +481,29 @@ def _consume_expressions(self) -> None:
     async def _rotate_writer(self, pending: list[ExpressiveTag]) -> None:
         """Finalize the current wire segment and open a new one led by the pending expression."""
         assert self._writer is not None
-        attributes = {ATTRIBUTE_TRANSCRIPTION_FINAL: "true"}
-        if self._track_id:
-            attributes[ATTRIBUTE_TRANSCRIPTION_TRACK_ID] = self._track_id
-        await self._writer.aclose(attributes=attributes)
+        old_writer = self._writer
 
+        # open the replacement first: closing first would strand self._writer on a closed
+        # writer if stream_text fails, silently dropping the rest of the turn
+        prev_id = self._current_id
         self._current_id = utils.shortuuid("SG_")
-        self._writer = await self._create_text_writer(
-            extra_attributes=expression_attribute(pending)
-        )
+        try:
+            new_writer = await self._create_text_writer(
+                extra_attributes=expression_attribute(pending)
+            )
+        except BaseException:
+            self._current_id = prev_id
+            raise
+
+        self._writer = new_writer
         self._writer_expression_sent = True
         self._writer_has_text = False
 
+        attributes = {ATTRIBUTE_TRANSCRIPTION_FINAL: "true"}
+        if self._track_id:
+            attributes[ATTRIBUTE_TRANSCRIPTION_TRACK_ID] = self._track_id
+        await old_writer.aclose(attributes=attributes)
+
     async def _capture_delta(self, piece: str, timing_src: str) -> None:
         clean_text = self._stripper.push(piece)
         if not self._room.isconnected():
@@ -548,10 +559,14 @@ async def capture_text(self, text: str) -> None:
                     await self._capture_delta(piece, text)
             else:  # always create a new writer
                 clean_text, self._segment_tags = split_all_markup(text)
-                if not clean_text or not self._room.isconnected():
+                if not clean_text:
                     return
+                # record before the connection check: flush() republishes _latest_text, so
+                # skipping it while disconnected would finalize an earlier chunk's text
                 payload = self._encode(clean_text, text)
                 self._latest_text = payload
+                if not self._room.isconnected():
+                    return
                 tmp_writer = await self._create_text_writer(
                     extra_attributes=expression_attribute(self._segment_tags)
                 )
@@ -587,9 +602,7 @@ async def _flush_task(
             logger.warning("failed to publish agent transcription to room: %s", e)
 
     def flush(self) -> None:
-        # only emit on a segment that captured text (keeps lk.transcription cadence intact).
-        # The closing header carries the expression only as a fallback when the stream
-        # never got one — e.g. the tag only completed in the flush remainder.
+        # only emit on a segment that captured text (keeps lk.transcription cadence intact)
         if self._participant_identity is None or not self._capturing:
             return
 
@@ -600,8 +613,9 @@ def flush(self) -> None:
         extra_attributes: dict[str, str] | None = None
         if self._is_delta_stream:
             remaining = self._stripper.flush()
-            if not self._writer_expression_sent:
-                extra_attributes = expression_attribute(self._pending_expressions())
+            # lk.expression rides the opening header only (see
+            # ATTRIBUTE_TRANSCRIPTION_EXPRESSION); one completing in the flush remainder
+            # has no text left to apply to
         else:
             remaining = ""
             extra_attributes = expression_attribute(self._segment_tags)