Skip to content

Commit 8e7c706

Browse files
tim-smarttim-smart
authored
Bound pending SSE decoder state (#6777)
1 parent 5f63adb commit 8e7c706

17 files changed

Lines changed: 247 additions & 42 deletions

File tree

‎.changeset/eff-218-bound-sse-pending-state.md‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,10 @@
1+
---
2+
"effect": patch
3+
"@effect/ai-anthropic": patch
4+
"@effect/ai-openai": patch
5+
"@effect/ai-openai-compat": patch
6+
"@effect/ai-openrouter": patch
7+
"@effect/openapi-generator": patch
8+
---
9+
10+
Bound pending SSE decoder state with a configurable maximum event size.

‎packages/ai/anthropic/src/AnthropicClient.ts‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -52,7 +52,7 @@ export interface Service {
5252
schema: S
5353
) => (request: HttpClientRequest.HttpClientRequest) => Stream.Stream<
5454
S["Type"],
55-
HttpClientError.HttpClientError | Schema.SchemaError | Sse.Retry,
55+
HttpClientError.HttpClientError | Schema.SchemaError | Sse.Retry | Sse.SseError,
5656
S["DecodingServices"]
5757
>
5858

@@ -259,7 +259,7 @@ export const make = Effect.fnUntraced(
259259
<S extends Sse.EventCodec>(schema: S) =>
260260
(request: HttpClientRequest.HttpClientRequest): Stream.Stream<
261261
S["Type"],
262-
HttpClientError.HttpClientError | Schema.SchemaError | Sse.Retry,
262+
HttpClientError.HttpClientError | Schema.SchemaError | Sse.Retry | Sse.SseError,
263263
S["DecodingServices"]
264264
> =>
265265
httpClientOk.execute(request).pipe(
@@ -312,6 +312,7 @@ export const make = Effect.fnUntraced(
312312
Stream.catchTags({
313313
// TODO: handle SSE retries
314314
Retry: (error) => Stream.die(error),
315+
SseError: (error) => Stream.fail(Errors.mapSseError(error, "createMessageStream")),
315316
HttpClientError: (error) => Stream.fromEffect(Errors.mapHttpClientError(error, "createMessageStream")),
316317
SchemaError: (error) => Stream.fail(Errors.mapSchemaError(error, "createMessageStream"))
317318
})

‎packages/ai/anthropic/src/internal/errors.ts‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@ import * as Redactable from "effect/Redactable"
88
import * as Schema from "effect/Schema"
99
import * as AiError from "effect/unstable/ai/AiError"
1010
import type * as Response from "effect/unstable/ai/Response"
11+
import type * as Sse from "effect/unstable/encoding/Sse"
1112
import type * as HttpClientError from "effect/unstable/http/HttpClientError"
1213
import type * as HttpClientRequest from "effect/unstable/http/HttpClientRequest"
1314
import type * as HttpClientResponse from "effect/unstable/http/HttpClientResponse"
@@ -52,6 +53,17 @@ export const mapSchemaError = dual<
5253
reason: AiError.InvalidOutputError.fromSchemaError(error)
5354
}))
5455

56+
/** @internal */
57+
export const mapSseError = dual<
58+
(method: string) => (error: Sse.SseError) => AiError.AiError,
59+
(error: Sse.SseError, method: string) => AiError.AiError
60+
>(2, (error, method) =>
61+
AiError.make({
62+
module: "AnthropicClient",
63+
method,
64+
reason: new AiError.InvalidOutputError({ description: error.message })
65+
}))
66+
5567
/** @internal */
5668
export const mapClientError = dual<
5769
(method: string) => (error: Generated.AnthropicClientError<string, AnthropicClientErrorBody>) => AiError.AiError,

‎packages/ai/openai-compat/src/OpenAiClient.ts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -218,6 +218,7 @@ export const make = Effect.fnUntraced(
218218
Stream.takeUntil((event) => event === "[DONE]"),
219219
Stream.catchTags({
220220
Retry: (error) => Stream.die(error),
221+
SseError: (error) => Stream.fail(Errors.mapSseError(error, "createResponseStream")),
221222
HttpClientError: (error) => Stream.fromEffect(Errors.mapHttpClientError(error, "createResponseStream"))
222223
})
223224
) as any

‎packages/ai/openai-compat/src/internal/errors.ts‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,7 @@ import * as SchemaTransformation from "effect/SchemaTransformation"
1010
import * as String from "effect/String"
1111
import * as AiError from "effect/unstable/ai/AiError"
1212
import type * as Response from "effect/unstable/ai/Response"
13+
import type * as Sse from "effect/unstable/encoding/Sse"
1314
import type * as HttpClientError from "effect/unstable/http/HttpClientError"
1415
import type * as HttpClientRequest from "effect/unstable/http/HttpClientRequest"
1516
import type * as HttpClientResponse from "effect/unstable/http/HttpClientResponse"
@@ -49,6 +50,17 @@ export const mapSchemaError = dual<
4950
reason: AiError.InvalidOutputError.fromSchemaError(error)
5051
}))
5152

