spb/zyquo-agent Public MIT

1//2//  StreamingService.swift3//  Zyquo Agent4//5//  Author: Simon-Pierre Boucher6//  Mail: contact@spboucher.ai7//8//  Ported verbatim from Zyquo Cloud (only the User-Agent string changed).9//1011import Foundation1213/// One Server-Sent Event as parsed off the wire.14struct SSEEvent {15    /// The `event:` field, if the stream names its events (Anthropic does).16    var event: String?17    /// Joined `data:` lines.18    var data: String19}2021/// Incremental SSE parser. Feed it raw lines (or byte chunks split on newlines)22/// and it yields complete events at blank-line boundaries, ignoring `:` comment23/// lines (DeepSeek sends `: keep-alive`) and unknown fields.24struct SSEParser {25    private var currentEvent: String?26    private var currentData: [String] = []2728    /// Consumes one line (without its trailing newline). Returns a completed29    /// event when the line is the blank separator, else nil.30    mutating func consume(line: String) -> SSEEvent? {31        if line.isEmpty {32            guard !currentData.isEmpty || currentEvent != nil else { return nil }33            let event = SSEEvent(event: currentEvent, data: currentData.joined(separator: "\n"))34            currentEvent = nil35            currentData = []36            return event.data.isEmpty && event.event == nil ? nil : event37        }38        if line.hasPrefix(":") { return nil } // comment / keep-alive39        if line.hasPrefix("event:") {40            currentEvent = String(line.dropFirst(6)).trimmingCharacters(in: .whitespaces)41        } else if line.hasPrefix("data:") {42            var value = String(line.dropFirst(5))43            if value.hasPrefix(" ") { value.removeFirst() }44            currentData.append(value)45        }46        // id:/retry:/unknown fields are ignored.47        return nil48    }49}5051/// Shared networking for all provider clients: request construction helpers and52/// an SSE line stream over URLSession.53enum StreamingService {54    /// URLSession tuned for long-lived streaming responses.55    static let session: URLSession = {56        let config = URLSessionConfiguration.default57        config.timeoutIntervalForRequest = 12058        config.timeoutIntervalForResource = 90059        config.httpAdditionalHeaders = ["User-Agent": "ZyquoAgent/1.0 (macOS)"]60        return URLSession(configuration: config)61    }()6263    /// POSTs `body` as JSON and returns the SSE events of the response.64    /// Throws `ProviderError` on non-2xx status (reading the full error body).65    static func sseEvents(66        for request: URLRequest,67        provider: ProviderID68    ) -> AsyncThrowingStream<SSEEvent, Error> {69        AsyncThrowingStream { continuation in70            let task = Task {71                do {72                    let (bytes, response) = try await session.bytes(for: request)73                    guard let http = response as? HTTPURLResponse else {74                        throw ProviderError.invalidResponse(provider, detail: "not an HTTP response")75                    }76                    guard (200..<300).contains(http.statusCode) else {77                        var body = Data()78                        for try await byte in bytes { body.append(byte) }79                        throw ProviderError.from(status: http.statusCode, body: body, provider: provider)80                    }81                    // NOTE: AsyncBytes.lines skips empty lines, which are the82                    // SSE event separators — split manually to preserve them.83                    var parser = SSEParser()84                    var lineBuffer = Data()85                    for try await byte in bytes {86                        if Task.isCancelled { break }87                        if byte == 0x0A { // \n88                            if lineBuffer.last == 0x0D { lineBuffer.removeLast() } // \r\n89                            let line = String(decoding: lineBuffer, as: UTF8.self)90                            lineBuffer.removeAll(keepingCapacity: true)91                            if let event = parser.consume(line: line) {92                                continuation.yield(event)93                            }94                        } else {95                            lineBuffer.append(byte)96                        }97                    }98                    // Flush a trailing line + event if the stream ended99                    // without a final newline / blank separator.100                    if !lineBuffer.isEmpty {101                        let line = String(decoding: lineBuffer, as: UTF8.self)102                        if let event = parser.consume(line: line) {103                            continuation.yield(event)104                        }105                    }106                    if let event = parser.consume(line: "") {107                        continuation.yield(event)108                    }109                    continuation.finish()110                } catch is CancellationError {111                    continuation.finish(throwing: ProviderError.cancelled)112                } catch let error as ProviderError {113                    continuation.finish(throwing: error)114                } catch {115                    continuation.finish(throwing: ProviderError.networkError(underlying: error))116                }117            }118            continuation.onTermination = { _ in task.cancel() }119        }120    }121122    /// Non-streaming JSON POST with exponential backoff on 429/5xx (3 attempts).123    /// Returns the response body data.124    static func postJSON(125        _ request: URLRequest,126        provider: ProviderID127    ) async throws -> Data {128        let maxAttempts = 3129        var lastError: ProviderError = .invalidResponse(provider, detail: "no attempts made")130        for attempt in 1...maxAttempts {131            do {132                let (data, response) = try await session.data(for: request)133                guard let http = response as? HTTPURLResponse else {134                    throw ProviderError.invalidResponse(provider, detail: "not an HTTP response")135                }136                guard (200..<300).contains(http.statusCode) else {137                    let error = ProviderError.from(status: http.statusCode, body: data, provider: provider)138                    if attempt < maxAttempts, http.statusCode == 429 || http.statusCode >= 500 {139                        lastError = error140                        let retryAfter = (response as? HTTPURLResponse)?141                            .value(forHTTPHeaderField: "Retry-After").flatMap(Double.init)142                        let delay = retryAfter ?? pow(2, Double(attempt)) * 2 // 4s, 8s143                        try await Task.sleep(nanoseconds: UInt64(delay * 1_000_000_000))144                        continue145                    }146                    throw error147                }148                return data149            } catch let error as ProviderError {150                throw error151            } catch is CancellationError {152                throw ProviderError.cancelled153            } catch {154                throw ProviderError.networkError(underlying: error)155            }156        }157        throw lastError158    }159160    /// GET returning decoded JSON data, with the same error mapping.161    static func getJSON(162        _ request: URLRequest,163        provider: ProviderID164    ) async throws -> Data {165        try await postJSON(request, provider: provider)166    }167}168