diff --git a/Sources/GlossCore/BabelDOCBatchCoordinator.swift b/Sources/GlossCore/BabelDOCBatchCoordinator.swift index 11c4ac7..66556b2 100644 --- a/Sources/GlossCore/BabelDOCBatchCoordinator.swift +++ b/Sources/GlossCore/BabelDOCBatchCoordinator.swift @@ -2,20 +2,17 @@ import Foundation actor BabelDOCBatchCoordinator { struct Configuration: Sendable { - let maximumBatchItems: Int let maximumBatchCharacters: Int let maximumConcurrentBatches: Int let fillDelayNanoseconds: UInt64 let refillDelayNanoseconds: UInt64 init( - maximumBatchItems: Int = 12, - maximumBatchCharacters: Int = 1_800, + maximumBatchCharacters: Int = 3_000, maximumConcurrentBatches: Int = 2, fillDelayNanoseconds: UInt64 = 25_000_000, refillDelayNanoseconds: UInt64 = 0 ) { - self.maximumBatchItems = max(1, maximumBatchItems) self.maximumBatchCharacters = max(1, maximumBatchCharacters) self.maximumConcurrentBatches = max(1, maximumConcurrentBatches) self.fillDelayNanoseconds = fillDelayNanoseconds @@ -39,15 +36,10 @@ actor BabelDOCBatchCoordinator { } self.init( - maximumBatchItems: integer( - "GLOSS_BABELDOC_BATCH_ITEMS", - default: defaults.maximumBatchItems, - range: 1...24 - ), maximumBatchCharacters: integer( "GLOSS_BABELDOC_BATCH_CHARACTERS", default: defaults.maximumBatchCharacters, - range: 200...4_000 + range: 200...12_000 ), maximumConcurrentBatches: integer( "GLOSS_BABELDOC_MODEL_CONCURRENCY", @@ -101,6 +93,9 @@ actor BabelDOCBatchCoordinator { private var activeBatchCount = 0 private var fillTask: Task? private var dispatchStateRevision = 0 + // Accounts for compact input/output JSON framing so tiny items cannot + // create an effectively unbounded model turn. + static let estimatedItemFramingCharacters = 32 init( broker: TranslationBroker, @@ -240,30 +235,32 @@ actor BabelDOCBatchCoordinator { let batch = takeNextBatch() guard !batch.isEmpty else { break } activeBatchCount += 1 - let characterCount = batch.reduce(0) { $0 + $1.brokerItem.text.count } + let sourceCharacterCount = batch.reduce(0) { + $0 + $1.brokerItem.text.count + } + let estimatedCharacterCount = batch.reduce(0) { + $0 + Self.estimatedCharacterCost(of: $1) + } let characterUtilization = min( 100, - characterCount * 100 / configuration.maximumBatchCharacters + estimatedCharacterCount * 100 / configuration.maximumBatchCharacters ) runtimeLog.write( "bridge", - "babeldoc_batch_dispatched items=\(batch.count) chars=\(characterCount) utilization_pct=\(characterUtilization) requests=\(Set(batch.map(\.requestID)).count) active_batches=\(activeBatchCount) pending_items=\(pendingItems.count)" + "babeldoc_batch_dispatched items=\(batch.count) chars=\(sourceCharacterCount) estimated_chars=\(estimatedCharacterCount) utilization_pct=\(characterUtilization) requests=\(Set(batch.map(\.requestID)).count) active_batches=\(activeBatchCount) pending_items=\(pendingItems.count)" ) let broker = self.broker let key = batch[0].key + let runtimeLog = self.runtimeLog Task { let result: Result<[TranslationOutput], Error> do { result = .success( - try await broker.translate( - TranslationBatchRequest( - items: batch.map(\.brokerItem), - targetLanguage: key.targetLanguage, - profile: .academic, - contentKind: .document, - context: key.context, - priority: .background - ) + try await Self.translateWithSplitRecovery( + batch: batch, + key: key, + broker: broker, + runtimeLog: runtimeLog ) ) } catch { @@ -274,10 +271,70 @@ actor BabelDOCBatchCoordinator { } } + private nonisolated static func translateWithSplitRecovery( + batch: [PendingItem], + key: BatchKey, + broker: TranslationBroker, + runtimeLog: GlossRuntimeLog, + depth: Int = 0 + ) async throws -> [TranslationOutput] { + do { + return try await broker.translate( + TranslationBatchRequest( + items: batch.map(\.brokerItem), + targetLanguage: key.targetLanguage, + profile: .academic, + contentKind: .document, + context: key.context, + priority: .background + ) + ) + } catch let error as TranslationError { + guard case .invalidResponse = error, batch.count > 1 else { throw error } + + let splitIndex = balancedSplitIndex(for: batch) + let left = Array(batch[.. Int { + precondition(batch.count > 1) + let targetCost = batch.reduce(0) { $0 + estimatedCharacterCost(of: $1) } / 2 + var accumulatedCost = 0 + for index in 1..= targetCost { return index } + } + return batch.count / 2 + } + private func takeNextBatch() -> [PendingItem] { guard let first = pendingItems.first else { return [] } var selectedIndices = [0] - var characterCount = first.brokerItem.text.count + var characterCount = Self.estimatedCharacterCost(of: first) let candidates = pendingItems.indices.dropFirst() .filter { pendingItems[$0].key == first.key } @@ -289,8 +346,7 @@ actor BabelDOCBatchCoordinator { } for index in candidates { - guard selectedIndices.count < configuration.maximumBatchItems else { break } - let itemCharacters = pendingItems[index].brokerItem.text.count + let itemCharacters = Self.estimatedCharacterCost(of: pendingItems[index]) guard characterCount + itemCharacters <= configuration.maximumBatchCharacters else { continue } @@ -309,20 +365,22 @@ actor BabelDOCBatchCoordinator { private func firstPendingBatchIsFull() -> Bool { guard let first = pendingItems.first else { return false } - var itemCount = 0 var characterCount = 0 for pending in pendingItems where pending.key == first.key { - if itemCount >= configuration.maximumBatchItems - || characterCount + pending.brokerItem.text.count - > configuration.maximumBatchCharacters + if characterCount + Self.estimatedCharacterCost(of: pending) + > configuration.maximumBatchCharacters { return true } - itemCount += 1 - characterCount += pending.brokerItem.text.count + characterCount += Self.estimatedCharacterCost(of: pending) } - return itemCount >= configuration.maximumBatchItems - || characterCount >= configuration.maximumBatchCharacters + return characterCount >= configuration.maximumBatchCharacters + } + + private nonisolated static func estimatedCharacterCost( + of pending: PendingItem + ) -> Int { + pending.brokerItem.text.count + estimatedItemFramingCharacters } private func finish( diff --git a/Sources/GlossCore/CodexAppServerClient.swift b/Sources/GlossCore/CodexAppServerClient.swift index 6bda4b2..a6e973d 100644 --- a/Sources/GlossCore/CodexAppServerClient.swift +++ b/Sources/GlossCore/CodexAppServerClient.swift @@ -61,6 +61,14 @@ public actor CodexAppServerClient: TranslationBackend { private static let defaultModelWaitHedgeNanoseconds: UInt64 = 8_000_000_000 private static let dispatchAwareModelWaitHedgeNanoseconds: UInt64 = 3_000_000_000 private static let defaultThreadRotationTurns = 10 + private static let defaultSparkStartIntervalNanoseconds: UInt64 = 500_000_000 + private static let defaultSparkCapacityRetryLimit = 4 + private static let sparkCapacityBackoffNanoseconds: [UInt64] = [ + 2_000_000_000, + 5_000_000_000, + 10_000_000_000, + 20_000_000_000, + ] static let disabledCodexFeatures = [ "shell_tool", "unified_exec", @@ -233,6 +241,8 @@ public actor CodexAppServerClient: TranslationBackend { private let threadRotationTurns: Int? private let maximumConcurrentTurns: Int private let maximumBackgroundConcurrentTurns: Int + private let sparkStartIntervalNanoseconds: UInt64 + private let sparkCapacityRetryLimit: Int private let glossaryStore: GlossaryStore private let dispatchState: TranslationDispatchState? private var process: Process? @@ -259,6 +269,8 @@ public actor CodexAppServerClient: TranslationBackend { private var initialized = false private var cachedAccountStatus: CodexAccountStatus? private var lastError: String? + private var nextSparkBackgroundStartAt: UInt64 = 0 + private var sparkCapacityCooldownUntil: UInt64 = 0 public init( environment: [String: String] = ProcessInfo.processInfo.environment, @@ -295,6 +307,8 @@ public actor CodexAppServerClient: TranslationBackend { environment, maximumConcurrentTurns: maximumConcurrentTurns ) + self.sparkStartIntervalNanoseconds = Self.readSparkStartIntervalNanoseconds(environment) + self.sparkCapacityRetryLimit = Self.readSparkCapacityRetryLimit(environment) self.glossaryStore = glossaryStore self.dispatchState = dispatchState } @@ -346,30 +360,16 @@ public actor CodexAppServerClient: TranslationBackend { && request.contentKind == .document && maximumConcurrentTurns > maximumBackgroundConcurrentTurns && modelWaitHedgeNanoseconds != nil - let primaryState = TurnAttemptState() let result: TurnAttemptResult do { - if hedgeEligible, let modelWaitHedgeNanoseconds { - result = try await runHedgedTurn( - request: request, - prompt: prompt, - primaryState: primaryState, - hedgeThresholdNanoseconds: modelWaitHedgeNanoseconds - ) - for output in result.outputs { - onOutput?(output) - } - } else { - result = try await runTurnAttempt( - request: request, - prompt: prompt, - kind: .primary, - attemptState: primaryState, - onOutput: onOutput, - acquisition: .scheduled - ) - } + result = try await runTranslationTurn( + request: request, + prompt: prompt, + onOutput: onOutput, + hedgeEligible: hedgeEligible, + hedgeThresholdNanoseconds: modelWaitHedgeNanoseconds + ) } catch { runtimeLog.write( "codex", @@ -398,6 +398,111 @@ public actor CodexAppServerClient: TranslationBackend { return result.outputs } + private func runTranslationTurn( + request: TranslationBatchRequest, + prompt: String, + onOutput: (@Sendable (TranslationOutput) -> Void)?, + hedgeEligible: Bool, + hedgeThresholdNanoseconds: UInt64? + ) async throws -> TurnAttemptResult { + let usesSparkPolicy = Self.shouldUseSparkDocumentPolicy( + model: model, + request: request + ) + let maximumAttempts = usesSparkPolicy ? sparkCapacityRetryLimit + 1 : 1 + + for attempt in 0.. now else { + nextSparkBackgroundStartAt = now + sparkStartIntervalNanoseconds + return + } + try await Task.sleep(nanoseconds: admissionTime - now) + } + } + + @discardableResult + private func scheduleSparkCapacityCooldown(retryIndex: Int) -> UInt64 { + let maximumBackoff = Self.sparkCapacityBackoffNanoseconds[ + min(retryIndex, Self.sparkCapacityBackoffNanoseconds.count - 1) + ] + let backoff = UInt64.random(in: 0...maximumBackoff) + sparkCapacityCooldownUntil = max( + sparkCapacityCooldownUntil, + DispatchTime.now().uptimeNanoseconds + backoff + ) + return backoff + } + + static func shouldUseSparkDocumentPolicy( + model: String?, + request: TranslationBatchRequest + ) -> Bool { + model?.localizedCaseInsensitiveContains("spark") == true + && request.priority == .background + && request.contentKind == .document + } + + static func isCapacityError(_ error: Error) -> Bool { + let message = error.localizedDescription.lowercased() + return message.contains("selected model is at capacity") + || message.contains("rate limit") + || message.contains("too many requests") + || message.contains("http 429") + || message.contains("status 429") + } + private enum ThreadAcquisition { case scheduled case opportunisticHedge @@ -1688,7 +1793,7 @@ public actor CodexAppServerClient: TranslationBackend { } static func compactModelItemID(for index: Int) -> String { - String(index, radix: 36) + String(index) } private static func modelJSONData(from output: String) throws -> Data { @@ -2123,6 +2228,26 @@ public actor CodexAppServerClient: TranslationBackend { return turns } + static func readSparkStartIntervalNanoseconds( + _ environment: [String: String] + ) -> UInt64 { + guard let rawValue = environment["GLOSS_SPARK_START_INTERVAL_MS"], + let milliseconds = UInt64(rawValue), + milliseconds <= 5_000 + else { return defaultSparkStartIntervalNanoseconds } + return milliseconds * 1_000_000 + } + + static func readSparkCapacityRetryLimit( + _ environment: [String: String] + ) -> Int { + guard let rawValue = environment["GLOSS_SPARK_CAPACITY_RETRIES"], + let retries = Int(rawValue), + (0...6).contains(retries) + else { return defaultSparkCapacityRetryLimit } + return retries + } + private static func readMaximumBackgroundConcurrentTurns( _ environment: [String: String], maximumConcurrentTurns: Int diff --git a/Sources/GlossCore/LoopbackServer.swift b/Sources/GlossCore/LoopbackServer.swift index ef2d1a0..4ead704 100644 --- a/Sources/GlossCore/LoopbackServer.swift +++ b/Sources/GlossCore/LoopbackServer.swift @@ -193,8 +193,7 @@ package final class LoopbackServer: @unchecked Sendable { private static let maximumBabelDOCChunkCharacters = 900 private static let defaultBabelDOCBatchConfiguration = BabelDOCBatchCoordinator.Configuration( - maximumBatchItems: 12, - maximumBatchCharacters: 1_800, + maximumBatchCharacters: 3_000, maximumConcurrentBatches: 2, fillDelayNanoseconds: 25_000_000, refillDelayNanoseconds: 0 @@ -251,7 +250,7 @@ package final class LoopbackServer: @unchecked Sendable { self.runtimeLog = runtimeLog runtimeLog.write( "bridge", - "babeldoc_batch_configuration max_items=\(babelDOCConfiguration.maximumBatchItems) max_chars=\(babelDOCConfiguration.maximumBatchCharacters) concurrency=\(babelDOCConfiguration.maximumConcurrentBatches) fill_delay_ms=\(babelDOCConfiguration.fillDelayNanoseconds / 1_000_000) refill_delay_ms=\(babelDOCConfiguration.refillDelayNanoseconds / 1_000_000)" + "babeldoc_batch_configuration max_estimated_chars=\(babelDOCConfiguration.maximumBatchCharacters) concurrency=\(babelDOCConfiguration.maximumConcurrentBatches) fill_delay_ms=\(babelDOCConfiguration.fillDelayNanoseconds / 1_000_000) refill_delay_ms=\(babelDOCConfiguration.refillDelayNanoseconds / 1_000_000)" ) } @@ -773,7 +772,7 @@ package final class LoopbackServer: @unchecked Sendable { ) runtimeLog.write( "bridge", - "babeldoc_translation_plan source_items=\(prompt.items.count) chunks=\(chunks.count) max_chunk_chars=\(Self.maximumBabelDOCChunkCharacters) max_batch_items=\(babelDOCBatchConfiguration.maximumBatchItems) max_batch_chars=\(babelDOCBatchConfiguration.maximumBatchCharacters)" + "babeldoc_translation_plan source_items=\(prompt.items.count) chunks=\(chunks.count) max_chunk_chars=\(Self.maximumBabelDOCChunkCharacters) max_batch_estimated_chars=\(babelDOCBatchConfiguration.maximumBatchCharacters)" ) let targetLanguage = prompt.targetLanguage diff --git a/Tests/GlossCoreTests/CodexAppServerClientTests.swift b/Tests/GlossCoreTests/CodexAppServerClientTests.swift index 131c472..6d2919b 100644 --- a/Tests/GlossCoreTests/CodexAppServerClientTests.swift +++ b/Tests/GlossCoreTests/CodexAppServerClientTests.swift @@ -116,9 +116,75 @@ final class CodexAppServerClientTests: XCTestCase { func testModelItemIDsStayCompactWithinLargeBatches() { XCTAssertEqual(CodexAppServerClient.compactModelItemID(for: 0), "0") - XCTAssertEqual(CodexAppServerClient.compactModelItemID(for: 10), "a") - XCTAssertEqual(CodexAppServerClient.compactModelItemID(for: 35), "z") - XCTAssertEqual(CodexAppServerClient.compactModelItemID(for: 36), "10") + XCTAssertEqual(CodexAppServerClient.compactModelItemID(for: 10), "10") + XCTAssertEqual(CodexAppServerClient.compactModelItemID(for: 35), "35") + XCTAssertEqual(CodexAppServerClient.compactModelItemID(for: 36), "36") + } + + func testSparkDocumentPolicyDoesNotApplyToLuna() { + let document = TranslationBatchRequest( + items: [TranslationItem(id: "pdf", text: "Paper")], + targetLanguage: "Chinese (Simplified)", + contentKind: .document, + priority: .background + ) + + XCTAssertTrue( + CodexAppServerClient.shouldUseSparkDocumentPolicy( + model: "gpt-5.3-codex-spark", + request: document + ) + ) + XCTAssertFalse( + CodexAppServerClient.shouldUseSparkDocumentPolicy( + model: "gpt-5.6-luna", + request: document + ) + ) + } + + func testSparkPacingAndCapacityRetryOverridesAreBounded() { + XCTAssertEqual( + CodexAppServerClient.readSparkStartIntervalNanoseconds([:]), + 500_000_000 + ) + XCTAssertEqual( + CodexAppServerClient.readSparkStartIntervalNanoseconds([ + "GLOSS_SPARK_START_INTERVAL_MS": "750" + ]), + 750_000_000 + ) + XCTAssertEqual(CodexAppServerClient.readSparkCapacityRetryLimit([:]), 4) + XCTAssertEqual( + CodexAppServerClient.readSparkCapacityRetryLimit([ + "GLOSS_SPARK_CAPACITY_RETRIES": "2" + ]), + 2 + ) + XCTAssertEqual( + CodexAppServerClient.readSparkCapacityRetryLimit([ + "GLOSS_SPARK_CAPACITY_RETRIES": "20" + ]), + 4 + ) + } + + func testSparkCapacityErrorsAreRecognizedWithoutMatchingUnrelatedFailures() { + XCTAssertTrue( + CodexAppServerClient.isCapacityError( + TranslationError.backendUnavailable("Selected model is at capacity") + ) + ) + XCTAssertTrue( + CodexAppServerClient.isCapacityError( + TranslationError.backendUnavailable("HTTP 429 Too Many Requests") + ) + ) + XCTAssertFalse( + CodexAppServerClient.isCapacityError( + TranslationError.invalidResponse("Missing translation item") + ) + ) } func testThreadRotationDefaultsToTenSuccessfulTurnsAndCanBeDisabled() { diff --git a/Tests/GlossCoreTests/LoopbackServerTests.swift b/Tests/GlossCoreTests/LoopbackServerTests.swift index abdd444..7c6ad22 100644 --- a/Tests/GlossCoreTests/LoopbackServerTests.swift +++ b/Tests/GlossCoreTests/LoopbackServerTests.swift @@ -306,10 +306,14 @@ final class LoopbackServerTests: XCTestCase { ) XCTAssertEqual(longCompletion.response.statusCode, 200) let chunkRequests = await backend.recordedRequests() - XCTAssertGreaterThan(chunkRequests.count, 1) + XCTAssertEqual(chunkRequests.count, 1) + XCTAssertGreaterThan(chunkRequests.reduce(0) { $0 + $1.items.count }, 1) XCTAssertTrue( chunkRequests.allSatisfy { - $0.items.reduce(0) { $0 + $1.text.count } <= 1_800 + $0.items.reduce(0) { + $0 + $1.text.count + + BabelDOCBatchCoordinator.estimatedItemFramingCharacters + } <= 3_000 } ) XCTAssertTrue( @@ -511,19 +515,18 @@ final class LoopbackServerTests: XCTestCase { XCTAssertEqual(Set(requests[0].items.map(\.id)).count, 2) } - func testBabelDOCBatchCoordinatorKeepsModelRequestsBounded() async throws { + func testBabelDOCBatchCoordinatorUsesEstimatedCharacterBudgetWithoutFixedItemLimit() async throws { let backend = BridgeBackend() let coordinator = BabelDOCBatchCoordinator( broker: TranslationBroker(backend: backend), configuration: .init( - maximumBatchItems: 2, - maximumBatchCharacters: 100, + maximumBatchCharacters: 800, maximumConcurrentBatches: 2, fillDelayNanoseconds: 0 ) ) - let items = (0..<5).map { - TranslationItem(id: "item-\($0)", text: "Paragraph \($0)") + let items = (0..<25).map { + TranslationItem(id: "item-\($0)", text: "\($0)") } let outputs = try await coordinator.translate( @@ -534,8 +537,41 @@ final class LoopbackServerTests: XCTestCase { XCTAssertEqual(outputs.map(\.id), items.map(\.id)) let requests = await backend.recordedRequests() - XCTAssertEqual(requests.count, 3) - XCTAssertTrue(requests.allSatisfy { $0.items.count <= 2 }) + XCTAssertEqual(requests.count, 2) + XCTAssertGreaterThan(requests.map(\.items.count).max() ?? 0, 12) + XCTAssertTrue( + requests.allSatisfy { + $0.items.reduce(0) { + $0 + $1.text.count + + BabelDOCBatchCoordinator.estimatedItemFramingCharacters + } <= 800 + } + ) + } + + func testBabelDOCBatchCoordinatorSplitsInvalidModelOutputUntilItRecovers() async throws { + let backend = SplitRecoveryBackend(maximumReliableItems: 2) + let coordinator = BabelDOCBatchCoordinator( + broker: TranslationBroker(backend: backend), + configuration: .init( + maximumBatchCharacters: 2_000, + maximumConcurrentBatches: 1, + fillDelayNanoseconds: 0 + ) + ) + let items = (0..<8).map { + TranslationItem(id: "item-\($0)", text: "Paragraph \($0)") + } + + let outputs = try await coordinator.translate( + items: items, + targetLanguage: "Chinese (Simplified)", + context: "PDF" + ) + + XCTAssertEqual(outputs.map(\.id), items.map(\.id)) + let requestSizes = await backend.requestSizes().sorted() + XCTAssertEqual(requestSizes, [2, 2, 2, 2, 4, 4, 8]) } func testBabelDOCBatchCoordinatorFillsAroundAnItemThatDoesNotFit() async throws { @@ -543,8 +579,7 @@ final class LoopbackServerTests: XCTestCase { let coordinator = BabelDOCBatchCoordinator( broker: TranslationBroker(backend: backend), configuration: .init( - maximumBatchItems: 4, - maximumBatchCharacters: 1_000, + maximumBatchCharacters: 1_064, maximumConcurrentBatches: 1, fillDelayNanoseconds: 0 ) @@ -568,7 +603,10 @@ final class LoopbackServerTests: XCTestCase { XCTAssertEqual(requests.map { $0.items.count }, [2, 2]) XCTAssertTrue( requests.allSatisfy { - $0.items.reduce(0) { $0 + $1.text.count } == 1_000 + $0.items.reduce(0) { + $0 + $1.text.count + + BabelDOCBatchCoordinator.estimatedItemFramingCharacters + } == 1_064 } ) } @@ -576,7 +614,6 @@ final class LoopbackServerTests: XCTestCase { func testBabelDOCBatchConfigurationReadsBoundedEnvironmentOverrides() { let configuration = BabelDOCBatchCoordinator.Configuration( environment: [ - "GLOSS_BABELDOC_BATCH_ITEMS": "12", "GLOSS_BABELDOC_BATCH_CHARACTERS": "1800", "GLOSS_BABELDOC_MODEL_CONCURRENCY": "3", "GLOSS_BABELDOC_FILL_DELAY_MS": "75", @@ -584,7 +621,6 @@ final class LoopbackServerTests: XCTestCase { ] ) - XCTAssertEqual(configuration.maximumBatchItems, 12) XCTAssertEqual(configuration.maximumBatchCharacters, 1_800) XCTAssertEqual(configuration.maximumConcurrentBatches, 3) XCTAssertEqual(configuration.fillDelayNanoseconds, 75_000_000) @@ -594,7 +630,6 @@ final class LoopbackServerTests: XCTestCase { func testBabelDOCBatchConfigurationRejectsOutOfRangeOverrides() { let configuration = BabelDOCBatchCoordinator.Configuration( environment: [ - "GLOSS_BABELDOC_BATCH_ITEMS": "100", "GLOSS_BABELDOC_BATCH_CHARACTERS": "20", "GLOSS_BABELDOC_MODEL_CONCURRENCY": "0", "GLOSS_BABELDOC_FILL_DELAY_MS": "-1", @@ -602,8 +637,7 @@ final class LoopbackServerTests: XCTestCase { ] ) - XCTAssertEqual(configuration.maximumBatchItems, 12) - XCTAssertEqual(configuration.maximumBatchCharacters, 1_800) + XCTAssertEqual(configuration.maximumBatchCharacters, 3_000) XCTAssertEqual(configuration.maximumConcurrentBatches, 2) XCTAssertEqual(configuration.fillDelayNanoseconds, 25_000_000) XCTAssertEqual(configuration.refillDelayNanoseconds, 0) @@ -751,6 +785,29 @@ private actor BridgeBackend: TranslationBackend { } } +private actor SplitRecoveryBackend: TranslationBackend { + private let maximumReliableItems: Int + private var sizes: [Int] = [] + + init(maximumReliableItems: Int) { + self.maximumReliableItems = maximumReliableItems + } + + func translate(_ request: TranslationBatchRequest) async throws -> [TranslationOutput] { + sizes.append(request.items.count) + guard request.items.count <= maximumReliableItems else { + throw TranslationError.invalidResponse("Simulated incomplete structured output.") + } + return request.items.map { + TranslationOutput(id: $0.id, text: "translated:\($0.text)") + } + } + + func requestSizes() -> [Int] { + sizes + } +} + private actor CancellableBridgeBackend: TranslationBackend { private(set) var hasStarted = false private(set) var wasCancelled = false diff --git a/Tests/GlossCoreTests/SparkPackageBenchmarkTests.swift b/Tests/GlossCoreTests/SparkPackageBenchmarkTests.swift new file mode 100644 index 0000000..7c60d75 --- /dev/null +++ b/Tests/GlossCoreTests/SparkPackageBenchmarkTests.swift @@ -0,0 +1,337 @@ +import Foundation +import PDFKit +import XCTest + +@testable import GlossCore + +final class SparkPackageBenchmarkTests: XCTestCase { + private struct Result: Codable { + let model: String? + let repeatIndex: Int + let budget: Int + let attempts: Int + let sourceItems: Int + let sourceCharacters: Int + let estimatedCharacters: Int + let translatedCharacters: Int + let turns: Int + let wallMilliseconds: Int + let preparationMilliseconds: Int + let queueWaitMilliseconds: Int + let modelWaitMilliseconds: Int + let outputStreamMilliseconds: Int + let cumulativeTurnMilliseconds: Int + let error: String? + } + + private struct PackageRun { + let outputs: [TranslationOutput] + let performance: TranslationDispatchState.DocumentPerformanceSnapshot + let wallMilliseconds: Int + let attempts: Int + let error: String? + } + + func testLiveSparkPackageBenchmarkWhenRequested() async throws { + let processEnvironment = ProcessInfo.processInfo.environment + guard processEnvironment["GLOSS_RUN_SPARK_PACKAGE_BENCHMARK"] == "1" else { + throw XCTSkip( + "Set GLOSS_RUN_SPARK_PACKAGE_BENCHMARK=1 and provide input/output paths." + ) + } + + let inputPath = try XCTUnwrap(processEnvironment["GLOSS_SPARK_BENCHMARK_INPUT"]) + let outputPath = try XCTUnwrap(processEnvironment["GLOSS_SPARK_BENCHMARK_OUTPUT"]) + let model = + processEnvironment["GLOSS_SPARK_BENCHMARK_MODEL"] + ?? "gpt-5.3-codex-spark" + let budgets = try parseBudgets( + processEnvironment["GLOSS_SPARK_BENCHMARK_BUDGETS"] + ?? "1800,3000,4500,6000,8000" + ) + let repeatCount = max( + 1, + processEnvironment["GLOSS_SPARK_BENCHMARK_REPEAT"].flatMap(Int.init) ?? 3 + ) + let sourceCharacterLimit = max( + budgets.max() ?? 8_000, + processEnvironment["GLOSS_SPARK_BENCHMARK_SOURCE_CHARACTERS"] + .flatMap(Int.init) ?? 18_000 + ) + let items = try sourceItems( + from: URL(fileURLWithPath: inputPath), + characterLimit: sourceCharacterLimit + ) + let sourceCharacters = items.reduce(0) { $0 + $1.text.count } + let estimatedCharacters = items.reduce(0) { + $0 + $1.text.count + BabelDOCBatchCoordinator.estimatedItemFramingCharacters + } + + let outputURL = URL(fileURLWithPath: outputPath) + let outputDirectory = outputURL.deletingLastPathComponent() + try FileManager.default.createDirectory( + at: outputDirectory, + withIntermediateDirectories: true + ) + let runtimeLog = GlossRuntimeLog(directory: outputDirectory) + try runtimeLog.prepare() + + var clientEnvironment = processEnvironment + clientEnvironment["GLOSS_CODEX_MODEL_WAIT_HEDGE_SECONDS"] = "off" + clientEnvironment["GLOSS_CODEX_MAX_CONCURRENCY"] = "3" + clientEnvironment["GLOSS_CODEX_BACKGROUND_CONCURRENCY"] = "2" + clientEnvironment["GLOSS_CODEX_THREAD_ROTATION_TURNS"] = "10" + let dispatchState = TranslationDispatchState() + let client = CodexAppServerClient( + environment: clientEnvironment, + timeoutSeconds: 180, + model: model, + reasoningEffort: .low, + documentReasoningEffort: .low, + dispatchState: dispatchState + ) + let center = TranslationDispatchCenter( + backend: client, + configuration: .init( + maximumConcurrentJobs: 3, + maximumBackgroundJobs: 2 + ), + runtimeLog: runtimeLog, + dispatchState: dispatchState + ) + + do { + _ = try await retryOnCapacity(maximumAttempts: 6) { + try await client.translate( + TranslationBatchRequest( + items: Array(items.prefix(2)), + targetLanguage: "Chinese (Simplified)", + profile: .academic, + contentKind: .document, + context: "Spark package benchmark warm-up.", + priority: .background + ) + ) + } + + var results = loadResults(from: outputURL) + for repeatIndex in 1...repeatCount { + let order = benchmarkOrder(budgets, repeatIndex: repeatIndex) + for budget in order { + if results.contains(where: { + $0.repeatIndex == repeatIndex && $0.budget == budget + }) { + continue + } + let packageRun = try await runPackage( + items: items, + budget: budget, + center: center, + dispatchState: dispatchState, + runtimeLog: runtimeLog + ) + let outputs = packageRun.outputs + let performance = packageRun.performance + if packageRun.error == nil { + XCTAssertEqual(outputs.map(\.id), items.map(\.id)) + XCTAssertTrue(outputs.allSatisfy { !$0.text.isEmpty }) + } + + let result = Result( + model: model, + repeatIndex: repeatIndex, + budget: budget, + attempts: packageRun.attempts, + sourceItems: items.count, + sourceCharacters: sourceCharacters, + estimatedCharacters: estimatedCharacters, + translatedCharacters: outputs.reduce(0) { $0 + $1.text.count }, + turns: performance.completedTurns, + wallMilliseconds: packageRun.wallMilliseconds, + preparationMilliseconds: performance.preparationMilliseconds, + queueWaitMilliseconds: performance.queueWaitMilliseconds, + modelWaitMilliseconds: performance.modelWaitMilliseconds, + outputStreamMilliseconds: performance.outputStreamMilliseconds, + cumulativeTurnMilliseconds: performance.totalTurnMilliseconds, + error: packageRun.error + ) + results.append(result) + try write(results, to: outputURL) + print( + "SPARK_PACKAGE_BENCHMARK model=\(model) repeat=\(repeatIndex) budget=\(budget) attempts=\(result.attempts) items=\(items.count) source_chars=\(sourceCharacters) estimated_chars=\(estimatedCharacters) turns=\(result.turns) wall_ms=\(result.wallMilliseconds) model_wait_ms=\(result.modelWaitMilliseconds) output_ms=\(result.outputStreamMilliseconds) translated_chars=\(result.translatedCharacters) error=\(result.error ?? "none")" + ) + } + } + await client.stop() + } catch { + await client.stop() + throw error + } + } + + private func parseBudgets(_ value: String) throws -> [Int] { + let budgets = value.split(separator: ",").compactMap { + Int($0.trimmingCharacters(in: .whitespaces)) + } + guard !budgets.isEmpty, budgets.allSatisfy({ (200...12_000).contains($0) }) else { + throw XCTSkip("Spark package budgets must be within 200...12000.") + } + return Array(Set(budgets)).sorted() + } + + private func runPackage( + items: [TranslationItem], + budget: Int, + center: TranslationDispatchCenter, + dispatchState: TranslationDispatchState, + runtimeLog: GlossRuntimeLog + ) async throws -> PackageRun { + for attempt in 1...4 { + let broker = TranslationBroker(backend: center, cacheLimit: 0) + let coordinator = BabelDOCBatchCoordinator( + broker: broker, + configuration: .init( + maximumBatchCharacters: budget, + maximumConcurrentBatches: 2, + fillDelayNanoseconds: 0, + refillDelayNanoseconds: 0 + ), + runtimeLog: runtimeLog, + dispatchState: dispatchState + ) + let runID = UUID() + await dispatchState.beginDocumentPerformanceRun(id: runID) + let startedAt = DispatchTime.now().uptimeNanoseconds + do { + let outputs = try await coordinator.translate( + items: items, + targetLanguage: "Chinese (Simplified)", + context: "Layout-preserving academic PDF translation benchmark." + ) + let wallMilliseconds = Int( + (DispatchTime.now().uptimeNanoseconds - startedAt) / 1_000_000 + ) + let snapshot = await dispatchState.endDocumentPerformanceRun(id: runID) + return PackageRun( + outputs: outputs, + performance: try XCTUnwrap(snapshot), + wallMilliseconds: wallMilliseconds, + attempts: attempt, + error: nil + ) + } catch { + try await waitUntilIdle(center) + let snapshot = await dispatchState.endDocumentPerformanceRun(id: runID) + if attempt < 4, isCapacityError(error) { + try await Task.sleep(for: .seconds(attempt * 5)) + continue + } + return PackageRun( + outputs: [], + performance: try XCTUnwrap(snapshot), + wallMilliseconds: Int( + (DispatchTime.now().uptimeNanoseconds - startedAt) / 1_000_000 + ), + attempts: attempt, + error: error.localizedDescription + ) + } + } + throw TranslationError.backendUnavailable("Spark package benchmark exhausted retries.") + } + + private func retryOnCapacity( + maximumAttempts: Int, + operation: () async throws -> T + ) async throws -> T { + for attempt in 1...maximumAttempts { + do { + return try await operation() + } catch { + guard attempt < maximumAttempts, isCapacityError(error) else { throw error } + try await Task.sleep(for: .seconds(min(20, attempt * 5))) + } + } + throw TranslationError.backendUnavailable("Spark package benchmark exhausted retries.") + } + + private func waitUntilIdle(_ center: TranslationDispatchCenter) async throws { + for _ in 0..<1_800 { + if await center.snapshot().activeJobs == 0 { return } + try await Task.sleep(for: .milliseconds(100)) + } + throw TranslationError.timedOut("Spark package benchmark drain") + } + + private func isCapacityError(_ error: Error) -> Bool { + error.localizedDescription.localizedCaseInsensitiveContains("capacity") + } + + private func benchmarkOrder(_ budgets: [Int], repeatIndex: Int) -> [Int] { + guard budgets.count > 1 else { return budgets } + if repeatIndex.isMultiple(of: 2) { + return budgets.reversed() + } + let offset = ((repeatIndex - 1) / 2) % budgets.count + return Array(budgets[offset...] + budgets[.. [TranslationItem] { + guard let document = PDFDocument(url: inputURL) else { + throw XCTSkip("Cannot read benchmark PDF at \(inputURL.path).") + } + var items: [TranslationItem] = [] + var characterCount = 0 + for pageIndex in 0..= 3, + letterCount >= 12 + else { continue } + for chunk in chunks(of: line, maximumCharacters: 900) { + items.append( + TranslationItem(id: "item-\(items.count)", text: chunk) + ) + characterCount += chunk.count + if characterCount >= characterLimit { return items } + } + } + } + guard characterCount >= min(characterLimit, 1_000) else { + throw XCTSkip("Benchmark PDF did not provide enough extractable text.") + } + return items + } + + private func chunks(of text: String, maximumCharacters: Int) -> [String] { + var result: [String] = [] + var remainder = text[...] + while remainder.count > maximumCharacters { + let boundary = remainder.index(remainder.startIndex, offsetBy: maximumCharacters) + let prefix = remainder[.. [Result] { + guard let data = try? Data(contentsOf: outputURL) else { return [] } + return (try? JSONDecoder().decode([Result].self, from: data)) ?? [] + } +} diff --git a/Tests/GlossCoreTests/TranslationDispatchCenterTests.swift b/Tests/GlossCoreTests/TranslationDispatchCenterTests.swift index aecf300..fc8125a 100644 --- a/Tests/GlossCoreTests/TranslationDispatchCenterTests.swift +++ b/Tests/GlossCoreTests/TranslationDispatchCenterTests.swift @@ -125,8 +125,7 @@ final class TranslationDispatchCenterTests: XCTestCase { let coordinator = BabelDOCBatchCoordinator( broker: TranslationBroker(backend: center), configuration: .init( - maximumBatchItems: 1, - maximumBatchCharacters: 100, + maximumBatchCharacters: 5, maximumConcurrentBatches: 2, fillDelayNanoseconds: 0 ),