From 03a2bb02219da6c26c4d77ed91a54941212b536d Mon Sep 17 00:00:00 2001 From: Codex Date: Thu, 16 Jul 2026 23:21:04 -0500 Subject: [PATCH 1/4] Make speaker merges atomic --- .../Protocols/SpeakerStore.swift | 2 +- .../Speaker/SpeakerDatabase.swift | 72 +++++++++ .../Speaker/SpeakerNamingCoordinator.swift | 18 ++- .../Speaker/SpeakerProfileMerger.swift | 115 +++++++++----- .../Speaker/SpeakerProfileProvenance.swift | 81 ++++++---- .../SpeakerPeopleSettingsSection.swift | 7 +- .../SpeakerProfileMergerTests.swift | 146 +++++++++++++++++- .../SpeakerProfileProvenanceTests.swift | 14 +- .../SpeakerTests/SpeakerProvenanceTests.swift | 10 +- .../SpeakerNamingSimulationRunner.swift | 12 +- 10 files changed, 374 insertions(+), 103 deletions(-) diff --git a/Sources/TranscriptedCore/Protocols/SpeakerStore.swift b/Sources/TranscriptedCore/Protocols/SpeakerStore.swift index 24ca77a0c..aaa4053f1 100644 --- a/Sources/TranscriptedCore/Protocols/SpeakerStore.swift +++ b/Sources/TranscriptedCore/Protocols/SpeakerStore.swift @@ -34,7 +34,7 @@ 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 /// Merge profiles that share the same display name func mergeProfilesByName() diff --git a/Sources/TranscriptedCore/Speaker/SpeakerDatabase.swift b/Sources/TranscriptedCore/Speaker/SpeakerDatabase.swift index 2e9043405..789a4fdbe 100644 --- a/Sources/TranscriptedCore/Speaker/SpeakerDatabase.swift +++ b/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 @@ -193,6 +203,68 @@ 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 + } + } + + 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 { diff --git a/Sources/TranscriptedCore/Speaker/SpeakerNamingCoordinator.swift b/Sources/TranscriptedCore/Speaker/SpeakerNamingCoordinator.swift index e1cc51003..344a3629b 100644 --- a/Sources/TranscriptedCore/Speaker/SpeakerNamingCoordinator.swift +++ b/Sources/TranscriptedCore/Speaker/SpeakerNamingCoordinator.swift @@ -164,10 +164,18 @@ extension TranscriptionTaskManager { let resolvedURL = finalization.resolvedURL if didFinalizeTranscript { - Self.applyPlannedNamingMutations(plannedChanges.mutations, speakerDB: speakerDB) - if let deferredReviewPlan { - Self.applyPlannedNamingMutations(deferredReviewPlan.mutations, speakerDB: speakerDB) + do { + 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]) + didFinalizeTranscript = false } + } + + if didFinalizeTranscript { speakerDB.recordMatchOutcomes(Self.plannedMatchOutcomes( for: plannedChanges.resolvedUpdates, clipsBySpeakerId: clipsBySpeakerId, @@ -647,11 +655,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): diff --git a/Sources/TranscriptedCore/Speaker/SpeakerProfileMerger.swift b/Sources/TranscriptedCore/Speaker/SpeakerProfileMerger.swift index 67446434c..98d09ce10 100644 --- a/Sources/TranscriptedCore/Speaker/SpeakerProfileMerger.swift +++ b/Sources/TranscriptedCore/Speaker/SpeakerProfileMerger.swift @@ -152,13 +152,13 @@ extension SpeakerDatabase { /// Merge source profile into target profile. /// Blends embeddings weighted by call count, sums call counts, bumps confidence. /// Transfers name from source if target has none. Deletes source profile. - public func mergeProfiles(sourceId: UUID, into targetId: UUID) { - queue.sync { - mergeProfilesImpl(sourceId: sourceId, into: targetId) + public func mergeProfiles(sourceId: UUID, into targetId: UUID) throws { + try queue.sync { + try mergeProfilesImpl(sourceId: sourceId, into: targetId) } } - func mergeProfilesImpl(sourceId: UUID, into targetId: UUID, kind: String = SpeakerMergeKind.explicit) { + func mergeProfilesImpl(sourceId: UUID, into targetId: UUID, kind: String = SpeakerMergeKind.explicit) throws { guard let source = getSpeakerImpl(id: sourceId), let target = getSpeakerImpl(id: targetId) else { AppLogger.speakers.warning("Merge failed — profile not found", [ @@ -187,76 +187,97 @@ extension SpeakerDatabase { } let normalized = SpeakerVectorMath.l2Normalize(blended) - // Transfer name from source if target has none - if target.displayName == nil, let name = source.displayName { - setDisplayNameImpl(id: targetId, name: name, source: source.nameSource ?? NameSource.userManual) - } - // Update target: blended embedding, summed call count, bumped confidence let isoFormatter = ISO8601DateFormatter() let now = isoFormatter.string(from: Date()) let newCallCount = target.callCount + source.callCount let newConfidence = min(1.0, target.confidence + 0.15) - // Wrap UPDATE + DELETE in a transaction so a crash between them can't orphan data - transaction { + let mergedDisplayName = target.displayName ?? source.displayName + let mergedNameSource = target.displayName == nil && source.displayName != nil + ? (source.nameSource ?? NameSource.userManual) + : target.nameSource + + // Every persistent identity mutation belongs to this one throwing transaction. + try throwingTransaction { // Safety net: snapshot both profiles + re-point provenance BEFORE the blend // overwrites the target and the source row is deleted, so a wrong merge can // be reconstructed into two distinct profiles later. Same transaction → atomic. - recordMergeEventImpl(source: source, target: target, kind: kind) + try recordMergeEventImpl(source: source, target: target, kind: kind) let sql = """ - UPDATE speakers SET embedding = ?, last_seen = ?, call_count = ?, confidence = ? + UPDATE speakers + SET display_name = ?, name_source = ?, embedding = ?, last_seen = ?, call_count = ?, confidence = ? WHERE id = ?; """ - var statement: OpaquePointer? - if sqlite3_prepare_v2(db, sql, -1, &statement, nil) == SQLITE_OK { - let embeddingData = normalized.withUnsafeBufferPointer { Data(buffer: $0) } - sqlite3_bind_blob(statement, 1, (embeddingData as NSData).bytes, Int32(embeddingData.count), SQLITE_TRANSIENT) - sqlite3_bind_text(statement, 2, (now as NSString).utf8String, -1, SQLITE_TRANSIENT) - sqlite3_bind_int(statement, 3, Int32(newCallCount)) - sqlite3_bind_double(statement, 4, newConfidence) - sqlite3_bind_text(statement, 5, (targetId.uuidString as NSString).utf8String, -1, SQLITE_TRANSIENT) - if sqlite3_step(statement) != SQLITE_DONE { - AppLogger.speakers.error("Failed to update target in merge", ["sqlite_error": dbErrorMessage(), "targetId": targetId.uuidString]) - } + let statement = try prepareStatement(sql, operation: "prepare merge target update") + defer { sqlite3_finalize(statement) } + if let mergedDisplayName { + sqlite3_bind_text(statement, 1, (mergedDisplayName as NSString).utf8String, -1, SQLITE_TRANSIENT) } else { - AppLogger.speakers.error("Failed to prepare merge update", ["sqlite_error": dbErrorMessage()]) + sqlite3_bind_null(statement, 1) } - sqlite3_finalize(statement) + if let mergedNameSource { + sqlite3_bind_text(statement, 2, (mergedNameSource as NSString).utf8String, -1, SQLITE_TRANSIENT) + } else { + sqlite3_bind_null(statement, 2) + } + let embeddingData = normalized.withUnsafeBufferPointer { Data(buffer: $0) } + sqlite3_bind_blob(statement, 3, (embeddingData as NSData).bytes, Int32(embeddingData.count), SQLITE_TRANSIENT) + sqlite3_bind_text(statement, 4, (now as NSString).utf8String, -1, SQLITE_TRANSIENT) + sqlite3_bind_int(statement, 5, Int32(newCallCount)) + sqlite3_bind_double(statement, 6, newConfidence) + sqlite3_bind_text(statement, 7, (targetId.uuidString as NSString).utf8String, -1, SQLITE_TRANSIENT) + try requireDone(statement, operation: "step merge target update", expectedChanges: 1) // Exemplars are a rebuildable cache of gated centroids tied to a specific identity/average. // The merge blends a second voice into the target and deletes the source, so drop both // sets: the source's would orphan, and the target's no longer match its changed average. // They re-accumulate from future confident matches. Un-merge restores neither (both start // empty), which degrades to single-average matching — never wrong, just less enriched. - deleteExemplarsImpl(profileId: sourceId) - deleteExemplarsImpl(profileId: targetId) + try deleteIdentityRowsForMerge( + table: "speaker_exemplars", + profileId: sourceId, + operation: "delete source exemplars" + ) + try deleteIdentityRowsForMerge( + table: "speaker_exemplars", + profileId: targetId, + operation: "delete target exemplars" + ) // Negative exemplars are also identity-bound. The source profile is deleted, so its // rejected samples must not remain as orphaned voice embeddings in the database. - deleteNegativeExemplarsImpl(profileId: sourceId) + try deleteIdentityRowsForMerge( + table: "speaker_negative_exemplars", + profileId: sourceId, + operation: "delete source negative exemplars" + ) // Delete source profile let deleteSql = "DELETE FROM speakers WHERE id = ?;" - var deleteStmt: OpaquePointer? - if sqlite3_prepare_v2(db, deleteSql, -1, &deleteStmt, nil) == SQLITE_OK { - sqlite3_bind_text(deleteStmt, 1, (sourceId.uuidString as NSString).utf8String, -1, SQLITE_TRANSIENT) - if sqlite3_step(deleteStmt) != SQLITE_DONE { - AppLogger.speakers.error("Failed to delete source in merge", ["sqlite_error": dbErrorMessage(), "sourceId": sourceId.uuidString]) - } - } else { - AppLogger.speakers.error("Failed to prepare merge delete", ["sqlite_error": dbErrorMessage()]) - } - sqlite3_finalize(deleteStmt) + let deleteStmt = try prepareStatement(deleteSql, operation: "prepare merge source delete") + defer { sqlite3_finalize(deleteStmt) } + sqlite3_bind_text(deleteStmt, 1, (sourceId.uuidString as NSString).utf8String, -1, SQLITE_TRANSIENT) + try requireDone(deleteStmt, operation: "step merge source delete", expectedChanges: 1) } AppLogger.speakers.info("Merged profiles", [ "source": source.displayName ?? "\(sourceId)", - "target": target.displayName ?? "\(targetId)", + "target": mergedDisplayName ?? "\(targetId)", "newCallCount": "\(newCallCount)" ]) } + private func deleteIdentityRowsForMerge(table: String, profileId: UUID, operation: String) throws { + let statement = try prepareStatement( + "DELETE FROM \(table) WHERE profile_id = ?;", + operation: "prepare \(operation)" + ) + defer { sqlite3_finalize(statement) } + sqlite3_bind_text(statement, 1, (profileId.uuidString as NSString).utf8String, -1, SQLITE_TRANSIENT) + try requireDone(statement, operation: "step \(operation)") + } + // MARK: - Duplicate Merging /// Scan all profiles for likely duplicates and merge them. @@ -316,7 +337,12 @@ extension SpeakerDatabase { } // Merge via mergeProfilesImpl (blends embeddings, transfers name, deletes source) - mergeProfilesImpl(sourceId: absorbed.id, into: keeper.id, kind: SpeakerMergeKind.duplicate) + do { + try mergeProfilesImpl(sourceId: absorbed.id, into: keeper.id, kind: SpeakerMergeKind.duplicate) + } catch { + AppLogger.speakers.error("Duplicate speaker merge failed", ["error": error.localizedDescription]) + return + } mergedIds.insert(absorbed.id) mergeCount += 1 @@ -385,7 +411,12 @@ extension SpeakerDatabase { let keeper = sorted[0] for source in sorted.dropFirst() { - mergeProfilesImpl(sourceId: source.id, into: keeper.id, kind: SpeakerMergeKind.byName) + do { + try mergeProfilesImpl(sourceId: source.id, into: keeper.id, kind: SpeakerMergeKind.byName) + } catch { + AppLogger.speakers.error("Same-name speaker merge failed", ["error": error.localizedDescription]) + return + } mergeCount += 1 } diff --git a/Sources/TranscriptedCore/Speaker/SpeakerProfileProvenance.swift b/Sources/TranscriptedCore/Speaker/SpeakerProfileProvenance.swift index 2dd5f225c..5e9de641c 100644 --- a/Sources/TranscriptedCore/Speaker/SpeakerProfileProvenance.swift +++ b/Sources/TranscriptedCore/Speaker/SpeakerProfileProvenance.swift @@ -169,18 +169,19 @@ extension SpeakerDatabase { /// Snapshot both profiles, re-point the source's provenance onto the target, and /// drop a merge marker — all on the open connection so the caller can run it inside /// the same transaction as the embedding blend + source delete. Must be called on - /// `queue`, inside a transaction. Returns the new merge-event id (nil on failure). + /// `queue`, inside a transaction. Returns the new merge-event id. @discardableResult - func recordMergeEventImpl(source: SpeakerProfile, target: SpeakerProfile, kind: String) -> UUID? { - guard isDatabaseOpen else { return nil } + func recordMergeEventImpl(source: SpeakerProfile, target: SpeakerProfile, kind: String) throws -> UUID { + guard isDatabaseOpen else { + throw SQLiteOperationError(operation: "record merge event", code: SQLITE_MISUSE, detail: "database not open") + } guard let sourceSnapshot = Self.encodeProfileSnapshot(source), let targetSnapshot = Self.encodeProfileSnapshot(target) else { - AppLogger.speakers.error("Failed to encode merge snapshot — skipping provenance", ["sourceId": source.id.uuidString]) - return nil + throw SQLiteOperationError(operation: "encode merge snapshots", code: SQLITE_ERROR, detail: "snapshot encoding failed") } let eventId = UUID() - let movedIds = provenanceIdsImpl(forProfileId: source.id) + let movedIds = try provenanceIdsForMergeImpl(forProfileId: source.id) let movedIdsJSON = (try? JSONEncoder().encode(movedIds.map { $0.uuidString })) .flatMap { String(data: $0, encoding: .utf8) } ?? "[]" let now = ISO8601DateFormatter().string(from: Date()) @@ -190,12 +191,8 @@ extension SpeakerDatabase { (id, target_id, source_id, kind, source_snapshot, target_snapshot, moved_provenance_ids, merged_at, undone_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, NULL); """ - var statement: OpaquePointer? - guard sqlite3_prepare_v2(db, insertEvent, -1, &statement, nil) == SQLITE_OK else { - AppLogger.speakers.error("Failed to prepare merge-event insert", ["sqlite_error": dbErrorMessage()]) - sqlite3_finalize(statement) - return nil - } + let statement = try prepareStatement(insertEvent, operation: "prepare merge-event insert") + defer { sqlite3_finalize(statement) } sqlite3_bind_text(statement, 1, (eventId.uuidString as NSString).utf8String, -1, SQLITE_TRANSIENT) sqlite3_bind_text(statement, 2, (target.id.uuidString as NSString).utf8String, -1, SQLITE_TRANSIENT) sqlite3_bind_text(statement, 3, (source.id.uuidString as NSString).utf8String, -1, SQLITE_TRANSIENT) @@ -204,42 +201,58 @@ extension SpeakerDatabase { sqlite3_bind_text(statement, 6, (targetSnapshot as NSString).utf8String, -1, SQLITE_TRANSIENT) sqlite3_bind_text(statement, 7, (movedIdsJSON as NSString).utf8String, -1, SQLITE_TRANSIENT) sqlite3_bind_text(statement, 8, (now as NSString).utf8String, -1, SQLITE_TRANSIENT) - if sqlite3_step(statement) != SQLITE_DONE { - AppLogger.speakers.error("Failed to insert merge event", ["sqlite_error": dbErrorMessage()]) - sqlite3_finalize(statement) - return nil - } - sqlite3_finalize(statement) + try requireDone(statement, operation: "step merge-event insert", expectedChanges: 1) // Re-point the absorbed profile's provenance onto the keeper. - execBind( + let rePointStmt = try prepareStatement( "UPDATE speaker_provenance SET profile_id = ? WHERE profile_id = ?;", - [target.id.uuidString, source.id.uuidString], - label: "re-point provenance" + operation: "prepare re-point merge provenance" ) + defer { sqlite3_finalize(rePointStmt) } + sqlite3_bind_text(rePointStmt, 1, (target.id.uuidString as NSString).utf8String, -1, SQLITE_TRANSIENT) + sqlite3_bind_text(rePointStmt, 2, (source.id.uuidString as NSString).utf8String, -1, SQLITE_TRANSIENT) + try requireDone(rePointStmt, operation: "step re-point merge provenance") // Marker row so the keeper's history shows what fused in. let marker = """ INSERT INTO speaker_provenance (id, profile_id, kind, embedding, source_profile_id, merge_event_id, recorded_at) VALUES (?, ?, ?, NULL, ?, ?, ?); """ - var markerStmt: OpaquePointer? - if sqlite3_prepare_v2(db, marker, -1, &markerStmt, nil) == SQLITE_OK { - sqlite3_bind_text(markerStmt, 1, (UUID().uuidString as NSString).utf8String, -1, SQLITE_TRANSIENT) - sqlite3_bind_text(markerStmt, 2, (target.id.uuidString as NSString).utf8String, -1, SQLITE_TRANSIENT) - sqlite3_bind_text(markerStmt, 3, (SpeakerProvenanceKind.merge as NSString).utf8String, -1, SQLITE_TRANSIENT) - sqlite3_bind_text(markerStmt, 4, (source.id.uuidString as NSString).utf8String, -1, SQLITE_TRANSIENT) - sqlite3_bind_text(markerStmt, 5, (eventId.uuidString as NSString).utf8String, -1, SQLITE_TRANSIENT) - sqlite3_bind_text(markerStmt, 6, (now as NSString).utf8String, -1, SQLITE_TRANSIENT) - if sqlite3_step(markerStmt) != SQLITE_DONE { - AppLogger.speakers.error("Failed to insert merge marker", ["sqlite_error": dbErrorMessage()]) - } - } - sqlite3_finalize(markerStmt) + let markerStmt = try prepareStatement(marker, operation: "prepare merge marker insert") + defer { sqlite3_finalize(markerStmt) } + sqlite3_bind_text(markerStmt, 1, (UUID().uuidString as NSString).utf8String, -1, SQLITE_TRANSIENT) + sqlite3_bind_text(markerStmt, 2, (target.id.uuidString as NSString).utf8String, -1, SQLITE_TRANSIENT) + sqlite3_bind_text(markerStmt, 3, (SpeakerProvenanceKind.merge as NSString).utf8String, -1, SQLITE_TRANSIENT) + sqlite3_bind_text(markerStmt, 4, (source.id.uuidString as NSString).utf8String, -1, SQLITE_TRANSIENT) + sqlite3_bind_text(markerStmt, 5, (eventId.uuidString as NSString).utf8String, -1, SQLITE_TRANSIENT) + sqlite3_bind_text(markerStmt, 6, (now as NSString).utf8String, -1, SQLITE_TRANSIENT) + try requireDone(markerStmt, operation: "step merge marker insert", expectedChanges: 1) return eventId } + private func provenanceIdsForMergeImpl(forProfileId profileId: UUID) throws -> [UUID] { + let statement = try prepareStatement( + "SELECT id FROM speaker_provenance WHERE profile_id = ?;", + operation: "prepare merge provenance lookup" + ) + defer { sqlite3_finalize(statement) } + sqlite3_bind_text(statement, 1, (profileId.uuidString as NSString).utf8String, -1, SQLITE_TRANSIENT) + + var ids: [UUID] = [] + while true { + let result = sqlite3_step(statement) + if result == SQLITE_DONE { return ids } + guard result == SQLITE_ROW else { + throw SQLiteOperationError(operation: "step merge provenance lookup", code: result, detail: dbErrorMessage()) + } + if let value = sqlite3_column_text(statement, 0).map(String.init(cString:)), + let id = UUID(uuidString: value) { + ids.append(id) + } + } + } + // MARK: - Public queries /// Recent merges that can still be undone, newest first. diff --git a/Sources/UI/Settings/SpeakerPeopleSettingsSection.swift b/Sources/UI/Settings/SpeakerPeopleSettingsSection.swift index fd8893c20..e0b462151 100644 --- a/Sources/UI/Settings/SpeakerPeopleSettingsSection.swift +++ b/Sources/UI/Settings/SpeakerPeopleSettingsSection.swift @@ -329,7 +329,12 @@ final class SpeakerPeopleSettingsViewModel: ObservableObject { preferredClipsDirectory: preferredClipsDirectory, legacyClipsDirectory: legacyClipsDirectory ) - speakerDatabase.mergeProfiles(sourceId: sourceId, into: targetId) + do { + try speakerDatabase.mergeProfiles(sourceId: sourceId, into: targetId) + } catch { + AppLogger.speakers.error("Manual speaker merge failed", ["error": error.localizedDescription]) + return + } Self.deleteClips( for: sourceId, preferredClipsDirectory: preferredClipsDirectory, diff --git a/Tests/TranscriptedCoreTests/SpeakerTests/SpeakerProfileMergerTests.swift b/Tests/TranscriptedCoreTests/SpeakerTests/SpeakerProfileMergerTests.swift index 38ad432a7..5e7d34be2 100644 --- a/Tests/TranscriptedCoreTests/SpeakerTests/SpeakerProfileMergerTests.swift +++ b/Tests/TranscriptedCoreTests/SpeakerTests/SpeakerProfileMergerTests.swift @@ -1,4 +1,5 @@ import XCTest +import SQLite3 @testable import TranscriptedCore @available(macOS 14.0, *) @@ -134,7 +135,7 @@ final class SpeakerProfileMergerTests: XCTestCase { let sourceBefore = try XCTUnwrap(database.getSpeaker(id: source.id)) database.recordNegativeExemplar(profileId: source.id, embedding: [Float](repeating: 0.12, count: 256)) - database.mergeProfiles(sourceId: source.id, into: target.id) + try database.mergeProfiles(sourceId: source.id, into: target.id) XCTAssertNil(database.getSpeaker(id: source.id), "source profile is deleted after merge") XCTAssertTrue( @@ -148,6 +149,76 @@ final class SpeakerProfileMergerTests: XCTestCase { XCTAssertEqual(merged.confidence, min(1.0, targetBefore.confidence + 0.15), accuracy: 0.0001, "confidence bumps") } + func testMergeProfilesRollsBackAndThrowsWhenTargetUpdateCannotBePrepared() throws { + let (source, target) = makeNamedSourceAndUnnamedTarget() + let result = database.queue.sync { + sqlite3_set_authorizer(database.db, { _, action, tableName, _, _, _ in + guard action == SQLITE_UPDATE, + let tableName, + String(cString: tableName) == "speakers" else { + return SQLITE_OK + } + return SQLITE_DENY + }, nil) + } + XCTAssertEqual(result, SQLITE_OK) + defer { + _ = database.queue.sync { sqlite3_set_authorizer(database.db, nil, nil) } + } + + assertMergeThrows(sourceId: source.id, targetId: target.id, operation: "prepare merge target update") + try assertMergeDidNotMutate(source: source, target: target) + } + + func testMergeProfilesRollsBackAndThrowsWhenTargetUpdateStepFails() throws { + let (source, target) = makeNamedSourceAndUnnamedTarget() + try executeSQL(""" + CREATE TRIGGER fail_merge_target_update + BEFORE UPDATE ON speakers + WHEN OLD.id = '\(target.id.uuidString)' AND NEW.call_count > OLD.call_count + BEGIN + SELECT RAISE(ABORT, 'forced merge update failure'); + END; + """) + + assertMergeThrows(sourceId: source.id, targetId: target.id, operation: "step merge target update") + try assertMergeDidNotMutate(source: source, target: target) + } + + func testMergeProfilesThrowsWhenTransactionCannotBegin() throws { + let (source, target) = makeNamedSourceAndUnnamedTarget() + + let thrown = database.queue.sync { () -> Error? in + guard sqlite3_exec(database.db, "BEGIN EXCLUSIVE", nil, nil, nil) == SQLITE_OK else { + return TestFailure.couldNotStartTransaction + } + defer { sqlite3_exec(database.db, "ROLLBACK", nil, nil, nil) } + do { + try database.mergeProfilesImpl(sourceId: source.id, into: target.id) + return nil + } catch { + return error + } + } + + assertSQLiteOperation(thrown, equals: "begin speaker transaction") + try assertMergeDidNotMutate(source: source, target: target) + } + + func testMergeProfilesRollsBackAndThrowsWhenCommitFails() throws { + let (source, target) = makeNamedSourceAndUnnamedTarget() + try executeSQL("PRAGMA foreign_keys = ON;") + try executeSQL(""" + CREATE TABLE merge_commit_guard ( + source_id TEXT NOT NULL REFERENCES speakers(id) DEFERRABLE INITIALLY DEFERRED + ); + INSERT INTO merge_commit_guard (source_id) VALUES ('\(source.id.uuidString)'); + """) + + assertMergeThrows(sourceId: source.id, targetId: target.id, operation: "commit speaker transaction") + try assertMergeDidNotMutate(source: source, target: target) + } + func testMergeProfilesByNameCollapsesSameNameProfiles() throws { // Distinct embeddings on purpose: same-name merge ignores similarity, so // four "Jenny Wen" profiles must still collapse into one. @@ -199,6 +270,77 @@ final class SpeakerProfileMergerTests: XCTestCase { XCTAssertNil(database.getSpeaker(id: profileId)) XCTAssertTrue(database.negativeExemplars(profileId: profileId).isEmpty) } + + private func makeNamedSourceAndUnnamedTarget() -> (source: SpeakerProfile, target: SpeakerProfile) { + let target = database.addOrUpdateSpeaker( + embedding: [Float](repeating: 0.30, count: 256), + existingId: nil + ) + let source = database.addOrUpdateSpeaker( + embedding: [Float](repeating: 0.31, count: 256), + existingId: nil + ) + database.setDisplayName(id: source.id, name: "Jenny Wen", source: NameSource.userManual) + database.recordNegativeExemplar( + profileId: source.id, + embedding: [Float](repeating: 0.12, count: 256) + ) + return (source, target) + } + + private func assertMergeThrows(sourceId: UUID, targetId: UUID, operation: String) { + XCTAssertThrowsError(try database.mergeProfiles(sourceId: sourceId, into: targetId)) { + self.assertSQLiteOperation($0, equals: operation) + } + } + + private func assertSQLiteOperation( + _ error: Error?, + equals operation: String, + file: StaticString = #filePath, + line: UInt = #line + ) { + guard let sqliteError = error as? SpeakerDatabase.SQLiteOperationError else { + XCTFail("Expected SQLiteOperationError, got \(String(describing: error))", file: file, line: line) + return + } + XCTAssertEqual(sqliteError.operation, operation, file: file, line: line) + } + + private func assertMergeDidNotMutate( + source: SpeakerProfile, + target: SpeakerProfile, + file: StaticString = #filePath, + line: UInt = #line + ) throws { + let sourceAfter = try XCTUnwrap(database.getSpeaker(id: source.id), file: file, line: line) + let targetAfter = try XCTUnwrap(database.getSpeaker(id: target.id), file: file, line: line) + XCTAssertEqual(sourceAfter.displayName, "Jenny Wen", file: file, line: line) + XCTAssertEqual(sourceAfter.callCount, source.callCount, file: file, line: line) + XCTAssertNil(targetAfter.displayName, file: file, line: line) + XCTAssertEqual(targetAfter.callCount, target.callCount, file: file, line: line) + XCTAssertFalse(database.negativeExemplars(profileId: source.id).isEmpty, file: file, line: line) + XCTAssertTrue(database.recentUndoableMerges().isEmpty, file: file, line: line) + } + + private func executeSQL(_ sql: String) throws { + try database.queue.sync { + var message: UnsafeMutablePointer? + let result = sqlite3_exec(database.db, sql, nil, nil, &message) + defer { sqlite3_free(message) } + guard result == SQLITE_OK else { + throw NSError( + domain: "SpeakerProfileMergerTests.SQLite", + code: Int(result), + userInfo: [NSLocalizedDescriptionKey: message.map { String(cString: $0) } ?? "unknown SQLite error"] + ) + } + } + } + + private enum TestFailure: Error { + case couldNotStartTransaction + } } @available(macOS 14.0, *) @@ -226,7 +368,7 @@ private final class DefaultMergeFallbackSpeakerStore: SpeakerStore, @unchecked S func setDisplayName(id _: UUID, name _: String, source _: String) {} func restoreProfile(_: SpeakerProfile) {} func deleteSpeaker(id _: UUID) {} - func mergeProfiles(sourceId _: UUID, into _: UUID) {} + func mergeProfiles(sourceId _: UUID, into _: UUID) throws {} func mergeProfilesByName() {} func mergeDuplicates() { diff --git a/Tests/TranscriptedCoreTests/SpeakerTests/SpeakerProfileProvenanceTests.swift b/Tests/TranscriptedCoreTests/SpeakerTests/SpeakerProfileProvenanceTests.swift index 243410985..eaa62a7bd 100644 --- a/Tests/TranscriptedCoreTests/SpeakerTests/SpeakerProfileProvenanceTests.swift +++ b/Tests/TranscriptedCoreTests/SpeakerTests/SpeakerProfileProvenanceTests.swift @@ -119,9 +119,9 @@ final class SpeakerProfileProvenanceTests: XCTestCase { database.setDisplayName(id: keeper.id, name: "Keeper", source: NameSource.userManual) database.setDisplayName(id: third.id, name: "Charlie", source: NameSource.userManual) - database.mergeProfiles(sourceId: first.id, into: keeper.id) - database.mergeProfiles(sourceId: second.id, into: keeper.id) - database.mergeProfiles(sourceId: third.id, into: keeper.id) + try database.mergeProfiles(sourceId: first.id, into: keeper.id) + try database.mergeProfiles(sourceId: second.id, into: keeper.id) + try database.mergeProfiles(sourceId: third.id, into: keeper.id) let all = database.recentUndoableMerges(limit: 25) XCTAssertEqual(all.count, 3) @@ -141,7 +141,7 @@ final class SpeakerProfileProvenanceTests: XCTestCase { func testRecentUndoableMergesExcludesUndoneMerges() throws { let source = database.addOrUpdateSpeaker(embedding: embedding(axis: 5), existingId: nil) let target = database.addOrUpdateSpeaker(embedding: embedding(axis: 6), existingId: nil) - database.mergeProfiles(sourceId: source.id, into: target.id) + try database.mergeProfiles(sourceId: source.id, into: target.id) XCTAssertTrue(database.recentUndoableMerges(limit: 25).contains { $0.sourceId == source.id }) XCTAssertTrue(database.unmergeMostRecent(forTargetId: target.id)) @@ -151,7 +151,7 @@ final class SpeakerProfileProvenanceTests: XCTestCase { func testRecentUndoableMergesWithNonPositiveLimitReturnsEmpty() throws { let source = database.addOrUpdateSpeaker(embedding: embedding(axis: 7), existingId: nil) let target = database.addOrUpdateSpeaker(embedding: embedding(axis: 8), existingId: nil) - database.mergeProfiles(sourceId: source.id, into: target.id) + try database.mergeProfiles(sourceId: source.id, into: target.id) XCTAssertTrue(database.recentUndoableMerges(limit: 0).isEmpty) XCTAssertTrue(database.recentUndoableMerges(limit: -5).isEmpty) @@ -189,7 +189,7 @@ final class SpeakerProfileProvenanceTests: XCTestCase { func testUnmergeFailsWhenCalledTwiceOnSameEvent() throws { let source = database.addOrUpdateSpeaker(embedding: embedding(axis: 50), existingId: nil) let target = database.addOrUpdateSpeaker(embedding: embedding(axis: 51), existingId: nil) - database.mergeProfiles(sourceId: source.id, into: target.id) + try database.mergeProfiles(sourceId: source.id, into: target.id) let record = try XCTUnwrap(database.undoableMerge(forTargetId: target.id)) XCTAssertTrue(database.unmerge(mergeId: record.id)) @@ -263,7 +263,7 @@ final class SpeakerProfileProvenanceTests: XCTestCase { let source = db1!.addOrUpdateSpeaker(embedding: embedding(axis: 90), existingId: nil) let target = db1!.addOrUpdateSpeaker(embedding: embedding(axis: 91), existingId: nil) db1!.setDisplayName(id: source.id, name: "Pat", source: NameSource.userManual) - db1!.mergeProfiles(sourceId: source.id, into: target.id) + try db1!.mergeProfiles(sourceId: source.id, into: target.id) let contributionCountBefore = db1!.contributions(forProfileId: target.id).count db1 = nil // close the connection before reopening the same file diff --git a/Tests/TranscriptedCoreTests/SpeakerTests/SpeakerProvenanceTests.swift b/Tests/TranscriptedCoreTests/SpeakerTests/SpeakerProvenanceTests.swift index 52addc28b..d7a3a880b 100644 --- a/Tests/TranscriptedCoreTests/SpeakerTests/SpeakerProvenanceTests.swift +++ b/Tests/TranscriptedCoreTests/SpeakerTests/SpeakerProvenanceTests.swift @@ -53,7 +53,7 @@ final class SpeakerProvenanceTests: XCTestCase { // Profiles are genuinely distinct going in. XCTAssertLessThan(cosine(preSource, preTarget), 0.5) - database.mergeProfiles(sourceId: source.id, into: target.id) + try database.mergeProfiles(sourceId: source.id, into: target.id) // Merge fused them: source is gone, only the keeper remains. XCTAssertNil(database.getSpeaker(id: source.id)) @@ -111,7 +111,7 @@ final class SpeakerProvenanceTests: XCTestCase { let source = database.addOrUpdateSpeaker(embedding: embedding(axis: 5), existingId: nil) let target = database.addOrUpdateSpeaker(embedding: embedding(axis: 6), existingId: nil) - database.mergeProfiles(sourceId: source.id, into: target.id) + try database.mergeProfiles(sourceId: source.id, into: target.id) let afterMerge = database.contributions(forProfileId: target.id) XCTAssertTrue( @@ -132,9 +132,9 @@ final class SpeakerProvenanceTests: XCTestCase { let second = database.addOrUpdateSpeaker(embedding: embedding(axis: 11), existingId: nil) let keeper = database.addOrUpdateSpeaker(embedding: embedding(axis: 12), existingId: nil) - database.mergeProfiles(sourceId: first.id, into: keeper.id) + try database.mergeProfiles(sourceId: first.id, into: keeper.id) let firstEvent = try XCTUnwrap(database.undoableMerge(forTargetId: keeper.id)) - database.mergeProfiles(sourceId: second.id, into: keeper.id) + try database.mergeProfiles(sourceId: second.id, into: keeper.id) // The older merge can't be undone while a newer merge still sits on top. XCTAssertFalse(database.unmerge(mergeId: firstEvent.id)) @@ -163,7 +163,7 @@ final class SpeakerProvenanceTests: XCTestCase { let source = database.addOrUpdateSpeaker(embedding: embedding(axis: 30), existingId: nil) let target = database.addOrUpdateSpeaker(embedding: embedding(axis: 31), existingId: nil) - database.mergeProfiles(sourceId: source.id, into: target.id) + try database.mergeProfiles(sourceId: source.id, into: target.id) // The keeper picks up another recording AFTER the merge. _ = database.addOrUpdateSpeaker(embedding: embedding(axis: 31), existingId: target.id) diff --git a/Tests/TranscriptedCoreTests/SpeakerTests/Support/SpeakerNamingSimulationRunner.swift b/Tests/TranscriptedCoreTests/SpeakerTests/Support/SpeakerNamingSimulationRunner.swift index 5fea1d77d..3ddb6d8e6 100644 --- a/Tests/TranscriptedCoreTests/SpeakerTests/Support/SpeakerNamingSimulationRunner.swift +++ b/Tests/TranscriptedCoreTests/SpeakerTests/Support/SpeakerNamingSimulationRunner.swift @@ -985,7 +985,7 @@ final class SpeakerNamingSimulationRunner { ) else { throw SimulationError.updateFailed(transcriptURL.lastPathComponent) } - applyDatabaseUpdates(regular, classification: classification, speakerDB: speakerDB) + try applyDatabaseUpdates(regular, classification: classification, speakerDB: speakerDB) } if !collapsed.isEmpty { @@ -1115,7 +1115,7 @@ final class SpeakerNamingSimulationRunner { _ updates: [SpeakerNameUpdate], classification: Classification, speakerDB: SpeakerDatabase - ) { + ) throws { for update in updates { let resolvedId = update.resolvedPersistentSpeakerId ?? update.persistentSpeakerId let context = classification.context( @@ -1126,14 +1126,14 @@ final class SpeakerNamingSimulationRunner { switch update.action { case .named: if resolvedId != update.persistentSpeakerId { - speakerDB.mergeProfiles(sourceId: update.persistentSpeakerId, into: resolvedId) + try speakerDB.mergeProfiles(sourceId: update.persistentSpeakerId, into: resolvedId) } speakerDB.setDisplayName(id: resolvedId, name: update.newName, source: NameSource.userManual) speakerDB.resetDisputeCount(id: resolvedId) case .confirmed: if resolvedId != update.persistentSpeakerId { - speakerDB.mergeProfiles(sourceId: update.persistentSpeakerId, into: resolvedId) + try speakerDB.mergeProfiles(sourceId: update.persistentSpeakerId, into: resolvedId) } speakerDB.setDisplayName(id: resolvedId, name: update.newName, source: NameSource.userManual) speakerDB.resetDisputeCount(id: resolvedId) @@ -1155,7 +1155,7 @@ final class SpeakerNamingSimulationRunner { speakerDB.resetDisputeCount(id: resolvedId) case .merged(let targetProfileId): - speakerDB.mergeProfiles(sourceId: update.persistentSpeakerId, into: targetProfileId) + try speakerDB.mergeProfiles(sourceId: update.persistentSpeakerId, into: targetProfileId) speakerDB.resetDisputeCount(id: targetProfileId) case .collapsedToMe, .discardedFromDatabase: @@ -1230,7 +1230,7 @@ final class SpeakerNamingSimulationRunner { let target = exactNamedTarget(named: targetName, excluding: sourceId, speakerDB: state.speakerDB) else { return PostActionOutcome(checks: 1, successes: 0, notes: ["missing merge target \(targetName)"]) } - state.speakerDB.mergeProfiles(sourceId: sourceId, into: target.id) + try state.speakerDB.mergeProfiles(sourceId: sourceId, into: target.id) TranscriptSaver.retroactivelyMergeSpeaker( sourceDbId: sourceId, targetDbId: target.id, From 77d690bbd92df47773cb20050eb688c55c88bd5f Mon Sep 17 00:00:00 2001 From: Codex Date: Fri, 17 Jul 2026 06:21:10 -0500 Subject: [PATCH 2/4] Keep speaker clip merge rollback-safe --- .../Protocols/SpeakerStore.swift | 7 ++ .../Speaker/SpeakerDatabase.swift | 19 +++ .../Speaker/SpeakerNamingCoordinator.swift | 38 ++++-- .../SpeakerNegativeExemplarStore.swift | 4 + .../Speaker/SpeakerProfileMerger.swift | 74 ++++++++--- .../SpeakerPeopleSettingsSection.swift | 32 +++-- ...kerProfileMergeSideEffectCoordinator.swift | 11 ++ ...ofileMergeSideEffectCoordinatorTests.swift | 53 ++++++++ .../SpeakerNamingCoordinatorTests.swift | 117 ++++++++++++++++++ .../SpeakerProfileMergerTests.swift | 78 ++++++++++++ scripts/entrypoints/run-tests.sh | 1 + 11 files changed, 396 insertions(+), 38 deletions(-) create mode 100644 Sources/UI/Settings/SpeakerProfileMergeSideEffectCoordinator.swift create mode 100644 Tests/SpeakerProfileMergeSideEffectCoordinatorTests.swift diff --git a/Sources/TranscriptedCore/Protocols/SpeakerStore.swift b/Sources/TranscriptedCore/Protocols/SpeakerStore.swift index aaa4053f1..8a269c5fd 100644 --- a/Sources/TranscriptedCore/Protocols/SpeakerStore.swift +++ b/Sources/TranscriptedCore/Protocols/SpeakerStore.swift @@ -36,6 +36,9 @@ public protocol SpeakerStore: Sendable { /// Merge two speaker profiles (source absorbed into target) 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) { guard protectedIds.isEmpty else { return } mergeDuplicates() diff --git a/Sources/TranscriptedCore/Speaker/SpeakerDatabase.swift b/Sources/TranscriptedCore/Speaker/SpeakerDatabase.swift index 789a4fdbe..6ed01a8e7 100644 --- a/Sources/TranscriptedCore/Speaker/SpeakerDatabase.swift +++ b/Sources/TranscriptedCore/Speaker/SpeakerDatabase.swift @@ -31,12 +31,18 @@ public final class SpeakerDatabase: @unchecked Sendable { var isDatabaseOpen = false let dbPath: URL let queue = DispatchQueue(label: "com.transcripted.speakerdb", qos: .utility) + private let queueSpecificKey = DispatchSpecificKey() + + 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 @@ -222,6 +228,16 @@ public final class SpeakerDatabase: @unchecked Sendable { } } + public func performMutationBatch(_ mutations: () throws -> Void) throws { + if isExecutingOnQueue { + try mutations() + return + } + try queue.sync { + try throwingTransaction(mutations) + } + } + func prepareStatement(_ sql: String, operation: String) throws -> OpaquePointer? { var statement: OpaquePointer? let result = sqlite3_prepare_v2(db, sql, -1, &statement, nil) @@ -298,6 +314,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) } diff --git a/Sources/TranscriptedCore/Speaker/SpeakerNamingCoordinator.swift b/Sources/TranscriptedCore/Speaker/SpeakerNamingCoordinator.swift index 344a3629b..77ed455b4 100644 --- a/Sources/TranscriptedCore/Speaker/SpeakerNamingCoordinator.swift +++ b/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,23 +162,33 @@ 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 { - do { - 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]) - didFinalizeTranscript = false - } - } - if didFinalizeTranscript { speakerDB.recordMatchOutcomes(Self.plannedMatchOutcomes( for: plannedChanges.resolvedUpdates, diff --git a/Sources/TranscriptedCore/Speaker/SpeakerNegativeExemplarStore.swift b/Sources/TranscriptedCore/Speaker/SpeakerNegativeExemplarStore.swift index b398eb2c4..8c2dc9635 100644 --- a/Sources/TranscriptedCore/Speaker/SpeakerNegativeExemplarStore.swift +++ b/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) } } diff --git a/Sources/TranscriptedCore/Speaker/SpeakerProfileMerger.swift b/Sources/TranscriptedCore/Speaker/SpeakerProfileMerger.swift index 98d09ce10..7f9e004b8 100644 --- a/Sources/TranscriptedCore/Speaker/SpeakerProfileMerger.swift +++ b/Sources/TranscriptedCore/Speaker/SpeakerProfileMerger.swift @@ -6,18 +6,40 @@ import SQLite3 @available(macOS 14.0, *) extension SpeakerDatabase { + enum ProfileMergeError: Error, LocalizedError { + case profileNotFound(sourceId: UUID, targetId: UUID) + case invalidEmbeddingState(sourceId: UUID, targetId: UUID) + + var errorDescription: String? { + switch self { + case .profileNotFound: + return "Speaker merge could not find both profiles" + case .invalidEmbeddingState: + return "Speaker merge found incompatible profile embeddings" + } + } + } + /// Set the display name for a speaker with provenance tracking /// - Parameters: /// - id: Speaker profile UUID /// - name: Display name to set /// - source: Where the name came from (NameSource.userManual) public func setDisplayName(id: UUID, name: String, source: String = NameSource.userManual) { + if isExecutingOnQueue { + setDisplayNameImpl(id: id, name: name, source: source) + return + } queue.sync { setDisplayNameImpl(id: id, name: name, source: source) } } public func restoreProfile(_ profile: SpeakerProfile) { + if isExecutingOnQueue { + restoreProfileImpl(profile) + return + } queue.sync { restoreProfileImpl(profile) } @@ -95,7 +117,7 @@ extension SpeakerDatabase { /// Increment the dispute count for a speaker (inference disagreed with DB name) public func incrementDisputeCount(id: UUID) { - queue.sync { [self] in + let increment = { [self] in guard isDatabaseOpen else { return } let sql = "UPDATE speakers SET dispute_count = dispute_count + 1 WHERE id = ?;" var statement: OpaquePointer? @@ -109,11 +131,12 @@ extension SpeakerDatabase { } sqlite3_finalize(statement) } + if isExecutingOnQueue { increment() } else { queue.sync(execute: increment) } } /// Reset dispute count (after user manual rename or name confirmed) public func resetDisputeCount(id: UUID) { - queue.sync { [self] in + let reset = { [self] in guard isDatabaseOpen else { return } let sql = "UPDATE speakers SET dispute_count = 0 WHERE id = ?;" var statement: OpaquePointer? @@ -127,6 +150,7 @@ extension SpeakerDatabase { } sqlite3_finalize(statement) } + if isExecutingOnQueue { reset() } else { queue.sync(execute: reset) } } // MARK: - Profile Lookup by Name @@ -153,18 +177,31 @@ extension SpeakerDatabase { /// Blends embeddings weighted by call count, sums call counts, bumps confidence. /// Transfers name from source if target has none. Deletes source profile. public func mergeProfiles(sourceId: UUID, into targetId: UUID) throws { + if isExecutingOnQueue { + try mergeProfilesImpl( + sourceId: sourceId, + into: targetId, + transactionAlreadyOpen: sqlite3_get_autocommit(db) == 0 + ) + return + } try queue.sync { try mergeProfilesImpl(sourceId: sourceId, into: targetId) } } - func mergeProfilesImpl(sourceId: UUID, into targetId: UUID, kind: String = SpeakerMergeKind.explicit) throws { + func mergeProfilesImpl( + sourceId: UUID, + into targetId: UUID, + kind: String = SpeakerMergeKind.explicit, + transactionAlreadyOpen: Bool = false + ) throws { guard let source = getSpeakerImpl(id: sourceId), let target = getSpeakerImpl(id: targetId) else { AppLogger.speakers.warning("Merge failed — profile not found", [ "sourceId": "\(sourceId)", "targetId": "\(targetId)" ]) - return + throw ProfileMergeError.profileNotFound(sourceId: sourceId, targetId: targetId) } // Blend embeddings weighted by call count (stronger profile dominates) @@ -177,7 +214,7 @@ extension SpeakerDatabase { "targetDim": "\(target.embedding.count)", "totalCalls": "\(Int(totalCalls))" ]) - return + throw ProfileMergeError.invalidEmbeddingState(sourceId: sourceId, targetId: targetId) } let sourceWeight = Float(source.callCount) / totalCalls @@ -198,8 +235,7 @@ extension SpeakerDatabase { ? (source.nameSource ?? NameSource.userManual) : target.nameSource - // Every persistent identity mutation belongs to this one throwing transaction. - try throwingTransaction { + let performMerge = { // Safety net: snapshot both profiles + re-point provenance BEFORE the blend // overwrites the target and the source row is deleted, so a wrong merge can // be reconstructed into two distinct profiles later. Same transaction → atomic. @@ -261,11 +297,20 @@ extension SpeakerDatabase { try requireDone(deleteStmt, operation: "step merge source delete", expectedChanges: 1) } - AppLogger.speakers.info("Merged profiles", [ - "source": source.displayName ?? "\(sourceId)", - "target": mergedDisplayName ?? "\(targetId)", - "newCallCount": "\(newCallCount)" - ]) + if transactionAlreadyOpen { + try performMerge() + } else { + // Every persistent identity mutation belongs to this one throwing transaction. + try throwingTransaction(performMerge) + } + + if !transactionAlreadyOpen { + AppLogger.speakers.info("Merged profiles", [ + "source": source.displayName ?? "\(sourceId)", + "target": mergedDisplayName ?? "\(targetId)", + "newCallCount": "\(newCallCount)" + ]) + } } private func deleteIdentityRowsForMerge(table: String, profileId: UUID, operation: String) throws { @@ -341,12 +386,13 @@ extension SpeakerDatabase { try mergeProfilesImpl(sourceId: absorbed.id, into: keeper.id, kind: SpeakerMergeKind.duplicate) } catch { AppLogger.speakers.error("Duplicate speaker merge failed", ["error": error.localizedDescription]) - return + continue } mergedIds.insert(absorbed.id) mergeCount += 1 AppLogger.speakers.info("Merged duplicate speaker", ["absorbed": absorbed.displayName ?? "unnamed", "keeper": keeper.displayName ?? "unnamed", "similarity": String(format: "%.3f", similarity)]) + if absorbed.id == speakers[i].id { break } } } @@ -415,7 +461,7 @@ extension SpeakerDatabase { try mergeProfilesImpl(sourceId: source.id, into: keeper.id, kind: SpeakerMergeKind.byName) } catch { AppLogger.speakers.error("Same-name speaker merge failed", ["error": error.localizedDescription]) - return + continue } mergeCount += 1 } diff --git a/Sources/UI/Settings/SpeakerPeopleSettingsSection.swift b/Sources/UI/Settings/SpeakerPeopleSettingsSection.swift index e0b462151..18ed40647 100644 --- a/Sources/UI/Settings/SpeakerPeopleSettingsSection.swift +++ b/Sources/UI/Settings/SpeakerPeopleSettingsSection.swift @@ -323,23 +323,31 @@ final class SpeakerPeopleSettingsViewModel: ObservableObject { let legacyClipsDirectory = self.legacyClipsDirectory DispatchQueue.global(qos: .userInitiated).async { [weak self] in - Self.promoteClipIfNeeded( - from: sourceId, - to: targetId, - preferredClipsDirectory: preferredClipsDirectory, - legacyClipsDirectory: legacyClipsDirectory - ) do { - try speakerDatabase.mergeProfiles(sourceId: sourceId, into: targetId) + try SpeakerProfileMergeSideEffectCoordinator.merge( + databaseMerge: { + try speakerDatabase.mergeProfiles(sourceId: sourceId, into: targetId) + }, + promoteClip: { + Self.promoteClipIfNeeded( + from: sourceId, + to: targetId, + preferredClipsDirectory: preferredClipsDirectory, + legacyClipsDirectory: legacyClipsDirectory + ) + }, + deleteSourceClips: { + Self.deleteClips( + for: sourceId, + preferredClipsDirectory: preferredClipsDirectory, + legacyClipsDirectory: legacyClipsDirectory + ) + } + ) } catch { AppLogger.speakers.error("Manual speaker merge failed", ["error": error.localizedDescription]) return } - Self.deleteClips( - for: sourceId, - preferredClipsDirectory: preferredClipsDirectory, - legacyClipsDirectory: legacyClipsDirectory - ) let resolvedName = speakerDatabase.getSpeaker(id: targetId)?.displayName ?? targetName diff --git a/Sources/UI/Settings/SpeakerProfileMergeSideEffectCoordinator.swift b/Sources/UI/Settings/SpeakerProfileMergeSideEffectCoordinator.swift new file mode 100644 index 000000000..31c09341f --- /dev/null +++ b/Sources/UI/Settings/SpeakerProfileMergeSideEffectCoordinator.swift @@ -0,0 +1,11 @@ +enum SpeakerProfileMergeSideEffectCoordinator { + static func merge( + databaseMerge: () throws -> Void, + promoteClip: () -> Void, + deleteSourceClips: () -> Void + ) rethrows { + try databaseMerge() + promoteClip() + deleteSourceClips() + } +} diff --git a/Tests/SpeakerProfileMergeSideEffectCoordinatorTests.swift b/Tests/SpeakerProfileMergeSideEffectCoordinatorTests.swift new file mode 100644 index 000000000..f032175f0 --- /dev/null +++ b/Tests/SpeakerProfileMergeSideEffectCoordinatorTests.swift @@ -0,0 +1,53 @@ +import Foundation + +private enum SpeakerProfileMergeSideEffectTestError: Error { + case mergeFailed +} + +func testSpeakerProfileMergeSideEffectCoordinator() { + runSuite("Speaker clip side effects follow the committed database merge") { + var operations: [String] = [] + + do { + try SpeakerProfileMergeSideEffectCoordinator.merge( + databaseMerge: { () throws -> Void in operations.append("database") }, + promoteClip: { operations.append("promote") }, + deleteSourceClips: { operations.append("delete-source") } + ) + } catch { + assertTrue(false, "Unexpected merge error: \(error)") + } + + assertEqual( + operations, + ["database", "promote", "delete-source"], + "clip promotion and deletion should follow the committed database merge" + ) + } + + runSuite("A throwing database merge leaves speaker clips untouched") { + var operations: [String] = [] + + do { + try SpeakerProfileMergeSideEffectCoordinator.merge( + databaseMerge: { + operations.append("database") + throw SpeakerProfileMergeSideEffectTestError.mergeFailed + }, + promoteClip: { operations.append("promote") }, + deleteSourceClips: { operations.append("delete-source") } + ) + assertTrue(false, "Expected database merge failure") + } catch SpeakerProfileMergeSideEffectTestError.mergeFailed { + // Expected. Filesystem side effects must not run after rollback. + } catch { + assertTrue(false, "Unexpected merge error: \(error)") + } + + assertEqual( + operations, + ["database"], + "a throwing database merge must leave both source and target clips untouched" + ) + } +} diff --git a/Tests/TranscriptedCoreTests/SpeakerTests/SpeakerNamingCoordinatorTests.swift b/Tests/TranscriptedCoreTests/SpeakerTests/SpeakerNamingCoordinatorTests.swift index ec49bd42c..d0561ddb2 100644 --- a/Tests/TranscriptedCoreTests/SpeakerTests/SpeakerNamingCoordinatorTests.swift +++ b/Tests/TranscriptedCoreTests/SpeakerTests/SpeakerNamingCoordinatorTests.swift @@ -1,6 +1,7 @@ import XCTest import Combine import FluidAudio +import SQLite3 @testable import TranscriptedCore @available(macOS 14.0, *) @@ -2120,6 +2121,122 @@ final class SpeakerNamingCoordinatorTests: XCTestCase { XCTAssertFalse(FileManager.default.fileExists(atPath: systemURL.path)) } + @MainActor + func testHandleNamingCompleteRestoresTranscriptWhenDatabaseMergeFails() async throws { + let harness = try makeHarness() + let transcriptId = UUID() + let source = harness.speakerDB.addOrUpdateSpeaker( + embedding: [Float](repeating: 0.25, count: 256), + existingId: nil + ) + let target = harness.speakerDB.addOrUpdateSpeaker( + embedding: [Float](repeating: 0.26, count: 256), + existingId: nil + ) + harness.speakerDB.setDisplayName(id: target.id, name: "Sarah Graham", source: NameSource.userManual) + + let transcriptURL = harness.paths.transcripts.appendingPathComponent("Merge_Rollback.md") + let micURL = harness.paths.audioCaptures.appendingPathComponent("merge-rollback-mic.wav") + let systemURL = harness.paths.audioCaptures.appendingPathComponent("merge-rollback-system.wav") + let clipURL = harness.paths.speakerClips.appendingPathComponent("merge-rollback-clip.wav") + let speakers = [ + MarkdownSpeaker( + id: "1", + persistentSpeakerId: source.id, + name: "Speaker 1", + confidence: "unknown", + source: "db_pending" + ) + ] + let utterances = [ + MarkdownUtterance( + timestamp: "00:01", + source: "System", + label: "Speaker 1", + text: "Thanks for joining." + ) + ] + let originalTranscript = sampleTranscript( + transcriptId: transcriptId, + speakers: speakers, + utterances: utterances, + breakdownEntries: [ + BreakdownEntry(name: "Speaker 1", utterances: 1, wordCount: 3, duration: "00:03") + ] + ) + try originalTranscript.write(to: transcriptURL, atomically: true, encoding: .utf8) + try Data().write(to: micURL) + try Data().write(to: systemURL) + try Data().write(to: clipURL) + + let authorizerResult = harness.speakerDB.queue.sync { + sqlite3_set_authorizer(harness.speakerDB.db, { _, action, tableName, _, _, _ in + guard action == SQLITE_UPDATE, + let tableName, + String(cString: tableName) == "speakers" else { + return SQLITE_OK + } + return SQLITE_DENY + }, nil) + } + XCTAssertEqual(authorizerResult, SQLITE_OK) + defer { + _ = harness.speakerDB.queue.sync { + sqlite3_set_authorizer(harness.speakerDB.db, nil, nil) + } + } + + harness.manager.speakerNamingRequest = SpeakerNamingRequest( + speakers: [], + transcriptURL: transcriptURL, + transcriptId: transcriptId, + systemAudioURL: systemURL, + micAudioURL: micURL, + onComplete: { _ in } + ) + harness.manager.handleNamingComplete( + updates: [ + SpeakerNameUpdate( + persistentSpeakerId: source.id, + diarizerSpeakerId: "1", + newName: "Sarah Graham", + previousName: nil, + action: .named + ) + ], + transcriptURL: transcriptURL, + transcriptId: transcriptId, + transcriptionResult: sampleTranscriptionResult(speakers: speakers, utterances: utterances), + micURL: micURL, + systemURL: systemURL, + clips: [ + SpeakerNamingEntry( + id: source.id, + diarizerSpeakerId: "1", + clipURL: clipURL, + sampleText: "Thanks for joining.", + currentName: nil, + matchSimilarity: nil, + needsNaming: true, + needsConfirmation: false, + sessionEmbedding: [Float](repeating: 0.25, count: 256) + ) + ] + ) + + try await waitUntil { + if case .failed(message: "Failed to finalize speaker names") = harness.manager.displayStatus, + harness.manager.speakerNamingRequest == nil { + return true + } + return false + } + + XCTAssertEqual(try String(contentsOf: transcriptURL, encoding: .utf8), originalTranscript) + XCTAssertNotNil(harness.speakerDB.getSpeaker(id: source.id)) + XCTAssertNotNil(harness.speakerDB.getSpeaker(id: target.id)) + } + @MainActor func testHandleNamingCompleteResolvesRenamedTranscriptByStableId() async throws { let harness = try makeHarness() diff --git a/Tests/TranscriptedCoreTests/SpeakerTests/SpeakerProfileMergerTests.swift b/Tests/TranscriptedCoreTests/SpeakerTests/SpeakerProfileMergerTests.swift index 5e7d34be2..fa4f7a0dd 100644 --- a/Tests/TranscriptedCoreTests/SpeakerTests/SpeakerProfileMergerTests.swift +++ b/Tests/TranscriptedCoreTests/SpeakerTests/SpeakerProfileMergerTests.swift @@ -149,6 +149,46 @@ final class SpeakerProfileMergerTests: XCTestCase { XCTAssertEqual(merged.confidence, min(1.0, targetBefore.confidence + 0.15), accuracy: 0.0001, "confidence bumps") } + func testMergeProfilesThrowsWhenEitherProfileIsMissing() throws { + let target = database.addOrUpdateSpeaker( + embedding: [Float](repeating: 0.30, count: 256), + existingId: nil + ) + let missingSourceId = UUID() + + XCTAssertThrowsError(try database.mergeProfiles(sourceId: missingSourceId, into: target.id)) { error in + guard case SpeakerDatabase.ProfileMergeError.profileNotFound( + sourceId: missingSourceId, + targetId: target.id + ) = error else { + XCTFail("Expected profileNotFound, got \(error)") + return + } + } + } + + func testAtomicMergeBatchRollsBackEarlierMergeWhenLaterMergeFails() throws { + let firstSource = database.addOrUpdateSpeaker( + embedding: [Float](repeating: 0.20, count: 256), + existingId: nil + ) + let firstTarget = database.addOrUpdateSpeaker( + embedding: [Float](repeating: 0.30, count: 256), + existingId: nil + ) + let firstSourceBefore = try XCTUnwrap(database.getSpeaker(id: firstSource.id)) + let firstTargetBefore = try XCTUnwrap(database.getSpeaker(id: firstTarget.id)) + + XCTAssertThrowsError(try database.performMutationBatch { + try database.mergeProfiles(sourceId: firstSource.id, into: firstTarget.id) + try database.mergeProfiles(sourceId: UUID(), into: firstTarget.id) + }) + + XCTAssertEqual(database.getSpeaker(id: firstSource.id), firstSourceBefore) + XCTAssertEqual(database.getSpeaker(id: firstTarget.id), firstTargetBefore) + XCTAssertTrue(database.recentUndoableMerges().isEmpty) + } + func testMergeProfilesRollsBackAndThrowsWhenTargetUpdateCannotBePrepared() throws { let (source, target) = makeNamedSourceAndUnnamedTarget() let result = database.queue.sync { @@ -246,6 +286,44 @@ final class SpeakerProfileMergerTests: XCTestCase { XCTAssertEqual(survivors.count, 1, "identical unnamed embeddings above threshold merge into one") } + func testMergeDuplicatesContinuesAfterOuterProfileIsAbsorbed() { + let now = Date() + let lowCallOuter = database.addOrUpdateSpeaker( + embedding: [Float](repeating: 0.25, count: 256), + existingId: nil + ) + let highCallKeeper = database.addOrUpdateSpeaker( + embedding: [Float](repeating: 0.25, count: 256), + existingId: nil + ) + let remaining = database.addOrUpdateSpeaker( + embedding: [Float](repeating: 0.25, count: 256), + existingId: nil + ) + database.restoreProfile(SpeakerProfile( + id: lowCallOuter.id, displayName: nil, nameSource: nil, + embedding: lowCallOuter.embedding, firstSeen: now, lastSeen: now, + callCount: 1, confidence: 0.5, disputeCount: 0 + )) + database.restoreProfile(SpeakerProfile( + id: highCallKeeper.id, displayName: nil, nameSource: nil, + embedding: highCallKeeper.embedding, firstSeen: now, lastSeen: now.addingTimeInterval(-60), + callCount: 3, confidence: 0.7, disputeCount: 0 + )) + database.restoreProfile(SpeakerProfile( + id: remaining.id, displayName: nil, nameSource: nil, + embedding: remaining.embedding, firstSeen: now, lastSeen: now.addingTimeInterval(-120), + callCount: 1, confidence: 0.5, disputeCount: 0 + )) + + database.mergeDuplicates(threshold: 0.6) + + let survivors = [lowCallOuter.id, highCallKeeper.id, remaining.id].compactMap { + database.getSpeaker(id: $0) + } + XCTAssertEqual(survivors.map(\.id), [highCallKeeper.id]) + } + func testPruneWeakProfilesRemovesNegativeExemplarsWithProfile() { let profileId = UUID() _ = database.addOrUpdateSpeaker( diff --git a/scripts/entrypoints/run-tests.sh b/scripts/entrypoints/run-tests.sh index 0f823cdee..01473dd17 100755 --- a/scripts/entrypoints/run-tests.sh +++ b/scripts/entrypoints/run-tests.sh @@ -393,6 +393,7 @@ APP_SOURCES=( "Sources/UI/Shared/HomeMeetingRename.swift" "Sources/UI/Settings/HomeMeetingSummaryBetaPresentationPolicy.swift" "Sources/UI/Settings/SpeakerVoiceRowPresentation.swift" + "Sources/UI/Settings/SpeakerProfileMergeSideEffectCoordinator.swift" "Sources/UI/Settings/HomeFailedMeetingInlinePresentation.swift" "Sources/UI/Settings/FailedMeetingRecoveryPresentation.swift" "Sources/UI/Settings/HomeMeetingPreviewFormatter.swift" From 6b3fb9df6ed377e3f96736bf9a1d1c6f7b1c343e Mon Sep 17 00:00:00 2001 From: Codex Date: Fri, 17 Jul 2026 07:17:54 -0500 Subject: [PATCH 3/4] Fix speaker merge package build --- .../Speaker/SpeakerProfileMerger.swift | 2 +- .../SpeakerProfileMergerTests.swift | 26 +++++++++++++++++-- 2 files changed, 25 insertions(+), 3 deletions(-) diff --git a/Sources/TranscriptedCore/Speaker/SpeakerProfileMerger.swift b/Sources/TranscriptedCore/Speaker/SpeakerProfileMerger.swift index 7f9e004b8..aa03d7db5 100644 --- a/Sources/TranscriptedCore/Speaker/SpeakerProfileMerger.swift +++ b/Sources/TranscriptedCore/Speaker/SpeakerProfileMerger.swift @@ -235,7 +235,7 @@ extension SpeakerDatabase { ? (source.nameSource ?? NameSource.userManual) : target.nameSource - let performMerge = { + let performMerge = { [self] in // Safety net: snapshot both profiles + re-point provenance BEFORE the blend // overwrites the target and the source row is deleted, so a wrong merge can // be reconstructed into two distinct profiles later. Same transaction → atomic. diff --git a/Tests/TranscriptedCoreTests/SpeakerTests/SpeakerProfileMergerTests.swift b/Tests/TranscriptedCoreTests/SpeakerTests/SpeakerProfileMergerTests.swift index fa4f7a0dd..b44ca74cc 100644 --- a/Tests/TranscriptedCoreTests/SpeakerTests/SpeakerProfileMergerTests.swift +++ b/Tests/TranscriptedCoreTests/SpeakerTests/SpeakerProfileMergerTests.swift @@ -184,11 +184,33 @@ final class SpeakerProfileMergerTests: XCTestCase { try database.mergeProfiles(sourceId: UUID(), into: firstTarget.id) }) - XCTAssertEqual(database.getSpeaker(id: firstSource.id), firstSourceBefore) - XCTAssertEqual(database.getSpeaker(id: firstTarget.id), firstTargetBefore) + assertProfile(database.getSpeaker(id: firstSource.id), equals: firstSourceBefore) + assertProfile(database.getSpeaker(id: firstTarget.id), equals: firstTargetBefore) XCTAssertTrue(database.recentUndoableMerges().isEmpty) } + private func assertProfile( + _ actual: SpeakerProfile?, + equals expected: SpeakerProfile, + file: StaticString = #filePath, + line: UInt = #line + ) { + guard let actual else { + XCTFail("Expected speaker profile \(expected.id)", file: file, line: line) + return + } + XCTAssertEqual(actual.id, expected.id, file: file, line: line) + XCTAssertEqual(actual.displayName, expected.displayName, file: file, line: line) + XCTAssertEqual(actual.nameSource, expected.nameSource, file: file, line: line) + XCTAssertEqual(actual.embedding, expected.embedding, file: file, line: line) + XCTAssertEqual(actual.exemplars, expected.exemplars, file: file, line: line) + XCTAssertEqual(actual.firstSeen, expected.firstSeen, file: file, line: line) + XCTAssertEqual(actual.lastSeen, expected.lastSeen, file: file, line: line) + XCTAssertEqual(actual.callCount, expected.callCount, file: file, line: line) + XCTAssertEqual(actual.confidence, expected.confidence, file: file, line: line) + XCTAssertEqual(actual.disputeCount, expected.disputeCount, file: file, line: line) + } + func testMergeProfilesRollsBackAndThrowsWhenTargetUpdateCannotBePrepared() throws { let (source, target) = makeNamedSourceAndUnnamedTarget() let result = database.queue.sync { From 0936030abfcbe0fb69c7bc3c3dd64f0f428fb3a0 Mon Sep 17 00:00:00 2001 From: r3dbars Date: Sat, 18 Jul 2026 07:21:31 -0500 Subject: [PATCH 4/4] Propagate speaker naming persistence failures --- .../Speaker/SpeakerDatabase.swift | 32 ++++++++++++++++++- .../SpeakerNegativeExemplarStore.swift | 5 +++ .../Speaker/SpeakerProfileMerger.swift | 20 ++++++++++-- .../SpeakerNamingCoordinatorTests.swift | 4 +-- 4 files changed, 56 insertions(+), 5 deletions(-) diff --git a/Sources/TranscriptedCore/Speaker/SpeakerDatabase.swift b/Sources/TranscriptedCore/Speaker/SpeakerDatabase.swift index 6ed01a8e7..66ea78f46 100644 --- a/Sources/TranscriptedCore/Speaker/SpeakerDatabase.swift +++ b/Sources/TranscriptedCore/Speaker/SpeakerDatabase.swift @@ -29,6 +29,10 @@ 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() @@ -230,14 +234,35 @@ public final class SpeakerDatabase: @unchecked Sendable { 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 { - try throwingTransaction(mutations) + 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) @@ -325,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) } @@ -370,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) @@ -432,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) diff --git a/Sources/TranscriptedCore/Speaker/SpeakerNegativeExemplarStore.swift b/Sources/TranscriptedCore/Speaker/SpeakerNegativeExemplarStore.swift index 8c2dc9635..e02ec0afe 100644 --- a/Sources/TranscriptedCore/Speaker/SpeakerNegativeExemplarStore.swift +++ b/Sources/TranscriptedCore/Speaker/SpeakerNegativeExemplarStore.swift @@ -50,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 } @@ -63,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) } @@ -75,6 +77,7 @@ extension SpeakerDatabase { "sqlite_error": dbErrorMessage(), "profileId": profileId.uuidString ]) + recordMutationFailure(operation: "insert negative exemplar", code: sqlite3_errcode(db)) } sqlite3_finalize(statement) @@ -96,6 +99,7 @@ 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) @@ -103,6 +107,7 @@ extension SpeakerDatabase { 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) } diff --git a/Sources/TranscriptedCore/Speaker/SpeakerProfileMerger.swift b/Sources/TranscriptedCore/Speaker/SpeakerProfileMerger.swift index aa03d7db5..0a2f4df13 100644 --- a/Sources/TranscriptedCore/Speaker/SpeakerProfileMerger.swift +++ b/Sources/TranscriptedCore/Speaker/SpeakerProfileMerger.swift @@ -48,6 +48,7 @@ extension SpeakerDatabase { func setDisplayNameImpl(id: UUID, name: String, source: String) { guard isDatabaseOpen else { AppLogger.speakers.error("setDisplayName failed — database not open", ["speakerId": id.uuidString, "name": name]) + recordMutationFailure(operation: "set display name", code: SQLITE_MISUSE) return } let sql = "UPDATE speakers SET display_name = ?, name_source = ? WHERE id = ?;" @@ -58,9 +59,11 @@ extension SpeakerDatabase { sqlite3_bind_text(statement, 3, (id.uuidString as NSString).utf8String, -1, SQLITE_TRANSIENT) if sqlite3_step(statement) != SQLITE_DONE { AppLogger.speakers.error("Failed to set display name", ["sqlite_error": dbErrorMessage(), "id": id.uuidString]) + recordMutationFailure(operation: "set display name", code: sqlite3_errcode(db)) } } else { AppLogger.speakers.error("Failed to prepare setDisplayName", ["sqlite_error": dbErrorMessage()]) + recordMutationFailure(operation: "prepare set display name", code: sqlite3_errcode(db)) } sqlite3_finalize(statement) } @@ -68,6 +71,7 @@ extension SpeakerDatabase { func restoreProfileImpl(_ profile: SpeakerProfile) { guard isDatabaseOpen else { AppLogger.speakers.error("restoreProfile failed — database not open", ["speakerId": profile.id.uuidString]) + recordMutationFailure(operation: "restore profile snapshot", code: SQLITE_MISUSE) return } @@ -108,9 +112,11 @@ extension SpeakerDatabase { if sqlite3_step(statement) != SQLITE_DONE { AppLogger.speakers.error("Failed to restore profile snapshot", ["sqlite_error": dbErrorMessage(), "id": profile.id.uuidString]) + recordMutationFailure(operation: "restore profile snapshot", code: sqlite3_errcode(db)) } } else { AppLogger.speakers.error("Failed to prepare restoreProfile", ["sqlite_error": dbErrorMessage()]) + recordMutationFailure(operation: "prepare restore profile", code: sqlite3_errcode(db)) } sqlite3_finalize(statement) } @@ -118,16 +124,21 @@ extension SpeakerDatabase { /// Increment the dispute count for a speaker (inference disagreed with DB name) public func incrementDisputeCount(id: UUID) { let increment = { [self] in - guard isDatabaseOpen else { return } + guard isDatabaseOpen else { + recordMutationFailure(operation: "increment dispute count", code: SQLITE_MISUSE) + return + } let sql = "UPDATE speakers SET dispute_count = dispute_count + 1 WHERE id = ?;" var statement: OpaquePointer? if sqlite3_prepare_v2(db, sql, -1, &statement, nil) == SQLITE_OK { sqlite3_bind_text(statement, 1, (id.uuidString as NSString).utf8String, -1, SQLITE_TRANSIENT) if sqlite3_step(statement) != SQLITE_DONE { AppLogger.speakers.error("Failed to increment dispute count", ["sqlite_error": dbErrorMessage()]) + recordMutationFailure(operation: "increment dispute count", code: sqlite3_errcode(db)) } } else { AppLogger.speakers.error("Failed to prepare incrementDisputeCount", ["sqlite_error": dbErrorMessage()]) + recordMutationFailure(operation: "prepare increment dispute count", code: sqlite3_errcode(db)) } sqlite3_finalize(statement) } @@ -137,16 +148,21 @@ extension SpeakerDatabase { /// Reset dispute count (after user manual rename or name confirmed) public func resetDisputeCount(id: UUID) { let reset = { [self] in - guard isDatabaseOpen else { return } + guard isDatabaseOpen else { + recordMutationFailure(operation: "reset dispute count", code: SQLITE_MISUSE) + return + } let sql = "UPDATE speakers SET dispute_count = 0 WHERE id = ?;" var statement: OpaquePointer? if sqlite3_prepare_v2(db, sql, -1, &statement, nil) == SQLITE_OK { sqlite3_bind_text(statement, 1, (id.uuidString as NSString).utf8String, -1, SQLITE_TRANSIENT) if sqlite3_step(statement) != SQLITE_DONE { AppLogger.speakers.error("Failed to reset dispute count", ["sqlite_error": dbErrorMessage()]) + recordMutationFailure(operation: "reset dispute count", code: sqlite3_errcode(db)) } } else { AppLogger.speakers.error("Failed to prepare resetDisputeCount", ["sqlite_error": dbErrorMessage()]) + recordMutationFailure(operation: "prepare reset dispute count", code: sqlite3_errcode(db)) } sqlite3_finalize(statement) } diff --git a/Tests/TranscriptedCoreTests/SpeakerTests/SpeakerNamingCoordinatorTests.swift b/Tests/TranscriptedCoreTests/SpeakerTests/SpeakerNamingCoordinatorTests.swift index d0561ddb2..5b4183a76 100644 --- a/Tests/TranscriptedCoreTests/SpeakerTests/SpeakerNamingCoordinatorTests.swift +++ b/Tests/TranscriptedCoreTests/SpeakerTests/SpeakerNamingCoordinatorTests.swift @@ -3554,8 +3554,8 @@ private final class BlockingSpeakerStore: SpeakerStore, @unchecked Sendable { base.deleteSpeaker(id: id) } - func mergeProfiles(sourceId: UUID, into targetId: UUID) { - base.mergeProfiles(sourceId: sourceId, into: targetId) + func mergeProfiles(sourceId: UUID, into targetId: UUID) throws { + try base.mergeProfiles(sourceId: sourceId, into: targetId) } func mergeProfilesByName() {