feat(sync): 增强同步引擎配额校验、断点进度、SHA256 去重与版本冲突检测

CelestiaTrace/Models/CelestiaSession.swift

@@ -47,7 +47,10 @@ final class CelestiaSession {
     }
     
     var noteCount: Int {
-        events.filter { $0.eventType == "NOTE" || $0.eventType == "MARKER" }.count
+        events.filter {
+            $0.eventType == "NOTE"
+                || ($0.eventType == "MARKER" && !$0.isContinuationMarker)
+        }.count
     }
     
     var syncState: SessionSyncState {
@@ -55,6 +58,15 @@ final class CelestiaSession {
         set { syncStateRaw = newValue.rawValue }
     }
 
+    /// The revision represented by the current local content.
+    /// Unsynced edits form the next revision based on the last known server revision.
+    var localRevision: Int64 {
+        if isSynced && syncState == .synced {
+            return max(serverRevision, 1)
+        }
+        return max(serverRevision + 1, 1)
+    }
+
     init(id: UUID = UUID(), title: String, localAudioPath: String? = nil, startTime: Date = Date()) {
         self.id = id
         self.title = title

CelestiaTrace/Services/Network/APIClient.swift

@@ -1,6 +1,9 @@
 import Foundation
 import Security
 
+// Remote contract: Docs/RemoteAPI.openapi.yaml
+// Keep the spec and these request/response models in sync.
+
 enum HTTPMethod: String {
     case get = "GET"
     case post = "POST"
@@ -14,6 +17,11 @@ struct APIEnvelope<T: Decodable>: Decodable {
     let data: T?
 }
 
+private struct APIErrorEnvelope: Decodable {
+    let code: Int
+    let message: String
+}
+
 struct EmptyAPIData: Decodable { }
 
 enum APIError: LocalizedError {
@@ -25,6 +33,7 @@ enum APIError: LocalizedError {
     case transport(String)
     case decoding(String)
     case missingData
+    case storageQuotaExceeded(requiredBytes: Int64, remainingBytes: Int64)
 
     var errorDescription: String? {
         switch self {
@@ -36,6 +45,10 @@ enum APIError: LocalizedError {
         case .transport(let message): return "网络连接失败：\(message)"
         case .decoding(let message): return "服务器数据解析失败：\(message)"
         case .missingData: return "服务器响应缺少必要数据"
+        case .storageQuotaExceeded(let requiredBytes, let remainingBytes):
+            let formatter = ByteCountFormatter()
+            formatter.countStyle = .file
+            return "云端空间不足：需要 \(formatter.string(fromByteCount: requiredBytes))，剩余 \(formatter.string(fromByteCount: remainingBytes))"
         }
     }
 }
@@ -182,13 +195,28 @@ actor APIClient {
         fileURL: URL,
         fileName: String,
         mimeType: String,
-        fields: [String: String]
+        fields: [String: String],
+        progress: @escaping @Sendable (Double) -> Void
     ) async throws -> T {
         do {
-            return try await performUpload(path, fileURL: fileURL, fileName: fileName, mimeType: mimeType, fields: fields)
+            return try await performUpload(
+                path,
+                fileURL: fileURL,
+                fileName: fileName,
+                mimeType: mimeType,
+                fields: fields,
+                progress: progress
+            )
         } catch APIError.unauthorized {
             try await refreshTokens()
-            return try await performUpload(path, fileURL: fileURL, fileName: fileName, mimeType: mimeType, fields: fields)
+            return try await performUpload(
+                path,
+                fileURL: fileURL,
+                fileName: fileName,
+                mimeType: mimeType,
+                fields: fields,
+                progress: progress
+            )
         }
     }
 
@@ -227,14 +255,16 @@ actor APIClient {
             try await refreshTokens()
             return try await execute(path, method: method, body: body, authenticated: true, canRefresh: false)
         }
+        guard (200..<300).contains(http.statusCode) else {
+            throw Self.serverError(from: data, fallbackStatusCode: http.statusCode)
+        }
         let envelope: APIEnvelope<T>
         do {
             envelope = try Self.decoder.decode(APIEnvelope<T>.self, from: data)
         } catch {
             throw APIError.decoding(error.localizedDescription)
         }
-        guard (200..<300).contains(http.statusCode), envelope.code == 0 else {
-            if http.statusCode == 401 { throw APIError.unauthorized }
+        guard envelope.code == 0 else {
             throw APIError.server(code: envelope.code, message: envelope.message)
         }
         return envelope
@@ -259,7 +289,8 @@ actor APIClient {
         fileURL: URL,
         fileName: String,
         mimeType: String,
-        fields: [String: String]
+        fields: [String: String],
+        progress: @escaping @Sendable (Double) -> Void
     ) async throws -> T {
         guard let token = tokenStore.load()?.accessToken else { throw APIError.unauthorized }
         let boundary = "CelestiaBoundary-\(UUID().uuidString)"
@@ -290,19 +321,27 @@ actor APIClient {
         let data: Data
         let response: URLResponse
         do {
-            (data, response) = try await session.upload(for: request, fromFile: temporaryURL)
+            let progressDelegate = UploadProgressDelegate(progress: progress)
+            (data, response) = try await session.upload(
+                for: request,
+                fromFile: temporaryURL,
+                delegate: progressDelegate
+            )
         } catch {
             throw APIError.transport(error.localizedDescription)
         }
         guard let http = response as? HTTPURLResponse else { throw APIError.invalidResponse }
         if http.statusCode == 401 { throw APIError.unauthorized }
+        guard (200..<300).contains(http.statusCode) else {
+            throw Self.serverError(from: data, fallbackStatusCode: http.statusCode)
+        }
         let envelope: APIEnvelope<T>
         do {
             envelope = try Self.decoder.decode(APIEnvelope<T>.self, from: data)
         } catch {
             throw APIError.decoding(error.localizedDescription)
         }
-        guard (200..<300).contains(http.statusCode), envelope.code == 0 else {
+        guard envelope.code == 0 else {
             throw APIError.server(code: envelope.code, message: envelope.message)
         }
         guard let result = envelope.data else { throw APIError.missingData }
@@ -390,4 +429,33 @@ actor APIClient {
             return false
         }
     }
+
+    private static func serverError(from data: Data, fallbackStatusCode: Int) -> APIError {
+        if fallbackStatusCode == 401 {
+            return .unauthorized
+        }
+        if let envelope = try? decoder.decode(APIErrorEnvelope.self, from: data) {
+            return .server(code: envelope.code, message: envelope.message)
+        }
+        return .server(code: fallbackStatusCode, message: "服务器请求失败（\(fallbackStatusCode)）")
+    }
+}
+
+private final class UploadProgressDelegate: NSObject, URLSessionTaskDelegate, @unchecked Sendable {
+    private let progress: @Sendable (Double) -> Void
+
+    init(progress: @escaping @Sendable (Double) -> Void) {
+        self.progress = progress
+    }
+
+    func urlSession(
+        _ session: URLSession,
+        task: URLSessionTask,
+        didSendBodyData bytesSent: Int64,
+        totalBytesSent: Int64,
+        totalBytesExpectedToSend: Int64
+    ) {
+        guard totalBytesExpectedToSend > 0 else { return }
+        progress(min(max(Double(totalBytesSent) / Double(totalBytesExpectedToSend), 0), 1))
+    }
 }

CelestiaTrace/Services/Network/NetworkServiceProtocol.swift

@@ -2,8 +2,18 @@ import Foundation
 
 protocol NetworkServiceProtocol {
     func fetchSessions() async throws -> [RemoteSession]
-    func syncSession(_ session: CelestiaSession) async throws -> RemoteSession
-    func uploadAsset(sessionID: String, clientID: String, kind: String, fileURL: URL) async throws -> RemoteAsset
+    func syncSession(
+        _ session: CelestiaSession,
+        deletedEventClientIDs: [String]
+    ) async throws -> RemoteSession
+    func fetchStorageQuota() async throws -> StorageQuota
+    func uploadAsset(
+        sessionID: String,
+        clientID: String,
+        kind: String,
+        fileURL: URL,
+        progress: @escaping @Sendable (Double) -> Void
+    ) async throws -> AssetUploadResult
     func downloadAsset(sessionID: String, asset: RemoteAsset, destinationURL: URL) async throws
     func recordSyncCheckpoint() async throws
 }

CelestiaTrace/Services/Network/RemoteNetworkService.swift

@@ -1,6 +1,10 @@
 import Foundation
 import SwiftData
 import UniformTypeIdentifiers
+import Network
+import CryptoKit
+
+// Remote contract: Docs/RemoteAPI.openapi.yaml
 
 struct RemoteAsset: Decodable {
     let id: String
@@ -10,6 +14,18 @@ struct RemoteAsset: Decodable {
     let mimeType: String
     let sizeBytes: Int64
     let sha256: String
+    let createdAt: Date?
+}
+
+struct AssetUploadResult {
+    let asset: RemoteAsset
+    let sessionRevision: Int64?
+}
+
+struct StorageQuota: Decodable {
+    let totalBytes: Int64
+    let usedBytes: Int64
+    let remainingBytes: Int64
 }
 
 struct RemoteEvent: Decodable {
@@ -42,6 +58,8 @@ private struct SessionUpload: Encodable {
     let endTime: Date?
     let durationMs: Int64
     let events: [EventUpload]
+    let baseRevision: Int64?
+    let deletedEventClientIds: [String]
 }
 
 private struct EventUpload: Encodable {
@@ -57,7 +75,55 @@ private struct SyncCheckpoint: Decodable {
     let syncedAt: Date
 }
 
+private struct AssetUploadPayload: Decodable {
+    let asset: RemoteAsset
+    let sessionRevision: Int64?
+
+    private enum CodingKeys: String, CodingKey {
+        case asset
+        case sessionRevision
+    }
+
+    init(from decoder: Decoder) throws {
+        if let container = try? decoder.container(keyedBy: CodingKeys.self),
+           container.contains(.asset) {
+            asset = try container.decode(RemoteAsset.self, forKey: .asset)
+            sessionRevision = try container.decodeIfPresent(Int64.self, forKey: .sessionRevision)
+        } else {
+            asset = try RemoteAsset(from: decoder)
+            sessionRevision = nil
+        }
+    }
+}
+
+private struct ChunkUploadInitBody: Encodable {
+    let clientId: String
+    let kind: String
+    let fileName: String
+    let mimeType: String
+    let fileSize: Int64
+    let chunkSize: Int
+}
+
+private struct ChunkUploadInitResponse: Decodable {
+    let uploadId: String
+    let totalChunks: Int
+    let chunkSize: Int
+}
+
+private struct ChunkUploadProgressResponse: Decodable {
+    let uploadedChunks: Int
+    let totalChunks: Int
+}
+
+private struct ChunkUploadCompleteBody: Encodable {
+    let uploadId: String
+    let totalChunks: Int
+}
+
 final class RemoteNetworkService: NetworkServiceProtocol {
+    private static let chunkedUploadThreshold: Int64 = 16 * 1024 * 1024
+    private static let preferredChunkSize = 8 * 1024 * 1024
     private let client: APIClient
 
     init(client: APIClient = .shared) {
@@ -68,7 +134,10 @@ final class RemoteNetworkService: NetworkServiceProtocol {
         try await client.request("sessions?includeDeleted=true", authenticated: true)
     }
 
-    func syncSession(_ session: CelestiaSession) async throws -> RemoteSession {
+    func syncSession(
+        _ session: CelestiaSession,
+        deletedEventClientIDs: [String]
+    ) async throws -> RemoteSession {
         let payload = SessionUpload(
             clientId: session.id.uuidString,
             title: session.title,
@@ -84,22 +153,48 @@ final class RemoteNetworkService: NetworkServiceProtocol {
                     voiceStartOffsetMs: $0.voiceStartOffsetMs,
                     voiceEndOffsetMs: $0.voiceEndOffsetMs
                 )
-            }
+            },
+            baseRevision: session.serverRevision > 0 ? session.serverRevision : nil,
+            deletedEventClientIds: deletedEventClientIDs
         )
         let encoder = JSONEncoder()
         encoder.dateEncodingStrategy = .iso8601
         return try await client.request("sessions", method: .post, body: try encoder.encode(payload), authenticated: true)
     }
 
-    func uploadAsset(sessionID: String, clientID: String, kind: String, fileURL: URL) async throws -> RemoteAsset {
+    func fetchStorageQuota() async throws -> StorageQuota {
+        try await client.request("storage/quota", authenticated: true)
+    }
+
+    func uploadAsset(
+        sessionID: String,
+        clientID: String,
+        kind: String,
+        fileURL: URL,
+        progress: @escaping @Sendable (Double) -> Void
+    ) async throws -> AssetUploadResult {
         let mimeType = UTType(filenameExtension: fileURL.pathExtension)?.preferredMIMEType ?? "application/octet-stream"
-        return try await client.upload(
+        let size = try fileSize(of: fileURL)
+        if size >= Self.chunkedUploadThreshold {
+            return try await uploadAssetInChunks(
+                sessionID: sessionID,
+                clientID: clientID,
+                kind: kind,
+                fileURL: fileURL,
+                fileSize: size,
+                mimeType: mimeType,
+                progress: progress
+            )
+        }
+        let payload: AssetUploadPayload = try await client.upload(
             "sessions/\(sessionID)/assets",
             fileURL: fileURL,
             fileName: fileURL.lastPathComponent,
             mimeType: mimeType,
-            fields: ["clientId": clientID, "kind": kind]
+            fields: ["clientId": clientID, "kind": kind],
+            progress: progress
         )
+        return AssetUploadResult(asset: payload.asset, sessionRevision: payload.sessionRevision)
     }
 
     func downloadAsset(sessionID: String, asset: RemoteAsset, destinationURL: URL) async throws {
@@ -109,6 +204,82 @@ final class RemoteNetworkService: NetworkServiceProtocol {
     func recordSyncCheckpoint() async throws {
         let _: SyncCheckpoint = try await client.request("sync/trigger", method: .post, authenticated: true)
     }
+
+    private func uploadAssetInChunks(
+        sessionID: String,
+        clientID: String,
+        kind: String,
+        fileURL: URL,
+        fileSize: Int64,
+        mimeType: String,
+        progress: @escaping @Sendable (Double) -> Void
+    ) async throws -> AssetUploadResult {
+        let body = ChunkUploadInitBody(
+            clientId: clientID,
+            kind: kind,
+            fileName: fileURL.lastPathComponent,
+            mimeType: mimeType,
+            fileSize: fileSize,
+            chunkSize: Self.preferredChunkSize
+        )
+        let encoder = JSONEncoder()
+        let upload: ChunkUploadInitResponse = try await client.request(
+            "sessions/\(sessionID)/assets/init",
+            method: .post,
+            body: try encoder.encode(body),
+            authenticated: true
+        )
+
+        let input = try FileHandle(forReadingFrom: fileURL)
+        defer { try? input.close() }
+        for chunkIndex in 0..<upload.totalChunks {
+            let offset = UInt64(chunkIndex) * UInt64(upload.chunkSize)
+            try input.seek(toOffset: offset)
+            guard let data = try input.read(upToCount: upload.chunkSize), !data.isEmpty else {
+                throw APIError.transport("读取上传分片失败")
+            }
+            let temporaryURL = FileManager.default.temporaryDirectory
+                .appendingPathComponent("celestia-\(upload.uploadId)-\(chunkIndex).chunk")
+            try data.write(to: temporaryURL, options: .atomic)
+            defer { try? FileManager.default.removeItem(at: temporaryURL) }
+
+            let chunkStart = Double(offset) / Double(fileSize)
+            let chunkSpan = Double(data.count) / Double(fileSize)
+            let _: ChunkUploadProgressResponse = try await client.upload(
+                "sessions/\(sessionID)/assets/chunk",
+                fileURL: temporaryURL,
+                fileName: "chunk-\(chunkIndex)",
+                mimeType: "application/octet-stream",
+                fields: [
+                    "uploadId": upload.uploadId,
+                    "chunkIndex": String(chunkIndex)
+                ]
+            ) { fraction in
+                progress(min(max(chunkStart + chunkSpan * fraction, 0), 1))
+            }
+        }
+
+        let completeBody = ChunkUploadCompleteBody(
+            uploadId: upload.uploadId,
+            totalChunks: upload.totalChunks
+        )
+        let payload: AssetUploadPayload = try await client.request(
+            "sessions/\(sessionID)/assets/complete",
+            method: .post,
+            body: try encoder.encode(completeBody),
+            authenticated: true
+        )
+        progress(1)
+        return AssetUploadResult(asset: payload.asset, sessionRevision: payload.sessionRevision)
+    }
+
+    private func fileSize(of url: URL) throws -> Int64 {
+        let values = try url.resourceValues(forKeys: [.fileSizeKey])
+        guard let size = values.fileSize else {
+            throw APIError.transport("无法读取待上传文件大小")
+        }
+        return Int64(size)
+    }
 }
 
 struct RemoteBoundDevice: Decodable {
@@ -175,6 +346,37 @@ final class DeviceCloudService {
     }
 }
 
+final class NetworkStatusMonitor: ObservableObject, @unchecked Sendable {
+    static let shared = NetworkStatusMonitor()
+
+    @Published private(set) var isConnected = true
+    @Published private(set) var isWiFi = false
+    @Published private(set) var isCellular = false
+
+    private let monitor = NWPathMonitor()
+    private let queue = DispatchQueue(label: "com.celestia.trace.network-path")
+
+    private init() {
+        monitor.pathUpdateHandler = { [weak self] path in
+            let connected = path.status == .satisfied
+            let usesWiFi = path.usesInterfaceType(.wifi)
+            let usesCellular = path.usesInterfaceType(.cellular)
+            DispatchQueue.main.async {
+                self?.isConnected = connected
+                self?.isWiFi = usesWiFi
+                self?.isCellular = usesCellular
+            }
+        }
+        monitor.start(queue: queue)
+    }
+
+    var connectionName: String {
+        if isWiFi { return "Wi-Fi" }
+        if isCellular { return "蜂窝网络" }
+        return isConnected ? "当前网络" : "网络"
+    }
+}
+
 @MainActor
 final class SyncManager: ObservableObject {
     static let shared = SyncManager()
@@ -184,14 +386,32 @@ final class SyncManager: ObservableObject {
     @Published private(set) var lastErrorMessage: String?
     @Published private(set) var completedCount = 0
     @Published private(set) var totalCount = 0
+    @Published private(set) var activeSessionID: UUID?
+    @Published private(set) var syncProgress: Double = 0
 
     private let service: NetworkServiceProtocol
+    private var activeSyncTask: Task<Bool, Never>?
 
     init(service: NetworkServiceProtocol = RemoteNetworkService()) {
         self.service = service
         lastSyncDate = UserDefaults.standard.object(forKey: "com.celestia.trace.last_server_sync") as? Date
     }
 
+    func estimatedUploadBytes(for session: CelestiaSession) -> Int64 {
+        var urls: [URL] = []
+        if let url = AudioPathHelper.resolveURL(for: session.localAudioPath) {
+            urls.append(url)
+        }
+        urls.append(contentsOf: session.events.compactMap { event in
+            guard event.eventType == "PHOTO" else { return nil }
+            return AudioPathHelper.resolveURL(for: event.localFilePath)
+        })
+        return urls.reduce(into: 0) { total, url in
+            let values = try? url.resourceValues(forKeys: [.fileSizeKey])
+            total += Int64(values?.fileSize ?? 0)
+        }
+    }
+
     func sync(
         sessions: [CelestiaSession],
         modelContext: ModelContext,
@@ -199,38 +419,107 @@ final class SyncManager: ObservableObject {
     ) async -> Bool {
         guard !isSyncing else { return false }
         isSyncing = true
+        let task = Task { @MainActor [self] in
+            await performSync(
+                sessions: sessions,
+                modelContext: modelContext,
+                userID: userID
+            )
+        }
+        activeSyncTask = task
+        let result = await withTaskCancellationHandler {
+            await task.value
+        } onCancel: {
+            task.cancel()
+        }
+        activeSyncTask = nil
+        isSyncing = false
+        activeSessionID = nil
+        return result
+    }
+
+    func pauseSync(sessionID: UUID) {
+        guard isSyncing, activeSessionID == sessionID else { return }
+        activeSyncTask?.cancel()
+    }
+
+    private func performSync(
+        sessions: [CelestiaSession],
+        modelContext: ModelContext,
+        userID: String
+    ) async -> Bool {
         lastErrorMessage = nil
         completedCount = 0
+        activeSessionID = nil
+        syncProgress = 0
         totalCount = sessions.filter {
             ($0.ownerUserID == nil || $0.ownerUserID == userID) && (!$0.isSynced || $0.cloudSessionId == nil)
         }.count
-        defer { isSyncing = false }
-
         do {
             let remoteSessions = try await service.fetchSessions()
+            try Task.checkCancellation()
             let mergedSessions = merge(remoteSessions, into: sessions, modelContext: modelContext, userID: userID)
             for (remote, local) in mergedSessions {
                 try await downloadMissingAssets(for: local, remote: remote)
+                try Task.checkCancellation()
             }
             try modelContext.save()
 
             let eligible = sessions.filter {
-                ($0.ownerUserID == nil || $0.ownerUserID == userID) && (!$0.isSynced || $0.cloudSessionId == nil)
+                ($0.ownerUserID == nil || $0.ownerUserID == userID)
+                    && (!$0.isSynced || $0.cloudSessionId == nil)
+                    && $0.syncState != .conflict
             }
             var failures: [String] = []
-            for session in eligible {
+            let eligibleCount = max(eligible.count, 1)
+            for (index, session) in eligible.enumerated() {
+                activeSessionID = session.id
+                let sessionStartProgress = Double(index) / Double(eligibleCount)
+                let sessionProgressSpan = 1.0 / Double(eligibleCount)
+                syncProgress = sessionStartProgress
                 session.ownerUserID = userID
                 session.syncState = .syncing
                 session.lastSyncError = nil
                 do {
-                    let remote = try await service.syncSession(session)
+                    let previousRemote = remoteSessions.first {
+                        $0.id == session.cloudSessionId
+                            || $0.clientId?.caseInsensitiveCompare(session.id.uuidString) == .orderedSame
+                    }
+                    let localEventIDs = Set(session.events.map { $0.id.uuidString.lowercased() })
+                    let deletedEventClientIDs = previousRemote?.events?
+                        .compactMap(\.clientId)
+                        .filter { !localEventIDs.contains($0.lowercased()) } ?? []
+                    let remote = try await service.syncSession(
+                        session,
+                        deletedEventClientIDs: deletedEventClientIDs
+                    )
+                    try Task.checkCancellation()
+                    syncProgress = sessionStartProgress + sessionProgressSpan * 0.12
                     session.cloudSessionId = remote.id
                     session.serverRevision = remote.revision
-                    try await uploadLocalAssets(for: session, remote: remote)
+                    try await uploadLocalAssets(for: session, remote: remote) { [weak self] assetProgress in
+                        Task { @MainActor [weak self] in
+                            self?.syncProgress = sessionStartProgress
+                                + sessionProgressSpan * (0.12 + assetProgress * 0.83)
+                        }
+                    }
+                    try Task.checkCancellation()
                     session.isSynced = true
                     session.syncState = .synced
                     session.lastSyncedAt = Date()
                     completedCount += 1
+                    syncProgress = sessionStartProgress + sessionProgressSpan
+                } catch where Task.isCancelled {
+                    session.isSynced = false
+                    session.syncState = .pending
+                    session.lastSyncError = nil
+                    syncProgress = 0
+                    try? modelContext.save()
+                    return false
+                } catch APIError.server(let code, _) where code == 409 {
+                    session.isSynced = false
+                    session.syncState = .conflict
+                    session.lastSyncError = "本地版本 \(session.serverRevision) 与云端版本不一致"
                 } catch {
                     session.isSynced = false
                     session.syncState = .failed
@@ -239,6 +528,12 @@ final class SyncManager: ObservableObject {
                 }
                 try modelContext.save()
             }
+            let conflicts = sessions.filter { $0.syncState == .conflict }
+            if !conflicts.isEmpty {
+                failures.append(contentsOf: conflicts.map {
+                    "\($0.title)：本地和云端版本不一致"
+                })
+            }
             guard failures.isEmpty else {
                 lastErrorMessage = failures.joined(separator: "\n")
                 return false
@@ -248,22 +543,35 @@ final class SyncManager: ObservableObject {
             lastSyncDate = now
             UserDefaults.standard.set(now, forKey: "com.celestia.trace.last_server_sync")
             return true
+        } catch where Task.isCancelled {
+            syncProgress = 0
+            return false
         } catch {
             lastErrorMessage = error.localizedDescription
             return false
         }
     }
 
-    private func uploadLocalAssets(for session: CelestiaSession, remote: RemoteSession) async throws {
+    private func uploadLocalAssets(
+        for session: CelestiaSession,
+        remote: RemoteSession,
+        progress: @escaping @Sendable (Double) -> Void
+    ) async throws {
         guard let cloudID = session.cloudSessionId else { throw APIError.missingData }
         let existingIDs = Set((remote.assets ?? []).map(\.clientId))
+        var uploads: [(clientID: String, kind: String, url: URL, size: Int64)] = []
         if let path = session.localAudioPath {
             guard let url = AudioPathHelper.resolveURL(for: path) else {
                 throw APIError.transport("本地录音文件不存在：\(path)")
             }
-            let clientID = "\(session.id.uuidString)-audio"
-            if !existingIDs.contains(clientID) {
-                _ = try await service.uploadAsset(sessionID: cloudID, clientID: clientID, kind: "AUDIO", fileURL: url)
+            let sha256 = try fileSHA256(of: url)
+            let hasMatchingAudio = (remote.assets ?? []).contains {
+                $0.kind.uppercased() == "AUDIO"
+                    && $0.sha256.caseInsensitiveCompare(sha256) == .orderedSame
+            }
+            if !hasMatchingAudio {
+                let clientID = "\(session.id.uuidString)-audio-\(sha256.prefix(16))"
+                uploads.append((clientID, "AUDIO", url, fileSize(of: url)))
             }
         }
         for event in session.events where event.eventType == "PHOTO" {
@@ -273,18 +581,75 @@ final class SyncManager: ObservableObject {
             }
             let clientID = "\(event.id.uuidString)-photo"
             if !existingIDs.contains(clientID) {
-                _ = try await service.uploadAsset(sessionID: cloudID, clientID: clientID, kind: "PHOTO", fileURL: url)
+                uploads.append((clientID, "PHOTO", url, fileSize(of: url)))
+            }
+        }
+
+        let totalBytes = max(uploads.reduce(Int64(0)) { $0 + $1.size }, 1)
+        let requiredBytes = uploads.reduce(Int64(0)) { $0 + $1.size }
+        if requiredBytes > 0 {
+            let quota = try await service.fetchStorageQuota()
+            guard requiredBytes <= quota.remainingBytes else {
+                throw APIError.storageQuotaExceeded(
+                    requiredBytes: requiredBytes,
+                    remainingBytes: quota.remainingBytes
+                )
             }
         }
+        var completedBytes: Int64 = 0
+        progress(uploads.isEmpty ? 1 : 0)
+        for upload in uploads {
+            let bytesBeforeUpload = completedBytes
+            let result = try await service.uploadAsset(
+                sessionID: cloudID,
+                clientID: upload.clientID,
+                kind: upload.kind,
+                fileURL: upload.url
+            ) { fraction in
+                let sentBytes = Double(bytesBeforeUpload) + Double(upload.size) * fraction
+                progress(min(max(sentBytes / Double(totalBytes), 0), 1))
+            }
+            if let revision = result.sessionRevision {
+                session.serverRevision = revision
+            }
+            completedBytes += upload.size
+            progress(Double(completedBytes) / Double(totalBytes))
+        }
+    }
+
+    private func fileSize(of url: URL) -> Int64 {
+        let values = try? url.resourceValues(forKeys: [.fileSizeKey])
+        return Int64(values?.fileSize ?? 0)
+    }
+
+    private func fileSHA256(of url: URL) throws -> String {
+        let input = try FileHandle(forReadingFrom: url)
+        defer { try? input.close() }
+        var hasher = SHA256()
+        while let data = try input.read(upToCount: 1_024 * 1_024),
+              !data.isEmpty {
+            hasher.update(data: data)
+        }
+        return hasher.finalize().map { String(format: "%02x", $0) }.joined()
     }
 
     private func downloadMissingAssets(for session: CelestiaSession, remote: RemoteSession) async throws {
         let assets = remote.assets ?? []
-        if (session.localAudioPath == nil || AudioPathHelper.resolveURL(for: session.localAudioPath) == nil),
-           let audio = assets.first(where: { $0.kind.uppercased() == "AUDIO" }) {
-            let destination = downloadDestination(for: audio)
-            try await service.downloadAsset(sessionID: remote.id, asset: audio, destinationURL: destination)
-            session.localAudioPath = AudioPathHelper.relativePath(from: destination.path)
+        if let audio = assets
+            .filter({ $0.kind.uppercased() == "AUDIO" })
+            .max(by: {
+                ($0.createdAt ?? .distantPast) < ($1.createdAt ?? .distantPast)
+            }) {
+            let localURL = AudioPathHelper.resolveURL(for: session.localAudioPath)
+            let localHash = localURL.flatMap { try? fileSHA256(of: $0) }
+            let needsDownload = localURL == nil
+                || (session.isSynced
+                    && localHash?.caseInsensitiveCompare(audio.sha256) != .orderedSame)
+            if needsDownload {
+                let destination = downloadDestination(for: audio)
+                try await service.downloadAsset(sessionID: remote.id, asset: audio, destinationURL: destination)
+                session.localAudioPath = AudioPathHelper.relativePath(from: destination.path)
+            }
         }
 
         let photoAssets = Dictionary(uniqueKeysWithValues: assets
@@ -322,6 +687,12 @@ final class SyncManager: ObservableObject {
                     modelContext.delete(local)
                     continue
                 }
+                if !local.isSynced,
+                   remote.revision > local.serverRevision {
+                    local.syncState = .conflict
+                    local.lastSyncError = "本地版本 \(local.serverRevision) 与云端版本 \(remote.revision) 不一致"
+                    continue
+                }
                 if local.isSynced, remote.revision > local.serverRevision {
                     apply(remote, to: local, userID: userID)
                 }