diff --git a/Sources/Meeting/CLAUDE.md b/Sources/Meeting/CLAUDE.md index b7f234342..c0dea5339 100644 --- a/Sources/Meeting/CLAUDE.md +++ b/Sources/Meeting/CLAUDE.md @@ -21,7 +21,8 @@ - `MeetingFailureCopy.swift` — normalizes `MeetingFailureKind` values into user-facing titles and recovery copy - `MeetingFailureExplanation.swift` — maps meeting outcomes into retryability, artifact-retention, user-visible state, and privacy-safe telemetry fields - `MeetingFailureKind.swift` — canonical failure taxonomy that classifies raw meeting errors into stable machine-readable kinds -- `MeetingImportedAudioPreparer.swift` — copies imported recordings into app-managed scratch paths, derives titles, and prepares single-file meeting transcription jobs +- `ImportedTranscriptionQueueJournalState.swift` — privacy-minimal durable schema for imported-job identity, process ownership, and transcript/cleanup phase +- `MeetingImportedAudioPreparer.swift` — copies imported recordings into app-managed scratch paths and implements the imported-job journal. Journal sessions hold a per-job process lease and retain the record through transcript commit until authorized scratch cleanup or durable failed-queue handoff - `MeetingImportPreparationFailureCopy.swift` — maps an imported-audio preparation failure into the shared `MeetingFailureKind` taxonomy and user-facing retry copy, extracted out of `MeetingSessionController` so it stays unit-testable - `LiveMeetingCodexSession.swift` — app-owned sidecar writer for the opt-in live-meeting workspace used by Codex and Claude Cowork; it must not replace or mutate the normal saved meeting transcript pipeline - `LiveMeetingPreviewServer.swift` — loopback HTTP server that serves the live sidecar preview on a tokenized URL so the page updates in place without full-page refreshes while Transcripted is running @@ -60,6 +61,7 @@ 8. `MeetingSessionController.stopRecording(...)` awaits mic/system audio files from the bridge, then either starts transcription immediately or queues it behind the active job. 9. `MeetingSessionController.cancelRecording(...)` is only for explicit confirmed discard flows; it stops capture, removes scratch audio, and does not enqueue transcription. 10. `MeetingSessionController.importAudioFile(...)` routes standalone recordings through `MeetingImportedAudioPreparer` and into the same save / naming / restyling pipeline used by live captures. +11. Imported jobs use their journal UUID as the transcript UUID. Recovery skips a live lease owner and rejects non-regular scratch entries. Cleanup-pending records delete scratch; committed or stable-transcript records with uncertain cleanup move audio to the durable visible failed queue; only records with no commit proof replay transcription. 11. `TranscriptionTaskManager` runs one diarize → transcribe → save pipeline at a time. When `LocalSpeakerPreferences` is enabled, queued meeting work also asks the core pipeline to diarize the local mic channel instead of treating it as a single "You" speaker. 12. A subscription on `taskManager.$lastSavedTranscriptURL` runs `MeetingTranscriptStyler.restyleTranscript(...)` on a serialized background task (the restyle reads/rewrites the whole transcript and can rename artifacts, so it must stay off the main actor) and hops back to the main actor to update the recent-meetings UI state. The restyle fails closed when a transcript body has text the entry parser cannot understand instead of replacing it with the empty placeholder. 13. After a transcript is saved, `MeetingAudioStorageManager` compresses retained WAV audio to M4A and applies the user's retention setting. Launch and Settings changes also run a backfill pass over existing Transcripted meeting transcripts, and queue-tracked failed-meeting audio can be compressed only after the failed queue is updated to point at the converted files. diff --git a/Sources/Meeting/ImportedTranscriptionQueueJournalState.swift b/Sources/Meeting/ImportedTranscriptionQueueJournalState.swift new file mode 100644 index 000000000..a367ca830 --- /dev/null +++ b/Sources/Meeting/ImportedTranscriptionQueueJournalState.swift @@ -0,0 +1,80 @@ +import Foundation + +enum ImportedTranscriptionQueueJournalPhase: String, Codable, Equatable, Sendable { + case queued + case active + case transcriptCommitted + case scratchCleanupPending +} + +enum ImportedTranscriptionQueueJournalRecoveryAction: Equatable { + case replayTranscription + case cleanScratch + case handOffScratch +} + +struct ImportedTranscriptionQueueJournalOwner: Codable, Equatable, Sendable { + let processIdentifier: Int32 + let claimedAt: Date +} + +struct ImportedTranscriptionQueueJournalRecord: Codable, Equatable, Sendable { + let id: UUID + let audioFilename: String + let recordingDate: Date + let enqueuedAt: Date + let sttModelRawValue: String + var phase: ImportedTranscriptionQueueJournalPhase + var owner: ImportedTranscriptionQueueJournalOwner? + + init( + id: UUID, + audioFilename: String, + recordingDate: Date, + enqueuedAt: Date, + sttModelRawValue: String, + phase: ImportedTranscriptionQueueJournalPhase = .queued, + owner: ImportedTranscriptionQueueJournalOwner? = nil + ) { + self.id = id + self.audioFilename = audioFilename + self.recordingDate = recordingDate + self.enqueuedAt = enqueuedAt + self.sttModelRawValue = sttModelRawValue + self.phase = phase + self.owner = owner + } + + private enum CodingKeys: String, CodingKey { + case id + case audioFilename + case recordingDate + case enqueuedAt + case sttModelRawValue + case phase + case owner + } + + init(from decoder: Decoder) throws { + let values = try decoder.container(keyedBy: CodingKeys.self) + id = try values.decode(UUID.self, forKey: .id) + audioFilename = try values.decode(String.self, forKey: .audioFilename) + recordingDate = try values.decode(Date.self, forKey: .recordingDate) + enqueuedAt = try values.decode(Date.self, forKey: .enqueuedAt) + sttModelRawValue = try values.decode(String.self, forKey: .sttModelRawValue) + phase = try values.decodeIfPresent( + ImportedTranscriptionQueueJournalPhase.self, + forKey: .phase + ) ?? .queued + owner = try values.decodeIfPresent( + ImportedTranscriptionQueueJournalOwner.self, + forKey: .owner + ) + } +} + +enum ImportedTranscriptionQueueJournalError: Error { + case audioOutsideScratchDirectory + case claimFailed + case journalMissing +} diff --git a/Sources/Meeting/MeetingImportPreparationFailureCopy.swift b/Sources/Meeting/MeetingImportPreparationFailureCopy.swift index 11a9331c7..2ff1b5b44 100644 --- a/Sources/Meeting/MeetingImportPreparationFailureCopy.swift +++ b/Sources/Meeting/MeetingImportPreparationFailureCopy.swift @@ -1,5 +1,17 @@ import Foundation +enum ImportedAudioQueuePersistenceFailureCopy { + static let retryEntryMessage = + "Imported audio was saved after Transcripted couldn't add it to the transcription queue. Finish the transcript from Home." + + static func displayMessage(preservedForRelaunch: Bool) -> String { + if preservedForRelaunch { + return "Transcripted couldn't safely queue that import. The copied audio was saved for retry in Home." + } + return "Transcripted couldn't safely queue that import or save a retry entry. Import the original file again." + } +} + /// Maps an imported-audio preparation failure into the shared failure-kind /// taxonomy and user-facing copy. Extracted from `MeetingSessionController` /// so this classification stays unit-testable without instantiating the diff --git a/Sources/Meeting/MeetingImportedAudioPreparer.swift b/Sources/Meeting/MeetingImportedAudioPreparer.swift index 3d928c3f8..1f8d64ec0 100644 --- a/Sources/Meeting/MeetingImportedAudioPreparer.swift +++ b/Sources/Meeting/MeetingImportedAudioPreparer.swift @@ -1,5 +1,9 @@ import Foundation import AVFoundation +import Darwin +#if canImport(TranscriptedCore) +import TranscriptedCore +#endif import UniformTypeIdentifiers struct PreparedImportedMeetingAudio: Sendable { @@ -8,39 +12,21 @@ struct PreparedImportedMeetingAudio: Sendable { let recordingDate: Date } -struct ImportedTranscriptionQueueJournalRecord: Codable, Equatable, Sendable { - let id: UUID - let audioFilename: String - let suggestedTitle: String - let recordingDate: Date - let enqueuedAt: Date - let sttModelRawValue: String -} - -enum ImportedTranscriptionQueueJournalError: Error { - case audioOutsideScratchDirectory -} - -enum ImportedAudioQueuePersistenceFailureCopy { - static let retryEntryMessage = - "Imported audio was saved after Transcripted couldn't add it to the transcription queue. Finish the transcript from Home." - - static func displayMessage(preservedForRelaunch: Bool) -> String { - if preservedForRelaunch { - return "Transcripted couldn't safely queue that import. The copied audio was saved for retry in Home." - } - return "Transcripted couldn't safely queue that import or save a retry entry. Import the original file again." - } -} - enum ImportedTranscriptionQueueJournal { private static let filenamePrefix = "import-job-" private static let filenameExtension = "json" + private static let lockFilenameExtension = "lock" + private static let maximumRecordBytes = 64 * 1024 + + enum RecoveryAudioStatus: Equatable { + case regularFile + case missing + case unsafeEntry + } static func persist( id: UUID, audioURL: URL, - suggestedTitle: String, recordingDate: Date, enqueuedAt: Date = Date(), sttModelRawValue: String, @@ -61,15 +47,49 @@ enum ImportedTranscriptionQueueJournal { let record = ImportedTranscriptionQueueJournalRecord( id: id, audioFilename: normalizedAudioURL.lastPathComponent, - suggestedTitle: suggestedTitle, recordingDate: recordingDate, enqueuedAt: enqueuedAt, sttModelRawValue: sttModelRawValue ) - let data = try JSONEncoder().encode(record) - let destination = journalURL(for: id, in: journalDirectory) - try data.write(to: destination, options: [.atomic]) - fileManager.restrictFileToOwnerOnly(at: destination) + try write(record, journalDirectory: journalDirectory, fileManager: fileManager) + } + + /// Creates the initial record only after taking its process lease, so no + /// other process can observe recoverable work before its owner is live. + static func createClaimed( + id: UUID, + audioURL: URL, + recordingDate: Date, + enqueuedAt: Date = Date(), + sttModelRawValue: String, + journalDirectory: URL, + scratchDirectory: URL, + processIdentifier: Int32 = getpid(), + claimedAt: Date = Date(), + fileManager: FileManager = .default + ) throws -> ImportedTranscriptionQueueJournalSession { + let normalizedAudioURL = audioURL.standardizedFileURL + guard normalizedAudioURL.deletingLastPathComponent() == scratchDirectory.standardizedFileURL else { + throw ImportedTranscriptionQueueJournalError.audioOutsideScratchDirectory + } + let record = ImportedTranscriptionQueueJournalRecord( + id: id, + audioFilename: normalizedAudioURL.lastPathComponent, + recordingDate: recordingDate, + enqueuedAt: enqueuedAt, + sttModelRawValue: sttModelRawValue + ) + guard let session = try claim( + id: id, + journalDirectory: journalDirectory, + processIdentifier: processIdentifier, + claimedAt: claimedAt, + fileManager: fileManager, + recordToPublish: record + ) else { + throw ImportedTranscriptionQueueJournalError.claimFailed + } + return session } static func load( @@ -78,7 +98,7 @@ enum ImportedTranscriptionQueueJournal { ) -> [ImportedTranscriptionQueueJournalRecord] { guard let urls = try? fileManager.contentsOfDirectory( at: journalDirectory, - includingPropertiesForKeys: nil, + includingPropertiesForKeys: [.isRegularFileKey, .isSymbolicLinkKey, .fileSizeKey], options: [.skipsHiddenFiles] ) else { return [] } @@ -88,11 +108,7 @@ enum ImportedTranscriptionQueueJournal { && $0.deletingPathExtension().lastPathComponent.hasPrefix(filenamePrefix) } .compactMap { url in - guard let data = try? Data(contentsOf: url), - let record = try? JSONDecoder().decode( - ImportedTranscriptionQueueJournalRecord.self, - from: data - ), + guard let record = readRecord(from: url), !record.audioFilename.isEmpty, URL(fileURLWithPath: record.audioFilename).lastPathComponent == record.audioFilename else { return nil } @@ -106,12 +122,104 @@ enum ImportedTranscriptionQueueJournal { } } + @discardableResult static func remove( id: UUID, journalDirectory: URL, fileManager: FileManager = .default - ) { - try? fileManager.removeItem(at: journalURL(for: id, in: journalDirectory)) + ) -> Bool { + do { + try fileManager.removeItem(at: journalURL(for: id, in: journalDirectory)) + synchronizeDirectory(journalDirectory) + return true + } catch { + if (error as NSError).code == NSFileNoSuchFileError { + return true + } + return false + } + } + + /// Claims one journal with an advisory lock held for the lifetime of the + /// returned session. A second process receives `nil`; a crashed owner loses + /// the kernel lock automatically and the next process can recover the work. + static func claim( + id: UUID, + journalDirectory: URL, + processIdentifier: Int32 = getpid(), + claimedAt: Date = Date(), + fileManager: FileManager = .default, + recordToPublish: ImportedTranscriptionQueueJournalRecord? = nil + ) throws -> ImportedTranscriptionQueueJournalSession? { + fileManager.ensurePrivateDirectory( + at: journalDirectory, + context: "imported transcription queue journal" + ) + let lockURL = lockURL(for: id, in: journalDirectory) + let descriptor = Darwin.open( + lockURL.path, + O_CREAT | O_RDWR | O_CLOEXEC | O_NOFOLLOW, + S_IRUSR | S_IWUSR + ) + guard descriptor >= 0 else { + throw ImportedTranscriptionQueueJournalError.claimFailed + } + + var lockStatus = stat() + guard fstat(descriptor, &lockStatus) == 0, + (lockStatus.st_mode & S_IFMT) == S_IFREG else { + Darwin.close(descriptor) + throw ImportedTranscriptionQueueJournalError.claimFailed + } + + guard flock(descriptor, LOCK_EX | LOCK_NB) == 0 else { + Darwin.close(descriptor) + if errno == EWOULDBLOCK || errno == EAGAIN { + return nil + } + throw ImportedTranscriptionQueueJournalError.claimFailed + } + + do { + var record: ImportedTranscriptionQueueJournalRecord + if let recordToPublish { + var journalStatus = stat() + guard lstat(journalURL(for: id, in: journalDirectory).path, &journalStatus) != 0, + errno == ENOENT else { + throw ImportedTranscriptionQueueJournalError.claimFailed + } + record = recordToPublish + } else { + guard let loaded = load( + id: id, + journalDirectory: journalDirectory, + fileManager: fileManager + ) else { + throw ImportedTranscriptionQueueJournalError.journalMissing + } + record = loaded + } + if record.phase == .queued || record.phase == .active { + record.phase = .active + } + record.owner = ImportedTranscriptionQueueJournalOwner( + processIdentifier: processIdentifier, + claimedAt: claimedAt + ) + try write(record, journalDirectory: journalDirectory, fileManager: fileManager) + fileManager.restrictFileToOwnerOnly(at: lockURL) + return ImportedTranscriptionQueueJournalSession( + record: record, + journalDirectory: journalDirectory, + lockDescriptor: descriptor, + fileManager: fileManager + ) + } catch { + unlink(lockURL.path) + flock(descriptor, LOCK_UN) + Darwin.close(descriptor) + throw error + } } static func audioURL( @@ -124,6 +232,22 @@ enum ImportedTranscriptionQueueJournal { return scratchDirectory.appendingPathComponent(record.audioFilename, isDirectory: false) } + static func recoveryAudioStatus( + at audioURL: URL, + scratchDirectory: URL + ) -> RecoveryAudioStatus { + let normalizedURL = audioURL.standardizedFileURL + guard normalizedURL.deletingLastPathComponent() == scratchDirectory.standardizedFileURL else { + return .unsafeEntry + } + + var fileStatus = stat() + guard lstat(normalizedURL.path, &fileStatus) == 0 else { + return errno == ENOENT ? .missing : .unsafeEntry + } + return (fileStatus.st_mode & S_IFMT) == S_IFREG ? .regularFile : .unsafeEntry + } + static func isDuplicate( record: ImportedTranscriptionQueueJournalRecord, audioURL: URL, @@ -134,12 +258,235 @@ enum ImportedTranscriptionQueueJournal { || existingAudioURLs.contains(audioURL.standardizedFileURL) } + static func recoveryAction( + phase: ImportedTranscriptionQueueJournalPhase, + stableTranscriptExists: Bool + ) -> ImportedTranscriptionQueueJournalRecoveryAction { + if phase == .scratchCleanupPending { + return .cleanScratch + } + if phase == .transcriptCommitted || stableTranscriptExists { + return .handOffScratch + } + return .replayTranscription + } + private static func journalURL(for id: UUID, in directory: URL) -> URL { directory.appendingPathComponent( "\(filenamePrefix)\(id.uuidString).\(filenameExtension)", isDirectory: false ) } + + fileprivate static func lockURL(for id: UUID, in directory: URL) -> URL { + directory.appendingPathComponent( + "\(filenamePrefix)\(id.uuidString).\(lockFilenameExtension)", + isDirectory: false + ) + } + + fileprivate static func load( + id: UUID, + journalDirectory: URL, + fileManager: FileManager + ) -> ImportedTranscriptionQueueJournalRecord? { + let url = journalURL(for: id, in: journalDirectory) + guard let record = readRecord(from: url), + record.id == id, + !record.audioFilename.isEmpty, + URL(fileURLWithPath: record.audioFilename).lastPathComponent == record.audioFilename + else { return nil } + return record + } + + fileprivate static func write( + _ record: ImportedTranscriptionQueueJournalRecord, + journalDirectory: URL, + fileManager: FileManager + ) throws { + let data = try JSONEncoder().encode(record) + let destination = journalURL(for: record.id, in: journalDirectory) + let temporary = journalDirectory.appendingPathComponent( + ".\(filenamePrefix)\(record.id.uuidString)-\(UUID().uuidString).tmp", + isDirectory: false + ) + let descriptor = Darwin.open( + temporary.path, + O_CREAT | O_EXCL | O_WRONLY | O_CLOEXEC | O_NOFOLLOW, + S_IRUSR | S_IWUSR + ) + guard descriptor >= 0 else { throw posixError() } + + do { + try data.withUnsafeBytes { bytes in + guard let base = bytes.baseAddress else { return } + var offset = 0 + while offset < bytes.count { + let written = Darwin.write( + descriptor, + base.advanced(by: offset), + bytes.count - offset + ) + guard written > 0 else { throw posixError() } + offset += written + } + } + guard fsync(descriptor) == 0 else { throw posixError() } + guard rename(temporary.path, destination.path) == 0 else { throw posixError() } + synchronizeDirectory(journalDirectory) + Darwin.close(descriptor) + } catch { + Darwin.close(descriptor) + unlink(temporary.path) + throw error + } + fileManager.restrictFileToOwnerOnly(at: destination) + } + + private static func readRecord(from url: URL) -> ImportedTranscriptionQueueJournalRecord? { + let descriptor = Darwin.open(url.path, O_RDONLY | O_CLOEXEC | O_NOFOLLOW) + guard descriptor >= 0 else { return nil } + defer { Darwin.close(descriptor) } + + var status = stat() + guard fstat(descriptor, &status) == 0, + (status.st_mode & S_IFMT) == S_IFREG, + status.st_size >= 0, + status.st_size <= maximumRecordBytes else { + return nil + } + + let handle = FileHandle(fileDescriptor: descriptor, closeOnDealloc: false) + guard let data = try? handle.read(upToCount: maximumRecordBytes + 1), + data.count <= maximumRecordBytes else { + return nil + } + return try? JSONDecoder().decode( + ImportedTranscriptionQueueJournalRecord.self, + from: data + ) + } + + private static func synchronizeDirectory(_ directory: URL) { + let descriptor = Darwin.open(directory.path, O_RDONLY | O_CLOEXEC) + guard descriptor >= 0 else { return } + _ = fsync(descriptor) + Darwin.close(descriptor) + } + + private static func posixError() -> POSIXError { + POSIXError(POSIXErrorCode(rawValue: errno) ?? .EIO) + } +} + +final class ImportedTranscriptionQueueJournalSession: ImportedTranscriptionRecoverySession, @unchecked Sendable { + let jobID: UUID + + private let journalDirectory: URL + private let fileManager: FileManager + private let stateLock = NSLock() + private var record: ImportedTranscriptionQueueJournalRecord + private var lockDescriptor: Int32 + + fileprivate init( + record: ImportedTranscriptionQueueJournalRecord, + journalDirectory: URL, + lockDescriptor: Int32, + fileManager: FileManager + ) { + jobID = record.id + self.record = record + self.journalDirectory = journalDirectory + self.lockDescriptor = lockDescriptor + self.fileManager = fileManager + } + + deinit { + releaseLock() + } + + func transcriptCommitConfirmed() { + stateLock.withLock { + guard lockDescriptor >= 0 else { return } + guard record.phase != .scratchCleanupPending else { return } + var updatedRecord = record + updatedRecord.phase = .transcriptCommitted + do { + try ImportedTranscriptionQueueJournal.write( + updatedRecord, + journalDirectory: journalDirectory, + fileManager: fileManager + ) + record = updatedRecord + } catch { + return + } + } + } + + func prepareForScratchCleanup() -> Bool { + stateLock.withLock { + guard lockDescriptor >= 0 else { return false } + guard record.phase != .scratchCleanupPending else { return true } + var updatedRecord = record + updatedRecord.phase = .scratchCleanupPending + do { + try ImportedTranscriptionQueueJournal.write( + updatedRecord, + journalDirectory: journalDirectory, + fileManager: fileManager + ) + record = updatedRecord + return true + } catch { + return false + } + } + } + + func scratchCleanupConfirmed() { + finish() + } + + func failedQueueHandoffConfirmed() { + finish() + } + + func supersededRecoveryConfirmed() { + finish() + } + + var phase: ImportedTranscriptionQueueJournalPhase { + stateLock.withLock { record.phase } + } + + private func finish() { + stateLock.withLock { + guard lockDescriptor >= 0 else { return } + let didRemoveJournal = ImportedTranscriptionQueueJournal.remove( + id: jobID, + journalDirectory: journalDirectory, + fileManager: fileManager + ) + if didRemoveJournal { + unlink(ImportedTranscriptionQueueJournal.lockURL(for: jobID, in: journalDirectory).path) + } + releaseLockWhileStateLocked() + } + } + + private func releaseLock() { + stateLock.withLock { + releaseLockWhileStateLocked() + } + } + + private func releaseLockWhileStateLocked() { + guard lockDescriptor >= 0 else { return } + flock(lockDescriptor, LOCK_UN) + Darwin.close(lockDescriptor) + lockDescriptor = -1 + } } enum MeetingImportedAudioPreparationError: LocalizedError, Equatable { diff --git a/Sources/Meeting/MeetingSessionController.swift b/Sources/Meeting/MeetingSessionController.swift index 23b07e699..a5b8c822a 100644 --- a/Sources/Meeting/MeetingSessionController.swift +++ b/Sources/Meeting/MeetingSessionController.swift @@ -1576,8 +1576,20 @@ final class MeetingSessionController: ObservableObject { ) case .imported(let audioURL, let suggestedTitle, let recordingDate): if reason == .userRequested { - try? FileManager.default.removeItem(at: audioURL) - transcriptionQueue.removeImportedJournal(for: job) + guard transcriptionQueue.prepareImportedScratchCleanup(for: job) else { + continue + } + do { + try FileManager.default.removeItem(at: audioURL) + transcriptionQueue.confirmImportedScratchCleanup(for: job) + } catch where (error as NSError).code == NSFileNoSuchFileError { + transcriptionQueue.confirmImportedScratchCleanup(for: job) + } catch { + AppLogger.pipeline.warning("Failed to discard queued imported scratch audio", [ + "file": audioURL.lastPathComponent, + "errorType": String(describing: type(of: error)) + ]) + } } else { if failedMeetingStore.preserveFailedMeetingForRetry( micAudioURL: nil, @@ -1586,7 +1598,7 @@ final class MeetingSessionController: ObservableObject { meetingTitle: suggestedTitle, recordingDate: recordingDate ) { - transcriptionQueue.removeImportedJournal(for: job) + transcriptionQueue.confirmImportedFailedQueueHandoff(for: job) } } } @@ -1678,6 +1690,11 @@ final class MeetingSessionController: ObservableObject { let activePreserved = taskManager.preserveActiveTranscriptionsForShutdown( errorMessage: "Meeting saved before quit. Audio is safe; finish the transcript from Home after reopening." ) + // An active imported pipeline can enqueue speaker review just before + // committing its transcript. Preserve that task first so review cleanup + // cannot retire the journal and delete its only scratch copy. Completed + // pipelines still finalize their typed recovery owner here. + taskManager.cleanupPendingNaming() let shouldFailPendingLiveHandoff = queuedPreserved > 0 || activePreserved > 0 || liveCodexSessionAwaitingFinalTranscript diff --git a/Sources/Meeting/TranscriptionQueueCoordinator.swift b/Sources/Meeting/TranscriptionQueueCoordinator.swift index 14b06b231..0ac02d577 100644 --- a/Sources/Meeting/TranscriptionQueueCoordinator.swift +++ b/Sources/Meeting/TranscriptionQueueCoordinator.swift @@ -43,6 +43,7 @@ final class TranscriptionQueueCoordinator { let stoppedAudioRecovery: DictationStoppedAudioRecovery? let promptTelemetryProperties: [String: String]? let promptRecordingStartedAt: Date? + var importedRecoverySession: ImportedTranscriptionQueueJournalSession? var captureDiagnostics: [String: String]? { switch kind { @@ -132,7 +133,8 @@ final class TranscriptionQueueCoordinator { sttModel: controller.sttRouter.selectedModel, stoppedAudioRecovery: nil, promptTelemetryProperties: promptTelemetryProperties, - promptRecordingStartedAt: promptRecordingStartedAt + promptRecordingStartedAt: promptRecordingStartedAt, + importedRecoverySession: nil ) if controller.liveCodexFinalTranscriptNeedsQueuedJobID && controller.liveCodexSessionAwaitingFinalTranscript { @@ -149,7 +151,7 @@ final class TranscriptionQueueCoordinator { startTrigger: MeetingSessionController.StartTrigger, stoppedAudioRecovery: DictationStoppedAudioRecovery? = nil ) throws -> QueueInsertionOutcome { - let job = QueuedTranscriptionJob( + var job = QueuedTranscriptionJob( id: UUID(), kind: .imported( audioURL: audioURL, @@ -160,12 +162,21 @@ final class TranscriptionQueueCoordinator { sttModel: controller.sttRouter.selectedModel, stoppedAudioRecovery: stoppedAudioRecovery, promptTelemetryProperties: nil, - promptRecordingStartedAt: nil + promptRecordingStartedAt: nil, + importedRecoverySession: nil ) - // Persist ownership before any asynchronous model preparation starts. - // Otherwise a force quit can lose an accepted immediate-start import. - try persistImportedJournal(for: job) + // Take the lease before publishing recoverable work, then keep it + // through every asynchronous queue and pipeline transition. + let session = try ImportedTranscriptionQueueJournal.createClaimed( + id: job.id, + audioURL: audioURL, + recordingDate: recordingDate, + sttModelRawValue: job.sttModel.rawValue, + journalDirectory: importedQueueJournalDirectory, + scratchDirectory: importedAudioScratchDirectory + ) + job.importedRecoverySession = session if canStartQueuedTranscriptionImmediately { startQueuedTranscription(job) @@ -338,12 +349,13 @@ final class TranscriptionQueueCoordinator { ) case .imported(let audioURL, let suggestedTitle, let recordingDate): controller.taskManager.startImportedTranscription( + taskId: job.id, audioURL: audioURL, outputFolder: MeetingStoragePaths.transcriptsFolder, meetingTitle: suggestedTitle, - recordingDate: recordingDate + recordingDate: recordingDate, + recoverySession: job.importedRecoverySession ) - removeImportedJournal(for: job) } } @@ -379,7 +391,7 @@ final class TranscriptionQueueCoordinator { ) } if preserved { - removeImportedJournal(for: job) + job.importedRecoverySession?.failedQueueHandoffConfirmed() } clearQueuedTranscriptionRuntimeDiagnosticsIfOwned(for: job, outcome: "model_recovery_failed") } @@ -457,6 +469,25 @@ final class TranscriptionQueueCoordinator { journalDirectory: importedQueueJournalDirectory ) let failedQueueAudioURLs = controller.failedMeetingStore.failedAudioURLs + let claimedRecords = records.compactMap { record -> ( + ImportedTranscriptionQueueJournalRecord, + ImportedTranscriptionQueueJournalSession + )? in + guard let recoverySession = try? ImportedTranscriptionQueueJournal.claim( + id: record.id, + journalDirectory: importedQueueJournalDirectory + ) else { + return nil + } + return (record, recoverySession) + } + // Scan only after every available lease is held. A prior owner can no + // longer publish a stable transcript after this snapshot and leave us + // with stale evidence that would replay the same job. + let existingTranscriptsByID = TranscriptSaver.existingTranscriptURLs( + in: MeetingStoragePaths.transcriptsFolder, + transcriptIds: Set(claimedRecords.map { $0.0.id }) + ) var recoveredJobIDs = Set(queuedTranscriptionJobs.map(\.id)) var recoveredAudioURLs = Set( queuedTranscriptionJobs.compactMap { job -> URL? in @@ -471,22 +502,15 @@ final class TranscriptionQueueCoordinator { } } var recovered = 0 - for record in records { + for (record, recoverySession) in claimedRecords { guard let audioURL = ImportedTranscriptionQueueJournal.audioURL( for: record, scratchDirectory: importedAudioScratchDirectory ) else { - ImportedTranscriptionQueueJournal.remove( - id: record.id, - journalDirectory: importedQueueJournalDirectory - ) continue } if failedQueueAudioURLs.contains(audioURL.standardizedFileURL) { - ImportedTranscriptionQueueJournal.remove( - id: record.id, - journalDirectory: importedQueueJournalDirectory - ) + recoverySession.failedQueueHandoffConfirmed() continue } if ImportedTranscriptionQueueJournal.isDuplicate( @@ -495,19 +519,53 @@ final class TranscriptionQueueCoordinator { existingJobIDs: recoveredJobIDs, existingAudioURLs: recoveredAudioURLs ) { - ImportedTranscriptionQueueJournal.remove( - id: record.id, - journalDirectory: importedQueueJournalDirectory - ) + recoverySession.supersededRecoveryConfirmed() continue } - guard FileManager.default.fileExists(atPath: audioURL.path) else { - ImportedTranscriptionQueueJournal.remove( - id: record.id, - journalDirectory: importedQueueJournalDirectory - ) + + let audioStatus = ImportedTranscriptionQueueJournal.recoveryAudioStatus( + at: audioURL, + scratchDirectory: importedAudioScratchDirectory + ) + if audioStatus == .missing { + recoverySession.scratchCleanupConfirmed() + continue + } + guard audioStatus == .regularFile else { + AppLogger.pipeline.warning("Rejected unsafe imported recovery scratch entry", [ + "jobId": record.id.uuidString + ]) + continue + } + + let stableTranscriptExists = existingTranscriptsByID[record.id] != nil + switch ImportedTranscriptionQueueJournal.recoveryAction( + phase: recoverySession.phase, + stableTranscriptExists: stableTranscriptExists + ) { + case .replayTranscription: + break + case .cleanScratch: + if removeRecoveredImportedScratchFile(audioURL) { + recoverySession.scratchCleanupConfirmed() + } + continue + case .handOffScratch: + if stableTranscriptExists { + recoverySession.transcriptCommitConfirmed() + } + if controller.failedMeetingStore.preserveFailedMeetingForRetry( + micAudioURL: nil, + systemAudioURL: audioURL, + errorMessage: "The transcript was saved. Imported audio was preserved because recovery could not confirm scratch cleanup.", + meetingTitle: "Imported audio", + recordingDate: record.recordingDate + ) { + recoverySession.failedQueueHandoffConfirmed() + } continue } + let model = TranscriptionModelChoice(rawValue: record.sttModelRawValue) ?? controller.sttRouter.selectedModel queuedTranscriptionJobs.append( @@ -515,14 +573,15 @@ final class TranscriptionQueueCoordinator { id: record.id, kind: .imported( audioURL: audioURL, - suggestedTitle: record.suggestedTitle, + suggestedTitle: "Imported audio", recordingDate: record.recordingDate ), startTrigger: .fileImport, sttModel: model, stoppedAudioRecovery: nil, promptTelemetryProperties: nil, - promptRecordingStartedAt: nil + promptRecordingStartedAt: nil, + importedRecoverySession: recoverySession ) ) recoveredJobIDs.insert(record.id) @@ -535,27 +594,32 @@ final class TranscriptionQueueCoordinator { return recovered } - func removeImportedJournal(for job: QueuedTranscriptionJob) { - guard case .imported = job.kind else { return } - ImportedTranscriptionQueueJournal.remove( - id: job.id, - journalDirectory: importedQueueJournalDirectory - ) + func prepareImportedScratchCleanup(for job: QueuedTranscriptionJob) -> Bool { + job.importedRecoverySession?.prepareForScratchCleanup() ?? true } - private func persistImportedJournal(for job: QueuedTranscriptionJob) throws { - guard case .imported(let audioURL, let suggestedTitle, let recordingDate) = job.kind else { - return + func confirmImportedScratchCleanup(for job: QueuedTranscriptionJob) { + job.importedRecoverySession?.scratchCleanupConfirmed() + } + + func confirmImportedFailedQueueHandoff(for job: QueuedTranscriptionJob) { + job.importedRecoverySession?.failedQueueHandoffConfirmed() + } + + private func removeRecoveredImportedScratchFile(_ audioURL: URL) -> Bool { + do { + try FileManager.default.removeItem(at: audioURL) + return true + } catch { + if (error as NSError).code == NSFileNoSuchFileError { + return true + } + AppLogger.pipeline.warning("Failed to clean committed imported scratch audio", [ + "file": audioURL.lastPathComponent, + "errorType": String(describing: type(of: error)) + ]) + return false } - try ImportedTranscriptionQueueJournal.persist( - id: job.id, - audioURL: audioURL, - suggestedTitle: suggestedTitle, - recordingDate: recordingDate, - sttModelRawValue: job.sttModel.rawValue, - journalDirectory: importedQueueJournalDirectory, - scratchDirectory: importedAudioScratchDirectory - ) } private func finalizeBackgroundTranscriptionStateIfNeeded() { @@ -621,7 +685,7 @@ final class TranscriptionQueueCoordinator { recordingDate: recordingDate ) { preservedCount += 1 - removeImportedJournal(for: job) + job.importedRecoverySession?.failedQueueHandoffConfirmed() } } } diff --git a/Sources/TranscriptedCore/CLAUDE.md b/Sources/TranscriptedCore/CLAUDE.md index 1f88457ad..016f731cf 100644 --- a/Sources/TranscriptedCore/CLAUDE.md +++ b/Sources/TranscriptedCore/CLAUDE.md @@ -10,7 +10,7 @@ - `Logging/` (5 files) — shared app logger (`AppLogger`, subsystem-scoped, os.Logger + JSONL), JSONL file logger (`FileLogger`), generic privacy text redactor, Core log metadata sanitizer, and `LogTailTrimmer` (shared truncate-in-place rotation used by `FileLogger` and by the app target's `AppLogSink`); see `docs/observability.md` for the full sink map, including how this `AppLogger` differs from `Sources/Observability/AppLogSink.swift` - `Models/` (5 files) — public data types: `TranscriptionResult`, `DisplayStatus`, `FailedTranscription`, `SpeakerMapping`, and recording-health metadata builders - `Pipeline/` (4 files) — transcription orchestration, pipeline runner, and task queue -- `Protocols/` (7 files) — host-injected seams: `SpeechToTextEngine`, `DiarizationEngine`, `SpeakerStore`, `TranscriptNotifier`, `AudioCaptureEngine`, `StatsStore`, `TranscriptStorage` +- `Protocols/` (8 files) — host-injected seams: `SpeechToTextEngine`, `DiarizationEngine`, `SpeakerStore`, `TranscriptNotifier`, `AudioCaptureEngine`, `StatsStore`, `TranscriptStorage`, and the typed `ImportedTranscriptionRecoverySession` ownership handoff - `Services/` (7 files) — DI container (`AppServices`), model bundle / download management, path indirection, recording validation, diarization, and failed-transcription persistence - `Speaker/` (27 files) — speaker DB (`SpeakerDatabase`, instance-based, injected via `AppServices`; no `.shared` singleton), embedding matching / clustering, embedding thresholds and segment re-embedding, clip extraction, naming policy / coordinator, people-review policy, profile merging + provenance, retroactive transcript updates, negative-exemplar policy/store, write-path policy, and the recognition lifeline: match-outcome store, profile-health demotion, and review prioritization (see `docs/speaker-recognition-metrics.md`) - `Stats/` (4 files) — recording stats database, models, queries, and service @@ -23,6 +23,7 @@ - `ModelBundleProvider` — lets hosts override where offline model bundles are resolved - `AppServices` — DI container over protocol-typed STT / diarization / speaker-store dependencies - `TranscriptionTaskManager` — host-facing queue and orchestration surface, including imported-audio jobs and optional local-speaker mic diarization when the app asks for it +- `ImportedTranscriptionRecoverySession` — keeps app-owned imported-audio recovery alive across Core transcript commit and deferred speaker-review scratch cleanup. Core must durably prepare cleanup before deleting scratch, and reports completion only after deletion, without an untyped terminal callback or parallel ownership map - `TranscriptNotifier` — optional callback channel for transcript-saved / failure notifications These seams exist specifically so the app can embed the library without adopting the old standalone Transcripted app assumptions. diff --git a/Sources/TranscriptedCore/Models/TranscriptionTypes.swift b/Sources/TranscriptedCore/Models/TranscriptionTypes.swift index cdd8bed93..00d521b26 100644 --- a/Sources/TranscriptedCore/Models/TranscriptionTypes.swift +++ b/Sources/TranscriptedCore/Models/TranscriptionTypes.swift @@ -235,6 +235,7 @@ public struct SpeakerNamingRequest { public let micAudioURL: URL? public let shouldRemoveTemporaryAudioOnCleanup: Bool public let sourceFailedTranscriptionId: UUID? + public let importedRecoverySession: (any ImportedTranscriptionRecoverySession)? public let onComplete: ([SpeakerNameUpdate]) -> Void public init( @@ -247,6 +248,7 @@ public struct SpeakerNamingRequest { micAudioURL: URL?, shouldRemoveTemporaryAudioOnCleanup: Bool = true, sourceFailedTranscriptionId: UUID? = nil, + importedRecoverySession: (any ImportedTranscriptionRecoverySession)? = nil, onComplete: @escaping ([SpeakerNameUpdate]) -> Void ) { self.speakers = speakers @@ -258,6 +260,7 @@ public struct SpeakerNamingRequest { self.micAudioURL = micAudioURL self.shouldRemoveTemporaryAudioOnCleanup = shouldRemoveTemporaryAudioOnCleanup self.sourceFailedTranscriptionId = sourceFailedTranscriptionId + self.importedRecoverySession = importedRecoverySession self.onComplete = onComplete } diff --git a/Sources/TranscriptedCore/Pipeline/TranscriptionPipelineRunner.swift b/Sources/TranscriptedCore/Pipeline/TranscriptionPipelineRunner.swift index 9e599af17..bd4308fc2 100644 --- a/Sources/TranscriptedCore/Pipeline/TranscriptionPipelineRunner.swift +++ b/Sources/TranscriptedCore/Pipeline/TranscriptionPipelineRunner.swift @@ -254,7 +254,8 @@ extension TranscriptionTaskManager { healthInfo: nil, splitLocalSpeakers: false, meetingTitle: meetingTitle, - recordingDate: recordingDate + recordingDate: recordingDate, + stableTranscriptId: taskId ) } @@ -274,7 +275,8 @@ extension TranscriptionTaskManager { sourceFailedTranscriptionId: UUID? = nil, removeSourceAudioAfterArchive: Bool = true, targetTranscriptURL: URL? = nil, - archiveRecordingAudio: Bool = true + archiveRecordingAudio: Bool = true, + stableTranscriptId: UUID? = nil ) async throws -> URL { let transcription = await MainActor.run { self.transcription } @@ -513,7 +515,9 @@ extension TranscriptionTaskManager { } let replacementTranscriptRollback = try ReplacementTranscriptRollback.capture(for: targetTranscriptURL) - let transcriptId = replacementTranscriptRollback?.transcriptId ?? UUID() + let transcriptId = replacementTranscriptRollback?.transcriptId + ?? stableTranscriptId + ?? UUID() try Task.checkCancellation() // Phase 2: Save transcript with speaker names let notifier: TranscriptNotifier? = await MainActor.run { @@ -725,6 +729,9 @@ extension TranscriptionTaskManager { replacementTranscriptRollback: replacementTranscriptRollback ) await MainActor.run { + let importedRecoverySession = self.importedRecoverySession( + taskId: taskId + ) self.enqueueSpeakerNamingRequest(SpeakerNamingRequest( speakers: capturedEntries, knownPeople: knownPeople, @@ -735,6 +742,7 @@ extension TranscriptionTaskManager { micAudioURL: micURL, shouldRemoveTemporaryAudioOnCleanup: shouldRemoveScratchAudio && sourceFailedTranscriptionId == nil, sourceFailedTranscriptionId: sourceFailedTranscriptionId, + importedRecoverySession: importedRecoverySession, onComplete: { [weak self] updates in self?.handleNamingComplete( updates: updates, @@ -745,7 +753,8 @@ extension TranscriptionTaskManager { systemURL: systemURL, shouldRemoveTemporaryAudio: shouldRemoveScratchAudio, sourceFailedTranscriptionId: sourceFailedTranscriptionId, - clips: capturedEntries + clips: capturedEntries, + importedRecoverySession: importedRecoverySession ) } )) @@ -803,8 +812,18 @@ extension TranscriptionTaskManager { // committed, so a late cancellation cannot delete transcript + retained // audio after scratch files are gone. if shouldRemoveScratchAudio, sourceFailedTranscriptionId == nil { - removeManagedCleanupFile(micURL, label: "completed mic scratch") - removeManagedCleanupFile(systemURL, label: "completed system scratch") + let cleanupPrepared = await MainActor.run { + self.prepareImportedTranscriptionScratchCleanup(taskId: taskId) + } + if cleanupPrepared { + let removedMic = removeManagedCleanupFile(micURL, label: "completed mic scratch") + let removedSystem = removeManagedCleanupFile(systemURL, label: "completed system scratch") + if removedMic && removedSystem { + await MainActor.run { + self.confirmImportedTranscriptionScratchCleanup(taskId: taskId) + } + } + } } return savedURL diff --git a/Sources/TranscriptedCore/Pipeline/TranscriptionTaskManager.swift b/Sources/TranscriptedCore/Pipeline/TranscriptionTaskManager.swift index b8e62ef66..e88122c3e 100644 --- a/Sources/TranscriptedCore/Pipeline/TranscriptionTaskManager.swift +++ b/Sources/TranscriptedCore/Pipeline/TranscriptionTaskManager.swift @@ -8,6 +8,14 @@ import AVFoundation @available(macOS 14.0, *) @MainActor public class TranscriptionTaskManager: ObservableObject { + private struct ActiveTaskAudio { + let micURL: URL? + let systemURL: URL? + let meetingTitle: String? + let recordingDate: Date? + let importedRecoverySession: (any ImportedTranscriptionRecoverySession)? + } + @Published public var activeCount: Int = 0 @Published public var justCompleted: Bool = false @Published public var displayStatus: DisplayStatus = .idle @@ -24,7 +32,7 @@ public class TranscriptionTaskManager: ObservableObject { private var savedTranscriptTaskIdsByTranscriptId: [UUID: UUID] = [:] private var savedTranscriptTaskIdsByURL: [URL: UUID] = [:] var activeTasks: [UUID: Task] = [:] - private var activeTaskAudio: [UUID: (micURL: URL?, systemURL: URL?, meetingTitle: String?, recordingDate: Date?)] = [:] + private var activeTaskAudio: [UUID: ActiveTaskAudio] = [:] private var preservedTaskIdsForShutdown: Set = [] private var intentionallyCancelledTaskIds: Set = [] private var committedTranscriptTaskIds: Set = [] @@ -185,11 +193,12 @@ public class TranscriptionTaskManager: ObservableObject { activeCount += 1 backgroundTaskCount += 1 - activeTaskAudio[task.id] = ( + activeTaskAudio[task.id] = ActiveTaskAudio( micURL: micURL, systemURL: systemURL, meetingTitle: meetingTitle, - recordingDate: recordingDate + recordingDate: recordingDate, + importedRecoverySession: nil ) publishNonFailureStatus(.gettingReady) @@ -327,14 +336,26 @@ public class TranscriptionTaskManager: ObservableObject { /// Imported files reuse the system-audio speaker path and are not added to the /// failed-transcription queue because the user can simply re-import the source file. public func startImportedTranscription( + taskId: UUID = UUID(), audioURL: URL, outputFolder: URL, meetingTitle: String? = nil, - recordingDate: Date? = nil + recordingDate: Date? = nil, + recoverySession: (any ImportedTranscriptionRecoverySession)? = nil ) { + precondition( + recoverySession == nil || recoverySession?.jobID == taskId, + "Imported recovery session must use the queued job identity" + ) if !activeTasks.isEmpty { AppLogger.pipeline.warning("Rejecting imported transcription — another pipeline is already active", ["activeCount": "\(activeTasks.count)"]) - removeRecordingFile(audioURL, label: "rejected imported recording") + if removeImportedRecordingFile( + audioURL, + recoverySession: recoverySession, + label: "rejected imported recording" + ) { + recoverySession?.scratchCleanupConfirmed() + } publishFailure( displayMessage: "Another transcript is already running. Wait for it to finish, then import the file again.", diagnosticMessage: "Transcription already in progress" @@ -346,7 +367,13 @@ public class TranscriptionTaskManager: ObservableObject { let minDuration: TimeInterval = 2.0 if let audioDuration = audioDuration(url: audioURL), audioDuration < minDuration { AppLogger.pipeline.info("Imported recording too short, skipping transcription", ["duration": String(format: "%.1fs", audioDuration)]) - removeRecordingFile(audioURL, label: "short imported recording") + if removeImportedRecordingFile( + audioURL, + recoverySession: recoverySession, + label: "short imported recording" + ) { + recoverySession?.scratchCleanupConfirmed() + } publishFailure( displayMessage: "That audio file is too short to transcribe. Choose audio that is at least two seconds long.", diagnosticMessage: "Recording too short" @@ -355,14 +382,14 @@ public class TranscriptionTaskManager: ObservableObject { return } - let taskId = UUID() activeCount += 1 backgroundTaskCount += 1 - activeTaskAudio[taskId] = ( + activeTaskAudio[taskId] = ActiveTaskAudio( micURL: audioURL, systemURL: nil, meetingTitle: meetingTitle, - recordingDate: recordingDate + recordingDate: recordingDate, + importedRecoverySession: recoverySession ) publishNonFailureStatus(.gettingReady) @@ -402,10 +429,22 @@ public class TranscriptionTaskManager: ObservableObject { return } if self.finishCancelledTaskIfNeeded(taskId: taskId, error: error) { - self.removeRecordingFile(audioURL, label: "cancelled imported recording") + if self.removeImportedRecordingFile( + audioURL, + recoverySession: recoverySession, + label: "cancelled imported recording" + ) { + recoverySession?.scratchCleanupConfirmed() + } return } - self.removeRecordingFile(audioURL, label: "failed imported recording") + if self.removeImportedRecordingFile( + audioURL, + recoverySession: recoverySession, + label: "failed imported recording" + ) { + recoverySession?.scratchCleanupConfirmed() + } let diagnosticMessage = Self.safeFailureDiagnosticMessage(for: error) self.publishFailure( displayMessage: Self.importedAudioFailureDisplayMessage(forDiagnosticMessage: diagnosticMessage), @@ -1604,8 +1643,9 @@ public class TranscriptionTaskManager: ObservableObject { func markTaskTranscriptCommitted(taskId: UUID) { committedTranscriptTaskIds.insert(taskId) - // Transcript is durably on disk — the crash-recovery journal for this - // recording's scratch audio is no longer needed. + activeTaskAudio[taskId]?.importedRecoverySession?.transcriptCommitConfirmed() + // The imported journal remains through scratch cleanup. The separate + // live-recording journal can retire once the transcript is durable. if let micURL = activeTaskAudio[taskId]?.micURL { MeetingRecordingJournalStore.removeJournal( forMicAudioURL: micURL, @@ -1652,8 +1692,13 @@ public class TranscriptionTaskManager: ObservableObject { intentionallyCancelledTaskIds.insert(taskId) task.cancel() if let audio = activeTaskAudio[taskId] { - removeManagedCleanupFile(audio.micURL, label: "cancelled live mic scratch") - removeManagedCleanupFile(audio.systemURL, label: "cancelled live system scratch") + if audio.importedRecoverySession?.prepareForScratchCleanup() != false { + let removedMic = removeManagedCleanupFile(audio.micURL, label: "cancelled live mic scratch") + let removedSystem = removeManagedCleanupFile(audio.systemURL, label: "cancelled live system scratch") + if removedMic && removedSystem { + audio.importedRecoverySession?.scratchCleanupConfirmed() + } + } } activeTaskAudio.removeValue(forKey: taskId) AppLogger.pipeline.info("Cancelled task", ["taskId": "\(taskId)"]) @@ -1691,15 +1736,17 @@ public class TranscriptionTaskManager: ObservableObject { var preservedCount = 0 for (taskId, audio) in activeAudio { - if addFailedTranscriptionRetainingAvailableAudio( + let didPersist = addFailedTranscriptionRetainingAvailableAudio( micAudioURL: audio.micURL, systemAudioURL: audio.systemURL, errorMessage: errorMessage, taskId: taskId, meetingTitle: audio.meetingTitle, recordingDate: audio.recordingDate - ) { + ) + if didPersist { preservedCount += 1 + audio.importedRecoverySession?.failedQueueHandoffConfirmed() } } return preservedCount @@ -1806,29 +1853,45 @@ public class TranscriptionTaskManager: ObservableObject { notifier.notifyTranscriptionFailed(errorMessage: errorMessage) } - nonisolated private func removeRecordingFile(_ url: URL, label: String) { + @discardableResult + nonisolated private func removeRecordingFile(_ url: URL, label: String) -> Bool { // Security: only delete scratch files inside Transcripted-managed cleanup roots. // `startImportedTranscription` accepts a URL from the caller, so without a containment // check a misuse or tampered in-memory request could unlink arbitrary user files. guard isSafeCleanupURL(url) else { AppLogger.pipeline.error("Refused to delete out-of-sandbox recording file", [ "label": label, - "path": url.path + "file": url.lastPathComponent ]) - return + return false } do { try FileManager.default.removeItem(at: url) + return true } catch { + if (error as NSError).code == NSFileNoSuchFileError { + return true + } AppLogger.pipeline.warning("Failed to remove recording file", [ "label": label, "file": url.lastPathComponent, "error": error.localizedDescription ]) + return false } } + @discardableResult + nonisolated private func removeImportedRecordingFile( + _ url: URL, + recoverySession: (any ImportedTranscriptionRecoverySession)?, + label: String + ) -> Bool { + guard recoverySession?.prepareForScratchCleanup() != false else { return false } + return removeRecordingFile(url, label: label) + } + func resolvedRetainedAudioDirectory() -> URL? { retainedAudioDirectoryProvider?() ?? retainedAudioDirectory } @@ -1845,9 +1908,22 @@ public class TranscriptionTaskManager: ObservableObject { .withAudioSources(audioSources) } - nonisolated func removeManagedCleanupFile(_ url: URL?, label: String) { - guard let url else { return } - removeRecordingFile(url, label: label) + @discardableResult + nonisolated func removeManagedCleanupFile(_ url: URL?, label: String) -> Bool { + guard let url else { return true } + return removeRecordingFile(url, label: label) + } + + func confirmImportedTranscriptionScratchCleanup(taskId: UUID) { + activeTaskAudio[taskId]?.importedRecoverySession?.scratchCleanupConfirmed() + } + + func prepareImportedTranscriptionScratchCleanup(taskId: UUID) -> Bool { + activeTaskAudio[taskId]?.importedRecoverySession?.prepareForScratchCleanup() ?? true + } + + func importedRecoverySession(taskId: UUID) -> (any ImportedTranscriptionRecoverySession)? { + activeTaskAudio[taskId]?.importedRecoverySession } nonisolated private func isSafeCleanupURL(_ url: URL) -> Bool { diff --git a/Sources/TranscriptedCore/Protocols/ImportedTranscriptionRecoverySession.swift b/Sources/TranscriptedCore/Protocols/ImportedTranscriptionRecoverySession.swift new file mode 100644 index 000000000..e3afa14ee --- /dev/null +++ b/Sources/TranscriptedCore/Protocols/ImportedTranscriptionRecoverySession.swift @@ -0,0 +1,26 @@ +import Foundation + +/// Durable ownership boundary for one imported-audio transcription. +/// +/// The app layer owns the journal and cross-process lease. TranscriptedCore +/// reports only the three transitions that make recovery safe to retire or +/// advance; it never decides where or how the journal is stored. +public protocol ImportedTranscriptionRecoverySession: AnyObject, Sendable { + var jobID: UUID { get } + + /// The transcript and its committed side effects are durable. Scratch audio + /// may still be needed for speaker review or cleanup, so this does not retire + /// the recovery record. + func transcriptCommitConfirmed() + + /// Durably records that scratch may now be deleted. Returns false when the + /// recovery record could not be advanced, in which case callers must retain + /// the audio. + func prepareForScratchCleanup() -> Bool + + /// The app-owned scratch entry is gone (or was already absent). + func scratchCleanupConfirmed() + + /// A durable failed-queue row now owns the audio. + func failedQueueHandoffConfirmed() +} diff --git a/Sources/TranscriptedCore/Speaker/SpeakerNamingCoordinator.swift b/Sources/TranscriptedCore/Speaker/SpeakerNamingCoordinator.swift index 77ed455b4..84812a7f5 100644 --- a/Sources/TranscriptedCore/Speaker/SpeakerNamingCoordinator.swift +++ b/Sources/TranscriptedCore/Speaker/SpeakerNamingCoordinator.swift @@ -47,7 +47,8 @@ extension TranscriptionTaskManager { systemURL: URL, shouldRemoveTemporaryAudio: Bool = true, sourceFailedTranscriptionId: UUID? = nil, - clips: [SpeakerNamingEntry] + clips: [SpeakerNamingEntry], + importedRecoverySession: (any ImportedTranscriptionRecoverySession)? = nil ) { let speakerDB = transcription.speakerDB let clipsDirectory = transcription.speakerClipsDirectory @@ -92,7 +93,8 @@ extension TranscriptionTaskManager { clips: clips, micURL: micURL, systemURL: systemURL, - shouldRemoveTemporaryAudio: shouldRemoveTemporaryAudio && sourceFailedTranscriptionId == nil + shouldRemoveTemporaryAudio: shouldRemoveTemporaryAudio && sourceFailedTranscriptionId == nil, + importedRecoverySession: importedRecoverySession ) Task { @MainActor in @@ -232,7 +234,8 @@ extension TranscriptionTaskManager { clips: clips, micURL: micURL, systemURL: systemURL, - shouldRemoveTemporaryAudio: shouldRemoveTemporaryAudio && !shouldKeepAudioForFailedRetry + shouldRemoveTemporaryAudio: shouldRemoveTemporaryAudio && !shouldKeepAudioForFailedRetry, + importedRecoverySession: importedRecoverySession ) } @@ -254,7 +257,8 @@ extension TranscriptionTaskManager { clips: clips, micURL: micURL, systemURL: systemURL, - shouldRemoveTemporaryAudio: shouldRemoveTemporaryAudio + shouldRemoveTemporaryAudio: shouldRemoveTemporaryAudio, + importedRecoverySession: importedRecoverySession ) case .metadataPublicationFailed: // The speaker clips are no longer needed, but retry audio must remain @@ -347,8 +351,13 @@ extension TranscriptionTaskManager { private func cleanupSpeakerNamingRequest(_ request: SpeakerNamingRequest) { if request.shouldRemoveTemporaryAudioOnCleanup { - removeManagedCleanupFile(request.micAudioURL, label: "pending mic audio") - removeManagedCleanupFile(request.systemAudioURL, label: "pending system audio") + if request.importedRecoverySession?.prepareForScratchCleanup() != false { + let removedMic = removeManagedCleanupFile(request.micAudioURL, label: "pending mic audio") + let removedSystem = removeManagedCleanupFile(request.systemAudioURL, label: "pending system audio") + if removedMic && removedSystem { + request.importedRecoverySession?.scratchCleanupConfirmed() + } + } } cleanupSpeakerClips(request.speakers) } @@ -390,12 +399,17 @@ extension TranscriptionTaskManager { clips: [SpeakerNamingEntry], micURL: URL?, systemURL: URL, - shouldRemoveTemporaryAudio: Bool + shouldRemoveTemporaryAudio: Bool, + importedRecoverySession: (any ImportedTranscriptionRecoverySession)? = nil ) { cleanupSpeakerClips(clips) guard shouldRemoveTemporaryAudio else { return } - removeManagedCleanupFile(micURL, label: "mic audio") - removeManagedCleanupFile(systemURL, label: "system audio") + guard importedRecoverySession?.prepareForScratchCleanup() != false else { return } + let removedMic = removeManagedCleanupFile(micURL, label: "mic audio") + let removedSystem = removeManagedCleanupFile(systemURL, label: "system audio") + if removedMic && removedSystem { + importedRecoverySession?.scratchCleanupConfirmed() + } } nonisolated private func cleanupSpeakerClips(_ clips: [SpeakerNamingEntry]) { diff --git a/Sources/TranscriptedCore/Storage/TranscriptSaver.swift b/Sources/TranscriptedCore/Storage/TranscriptSaver.swift index e9d8b46b8..5826e26e9 100644 --- a/Sources/TranscriptedCore/Storage/TranscriptSaver.swift +++ b/Sources/TranscriptedCore/Storage/TranscriptSaver.swift @@ -58,6 +58,53 @@ public class TranscriptSaver { return try fileUpdateQueue.sync(execute: update) } + /// Finds an already-written transcript by its stable frontmatter identity. + /// Recovery uses this to make a crash immediately after the atomic Markdown + /// write idempotent instead of producing a second timestamp-suffixed file. + public static func existingTranscriptURL( + in directory: URL, + transcriptId: UUID, + fileManager: FileManager = .default + ) -> URL? { + existingTranscriptURLs( + in: directory, + transcriptIds: [transcriptId], + fileManager: fileManager + )[transcriptId] + } + + /// Resolves a recovery batch in one bounded directory scan so many stale + /// journals cannot multiply transcript-directory I/O. + public static func existingTranscriptURLs( + in directory: URL, + transcriptIds: Set, + fileManager: FileManager = .default + ) -> [UUID: URL] { + guard !transcriptIds.isEmpty else { return [:] } + guard let files = try? fileManager.contentsOfDirectory( + at: directory, + includingPropertiesForKeys: [.isRegularFileKey, .isSymbolicLinkKey], + options: [.skipsHiddenFiles] + ) else { return [:] } + + var matches: [UUID: URL] = [:] + let markdownFiles = files + .filter { $0.pathExtension.lowercased() == "md" } + .sorted { $0.lastPathComponent < $1.lastPathComponent } + for url in markdownFiles { + guard let values = try? url.resourceValues(forKeys: [.isRegularFileKey, .isSymbolicLinkKey]), + values.isRegularFile == true, + values.isSymbolicLink != true, + let frontmatter = try? TranscriptFrontmatter.readValues(from: url), + let transcriptId = TranscriptFrontmatter.captureID(in: frontmatter), + transcriptIds.contains(transcriptId), + matches[transcriptId] == nil + else { continue } + matches[transcriptId] = url + } + return matches + } + /// Save transcript to file with automatic timestamped naming /// - Parameters: /// - text: The transcript text to save diff --git a/Tests/MeetingImportedAudioPreparerTests.swift b/Tests/MeetingImportedAudioPreparerTests.swift index 75110246a..18f43472c 100644 --- a/Tests/MeetingImportedAudioPreparerTests.swift +++ b/Tests/MeetingImportedAudioPreparerTests.swift @@ -510,7 +510,6 @@ func testMeetingImportedAudioPreparer() async { try! ImportedTranscriptionQueueJournal.persist( id: id, audioURL: audioURL, - suggestedTitle: "Customer Call", recordingDate: recordingDate, enqueuedAt: recordingDate.addingTimeInterval(10), sttModelRawValue: "parakeet", @@ -521,7 +520,18 @@ func testMeetingImportedAudioPreparer() async { let recovered = ImportedTranscriptionQueueJournal.load(journalDirectory: journalURL) assertEqual(recovered.count, 1, "a fresh journal reader should recover the accepted import") assertEqual(recovered.first?.id, id, "recovery should preserve the queued job identity") + assertEqual(recovered.first?.phase, .queued, "a newly accepted import should begin in the queued phase") + assertNil(recovered.first?.owner, "a queued journal should not claim a live process owner") assertEqual(recovered.first?.audioFilename, audioURL.lastPathComponent, "journals should store only the app-owned scratch filename") + let journalFile = try! FileManager.default.contentsOfDirectory( + at: journalURL, + includingPropertiesForKeys: nil + ).first { $0.pathExtension == "json" }! + let journalData = try! Data(contentsOf: journalFile) + assertFalse( + journalData.range(of: Data("suggestedTitle".utf8)) != nil, + "recovery journals must not persist an imported title" + ) assertEqual( recovered.first.flatMap { ImportedTranscriptionQueueJournal.audioURL(for: $0, scratchDirectory: scratchURL) @@ -550,7 +560,6 @@ func testMeetingImportedAudioPreparer() async { let unsafeRecord = ImportedTranscriptionQueueJournalRecord( id: UUID(), audioFilename: "../outside.wav", - suggestedTitle: "Unsafe", recordingDate: recordingDate, enqueuedAt: recordingDate, sttModelRawValue: "parakeet" @@ -569,7 +578,6 @@ func testMeetingImportedAudioPreparer() async { try! ImportedTranscriptionQueueJournal.persist( id: olderSourceNewerQueueID, audioURL: olderSourceNewerQueueAudioURL, - suggestedTitle: "Older Source Imported Second", recordingDate: recordingDate.addingTimeInterval(-86_400), enqueuedAt: recordingDate.addingTimeInterval(20), sttModelRawValue: "parakeet", @@ -590,6 +598,228 @@ func testMeetingImportedAudioPreparer() async { ) } + runSuite("Imported transcription journal lease excludes a second live process and survives owner death") { + let root = temporaryImportAudioPreparerRoot() + let scratchURL = root.appendingPathComponent("scratch", isDirectory: true) + let journalURL = root.appendingPathComponent("state/queue", isDirectory: true) + let audioURL = scratchURL.appendingPathComponent("lease.wav") + let id = UUID() + try! FileManager.default.createDirectory(at: scratchURL, withIntermediateDirectories: true) + try! Data([0, 1, 2, 3]).write(to: audioURL) + var firstOwner: ImportedTranscriptionQueueJournalSession? = try! ImportedTranscriptionQueueJournal.createClaimed( + id: id, + audioURL: audioURL, + recordingDate: Date(timeIntervalSince1970: 1_704_067_200), + sttModelRawValue: "parakeet", + journalDirectory: journalURL, + scratchDirectory: scratchURL, + processIdentifier: 101 + ) + assertNotNil(firstOwner, "the first process should claim the queued import") + assertEqual(firstOwner?.phase, .active, "claiming should durably mark the import active") + assertEqual( + ImportedTranscriptionQueueJournal.load(journalDirectory: journalURL).first?.owner?.processIdentifier, + 101, + "initial journal publication should already contain the live lease owner" + ) + let lockURL = journalURL.appendingPathComponent( + "import-job-\(id.uuidString).lock", + isDirectory: false + ) + assertEqual( + importedJournalLeaseProbe(lockURL: lockURL, expectLocked: true), + 0, + "a separate process must be unable to take a live owner's lease" + ) + + firstOwner = nil + assertEqual( + importedJournalLeaseProbe(lockURL: lockURL, expectLocked: false), + 0, + "a separate process should take the lease after the prior owner exits" + ) + let recoveredOwner = try! ImportedTranscriptionQueueJournal.claim( + id: id, + journalDirectory: journalURL, + processIdentifier: 202 + ) + assertNotNil(recoveredOwner, "the kernel lease should release automatically when the prior owner dies") + try! FileManager.default.removeItem(at: audioURL) + recoveredOwner?.scratchCleanupConfirmed() + assertTrue( + ImportedTranscriptionQueueJournal.load(journalDirectory: journalURL).isEmpty, + "confirmed scratch cleanup should retire the journal" + ) + assertFalse( + FileManager.default.fileExists(atPath: lockURL.path), + "terminal journal cleanup should remove its no-longer-needed lease file" + ) + } + + runSuite("Imported transcription committed phase survives a crash until scratch cleanup") { + let root = temporaryImportAudioPreparerRoot() + let scratchURL = root.appendingPathComponent("scratch", isDirectory: true) + let journalURL = root.appendingPathComponent("state/queue", isDirectory: true) + let audioURL = scratchURL.appendingPathComponent("committed.wav") + let id = UUID() + try! FileManager.default.createDirectory(at: scratchURL, withIntermediateDirectories: true) + try! Data([0, 1, 2, 3]).write(to: audioURL) + try! ImportedTranscriptionQueueJournal.persist( + id: id, + audioURL: audioURL, + recordingDate: Date(timeIntervalSince1970: 1_704_067_200), + sttModelRawValue: "parakeet", + journalDirectory: journalURL, + scratchDirectory: scratchURL + ) + + var owner = try! ImportedTranscriptionQueueJournal.claim(id: id, journalDirectory: journalURL) + owner?.transcriptCommitConfirmed() + assertEqual( + ImportedTranscriptionQueueJournal.load(journalDirectory: journalURL).first?.phase, + .transcriptCommitted, + "transcript commit should advance the journal without retiring scratch ownership" + ) + owner = nil + + var cleanupOwner = try! ImportedTranscriptionQueueJournal.claim(id: id, journalDirectory: journalURL) + assertEqual(cleanupOwner?.phase, .transcriptCommitted, "relaunch should preserve the committed cleanup phase") + assertEqual( + ImportedTranscriptionQueueJournal.recoveryAction( + phase: cleanupOwner?.phase ?? .active, + stableTranscriptExists: false + ), + .handOffScratch, + "transcript commit alone must hand off audio intentionally retained after an archive failure" + ) + assertTrue( + cleanupOwner?.prepareForScratchCleanup() == true, + "cleanup intent should become durable before scratch deletion" + ) + assertEqual( + ImportedTranscriptionQueueJournal.load(journalDirectory: journalURL).first?.phase, + .scratchCleanupPending, + "cleanup authorization should survive a crash before file removal" + ) + cleanupOwner = nil + let resumedCleanupOwner = try! ImportedTranscriptionQueueJournal.claim( + id: id, + journalDirectory: journalURL + ) + assertEqual( + resumedCleanupOwner?.phase, + .scratchCleanupPending, + "a new process should resume the authorized cleanup phase without replaying transcription" + ) + try! FileManager.default.removeItem(at: audioURL) + resumedCleanupOwner?.scratchCleanupConfirmed() + assertTrue( + ImportedTranscriptionQueueJournal.load(journalDirectory: journalURL).isEmpty, + "the committed journal should retire only after scratch is gone" + ) + } + + runSuite("Imported transcription recovery resolves both transcript crash boundaries") { + assertEqual( + ImportedTranscriptionQueueJournal.recoveryAction( + phase: .active, + stableTranscriptExists: true + ), + .handOffScratch, + "a stable transcript identity must hand off uncertain scratch after the Markdown write but before the phase marker" + ) + assertEqual( + ImportedTranscriptionQueueJournal.recoveryAction( + phase: .transcriptCommitted, + stableTranscriptExists: false + ), + .handOffScratch, + "a committed record without cleanup authorization must receive durable visible ownership" + ) + assertEqual( + ImportedTranscriptionQueueJournal.recoveryAction( + phase: .active, + stableTranscriptExists: false + ), + .replayTranscription, + "recovery should replay only when neither commit proof exists" + ) + assertEqual( + ImportedTranscriptionQueueJournal.recoveryAction( + phase: .scratchCleanupPending, + stableTranscriptExists: false + ), + .cleanScratch, + "recovery should finish an authorized cleanup idempotently" + ) + } + + runSuite("Imported transcription recovery rejects symlinks and non-regular scratch entries") { + let root = temporaryImportAudioPreparerRoot() + let scratchURL = root.appendingPathComponent("scratch", isDirectory: true) + let journalURL = root.appendingPathComponent("state/queue", isDirectory: true) + let outsideURL = root.appendingPathComponent("outside.wav") + let symlinkURL = scratchURL.appendingPathComponent("linked.wav") + let directoryURL = scratchURL.appendingPathComponent("folder.wav", isDirectory: true) + try! FileManager.default.createDirectory(at: scratchURL, withIntermediateDirectories: true) + try! Data([0, 1, 2, 3]).write(to: outsideURL) + try! FileManager.default.createSymbolicLink(at: symlinkURL, withDestinationURL: outsideURL) + try! FileManager.default.createDirectory(at: directoryURL, withIntermediateDirectories: true) + + assertEqual( + ImportedTranscriptionQueueJournal.recoveryAudioStatus( + at: symlinkURL, + scratchDirectory: scratchURL + ), + .unsafeEntry, + "recovery must not follow a scratch symlink" + ) + assertEqual( + ImportedTranscriptionQueueJournal.recoveryAudioStatus( + at: directoryURL, + scratchDirectory: scratchURL + ), + .unsafeEntry, + "recovery must not transcribe a non-regular scratch entry" + ) + assertEqual( + ImportedTranscriptionQueueJournal.recoveryAudioStatus( + at: outsideURL, + scratchDirectory: scratchURL + ), + .unsafeEntry, + "recovery input must remain directly inside app-owned scratch" + ) + + let jobId = UUID() + let regularURL = scratchURL.appendingPathComponent("regular.wav") + try! Data([4, 5, 6, 7]).write(to: regularURL) + try! ImportedTranscriptionQueueJournal.persist( + id: jobId, + audioURL: regularURL, + recordingDate: Date(timeIntervalSince1970: 1_704_067_200), + sttModelRawValue: "parakeet", + journalDirectory: journalURL, + scratchDirectory: scratchURL + ) + let outsideLockURL = root.appendingPathComponent("outside.lock") + try! Data().write(to: outsideLockURL) + try! FileManager.default.createSymbolicLink( + at: journalURL.appendingPathComponent("import-job-\(jobId.uuidString).lock"), + withDestinationURL: outsideLockURL + ) + var rejectedSymlinkLock = false + do { + _ = try ImportedTranscriptionQueueJournal.claim( + id: jobId, + journalDirectory: journalURL + ) + } catch { + rejectedSymlinkLock = true + } + assertTrue(rejectedSymlinkLock, "recovery must not follow a symlinked lease file") + } + runSuite("Imported queue persistence failure copy only claims durable retry ownership when persisted") { assertTrue( ImportedAudioQueuePersistenceFailureCopy.displayMessage(preservedForRelaunch: true) @@ -610,6 +840,33 @@ func testMeetingImportedAudioPreparer() async { } } +private func importedJournalLeaseProbe(lockURL: URL, expectLocked: Bool) -> Int32 { + let script = """ + import fcntl, os, sys + descriptor = os.open(sys.argv[1], os.O_RDWR) + observed_locked = False + try: + fcntl.flock(descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError: + observed_locked = True + expected_locked = sys.argv[2] == "locked" + os.close(descriptor) + sys.exit(0 if observed_locked == expected_locked else 92) + """ + let process = Process() + process.executableURL = URL(fileURLWithPath: "/usr/bin/python3") + process.arguments = ["-c", script, lockURL.path, expectLocked ? "locked" : "available"] + process.standardOutput = FileHandle.nullDevice + process.standardError = FileHandle.nullDevice + do { + try process.run() + process.waitUntilExit() + return process.terminationStatus + } catch { + return 90 + } +} + private func assertImportedAudioPreparationError( _ expected: MeetingImportedAudioPreparationError, sourceURL: URL, diff --git a/Tests/RepoCommandContractTests.swift b/Tests/RepoCommandContractTests.swift index 97b903077..39f760b1a 100644 --- a/Tests/RepoCommandContractTests.swift +++ b/Tests/RepoCommandContractTests.swift @@ -453,6 +453,7 @@ func testRepoCommandContract() { runSuite("Repo command contract - script edits map to syntax and owned checks") { let matrix = readRepoTextFile(".agents/test-matrix.yml") + let homeBenchmark = readRepoTextFile("scripts/dev/benchmark-home-recent-captures.sh") let expectedChecks = [ "bash -n scripts/entrypoints/build-deps.sh", "bash build-deps.sh --force", @@ -484,6 +485,11 @@ func testRepoCommandContract() { for check in expectedChecks { assertTrue(matrix.contains(check), "test matrix should include \(check)") } + assertTrue( + homeBenchmark.contains("Sources/TranscriptedCore/Protocols/ImportedTranscriptionRecoverySession.swift") + && homeBenchmark.contains("Sources/TranscriptedCore/Models/TranscriptionTypes.swift"), + "the raw-swiftc Home benchmark should include the recovery protocol used by transcription types" + ) } runSuite("Repo command contract - agent preflight executes the canonical matrix") { @@ -2284,12 +2290,13 @@ func testRepoCommandContract() { to: "private func enqueue(_ job: QueuedTranscriptionJob)" ) assertTrue( - importedEnqueueBlock.contains("try persistImportedJournal(for: job)") + importedEnqueueBlock.contains("ImportedTranscriptionQueueJournal.createClaimed(") + && importedEnqueueBlock.contains("job.importedRecoverySession = session") && importedEnqueueBlock.contains("queuedTranscriptionJobs.append(job)"), - "an accepted imported job should be journaled before it becomes crash-volatile queue state" + "an accepted imported job should publish its journal while already leased before it becomes crash-volatile queue state" ) let journalPersistIndex = importedEnqueueBlock.range( - of: "try persistImportedJournal(for: job)" + of: "ImportedTranscriptionQueueJournal.createClaimed(" )?.lowerBound ?? importedEnqueueBlock.endIndex let immediateStartCheckIndex = importedEnqueueBlock.range( of: "if canStartQueuedTranscriptionImmediately" @@ -2305,20 +2312,41 @@ func testRepoCommandContract() { let importedRecoveryBlock = sourceSlice( coordinatorContents, from: "func recoverImportedAudioJobs() -> Int {", - to: "func removeImportedJournal(for job: QueuedTranscriptionJob)" + to: "func confirmImportedScratchCleanup(for job: QueuedTranscriptionJob)" + ) + let recoveryClaimIndex = importedRecoveryBlock.range( + of: "let claimedRecords = records.compactMap" + )?.lowerBound ?? importedRecoveryBlock.endIndex + let transcriptScanIndex = importedRecoveryBlock.range( + of: "let existingTranscriptsByID = TranscriptSaver.existingTranscriptURLs(" + )?.lowerBound ?? importedRecoveryBlock.startIndex + assertTrue( + recoveryClaimIndex < transcriptScanIndex, + "recovery must hold every available job lease before its one batch transcript scan so commit evidence cannot race stale" ) assertTrue( importedRecoveryBlock.contains("controller.failedMeetingStore.failedAudioURLs") - && importedRecoveryBlock.contains("failedQueueAudioURLs.contains(audioURL.standardizedFileURL)"), + && importedRecoveryBlock.contains("failedQueueAudioURLs.contains(audioURL.standardizedFileURL)") + && importedRecoveryBlock.contains("recoverySession.failedQueueHandoffConfirmed()"), "relaunch recovery should use the failed-meeting store boundary and retire a journal whose audio already has durable failed-queue ownership" ) assertTrue( importedRecoveryBlock.contains("ImportedTranscriptionQueueJournal.isDuplicate") && importedRecoveryBlock.contains("recoveredJobIDs") && importedRecoveryBlock.contains("recoveredAudioURLs") - && importedRecoveryBlock.contains("ImportedTranscriptionQueueJournal.remove"), + && importedRecoveryBlock.contains("recoverySession.supersededRecoveryConfirmed()"), "relaunch recovery should retire duplicate journals instead of enqueueing the same import twice" ) + assertTrue( + importedRecoveryBlock.contains("recoveryAudioStatus(") + && importedRecoveryBlock.contains("guard audioStatus == .regularFile") + && coordinatorContents.contains("TranscriptSaver.existingTranscriptURLs(") + && importedRecoveryBlock.contains("recoveryAction(") + && importedRecoveryBlock.contains("case .cleanScratch:") + && importedRecoveryBlock.contains("case .handOffScratch:") + && importedRecoveryBlock.contains("preserveFailedMeetingForRetry("), + "relaunch recovery should reject unsafe inputs and resolve every state through replay, authorized cleanup, or durable visible handoff" + ) let importJournalFailureBlock = sourceSlice( controllerContents, from: "let preservedForRelaunch = failedMeetingStore.preserveFailedMeetingForRetry(", @@ -2341,15 +2369,12 @@ func testRepoCommandContract() { from: "func preserveQueuedTranscriptionJobsForShutdown(errorMessage: String) -> Int {", to: "return preservedCount" ) - let shutdownImportedBlock = sourceSlice( - shutdownQueuePreservationBlock, - from: "case .imported(let audioURL, let suggestedTitle, let recordingDate):", - to: "removeImportedJournal(for: job)" - ) assertTrue( - shutdownImportedBlock.contains("micAudioURL: nil") - && shutdownImportedBlock.contains("systemAudioURL: audioURL"), - "shutdown should preserve an imported single track with the same channel semantics as normal imported retry" + shutdownQueuePreservationBlock.contains("case .imported(let audioURL, let suggestedTitle, let recordingDate):") + && shutdownQueuePreservationBlock.contains("micAudioURL: nil") + && shutdownQueuePreservationBlock.contains("systemAudioURL: audioURL") + && shutdownQueuePreservationBlock.contains("job.importedRecoverySession?.failedQueueHandoffConfirmed()"), + "shutdown should retire an imported journal only after durable failed-queue ownership succeeds" ) let startQueuedBlock = sourceSlice( coordinatorContents, @@ -2834,6 +2859,20 @@ func testRepoCommandContract() { terminationBlock.contains("activeQueuedTranscriptionJobID = nil"), "shutdown preservation should clear the active queued job owner used for live sidecar final attachment" ) + assertTrue( + terminationBlock.contains("taskManager.cleanupPendingNaming()"), + "shutdown should finalize pending speaker-review scratch ownership before in-memory requests disappear" + ) + let activePreservationIndex = terminationBlock.range( + of: "let activePreserved = taskManager.preserveActiveTranscriptionsForShutdown(" + )?.lowerBound ?? terminationBlock.endIndex + let pendingNamingCleanupIndex = terminationBlock.range( + of: "taskManager.cleanupPendingNaming()" + )?.lowerBound ?? terminationBlock.startIndex + assertTrue( + activePreservationIndex < pendingNamingCleanupIndex, + "shutdown must preserve active imported audio before pending review cleanup can retire its journal" + ) assertTrue( terminationBlock.contains("let shutdownFailedTaskId = UUID()") && terminationBlock.contains("timedOutOwner: .failedMeeting(shutdownFailedTaskId)") @@ -2966,7 +3005,9 @@ func testRepoCommandContract() { ) assertTrue( cancelBlock.contains("if reason == .userRequested") - && cancelBlock.contains("try? FileManager.default.removeItem(at: audioURL)") + && cancelBlock.contains("prepareImportedScratchCleanup(for: job)") + && cancelBlock.contains("try FileManager.default.removeItem(at: audioURL)") + && cancelBlock.contains("confirmImportedScratchCleanup(for: job)") && cancelBlock.contains("Imported audio saved before cancellation"), "explicit user cancellation should delete cancelled imports while non-user cancellation keeps recovery audio" ) @@ -3297,7 +3338,9 @@ func testRepoCommandContract() { ) assertTrue( runnerContents.contains("let replacementTranscriptRollback = try ReplacementTranscriptRollback.capture(for: targetTranscriptURL)") - && runnerContents.contains("let transcriptId = replacementTranscriptRollback?.transcriptId ?? UUID()"), + && runnerContents.contains("let transcriptId = replacementTranscriptRollback?.transcriptId") + && runnerContents.contains("?? stableTranscriptId") + && runnerContents.contains("?? UUID()"), "replacement retranscription should snapshot the selected transcript and reuse its existing ID" ) assertTrue( diff --git a/Tests/TranscriptedCoreTests/PipelineTests/TranscriptionTaskManagerImportedAudioTests.swift b/Tests/TranscriptedCoreTests/PipelineTests/TranscriptionTaskManagerImportedAudioTests.swift index 942779de9..b2ceffc5d 100644 --- a/Tests/TranscriptedCoreTests/PipelineTests/TranscriptionTaskManagerImportedAudioTests.swift +++ b/Tests/TranscriptedCoreTests/PipelineTests/TranscriptionTaskManagerImportedAudioTests.swift @@ -4,6 +4,52 @@ import AVFoundation import FluidAudio @testable import TranscriptedCore +private final class ImportedRecoverySessionSpy: ImportedTranscriptionRecoverySession, @unchecked Sendable { + let jobID: UUID + + private let lock = NSLock() + private var transcriptCommits = 0 + private var scratchCleanupPreparations = 0 + private var scratchCleanups = 0 + private var failedQueueHandoffs = 0 + private var finished = false + + init(jobID: UUID) { + self.jobID = jobID + } + + var transcriptCommitCount: Int { lock.withLock { transcriptCommits } } + var scratchCleanupPreparationCount: Int { lock.withLock { scratchCleanupPreparations } } + var scratchCleanupCount: Int { lock.withLock { scratchCleanups } } + var failedQueueHandoffCount: Int { lock.withLock { failedQueueHandoffs } } + + func transcriptCommitConfirmed() { + lock.withLock { transcriptCommits += 1 } + } + + func prepareForScratchCleanup() -> Bool { + lock.withLock { + guard !finished else { return false } + scratchCleanupPreparations += 1 + return true + } + } + + func scratchCleanupConfirmed() { + lock.withLock { + scratchCleanups += 1 + finished = true + } + } + + func failedQueueHandoffConfirmed() { + lock.withLock { + failedQueueHandoffs += 1 + finished = true + } + } +} + @available(macOS 14.0, *) @MainActor extension TranscriptionTaskManagerMetadataTests { @@ -35,11 +81,15 @@ extension TranscriptionTaskManagerMetadataTests { .appendingPathComponent("audio", isDirectory: true) .appendingPathComponent("imported-meeting.wav") try writeMonoWAV(to: copiedImportURL, duration: 2.5) + let taskId = UUID() + let recoverySession = ImportedRecoverySessionSpy(jobID: taskId) manager.startImportedTranscription( + taskId: taskId, audioURL: copiedImportURL, outputFolder: tempDirectory.appendingPathComponent("transcripts"), - meetingTitle: "Imported customer call" + meetingTitle: "Imported customer call", + recoverySession: recoverySession ) try await waitUntil { @@ -59,7 +109,180 @@ extension TranscriptionTaskManagerMetadataTests { XCTAssertTrue(failed.micAudioURL.path.hasPrefix(retainedAudioDirectory.path + "/")) XCTAssertTrue(FileManager.default.fileExists(atPath: failed.micAudioURL.path)) XCTAssertFalse(FileManager.default.fileExists(atPath: copiedImportURL.path)) + XCTAssertEqual(recoverySession.failedQueueHandoffCount, 1) + XCTAssertEqual(recoverySession.scratchCleanupCount, 0) + + speech.release() + } + + func testShutdownPreservesActiveImportedAudioBeforePendingNamingCleanup() async throws { + let speech = BlockingMetadataStubSpeechToTextEngine(transcript: "Imported shutdown ordering.") + let retainedAudioDirectory = tempDirectory + .appendingPathComponent("transcripts", isDirectory: true) + .appendingPathComponent("audio", isDirectory: true) + let manager = makeManager( + speechToText: speech, + diarization: MetadataStubDiarizationEngine(segments: singleSpeakerSegments()), + retainedAudioDirectory: retainedAudioDirectory + ) + let copiedImportURL = tempDirectory + .appendingPathComponent("audio", isDirectory: true) + .appendingPathComponent("imported-precommit-review.wav") + try writeMonoWAV(to: copiedImportURL, duration: 2.5) + let taskId = UUID() + let recoverySession = ImportedRecoverySessionSpy(jobID: taskId) + + manager.startImportedTranscription( + taskId: taskId, + audioURL: copiedImportURL, + outputFolder: tempDirectory.appendingPathComponent("transcripts"), + recoverySession: recoverySession + ) + try await waitUntil { speech.didStart } + + manager.speakerNamingRequest = SpeakerNamingRequest( + speakers: [], + knownPeople: [], + recognizedPeopleCount: 0, + transcriptURL: tempDirectory.appendingPathComponent("precommit.md"), + transcriptId: taskId, + systemAudioURL: copiedImportURL, + micAudioURL: nil, + shouldRemoveTemporaryAudioOnCleanup: true, + sourceFailedTranscriptionId: nil, + importedRecoverySession: recoverySession, + onComplete: { _ in } + ) + + XCTAssertEqual( + manager.preserveActiveTranscriptionsForShutdown(errorMessage: "test shutdown ordering"), + 1 + ) + manager.cleanupPendingNaming() + + let failed = try XCTUnwrap(manager.failedTranscriptionManager.failedTranscriptions.first) + XCTAssertTrue(FileManager.default.fileExists(atPath: failed.micAudioURL.path)) + XCTAssertEqual(recoverySession.failedQueueHandoffCount, 1) + XCTAssertEqual(recoverySession.scratchCleanupPreparationCount, 0) + XCTAssertEqual(recoverySession.scratchCleanupCount, 0) + XCTAssertNil(manager.speakerNamingRequest) + + speech.release() + } + + func testImportedRecoverySessionUsesStableIdentityAndWaitsForScratchCleanup() async throws { + let speech = BlockingMetadataStubSpeechToTextEngine(transcript: "Imported terminal handoff.") + let manager = makeManager( + speechToText: speech, + diarization: MetadataStubDiarizationEngine(segments: singleSpeakerSegments()) + ) + let copiedImportURL = tempDirectory + .appendingPathComponent("audio", isDirectory: true) + .appendingPathComponent("imported-terminal-handoff.wav") + try writeMonoWAV(to: copiedImportURL, duration: 2.5) + + let taskId = UUID() + let recoverySession = ImportedRecoverySessionSpy(jobID: taskId) + manager.startImportedTranscription( + taskId: taskId, + audioURL: copiedImportURL, + outputFolder: tempDirectory.appendingPathComponent("transcripts"), + recoverySession: recoverySession + ) + + try await waitUntil { speech.didStart } + XCTAssertEqual(recoverySession.transcriptCommitCount, 0) + XCTAssertEqual(recoverySession.scratchCleanupPreparationCount, 0) + XCTAssertEqual(recoverySession.scratchCleanupCount, 0) + XCTAssertTrue(FileManager.default.fileExists(atPath: copiedImportURL.path)) + + speech.release() + try await waitUntil { + recoverySession.transcriptCommitCount == 1 + && manager.lastSavedTranscriptURL != nil + && manager.activeTasks.isEmpty + } + + let transcriptURL = try XCTUnwrap(manager.lastSavedTranscriptURL) + let frontmatter = try XCTUnwrap(try TranscriptFrontmatter.readValues(from: transcriptURL)) + XCTAssertEqual(TranscriptFrontmatter.captureID(in: frontmatter), taskId) + XCTAssertEqual(recoverySession.scratchCleanupPreparationCount, 0) + XCTAssertEqual(recoverySession.scratchCleanupCount, 0, "speaker review still owns scratch audio") + + XCTAssertTrue(manager.deferPendingSpeakerNamingReview(reason: "test cleanup")) + try await waitUntil { + recoverySession.scratchCleanupCount == 1 + } + XCTAssertEqual(recoverySession.scratchCleanupPreparationCount, 1) + XCTAssertFalse(FileManager.default.fileExists(atPath: copiedImportURL.path)) + } + func testImportedArchiveFailureDoesNotAuthorizeRecoveryCleanup() async throws { + let blockedArchiveParent = tempDirectory.appendingPathComponent("blocked-retained-audio") + try Data([0]).write(to: blockedArchiveParent) + let manager = makeManager( + speechToText: MetadataStubSpeechToTextEngine(transcript: "Synthetic imported archive failure."), + diarization: MetadataStubDiarizationEngine(segments: singleSpeakerSegments()), + retainedAudioDirectory: blockedArchiveParent.appendingPathComponent("audio") + ) + let copiedImportURL = tempDirectory + .appendingPathComponent("audio", isDirectory: true) + .appendingPathComponent("archive-failure.wav") + try writeMonoWAV(to: copiedImportURL, duration: 2.5) + let taskId = UUID() + let recoverySession = ImportedRecoverySessionSpy(jobID: taskId) + + manager.startImportedTranscription( + taskId: taskId, + audioURL: copiedImportURL, + outputFolder: tempDirectory.appendingPathComponent("transcripts"), + recoverySession: recoverySession + ) + + try await waitUntil { + recoverySession.transcriptCommitCount == 1 && manager.activeTasks.isEmpty + } + + XCTAssertEqual(recoverySession.scratchCleanupPreparationCount, 0) + XCTAssertEqual(recoverySession.scratchCleanupCount, 0) + XCTAssertTrue( + FileManager.default.fileExists(atPath: copiedImportURL.path), + "archive failure must retain the only audio copy" + ) + } + + func testFailedShutdownHandoffKeepsImportedRecoverySession() async throws { + let speech = BlockingMetadataStubSpeechToTextEngine(transcript: "Imported failed handoff.") + let blockedParent = tempDirectory.appendingPathComponent("blocked-failed-queue") + try Data([0]).write(to: blockedParent) + let manager = makeManager( + speechToText: speech, + diarization: MetadataStubDiarizationEngine(segments: singleSpeakerSegments()), + failedQueueURL: blockedParent.appendingPathComponent("failed.json") + ) + let copiedImportURL = tempDirectory + .appendingPathComponent("audio", isDirectory: true) + .appendingPathComponent("failed-handoff.wav") + try writeMonoWAV(to: copiedImportURL, duration: 2.5) + let taskId = UUID() + let recoverySession = ImportedRecoverySessionSpy(jobID: taskId) + + manager.startImportedTranscription( + taskId: taskId, + audioURL: copiedImportURL, + outputFolder: tempDirectory.appendingPathComponent("transcripts"), + recoverySession: recoverySession + ) + try await waitUntil { speech.didStart } + + XCTAssertEqual( + manager.preserveActiveTranscriptionsForShutdown(errorMessage: "test failed handoff"), + 0 + ) + XCTAssertEqual(recoverySession.failedQueueHandoffCount, 0) + XCTAssertEqual(recoverySession.scratchCleanupPreparationCount, 0) + XCTAssertEqual(recoverySession.scratchCleanupCount, 0) + XCTAssertTrue(FileManager.default.fileExists(atPath: copiedImportURL.path)) speech.release() } diff --git a/Tests/TranscriptedCoreTests/PipelineTests/TranscriptionTaskManagerMetadataTests.swift b/Tests/TranscriptedCoreTests/PipelineTests/TranscriptionTaskManagerMetadataTests.swift index 39750c477..5a417b389 100644 --- a/Tests/TranscriptedCoreTests/PipelineTests/TranscriptionTaskManagerMetadataTests.swift +++ b/Tests/TranscriptedCoreTests/PipelineTests/TranscriptionTaskManagerMetadataTests.swift @@ -483,13 +483,14 @@ final class TranscriptionTaskManagerMetadataTests: XCTestCase { diarization: (any DiarizationEngine)? = nil, retainedAudioDirectory: URL? = nil, retainedAudioDirectoryProvider: (() -> URL?)? = nil, - statsStore: (any StatsStore)? = nil + statsStore: (any StatsStore)? = nil, + failedQueueURL: URL? = nil ) -> TranscriptionTaskManager { let paths = CoreStoragePaths( transcripts: tempDirectory.appendingPathComponent("transcripts"), speakerDB: tempDirectory.appendingPathComponent("speakers.sqlite"), statsDB: tempDirectory.appendingPathComponent("stats.sqlite"), - failedQueue: tempDirectory.appendingPathComponent("failed_transcriptions.json"), + failedQueue: failedQueueURL ?? tempDirectory.appendingPathComponent("failed_transcriptions.json"), speakerClips: tempDirectory.appendingPathComponent("speaker_clips"), audioCaptures: tempDirectory.appendingPathComponent("audio"), logs: tempDirectory.appendingPathComponent("logs") diff --git a/Tests/TranscriptedCoreTests/StorageTests/TranscriptSaverRecoveryIdentityTests.swift b/Tests/TranscriptedCoreTests/StorageTests/TranscriptSaverRecoveryIdentityTests.swift new file mode 100644 index 000000000..8c046a75b --- /dev/null +++ b/Tests/TranscriptedCoreTests/StorageTests/TranscriptSaverRecoveryIdentityTests.swift @@ -0,0 +1,60 @@ +import Foundation +import XCTest +@testable import TranscriptedCore + +final class TranscriptSaverRecoveryIdentityTests: XCTestCase { + func testFindsStableTranscriptIdentityWithoutFollowingSymlinks() throws { + let root = FileManager.default.temporaryDirectory.appendingPathComponent( + "TranscriptSaverRecoveryIdentityTests-\(UUID().uuidString)", + isDirectory: true + ) + defer { try? FileManager.default.removeItem(at: root) } + try FileManager.default.createDirectory(at: root, withIntermediateDirectories: true) + + let transcriptId = UUID() + let expectedURL = root.appendingPathComponent("committed.md") + let unrelatedURL = root.appendingPathComponent("unrelated.md") + let outsideURL = root.deletingLastPathComponent().appendingPathComponent( + "outside-\(UUID().uuidString).md" + ) + defer { try? FileManager.default.removeItem(at: outsideURL) } + + try syntheticTranscript(id: UUID()).write( + to: unrelatedURL, + atomically: true, + encoding: .utf8 + ) + try syntheticTranscript(id: transcriptId).write( + to: expectedURL, + atomically: true, + encoding: .utf8 + ) + try syntheticTranscript(id: transcriptId).write( + to: outsideURL, + atomically: true, + encoding: .utf8 + ) + try FileManager.default.createSymbolicLink( + at: root.appendingPathComponent("linked.md"), + withDestinationURL: outsideURL + ) + + XCTAssertEqual( + TranscriptSaver.existingTranscriptURL(in: root, transcriptId: transcriptId)? + .resolvingSymlinksInPath(), + expectedURL.resolvingSymlinksInPath() + ) + XCTAssertNil( + TranscriptSaver.existingTranscriptURL(in: root, transcriptId: UUID()) + ) + } + + private func syntheticTranscript(id: UUID) -> String { + """ + --- + transcript_id: \(id.uuidString) + --- + Synthetic recovery fixture. + """ + } +} diff --git a/scripts/dev/benchmark-home-recent-captures.sh b/scripts/dev/benchmark-home-recent-captures.sh index e7c15a408..16c5a14ff 100755 --- a/scripts/dev/benchmark-home-recent-captures.sh +++ b/scripts/dev/benchmark-home-recent-captures.sh @@ -20,6 +20,7 @@ swiftc \ "$ROOT_DIR/Sources/Meeting/LocalMeetingSummarizer.swift" \ "$ROOT_DIR/Sources/Support/LocalMeetingSummaryPreferences.swift" \ "$ROOT_DIR/Sources/TranscriptedCore/Speaker/SpeakerProfile.swift" \ + "$ROOT_DIR/Sources/TranscriptedCore/Protocols/ImportedTranscriptionRecoverySession.swift" \ "$ROOT_DIR/Sources/TranscriptedCore/Models/TranscriptionTypes.swift" \ "$ROOT_DIR/Sources/TranscriptedCore/Storage/TranscriptFrontmatter.swift" \ "$ROOT_DIR/Sources/UI/Shared/MeetingAudioArchiveResolver.swift" \ diff --git a/scripts/entrypoints/run-e2e-smoke.sh b/scripts/entrypoints/run-e2e-smoke.sh index 8bb7b1f29..24a968d84 100755 --- a/scripts/entrypoints/run-e2e-smoke.sh +++ b/scripts/entrypoints/run-e2e-smoke.sh @@ -42,6 +42,7 @@ SWIFT_SOURCES=( "Sources/Meeting/LocalMeetingSummarizer.swift" "Sources/Meeting/MeetingArtifactRenamer.swift" "Sources/Meeting/MeetingTranscriptStyler.swift" + "Sources/TranscriptedCore/Protocols/ImportedTranscriptionRecoverySession.swift" "Sources/TranscriptedCore/Storage/TranscriptFrontmatter.swift" "Sources/TranscriptedCore/Models/TranscriptionTypes.swift" "Sources/TranscriptedCore/Models/FailedTranscription.swift" diff --git a/scripts/entrypoints/run-tests.sh b/scripts/entrypoints/run-tests.sh index e3da8634a..df01bdf9c 100755 --- a/scripts/entrypoints/run-tests.sh +++ b/scripts/entrypoints/run-tests.sh @@ -249,6 +249,7 @@ $test_entries EOF APP_SOURCES=( + "Sources/TranscriptedCore/Protocols/ImportedTranscriptionRecoverySession.swift" "Sources/Support/ActivationPolicyController.swift" "Sources/Support/TranscriptedPermissionKind.swift" "Sources/Support/TranscriptedPermissionAccess.swift" @@ -321,6 +322,7 @@ APP_SOURCES=( "Sources/Meeting/SustainedActivityConfirmer.swift" "Sources/Meeting/MeetingAudioInactivityDetector.swift" "Sources/Meeting/MeetingAudioStorageManager.swift" + "Sources/Meeting/ImportedTranscriptionQueueJournalState.swift" "Sources/Meeting/MeetingImportedAudioPreparer.swift" "Sources/Meeting/MeetingImportPreparationFailureCopy.swift" "Sources/Meeting/MeetingSessionUIPolicy.swift"