fix(mcp): resolve data race and reliability issues in user_collaboration tool

Three fixes for intermittent user_collaboration failures:

1. UserCollaborationTool: pendingResponses dictionary had a declared but
   unused NSLock. All reads/writes now go through lockedRead/lockedWrite
   helpers, preventing data races between Vapor's thread (submitUserResponse)
   and @mainactor polling (waitForUserResponse).

2. AgentOrchestrator: duplicate check for user response persistence was
   checking conversation.messages (old array) instead of messageBus.messages
   (source of truth), causing responses to sometimes not be persisted.

3. ChatWidget: collaboration UI state was cleared before the HTTP POST
   fired. On HTTP failure the response was silently lost. State now clears
   only after HTTP 200, with input text restored on failure for retry.

Bump version to 20260313.2.

Author: fewtarius

Info.plist

@@ -19,9 +19,9 @@
 	<key>CFBundlePackageType</key>
 	<string>APPL</string>
 	<key>CFBundleShortVersionString</key>
-	<string>20260313.1</string>
+	<string>20260313.2</string>
 	<key>CFBundleVersion</key>
-	<string>20260313.1</string>
+	<string>20260313.2</string>
 	<key>LSApplicationCategoryType</key>
 	<string>public.app-category.productivity</string>
 	<key>LSMinimumSystemVersion</key>

Resources/whats-new.json

