Make speaker profile merges atomic on current main

Sources/TranscriptedCore/Protocols/SpeakerStore.swift

@@ -34,7 +34,10 @@ public protocol SpeakerStore: Sendable {
     func deleteSpeaker(id: UUID)
 
     /// Merge two speaker profiles (source absorbed into target)
-    func mergeProfiles(sourceId: UUID, into targetId: UUID)
+    func mergeProfiles(sourceId: UUID, into targetId: UUID) throws
+
+    /// Apply an ordered naming mutation batch under one persistence boundary.
+    func performMutationBatch(_ mutations: () throws -> Void) throws
 
     /// Merge profiles that share the same display name
     func mergeProfilesByName()
@@ -76,6 +79,10 @@ public protocol SpeakerStore: Sendable {
 }
 
 public extension SpeakerStore {
+    func performMutationBatch(_ mutations: () throws -> Void) throws {
+        try mutations()
+    }
+
     func mergeDuplicates(protecting protectedIds: Set<UUID>) {
         guard protectedIds.isEmpty else { return }
         mergeDuplicates()

Sources/TranscriptedCore/Speaker/SpeakerDatabase.swift

@@ -9,6 +9,16 @@ import SQLite3
 @available(macOS 14.0, *)
 public final class SpeakerDatabase: @unchecked Sendable {
 
+    struct SQLiteOperationError: Error, LocalizedError {
+        let operation: String
+        let code: Int32
+        let detail: String
+
+        var errorDescription: String? {
+            "\(operation) failed (SQLite \(code)): \(detail)"
+        }
+    }
+
     var db: OpaquePointer?
     // Thread-safety invariant:
     // - Writes happen only during `init` (single-threaded construction) inside
@@ -19,14 +29,24 @@ public final class SpeakerDatabase: @unchecked Sendable {
     //   even though this field is a plain `var Bool`.
     // If a new caller reads this outside the queue, add a queue.sync hop.
     var isDatabaseOpen = false
+    // Set by best-effort legacy write helpers when they are used inside
+    // performMutationBatch. The batch promotes the failure to a thrown error so
+    // its caller can roll back related transcript changes.
+    var pendingMutationError: SQLiteOperationError?
     let dbPath: URL
     let queue = DispatchQueue(label: "com.transcripted.speakerdb", qos: .utility)
+    private let queueSpecificKey = DispatchSpecificKey<UInt8>()
+
+    var isExecutingOnQueue: Bool {
+        DispatchQueue.getSpecific(key: queueSpecificKey) != nil
+    }
 
     /// Public initializer that accepts a custom SQLite path.
     /// Used by tests and by callers that want to store the database outside the default
     /// `CoreStoragePaths.default` layout (e.g. a host app redirecting to its own data dir).
     public init(path: String) {
         dbPath = URL(fileURLWithPath: path)
+        queue.setSpecific(key: queueSpecificKey, value: 1)
         try? FileManager.default.createDirectory(
             at: dbPath.deletingLastPathComponent(),
             withIntermediateDirectories: true
@@ -193,6 +213,99 @@ public final class SpeakerDatabase: @unchecked Sendable {
         }
     }
 
+    /// Transaction boundary for writes whose caller must know whether persistence succeeded.
+    func throwingTransaction(_ block: () throws -> Void) throws {
+        try executeTransactionControl("BEGIN EXCLUSIVE", operation: "begin speaker transaction")
+
+        do {
+            try block()
+        } catch {
+            rollbackAfterFailure()
+            throw error
+        }
+
+        do {
+            try executeTransactionControl("COMMIT", operation: "commit speaker transaction")
+        } catch {
+            rollbackAfterFailure()
+            throw error
+        }
+    }
+
+    public func performMutationBatch(_ mutations: () throws -> Void) throws {
+        if isExecutingOnQueue {
+            pendingMutationError = nil
+            defer { pendingMutationError = nil }
+            try mutations()
+            if let pendingMutationError {
+                throw pendingMutationError
+            }
+            return
+        }
+        try queue.sync {
+            pendingMutationError = nil
+            defer { pendingMutationError = nil }
+            try throwingTransaction {
+                try mutations()
+                if let pendingMutationError {
+                    throw pendingMutationError
+                }
+            }
+        }
+    }
+
+    func recordMutationFailure(operation: String, code: Int32) {
+        guard pendingMutationError == nil else { return }
+        pendingMutationError = SQLiteOperationError(
+            operation: operation,
+            code: code,
+            detail: dbErrorMessage()
+        )
+    }
+
+    func prepareStatement(_ sql: String, operation: String) throws -> OpaquePointer? {
+        var statement: OpaquePointer?
+        let result = sqlite3_prepare_v2(db, sql, -1, &statement, nil)
+        guard result == SQLITE_OK else {
+            sqlite3_finalize(statement)
+            throw sqliteOperationError(operation: operation, code: result)
+        }
+        return statement
+    }
+
+    func requireDone(_ statement: OpaquePointer?, operation: String, expectedChanges: Int32? = nil) throws {
+        let result = sqlite3_step(statement)
+        guard result == SQLITE_DONE else {
+            throw sqliteOperationError(operation: operation, code: result)
+        }
+        if let expectedChanges, sqlite3_changes(db) != expectedChanges {
+            throw SQLiteOperationError(
+                operation: operation,
+                code: SQLITE_NOTFOUND,
+                detail: "expected \(expectedChanges) changed row(s), got \(sqlite3_changes(db))"
+            )
+        }
+    }
+
+    private func executeTransactionControl(_ sql: String, operation: String) throws {
+        let result = sqlite3_exec(db, sql, nil, nil, nil)
+        guard result == SQLITE_OK else {
+            throw sqliteOperationError(operation: operation, code: result)
+        }
+    }
+
+    private func rollbackAfterFailure() {
+        guard sqlite3_get_autocommit(db) == 0 else { return }
+        let result = sqlite3_exec(db, "ROLLBACK", nil, nil, nil)
+        if result != SQLITE_OK {
+            AppLogger.speakers.error("Transaction ROLLBACK failed", ["sqlite_error": dbErrorMessage()])
+        }
+    }
+
+    private func sqliteOperationError(operation: String, code: Int32) -> SQLiteOperationError {
+        SQLiteOperationError(operation: operation, code: code, detail: dbErrorMessage())
+    }
+
     /// Log and return the sqlite3_errmsg for the current database connection
     func dbErrorMessage() -> String {
         if let db = db {
@@ -226,6 +339,9 @@ public final class SpeakerDatabase: @unchecked Sendable {
     /// call-count / last-seen (records the appearance without contaminating the identity).
     /// New-speaker inserts ignore `blendAlpha`. See `SpeakerWritePathPolicy`.
     public func addOrUpdateSpeaker(embedding: [Float], existingId: UUID?, blendAlpha: Float) -> SpeakerProfile {
+        if isExecutingOnQueue {
+            return addOrUpdateSpeakerImpl(embedding: embedding, existingId: existingId, blendAlpha: blendAlpha)
+        }
         return queue.sync {
             addOrUpdateSpeakerImpl(embedding: embedding, existingId: existingId, blendAlpha: blendAlpha)
         }
@@ -234,6 +350,7 @@ public final class SpeakerDatabase: @unchecked Sendable {
     private func addOrUpdateSpeakerImpl(embedding: [Float], existingId: UUID?, blendAlpha: Float) -> SpeakerProfile {
         guard isDatabaseOpen else {
             AppLogger.speakers.error("CRITICAL: addOrUpdateSpeaker returning in-memory-only profile — database not open, speaker will NOT be persisted", ["existingId": existingId?.uuidString ?? "new"])
+            recordMutationFailure(operation: "add or update speaker", code: SQLITE_MISUSE)
             return SpeakerProfile(id: existingId ?? UUID(), displayName: nil, nameSource: nil, embedding: embedding, firstSeen: Date(), lastSeen: Date(), callCount: 1, confidence: 0.5, disputeCount: 0)
         }
 
@@ -279,11 +396,13 @@ public final class SpeakerDatabase: @unchecked Sendable {
                 sqlite3_bind_text(statement, 4, (existingId.uuidString as NSString).utf8String, -1, SQLITE_TRANSIENT)
                 if sqlite3_step(statement) != SQLITE_DONE {
                     AppLogger.speakers.error("Failed to update speaker embedding", ["sqlite_error": dbErrorMessage(), "id": existingId.uuidString])
+                    recordMutationFailure(operation: "update speaker embedding", code: sqlite3_errcode(db))
                 } else {
                     sqlSucceeded = true
                 }
             } else {
                 AppLogger.speakers.error("Failed to prepare update speaker", ["sqlite_error": dbErrorMessage()])
+                recordMutationFailure(operation: "prepare update speaker", code: sqlite3_errcode(db))
             }
             sqlite3_finalize(statement)
 
@@ -341,11 +460,13 @@ public final class SpeakerDatabase: @unchecked Sendable {
             sqlite3_bind_text(statement, 4, (now as NSString).utf8String, -1, SQLITE_TRANSIENT)
             if sqlite3_step(statement) != SQLITE_DONE {
                 AppLogger.speakers.error("Failed to insert new speaker", ["sqlite_error": dbErrorMessage(), "id": newId.uuidString])
+                recordMutationFailure(operation: "insert new speaker", code: sqlite3_errcode(db))
             } else {
                 sqlSucceeded = true
             }
         } else {
             AppLogger.speakers.error("Failed to prepare insert speaker", ["sqlite_error": dbErrorMessage()])
+            recordMutationFailure(operation: "prepare insert speaker", code: sqlite3_errcode(db))
         }
         sqlite3_finalize(statement)
 

Sources/TranscriptedCore/Speaker/SpeakerNamingCoordinator.swift

@@ -128,6 +128,10 @@ extension TranscriptionTaskManager {
                 ) else {
                     return (didFinalize: false, resolvedURL: transcriptURL)
                 }
+                guard let originalTranscriptData = try? Data(contentsOf: resolvedURL) else {
+                    AppLogger.speakers.error("Speaker naming could not snapshot the transcript before update")
+                    return (didFinalize: false, resolvedURL: resolvedURL)
+                }
 
                 var didFinalize = visibleRegularUpdates.isEmpty || TranscriptSaver.updateSpeakerNames(
                     transcriptURL: resolvedURL,
@@ -158,16 +162,34 @@ extension TranscriptionTaskManager {
                     )
                 }
 
+                if didFinalize {
+                    do {
+                        try speakerDB.performMutationBatch {
+                            try Self.applyPlannedNamingMutations(plannedChanges.mutations, speakerDB: speakerDB)
+                            if let deferredReviewPlan {
+                                try Self.applyPlannedNamingMutations(deferredReviewPlan.mutations, speakerDB: speakerDB)
+                            }
+                        }
+                    } catch {
+                        AppLogger.speakers.error("Speaker naming persistence failed", ["error": error.localizedDescription])
+                        do {
+                            try originalTranscriptData.write(to: resolvedURL, options: .atomic)
+                            FileManager.default.restrictToOwnerOnly(atPath: resolvedURL.path)
+                        } catch {
+                            AppLogger.speakers.error("Speaker naming transcript rollback failed", [
+                                "error": error.localizedDescription
+                            ])
+                        }
+                        didFinalize = false
+                    }
+                }
+
                 return (didFinalize: didFinalize, resolvedURL: resolvedURL)
             }
             let didFinalizeTranscript = finalization.didFinalize
             let resolvedURL = finalization.resolvedURL
 
             if didFinalizeTranscript {
-                Self.applyPlannedNamingMutations(plannedChanges.mutations, speakerDB: speakerDB)
-                if let deferredReviewPlan {
-                    Self.applyPlannedNamingMutations(deferredReviewPlan.mutations, speakerDB: speakerDB)
-                }
                 speakerDB.recordMatchOutcomes(Self.plannedMatchOutcomes(
                     for: plannedChanges.resolvedUpdates,
                     clipsBySpeakerId: clipsBySpeakerId,
@@ -647,11 +669,11 @@ extension TranscriptionTaskManager {
     nonisolated private static func applyPlannedNamingMutations(
         _ mutations: [PlannedSpeakerMutation],
         speakerDB: any SpeakerStore
-    ) {
+    ) throws {
         for mutation in mutations {
             switch mutation {
             case .merge(let sourceId, let targetId):
-                speakerDB.mergeProfiles(sourceId: sourceId, into: targetId)
+                try speakerDB.mergeProfiles(sourceId: sourceId, into: targetId)
             case .setDisplayName(let id, let name):
                 speakerDB.setDisplayName(id: id, name: name, source: NameSource.userManual)
             case .restoreProfile(let profile):

Sources/TranscriptedCore/Speaker/SpeakerNegativeExemplarStore.swift

@@ -38,6 +38,10 @@ extension SpeakerDatabase {
     /// comparisons at match time are consistent. No-ops on an empty embedding.
     public func recordNegativeExemplar(profileId: UUID, embedding: [Float]) {
         guard !embedding.isEmpty else { return }
+        if isExecutingOnQueue {
+            recordNegativeExemplarImpl(profileId: profileId, embedding: embedding)
+            return
+        }
         queue.sync { recordNegativeExemplarImpl(profileId: profileId, embedding: embedding) }
     }
 
@@ -46,6 +50,7 @@ extension SpeakerDatabase {
             AppLogger.speakers.error("recordNegativeExemplar skipped — database not open", [
                 "profileId": profileId.uuidString
             ])
+            recordMutationFailure(operation: "record negative exemplar", code: SQLITE_MISUSE)
             return
         }
 
@@ -59,6 +64,7 @@ extension SpeakerDatabase {
         var statement: OpaquePointer?
         guard sqlite3_prepare_v2(db, sql, -1, &statement, nil) == SQLITE_OK else {
             AppLogger.speakers.error("Failed to prepare recordNegativeExemplar", ["sqlite_error": dbErrorMessage()])
+            recordMutationFailure(operation: "prepare record negative exemplar", code: sqlite3_errcode(db))
             return
         }
         let embeddingData = normalized.withUnsafeBufferPointer { Data(buffer: $0) }
@@ -71,6 +77,7 @@ extension SpeakerDatabase {
                 "sqlite_error": dbErrorMessage(),
                 "profileId": profileId.uuidString
             ])
+            recordMutationFailure(operation: "insert negative exemplar", code: sqlite3_errcode(db))
         }
         sqlite3_finalize(statement)
 
@@ -92,13 +99,15 @@ extension SpeakerDatabase {
         var statement: OpaquePointer?
         guard sqlite3_prepare_v2(db, sql, -1, &statement, nil) == SQLITE_OK else {
             AppLogger.speakers.error("Failed to prepare pruneNegativeExemplars", ["sqlite_error": dbErrorMessage()])
+            recordMutationFailure(operation: "prepare prune negative exemplars", code: sqlite3_errcode(db))
             return
         }
         sqlite3_bind_text(statement, 1, (profileId.uuidString as NSString).utf8String, -1, SQLITE_TRANSIENT)
         sqlite3_bind_text(statement, 2, (profileId.uuidString as NSString).utf8String, -1, SQLITE_TRANSIENT)
         sqlite3_bind_int(statement, 3, Int32(Self.negativeExemplarRetentionPerProfile))
         if sqlite3_step(statement) != SQLITE_DONE {
             AppLogger.speakers.error("Failed to prune negative exemplars", ["sqlite_error": dbErrorMessage()])
+            recordMutationFailure(operation: "prune negative exemplars", code: sqlite3_errcode(db))
         }
         sqlite3_finalize(statement)
     }