NEXT: Cloud and peer chat uploads take files of any size

Packages/macOS/CmuxNext/Sources/CmuxNextAgentPane/AgentPaneFilePut.swift

@@ -6,7 +6,7 @@ public nonisolated enum AgentPaneFilePutFailure: Error, Equatable, Sendable {
     /// This Mac's acpmux or the peer's takes no files (a build from before `_acpmux/file_put`), or
     /// the host has no acpmux: the chat takes no uploads and the composer keeps its own note.
     case unsupported
-    /// Past ``AgentPaneUpload/maximumPeerFileBytes``.
+    /// The peer refused the file as too large (`filePut.tooLarge`); this Mac sets no size limit.
     case tooLarge
     /// No answer, a refusal, or a write that failed on the peer.
     case failed
@@ -45,7 +45,6 @@ extension AcpmuxStatusClient {
     /// there. Throws ``AgentPaneFilePutFailure``.
     @concurrent static func putFile(socketPath: String, sessionId: String, name: String, data: Data,
                                     deadline: Duration = .seconds(120)) async throws -> String {
-        guard data.count <= AgentPaneUpload.maximumPeerFileBytes else { throw AgentPaneFilePutFailure.tooLarge }
         if data.count <= legacyFilePutBytes {
             return try await putFile(sessionId: sessionId, name: name, data: data) { params in
                 try await call(socketPath: socketPath, method: filePutMethod, params: params, deadline: deadline,
@@ -62,7 +61,6 @@ extension AcpmuxStatusClient {
     @concurrent static func putFile(socketPath: String, sessionId: String, name: String, file: URL,
                                     deadline: Duration = .seconds(120)) async throws -> String {
         let size = (try FileManager.default.attributesOfItem(atPath: file.path)[.size] as? NSNumber)?.intValue ?? 0
-        guard size <= AgentPaneUpload.maximumPeerFileBytes else { throw AgentPaneFilePutFailure.tooLarge }
         if size <= legacyFilePutBytes {
             return try await putFile(socketPath: socketPath, sessionId: sessionId, name: name,
                                      data: try Data(contentsOf: file), deadline: deadline)
@@ -76,7 +74,6 @@ extension AcpmuxStatusClient {
     /// ``putFile(socketPath:sessionId:name:data:deadline:)`` with `send`, which runs the request.
     static func putFile(sessionId: String, name: String, data: Data,
                         send: @Sendable (_ params: [String: any Sendable]) async throws -> [String: Any]) async throws -> String {
-        guard data.count <= AgentPaneUpload.maximumPeerFileBytes else { throw AgentPaneFilePutFailure.tooLarge }
         let result: [String: Any]
         do {
             result = try await send(["sessionId": sessionId, "fileName": name, "data": data.base64EncodedString()])
@@ -90,8 +87,7 @@ extension AcpmuxStatusClient {
     /// Streams a larger upload over begin/chunk/finish requests. Each request is a fresh bounded
     /// exchange, so a peer WebSocket never receives the entire base64 file in one frame.
     static func putFileChunked(sessionId: String, name: String, data: Data,
-                               send: @Sendable (_ method: String, _ params: [String: any Sendable]) async throws -> [String: Any]) async throws -> String {
-        guard data.count <= AgentPaneUpload.maximumPeerFileBytes else { throw AgentPaneFilePutFailure.tooLarge }
+                               send: @escaping @Sendable (_ method: String, _ params: [String: any Sendable]) async throws -> [String: Any]) async throws -> String {
         var upload: String?
         let started: [String: Any]
         do {
@@ -112,16 +108,14 @@ extension AcpmuxStatusClient {
             guard let path = finished["path"] as? String, !path.isEmpty else { throw AgentPaneFilePutFailure.failed }
             return path
         } catch {
-            if let upload {
-                try? await send(filePutCancelMethod, ["sessionId": sessionId, "uploadId": upload])
-            }
+            if let upload { await cancelStaged(upload, sessionId: sessionId, send: send) }
             if let failure = error as? AgentPaneFilePutFailure { throw failure }
             throw AgentPaneFilePutFailure(error)
         }
     }
 
     private static func putFileChunked(sessionId: String, name: String, size: Int, file: URL,
-                                       send: @Sendable (_ method: String, _ params: [String: any Sendable]) async throws -> [String: Any]) async throws -> String {
+                                       send: @escaping @Sendable (_ method: String, _ params: [String: any Sendable]) async throws -> [String: Any]) async throws -> String {
         var upload: String?
         do {
             let started = try await send(filePutBeginMethod, ["sessionId": sessionId, "fileName": name, "size": size])
@@ -144,9 +138,21 @@ extension AcpmuxStatusClient {
             guard let path = finished["path"] as? String, !path.isEmpty else { throw AgentPaneFilePutFailure.failed }
             return path
         } catch {
-            if let upload { try? await send(filePutCancelMethod, ["sessionId": sessionId, "uploadId": upload]) }
+            if let upload { await cancelStaged(upload, sessionId: sessionId, send: send) }
             if let failure = error as? AgentPaneFilePutFailure { throw failure }
             throw AgentPaneFilePutFailure(error)
         }
     }
+
+    /// Drops staged upload `upload` on the peer (`_acpmux/file_put_cancel`), from a fresh task: the
+    /// put's own task may be the one that was cancelled (the page's x), and a call made from it is
+    /// torn down before it is sent, which would leave the upload staged on the peer.
+    private static func cancelStaged(_ upload: String, sessionId: String,
+                                     send: @escaping @Sendable (_ method: String, _ params: [String: any Sendable]) async throws -> [String: Any]) async {
+        // task-owner: one cleanup call, bounded by the call's own deadline and awaited right here;
+        // unstructured so it does not inherit the put's cancellation.
+        await Task {
+            _ = try? await send(filePutCancelMethod, ["sessionId": sessionId, "uploadId": upload])
+        }.value
+    }
 }

Packages/macOS/CmuxNext/Sources/CmuxNextAgentPane/AgentPaneUpload.swift

@@ -14,6 +14,40 @@ public nonisolated struct AgentPaneUploadBatchEntry: Sendable, Equatable {
     }
 }
 
+/// A gesture authorizes an ordered batch. A live stage retains that authorization through both
+/// local staging and remote transfer; only an ownerless batch has an abandonment deadline.
+package nonisolated struct AgentPaneUploadBatch: Sendable {
+    private var remaining: [AgentPaneUploadBatchEntry]
+    private var owners: Set<String> = []
+    private var expires: Date
+    package static let lifetime: TimeInterval = 10 * 60
+
+    package init(_ manifest: [AgentPaneUploadBatchEntry], now: Date = Date()) {
+        remaining = manifest
+        expires = now.addingTimeInterval(Self.lifetime)
+    }
+
+    package var abandonmentDeadline: Date? { owners.isEmpty ? expires : nil }
+    package var isComplete: Bool { remaining.isEmpty && owners.isEmpty }
+    package func isAlive(at now: Date = Date()) -> Bool { !owners.isEmpty || expires > now }
+    package func accepts(name: String, size: Int, at now: Date = Date()) -> Bool {
+        isAlive(at: now) && remaining.first == AgentPaneUploadBatchEntry(name: name, size: size)
+    }
+
+    @discardableResult
+    package mutating func begin(name: String, size: Int, stage: String, at now: Date = Date()) -> Bool {
+        guard accepts(name: name, size: size, at: now), !owners.contains(stage) else { return false }
+        remaining.removeFirst()
+        owners.insert(stage)
+        return true
+    }
+
+    package mutating func release(stage: String, at now: Date = Date()) {
+        guard owners.remove(stage) != nil, owners.isEmpty else { return }
+        expires = now.addingTimeInterval(Self.lifetime)
+    }
+}
+
 public nonisolated enum AgentPaneUploadStageRequest: Sendable, Equatable {
     case begin(name: String, size: Int, batchToken: String?, batch: [AgentPaneUploadBatchEntry]?)
     case chunk(token: String, offset: Int, data: Data)
@@ -42,27 +76,17 @@ public nonisolated struct AgentPaneUpload: Sendable, Equatable, CustomStringConv
     public var debugDescription: String { description }
     public var customMirror: Mirror { Mirror(self, children: [:]) }
 
-    /// Local and peer transports stream the bytes in bounded chunks. One GiB is also the native
-    /// daemon's abuse guard; Cloud uploads keep their smaller policy in ``CloudUploads``.
-    public static let maximumFileBytes = 1 * 1024 * 1024 * 1024
-    public static let maximumTotalBytes = 1 * 1024 * 1024 * 1024
-    public static let maximumPeerFileBytes = maximumFileBytes
-
     /// The files of `chat.upload`'s params, nil when they break the contract: one or more files,
-    /// each with a name a folder holds as one file and base64 bytes, at most
-    /// ``maximumFileBytes`` each and ``maximumTotalBytes`` in all.
+    /// each with a name a folder holds as one file and base64 bytes, of any size. This one-frame
+    /// upload is the older page's; the page streams a file of any size through the staged upload
+    /// (``AgentPaneUploadStageRequest``) instead.
     static func files(_ params: [String: Any]?) -> [AgentPaneUpload]? {
         guard let params, Set(params.keys) == ["files"], let list = params["files"] as? [[String: Any]],
               !list.isEmpty else { return nil }
         var files: [AgentPaneUpload] = []
-        var total = 0
         for item in list {
             guard Set(item.keys) == ["name", "data"], let name = item["name"] as? String, isFileName(name),
-                  let text = item["data"] as? String, text.utf8.count <= (maximumFileBytes / 3 + 1) * 4,
-                  let data = Data(base64Encoded: text), data.count <= maximumFileBytes else { return nil }
-            let (sum, overflow) = total.addingReportingOverflow(data.count)
-            total = overflow ? Int.max : sum
-            guard total <= maximumTotalBytes else { return nil }
+                  let text = item["data"] as? String, let data = Data(base64Encoded: text) else { return nil }
             files.append(AgentPaneUpload(name: name, data: data))
         }
         return files