fix(streaming): typed provider errors and eof completion after terminal finish

Author: Tom-Ryder

Sources/AgentRunKit/Core/Streaming/StreamCompletion.swift

@@ -4,10 +4,13 @@ import Foundation
 public struct StreamCompletionDiagnostics: Sendable, Equatable {
     public let elapsed: Duration
     public let eventsObserved: Int
+    /// Whether the stream ended through the provider's own completion signal rather than completion inferred at EOF.
+    public let terminalMarkerSeen: Bool
 
-    public init(elapsed: Duration, eventsObserved: Int) {
+    public init(elapsed: Duration, eventsObserved: Int, terminalMarkerSeen: Bool) {
         self.elapsed = elapsed
         self.eventsObserved = eventsObserved
+        self.terminalMarkerSeen = terminalMarkerSeen
     }
 }
 

Sources/AgentRunKit/Core/Streaming/StreamProcessor.swift

@@ -84,6 +84,7 @@ private struct AudioAccumulator {
 
 private struct StreamAccumulation {
     let eventFactory: StreamEventFactory
+    let provider: ProviderIdentifier
     let started: ContinuousClock.Instant = .now
     var content = ""
     var reasoning = ""
@@ -96,6 +97,9 @@ private struct StreamAccumulation {
     var yieldedEvent = false
     var eventsObserved = 0
     var sawFinished = false
+    /// Every backend except OpenAI chat can complete only through its own terminal marker, so a stream
+    /// that never reports `.streamClosed` has necessarily seen one; the chat path overrides this at EOF.
+    var terminalMarkerSeen = true
 
     mutating func apply(
         _ input: RunStreamElement,
@@ -155,6 +159,8 @@ private struct StreamAccumulation {
             guard let iterationUsage else { return }
             totalUsage += iterationUsage
             usage = iterationUsage
+        case let .streamClosed(markerSeen):
+            terminalMarkerSeen = markerSeen
         }
     }
 
@@ -222,14 +228,20 @@ private struct StreamAccumulation {
 
     var diagnostics: StreamFailureDiagnostics {
         StreamFailureDiagnostics(
+            provider: provider,
             elapsed: ContinuousClock.now - started,
             eventsObserved: eventsObserved,
+            finishSignalSeen: sawFinished,
             lastEvent: nil
         )
     }
 
     var completionDiagnostics: StreamCompletionDiagnostics {
-        StreamCompletionDiagnostics(elapsed: ContinuousClock.now - started, eventsObserved: eventsObserved)
+        StreamCompletionDiagnostics(
+            elapsed: ContinuousClock.now - started,
+            eventsObserved: eventsObserved,
+            terminalMarkerSeen: terminalMarkerSeen
+        )
     }
 
     func streamFailure(reason: MalformedStreamReason) -> AgentError {
@@ -263,7 +275,7 @@ struct StreamProcessor {
         requestContext: RequestContext? = nil,
         requestMode: RunRequestMode = .auto
     ) async throws -> StreamIteration {
-        var state = StreamAccumulation(eventFactory: eventFactory)
+        var state = StreamAccumulation(eventFactory: eventFactory, provider: client.providerIdentifier)
         let eventObserver = requestContext?.onStreamEvent
         let completionObserver = requestContext?.onStreamComplete
 

Sources/AgentRunKit/Documentation.docc/Articles/LLMProviders.md

@@ -217,10 +217,23 @@ All clients accept a ``RetryPolicy`` controlling retry behavior on transient fai
 | `maxAttempts` | 3 | Total attempts before failing |
 | `baseDelay` | 1 second | Initial backoff duration |
 | `maxDelay` | 30 seconds | Cap on exponential backoff |
-| `streamStallTimeout` | nil | Restarts a stream if no delta arrives within this duration |
+| `streamStallTimeout` | nil | Fails a stream with ``StreamFailure/idleTimeout(diagnostics:)`` when no bytes arrive within this duration |
 
 Two static presets: `.default` (3 attempts, 1s base, 30s max) and `.none` (single attempt, no retries).
 
+Retries apply before a stream starts; once a 2xx byte stream is open, failures propagate to the caller as typed ``StreamFailure`` values rather than being retried.
+
+## Stream Termination
+
+Each streaming transport defines exactly which wire conditions end a stream successfully; anything else throws a typed ``StreamFailure``.
+
+- OpenAI-compatible Chat Completions: a `data: [DONE]` sentinel completes the stream immediately. A terminal non-`error` `finish_reason` also marks the turn complete, so a stream that ends at EOF after one is a successful completion (reported through ``StreamCompletionDiagnostics/terminalMarkerSeen``). Frames carrying a top-level `error` payload, or `finish_reason: "error"`, throw ``StreamFailure/providerError(code:message:diagnostics:)`` with the upstream code and message preserved. EOF with no finish signal throws ``StreamFailure/providerTerminationMissing(diagnostics:)``, and a `[DONE]` with no preceding finish signal throws ``StreamFailure/finishedDeltaMissing(diagnostics:)``.
+- Anthropic (and Vertex Anthropic): `message_stop` completes the stream; `error` events throw ``StreamFailure/providerError(code:message:diagnostics:)``; EOF before `message_stop` throws ``StreamFailure/providerTerminationMissing(diagnostics:)``.
+- Gemini (and Vertex Gemini): a chunk with `finishReason` completes the stream; error envelopes throw ``StreamFailure/providerError(code:message:diagnostics:)``.
+- Responses API: `response.completed` and `response.incomplete` complete the stream; `response.failed`, `response.error`, and standalone `error` events throw ``StreamFailure/providerError(code:message:diagnostics:)``.
+
+Every failure carries ``StreamFailureDiagnostics`` identifying the provider, elapsed time, events observed, and whether a finish signal had been seen before the failure.
+
 ```swift
 let client = OpenAIClient(
     apiKey: "sk-...",

Sources/AgentRunKit/LLM/Core/StreamDelta.swift

@@ -11,4 +11,6 @@ public enum StreamDelta: Sendable, Equatable {
     case audioTranscript(String)
     case audioStarted(id: String, expiresAt: Int)
     case finished(usage: TokenUsage?)
+    /// Emitted at most once at stream end; absent from backends that can only complete via their own terminal marker.
+    case streamClosed(terminalMarkerSeen: Bool)
 }

Sources/AgentRunKit/LLM/Providers/Anthropic/AnthropicClient.swift

@@ -141,7 +141,7 @@ public struct AnthropicClient: LLMClient, Sendable {
             urlRequest: urlRequest, session: session, retryPolicy: retryPolicy
         )
         requestContext?.onResponse?(httpResponse)
-        return try parseResponse(data)
+        return try parseResponse(data, provider: providerIdentifier)
     }
 
     public func stream(
@@ -411,14 +411,18 @@ extension AnthropicClient {
         return try buildJSONPostRequest(url: url, body: request, headers: headerMap)
     }
 
-    func parseResponse(_ data: Data) throws -> AssistantMessage {
+    func parseResponse(_ data: Data, provider: ProviderIdentifier) throws -> AssistantMessage {
         let response: AnthropicResponse
         do {
             response = try JSONDecoder().decode(AnthropicResponse.self, from: data)
         } catch let decodingError {
             if let err = try? JSONDecoder().decode(AnthropicErrorResponse.self, from: data),
                err.type == "error" {
-                throw AgentError.llmError(.other("\(err.error.type): \(err.error.message)"))
+                throw AgentError.llmError(.providerError(
+                    provider: provider,
+                    code: err.error.type,
+                    message: err.error.message
+                ))
             }
             throw AgentError.llmError(.decodingFailed(decodingError))
         }

Sources/AgentRunKit/LLM/Providers/Anthropic/AnthropicClientStreaming.swift

@@ -24,11 +24,11 @@ extension AnthropicClient {
 
         try await processSSEStream(
             bytes: bytes,
+            provider: providerIdentifier,
             stallTimeout: retryPolicy.streamStallTimeout
         ) { event, diagnostics in
             try await self.handleSSEEvent(
-                event, state: state, providerIdentifier: self.providerIdentifier,
-                diagnostics: diagnostics, continuation: continuation
+                event, state: state, diagnostics: diagnostics, continuation: continuation
             )
         }
         continuation.finish()
@@ -57,10 +57,11 @@ extension AnthropicClient {
 
         try await processSSEStream(
             bytes: bytes,
+            provider: providerIdentifier,
             stallTimeout: retryPolicy.streamStallTimeout
         ) { event, diagnostics in
             try await self.handleSSEEvent(
-                event, state: state, providerIdentifier: self.providerIdentifier, diagnostics: diagnostics
+                event, state: state, diagnostics: diagnostics
             ) { delta in
                 continuation.yield(.delta(delta))
             }
@@ -79,28 +80,23 @@ extension AnthropicClient {
     func handleSSEEvent(
         _ event: SSEEvent,
         state: AnthropicStreamState,
-        providerIdentifier: ProviderIdentifier = .anthropic,
         diagnostics: StreamFailureDiagnostics,
         continuation: AsyncThrowingStream<StreamDelta, Error>.Continuation
-    ) async throws -> Bool {
-        try await handleSSEEvent(
-            event, state: state, providerIdentifier: providerIdentifier, diagnostics: diagnostics
-        ) { delta in
+    ) async throws -> SSEDisposition {
+        try await handleSSEEvent(event, state: state, diagnostics: diagnostics) { delta in
             continuation.yield(delta)
         }
     }
 
     func handleSSEEvent(
         _ event: SSEEvent,
         state: AnthropicStreamState,
-        providerIdentifier: ProviderIdentifier = .anthropic,
         diagnostics: StreamFailureDiagnostics,
         yield: @Sendable (StreamDelta) -> Void
-    ) async throws -> Bool {
+    ) async throws -> SSEDisposition {
         try await handleSSEPayload(
             event.data,
             state: state,
-            providerIdentifier: providerIdentifier,
             diagnostics: diagnostics,
             yield: yield
         )
@@ -109,27 +105,25 @@ extension AnthropicClient {
     private func handleSSEPayload(
         _ payload: String,
         state: AnthropicStreamState,
-        providerIdentifier: ProviderIdentifier,
         diagnostics: StreamFailureDiagnostics,
         yield: @Sendable (StreamDelta) -> Void
-    ) async throws -> Bool {
+    ) async throws -> SSEDisposition {
         let event = try decodeEvent(AnthropicEventTypeOnly.self, from: Data(payload.utf8))
-        guard let eventType = event.type else { return false }
+        guard let eventType = event.type else { return .continue }
 
         return try await dispatchEvent(
             eventType, data: Data(payload.utf8),
-            state: state, providerIdentifier: providerIdentifier, diagnostics: diagnostics, yield: yield
+            state: state, diagnostics: diagnostics, yield: yield
         )
     }
 
     private func dispatchEvent(
         _ type: AnthropicSSEEvent,
         data: Data,
         state: AnthropicStreamState,
-        providerIdentifier: ProviderIdentifier,
         diagnostics: StreamFailureDiagnostics,
         yield: @Sendable (StreamDelta) -> Void
-    ) async throws -> Bool {
+    ) async throws -> SSEDisposition {
         switch type {
         case .messageStart:
             try await handleMessageStart(data: data, state: state)
@@ -152,11 +146,11 @@ extension AnthropicClient {
         case .messageStop:
             await state.markCompleted()
             await yield(.finished(usage: state.finalUsage()))
-            return true
+            return .complete
         case .error:
-            try handleError(data: data, providerIdentifier: providerIdentifier)
+            try handleError(data: data, diagnostics: diagnostics)
         }
-        return false
+        return .continue
     }
 
     private func handleBlockStart(
@@ -274,12 +268,12 @@ extension AnthropicClient {
         await state.setOutputTokens(event.usage?.outputTokens ?? 0)
     }
 
-    private func handleError(data: Data, providerIdentifier: ProviderIdentifier) throws {
+    private func handleError(data: Data, diagnostics: StreamFailureDiagnostics) throws {
         let event = try decodeEvent(AnthropicStreamErrorEvent.self, from: data)
         throw AgentError.llmError(.streamFailed(.providerError(
-            provider: providerIdentifier,
             code: event.error.type,
-            message: event.error.message
+            message: event.error.message,
+            diagnostics: diagnostics
         )))
     }
 

Sources/AgentRunKit/LLM/Providers/Gemini/GeminiClient.swift

@@ -58,7 +58,7 @@ public struct GeminiClient: LLMClient, Sendable {
             urlRequest: urlRequest, session: session, retryPolicy: retryPolicy
         )
         requestContext?.onResponse?(httpResponse)
-        return try parseResponse(data)
+        return try parseResponse(data, provider: providerIdentifier)
     }
 
     public func stream(
@@ -250,9 +250,13 @@ extension GeminiClient {
         return str
     }
 
-    func parseResponse(_ data: Data) throws -> AssistantMessage {
+    func parseResponse(_ data: Data, provider: ProviderIdentifier) throws -> AssistantMessage {
         if let err = try? JSONDecoder().decode(GeminiErrorResponse.self, from: data) {
-            throw AgentError.llmError(.other("\(err.error.status): \(err.error.message)"))
+            throw AgentError.llmError(.providerError(
+                provider: provider,
+                code: err.error.status,
+                message: err.error.message
+            ))
         }
 
         let response: GeminiResponse

Sources/AgentRunKit/LLM/Providers/Gemini/GeminiClientStreaming.swift

@@ -28,10 +28,11 @@ extension GeminiClient {
 
         try await processSSEStream(
             bytes: bytes,
+            provider: providerIdentifier,
             stallTimeout: retryPolicy.streamStallTimeout
-        ) { event, _ in
+        ) { event, diagnostics in
             try await self.handleSSEEvent(
-                event, state: state, providerIdentifier: self.providerIdentifier, continuation: continuation
+                event, state: state, diagnostics: diagnostics, continuation: continuation
             )
         }
         continuation.finish()
@@ -40,28 +41,28 @@ extension GeminiClient {
     func handleSSEEvent(
         _ event: SSEEvent,
         state: GeminiStreamState,
-        providerIdentifier: ProviderIdentifier = .gemini,
+        diagnostics: StreamFailureDiagnostics,
         continuation: AsyncThrowingStream<StreamDelta, Error>.Continuation
-    ) async throws -> Bool {
+    ) async throws -> SSEDisposition {
         try await handleSSEPayload(
-            event.data, state: state, providerIdentifier: providerIdentifier, continuation: continuation
+            event.data, state: state, diagnostics: diagnostics, continuation: continuation
         )
     }
 
     private func handleSSEPayload(
         _ payload: String,
         state: GeminiStreamState,
-        providerIdentifier: ProviderIdentifier,
+        diagnostics: StreamFailureDiagnostics,
         continuation: AsyncThrowingStream<StreamDelta, Error>.Continuation
-    ) async throws -> Bool {
+    ) async throws -> SSEDisposition {
         let data = Data(payload.utf8)
 
         if let errorResponse = try? JSONDecoder().decode(GeminiErrorResponse.self, from: data) {
             throw AgentError.llmError(
                 .streamFailed(.providerError(
-                    provider: providerIdentifier,
                     code: errorResponse.error.status,
-                    message: errorResponse.error.message
+                    message: errorResponse.error.message,
+                    diagnostics: diagnostics
                 ))
             )
         }
@@ -73,7 +74,7 @@ extension GeminiClient {
             throw AgentError.llmError(.decodingFailed(error))
         }
 
-        guard let candidate = response.candidates?.first else { return false }
+        guard let candidate = response.candidates?.first else { return .continue }
 
         for part in candidate.content?.parts ?? [] {
             if let functionCall = part.functionCall {
@@ -120,10 +121,10 @@ extension GeminiClient {
             }
 
             continuation.yield(.finished(usage: response.usageMetadata?.tokenUsage))
-            return true
+            return .complete
         }
 
-        return false
+        return .continue
     }
 }