53+
/** @internal */
54+
export const mapSseError = dual<
55+
(method: string) => (error: Sse.SseError) => AiError.AiError,
56+
(error: Sse.SseError, method: string) => AiError.AiError
57+
>(2, (error, method) =>
58+
AiError.make({
59+
module: "OpenAiClient",
60+
method,
61+
reason: new AiError.InvalidOutputError({ description: error.message })
62+
}))
63+
5264
/** @internal */
5365
export const mapHttpClientError = dual<
5466
(method: string) => (error: HttpClientError.HttpClientError) => Effect.Effect<never, AiError.AiError>,

‎packages/ai/openai/src/Generated.ts‎

Lines changed: 7 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -30441,7 +30441,7 @@ export const make = (
3044130441
request: HttpClientRequest.HttpClientRequest
3044230442
): Stream.Stream<
3044330443
{ readonly event: string; readonly id: string | undefined; readonly data: Type },
30444-
HttpClientError.HttpClientError | SchemaError | Sse.Retry,
30444+
HttpClientError.HttpClientError | SchemaError | Sse.Retry | Sse.SseError,
3044530445
DecodingServices
3044630446
> =>
3044730447
HttpClient.filterStatusOk(httpClient).execute(request).pipe(
@@ -32832,7 +32832,7 @@ export interface OpenAiClient {
3283232832
options: { readonly payload: typeof CreateSpeechRequestJson.Encoded }
3283332833
) => Stream.Stream<
3283432834
{ readonly event: string; readonly id: string | undefined; readonly data: typeof CreateSpeech200Sse.Type },
32835-
HttpClientError.HttpClientError | SchemaError | Sse.Retry,
32835+
HttpClientError.HttpClientError | SchemaError | Sse.Retry | Sse.SseError,
3283632836
typeof CreateSpeech200Sse.DecodingServices
3283732837
>
3283832838
/**
@@ -32868,7 +32868,7 @@ export interface OpenAiClient {
3286832868
options: { readonly payload: typeof CreateTranscriptionRequestFormData.Encoded }
3286932869
) => Stream.Stream<
3287032870
{ readonly event: string; readonly id: string | undefined; readonly data: typeof CreateTranscription200Sse.Type },
32871-
HttpClientError.HttpClientError | SchemaError | Sse.Retry,
32871+
HttpClientError.HttpClientError | SchemaError | Sse.Retry | Sse.SseError,
3287232872
typeof CreateTranscription200Sse.DecodingServices
3287332873
>
3287432874
/**
@@ -33058,7 +33058,7 @@ export interface OpenAiClient {
3305833058
options: { readonly payload: typeof CreateChatCompletionRequestJson.Encoded }
3305933059
) => Stream.Stream<
3306033060
{ readonly event: string; readonly id: string | undefined; readonly data: typeof CreateChatCompletion200Sse.Type },
33061-
HttpClientError.HttpClientError | SchemaError | Sse.Retry,
33061+
HttpClientError.HttpClientError | SchemaError | Sse.Retry | Sse.SseError,
3306233062
typeof CreateChatCompletion200Sse.DecodingServices
3306333063
>
3306433064
/**
@@ -33659,7 +33659,7 @@ export interface OpenAiClient {
3365933659
options: { readonly payload: typeof CreateImageEditRequestFormData.Encoded }
3366033660
) => Stream.Stream<
3366133661
{ readonly event: string; readonly id: string | undefined; readonly data: typeof CreateImageEdit200Sse.Type },
33662-
HttpClientError.HttpClientError | SchemaError | Sse.Retry,
33662+
HttpClientError.HttpClientError | SchemaError | Sse.Retry | Sse.SseError,
3366333663
typeof CreateImageEdit200Sse.DecodingServices
3366433664
>
3366533665
/**
@@ -33678,7 +33678,7 @@ export interface OpenAiClient {
3367833678
options: { readonly payload: typeof CreateImageRequestJson.Encoded }
3367933679
) => Stream.Stream<
3368033680
{ readonly event: string; readonly id: string | undefined; readonly data: typeof CreateImage200Sse.Type },
33681-
HttpClientError.HttpClientError | SchemaError | Sse.Retry,
33681+
HttpClientError.HttpClientError | SchemaError | Sse.Retry | Sse.SseError,
3368233682
typeof CreateImage200Sse.DecodingServices
3368333683
>
3368433684
/**
@@ -34879,7 +34879,7 @@ export interface OpenAiClient {
3487934879
options: { readonly payload: typeof CreateResponseRequestJson.Encoded }
3488034880
) => Stream.Stream<
3488134881
{ readonly event: string; readonly id: string | undefined; readonly data: typeof CreateResponse200Sse.Type },
34882-
HttpClientError.HttpClientError | SchemaError | Sse.Retry,
34882+
HttpClientError.HttpClientError | SchemaError | Sse.Retry | Sse.SseError,
3488334883
typeof CreateResponse200Sse.DecodingServices
3488434884
>
3488534885
/**

‎packages/ai/openai/src/OpenAiClient.ts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -276,6 +276,7 @@ export const make = Effect.fnUntraced(
276276
Stream.catchTags({
277277
// TODO: handle SSE retries
278278
Retry: (error) => Stream.die(error),
279+
SseError: (error) => Stream.fail(Errors.mapSseError(error, "createResponseStream")),
279280
HttpClientError: (error) => Stream.fromEffect(Errors.mapHttpClientError(error, "createResponseStream")),
280281
SchemaError: (error) => Stream.fail(Errors.mapSchemaError(error, "createResponseStream"))
281282
})

‎packages/ai/openai/src/internal/errors.ts‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@ import * as Redactable from "effect/Redactable"
88
import * as Schema from "effect/Schema"
99
import * as AiError from "effect/unstable/ai/AiError"
1010
import type * as Response from "effect/unstable/ai/Response"
11+
import type * as Sse from "effect/unstable/encoding/Sse"
1112
import type * as HttpClientError from "effect/unstable/http/HttpClientError"
1213
import type * as HttpClientRequest from "effect/unstable/http/HttpClientRequest"
1314
import type * as HttpClientResponse from "effect/unstable/http/HttpClientResponse"
@@ -42,6 +43,17 @@ export const mapSchemaError = dual<
4243
reason: AiError.InvalidOutputError.fromSchemaError(error)
4344
}))
4445

46+
/** @internal */
47+
export const mapSseError = dual<
48+
(method: string) => (error: Sse.SseError) => AiError.AiError,
49+
(error: Sse.SseError, method: string) => AiError.AiError
50+
>(2, (error, method) =>
51+
AiError.make({
52+
module: "OpenAiClient",
53+
method,
54+
reason: new AiError.InvalidOutputError({ description: error.message })
55+
}))
56+
4557
/** @internal */
4658
export const mapHttpClientError = dual<
4759
(method: string) => (error: HttpClientError.HttpClientError) => Effect.Effect<never, AiError.AiError>,

‎packages/ai/openrouter/src/Generated.ts‎

Lines changed: 7 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -32803,7 +32803,7 @@ export const make = (
3280332803
request: HttpClientRequest.HttpClientRequest
3280432804
): Stream.Stream<
3280532805
{ readonly event: string; readonly id: string | undefined; readonly data: Type },
32806-
HttpClientError.HttpClientError | SchemaError | Sse.Retry,
32806+
HttpClientError.HttpClientError | SchemaError | Sse.Retry | Sse.SseError,
3280732807
DecodingServices
3280832808
> =>
3280932809
HttpClient.filterStatusOk(httpClient).execute(request).pipe(
@@ -34824,7 +34824,7 @@ export interface OpenRouterClient {
3482434824
readonly id: string | undefined
3482534825
readonly data: typeof SendChatCompletionRequest200Sse.Type
3482634826
},
34827-
HttpClientError.HttpClientError | SchemaError | Sse.Retry,
34827+
HttpClientError.HttpClientError | SchemaError | Sse.Retry | Sse.SseError,
3482834828
typeof SendChatCompletionRequest200Sse.DecodingServices
3482934829
>
3483034830
/**
@@ -35005,7 +35005,7 @@ export interface OpenRouterClient {
3500535005
}
3500635006
) => Stream.Stream<
3500735007
{ readonly event: string; readonly id: string | undefined; readonly data: typeof CreateEmbeddings200Sse.Type },
35008-
HttpClientError.HttpClientError | SchemaError | Sse.Retry,
35008+
HttpClientError.HttpClientError | SchemaError | Sse.Retry | Sse.SseError,
3500935009
typeof CreateEmbeddings200Sse.DecodingServices
3501035010
>
3501135011
/**
@@ -35450,7 +35450,7 @@ export interface OpenRouterClient {
3545035450
}
3545135451
) => Stream.Stream<
3545235452
{ readonly event: string; readonly id: string | undefined; readonly data: typeof CreateImages200Sse.Type },
35453-
HttpClientError.HttpClientError | SchemaError | Sse.Retry,
35453+
HttpClientError.HttpClientError | SchemaError | Sse.Retry | Sse.SseError,
3545435454
typeof CreateImages200Sse.DecodingServices
3545535455
>
3545635456
/**
@@ -35619,7 +35619,7 @@ export interface OpenRouterClient {
3561935619
}
3562035620
) => Stream.Stream<
3562135621
{ readonly event: string; readonly id: string | undefined; readonly data: typeof CreateMessages200Sse.Type },
35622-
HttpClientError.HttpClientError | SchemaError | Sse.Retry,
35622+
HttpClientError.HttpClientError | SchemaError | Sse.Retry | Sse.SseError,
3562335623
typeof CreateMessages200Sse.DecodingServices
3562435624
>
3562535625
/**
@@ -35987,7 +35987,7 @@ export interface OpenRouterClient {
3598735987
}
3598835988
) => Stream.Stream<
3598935989
{ readonly event: string; readonly id: string | undefined; readonly data: typeof CreateRerank200Sse.Type },
35990-
HttpClientError.HttpClientError | SchemaError | Sse.Retry,
35990+
HttpClientError.HttpClientError | SchemaError | Sse.Retry | Sse.SseError,
3599135991
typeof CreateRerank200Sse.DecodingServices
3599235992
>
3599335993
/**
@@ -36028,7 +36028,7 @@ export interface OpenRouterClient {
3602836028
}
3602936029
) => Stream.Stream<
3603036030
{ readonly event: string; readonly id: string | undefined; readonly data: typeof CreateResponses200Sse.Type },
36031-
HttpClientError.HttpClientError | SchemaError | Sse.Retry,
36031+
HttpClientError.HttpClientError | SchemaError | Sse.Retry | Sse.SseError,
3603236032
typeof CreateResponses200Sse.DecodingServices
3603336033
>
3603436034
/**

‎packages/ai/openrouter/src/OpenRouterClient.ts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -228,6 +228,7 @@ export const make = Effect.fnUntraced(
228228
Stream.catchTags({
229229
// TODO: handle SSE retries
230230
Retry: (error) => Stream.die(error),
231+
SseError: (error) => Stream.fail(Errors.mapSseError(error, "createChatCompletionStream")),
231232
HttpClientError: (error) => Stream.fromEffect(Errors.mapHttpClientError(error, "createChatCompletionStream")),
232233
SchemaError: (error) => Stream.fail(Errors.mapSchemaError(error, "createChatCompletionStream"))
233234
})

0 commit comments

Comments
 (0)