diff --git a/FrameworkTests/SFTPSettingsSyncTests.swift b/FrameworkTests/SFTPSettingsSyncTests.swift index 781a98bd..a5a988ea 100644 --- a/FrameworkTests/SFTPSettingsSyncTests.swift +++ b/FrameworkTests/SFTPSettingsSyncTests.swift @@ -15,7 +15,7 @@ struct SFTPSettingsSyncTests { func importsExistingValuesOnlyOnce(global: Int?) throws { let fixture = try Fixture(global: global, perHost: global == nil ? nil : 2) defer { fixture.cleanUp() } - let expected = SFTPPreferences(globalLimit: global ?? 15, perHostLimit: global == nil ? 5 : 2) + let expected = SFTPPreferences(globalLimit: global ?? 30, perHostLimit: global == nil ? 10 : 2) #expect(try fixture.store.loadSFTP() == expected) #expect(fixture.defaults.integer(forKey: SettingsKey.sftpGlobalTransferLimit) == expected.globalLimit) let original = try #require(try fixture.store.journal.envelope(for: key)) @@ -52,7 +52,7 @@ struct SFTPSettingsSyncTests { } var localChanges: [SettingsMutationSet] = [] source.store.didRecordChanges = { localChanges.append($0) } - let value = SFTPPreferences(globalLimit: 100, perHostLimit: 50) + let value = SFTPPreferences(globalLimit: 100, perHostLimit: 50, strategy: .connectionsPreferred) try source.store.saveSFTP(value) #expect(localChanges.last?.keys == [key]) let outgoing = try #require(try source.store.journal.envelope(for: key)) @@ -71,6 +71,9 @@ struct SFTPSettingsSyncTests { #expect(try destination.store.loadSFTP() == value) #expect(destination.defaults.integer(forKey: SettingsKey.sftpGlobalTransferLimit) == 100) #expect(destination.defaults.integer(forKey: SettingsKey.sftpPerHostTransferLimit) == 50) + #expect( + destination.defaults.string(forKey: SettingsKey.sftpConcurrencyStrategy) + == SFTPConcurrencyStrategy.connectionsPreferred.rawValue) #expect(callbacks == 0) #expect(result.acceptedKeys == [key]) #expect(try !destination.store.journal.pendingEnvelopes().contains { $0.key == key }) @@ -122,7 +125,7 @@ struct SFTPSettingsSyncTests { #expect(try SettingsSyncCoordinator.payloadData(from: record) == bytes) #expect(throws: SettingsSyncPersistenceError.self) { try fixture.store.applyRemote([invalid]) } #expect(try fixture.store.loadSFTP() == SFTPPreferences()) - #expect(fixture.defaults.integer(forKey: SettingsKey.sftpGlobalTransferLimit) == 15) + #expect(fixture.defaults.integer(forKey: SettingsKey.sftpGlobalTransferLimit) == 30) } @Test func cloudRecordTypeMustMatchItsPayloadAndRestoredSystemFields() throws { @@ -154,7 +157,7 @@ struct SFTPSettingsSyncTests { #expect(throws: (any Error).self) { try fixture.store.saveSFTP(.init(globalLimit: 100, perHostLimit: 50)) } #expect(try fixture.store.loadSFTP() == SFTPPreferences()) #expect(try fixture.store.journal.envelope(for: key) == before) - #expect(fixture.defaults.integer(forKey: SettingsKey.sftpGlobalTransferLimit) == 15) + #expect(fixture.defaults.integer(forKey: SettingsKey.sftpGlobalTransferLimit) == 30) #expect(callbacks == 0) } diff --git a/FrameworkTests/SFTPSettingsTests.swift b/FrameworkTests/SFTPSettingsTests.swift index 6308b0ad..3f0aec08 100644 --- a/FrameworkTests/SFTPSettingsTests.swift +++ b/FrameworkTests/SFTPSettingsTests.swift @@ -6,9 +6,9 @@ import Testing @Suite @MainActor struct SFTPSettingsTests { @Test func defaultsAndRangesHaveOneSharedDefinition() { - #expect(SFTPPreferences() == SFTPPreferences(globalLimit: 15, perHostLimit: 5)) - #expect(SettingsKey.defaultSFTPGlobalTransferLimit == 15) - #expect(SettingsKey.defaultSFTPPerHostTransferLimit == 5) + #expect(SFTPPreferences() == SFTPPreferences(globalLimit: 30, perHostLimit: 10, strategy: .smart)) + #expect(SettingsKey.defaultSFTPGlobalTransferLimit == 30) + #expect(SettingsKey.defaultSFTPPerHostTransferLimit == 10) #expect(SettingsKey.sftpGlobalTransferLimitRange == 1...100) #expect(SettingsKey.sftpPerHostTransferLimitRange == 1...50) } @@ -37,7 +37,7 @@ struct SFTPSettingsTests { persistence.fail = false #expect(model.set(100, for: \.globalLimit)) #expect(model.storageError == nil) - #expect(received == [SFTPPreferences(globalLimit: 100, perHostLimit: 5)]) + #expect(received == [SFTPPreferences(globalLimit: 100, perHostLimit: 10)]) } @Test func remoteChangesNotifyOnceWithoutSavingAndLocalChangesCommitThePair() { @@ -55,6 +55,25 @@ struct SFTPSettingsTests { #expect(persistence.saves == 1) #expect(received.last == persistence.value) } + + @Test func strategyPersistsAndOlderJSONDefaultsToSmart() throws { + let legacy = try JSONDecoder().decode( + SFTPPreferences.self, + from: Data(#"{"globalLimit":15,"perHostLimit":5}"#.utf8)) + #expect(legacy == SFTPPreferences(globalLimit: 15, perHostLimit: 5, strategy: .smart)) + + let persistence = TestSFTPPreferencesPersistence() + let model = SFTPSettingsModel(persistence: persistence) + #expect(model.setStrategy(.channelsOnly)) + #expect(persistence.value.strategy == .channelsOnly) + #expect(persistence.saves == 1) + + #expect(throws: DecodingError.self) { + _ = try JSONDecoder().decode( + SFTPPreferences.self, + from: Data(#"{"globalLimit":30,"perHostLimit":10,"strategy":"future"}"#.utf8)) + } + } } @MainActor diff --git a/Frameworks/Resources/Localizable.xcstrings b/Frameworks/Resources/Localizable.xcstrings index 9e3af211..b31a0f90 100644 --- a/Frameworks/Resources/Localizable.xcstrings +++ b/Frameworks/Resources/Localizable.xcstrings @@ -4486,19 +4486,19 @@ } } }, - "Changes apply to queued transfers. Active transfers continue unchanged. If additional connections cannot open without authentication prompts, the session transfers files one at a time." : { + "Changes apply to queued transfers. Active transfers continue unchanged. If additional connections cannot open without authentication prompts, transfers use parallel channels on the existing connection." : { "extractionState" : "manual", "localizations" : { "zh-Hans" : { "stringUnit" : { "state" : "translated", - "value" : "更改会应用到排队中的传输。正在进行的传输不受影响。如果额外连接需要身份验证提示，会话将逐个传输文件。" + "value" : "更改会应用到排队中的传输。正在进行的传输不受影响。如果额外连接无法在不显示身份验证提示的情况下打开，传输将改用现有连接上的并行通道。" } }, "zh-Hant" : { "stringUnit" : { "state" : "translated", - "value" : "更改會套用至佇列中的傳輸。正在進行的傳輸不受影響。如果額外連線需要驗證提示，工作階段將逐一傳輸檔案。" + "value" : "更改會套用至佇列中的傳輸。正在進行的傳輸不受影響。如果額外連線無法在不顯示驗證提示的情況下開啟，傳輸將改用現有連線上的平行通道。" } } } @@ -5667,6 +5667,57 @@ } } }, + "Channels Only" : { + "extractionState" : "manual", + "localizations" : { + "zh-Hans" : { + "stringUnit" : { + "state" : "translated", + "value" : "仅通道" + } + }, + "zh-Hant" : { + "stringUnit" : { + "state" : "translated", + "value" : "僅通道" + } + } + } + }, + "Concurrent Strategy" : { + "extractionState" : "manual", + "localizations" : { + "zh-Hans" : { + "stringUnit" : { + "state" : "translated", + "value" : "并发策略" + } + }, + "zh-Hant" : { + "stringUnit" : { + "state" : "translated", + "value" : "並行策略" + } + } + } + }, + "Connections Preferred" : { + "extractionState" : "manual", + "localizations" : { + "zh-Hans" : { + "stringUnit" : { + "state" : "translated", + "value" : "连接优先" + } + }, + "zh-Hant" : { + "stringUnit" : { + "state" : "translated", + "value" : "連線優先" + } + } + } + }, "Concurrent Transfers" : { "extractionState" : "manual", "localizations" : { @@ -5684,6 +5735,40 @@ } } }, + "Smart" : { + "extractionState" : "manual", + "localizations" : { + "zh-Hans" : { + "stringUnit" : { + "state" : "translated", + "value" : "智能" + } + }, + "zh-Hant" : { + "stringUnit" : { + "state" : "translated", + "value" : "智慧" + } + } + } + }, + "Smart uses channels for files up to 64 MB and prefers additional connections for larger files. Transfers with an unknown size use channels." : { + "extractionState" : "manual", + "localizations" : { + "zh-Hans" : { + "stringUnit" : { + "state" : "translated", + "value" : "智能模式对不超过64 MB的文件使用通道，对更大的文件优先使用额外连接。大小未知的传输使用通道。" + } + }, + "zh-Hant" : { + "stringUnit" : { + "state" : "translated", + "value" : "智慧模式對不超過64 MB的檔案使用通道，對更大的檔案優先使用額外連線。大小未知的傳輸使用通道。" + } + } + } + }, "Configure Model Providers" : { "localizations" : { "zh-Hans" : { diff --git a/Frameworks/TermindApplication/FeatureModels/SFTPSettingsModel.swift b/Frameworks/TermindApplication/FeatureModels/SFTPSettingsModel.swift index f03719fb..9b491dd4 100644 --- a/Frameworks/TermindApplication/FeatureModels/SFTPSettingsModel.swift +++ b/Frameworks/TermindApplication/FeatureModels/SFTPSettingsModel.swift @@ -32,6 +32,21 @@ public final class SFTPSettingsModel { } } + @discardableResult + public func setStrategy(_ value: SFTPConcurrencyStrategy) -> Bool { + var next = preferences + next.strategy = value + guard next != preferences else { return true } + do { + try persistence?.saveSFTP(next) + applyRemoteSnapshot(next) + return true + } catch { + storageError = error.localizedDescription + return false + } + } + public func applyRemoteSnapshot(_ value: SFTPPreferences) { let next = value.normalized let changed = preferences != next diff --git a/Frameworks/TermindApplication/SettingsKey.swift b/Frameworks/TermindApplication/SettingsKey.swift index 79cda6e4..09e0e479 100644 --- a/Frameworks/TermindApplication/SettingsKey.swift +++ b/Frameworks/TermindApplication/SettingsKey.swift @@ -29,6 +29,7 @@ public nonisolated enum SettingsKey { public static let settingsSyncEnabled = "settings.sync.enabled" public static let settingsSyncEnvironment = "settings.sync.environment" public static let shellIntegration = "terminal.shellIntegration" + public static let sftpConcurrencyStrategy = "sftp.transfers.strategy" public static let sftpGlobalTransferLimit = "sftp.transfers.globalLimit" public static let sftpPerHostTransferLimit = "sftp.transfers.perHostLimit" public static let defaultSFTPGlobalTransferLimit = SFTPPreferences.defaultGlobalLimit diff --git a/Frameworks/TermindDomain/Transfers/SFTPPreferences.swift b/Frameworks/TermindDomain/Transfers/SFTPPreferences.swift index b35a6033..e700e8ed 100644 --- a/Frameworks/TermindDomain/Transfers/SFTPPreferences.swift +++ b/Frameworks/TermindDomain/Transfers/SFTPPreferences.swift @@ -1,23 +1,39 @@ import Foundation +public nonisolated enum SFTPConcurrencyStrategy: String, Codable, CaseIterable, Hashable, Sendable { + case smart + case channelsOnly = "channels_only" + case connectionsPreferred = "connections_preferred" + + public static let smartConnectionThreshold: Int64 = 64_000_000 +} + /// File-transfer concurrency shared by terminal uploads and SFTP browsers. public nonisolated struct SFTPPreferences: Codable, Hashable, Sendable { - public static let defaultGlobalLimit = 15 - public static let defaultPerHostLimit = 5 + public static let defaultGlobalLimit = 30 + public static let defaultPerHostLimit = 10 public static let globalLimitRange = 1...100 public static let perHostLimitRange = 1...50 public var globalLimit: Int public var perHostLimit: Int + public var strategy: SFTPConcurrencyStrategy - public init(globalLimit: Int = defaultGlobalLimit, perHostLimit: Int = defaultPerHostLimit) { + public init( + globalLimit: Int = defaultGlobalLimit, + perHostLimit: Int = defaultPerHostLimit, + strategy: SFTPConcurrencyStrategy = .smart + ) { self.globalLimit = min(max(globalLimit, Self.globalLimitRange.lowerBound), Self.globalLimitRange.upperBound) self.perHostLimit = min( max(perHostLimit, Self.perHostLimitRange.lowerBound), Self.perHostLimitRange.upperBound, self.globalLimit) + self.strategy = strategy } - public var normalized: Self { Self(globalLimit: globalLimit, perHostLimit: perHostLimit) } + public var normalized: Self { + Self(globalLimit: globalLimit, perHostLimit: perHostLimit, strategy: strategy) + } /// Decoded remote values must be validated before entering the sync journal. public var isValid: Bool { @@ -28,4 +44,22 @@ public nonisolated struct SFTPPreferences: Codable, Hashable, Sendable { public var availablePerHostRange: ClosedRange { Self.perHostLimitRange.lowerBound...min(Self.perHostLimitRange.upperBound, normalized.globalLimit) } + + private enum CodingKeys: String, CodingKey { + case globalLimit, perHostLimit, strategy + } + + public init(from decoder: any Decoder) throws { + let values = try decoder.container(keyedBy: CodingKeys.self) + globalLimit = try values.decodeIfPresent(Int.self, forKey: .globalLimit) ?? Self.defaultGlobalLimit + perHostLimit = try values.decodeIfPresent(Int.self, forKey: .perHostLimit) ?? Self.defaultPerHostLimit + strategy = try values.decodeIfPresent(SFTPConcurrencyStrategy.self, forKey: .strategy) ?? .smart + } + + public func encode(to encoder: any Encoder) throws { + var values = encoder.container(keyedBy: CodingKeys.self) + try values.encode(globalLimit, forKey: .globalLimit) + try values.encode(perHostLimit, forKey: .perHostLimit) + try values.encode(strategy, forKey: .strategy) + } } diff --git a/Frameworks/TermindPersistence/PersistentSFTPPreferences.swift b/Frameworks/TermindPersistence/PersistentSFTPPreferences.swift index 106d00a8..a1d43c86 100644 --- a/Frameworks/TermindPersistence/PersistentSFTPPreferences.swift +++ b/Frameworks/TermindPersistence/PersistentSFTPPreferences.swift @@ -19,11 +19,14 @@ extension SFTPPreferences { globalLimit: defaults.object(forKey: SettingsKey.sftpGlobalTransferLimit) as? Int ?? Self.defaultGlobalLimit, perHostLimit: defaults.object(forKey: SettingsKey.sftpPerHostTransferLimit) as? Int - ?? Self.defaultPerHostLimit) + ?? Self.defaultPerHostLimit, + strategy: defaults.string(forKey: SettingsKey.sftpConcurrencyStrategy) + .flatMap(SFTPConcurrencyStrategy.init(rawValue:)) ?? .smart) } public func mirror(to defaults: UserDefaults) { defaults.set(globalLimit, forKey: SettingsKey.sftpGlobalTransferLimit) defaults.set(perHostLimit, forKey: SettingsKey.sftpPerHostTransferLimit) + defaults.set(strategy.rawValue, forKey: SettingsKey.sftpConcurrencyStrategy) } } diff --git a/Frameworks/TermindPresentation/SFTP/SFTPSettingsFields.swift b/Frameworks/TermindPresentation/SFTP/SFTPSettingsFields.swift index 2b78aa72..59f1ef55 100644 --- a/Frameworks/TermindPresentation/SFTP/SFTPSettingsFields.swift +++ b/Frameworks/TermindPresentation/SFTP/SFTPSettingsFields.swift @@ -8,6 +8,17 @@ public struct SFTPSettingsFields: View { public init(model: SFTPSettingsModel) { self.model = model } public var body: some View { + Picker("Concurrent Strategy", selection: strategyBinding) { + Text("Smart").tag(SFTPConcurrencyStrategy.smart) + Text("Channels Only").tag(SFTPConcurrencyStrategy.channelsOnly) + Text("Connections Preferred").tag(SFTPConcurrencyStrategy.connectionsPreferred) + } + if model.preferences.strategy == .smart { + Text( + "Smart uses channels for files up to 64 MB and prefers additional connections for larger files. Transfers with an unknown size use channels." + ) + .font(.caption).foregroundStyle(.secondary) + } Stepper(value: binding(\.globalLimit), in: SFTPPreferences.globalLimitRange) { label("Concurrent Transfers", value: model.preferences.globalLimit) } @@ -15,7 +26,7 @@ public struct SFTPSettingsFields: View { label("Per-Host Transfers", value: model.preferences.perHostLimit) } Text( - "Changes apply to queued transfers. Active transfers continue unchanged. If additional connections cannot open without authentication prompts, the session transfers files one at a time." + "Changes apply to queued transfers. Active transfers continue unchanged. If additional connections cannot open without authentication prompts, transfers use parallel channels on the existing connection." ) .font(.caption).foregroundStyle(.secondary) if let error = model.storageError { Text(error).foregroundStyle(.red) } @@ -39,4 +50,8 @@ public struct SFTPSettingsFields: View { private func binding(_ keyPath: WritableKeyPath) -> Binding { Binding(get: { model.preferences[keyPath: keyPath] }, set: { model.set($0, for: keyPath) }) } + + private var strategyBinding: Binding { + Binding(get: { model.preferences.strategy }, set: { model.setStrategy($0) }) + } } diff --git a/Frameworks/TermindRemoteInfrastructure/Agent/AgentBuiltInSFTPTools.swift b/Frameworks/TermindRemoteInfrastructure/Agent/AgentBuiltInSFTPTools.swift index 9317ffef..628a17ba 100644 --- a/Frameworks/TermindRemoteInfrastructure/Agent/AgentBuiltInSFTPTools.swift +++ b/Frameworks/TermindRemoteInfrastructure/Agent/AgentBuiltInSFTPTools.swift @@ -13,8 +13,10 @@ nonisolated public protocol SFTPToolDefinition: AgentToolDefinition, AgentToolSc static var extraProperties: [String: AgentJSONValue] { get } static var extraRequired: [String] { get } static var presentation: AgentToolDescriptor.Presentation { get } + static var schedulesDataTransfer: Bool { get } var authentication: AgentSSHAuthenticationRegistry { get } var connections: AgentConnectionPool { get } + func estimatedByteCount(arguments: Arguments) throws -> Int64? func perform( client: any SFTPClientProtocol, path: String, arguments: Arguments, invocation: AgentToolInvocation @@ -27,6 +29,9 @@ nonisolated extension SFTPToolDefinition { } public static var extraProperties: [String: AgentJSONValue] { [:] } public static var extraRequired: [String] { [] } + public static var schedulesDataTransfer: Bool { false } + + public func estimatedByteCount(arguments _: Arguments) throws -> Int64? { nil } public var descriptor: AgentToolDescriptor { var properties: [String: AgentJSONValue] = [ @@ -60,7 +65,9 @@ nonisolated extension SFTPToolDefinition { let path = try AgentToolSupport.normalizeRemotePath(arguments.string("path")) return try await AgentToolSupport.withAuthenticatedSFTP( hostID: hostID, invocation: invocation, context: context, - authentication: authentication, pool: connections + authentication: authentication, pool: connections, + byteCount: try estimatedByteCount(arguments: arguments), + schedulesDataTransfer: Self.schedulesDataTransfer ) { client in try await perform( client: client, path: path, arguments: arguments, invocation: invocation @@ -131,6 +138,7 @@ nonisolated public struct SFTPReadTool: SFTPToolDefinition { "offset": integer(min: 1, max: 10_000_000), "limit": integer(min: 1, max: 2_000), ] public static let approvalPolicy: AgentToolDescriptor.ApprovalPolicy = .approve + public static let schedulesDataTransfer = true public static let presenter = AgentToolDetailPresenter( id: "builtin.sftp_read", present: present ) @@ -140,6 +148,7 @@ nonisolated public struct SFTPReadTool: SFTPToolDefinition { ) public let authentication: AgentSSHAuthenticationRegistry public let connections: AgentConnectionPool + public func estimatedByteCount(arguments _: Arguments) throws -> Int64? { 50 * 1_024 } public func perform( client: any SFTPClientProtocol, path: String, arguments: Arguments, invocation: AgentToolInvocation @@ -171,6 +180,7 @@ nonisolated public struct SFTPWriteTool: SFTPToolDefinition { public static let name = "sftp_write" public static let summary = "Write a bounded UTF-8 remote file atomically." public static let approvalPolicy: AgentToolDescriptor.ApprovalPolicy = .ask + public static let schedulesDataTransfer = true public static let extraProperties = ["content": string(max: 131_072), "overwrite": boolean()] public static let extraRequired = ["content"] public static let presenter = AgentToolDetailPresenter( @@ -184,6 +194,9 @@ nonisolated public struct SFTPWriteTool: SFTPToolDefinition { ) public let authentication: AgentSSHAuthenticationRegistry public let connections: AgentConnectionPool + public func estimatedByteCount(arguments: Arguments) throws -> Int64? { + Int64(try arguments.string("content").utf8.count) + } public func perform( client: any SFTPClientProtocol, path: String, arguments: Arguments, invocation: AgentToolInvocation @@ -620,11 +633,15 @@ nonisolated extension AgentToolSupport { hostID: Host.ID, invocation: AgentToolInvocation, context: AgentToolExecutionContext, authentication: AgentSSHAuthenticationRegistry, pool: AgentConnectionPool, + byteCount: Int64? = nil, + schedulesDataTransfer: Bool = false, operation: @escaping @Sendable (any SFTPClientProtocol) async throws -> Value ) async throws -> Value { try await authentication.withSFTP( hostID: hostID, invocationID: invocation.id, - interaction: context.sshInteraction, pool: pool + interaction: context.sshInteraction, pool: pool, + runID: context.runID, byteCount: byteCount, + schedulesDataTransfer: schedulesDataTransfer ) { client in do { return try await operation(client) diff --git a/Frameworks/TermindRemoteInfrastructure/Agent/AgentBuiltInSSHTools.swift b/Frameworks/TermindRemoteInfrastructure/Agent/AgentBuiltInSSHTools.swift index 7e6fe53c..ae786bcf 100644 --- a/Frameworks/TermindRemoteInfrastructure/Agent/AgentBuiltInSSHTools.swift +++ b/Frameworks/TermindRemoteInfrastructure/Agent/AgentBuiltInSSHTools.swift @@ -338,12 +338,11 @@ nonisolated public struct SSHExecTool: AgentToolDefinition, AgentToolSchemaBuild // turn serial, so nothing observes the host while it is // being changed. // - // The ceiling on what this buys is worth stating: commands - // aimed at one host queue behind each other inside - // `SSHExecSession` regardless, because a libssh2 session - // cannot be driven from two places at once. The win is - // several hosts at once, and host work overlapping `fetch`, - // `web_search`, and MCP. + // Safe commands may overlap even when aimed at one host: + // `SSHExecSession` gives each command its own Channel on the + // shared transport, while one fair owner serializes the actual + // libssh2 calls. Commands on separate hosts can also run at once, + // as can host work alongside `fetch`, `web_search`, and MCP. concurrency: safe ? .parallel : .sequential ) } diff --git a/Frameworks/TermindRemoteInfrastructure/Agent/AgentConnections.swift b/Frameworks/TermindRemoteInfrastructure/Agent/AgentConnections.swift index 98884bea..0bfd07aa 100644 --- a/Frameworks/TermindRemoteInfrastructure/Agent/AgentConnections.swift +++ b/Frameworks/TermindRemoteInfrastructure/Agent/AgentConnections.swift @@ -790,42 +790,66 @@ public actor AgentSSHAuthenticationRegistry { hostID: Host.ID, invocationID: AgentToolInvocation.ID, interaction: AgentSSHInteraction, pool: AgentConnectionPool, cancellation: TransferCancellationToken? = nil, + runID: UUID? = nil, + byteCount: Int64? = nil, + schedulesDataTransfer: Bool = false, + concurrencyPermit: SFTPConcurrencyPermit? = nil, operation: @escaping @Sendable (any SFTPClientProtocol) async throws -> Value ) async throws -> Value { - try await withSession( - hostID: hostID, invocationID: invocationID, interaction: interaction - ) { authentication in - let lease = try await self.openAuthenticated( - hostID: hostID, authentication: authentication, pool: pool, kind: .sftp, - evictsPooledConnection: false, cancellation: cancellation - ) { attempt in - let resolver: AgentConnectionPool.RequestResolving? - if let attempt { - let request = attempt.request - resolver = { _ in request } - } else { - resolver = nil - } - return try await pool.acquireSFTP( - hostID: hostID, cancellation: cancellation, - configuration: resolver) - } - do { - let value = try await operation(lease.client) - await pool.releaseSFTP(lease.id) - return value - } catch { - await pool.releaseSFTP(lease.id) - if let failure = error as? SFTPError { - switch failure { - case .notConnected, .connectionLost: - await pool.disconnect(hostID: hostID, kind: .sftp) - case .cancelled, .localFile, .operation: - break + let ownedPermit: SFTPConcurrencyPermit? + if schedulesDataTransfer, concurrencyPermit == nil { + ownedPermit = try await pool.sftpConcurrencyScheduler.acquire( + anchorHostID: hostID, + hostIDs: [hostID], + source: .agentRun(runID ?? invocationID), + byteCount: byteCount + ) + } else { + ownedPermit = nil + } + let permit = concurrencyPermit ?? ownedPermit + do { + let value = try await withSession( + hostID: hostID, invocationID: invocationID, interaction: interaction + ) { authentication in + let lease = try await self.openAuthenticated( + hostID: hostID, authentication: authentication, pool: pool, kind: .sftp, + evictsPooledConnection: false, cancellation: cancellation + ) { attempt in + let resolver: AgentConnectionPool.RequestResolving? + if let attempt { + let request = attempt.request + resolver = { _ in request } + } else { + resolver = nil } + return try await pool.acquireSFTP( + hostID: hostID, cancellation: cancellation, + configuration: resolver, + mode: permit?.schedulingMode ?? .channelsOnly) + } + do { + let value = try await operation(lease.client) + await pool.releaseSFTP(lease.id) + return value + } catch { + await pool.releaseSFTP(lease.id) + if let failure = error as? SFTPError { + switch failure { + case .notConnected, .connectionLost: + await pool.disconnect(hostID: hostID, kind: .sftp) + case .cancelled, .localFile, .operation: + break + } + } + throw error } - throw error } + await ownedPermit?.release() + return value + } catch { + await ownedPermit?.release() + throw error } } @@ -907,11 +931,11 @@ public actor AgentSSHAuthenticationRegistry { /// is what gives it back. /// /// Exec and SFTP remain separate lifecycle domains. Exec never opens a second -/// transport; overlapping SFTP leases prefer additional connections and fall -/// back to sibling Channels on the primary transport. +/// transport; SFTP data permits resolve their topology from the process-wide +/// preference, while control work and Channel fallback use the primary transport. public actor AgentConnectionPool { public nonisolated static let execSchedulingMode = SSHChannelSchedulingMode.channelsOnly - public nonisolated static let sftpSchedulingMode = SSHChannelSchedulingMode.connectionsPreferred + public nonisolated let sftpConcurrencyScheduler: SFTPConcurrencyScheduler public nonisolated enum Kind: String, Hashable, Sendable, CaseIterable { case exec, sftp } @@ -1009,7 +1033,8 @@ public actor AgentConnectionPool { /// SFTP Channel on the primary transport instead. func acquireSFTP( hostID: UUID, cancellation: TransferCancellationToken? = nil, - configuration override: RequestResolving? = nil + configuration override: RequestResolving? = nil, + mode: SSHChannelSchedulingMode = .connectionsPreferred ) async throws -> SFTPLease { try Task.checkCancellation() if cancellation?.isCancelled == true { throw SFTPError.cancelled } @@ -1029,7 +1054,7 @@ public actor AgentConnectionPool { } if let idle = sftpStates[hostID]?.records.values - .filter({ !$0.busy }) + .filter({ !$0.busy && (mode == .connectionsPreferred || $0.id == sftpStates[hostID]?.primaryID) }) .sorted(by: { $0.lastUsedAt < $1.lastUsedAt }).first { sftpStates[hostID]?.records[idle.id]?.busy = true @@ -1042,6 +1067,31 @@ public actor AgentConnectionPool { throw SFTPError.notConnected } + if mode == .channelsOnly { + do { + let sibling = try await primaryRecord.client.openSiblingChannel() + if cancellation?.isCancelled == true { + await sibling.disconnect() + throw SFTPError.cancelled + } + return registerSFTPLease( + hostID: hostID, recordID: nil, client: sibling, cancellation: cancellation) + } catch { + try Task.checkCancellation() + if cancellation?.isCancelled == true { throw SFTPError.cancelled } + if error is CancellationError { throw error } + if let failure = error as? SFTPError, case .cancelled = failure { throw failure } + while sftpStates[hostID]?.records[primaryRecord.id]?.busy == true { + try await Task.sleep(for: .milliseconds(10)) + try Task.checkCancellation() + if cancellation?.isCancelled == true { throw SFTPError.cancelled } + } + return try await acquireSFTP( + hostID: hostID, cancellation: cancellation, + configuration: override, mode: mode) + } + } + if state.capability == .unknown { let probe: SFTPProbe if let existing = sftpProbes[hostID] { @@ -1068,7 +1118,7 @@ public actor AgentConnectionPool { } return try await acquireSFTP( hostID: hostID, cancellation: cancellation, - configuration: override) + configuration: override, mode: mode) } catch { if sftpProbes[hostID]?.id == probe.id { sftpProbes[hostID] = nil @@ -1138,6 +1188,7 @@ public actor AgentConnectionPool { routeFingerprint: @escaping FingerprintResolving, openExec: @escaping ExecOpening, openSFTP: @escaping SFTPOpening, + sftpConcurrencyScheduler: SFTPConcurrencyScheduler = SFTPConcurrencyScheduler(), idleTimeout: Duration = AgentConnectionPool.defaultIdleTimeout, sweepInterval: Duration? = .seconds(30) ) { @@ -1145,6 +1196,7 @@ public actor AgentConnectionPool { self.routeFingerprint = routeFingerprint self.openExec = openExec self.openSFTP = openSFTP + self.sftpConcurrencyScheduler = sftpConcurrencyScheduler self.idleTimeout = idleTimeout sweeper = AgentIdleSweeper(interval: sweepInterval) } diff --git a/Frameworks/TermindRemoteInfrastructure/Agent/AgentHostServices.swift b/Frameworks/TermindRemoteInfrastructure/Agent/AgentHostServices.swift index 1bbcd97e..b006352f 100644 --- a/Frameworks/TermindRemoteInfrastructure/Agent/AgentHostServices.swift +++ b/Frameworks/TermindRemoteInfrastructure/Agent/AgentHostServices.swift @@ -24,7 +24,8 @@ public struct AgentHostServices { public static func live( store: HostLibraryModel, - coordinator: RemoteConnectionCoordinator = RemoteConnectionCoordinator() + coordinator: RemoteConnectionCoordinator = RemoteConnectionCoordinator(), + sftpConcurrencyScheduler: SFTPConcurrencyScheduler = SFTPConcurrencyScheduler() ) -> AgentHostServices { let resolver = AgentHostResolver(store: store) return AgentHostServices( @@ -33,7 +34,8 @@ public struct AgentHostServices { configuration: { try await resolver.configuration(for: $0) }, routeFingerprint: { try await resolver.routeFingerprint(for: $0) }, openExec: { try await coordinator.connectExec(using: $0) }, - openSFTP: { try await coordinator.connectSFTP(using: $0) } + openSFTP: { try await coordinator.connectSFTP(using: $0) }, + sftpConcurrencyScheduler: sftpConcurrencyScheduler ), authentication: AgentSSHAuthenticationRegistry(resolver: resolver), background: AgentBackgroundCommands() diff --git a/Frameworks/TermindRemoteInfrastructure/Agent/AgentScratchTransferTools.swift b/Frameworks/TermindRemoteInfrastructure/Agent/AgentScratchTransferTools.swift index f2a8601b..488f6181 100644 --- a/Frameworks/TermindRemoteInfrastructure/Agent/AgentScratchTransferTools.swift +++ b/Frameworks/TermindRemoteInfrastructure/Agent/AgentScratchTransferTools.swift @@ -54,7 +54,8 @@ nonisolated public struct ScratchPullTool: AgentToolDefinition, AgentToolSchemaB let hostID = try arguments.uuid("host_id") let data = try await AgentToolSupport.withAuthenticatedSFTP( hostID: hostID, invocation: invocation, context: context, - authentication: authentication, pool: pool + authentication: authentication, pool: pool, + schedulesDataTransfer: true ) { client in try await client.readFile(at: remote, maximumBytes: AgentScratchWorkspace.Limits.fileBytes) } @@ -137,7 +138,8 @@ nonisolated public struct ScratchPushTool: AgentToolDefinition, AgentToolSchemaB let hostID = try arguments.uuid("host_id") try await AgentToolSupport.withAuthenticatedSFTP( hostID: hostID, invocation: invocation, context: context, - authentication: authentication, pool: pool + authentication: authentication, pool: pool, + byteCount: Int64(file.bytes), schedulesDataTransfer: true ) { client in try await client.writeFile( file.data, at: remote, overwrite: arguments.optionalBool("overwrite") ?? false diff --git a/Frameworks/TermindRemoteInfrastructure/Agent/SFTPCopyTool.swift b/Frameworks/TermindRemoteInfrastructure/Agent/SFTPCopyTool.swift index 08608bf4..c34411e4 100644 --- a/Frameworks/TermindRemoteInfrastructure/Agent/SFTPCopyTool.swift +++ b/Frameworks/TermindRemoteInfrastructure/Agent/SFTPCopyTool.swift @@ -96,29 +96,67 @@ nonisolated public struct SFTPCopyTool: AgentToolDefinition, AgentToolSchemaBuil let request = try Request(invocation) let cancellation = TransferCancellationToken() return try await withTaskCancellationHandler { - try await authentication.withSFTP( - hostID: request.sourceHostID, invocationID: invocation.id, - interaction: context.sshInteraction, pool: connections, cancellation: cancellation - ) { source in - try await authentication.withSFTP( - hostID: request.destinationHostID, invocationID: invocation.id, - interaction: context.sshInteraction, pool: connections, cancellation: cancellation - ) { destination in - let outcome = await SFTPRemoteCopyOperation.run( - sourcePath: request.sourcePath, destinationPath: request.destinationPath, - sameHost: request.sourceHostID == request.destinationHostID, - source: source, destination: destination, policy: request.policy, cancellation: cancellation - ) { outcome, progress in - context.report(Self.copyResult(invocation, request, context, outcome, progress: progress)) + let permit = try await connections.sftpConcurrencyScheduler.acquire( + anchorHostID: request.destinationHostID, + hostIDs: [request.sourceHostID, request.destinationHostID], + source: .agentRun(context.runID), + byteCount: nil + ) + do { + let result = try await authentication.withSFTP( + hostID: request.sourceHostID, invocationID: invocation.id, + interaction: context.sshInteraction, pool: connections, cancellation: cancellation, + runID: context.runID, schedulesDataTransfer: true, concurrencyPermit: permit + ) { source in + if request.sourceHostID == request.destinationHostID, + permit.schedulingMode == .channelsOnly + { + return await performCopy( + source: source, destination: source, + request: request, invocation: invocation, + context: context, cancellation: cancellation) + } + return try await authentication.withSFTP( + hostID: request.destinationHostID, invocationID: invocation.id, + interaction: context.sshInteraction, pool: connections, cancellation: cancellation, + runID: context.runID, schedulesDataTransfer: true, concurrencyPermit: permit + ) { destination in + await performCopy( + source: source, destination: destination, + request: request, invocation: invocation, + context: context, cancellation: cancellation) } - return Self.copyResult(invocation, request, context, outcome) } + await permit.release() + return result + } catch { + await permit.release() + throw error } } onCancel: { cancellation.cancel() } } + private func performCopy( + source: any SFTPClientProtocol, + destination: any SFTPClientProtocol, + request: Request, + invocation: AgentToolInvocation, + context: AgentToolExecutionContext, + cancellation: TransferCancellationToken + ) async -> AgentToolResult { + let outcome = await SFTPRemoteCopyOperation.run( + sourcePath: request.sourcePath, destinationPath: request.destinationPath, + sameHost: request.sourceHostID == request.destinationHostID, + source: source, destination: destination, policy: request.policy, + cancellation: cancellation + ) { outcome, progress in + context.report(Self.copyResult(invocation, request, context, outcome, progress: progress)) + } + return Self.copyResult(invocation, request, context, outcome) + } + private static func copyResult( _ invocation: AgentToolInvocation, _ request: Request, _ context: AgentToolExecutionContext, _ outcome: SFTPRemoteCopyOperation.Outcome, progress: SFTPProgress? = nil diff --git a/Frameworks/TermindRemoteInfrastructure/Transfers/SFTPConcurrencyScheduler.swift b/Frameworks/TermindRemoteInfrastructure/Transfers/SFTPConcurrencyScheduler.swift new file mode 100644 index 00000000..ec348045 --- /dev/null +++ b/Frameworks/TermindRemoteInfrastructure/Transfers/SFTPConcurrencyScheduler.swift @@ -0,0 +1,220 @@ +import Foundation +import TermindDomain + +public nonisolated enum SFTPWorkSource: Hashable, Sendable { + case workspace(UUID) + case agentRun(UUID) + case batch(UUID) +} + +public actor SFTPConcurrencyPermit { + public nonisolated let schedulingMode: SSHChannelSchedulingMode + private var releaseAction: (@Sendable () async -> Void)? + + fileprivate init( + schedulingMode: SSHChannelSchedulingMode, + release: @escaping @Sendable () async -> Void + ) { + self.schedulingMode = schedulingMode + releaseAction = release + } + + public func release() async { + let action = releaseAction + releaseAction = nil + await action?() + } +} + +/// Process-wide admission for SFTP data work. +/// +/// Capacity is counted in logical file operations. A remote copy therefore +/// consumes one global slot and one slot on each distinct endpoint Host even +/// though its implementation may hold two SFTP leases. +public actor SFTPConcurrencyScheduler { + private struct Request { + var id: UUID + var anchorHostID: UUID + var hostIDs: Set + var source: SFTPWorkSource + var byteCount: Int64? + var exclusiveResourceIDs: Set + var continuation: CheckedContinuation + } + + private struct ActiveRequest { + var hostIDs: Set + var exclusiveResourceIDs: Set + } + + private var preferences: SFTPPreferences + private var requests: [UUID: Request] = [:] + private var queues: [UUID: [SFTPWorkSource: [UUID]]] = [:] + private var hostOrder: [UUID] = [] + private var sourceOrder: [UUID: [SFTPWorkSource]] = [:] + private var lastGrantedHost: UUID? + private var lastGrantedSource: [UUID: SFTPWorkSource] = [:] + private var active: [UUID: ActiveRequest] = [:] + private var activePerHost: [UUID: Int] = [:] + private var activeExclusiveResources: Set = [] + + public init(preferences: SFTPPreferences = SFTPPreferences()) { + self.preferences = preferences.normalized + } + + public func updatePreferences(_ value: SFTPPreferences) { + preferences = value.normalized + drain() + } + + public func acquire( + anchorHostID: UUID, + hostIDs: Set, + source: SFTPWorkSource, + byteCount: Int64?, + exclusiveResourceIDs: Set = [] + ) async throws -> SFTPConcurrencyPermit { + try Task.checkCancellation() + let id = UUID() + let normalizedHosts = hostIDs.union([anchorHostID]) + return try await withTaskCancellationHandler { + try await withCheckedThrowingContinuation { continuation in + let request = Request( + id: id, + anchorHostID: anchorHostID, + hostIDs: normalizedHosts, + source: source, + byteCount: byteCount, + exclusiveResourceIDs: exclusiveResourceIDs, + continuation: continuation + ) + requests[id] = request + if queues[anchorHostID] == nil { + queues[anchorHostID] = [:] + hostOrder.append(anchorHostID) + } + if queues[anchorHostID]?[source] == nil { + queues[anchorHostID]?[source] = [] + sourceOrder[anchorHostID, default: []].append(source) + } + queues[anchorHostID]?[source, default: []].append(id) + drain() + } + } onCancel: { + Task { await self.cancel(id) } + } + } + + public var activeCount: Int { active.count } + public var queuedCount: Int { requests.count } + + public func activeCount(for hostID: UUID) -> Int { + activePerHost[hostID, default: 0] + } + + private func cancel(_ id: UUID) { + guard let request = requests.removeValue(forKey: id) else { return } + removeQueuedID(id, request: request) + request.continuation.resume(throwing: CancellationError()) + pruneEmptyQueues() + drain() + } + + private func release(_ id: UUID) { + guard let request = active.removeValue(forKey: id) else { return } + for hostID in request.hostIDs { + let remaining = max(0, activePerHost[hostID, default: 1] - 1) + activePerHost[hostID] = remaining == 0 ? nil : remaining + } + activeExclusiveResources.subtract(request.exclusiveResourceIDs) + drain() + } + + private func drain() { + while active.count < preferences.globalLimit, let request = nextEligibleRequest() { + requests[request.id] = nil + removeQueuedID(request.id, request: request) + let permitID = UUID() + active[permitID] = ActiveRequest( + hostIDs: request.hostIDs, + exclusiveResourceIDs: request.exclusiveResourceIDs) + for hostID in request.hostIDs { activePerHost[hostID, default: 0] += 1 } + activeExclusiveResources.formUnion(request.exclusiveResourceIDs) + lastGrantedHost = request.anchorHostID + lastGrantedSource[request.anchorHostID] = request.source + let mode = schedulingMode(byteCount: request.byteCount) + request.continuation.resume( + returning: SFTPConcurrencyPermit(schedulingMode: mode) { [weak self] in + await self?.release(permitID) + }) + } + pruneEmptyQueues() + } + + private func nextEligibleRequest() -> Request? { + for hostID in rotated(hostOrder, after: lastGrantedHost) { + guard let hostQueues = queues[hostID] else { continue } + let sources = sourceOrder[hostID] ?? [] + for source in rotated(sources, after: lastGrantedSource[hostID]) { + guard let id = hostQueues[source]?.first, let request = requests[id] else { continue } + guard request.hostIDs.allSatisfy({ activePerHost[$0, default: 0] < preferences.perHostLimit }) + else { continue } + guard request.exclusiveResourceIDs.isDisjoint(with: activeExclusiveResources) else { continue } + return request + } + } + return nil + } + + private func schedulingMode(byteCount: Int64?) -> SSHChannelSchedulingMode { + switch preferences.strategy { + case .channelsOnly: + .channelsOnly + case .connectionsPreferred: + .connectionsPreferred + case .smart: + if let byteCount, byteCount > SFTPConcurrencyStrategy.smartConnectionThreshold { + .connectionsPreferred + } else { + .channelsOnly + } + } + } + + private func removeQueuedID(_ id: UUID, request: Request) { + guard var values = queues[request.anchorHostID]?[request.source] else { return } + values.removeAll { $0 == id } + queues[request.anchorHostID]?[request.source] = values + } + + private func pruneEmptyQueues() { + for hostID in hostOrder { + guard var hostQueues = queues[hostID] else { continue } + for source in sourceOrder[hostID] ?? [] where hostQueues[source]?.isEmpty != false { + hostQueues[source] = nil + sourceOrder[hostID]?.removeAll { $0 == source } + } + if hostQueues.isEmpty { + queues[hostID] = nil + sourceOrder[hostID] = nil + hostOrder.removeAll { $0 == hostID } + } else { + queues[hostID] = hostQueues + } + } + let activeHosts = Set(active.values.flatMap(\.hostIDs)) + for hostID in Array(lastGrantedSource.keys) + where queues[hostID] == nil && !activeHosts.contains(hostID) { + lastGrantedSource[hostID] = nil + } + if let lastGrantedHost, queues[lastGrantedHost] == nil, !activeHosts.contains(lastGrantedHost) { + self.lastGrantedHost = nil + } + } + + private func rotated(_ values: [Value], after last: Value?) -> [Value] { + guard !values.isEmpty, let last, let index = values.firstIndex(of: last) else { return values } + let split = values.index(after: index) + return Array(values[split...]) + Array(values[..] = [:] + private var preferencesUpdateTask: Task? private var activeTasks: [SFTPTransfer.ID: Task] = [:] - private var activePerHost: [Host.ID: Int] = [:] + private var activePermits: [SFTPTransfer.ID: SFTPConcurrencyPermit] = [:] private var activeBorrowedConnections: Set = [] private var remoteCopySources: [UUID: (source: SFTPRemoteCopySource, createdAt: Date)] = [:] public private(set) var globalLimit: Int public private(set) var perHostLimit: Int + public private(set) var strategy: SFTPConcurrencyStrategy public init( store: HostLibraryModel, clientFactory: @escaping ClientFactory, - globalLimit: Int? = nil, perHostLimit: Int? = nil + globalLimit: Int? = nil, perHostLimit: Int? = nil, + strategy: SFTPConcurrencyStrategy? = nil, + concurrencyScheduler: SFTPConcurrencyScheduler? = nil ) { self.store = store self.clientFactory = clientFactory @@ -127,15 +133,30 @@ public final class SFTPTransferCoordinator { globalLimit: globalLimit ?? defaults.object(forKey: SettingsKey.sftpGlobalTransferLimit) as? Int ?? SFTPPreferences.defaultGlobalLimit, perHostLimit: perHostLimit ?? defaults.object(forKey: SettingsKey.sftpPerHostTransferLimit) as? Int - ?? SFTPPreferences.defaultPerHostLimit) + ?? SFTPPreferences.defaultPerHostLimit, + strategy: strategy ?? defaults.string(forKey: SettingsKey.sftpConcurrencyStrategy) + .flatMap(SFTPConcurrencyStrategy.init(rawValue:)) ?? .smart) self.globalLimit = limits.globalLimit self.perHostLimit = limits.perHostLimit + self.strategy = limits.strategy + self.concurrencyScheduler = concurrencyScheduler ?? SFTPConcurrencyScheduler(preferences: limits) } public func updateLimits(global: Int, perHost: Int) { - let limits = SFTPPreferences(globalLimit: global, perHostLimit: perHost) + updatePreferences( + SFTPPreferences( + globalLimit: global, perHostLimit: perHost, + strategy: strategy + )) + } + + public func updatePreferences(_ value: SFTPPreferences) { + let limits = value.normalized globalLimit = limits.globalLimit perHostLimit = limits.perHostLimit + strategy = limits.strategy + preferencesUpdateTask?.cancel() + preferencesUpdateTask = Task { await concurrencyScheduler.updatePreferences(limits) } schedule() } @@ -330,6 +351,7 @@ public final class SFTPTransferCoordinator { guard canPause(id), let runtime = runtimes[id] else { return } runtimes[id]?.pauseRequested = true if activeTasks[id] == nil { + admissionTasks.removeValue(forKey: id)?.cancel() store.setSFTPTransferState(.paused, for: id) runtime.connection.managedSession?.suspendTransfer(id) } else { @@ -355,6 +377,7 @@ public final class SFTPTransferCoordinator { runtimes[id]?.pauseRequested = false runtime.cancellation.cancel() if activeTasks[id] == nil { + admissionTasks.removeValue(forKey: id)?.cancel() let task = Task { [self] in var state: SFTPTransfer.State = .cancelled do { try await cleanup(runtime, direction: transfer.direction, client: nil) } catch { @@ -370,6 +393,7 @@ public final class SFTPTransferCoordinator { activeTasks[id] = task } } else if transfer.state == .queued, activeTasks[id] == nil { + admissionTasks.removeValue(forKey: id)?.cancel() let sessions = managedSessions(for: id) guard let runtime = remoteCopyRuntimes.removeValue(forKey: id) else { return } activeTasks[id] = Task { [self] in @@ -418,8 +442,8 @@ public final class SFTPTransferCoordinator { } public func waitForCancellation() async { - while !activeTasks.isEmpty { - let tasks = Array(activeTasks.values) + while !activeTasks.isEmpty || !admissionTasks.isEmpty { + let tasks = Array(activeTasks.values) + Array(admissionTasks.values) for task in tasks { await task.value } } } @@ -458,36 +482,56 @@ public final class SFTPTransferCoordinator { public func clearFinished() { store.clearFinishedSFTPTransfers() } private func schedule() { - while activeTasks.count < globalLimit { - guard - let sftpTransfer = store.sftpTransfers.first(where: { sftpTransfer in - guard sftpTransfer.state == .queued, activeTasks[sftpTransfer.id] == nil, + for sftpTransfer in store.sftpTransfers where sftpTransfer.state == .queued { + guard activeTasks[sftpTransfer.id] == nil, admissionTasks[sftpTransfer.id] == nil, + runtimes[sftpTransfer.id]?.cancelRequested != true, + runtimes[sftpTransfer.id] != nil || remoteCopyRuntimes[sftpTransfer.id] != nil + else { continue } + let hostIDs = Set(hostIDs(for: sftpTransfer)) + guard let anchorHostID = anchorHostID(for: sftpTransfer), !hostIDs.isEmpty else { continue } + let source = workSource(for: sftpTransfer) + let exclusiveResourceIDs = Set(borrowedIDs(for: sftpTransfer)) + let scheduler = concurrencyScheduler + let preferencesUpdateTask = preferencesUpdateTask + admissionTasks[sftpTransfer.id] = Task { @MainActor [weak self] in + do { + await preferencesUpdateTask?.value + let permit = try await scheduler.acquire( + anchorHostID: anchorHostID, + hostIDs: hostIDs, + source: source, + byteCount: sftpTransfer.byteCount >= 0 ? sftpTransfer.byteCount : nil, + exclusiveResourceIDs: exclusiveResourceIDs + ) + guard let self else { + await permit.release() + return + } + admissionTasks[sftpTransfer.id] = nil + guard store.sftpTransfer(sftpTransfer.id)?.state == .queued, runtimes[sftpTransfer.id]?.cancelRequested != true, runtimes[sftpTransfer.id] != nil || remoteCopyRuntimes[sftpTransfer.id] != nil - else { return false } - guard managedSessions(for: sftpTransfer.id).allSatisfy(\.canStartTransfer) else { return false } - let borrowedIDs = borrowedIDs(for: sftpTransfer) - guard borrowedIDs.allSatisfy({ !activeBorrowedConnections.contains($0) }) else { - return false + else { + await permit.release() + return } - return hostIDs(for: sftpTransfer).allSatisfy { - activePerHost[$0, default: 0] < perHostLimit - } - }) - else { return } - start(sftpTransfer) + activePermits[sftpTransfer.id] = permit + start(sftpTransfer, mode: permit.schedulingMode) + } catch { + self?.admissionTasks[sftpTransfer.id] = nil + } + } } } - private func start(_ sftpTransfer: SFTPTransfer) { + private func start(_ sftpTransfer: SFTPTransfer, mode: SSHChannelSchedulingMode) { for session in managedSessions(for: sftpTransfer.id) { session.reserveSlot() } if let runtime = remoteCopyRuntimes[sftpTransfer.id] { - startRemoteCopy(sftpTransfer, runtime: runtime) + startRemoteCopy(sftpTransfer, runtime: runtime, mode: mode) return } guard let hostID = sftpTransfer.hostID, let runtime = runtimes[sftpTransfer.id] else { return } store.setSFTPTransferState(.running, for: sftpTransfer.id) - activePerHost[hostID, default: 0] += 1 let borrowedIDs = Array(runtime.connection.borrowedID.map { [$0] } ?? []) activeBorrowedConnections.formUnion(borrowedIDs) let task = Task { [weak self] in @@ -497,7 +541,8 @@ public final class SFTPTransferCoordinator { var lease: ClientLease? var result: SFTPTransfer.State do { - let acquired = try await clientLease(for: runtime.connection, cancellation: runtime.cancellation) + let acquired = try await clientLease( + for: runtime.connection, mode: mode, cancellation: runtime.cancellation) lease = acquired let client = acquired.client let progress: @Sendable (SFTPProgress) -> Void = { [weak self] value in @@ -572,10 +617,13 @@ public final class SFTPTransferCoordinator { } } - private func startRemoteCopy(_ sftpTransfer: SFTPTransfer, runtime: RemoteCopyRuntime) { + private func startRemoteCopy( + _ sftpTransfer: SFTPTransfer, + runtime: RemoteCopyRuntime, + mode: SSHChannelSchedulingMode + ) { let involvedHostIDs = Array(Set([runtime.sourceHostID, runtime.destinationHostID])) store.setSFTPTransferState(.running, for: sftpTransfer.id) - for hostID in involvedHostIDs { activePerHost[hostID, default: 0] += 1 } let borrowedIDs = [ runtime.sourceConnection.borrowedID, runtime.destinationConnection.borrowedID, @@ -591,18 +639,18 @@ public final class SFTPTransferCoordinator { let destinationID = runtime.destinationConnection.managedSession?.id if let sourceID, sourceID == destinationID { sourceLease = try await clientLease( - for: runtime.sourceConnection, cancellation: runtime.cancellation) + for: runtime.sourceConnection, mode: mode, cancellation: runtime.cancellation) destinationLease = sourceLease } else if let sourceID, let destinationID, sourceID.uuidString > destinationID.uuidString { destinationLease = try await clientLease( - for: runtime.destinationConnection, cancellation: runtime.cancellation) + for: runtime.destinationConnection, mode: mode, cancellation: runtime.cancellation) sourceLease = try await clientLease( - for: runtime.sourceConnection, cancellation: runtime.cancellation) + for: runtime.sourceConnection, mode: mode, cancellation: runtime.cancellation) } else { sourceLease = try await clientLease( - for: runtime.sourceConnection, cancellation: runtime.cancellation) + for: runtime.sourceConnection, mode: mode, cancellation: runtime.cancellation) destinationLease = try await clientLease( - for: runtime.destinationConnection, cancellation: runtime.cancellation) + for: runtime.destinationConnection, mode: mode, cancellation: runtime.cancellation) } guard let acquiredSource = sourceLease, let acquiredDestination = destinationLease else { throw SFTPError.notConnected @@ -650,12 +698,30 @@ public final class SFTPTransferCoordinator { return sftpTransfer.hostID.map { [$0] } ?? [] } + private func anchorHostID(for sftpTransfer: SFTPTransfer) -> Host.ID? { + sftpTransfer.destinationHostID ?? sftpTransfer.sourceHostID ?? sftpTransfer.hostID + } + + private func workSource(for sftpTransfer: SFTPTransfer) -> SFTPWorkSource { + if let runtime = remoteCopyRuntimes[sftpTransfer.id], + let session = runtime.destinationConnection.managedSession + { + return .workspace(session.id) + } + if let session = runtimes[sftpTransfer.id]?.connection.managedSession { + return .workspace(session.id) + } + return .batch(sftpTransfer.batchID) + } + private func clientLease( - for connection: SFTPTransferConnection, cancellation: TransferCancellationToken? = nil + for connection: SFTPTransferConnection, + mode: SSHChannelSchedulingMode = .channelsOnly, + cancellation: TransferCancellationToken? = nil ) async throws -> ClientLease { switch connection { case .session(let session): - return try await session.acquire(cancellation: cancellation) + return try await session.acquire(mode: mode, cancellation: cancellation) case .saved(let configuration): let client = clientFactory() do { @@ -735,20 +801,19 @@ public final class SFTPTransferCoordinator { session.releaseSlot() await session.releaseTransfer(id) } - activeTasks.removeValue(forKey: id) - for hostID in hostIDs { - activePerHost[hostID] = max(0, activePerHost[hostID, default: 1] - 1) - } - releaseBorrowedConnections(borrowedIDs) if state == .paused { store.setSFTPTransferState(.paused, for: id) - schedule() - return + } else { + let completion = + runtimes.removeValue(forKey: id)?.completion + ?? remoteCopyRuntimes.removeValue(forKey: id)?.completion + store.finishSFTPTransfer(id, state: state) + completion?(state) } - let completion = - runtimes.removeValue(forKey: id)?.completion - ?? remoteCopyRuntimes.removeValue(forKey: id)?.completion - store.finishSFTPTransfer(id, state: state) - completion?(state) + activeTasks.removeValue(forKey: id) + let permit = activePermits.removeValue(forKey: id) + releaseBorrowedConnections(borrowedIDs) + await permit?.release() + schedule() } } diff --git a/Frameworks/TermindRemoteInfrastructure/Transfers/SFTPTransferSession.swift b/Frameworks/TermindRemoteInfrastructure/Transfers/SFTPTransferSession.swift index fe0cf202..059c7d21 100644 --- a/Frameworks/TermindRemoteInfrastructure/Transfers/SFTPTransferSession.swift +++ b/Frameworks/TermindRemoteInfrastructure/Transfers/SFTPTransferSession.swift @@ -2,11 +2,10 @@ import Foundation import TermindApplication /// A browser and its queued transfers share ownership of one authenticated connection. -/// Extra connections are always opened without interaction, using this generation's snapshot. +/// The admitted topology decides between sibling Channels and additional connections; +/// additional connections are always opened without interaction from this generation's snapshot. @MainActor public final class SFTPTransferSession { - public nonisolated static let schedulingMode = SSHChannelSchedulingMode.connectionsPreferred - public let id = UUID() public let configuration: SSHConnectionConfiguration private var clientFactory: SFTPTransferCoordinator.ClientFactory @@ -81,10 +80,30 @@ public final class SFTPTransferSession { await old?.disconnect() } - func acquire(cancellation: TransferCancellationToken? = nil) async throws -> SFTPTransferLease { + func acquire( + mode: SSHChannelSchedulingMode, + cancellation: TransferCancellationToken? = nil + ) async throws -> SFTPTransferLease { try checkCancellation(cancellation) while probing { try await waitForPrimary(cancellation: cancellation) } - if capability != .channels, primary != nil, !(prefersPrimary && !primaryBusy) { + if mode == .channelsOnly, let primary, !(prefersPrimary && !primaryBusy) { + do { + let client = try await primary.openSiblingChannel() + try checkCancellation(cancellation) + return SFTPTransferLease(client: client) { await client.disconnect() } + } catch { + try checkCancellation(cancellation) + if error is CancellationError { throw error } + if let failure = error as? SFTPError, case .cancelled = failure { throw failure } + // A server may accept only the browser's original SFTP + // subsystem. Keep the Channel-only transport contract and + // serialize this lease on that primary rather than opening a + // second SSH connection or pausing the transfer. + } + } + if mode == .connectionsPreferred, capability != .channels, primary != nil, + !(prefersPrimary && !primaryBusy) + { let isProbe = capability == .unknown if isProbe { probing = true } defer { @@ -106,12 +125,18 @@ public final class SFTPTransferSession { availabilityChanged?() } } - if capability == .channels, let primary { - try checkCancellation(cancellation) - let client = try await primary.openSiblingChannel() - try checkCancellation(cancellation) - availabilityChanged?() - return SFTPTransferLease(client: client) { await client.disconnect() } + if mode == .connectionsPreferred, capability == .channels, let primary { + do { + try checkCancellation(cancellation) + let client = try await primary.openSiblingChannel() + try checkCancellation(cancellation) + availabilityChanged?() + return SFTPTransferLease(client: client) { await client.disconnect() } + } catch { + try checkCancellation(cancellation) + if error is CancellationError { throw error } + if let failure = error as? SFTPError, case .cancelled = failure { throw failure } + } } while primaryBusy { try await waitForPrimary(cancellation: cancellation) diff --git a/Memory/ARCHITECTURE.md b/Memory/ARCHITECTURE.md index 5bf97afa..9014837e 100644 --- a/Memory/ARCHITECTURE.md +++ b/Memory/ARCHITECTURE.md @@ -164,19 +164,24 @@ AgentKit is a separately versioned remote package; its sibling working tree is n - Domain owns `RemoteFile`, Application `RemoteFileSystem`/`MobileSFTPSessionModel`, Remote Infrastructure libssh2 protocols, Presentation symbols/kinds. Device-local History changes only after fresh successful listings/downloads, never stale/failed/ cancelled/type-changed work. -- Generation-scoped `SFTPTransferSession` outlives panes. First additional connection silently probes the full route: success - enables connection-parallel workers; failure opens sibling SFTP Channels on the primary transport, never the browser's control - Channel. Later failure degrades pending work without replaying active operations. Cancellation caches no failure. Terminal - preflight donates its successful connection. Refresh durable trust, never promote once-only trust. -- App queue defaults 15 active files/five per Host.ID across sessions; ranges 1–100 and 1–min(50, global). The unit is an active - SFTP lease: an independent connection and a sibling Channel consume the same slot. `SFTPSettingsModel` scheduling outlives - Settings. New limits affect future work only. Pane close releases only its ownership; queued work retains the primary - connection, paused work configuration/checkpoints but no connection. Retired generations cannot replace snapshots. +- Generation-scoped `SFTPTransferSession` outlives panes. SFTP data work resolves one topology per admitted file: Channels Only + opens sibling SFTP Channels, Connections Preferred silently probes the full route and falls back to siblings, and Smart uses + Channels through 64 MB/64,000,000 bytes (including unknown sizes) before preferring connections. A failed silent probe degrades later work + without replaying active operations; cancellation caches no failure. Browser control Channels stay independent, terminal + preflight may donate its successful connection, and durable trust refresh never promotes once-only trust. +- One process scheduler covers Workspace/terminal/edit transfers and Agent data operations. It defaults to 30 active files/ten + per Host.ID (ranges 1–100 and 1–min(50, global)), rotating Hosts then Workspace/Agent-run/batch sources while retaining FIFO + within a source. A Channel and an independent connection consume the same logical-file capacity. Cross-Host copy consumes one + global slot and one slot on each endpoint Host. Browser and Agent metadata/control operations do not consume data slots. + `SFTPSettingsModel` scheduling outlives Settings; new limits/topology affect queued work only. Pane close releases only its + ownership; queued work retains the primary connection, paused work configuration/checkpoints but no connection. - Checkpoints, references, acknowledged offsets and digests are memory-only. Pause/loss retains partial destinations; explicit resume verifies source/prefix. No ordinary staging/bookmarks/restart recovery or silent replay. Cancel/Host delete removes partials and reports failed cleanup. Mobile expiration pauses upload/download, cancels copies; desktop quit allows three seconds. Same-connection copies alternate bounded reads/writes, validate size/metadata/EOF and atomically commit temporary files. -- Shared planners retain directory-only batch destinations/mutation counts. Apps own UI, cancellation/errors/refresh and deletion. +- Shared planners retain directory-only batch destinations/mutation counts. In-app SFTP-to-SFTP drag/copy consumes the private + remote payload before asking for its Finder file promise, streams directly between clients without local staging, and exposes + exactly one `.copy` queue row per regular file. Apps own UI, cancellation/errors/refresh and deletion. `SFTPRemoteConflictResolver` validates recursive child paths, subtree mapping/Keep Both: upload blocks strict descendants; copy also blocks conflicting root. Desktop context menus snapshot reconciled selection; appearance never mutates it. - Terminal drop preparation is cancellable/generation-scoped; late results cannot enqueue. Nonmodal notices cannot replace same-batch @@ -197,10 +202,12 @@ AgentKit is a separately versioned remote package; its sibling working tree is n ### Agent/MCP byte channels -- Agent SSH keeps one Host transport and multiplexes concurrent exec Channels. Agent SFTP uses connection-preferred scheduling - with sibling-Channel fallback and a 180-second idle pool across conversations/MCP. `sftp_copy` belongs to Agent SFTP: regular - trees to existing directories, browser conflict/merge policy, preserved sources and two ordered endpoint leases; same-Host - copies prefer two connections and fall back to two Channels. Active exec/SFTP leases block idle sweeping; background commands +- Agent SSH keeps one Host transport and multiplexes concurrent exec Channels. Agent SFTP data work shares the process SFTP + scheduler and resolves Smart/Channel-only/connection-preferred topology by the synced preference; metadata/control calls use + Channels without consuming data capacity. Additional Agent connections retain a 180-second idle pool across conversations/MCP. + `sftp_copy` belongs to Agent SFTP: regular trees to existing directories, browser conflict/merge policy and preserved sources. + Cross-Host copies take two ordered endpoint leases; same-Host copies use two leases for Connections Preferred and one SFTP + Channel for Channel-only/Smart-unknown work. Active exec/SFTP leases block idle sweeping; background commands retain only their Channel, not the whole transport. Started operations never replay. No model bytes/staging/scratch. - macOS MCP exposes the same definition; Bearer/tool switches/Full access remain boundaries, auth uses external interaction. Stop awaits copies. Audit metadata/counts/endpoints/policy, never bodies. Stdio only forwards protocol; mobile has no local MCP listener. @@ -348,7 +355,8 @@ factories/Sendable buffer handlers. Timers/workers enter explicit short MainActo - Frozen V13 Settings; V14 terminal, V15 Metrics journal/account/quarantine, V21 SFTP. Commit data+intent atomically, then mirror defaults; remote apply refreshes without echo. Separate records cover providers/models/MCP/skills/roles/search/terminal/custom text/SFTP limits. First SFTP row imports prior local values once, missing 15/5; future preparation uses persisted truth. `v1/sftp/default` uses `SFTPPreferencesItem` in same zone/ - V1 envelope so old clients ignore it before decode. Invalid limits/type mismatches quarantine. Running Agent snapshots remain fixed. + V1 envelope so old clients ignore it before decode. Missing strategy decodes as Smart; new/missing records default to 30/10 while + existing saved limits remain unchanged. Invalid limits/type mismatches quarantine. Running Agent snapshots remain fixed. - Live role records sync explicit choices, not resolved fallbacks: nil generator = Same as Chat, nil Guardian = Automatic. Read legacy `securityReview`, migrate to canonical `guardian` without reusing system fields. Conversations/approval mode/last-used model/listener config/ unrelated defaults stay local. Listener token is this-device-only Data Protection Keychain. diff --git a/Memory/REFACTOR.md b/Memory/REFACTOR.md index d13dcb8c..99e8ba74 100644 --- a/Memory/REFACTOR.md +++ b/Memory/REFACTOR.md @@ -175,3 +175,22 @@ never justifies extraction. Keep schemas, native lists/terminal teardown and loc reproducible `LaunchFileLogServiceTests.loggerDropsUnapprovedSensitiveValues` failure; one concurrent AgentRuntime cancellation failure passed immediately when its 20-test suite was rerun. Recursive format also retains the unrelated existing `LoggingFilesSettings.swift:110` finding. + +### 2026-09-19 — fair SFTP admission and selectable topology (complete) + +- **Evidence:** the app transfer coordinator admitted the first eligible queued row, so one Workspace could fill every + Host slot, while Agent SFTP leases bypassed the 15/5 limits entirely. User and Agent pools also fixed their topology + at connection-preferred even when a small transfer would not repay another SSH handshake. Internal SFTP drag providers + advertised both a private remote payload and a Finder file promise; reading both representations could start a redundant + download beside the direct remote copy. +- **Action:** add one process scheduler shared by Workspace/terminal/edit and Agent data work, rotating Hosts and then stable + Workspace/Agent-run/batch sources. Resolve Smart (64 MB decimal), Channels Only and Connections Preferred only after admission; + retain feature-owned connection lifetimes and connection-to-Channel fallback. Extend the existing synced SFTP preference + record compatibly and consume the private drag payload without materializing its local-file promise. +- **Result:** new/missing settings default to Smart and 30/10 while saved values remain unchanged. Logical file operations + consume identical capacity whether backed by a Channel or connection; cross-Host copy counts once globally and once per + endpoint. Agent control operations remain outside data capacity, `sftp_copy` uses one composite permit, and in-app remote + copy exposes one `.copy` row per regular file without local staging. This is additive scheduling work, not a reduction claim. +- **Verification:** macOS and generic iOS Simulator builds passed. Focused suites passed for scheduler/session topology, + settings persistence/sync, Agent authentication/MCP, Agent/Workspace copy and native drag payload selection. Chinese + localization and diff checks passed; final full-suite/format status is recorded in the task handoff. diff --git a/TFI/App/MobileAppModel.swift b/TFI/App/MobileAppModel.swift index 0dcc3fff..81e7c720 100644 --- a/TFI/App/MobileAppModel.swift +++ b/TFI/App/MobileAppModel.swift @@ -178,12 +178,16 @@ final class MobileAppModel { service: AppleSpeechTranscriptionService(audioInput: speechAudio) ) speechAudio.onInterruption = { [weak speech] in speech?.cancelAll() } + let sftpConcurrencyScheduler = SFTPConcurrencyScheduler(preferences: sftpSettings.preferences) assistant = MobileAssistantModel( configuration: ai, repository: MemoryIndexedAgentRepository( base: SwiftDataAgentRunRepository(context: ModelContext(container)), memory: memory), memory: memory, - hostServices: AgentHostServices.live(store: library, coordinator: connectionCoordinator), + hostServices: AgentHostServices.live( + store: library, + coordinator: connectionCoordinator, + sftpConcurrencyScheduler: sftpConcurrencyScheduler), metrics: metrics.core, skills: skills, authorize: { [pro] in pro.require(.assistant) }, speech: speech, @@ -197,7 +201,10 @@ final class MobileAppModel { forwardings = dependencies.makeForwardings(library) transfers = SFTPTransferCoordinator( store: library, clientFactory: { connectionCoordinator.makeSFTPClient() }, - globalLimit: sftpSettings.preferences.globalLimit, perHostLimit: sftpSettings.preferences.perHostLimit) + globalLimit: sftpSettings.preferences.globalLimit, + perHostLimit: sftpSettings.preferences.perHostLimit, + strategy: sftpSettings.preferences.strategy, + concurrencyScheduler: sftpConcurrencyScheduler) sftpEdits = SFTPEditService(transfers: transfers, host: { [weak library] in library?.host($0) }) library.willDeleteHost = { [weak sftpEdits, weak forwardings] hostID, _ in sftpEdits?.invalidate(hostID: hostID) @@ -232,7 +239,7 @@ final class MobileAppModel { } vaultSync?.didApplyRemote = { [weak self] in self?.applyRemoteVault($0, keys: $1) } sftpSettings.didChange = { [weak transfers] preferences in - transfers?.updateLimits(global: preferences.globalLimit, perHost: preferences.perHostLimit) + transfers?.updatePreferences(preferences) } settingsSync?.didApplyRemote = { [weak self] snapshot, _ in self?.ai.applyRemoteSnapshot(snapshot.configuration, webSearch: snapshot.webSearch) diff --git a/TFITests/MobileSettingsPortTests.swift b/TFITests/MobileSettingsPortTests.swift index cc1601c9..af632886 100644 --- a/TFITests/MobileSettingsPortTests.swift +++ b/TFITests/MobileSettingsPortTests.swift @@ -13,16 +13,21 @@ import Testing struct MobileSettingsPortTests { @Test func sftpSettingsUseTheSharedPreferencesAndUpdateWithoutOpeningSettings() throws { let app = try MobileAppModel(inMemory: true) - #expect(app.sftpSettings.preferences == SFTPPreferences(globalLimit: 15, perHostLimit: 5)) - #expect(app.transfers.globalLimit == 15) - #expect(app.transfers.perHostLimit == 5) + #expect(app.sftpSettings.preferences == SFTPPreferences(globalLimit: 30, perHostLimit: 10)) + #expect(app.transfers.globalLimit == 30) + #expect(app.transfers.perHostLimit == 10) + #expect(app.transfers.strategy == .smart) #expect(app.sftpSettings.set(100, for: \.globalLimit)) #expect(app.sftpSettings.set(50, for: \.perHostLimit)) + #expect(app.sftpSettings.setStrategy(.connectionsPreferred)) #expect(app.transfers.globalLimit == 100) #expect(app.transfers.perHostLimit == 50) - app.sftpSettings.applyRemoteSnapshot(SFTPPreferences(globalLimit: 3, perHostLimit: 2)) + #expect(app.transfers.strategy == .connectionsPreferred) + app.sftpSettings.applyRemoteSnapshot( + SFTPPreferences(globalLimit: 3, perHostLimit: 2, strategy: .channelsOnly)) #expect(app.transfers.globalLimit == 3) #expect(app.transfers.perHostLimit == 2) + #expect(app.transfers.strategy == .channelsOnly) } @Test func sftpCoordinatorAppliesNormalizedSettingsImmediately() { diff --git a/TFM/App/AppModel.swift b/TFM/App/AppModel.swift index 0a4428c6..042307b4 100644 --- a/TFM/App/AppModel.swift +++ b/TFM/App/AppModel.swift @@ -168,7 +168,11 @@ final class AppModel { serialPorts = dependencies.serialPortService agentRunRepository = MemoryIndexedAgentRepository( base: dependencies.agentRunRepository ?? InMemoryAgentRunRepository(), memory: memory) - let agentHostServices = AgentHostServices.live(store: hosts, coordinator: connectionCoordinator) + let sftpConcurrencyScheduler = SFTPConcurrencyScheduler(preferences: sftpSettings.preferences) + let agentHostServices = AgentHostServices.live( + store: hosts, + coordinator: connectionCoordinator, + sftpConcurrencyScheduler: sftpConcurrencyScheduler) agentHosts = agentHostServices.resolver agentAuthentication = agentHostServices.authentication agentConnections = agentHostServices.connections @@ -182,7 +186,10 @@ final class AppModel { sftpTransfers = SFTPTransferCoordinator( store: hosts, clientFactory: { connectionCoordinator.makeSFTPClient() }, - globalLimit: sftpSettings.preferences.globalLimit, perHostLimit: sftpSettings.preferences.perHostLimit + globalLimit: sftpSettings.preferences.globalLimit, + perHostLimit: sftpSettings.preferences.perHostLimit, + strategy: sftpSettings.preferences.strategy, + concurrencyScheduler: sftpConcurrencyScheduler ) sftpEdits = SFTPEditService(transfers: sftpTransfers, host: { [weak hosts] in hosts?.host($0) }) self.ai = ai @@ -263,7 +270,7 @@ final class AppModel { self?.applyRemoteVaultSnapshot(snapshot, changedKeys: keys) } sftpSettings.didChange = { [weak sftpTransfers] preferences in - sftpTransfers?.updateLimits(global: preferences.globalLimit, perHost: preferences.perHostLimit) + sftpTransfers?.updatePreferences(preferences) } settingsSync?.didApplyRemote = { [weak self] snapshot, _ in self?.ai.applyRemoteSnapshot(snapshot.configuration, webSearch: snapshot.webSearch) diff --git a/TFM/Application/MCPServer/MCPServerTools.swift b/TFM/Application/MCPServer/MCPServerTools.swift index f4828e7a..a23dd78b 100644 --- a/TFM/Application/MCPServer/MCPServerTools.swift +++ b/TFM/Application/MCPServer/MCPServerTools.swift @@ -49,7 +49,9 @@ final class MCPServerTools { let handler = await externalAuthentication.handler(clientName: client, runID: runID) return try await authentication.withSFTP( hostID: request.hostID, invocationID: invocationID, interaction: handler.sshInteraction, - pool: connections, cancellation: token + pool: connections, cancellation: token, + runID: runID, byteCount: request.byteCount, + schedulesDataTransfer: true ) { client in try await SFTPStreamUploadOperation.run( client: client, remotePath: request.remotePath, byteCount: request.byteCount, @@ -63,7 +65,8 @@ final class MCPServerTools { let handler = await externalAuthentication.handler(clientName: clientName, runID: runID) try await authentication.withSFTP( hostID: request.hostID, invocationID: invocationID, interaction: handler.sshInteraction, - pool: connections, cancellation: token + pool: connections, cancellation: token, + runID: runID, schedulesDataTransfer: true ) { client in let stream = try await client.openDownloadStream( at: request.remotePath, maximumBytes: Int64(MCPTransferBudget.maximumBytes), cancellation: token diff --git a/TFM/UI/Components/FileURLDropTarget.swift b/TFM/UI/Components/FileURLDropTarget.swift index 710f001d..d4e3071b 100644 --- a/TFM/UI/Components/FileURLDropTarget.swift +++ b/TFM/UI/Components/FileURLDropTarget.swift @@ -223,13 +223,28 @@ nonisolated struct NativeDropPayload: Sendable { /// `isEnabled` for why refusing the hit test is not enough on its own. func supportedPayload(from pasteboard: NSPasteboard) -> NativeDropPayload? { guard isEnabled else { return nil } - let payload = Self.payload(from: pasteboard).restricted(to: acceptedKinds) + let payload = Self.payload( + from: pasteboard, + preferSFTPItems: acceptedKinds.contains(.sftpItems) + ).restricted(to: acceptedKinds) return payload.preferredKind == nil ? nil : payload } - static func payload(from pasteboard: NSPasteboard) -> NativeDropPayload { + static func payload( + from pasteboard: NSPasteboard, + preferSFTPItems: Bool = true + ) -> NativeDropPayload { let sftpType = NSPasteboard.PasteboardType(UTType.termindSFTPItems.identifier) - var payload = NativeDropPayload(sftpItems: pasteboard.data(forType: sftpType)) + let sftpItems = pasteboard.data(forType: sftpType) + if preferSFTPItems, let sftpItems { + // The same provider also advertises a promised local file for + // Finder. Asking that representation for a URL materializes + // the promise and starts a download. An in-app SFTP target + // must consume only the private payload, otherwise one drag + // creates both a remote copy and a redundant download/upload. + return NativeDropPayload(sftpItems: sftpItems) + } + var payload = NativeDropPayload(sftpItems: sftpItems) for item in pasteboard.pasteboardItems ?? [] { if let value = item.string(forType: .fileURL), diff --git a/TFMTests/FileURLDropTargetTests.swift b/TFMTests/FileURLDropTargetTests.swift index 8c402a5d..daff2bb3 100644 --- a/TFMTests/FileURLDropTargetTests.swift +++ b/TFMTests/FileURLDropTargetTests.swift @@ -98,6 +98,31 @@ import TermindRemoteInfrastructure #expect(target.supportedPayload(from: pasteboard)?.preferredKind == .sftpItems) } + @Test + func internalSFTPPayloadDoesNotRequestItsPromisedLocalFile() throws { + let pasteboard = stagedPasteboard() + defer { pasteboard.releaseGlobally() } + let item = NSPasteboardItem() + let remote = try JSONEncoder().encode(SFTPRemoteItemsPayload(token: UUID())) + item.setData( + remote, + forType: NSPasteboard.PasteboardType(UTType.termindSFTPItems.identifier)) + item.setString( + URL(fileURLWithPath: "/tmp/materialized-by-a-file-promise").absoluteString, + forType: .fileURL) + pasteboard.writeObjects([item]) + + let payload = FileURLDropNSView.payload(from: pasteboard) + + #expect(payload.sftpItems == remote) + #expect(payload.fileURLs.isEmpty) + #expect(payload.preferredKind == .sftpItems) + + let terminalPayload = FileURLDropNSView.payload( + from: pasteboard, preferSFTPItems: false) + #expect(terminalPayload.fileURLs.count == 1) + } + @Test func finderImagePrefersItsFileURLAndFilename() throws { let pasteboard = stagedPasteboard() diff --git a/TFMTests/SFTPConcurrencySchedulerTests.swift b/TFMTests/SFTPConcurrencySchedulerTests.swift new file mode 100644 index 00000000..7a0a8910 --- /dev/null +++ b/TFMTests/SFTPConcurrencySchedulerTests.swift @@ -0,0 +1,183 @@ +import Foundation +import TermindDomain +import TermindRemoteInfrastructure +import Testing + +@Suite +struct SFTPConcurrencySchedulerTests { + @Test func smartResolvesTheSixtyFourMBBoundaryAndUnknownSizeToChannels() async throws { + #expect(SFTPConcurrencyStrategy.smartConnectionThreshold == 64_000_000) + let scheduler = SFTPConcurrencyScheduler( + preferences: SFTPPreferences(globalLimit: 3, perHostLimit: 3, strategy: .smart)) + let host = UUID() + let source = SFTPWorkSource.workspace(UUID()) + + let unknown = try await scheduler.acquire( + anchorHostID: host, hostIDs: [host], source: source, byteCount: nil) + let boundary = try await scheduler.acquire( + anchorHostID: host, hostIDs: [host], source: source, + byteCount: SFTPConcurrencyStrategy.smartConnectionThreshold) + let large = try await scheduler.acquire( + anchorHostID: host, hostIDs: [host], source: source, + byteCount: SFTPConcurrencyStrategy.smartConnectionThreshold + 1) + + #expect(unknown.schedulingMode == .channelsOnly) + #expect(boundary.schedulingMode == .channelsOnly) + #expect(large.schedulingMode == .connectionsPreferred) + await unknown.release() + await boundary.release() + await large.release() + } + + @Test func sameHostRotatesFromWorkspaceToAgentRun() async throws { + let scheduler = SFTPConcurrencyScheduler( + preferences: SFTPPreferences(globalLimit: 1, perHostLimit: 1, strategy: .channelsOnly)) + let host = UUID() + let workspace = SFTPWorkSource.workspace(UUID()) + let agent = SFTPWorkSource.agentRun(UUID()) + let first = try await scheduler.acquire( + anchorHostID: host, hostIDs: [host], source: workspace, byteCount: 1) + let recorder = GrantRecorder() + let workspaceTask = queuedAcquire( + scheduler: scheduler, host: host, source: workspace, label: "workspace", recorder: recorder) + try await eventually { await scheduler.queuedCount == 1 } + let agentTask = queuedAcquire( + scheduler: scheduler, host: host, source: agent, label: "agent", recorder: recorder) + try await eventually { await scheduler.queuedCount == 2 } + + await first.release() + try await eventually { await recorder.values.count == 1 } + let grants = await recorder.values + #expect(grants == ["agent"]) + + let winner = grants.first == "agent" ? agentTask : workspaceTask + let loser = grants.first == "agent" ? workspaceTask : agentTask + let winnerPermit = await winner.value + await winnerPermit?.release() + loser.cancel() + _ = await loser.value + } + + @Test func hostsRotateBeforeSources() async throws { + let scheduler = SFTPConcurrencyScheduler( + preferences: SFTPPreferences(globalLimit: 1, perHostLimit: 1, strategy: .channelsOnly)) + let firstHost = UUID() + let secondHost = UUID() + let source = SFTPWorkSource.workspace(UUID()) + let first = try await scheduler.acquire( + anchorHostID: firstHost, hostIDs: [firstHost], source: source, byteCount: 1) + let recorder = GrantRecorder() + let sameHostTask = queuedAcquire( + scheduler: scheduler, host: firstHost, source: source, label: "first", recorder: recorder) + try await eventually { await scheduler.queuedCount == 1 } + let otherHostTask = queuedAcquire( + scheduler: scheduler, host: secondHost, source: source, label: "second", recorder: recorder) + try await eventually { await scheduler.queuedCount == 2 } + + await first.release() + try await eventually { await recorder.values.count == 1 } + let grants = await recorder.values + #expect(grants == ["second"]) + + let winner = grants.first == "second" ? otherHostTask : sameHostTask + let loser = grants.first == "second" ? sameHostTask : otherHostTask + let winnerPermit = await winner.value + await winnerPermit?.release() + loser.cancel() + _ = await loser.value + } + + @Test func crossHostCopyUsesOneGlobalSlotAndOneSlotOnEachHost() async throws { + let scheduler = SFTPConcurrencyScheduler( + preferences: SFTPPreferences(globalLimit: 2, perHostLimit: 1, strategy: .channelsOnly)) + let sourceHost = UUID() + let destinationHost = UUID() + let unrelatedHost = UUID() + let copy = try await scheduler.acquire( + anchorHostID: destinationHost, + hostIDs: [sourceHost, destinationHost], + source: .workspace(UUID()), + byteCount: 1) + let unrelated = try await scheduler.acquire( + anchorHostID: unrelatedHost, + hostIDs: [unrelatedHost], + source: .workspace(UUID()), + byteCount: 1) + + #expect(await scheduler.activeCount == 2) + #expect(await scheduler.activeCount(for: sourceHost) == 1) + #expect(await scheduler.activeCount(for: destinationHost) == 1) + #expect(await scheduler.activeCount(for: unrelatedHost) == 1) + await copy.release() + await unrelated.release() + } + + @Test func fiveSameHostSlotsSplitThreeTwoAcrossWorkspaces() async throws { + let scheduler = SFTPConcurrencyScheduler( + preferences: SFTPPreferences(globalLimit: 1, perHostLimit: 5, strategy: .channelsOnly)) + let host = UUID() + let blocker = try await scheduler.acquire( + anchorHostID: host, hostIDs: [host], source: .batch(UUID()), byteCount: 1) + let recorder = GrantRecorder() + let first = SFTPWorkSource.workspace(UUID()) + let second = SFTPWorkSource.workspace(UUID()) + var tasks: [Task] = [] + for _ in 0..<4 { + tasks.append( + queuedAcquire( + scheduler: scheduler, host: host, source: first, + label: "first", recorder: recorder)) + tasks.append( + queuedAcquire( + scheduler: scheduler, host: host, source: second, + label: "second", recorder: recorder)) + } + try await eventually { await scheduler.queuedCount == 8 } + await scheduler.updatePreferences( + SFTPPreferences(globalLimit: 5, perHostLimit: 5, strategy: .channelsOnly)) + await blocker.release() + try await eventually { await recorder.values.count == 5 } + + let grants = await recorder.values + #expect(grants.filter { $0 == "first" }.count == 3) + #expect(grants.filter { $0 == "second" }.count == 2) + + for task in tasks { task.cancel() } + for task in tasks { + let permit = await task.value + await permit?.release() + } + } + + private func queuedAcquire( + scheduler: SFTPConcurrencyScheduler, + host: UUID, + source: SFTPWorkSource, + label: String, + recorder: GrantRecorder + ) -> Task { + Task { + do { + let permit = try await scheduler.acquire( + anchorHostID: host, hostIDs: [host], source: source, byteCount: 1) + await recorder.append(label) + return permit + } catch { + return nil + } + } + } + + private func eventually(_ condition: () async -> Bool) async throws { + for _ in 0..<500 { + if await condition() { return } + try await Task.sleep(for: .milliseconds(10)) + } + try #require(await condition()) + } +} + +private actor GrantRecorder { + private(set) var values: [String] = [] + func append(_ value: String) { values.append(value) } +} diff --git a/TFMTests/SFTPCopyToolTests.swift b/TFMTests/SFTPCopyToolTests.swift index 914de3b3..93b8b524 100644 --- a/TFMTests/SFTPCopyToolTests.swift +++ b/TFMTests/SFTPCopyToolTests.swift @@ -124,7 +124,7 @@ struct SFTPCopyToolTests { @Test func sameHostUsesIndependentConnectionsAndRejectsSelfCopies() async throws { - let fixture = CopyFixture(sameHost: true) + let fixture = CopyFixture(sameHost: true, strategy: .connectionsPreferred) await fixture.source.directory("/target") await fixture.source.put("/source/a.bin", bytes: Data([1])) #expect(try await !fixture.copy().isError) @@ -139,6 +139,17 @@ struct SFTPCopyToolTests { #expect(await fixture.source.nodes["/source/tree/nested/tree"] == nil) } + @Test + func smartUnknownSameHostCopyUsesOneChannel() async throws { + let fixture = CopyFixture(sameHost: true) + await fixture.source.directory("/target") + await fixture.source.put("/source/a.bin", bytes: Data([1])) + + #expect(try await !fixture.copy().isError) + #expect(await fixture.backend.opens == 1) + #expect(await fixture.backend.siblingChannels == 0) + } + @Test func detectsChangedSourceAndReportsPartialCompletionWithoutRetry() async throws { let fixture = CopyFixture() @@ -333,7 +344,10 @@ private final class CopyFixture { let authentication: AgentSSHAuthenticationRegistry var tool: SFTPCopyTool { .init(hosts: resolver, authentication: authentication, connections: pool) } - init(sameHost: Bool = false) { + init( + sameHost: Bool = false, + strategy: SFTPConcurrencyStrategy = .smart + ) { destinationID = sameHost ? sourceID : UUID() let sourceID = sourceID let destinationID = destinationID @@ -362,7 +376,10 @@ private final class CopyFixture { pool = AgentConnectionPool( configuration: { configuration($0) }, routeFingerprint: { _ in "test" }, openExec: { _ in throw AgentToolError.invalidHost }, - openSFTP: { try await backend.open($0.configuration.hostID) }, sweepInterval: nil + openSFTP: { try await backend.open($0.configuration.hostID) }, + sftpConcurrencyScheduler: SFTPConcurrencyScheduler( + preferences: SFTPPreferences(strategy: strategy)), + sweepInterval: nil ) } @@ -457,12 +474,14 @@ private actor CopyBackend { var filesystems: [UUID: CopyFileSystem] var opens = 0 var disconnects = 0 + var siblingChannels = 0 init(filesystems: [UUID: CopyFileSystem]) { self.filesystems = filesystems } func open(_ id: UUID) throws -> CopyClient { opens += 1 return CopyClient(filesystem: try #require(filesystems[id]), backend: self) } func disconnected() { disconnects += 1 } + func openedSibling() { siblingChannels += 1 } } private actor CopyClient: SFTPClientProtocol { @@ -473,6 +492,10 @@ private actor CopyClient: SFTPClientProtocol { self.backend = backend } func connect(using configuration: SSHConnectionConfiguration, progress: SSHRouteProgressHandler?) async throws {} + func openSiblingChannel() async throws -> any SFTPClientProtocol { + await backend.openedSibling() + return CopyClient(filesystem: filesystem, backend: backend) + } func disconnect() async { await backend.disconnected() } func homeDirectory() async throws -> String { "/" } func listDirectory(at path: String) async throws -> [RemoteFile] { await filesystem.children(path) } diff --git a/TFMTests/SFTPSettingsIntegrationTests.swift b/TFMTests/SFTPSettingsIntegrationTests.swift index 5595827a..1f0f3a14 100644 --- a/TFMTests/SFTPSettingsIntegrationTests.swift +++ b/TFMTests/SFTPSettingsIntegrationTests.swift @@ -12,15 +12,19 @@ struct SFTPSettingsIntegrationTests { let settings = SFTPSettingsModel() let app = AppModel(hosts: HostLibraryModel(), sftpSettings: settings) defer { app.shutdownForTermination() } - #expect(app.sftpTransfers.globalLimit == 15) - #expect(app.sftpTransfers.perHostLimit == 5) + #expect(app.sftpTransfers.globalLimit == 30) + #expect(app.sftpTransfers.perHostLimit == 10) + #expect(app.sftpTransfers.strategy == .smart) #expect(settings.set(100, for: \.globalLimit)) #expect(settings.set(50, for: \.perHostLimit)) + #expect(settings.setStrategy(.connectionsPreferred)) #expect(app.sftpTransfers.globalLimit == 100) #expect(app.sftpTransfers.perHostLimit == 50) - settings.applyRemoteSnapshot(SFTPPreferences(globalLimit: 3, perHostLimit: 2)) + #expect(app.sftpTransfers.strategy == .connectionsPreferred) + settings.applyRemoteSnapshot(SFTPPreferences(globalLimit: 3, perHostLimit: 2, strategy: .channelsOnly)) #expect(app.sftpTransfers.globalLimit == 3) #expect(app.sftpTransfers.perHostLimit == 2) + #expect(app.sftpTransfers.strategy == .channelsOnly) } @Test(.requiresVisibleWindow, arguments: ["en", "zh-Hans", "zh-Hant"]) diff --git a/TFMTests/SFTPTransferSessionTests.swift b/TFMTests/SFTPTransferSessionTests.swift index e1de4c0e..84b8208e 100644 --- a/TFMTests/SFTPTransferSessionTests.swift +++ b/TFMTests/SFTPTransferSessionTests.swift @@ -14,7 +14,8 @@ struct SFTPTransferSessionTests { let configuration = SSHConnectionConfiguration( hostID: UUID(), host: "fixture", port: 22, username: "user", credentials: [], expectedHostKeys: []) let coordinator = SFTPTransferCoordinator( - store: HostLibraryModel(), clientFactory: { TransferSessionClient(harness: harness) }) + store: HostLibraryModel(), clientFactory: { TransferSessionClient(harness: harness) }, + strategy: .connectionsPreferred) let browser = SFTPSessionModel( hostID: configuration.hostID, initialPath: "/", client: primary, configuration: configuration) let workspace = Workspace(pane: Pane(.sftpSession(browser))) @@ -88,6 +89,21 @@ struct SFTPTransferSessionTests { #expect(await harness.opens == 1) } + @Test func channelsOnlyUsesSiblingChannelsWithoutOpeningConnections() async throws { + let harness = TransferSessionHarness(behaviors: []) + let primary = TransferSessionClient(harness: harness, primary: true) + let (session, coordinator) = fixture( + primary: primary, harness: harness, limit: 3, strategy: .channelsOnly) + enqueue(3, session: session, coordinator: coordinator) + try await eventually { await harness.started == 3 } + #expect(await harness.opens == 0) + #expect(await harness.siblingChannels == 3) + #expect(await harness.maximumActive == 3) + await harness.releaseAll() + try await eventually { coordinator.activeCount == 0 } + await session.releaseOwner() + } + @Test func emptyCredentialsCanOpenParallelConnectionsWithinTheLimits() async throws { let harness = TransferSessionHarness(behaviors: []) let primary = TransferSessionClient(harness: harness, primary: true) @@ -180,7 +196,7 @@ struct SFTPTransferSessionTests { hostID: UUID(), host: "terminal", port: 22, username: "user", credentials: [], expectedHostKeys: []) let coordinator = SFTPTransferCoordinator( store: HostLibraryModel(), clientFactory: { TransferSessionClient(harness: harness) }, - globalLimit: 15, perHostLimit: 5) + globalLimit: 15, perHostLimit: 5, strategy: .connectionsPreferred) let upload = SFTPSessionModel( hostID: configuration.hostID, initialPath: "/destination", client: primary, configuration: configuration, wasOpenedSilently: true) @@ -215,7 +231,7 @@ struct SFTPTransferSessionTests { let harness = TransferSessionHarness(behaviors: []) let coordinator = SFTPTransferCoordinator( store: HostLibraryModel(), clientFactory: { TransferSessionClient(harness: harness) }, - globalLimit: 15, perHostLimit: 5) + globalLimit: 15, perHostLimit: 5, strategy: .connectionsPreferred) let configuration = SSHConnectionConfiguration( hostID: UUID(), host: "same-host", port: 22, username: "user", credentials: [], expectedHostKeys: []) let sessions = (0..<2).map { _ in @@ -237,7 +253,8 @@ struct SFTPTransferSessionTests { private func fixture( primary: TransferSessionClient, harness: TransferSessionHarness, limit: Int = 4, - wasOpenedSilently: Bool = false + wasOpenedSilently: Bool = false, + strategy: SFTPConcurrencyStrategy = .connectionsPreferred ) -> (SFTPTransferSession, SFTPTransferCoordinator) { let session = SFTPTransferSession( configuration: SSHConnectionConfiguration( @@ -246,7 +263,7 @@ struct SFTPTransferSessionTests { client: primary, wasOpenedSilently: wasOpenedSilently) let coordinator = SFTPTransferCoordinator( store: HostLibraryModel(), clientFactory: { TransferSessionClient(harness: harness) }, - globalLimit: limit, perHostLimit: limit) + globalLimit: limit, perHostLimit: limit, strategy: strategy) return (session, coordinator) }