feat(events): streaming/partial message reconciliation, textOnly serialization

- Message.isPartial (true while an agent message is streaming in from
  agent_chat_response_part chunks, or a user transcript is tentative;
  false once finalized or for locally-appended messages) -- lands here,
  not with the rest of the event-model changes, because adding the field
  while this file's message-store logic was still unmigrated was
  verified to cause runtime data corruption in unrelated codepaths
  (a sendToolResult call started publishing garbage that looked like an
  unrelated feedback event) -- a partial-migration hazard specific to
  that combination, not present in either the fully-old or fully-new state.
- Conversation+Events.swift: insertUserTranscript/upsertAgentMessage
  replaced by applyUserTranscript/applyTentativeUserTranscript/
  applyAgentResponse/applyAgentResponsePart, backed by shared
  messageIndex/isNewerThanHighestEventId/appendMessage helpers --
  guards against reopening a finalized message or applying a stale/
  out-of-order part, which the old version didn't have. Ping replies
  now fire via a detached Task instead of being awaited inline, so a
  slow publish doesn't stall the serialized event-handling queue.
  callbacks.onPing added.
  Two behavior changes worth explicit reviewer sign-off, not mentioned
  in the v4 import's stated deviations: the .agentToolResponse handler
  for toolName == "end_call" no longer calls endConversation() (grep
  confirms nothing else does either -- server-driven auto-hangup appears
  gone), and onCanSendFeedbackChange/lastFeedbackSubmittedEventId
  tracking is removed entirely with no replacement.
- EventSerializer/OutgoingEvents: config.textOnly (flattened from
  conversationOverrides.textOnly) and DynamicVariableValue-based
  dynamic_variables serialization (was String-only) now that
  ConversationConfig carries them. ClientToolResultEvent's Any-based
  init is no longer deprecated (the Encodable overload it pointed
  reviewers to is gone -- see ConversationClient in the next commit)
  and throws ConversationError.invalidToolResult instead of silently
  stringifying an un-encodable result.

Author: renal128

Sources/ElevenLabs/Internal/Conversation/AgentStateManager.swift

@@ -15,8 +15,8 @@ private typealias TimerTask = Task<Void, Never>
 
 @MainActor
 final class AgentStateManager {
-    private(set) var currentState: ElevenLabs.AgentState = .listening
-    var onStateChange: ((ElevenLabs.AgentState) -> Void)?
+    private(set) var currentState: AgentState = .listening
+    var onStateChange: ((AgentState) -> Void)?
 
     private let configuration: AgentStateConfiguration
 
@@ -90,7 +90,7 @@ final class AgentStateManager {
         }
     }
 
