fix(realtime): support externally transcribed user turns

livekit-agents/livekit/agents/__init__.py

@@ -28,6 +28,7 @@
     APITimeoutError,
     AssignmentTimeoutError,
     UnexpectedModelBehavior,
+    UserTurnCommitAbortedError,
     create_api_error_from_http,
 )
 from .job import (
@@ -142,6 +143,7 @@
     EndpointingOptions,
     InterruptionOptions,
     PreemptiveGenerationOptions,
+    RealtimeInputMode,
     TurnHandlingOptions,
     UserTurnLimitOptions,
 )
@@ -236,6 +238,7 @@ def __getattr__(name: str) -> typing.Any:
     "Agent",
     "AgentTask",
     "AssignmentTimeoutError",
+    "UserTurnCommitAbortedError",
     "UnexpectedModelBehavior",
     "APIConnectionError",
     "APIError",
@@ -280,6 +283,7 @@ def __getattr__(name: str) -> typing.Any:
     "AMD",
     "AMDCategory",
     "AMDPredictionEvent",
+    "RealtimeInputMode",
     "TurnHandlingOptions",
     "EndpointingOptions",
     "InterruptionOptions",

livekit-agents/livekit/agents/_exceptions.py

@@ -14,6 +14,18 @@ class AssignmentTimeoutError(Exception):
     pass
 
 
+class UserTurnCommitAbortedError(RuntimeError):
+    """Raised when a manually committed user turn is abandoned before STT settles.
+
+    ``partial_transcript`` contains all final and interim text accumulated for that turn at
+    the abandonment boundary.
+    """
+
+    def __init__(self, partial_transcript: str) -> None:
+        super().__init__("the manually committed user turn was abandoned before completion")
+        self.partial_transcript = partial_transcript
+
+
 # errors used by our plugins
 
 

livekit-agents/livekit/agents/llm/realtime.py

@@ -5,6 +5,7 @@
 from abc import ABC, abstractmethod
 from collections.abc import AsyncIterable, Awaitable
 from dataclasses import dataclass
+from enum import Enum
 from types import TracebackType
 from typing import Generic, Literal, TypeVar
 
@@ -104,6 +105,18 @@ def __init__(self, message: str, *, code: str | None = None) -> None:
         self.code = code
 
 
+class _UserMessageSyncStatus(str, Enum):
+    ACCEPTED = "accepted"
+    REJECTED = "rejected"
+    UNKNOWN = "unknown"
+
+
+@dataclass(frozen=True)
+class _UserMessageSyncResult:
+    status: _UserMessageSyncStatus
+    error: RealtimeError | None = None
+
+
 class RealtimeModel:
     def __init__(self, *, capabilities: RealtimeCapabilities) -> None:
         self._capabilities = capabilities
@@ -250,6 +263,23 @@ async def update_chat_ctx(
         self, chat_ctx: ChatContext
     ) -> None: ...  # can raise RealtimeError on Timeout
 
+    async def _sync_user_message(
+        self, chat_ctx: ChatContext, message_id: str
+    ) -> _UserMessageSyncResult:
+        """Synchronize one finalized user item without changing the public provider API."""
+
+        del message_id  # the compatibility implementation updates the complete context
+        try:
+            await self.update_chat_ctx(chat_ctx)
+        except RealtimeError as error:
+            return _UserMessageSyncResult(_UserMessageSyncStatus.UNKNOWN, error)
+        return _UserMessageSyncResult(_UserMessageSyncStatus.ACCEPTED)
+
+    def _exclude_chat_ctx_item_from_replay(self, item_id: str) -> None:
+        """Keep a locally observable item out of adapter-driven provider replay."""
+
+        del item_id  # only session wrappers that rebuild provider state need to track this
+
     @abstractmethod
     async def update_tools(self, tools: list[Tool]) -> None: ...