ios: add multi-polling support
This commit is contained in:
@@ -116,6 +116,10 @@ actor SybilAPIClient: SybilAPIClienting {
|
||||
try await request("/v1/models", method: "GET", responseType: ModelCatalogResponse.self)
|
||||
}
|
||||
|
||||
func getActiveRuns() async throws -> ActiveRunsResponse {
|
||||
try await request("/v1/active-runs", method: "GET", responseType: ActiveRunsResponse.self)
|
||||
}
|
||||
|
||||
func runCompletionStream(
|
||||
body: CompletionStreamRequest,
|
||||
onEvent: @escaping @Sendable (CompletionStreamEvent) async -> Void
|
||||
@@ -133,43 +137,35 @@ actor SybilAPIClient: SybilAPIClienting {
|
||||
)
|
||||
|
||||
try await stream(request: request) { eventName, dataText in
|
||||
switch eventName {
|
||||
case "meta":
|
||||
let payload: CompletionStreamMeta = try Self.decodeEvent(dataText, as: CompletionStreamMeta.self, eventName: eventName)
|
||||
await onEvent(.meta(payload))
|
||||
case "tool_call":
|
||||
let payload: CompletionStreamToolCall = try Self.decodeEvent(dataText, as: CompletionStreamToolCall.self, eventName: eventName)
|
||||
await onEvent(.toolCall(payload))
|
||||
case "delta":
|
||||
let payload: CompletionStreamDelta = try Self.decodeEvent(dataText, as: CompletionStreamDelta.self, eventName: eventName)
|
||||
await onEvent(.delta(payload))
|
||||
case "done":
|
||||
do {
|
||||
let payload: CompletionStreamDone = try Self.decodeEvent(dataText, as: CompletionStreamDone.self, eventName: eventName)
|
||||
await onEvent(.done(payload))
|
||||
} catch {
|
||||
if let recovered = Self.decodeLastJSONLine(dataText, as: CompletionStreamDone.self) {
|
||||
SybilLog.warning(
|
||||
SybilLog.network,
|
||||
"Recovered chat stream done payload from concatenated SSE data"
|
||||
)
|
||||
await onEvent(.done(recovered))
|
||||
} else {
|
||||
throw error
|
||||
}
|
||||
}
|
||||
case "error":
|
||||
let payload: StreamErrorPayload = try Self.decodeEvent(dataText, as: StreamErrorPayload.self, eventName: eventName)
|
||||
await onEvent(.error(payload))
|
||||
default:
|
||||
SybilLog.warning(SybilLog.network, "Ignoring unknown chat stream event '\(eventName)'")
|
||||
await onEvent(.ignored)
|
||||
}
|
||||
try await Self.handleCompletionStreamEvent(eventName: eventName, dataText: dataText, onEvent: onEvent)
|
||||
}
|
||||
|
||||
SybilLog.info(SybilLog.network, "Chat stream completed")
|
||||
}
|
||||
|
||||
func attachCompletionStream(
|
||||
chatID: String,
|
||||
onEvent: @escaping @Sendable (CompletionStreamEvent) async -> Void
|
||||
) async throws {
|
||||
let request = try makeRequest(
|
||||
path: "/v1/chats/\(chatID)/stream/attach",
|
||||
method: "POST",
|
||||
body: nil,
|
||||
acceptsSSE: true
|
||||
)
|
||||
|
||||
SybilLog.info(
|
||||
SybilLog.network,
|
||||
"Attaching chat stream POST \(request.url?.absoluteString ?? "<unknown>")"
|
||||
)
|
||||
|
||||
try await stream(request: request) { eventName, dataText in
|
||||
try await Self.handleCompletionStreamEvent(eventName: eventName, dataText: dataText, onEvent: onEvent)
|
||||
}
|
||||
|
||||
SybilLog.info(SybilLog.network, "Attached chat stream completed")
|
||||
}
|
||||
|
||||
func runSearchStream(
|
||||
searchID: String,
|
||||
body: SearchRunRequest,
|
||||
@@ -188,34 +184,35 @@ actor SybilAPIClient: SybilAPIClienting {
|
||||
)
|
||||
|
||||
try await stream(request: request) { eventName, dataText in
|
||||
switch eventName {
|
||||
case "search_results":
|
||||
let payload: SearchResultsPayload = try Self.decodeEvent(dataText, as: SearchResultsPayload.self, eventName: eventName)
|
||||
await onEvent(.searchResults(payload))
|
||||
case "search_error":
|
||||
let payload: SearchErrorPayload = try Self.decodeEvent(dataText, as: SearchErrorPayload.self, eventName: eventName)
|
||||
await onEvent(.searchError(payload))
|
||||
case "answer":
|
||||
let payload: SearchAnswerPayload = try Self.decodeEvent(dataText, as: SearchAnswerPayload.self, eventName: eventName)
|
||||
await onEvent(.answer(payload))
|
||||
case "answer_error":
|
||||
let payload: SearchErrorPayload = try Self.decodeEvent(dataText, as: SearchErrorPayload.self, eventName: eventName)
|
||||
await onEvent(.answerError(payload))
|
||||
case "done":
|
||||
let payload: SearchDonePayload = try Self.decodeEvent(dataText, as: SearchDonePayload.self, eventName: eventName)
|
||||
await onEvent(.done(payload))
|
||||
case "error":
|
||||
let payload: StreamErrorPayload = try Self.decodeEvent(dataText, as: StreamErrorPayload.self, eventName: eventName)
|
||||
await onEvent(.error(payload))
|
||||
default:
|
||||
SybilLog.warning(SybilLog.network, "Ignoring unknown search stream event '\(eventName)'")
|
||||
await onEvent(.ignored)
|
||||
}
|
||||
try await Self.handleSearchStreamEvent(eventName: eventName, dataText: dataText, onEvent: onEvent)
|
||||
}
|
||||
|
||||
SybilLog.info(SybilLog.network, "Search stream completed")
|
||||
}
|
||||
|
||||
func attachSearchStream(
|
||||
searchID: String,
|
||||
onEvent: @escaping @Sendable (SearchStreamEvent) async -> Void
|
||||
) async throws {
|
||||
let request = try makeRequest(
|
||||
path: "/v1/searches/\(searchID)/run/stream/attach",
|
||||
method: "POST",
|
||||
body: nil,
|
||||
acceptsSSE: true
|
||||
)
|
||||
|
||||
SybilLog.info(
|
||||
SybilLog.network,
|
||||
"Attaching search stream POST \(request.url?.absoluteString ?? "<unknown>")"
|
||||
)
|
||||
|
||||
try await stream(request: request) { eventName, dataText in
|
||||
try await Self.handleSearchStreamEvent(eventName: eventName, dataText: dataText, onEvent: onEvent)
|
||||
}
|
||||
|
||||
SybilLog.info(SybilLog.network, "Attached search stream completed")
|
||||
}
|
||||
|
||||
private func request<Response: Decodable>(
|
||||
_ path: String,
|
||||
method: String,
|
||||
@@ -498,6 +495,75 @@ actor SybilAPIClient: SybilAPIClienting {
|
||||
return try? Self.decodeJSON(type, from: data)
|
||||
}
|
||||
|
||||
private static func handleCompletionStreamEvent(
|
||||
eventName: String,
|
||||
dataText: String,
|
||||
onEvent: @escaping @Sendable (CompletionStreamEvent) async -> Void
|
||||
) async throws {
|
||||
switch eventName {
|
||||
case "meta":
|
||||
let payload: CompletionStreamMeta = try Self.decodeEvent(dataText, as: CompletionStreamMeta.self, eventName: eventName)
|
||||
await onEvent(.meta(payload))
|
||||
case "tool_call":
|
||||
let payload: CompletionStreamToolCall = try Self.decodeEvent(dataText, as: CompletionStreamToolCall.self, eventName: eventName)
|
||||
await onEvent(.toolCall(payload))
|
||||
case "delta":
|
||||
let payload: CompletionStreamDelta = try Self.decodeEvent(dataText, as: CompletionStreamDelta.self, eventName: eventName)
|
||||
await onEvent(.delta(payload))
|
||||
case "done":
|
||||
do {
|
||||
let payload: CompletionStreamDone = try Self.decodeEvent(dataText, as: CompletionStreamDone.self, eventName: eventName)
|
||||
await onEvent(.done(payload))
|
||||
} catch {
|
||||
if let recovered = Self.decodeLastJSONLine(dataText, as: CompletionStreamDone.self) {
|
||||
SybilLog.warning(
|
||||
SybilLog.network,
|
||||
"Recovered chat stream done payload from concatenated SSE data"
|
||||
)
|
||||
await onEvent(.done(recovered))
|
||||
} else {
|
||||
throw error
|
||||
}
|
||||
}
|
||||
case "error":
|
||||
let payload: StreamErrorPayload = try Self.decodeEvent(dataText, as: StreamErrorPayload.self, eventName: eventName)
|
||||
await onEvent(.error(payload))
|
||||
default:
|
||||
SybilLog.warning(SybilLog.network, "Ignoring unknown chat stream event '\(eventName)'")
|
||||
await onEvent(.ignored)
|
||||
}
|
||||
}
|
||||
|
||||
private static func handleSearchStreamEvent(
|
||||
eventName: String,
|
||||
dataText: String,
|
||||
onEvent: @escaping @Sendable (SearchStreamEvent) async -> Void
|
||||
) async throws {
|
||||
switch eventName {
|
||||
case "search_results":
|
||||
let payload: SearchResultsPayload = try Self.decodeEvent(dataText, as: SearchResultsPayload.self, eventName: eventName)
|
||||
await onEvent(.searchResults(payload))
|
||||
case "search_error":
|
||||
let payload: SearchErrorPayload = try Self.decodeEvent(dataText, as: SearchErrorPayload.self, eventName: eventName)
|
||||
await onEvent(.searchError(payload))
|
||||
case "answer":
|
||||
let payload: SearchAnswerPayload = try Self.decodeEvent(dataText, as: SearchAnswerPayload.self, eventName: eventName)
|
||||
await onEvent(.answer(payload))
|
||||
case "answer_error":
|
||||
let payload: SearchErrorPayload = try Self.decodeEvent(dataText, as: SearchErrorPayload.self, eventName: eventName)
|
||||
await onEvent(.answerError(payload))
|
||||
case "done":
|
||||
let payload: SearchDonePayload = try Self.decodeEvent(dataText, as: SearchDonePayload.self, eventName: eventName)
|
||||
await onEvent(.done(payload))
|
||||
case "error":
|
||||
let payload: StreamErrorPayload = try Self.decodeEvent(dataText, as: StreamErrorPayload.self, eventName: eventName)
|
||||
await onEvent(.error(payload))
|
||||
default:
|
||||
SybilLog.warning(SybilLog.network, "Ignoring unknown search stream event '\(eventName)'")
|
||||
await onEvent(.ignored)
|
||||
}
|
||||
}
|
||||
|
||||
private static func flushSSEEvent(
|
||||
eventName: inout String,
|
||||
dataLines: inout [String]
|
||||
|
||||
Reference in New Issue
Block a user