@@ -1,7 +1,7 @@
 {
   "releases": [
     {
-      "version": "20260313.1",
+      "version": "20260313.2",
       "release_date": "March 13, 2026",
       "highlights": [
         {

Sources/APIFramework/AgentOrchestrator.swift

@@ -2134,12 +2134,14 @@ public class AgentOrchestrator: ObservableObject, IterationController {
                     ])
 
                     /// PERSIST MESSAGE: Add user's response to conversation (PINNED for context preservation)
+                    /// Must use Task @MainActor for Swift 6 concurrency compliance.
                     if let convId = conversationId {
                         Task { @MainActor in
                             if let conversation = self.conversationManager.conversations.first(where: { $0.id == convId }) {
-                                let isDuplicate = conversation.messages.contains(where: {
+                                /// Check MessageBus messages (source of truth) not conversation.messages
+                                let isDuplicate = conversation.messageBus?.messages.contains(where: {
                                     $0.isFromUser && $0.content == userInput
-                                })
+                                }) ?? false
 
                                 if !isDuplicate {
                                     let messageId = conversation.messageBus?.addUserMessage(
@@ -2152,6 +2154,10 @@ public class AgentOrchestrator: ObservableObject, IterationController {
                                         "messageId": .string(messageId?.uuidString ?? "unknown"),
                                         "conversationId": .string(convId.uuidString)
                                     ])
+                                } else {
+                                    self.logger.debug("USER_COLLAB: Skipping duplicate user response persistence", metadata: [
+                                        "toolCallId": .string(toolCallId)
+                                    ])
                                 }
                             }
                         }

Sources/MCPFramework/Tools/UserCollaborationTool.swift

@@ -73,9 +73,31 @@ public class UserCollaborationTool: MCPTool, @unchecked Sendable {
     private let logger = Logger(label: "com.sam.mcp.UserCollaborationTool")
 
     /// Shared state for pending user responses Key: toolCallId (UUID string), Value: PendingResponse.
-    nonisolated(unsafe) private static var pendingResponses: [String: PendingResponse] = [:]
+    /// Thread-safe access: ALL reads/writes MUST go through lockedRead/lockedWrite helpers.
+    nonisolated(unsafe) private static var _pendingResponses: [String: PendingResponse] = [:]
     private static let responseLock = NSLock()
 
+    /// Thread-safe read from pendingResponses.
+    private static func lockedRead<T>(_ body: ([String: PendingResponse]) -> T) -> T {
+        responseLock.lock()
+        defer { responseLock.unlock() }
+        return body(_pendingResponses)
+    }
+
+    /// Thread-safe write to pendingResponses.
+    private static func lockedWrite(_ body: (inout [String: PendingResponse]) -> Void) {
+        responseLock.lock()
+        defer { responseLock.unlock() }
+        body(&_pendingResponses)
+    }
+
+    /// Thread-safe write with return value.
+    private static func lockedWrite<T>(_ body: (inout [String: PendingResponse]) -> T) -> T {
+        responseLock.lock()
+        defer { responseLock.unlock() }
+        return body(&_pendingResponses)
+    }
+
     public init() {}
 
     public func initialize() async throws {
@@ -153,7 +175,7 @@ public class UserCollaborationTool: MCPTool, @unchecked Sendable {
             requestedAt: Date()
         )
 
-        Self.pendingResponses[toolCallId] = pending
+        Self.lockedWrite { $0[toolCallId] = pending }
 
         /// Send SSE event to notify UI NOTE: This will be handled by the streaming handler in APIHandler The event format: { type: "user_input_required", toolCallId, prompt, context?.
         await notifyUIForInput(toolCallId: toolCallId, prompt: prompt, context: userContext, conversationId: context.conversationId)
@@ -170,7 +192,7 @@ public class UserCollaborationTool: MCPTool, @unchecked Sendable {
         )
 
         /// Clean up pending response.
-        Self.pendingResponses.removeValue(forKey: toolCallId)
+        Self.lockedWrite { $0.removeValue(forKey: toolCallId) }
 
         return result
     }
@@ -189,18 +211,25 @@ public class UserCollaborationTool: MCPTool, @unchecked Sendable {
 
             /// Log every 10 polls (every second) to track that we're still waiting
             if pollCount % 10 == 0 {
-                let elapsed = Date().timeIntervalSince(startTime)
+               let elapsed = Date().timeIntervalSince(startTime)
+                let (hasPending, hasResponse) = Self.lockedRead { dict in
+                    (dict[toolCallId] != nil, dict[toolCallId]?.userResponse != nil)
+                }
                 logger.debug("COLLAB_DEBUG: Still waiting for user response", metadata: [
                     "toolCallId": .string(toolCallId),
                     "pollCount": .stringConvertible(pollCount),
                     "elapsedSeconds": .stringConvertible(elapsed),
-                    "hasPending": .stringConvertible(Self.pendingResponses[toolCallId] != nil),
-                    "hasResponse": .stringConvertible(Self.pendingResponses[toolCallId]?.userResponse != nil)
+                    "hasPending": .stringConvertible(hasPending),
+                    "hasResponse": .stringConvertible(hasResponse)
                 ])
             }
-            
-            /// Check if response received.
-            if let pending = Self.pendingResponses[toolCallId], let response = pending.userResponse {
+
+            /// Check if response received (thread-safe read).
+            let maybeResponse = Self.lockedRead { dict -> String? in
+                dict[toolCallId]?.userResponse
+            }
+
+            if let response = maybeResponse {
 
                 let waitTime = Date().timeIntervalSince(startTime)
                 logger.info("USER_RESPONDED: User response received after \(String(format: "%.2f", waitTime))s", metadata: [
@@ -283,37 +312,42 @@ public class UserCollaborationTool: MCPTool, @unchecked Sendable {
     /// Submit user response for a pending tool call (called from API endpoint).
     public static func submitUserResponse(toolCallId: String, userInput: String) -> Bool {
         let logger = Logger(label: "com.sam.mcp.UserCollaborationTool.submitResponse")
+        let pendingCount = lockedRead { $0.count }
+        let hasPending = lockedRead { $0[toolCallId] != nil }
         logger.info("COLLAB_DEBUG: submitUserResponse called", metadata: [
             "toolCallId": .string(toolCallId),
             "userInputLength": .stringConvertible(userInput.count),
-            "pendingCount": .stringConvertible(pendingResponses.count),
-            "hasPending": .stringConvertible(pendingResponses[toolCallId] != nil)
+            "pendingCount": .stringConvertible(pendingCount),
+            "hasPending": .stringConvertible(hasPending)
         ])
 
-        guard var pending = pendingResponses[toolCallId] else {
+        /// Thread-safe update: read, modify, write back under lock.
+        let result: (found: Bool, conversationId: UUID?) = lockedWrite { dict in
+            guard var pending = dict[toolCallId] else {
+                return (false, nil)
+            }
+            pending.userResponse = userInput
+            pending.respondedAt = Date()
+            dict[toolCallId] = pending
+            return (true, pending.conversationId)
+        }
+
+        guard result.found else {
+            let availableIds = lockedRead { Array($0.keys).joined(separator: ", ") }
             logger.error("COLLAB_DEBUG: No pending response found for toolCallId", metadata: [
                 "toolCallId": .string(toolCallId),
-                "availableToolCallIds": .string(Array(pendingResponses.keys).joined(separator: ", "))
+                "availableToolCallIds": .string(availableIds)
             ])
             return false
         }
 
-        logger.debug("COLLAB_DEBUG: Updating pending response", metadata: [
-            "toolCallId": .string(toolCallId),
-            "waitTimeSeconds": .stringConvertible(Date().timeIntervalSince(pending.requestedAt))
-        ])
-
-        pending.userResponse = userInput
-        pending.respondedAt = Date()
-        pendingResponses[toolCallId] = pending
-
-        logger.debug("COLLAB_DEBUG: Pending response updated, posting notification")
+        logger.debug("COLLAB_DEBUG: Pending response updated under lock, posting notification")
 
         /// Notify that user response was received so AgentOrchestrator can emit it as streaming chunk
         ToolNotificationCenter.shared.postUserResponseReceived(
             toolCallId: toolCallId,
             userInput: userInput,
-            conversationId: pending.conversationId
+            conversationId: result.conversationId
         )
 
         logger.info("COLLAB_DEBUG: Notification posted successfully", metadata: [
@@ -325,12 +359,12 @@ public class UserCollaborationTool: MCPTool, @unchecked Sendable {
 
     /// Get pending response info (for debugging/monitoring).
     public static func getPendingResponse(toolCallId: String) -> PendingResponse? {
-        return pendingResponses[toolCallId]
+        return lockedRead { $0[toolCallId] }
     }
 
     /// Get all pending responses (for debugging/monitoring).
     public static func getAllPendingResponses() -> [PendingResponse] {
-        return Array(pendingResponses.values)
+        return lockedRead { Array($0.values) }
     }
 }
 

Sources/UserInterface/Chat/ChatWidget.swift

@@ -3352,13 +3352,10 @@ public struct ChatWidget: View {
 
         logger.info("USER_COLLAB: Submitting user response for collaboration tool call: \(toolCallId)")
 
-        /// Clear message text and reset collaboration state immediately
-        /// The user's response will appear via streaming from AgentOrchestrator
+        /// Clear message text immediately for UX responsiveness, but keep collaboration
+        /// state until HTTP succeeds so we can retry or restore on failure.
+        let savedInput = messageText
         messageText = ""
-        isAwaitingUserInput = false
-        userCollaborationPrompt = ""
-        userCollaborationContext = nil
-        userCollaborationToolCallId = nil
 
         /// Submit response to API endpoint
         /// API will add to MessageBus and AgentOrchestrator will emit as streaming chunk
@@ -3381,14 +3378,29 @@ public struct ChatWidget: View {
                 let (_, response) = try await URLSession.shared.data(for: request)
 
                 guard let httpResponse = response as? HTTPURLResponse, httpResponse.statusCode == 200 else {
-                    logger.error("USER_COLLAB: Failed to submit user response - HTTP error")
+                    logger.error("USER_COLLAB: Failed to submit user response - HTTP \((response as? HTTPURLResponse)?.statusCode ?? -1)")
+                    /// Restore input so user can retry
+                    await MainActor.run {
+                        self.messageText = savedInput
+                    }
                     return
                 }
 
-                logger.debug("User response submitted successfully")
-                /// Streaming will automatically resume after tool unblocks.
+                logger.debug("User response submitted successfully - clearing collaboration state")
+
+                /// Clear collaboration state only after successful submission
+                await MainActor.run {
+                    self.isAwaitingUserInput = false
+                    self.userCollaborationPrompt = ""
+                    self.userCollaborationContext = nil
+                    self.userCollaborationToolCallId = nil
+                }
             } catch {
                 logger.error("Failed to submit user response: \(error)")
+                /// Restore input so user can retry
+                await MainActor.run {
+                    self.messageText = savedInput
+                }
             }
         }
     }