-    private func transitionTo(_ newState: ElevenLabs.AgentState) {
+    private func transitionTo(_ newState: AgentState) {
         guard newState != currentState else { return }
         currentState = newState
         onStateChange?(newState)

Sources/ElevenLabs/Internal/Conversation/Conversation+Events.swift

@@ -8,132 +8,208 @@ extension Conversation {
     func handleIncomingEvent(_ event: IncomingEvent) async {
         switch event {
         case let .userTranscript(e):
-            insertUserTranscript(content: e.transcript, eventId: e.eventId)
+            applyUserTranscript(content: e.transcript, eventId: e.eventId)
+            callbacks.onUserTranscript?(e.transcript, e.eventId)
             agentStateManager?.processSignal(.userTranscript)
-            options.onUserTranscript?(e.transcript, e.eventId)
+
+        case let .tentativeUserTranscript(e):
+            applyTentativeUserTranscript(content: e.transcript, eventId: e.eventId)
+            callbacks.onTentativeUserTranscript?(e.transcript, e.eventId)
 
         case let .agentResponse(e):
-            upsertAgentMessage(content: e.response, eventId: e.eventId)
-            lastAgentEventId = e.eventId
+            applyAgentResponse(content: e.response, eventId: e.eventId)
+            callbacks.onAgentResponse?(e.response, e.eventId)
             agentStateManager?.processSignal(.agentResponse)
-            options.onAgentResponse?(e.response, e.eventId)
-            if lastFeedbackSubmittedEventId.map({ e.eventId > $0 }) ?? true {
-                options.onCanSendFeedbackChange?(true)
-            }
 
         case let .agentResponseCorrection(correction):
-            upsertAgentMessage(content: correction.correctedAgentResponse, eventId: correction.eventId)
-            options.onAgentResponseCorrection?(
+            applyAgentResponse(
+                content: correction.correctedAgentResponse,
+                eventId: correction.eventId
+            )
+            callbacks.onAgentResponseCorrection?(
                 correction.originalAgentResponse,
                 correction.correctedAgentResponse,
                 correction.eventId
             )
 
+        case let .agentChatResponsePart(e):
+            applyAgentResponsePart(text: e.text, type: e.type, eventId: e.eventId)
+            callbacks.onAgentResponsePart?(e.text, e.type, e.eventId)
+
         case let .agentResponseMetadata(metadata):
-            options.onAgentResponseMetadata?(
+            callbacks.onAgentResponseMetadata?(
                 metadata.metadataData,
                 metadata.eventId
             )
 
-        case let .agentChatResponsePart(e):
-            let existing = messages.last(where: { $0.role == .agent && $0.eventId == e.eventId })?.content ?? ""
-            upsertAgentMessage(content: existing + e.text, eventId: e.eventId)
-
         case let .audio(audioEvent):
-            latestAudioEvent = audioEvent
-            latestAudioAlignment = audioEvent.alignment
             if let alignment = audioEvent.alignment {
-                options.onAudioAlignment?(alignment)
+                callbacks.onAudioAlignment?(alignment)
             }
 
         case let .interruption(interruptionEvent):
             speakingTimer?.cancel()
-            applyStateSignal(.interruption, fallback: .listening)
-            options.onInterruption?(interruptionEvent.eventId)
-            options.onCanSendFeedbackChange?(false)
+            isAgentSpeaking = false
+            feedAgentState(.interruption, fallback: .listening)
+            callbacks.onInterruption?(interruptionEvent.eventId)
 
         case let .conversationMetadata(metadata):
-            // Store the conversation metadata for public access
             conversationMetadata = metadata
-            options.onConversationMetadata?(metadata)
+            // This event completes the startup handshake: release any waiter
+            // blocking `connect()` on metadata receipt.
+            resumeConversationMetadataWaiter()
 
         case let .ping(p):
-            // Respond to ping with pong
-            let pong = OutgoingEvent.pong(PongEvent(eventId: p.eventId))
-            try? await publish(pong)
-
-        case let .clientToolCall(toolCall):
-            // Add to pending tool calls for the app to handle
-            options.onUnhandledClientToolCall?(toolCall)
-            pendingToolCalls.append(toolCall)
+            if let pingMs = p.pingMs {
+                callbacks.onPing?(pingMs)
+            }
+            // Send pong off the serialized handler loop: awaiting the publish
+            // here would let a slow transport stall delivery of every queued
+            // event behind this heartbeat. Pong is keyed by `eventId`, so
+            // out-of-order delivery is fine.
+            let eventId = p.eventId
+            Task { @MainActor [weak self] in
+                try? await self?.publish(.pong(PongEvent(eventId: eventId)))
+            }
 
         case let .vadScore(vad):
+            callbacks.onVadScore?(vad.vadScore)
             agentStateManager?.processSignal(.vadScore(vad.vadScore))
-            options.onVadScore?(vad.vadScore)
-
-        case let .agentToolResponse(toolResponse):
-            applyStateSignal(.agentToolResponse, fallback: .listening)
 
-            if toolResponse.toolName == "end_call" {
-                await endConversation()
-            }
-            options.onAgentToolResponse?(toolResponse)
+        case let .clientToolCall(toolCall):
+            // Append before invoking the callback so a handler that inspects
+            // `pendingToolCalls` (directly or via the mirrored client property)
+            // already sees the new call.
+            pendingToolCalls.append(toolCall)
+            callbacks.onClientToolCall?(toolCall)
 
         case let .agentToolRequest(toolRequest):
-            applyStateSignal(.agentToolRequest, fallback: .thinking)
-            options.onAgentToolRequest?(toolRequest)
+            feedAgentState(.agentToolRequest, fallback: .thinking)
+            callbacks.onAgentToolRequest?(toolRequest)
 
-        case .tentativeUserTranscript:
-            // Tentative user transcript (in-progress transcription)
-            break
+        case let .agentToolResponse(toolResponse):
+            feedAgentState(.agentToolResponse, fallback: .listening)
+            callbacks.onAgentToolResponse?(toolResponse)
 
         case let .mcpToolCall(toolCall):
-            // Update or append MCP tool call based on toolCallId
             if let index = mcpToolCalls.firstIndex(where: { $0.toolCallId == toolCall.toolCallId }) {
                 mcpToolCalls[index] = toolCall
             } else {
                 mcpToolCalls.append(toolCall)
             }
 
         case let .mcpConnectionStatus(status):
-            // Update MCP connection status
             mcpConnectionStatus = status
 
         case let .error(errorEvent):
-            logger.error("Received error event from server: code=\(errorEvent.code), message=\(errorEvent.message ?? "none")")
-            options.onError?(.serverError(errorEvent))
+            logger.error("Received error event from server: code=\(errorEvent.code), name=\(errorEvent.errorName ?? "none"), message=\(errorEvent.message ?? "none")")
+            callbacks.onError?(.serverError(errorEvent))
         }
     }
 
-    /// Inserts the user transcript before the agent message with the same `eventId`
-    /// if one exists, since the agent's response may be received before the transcript.
-    private func insertUserTranscript(content: String, eventId: Int) {
-        let message = Message(
-            id: UUID().uuidString,
-            role: .user,
-            content: content,
-            timestamp: Date(),
-            eventId: eventId
-        )
-        if let agentIdx = messages.firstIndex(where: { $0.role == .agent && $0.eventId == eventId }) {
-            messages.insert(message, at: agentIdx)
+    // MARK: - Transcript / response reconciliation
+    //
+    // `messages` is keyed by role + event id and only ever appended, never
+    // reordered, so order follows the arrival of finalized text:
+    //   * Finalized text (`agent_response`, `agent_response_correction`,
+    //     `user_transcript`) is always recorded — a matching event id updates in
+    //     place, otherwise it's appended (even out of order).
+    //   * Streaming parts (`agent_chat_response_part`, `tentative_user_transcript`)
+    //     only open a new partial when their event id is newer than the role's
+    //     highest; otherwise they're stale and ignored.
+    // Event ids stay unique per role; partial user transcripts are cleared on
+    // every tentative/final user transcript.
+
+    /// `agent_chat_response_part`: accumulates streamed text. A finalized message
+    /// (`.stop` already seen) is never reopened, and a stale part (older than the
+    /// agent's highest event id) never opens a new bubble.
+    private func applyAgentResponsePart(text: String, type: AgentChatResponsePartType, eventId: Int) {
+        let isPartial = type != .stop
+        guard let idx = messageIndex(role: .agent, eventId: eventId) else {
+            if isNewerThanHighestEventId(role: .agent, eventId: eventId) {
+                appendMessage(role: .agent, content: text, eventId: eventId, isPartial: isPartial)
+            }
+            return
+        }
+        guard messages[idx].isPartial else { return }
+        messages[idx] = messages[idx].updating(content: messages[idx].content + text, eventId: eventId, isPartial: isPartial)
+    }
+
+    /// `agent_response` | `agent_response_correction`: the finalized response for a
+    /// turn. Replaces the matching message in place, or records it (appending,
+    /// even out of order) when no slot exists yet.
+    private func applyAgentResponse(content: String, eventId: Int) {
+        if let idx = messageIndex(role: .agent, eventId: eventId) {
+            messages[idx] = messages[idx].updating(content: content, eventId: eventId, isPartial: false)
         } else {
-            messages.append(message)
+            appendMessage(role: .agent, content: content, eventId: eventId, isPartial: false)
         }
     }
 
-    private func upsertAgentMessage(content: String, eventId: Int) {
-        if let idx = messages.lastIndex(where: { $0.role == .agent && $0.eventId == eventId }) {
-            let existing = messages[idx]
-            messages[idx] = Message(
-                id: existing.id,
-                role: .agent,
-                content: content,
-                timestamp: existing.timestamp,
-                eventId: eventId
-            )
+    /// `user_transcript`: the finalized user transcript. Finalizes the matching
+    /// in-progress partial, or records it (appending, even out of order) when no
+    /// slot exists; then drops any leftover partial (a tentative that never
+    /// produced its own final).
+    private func applyUserTranscript(content: String, eventId: Int) {
+        if let idx = messageIndex(role: .user, eventId: eventId) {
+            messages[idx] = messages[idx].updating(content: content, eventId: eventId, isPartial: false)
         } else {
-            appendMessage(role: .agent, content: content, eventId: eventId)
+            appendMessage(role: .user, content: content, eventId: eventId, isPartial: false)
         }
+        // A finalized transcript ends the turn, so any leftover in-progress
+        // partial is stale and removed.
+        messages.removeAll { $0.role == .user && $0.isPartial }
+    }
+
+    /// `tentative_user_transcript`: the in-progress user transcript. Supersedes any
+    /// existing partial, then surfaces a fresh one if it belongs to a turn newer
+    /// than the user's highest event id.
+    private func applyTentativeUserTranscript(content: String, eventId: Int) {
+        messages.removeAll { $0.role == .user && $0.isPartial }
+        guard isNewerThanHighestEventId(role: .user, eventId: eventId) else { return }
+        appendMessage(role: .user, content: content, eventId: eventId, isPartial: true)
+    }
+
+    /// Index of the `role` message with exactly `eventId`, scanning tail-first so
+    /// the common "touch the latest message" case is cheap. Event ids are unique
+    /// per role (matches update in place), so first/last match the same element.
+    private func messageIndex(role: Message.Role, eventId: Int) -> Int? {
+        messages.lastIndex { $0.role == role && $0.eventId == eventId }
+    }
+
+    /// Whether `eventId` is greater than the highest event id recorded for `role`.
+    /// Uses the max rather than the last message, because finalized responses can
+    /// append out of order and locally-sent messages carry no event id (skipped).
+    private func isNewerThanHighestEventId(role: Message.Role, eventId: Int) -> Bool {
+        guard let highest = messages.compactMap({ $0.role == role ? $0.eventId : nil }).max() else { return true }
+        return eventId > highest
+    }
+
+    private func appendMessage(role: Message.Role, content: String, eventId: Int, isPartial: Bool) {
+        messages.append(
+            Message(
+                id: UUID().uuidString,
+                role: role,
+                content: content,
+                timestamp: Date(),
+                eventId: eventId,
+                isPartial: isPartial
+            )
+        )
+    }
+}
+
+private extension Message {
+    /// A copy with new `content`/`eventId`/`isPartial`, preserving the stable
+    /// `id`, `role`, and `timestamp` so SwiftUI identity and ordering hold.
+    func updating(content: String, eventId: Int?, isPartial: Bool) -> Message {
+        Message(
+            id: id,
+            role: role,
+            content: content,
+            timestamp: timestamp,
+            eventId: eventId,
+            isPartial: isPartial
+        )
     }
 }

Sources/ElevenLabs/Internal/Utilities/EventSerializer.swift

@@ -10,9 +10,6 @@ enum EventSerializer {
             json["type"] = "pong"
             json["event_id"] = pongEvent.eventId
 
-        case let .userAudio(audioEvent):
-            json["user_audio_chunk"] = audioEvent.audioChunk
-
         case let .conversationInit(initEvent):
             json["type"] = "conversation_initiation_client_data"
             if let config = initEvent.config {
@@ -29,7 +26,9 @@ enum EventSerializer {
             json["tool_call_id"] = resultEvent.toolCallId
             json["result"] = resultEvent.result
             json["is_error"] = resultEvent.isError
-            json["error_type"] = resultEvent.errorType?.rawValue
+            if let errorType = resultEvent.errorType {
+                json["error_type"] = errorType.rawValue
+            }
 
         case let .contextualUpdate(updateEvent):
             json["type"] = "contextual_update"
@@ -93,32 +92,19 @@ enum EventSerializer {
         }
 
         // Conversation overrides
-        if let conversationOverrides = config.conversationOverrides {
-            var conversation: [String: Any] = [:]
-            if conversationOverrides.textOnly {
-                conversation["text_only"] = true
-            }
-            if let clientEvents = conversationOverrides.clientEvents {
-                conversation["client_events"] = clientEvents
-            }
-            if !conversation.isEmpty {
-                configOverride["conversation"] = conversation
-            }
+        if config.textOnly {
+            configOverride["conversation"] = ["text_only": true]
         }
 
         if !configOverride.isEmpty {
             json["conversation_config_override"] = configOverride
         }
 
-        if let customBody = config.customLlmExtraBody {
-            json["custom_llm_extra_body"] = customBody
-        }
-
         if let dynamicVars = config.dynamicVariables {
-            json["dynamic_variables"] = dynamicVars
+            json["dynamic_variables"] = dynamicVars.mapValues(\.jsonObject)
         }
 
-        // Add source_info (equivalent to client in React Native)
+        // Identify the SDK to the orchestrator.
         var sourceInfo: [String: Any] = [:]
         sourceInfo["source"] = "swift_sdk"
         sourceInfo["version"] = SDKVersion.version

Sources/ElevenLabs/Public/Conversation/Models/Message.swift

@@ -7,9 +7,31 @@ public struct Message: Identifiable, Sendable {
     public let timestamp: Date
     /// Server-assigned event id used for per-message operations like `sendFeedback`; `nil` for locally appended messages.
     public let eventId: Int?
+    /// Whether the message is still being assembled and may change. `true` while
+    /// an agent message is streaming in from `agent_chat_response_part` chunks,
+    /// or while a user message reflects an in-progress (tentative) transcript.
+    /// It flips to `false` once the finalized agent response or user transcript
+    /// arrives. Locally appended messages are always final (`false`).
+    public let isPartial: Bool
 
     public enum Role: Sendable {
         case user
         case agent
     }
+
+    init(
+        id: String,
+        role: Role,
+        content: String,
+        timestamp: Date,
+        eventId: Int?,
+        isPartial: Bool = false
+    ) {
+        self.id = id
+        self.role = role
+        self.content = content
+        self.timestamp = timestamp
+        self.eventId = eventId
+        self.isPartial = isPartial
+    }
 }