fix(ai-openai): decode nested Responses API error stream events

.changeset/openai-tolerant-error-stream-event.md

@@ -0,0 +1,9 @@
+---
+"@effect/ai-openai": patch
+---
+
+Decode OpenAI Responses API `error` stream events whose payload is nested under `error`.
+
+The Responses API documents the `error` stream event with `code`, `message`, and `param` at the top level, but mid-stream errors (for example quota exhaustion) instead emit the standard error envelope nested under `error` — `{ "type": "error", "error": { "code", "message", "param", ... }, "sequence_number" }`. The strict schema only matched the flat shape, so a nested error event failed to decode and aborted the whole stream with an opaque schema error (`Expected UnknownResponseStreamEvent … Missing key at ["data"]["code"]`) instead of surfacing the actual error.
+
+`ResponseErrorEvent` now accepts both wire shapes and normalizes them to the documented shape, so the error's `code` and `message` always surface and the decoded type is unchanged.

packages/ai/openai/src/OpenAiSchema.ts

@@ -9,6 +9,7 @@
 import * as Effect from "effect/Effect"
 import * as Predicate from "effect/Predicate"
 import * as Schema from "effect/Schema"
+import * as SchemaTransformation from "effect/SchemaTransformation"
 
 const UnknownRecord = Schema.Record(Schema.String, Schema.Unknown)
 
@@ -1032,7 +1033,7 @@ const ResponseImageGenerationCallPartialImageEvent = Schema.Struct({
   partial_image_b64: Schema.String
 })
 
-const ResponseErrorEvent = Schema.Struct({
+const ResponseErrorEventDecoded = Schema.Struct({
   type: Schema.Literal("error"),
   code: Schema.NullOr(Schema.String),
   message: Schema.String,
@@ -1041,6 +1042,43 @@ const ResponseErrorEvent = Schema.Struct({
   status: Schema.optionalKey(Schema.Int)
 })
 
+// The Responses API documents the `error` stream event with `code`, `message`,
+// and `param` at the top level, but mid-stream errors (for example quota
+// exhaustion) instead emit the standard error envelope nested under `error`
+// (`{ type: "error", error: { code, message, param, ... }, sequence_number }`).
+// Accept both wire shapes and normalize them to the documented shape, so an
+// error event — the one event that most needs to surface — never fails to
+// decode and abort the whole stream with an opaque schema error.
+const ResponseErrorEvent = Schema.Struct({
+  type: Schema.Literal("error"),
+  code: Schema.optionalKey(Schema.NullOr(Schema.String)),
+  message: Schema.optionalKey(Schema.String),
+  param: Schema.optionalKey(Schema.NullOr(Schema.String)),
+  sequence_number: Schema.optionalKey(Schema.Int),
+  status: Schema.optionalKey(Schema.Int),
+  error: Schema.optionalKey(Schema.Struct({
+    type: Schema.optionalKey(Schema.NullOr(Schema.String)),
+    code: Schema.optionalKey(Schema.NullOr(Schema.String)),
+    message: Schema.optionalKey(Schema.String),
+    param: Schema.optionalKey(Schema.NullOr(Schema.String))
+  }))
+}).pipe(
+  Schema.decodeTo(
+    ResponseErrorEventDecoded,
+    SchemaTransformation.transform({
+      decode: (input) => ({
+        type: "error" as const,
+        code: input.code ?? input.error?.code ?? null,
+        message: input.message ?? input.error?.message ?? "An unknown error occurred",
+        param: input.param ?? input.error?.param ?? null,
+        sequence_number: input.sequence_number ?? 0,
+        ...(input.status !== undefined ? { status: input.status } : {})
+      }),
+      encode: (value) => value
+    })
+  )
+)
+
 const knownResponseStreamEventTypes = new Set([
   "response.created",
   "response.completed",

packages/ai/openai/test/OpenAiSchema.test.ts

@@ -283,6 +283,45 @@ describe("OpenAiSchema", () => {
       assert.isDefined(malformed)
     }))
 
+  it("decodes the error event whether the payload is flat or nested under `error`", () => {
+    // The documented shape carries `code`/`message`/`param` at the top level.
+    const flat = Schema.decodeUnknownSync(OpenAiSchema.ResponseStreamEvent)({
+      type: "error",
+      code: "ERR",
+      message: "boom",
+      param: null,
+      sequence_number: 1
+    })
+    assert.deepStrictEqual(flat, {
+      type: "error",
+      code: "ERR",
+      message: "boom",
+      param: null,
+      sequence_number: 1
+    })
+
+    // Mid-stream errors (e.g. quota exhaustion) instead nest the standard error
+    // envelope under `error`. This must decode too — not abort the stream — and
+    // normalize to the documented shape so the message and code still surface.
+    const nested = Schema.decodeUnknownSync(OpenAiSchema.ResponseStreamEvent)({
+      type: "error",
+      error: {
+        type: "insufficient_quota",
+        code: "credit_balance_exhausted",
+        message: "You have no credits remaining.",
+        param: null
+      },
+      sequence_number: 2
+    })
+    assert.deepStrictEqual(nested, {
+      type: "error",
+      code: "credit_balance_exhausted",
+      message: "You have no credits remaining.",
+      param: null,
+      sequence_number: 2
+    })
+  })
+
   it("decodes embedding response variants (numeric + string/base64)", () => {
     const numeric = Schema.decodeUnknownSync(OpenAiSchema.CreateEmbeddingResponse)({
       object: "list",