fix(desktop): replace from-segments with force-process on stop (#6356)

Author: beastoin

desktop/Backend-Rust/src/routes/conversations.rs

@@ -157,14 +157,28 @@ async fn get_conversations_count(
 }
 
 /// POST /v1/conversations/from-segments - Create conversation from transcript
-/// Copied from Python create_conversation_from_segments
+/// DEPRECATED: Desktop now uses Python POST /v1/conversations (force-process) instead.
+/// This endpoint will return 410 Gone after 24 hours from deploy.
+/// See: https://github.com/BasedHardware/omi/issues/6355
 async fn create_conversation_from_segments(
     State(state): State<AppState>,
     user: AuthUser,
     Json(request): Json<CreateConversationRequest>,
 ) -> Result<Json<CreateConversationResponse>, (StatusCode, String)> {
-    tracing::info!(
-        "Creating conversation for user {} from {} segments",
+    // Deprecation: return 410 Gone after 24h from deploy
+    if is_from_segments_deprecated() {
+        tracing::warn!(
+            "from-segments endpoint expired for user {} (deprecated)",
+            user.uid,
+        );
+        return Err((
+            StatusCode::GONE,
+            "This endpoint is deprecated. Desktop app now uses Python POST /v1/conversations.".to_string(),
+        ));
+    }
+
+    tracing::warn!(
+        "DEPRECATED: Creating conversation for user {} from {} segments — use Python POST /v1/conversations instead",
         user.uid,
         request.transcript_segments.len()
     );
@@ -1126,6 +1140,21 @@ async fn get_shared_conversation(
     Ok(Json(response))
 }
 
+/// Check if the from-segments endpoint has been deprecated (>24h since DEPRECATION_TIMESTAMP).
+fn is_from_segments_deprecated() -> bool {
+    is_from_segments_deprecated_at(chrono::Utc::now().timestamp())
+}
+
+/// Testable version: check deprecation at a given timestamp.
+fn is_from_segments_deprecated_at(now: i64) -> bool {
+    if let Ok(ts) = std::env::var("DEPRECATION_TIMESTAMP") {
+        if let Ok(deploy_time) = ts.parse::<i64>() {
+            return now - deploy_time > 86400;
+        }
+    }
+    false
+}
+
 pub fn conversations_routes() -> Router<AppState> {
     Router::new()
         .route("/v1/conversations", get(get_conversations))
@@ -1158,3 +1187,41 @@ pub fn conversations_routes() -> Router<AppState> {
         )
 }
 
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_deprecation_not_set() {
+        // No env var → not deprecated
+        std::env::remove_var("DEPRECATION_TIMESTAMP");
+        assert!(!is_from_segments_deprecated_at(1_000_000));
+    }
+
+    #[test]
+    fn test_deprecation_within_24h() {
+        let deploy_time: i64 = 1_700_000_000;
+        std::env::set_var("DEPRECATION_TIMESTAMP", deploy_time.to_string());
+        // 23h59m later → not deprecated
+        let now = deploy_time + 86399;
+        assert!(!is_from_segments_deprecated_at(now));
+        std::env::remove_var("DEPRECATION_TIMESTAMP");
+    }
+
+    #[test]
+    fn test_deprecation_after_24h() {
+        let deploy_time: i64 = 1_700_000_000;
+        std::env::set_var("DEPRECATION_TIMESTAMP", deploy_time.to_string());
+        // 24h + 1s later → deprecated
+        let now = deploy_time + 86401;
+        assert!(is_from_segments_deprecated_at(now));
+        std::env::remove_var("DEPRECATION_TIMESTAMP");
+    }
+
+    #[test]
+    fn test_deprecation_invalid_value() {
+        std::env::set_var("DEPRECATION_TIMESTAMP", "not-a-number");
+        assert!(!is_from_segments_deprecated_at(1_000_000));
+        std::env::remove_var("DEPRECATION_TIMESTAMP");
+    }
+}

desktop/Desktop/Sources/APIClient.swift

@@ -1191,89 +1191,35 @@ struct ServerMemory: Codable, Identifiable {
     }
 }
 
-// MARK: - Create Conversation API
+// MARK: - Force Process Conversation API
 
 extension APIClient {
 
-    /// Request model for creating a conversation from transcript segments
-    struct CreateConversationFromSegmentsRequest: Encodable {
-        let transcriptSegments: [TranscriptSegmentRequest]
-        let source: String
-        let startedAt: String
-        let finishedAt: String
-        let language: String
-        let timezone: String
-        let inputDeviceName: String?
-
-        enum CodingKeys: String, CodingKey {
-            case transcriptSegments = "transcript_segments"
-            case source
-            case startedAt = "started_at"
-            case finishedAt = "finished_at"
-            case language
-            case timezone
-            case inputDeviceName = "input_device_name"
-        }
-    }
-
-    struct TranscriptSegmentRequest: Encodable {
-        let id: String?
-        let text: String
-        let speaker: String
-        let speakerId: Int
-        let isUser: Bool
-        let personId: String?
-        let start: Double
-        let end: Double
-
-        enum CodingKeys: String, CodingKey {
-            case id, text, speaker
-            case speakerId = "speaker_id"
-            case isUser = "is_user"
-            case personId = "person_id"
-            case start, end
-        }
-    }
-
-    struct CreateConversationResponse: Decodable {
-        let id: String
-        let status: String
-        let discarded: Bool
+    /// Response from Python POST /v1/conversations (force-process)
+    struct ForceProcessConversationResponse: Decodable {
+        let conversation: ServerConversation
     }
 
-    /// Creates a conversation from transcript segments
-    /// Endpoint: POST /v1/conversations/from-segments (local backend)
-    /// - Parameters:
-    ///   - segments: Transcript segments to include
-    ///   - startedAt: When the recording started
-    ///   - finishedAt: When the recording finished
-    ///   - source: Source of the conversation (e.g., "desktop", "omi", "bee")
-    ///   - language: Language code for transcription
-    ///   - timezone: User's timezone
-    ///   - inputDeviceName: Name of the input device (microphone or BLE device)
-    func createConversationFromSegments(
-        segments: [TranscriptSegmentRequest],
-        startedAt: Date,
-        finishedAt: Date,
-        source: ConversationSource = .desktop,
-        language: String = "en",
-        timezone: String = "UTC",
-        inputDeviceName: String? = nil
-    ) async throws -> CreateConversationResponse {
-        let formatter = ISO8601DateFormatter()
-        formatter.formatOptions = [.withInternetDateTime, .withFractionalSeconds]
-
-        let request = CreateConversationFromSegmentsRequest(
-            transcriptSegments: segments,
-            source: source.rawValue,
-            startedAt: formatter.string(from: startedAt),
-            finishedAt: formatter.string(from: finishedAt),
-            language: language,
-            timezone: timezone,
-            inputDeviceName: inputDeviceName
-        )
+    /// Force-process the current in-progress conversation on the Python backend.
+    /// Endpoint: POST /v1/conversations (Python backend)
+    /// This is the same endpoint the mobile app uses when stopping phone mic recording.
+    /// The Python backend finds the in-progress conversation via Redis and processes it.
+    /// Returns the processed conversation on success, nil on 404 (already processed).
+    /// Throws on other errors.
+    func forceProcessConversation() async throws -> ServerConversation? {
+        struct EmptyBody: Encodable {}
 
-        return try await post("v1/conversations/from-segments", body: request)
+        do {
+            let response: ForceProcessConversationResponse = try await post(
+                "v1/conversations",
+                body: EmptyBody(),
+                customBaseURL: pythonBackendURL
+            )
+            return response.conversation
+        } catch APIError.httpError(let statusCode) where statusCode == 404 {
+            // 404 = no in-progress conversation found — WS close handler already processed it
+            return nil
+        }
     }
 }
 

desktop/Desktop/Sources/AppState.swift

@@ -26,10 +26,17 @@ enum FinishConversationResult {
 
 @MainActor
 class AppState: ObservableObject {
+  /// Weak reference to the current AppState instance, set on init.
+  /// Used by background services (e.g. TranscriptionRetryService) to check recording state.
+  static weak var current: AppState?
+
   @AppStorage("hasCompletedOnboarding") var hasCompletedOnboarding = false
 
   // Transcription state
   @Published var isTranscribing = false
+  /// Monotonically increasing counter — incremented each time a new recording starts.
+  /// Used to detect if a new recording began during the post-stop force-process delay.
+  private(set) var recordingGeneration: UInt64 = 0
   @Published var isSavingConversation = false
   // currentTranscript is internal-only (not observed by views), so no @Published needed
   private var currentTranscript: String = ""
@@ -195,6 +202,9 @@ class AppState: ObservableObject {
   private var bluetoothStateCancellable: AnyCancellable?
 
   init() {
+    // Register as the current instance so background services can check recording state
+    AppState.current = self
+
     // Load API key from environment or .env file
     loadEnvironment()
 
@@ -1331,6 +1341,7 @@ class AppState: ObservableObject {
       )
 
       isTranscribing = true
+      recordingGeneration &+= 1
       AssistantSettings.shared.transcriptionEnabled = true
       audioSource = effectiveSource
       currentTranscript = ""
@@ -1538,22 +1549,89 @@ class AppState: ObservableObject {
 
   /// Stop real-time transcription
   /// The Python backend handles conversation lifecycle automatically — disconnecting the WebSocket
-  /// triggers conversation processing on the backend side.
+  /// triggers conversation processing on the backend side. We also call force-process to ensure
+  /// the conversation is finalized, preventing the retry service from creating duplicates.
   func stopTranscription() {
+    // Capture session metadata BEFORE clearing state (clearTranscriptionState sets sessionId to nil)
+    let capturedSessionId = currentSessionId
+    let capturedStartTime = recordingStartTime
+    let generationAtStop = recordingGeneration
+
     stopAudioCapture()
     clearTranscriptionState()
 
-    // Backend processes the conversation when WebSocket disconnects.
-    // The local session stays as pendingUpload — the retry service will either:
-    // 1. Find the conversation on the backend (duplicate check) and mark it completed, or
-    // 2. Re-upload if backend processing failed (recovery path).
-    // We don't optimistically mark as completed because that orphans the session if backend fails.
+    // After WS close, the Python backend processes the conversation automatically.
+    // Call force-process to ensure finalization and get the backend conversation ID.
+    // This prevents the retry service from picking up the pendingUpload session.
     Task {
-      try? await Task.sleep(nanoseconds: 5_000_000_000)  // 5s for backend to process
+      try? await Task.sleep(nanoseconds: 3_000_000_000)  // 3s for backend to process after WS close
+
+      // If a new recording started during the delay, skip force-process — it would
+      // finalize the NEW conversation instead of the one we just stopped.
+      // The retry service will reconcile the old session by timestamp matching.
+      guard self.recordingGeneration == generationAtStop else {
+        log("Transcription: New recording started during delay, skipping force-process for session \(capturedSessionId.map(String.init) ?? "nil")")
+        return
+      }
+
+      do {
+        if let conversation = try await APIClient.shared.forceProcessConversation() {
+          // Validate the returned conversation matches the session we just stopped
+          if let sessionId = capturedSessionId, let startTime = capturedStartTime,
+             let convStarted = conversation.startedAt,
+             abs(convStarted.timeIntervalSince(startTime)) < 10,
+             conversation.source == .desktop {
+            try? await TranscriptionStorage.shared.markSessionCompleted(
+              id: sessionId, backendId: conversation.id)
+            log("Transcription: Force-processed conversation \(conversation.id), session \(sessionId) completed")
+          } else if let sessionId = capturedSessionId, let startTime = capturedStartTime {
+            // Force-process returned a different conversation — fall back to reconciliation
+            log("Transcription: Force-processed conversation \(conversation.id) does not match session \(sessionId), reconciling by timestamp")
+            await reconcileSession(sessionId: sessionId, startTime: startTime)
+          }
+        } else {
+          // 404: No in-progress conversation — WS close handler already processed it.
+          // Reconcile by checking if a matching conversation exists on the backend.
+          if let sessionId = capturedSessionId, let startTime = capturedStartTime {
+            await reconcileSession(sessionId: sessionId, startTime: startTime)
+          }
+        }
+      } catch {
+        // Other error — leave session as pendingUpload for retry service to reconcile
+        logError("Transcription: Force-process failed, retry service will reconcile", error: error)
+      }
+
       await loadConversations()
     }
   }
 
+  /// Reconcile a local session by checking if a matching conversation exists on the backend.
+  /// If found, marks the session as completed. Otherwise leaves it as pendingUpload for retry.
+  private func reconcileSession(sessionId: Int64, startTime: Date) async {
+    do {
+      let conversations = try await APIClient.shared.getConversations(
+        limit: 5,
+        includeDiscarded: true,
+        startDate: startTime.addingTimeInterval(-5),
+        endDate: Date().addingTimeInterval(5)
+      )
+      if let match = conversations.first(where: { conv in
+        guard let convStarted = conv.startedAt else { return false }
+        // Must be a desktop conversation with matching start time
+        guard conv.source == .desktop else { return false }
+        return abs(convStarted.timeIntervalSince(startTime)) < 10
+      }) {
+        try await TranscriptionStorage.shared.markSessionCompleted(
+          id: sessionId, backendId: match.id)
+        log("Transcription: Reconciled session \(sessionId) → backend conversation \(match.id)")
+      } else {
+        log("Transcription: No matching backend conversation found for session \(sessionId), leaving for retry")
+      }
+    } catch {
+      logError("Transcription: Reconciliation failed for session \(sessionId)", error: error)
+    }
+  }
+
   /// Finish the current conversation and keep recording for a new one.
   /// Disconnects the WebSocket (triggers backend conversation processing) then reconnects.
   func finishConversation() async -> FinishConversationResult {

desktop/Desktop/Sources/Rewind/Core/TranscriptionStorage.swift

@@ -117,7 +117,8 @@ actor TranscriptionStorage {
         log("TranscriptionStorage: Completed session \(id) (backendId: \(backendId))")
     }
 
-    /// Mark session as failed with error
+    /// Mark session as failed with error.
+    /// No-op if the session is already completed (prevents race with concurrent completion).
     func markSessionFailed(id: Int64, error: String) async throws {
         let db = try await ensureInitialized()
 
@@ -126,6 +127,12 @@ actor TranscriptionStorage {
                 throw TranscriptionStorageError.sessionNotFound
             }
 
+            // Don't regress a completed session back to failed
+            guard record.status != .completed else {
+                log("TranscriptionStorage: Skipping markSessionFailed for already-completed session \(id)")
+                return
+            }
+
             record.status = .failed
             record.lastError = error
             record.updatedAt = Date()
@@ -135,7 +142,8 @@ actor TranscriptionStorage {
         log("TranscriptionStorage: Failed session \(id) (error: \(error))")
     }
 
-    /// Increment retry count for a session
+    /// Increment retry count for a session.
+    /// No-op if the session is already completed (prevents race with concurrent completion).
     func incrementRetryCount(id: Int64) async throws {
         let db = try await ensureInitialized()
 
@@ -144,6 +152,12 @@ actor TranscriptionStorage {
                 throw TranscriptionStorageError.sessionNotFound
             }
 
+            // Don't modify a completed session
+            guard record.status != .completed else {
+                log("TranscriptionStorage: Skipping incrementRetryCount for already-completed session \(id)")
+                return
+            }
+
             record.retryCount += 1
             record.updatedAt = Date()
             try record.update(database)