1import { Generated, OpenAiClient, OpenAiLanguageModel } from "@effect/ai-openai"
2import * as OpenAiSchema from "@effect/ai-openai/OpenAiSchema"
3import { assert, describe, it } from "@effect/vitest"
4import { Effect, Layer, Schema, Stream } from "effect"
5import { LanguageModel } from "effect/ai"
6import * as Sse from "effect/encoding/Sse"
7import { HttpClientRequest, HttpClientResponse } from "effect/http"
8
9const makeResponse = (overrides: Record<string, unknown> = {}) => ({
10 id: "resp_123",
11 object: "response",
12 model: "gpt-4o-mini",
13 status: "completed",
14 created_at: 1,
15 output: [],
16 ...overrides
17})
18
19describe("OpenAiSchema", () => {
20 describe("web search sources", () => {
21 it("preserves API sources in response history", () => {
22 const item = {
23 type: "web_search_call",
24 id: "ws_123",
25 status: "completed",
26 action: { type: "search", sources: [{ type: "api", name: "oai-weather" }] }
27 } as const
28
29 assert.deepStrictEqual(Schema.decodeUnknownSync(Generated.InputItem)(item), item)
30 })
31
32 it.each([
33 { type: "url" },
34 { type: "api" },
35 { type: "api", name: null },
36 { type: "unknown", name: "oai-weather" }
37 ])("rejects malformed search sources: %j", (source) => {
38 const item = {
39 type: "web_search_call",
40 id: "ws_123",
41 status: "completed",
42 action: { type: "search", sources: [source] }
43 }
44
45 assert.throws(() => Schema.decodeUnknownSync(Generated.WebSearchToolCall)(item))
46 assert.throws(() => Schema.decodeUnknownSync(Generated.InputItem)(item))
47 })
48 })
49
50 it("accepts max reasoning effort", () => {
51 const decoded = Schema.decodeUnknownSync(OpenAiSchema.CreateResponse)({
52 reasoning: { effort: "max" }
53 })
54
55 assert.strictEqual(decoded.reasoning?.effort, "max")
56 })
57
58 it("decodes a representative response payload", () => {
59 const decoded = Schema.decodeUnknownSync(OpenAiSchema.Response)({
60 ...makeResponse(),
61 output: [
62 {
63 id: "msg_1",
64 type: "message",
65 role: "assistant",
66 status: "completed",
67 content: [
68 {
69 type: "output_text",
70 text: "hello",
71 annotations: [
72 {
73 type: "url_citation",
74 url: "https://example.com",
75 start_index: 0,
76 end_index: 5,
77 title: "example"
78 }
79 ]
80 },
81 {
82 type: "refusal",
83 refusal: "cannot comply"
84 }
85 ]
86 },
87 {
88 id: "fc_1",
89 type: "function_call",
90 call_id: "call_1",
91 name: "lookup",
92 arguments: "{}",
93 status: "completed"
94 },
95 {
96 id: "reasoning_1",
97 type: "reasoning",
98 summary: [
99 {
100 type: "summary_text",
101 text: "thinking"
102 }
103 ]
104 }
105 ],
106 usage: {
107 input_tokens: 10,
108 output_tokens: 20,
109 total_tokens: 30
110 }
111 })
112
113 assert.strictEqual(decoded.id, "resp_123")
114 assert.strictEqual(decoded.output.length, 3)
115 assert.strictEqual(decoded.output[0].type, "message")
116 if (decoded.output[0].type === "message") {
117 assert.strictEqual(decoded.output[0].content[0].type, "output_text")
118 }
119 })
120
121 it("decodes image generation lifecycle statuses", () => {
122 for (const status of ["generating", "failed"]) {
123 const decoded = Schema.decodeUnknownSync(OpenAiSchema.Response)({
124 ...makeResponse(),
125 output: [{
126 id: "image_1",
127 type: "image_generation_call",
128 status,
129 result: null
130 }]
131 })
132
133 assert.strictEqual(decoded.output[0].type, "image_generation_call")
134 if (decoded.output[0].type === "image_generation_call") {
135 assert.strictEqual(decoded.output[0].status, status)
136 }
137 }
138 })
139
140 it("decodes required stream events", () => {
141 const response = makeResponse({ status: "in_progress" })
142 const applyPatchItem = {
143 id: "ap_1",
144 type: "apply_patch_call",
145 call_id: "call_ap",
146 operation: {
147 type: "update_file",
148 path: "README.md",
149 diff: "@@"
150 }
151 }
152
153 const events = [
154 { type: "response.created", sequence_number: 1, response },
155 {
156 type: "response.completed",
157 sequence_number: 2,
158 response: makeResponse()
159 },
160 { type: "response.incomplete", sequence_number: 3, response: makeResponse({ status: "incomplete" }) },
161 { type: "response.failed", sequence_number: 4, response: makeResponse({ status: "failed" }) },
162 { type: "response.output_item.added", sequence_number: 5, output_index: 0, item: applyPatchItem },
163 { type: "response.output_item.done", sequence_number: 6, output_index: 0, item: applyPatchItem },
164 {
165 type: "response.output_text.delta",
166 sequence_number: 7,
167 item_id: "msg_1",
168 output_index: 0,
169 content_index: 0,
170 delta: "hel"
171 },
172 {
173 type: "response.output_text.annotation.added",
174 sequence_number: 8,
175 item_id: "msg_1",
176 output_index: 0,
177 content_index: 0,
178 annotation_index: 0,
179 annotation: {
180 type: "file_path",
181 file_id: "file_1",
182 index: 0
183 }
184 },
185 {
186 type: "response.reasoning_summary_part.added",
187 sequence_number: 9,
188 item_id: "reasoning_1",
189 output_index: 1,
190 summary_index: 0,
191 part: { type: "summary_text", text: "thinking" }
192 },
193 {
194 type: "response.reasoning_summary_part.done",
195 sequence_number: 10,
196 item_id: "reasoning_1",
197 output_index: 1,
198 summary_index: 0,
199 part: { type: "summary_text", text: "thinking" }
200 },
201 {
202 type: "response.reasoning_summary_text.delta",
203 sequence_number: 11,
204 item_id: "reasoning_1",
205 output_index: 1,
206 summary_index: 0,
207 delta: "..."
208 },
209 {
210 type: "response.function_call_arguments.delta",
211 sequence_number: 12,
212 item_id: "fc_1",
213 output_index: 2,
214 delta: "{"
215 },
216 {
217 type: "response.function_call_arguments.done",
218 sequence_number: 13,
219 item_id: "fc_1",
220 output_index: 2,
221 arguments: "{}"
222 },
223 {
224 type: "response.code_interpreter_call_code.delta",
225 sequence_number: 14,
226 item_id: "code_1",
227 output_index: 3,
228 delta: "print"
229 },
230 {
231 type: "response.code_interpreter_call_code.done",
232 sequence_number: 15,
233 item_id: "code_1",
234 output_index: 3,
235 code: "print('ok')"
236 },
237 {
238 type: "response.apply_patch_call_operation_diff.delta",
239 sequence_number: 16,
240 item_id: "ap_1",
241 output_index: 4,
242 delta: "@@"
243 },
244 {
245 type: "response.apply_patch_call_operation_diff.done",
246 sequence_number: 17,
247 item_id: "ap_1",
248 output_index: 4,
249 delta: "@@"
250 },
251 {
252 type: "response.image_generation_call.partial_image",
253 sequence_number: 18,
254 item_id: "img_1",
255 output_index: 5,
256 partial_image_b64: "AQID"
257 }
258 ]
259
260 for (const event of events) {
261 const decoded = Schema.decodeUnknownSync(OpenAiSchema.ResponseStreamEvent)(event)
262 assert.strictEqual(decoded.type, event.type)
263 }
264 })
265
266 it.effect("keeps keepalive and unknown events tolerant in SSE decoding", () =>
267 Effect.gen(function*() {
268 const sseBody = [
269 {
270 type: "response.created",
271 sequence_number: 1,
272 response: makeResponse({ status: "in_progress" })
273 },
274 {
275 type: "keepalive",
276 sequence_number: 2,
277 heartbeat: true
278 },
279 {
280 type: "provider.future_event",
281 sequence_number: 3,
282 nested: { ok: true }
283 },
284 {
285 type: "response.completed",
286 sequence_number: 4,
287 response: makeResponse({ status: "completed" })
288 }
289 ].map((event) => `data: ${JSON.stringify(event)}\n\n`).join("")
290
291 const events = yield* Stream.fromIterable([sseBody]).pipe(
292 Stream.pipeThroughChannel(Sse.decodeDataSchema(OpenAiSchema.ResponseStreamEvent)),
293 Stream.map((event) => event.data),
294 Stream.runCollect
295 )
296
297 const decoded = globalThis.Array.from(events)
298 assert.strictEqual(decoded.length, 4)
299 assert.strictEqual(decoded[1].type, "keepalive")
300 assert.strictEqual(decoded[2].type, "provider.future_event")
301 if (decoded[1].type === "keepalive") {
302 assert.strictEqual(decoded[1].sequence_number, 2)
303 }
304 }))
305
306 it.effect("does not silently decode malformed known events as unknown", () =>
307 Effect.gen(function*() {
308 const malformed = yield* Schema.decodeUnknownEffect(OpenAiSchema.ResponseStreamEvent)({
309 type: "response.completed",
310 sequence_number: 1
311 }).pipe(Effect.flip)
312
313 assert.isDefined(malformed)
314 }))
315
316 it("decodes the error event whether the payload is flat or nested under `error`", () => {
317 const flat = Schema.decodeUnknownSync(OpenAiSchema.ResponseStreamEvent)({
318 type: "error",
319 code: "ERR",
320 message: "boom",
321 param: null,
322 sequence_number: 1
323 })
324 assert.deepStrictEqual(flat, {
325 type: "error",
326 code: "ERR",
327 message: "boom",
328 param: null,
329 sequence_number: 1
330 })
331
332 const nested = Schema.decodeUnknownSync(OpenAiSchema.ResponseStreamEvent)({
333 type: "error",
334 error: {
335 type: "insufficient_quota",
336 code: "credit_balance_exhausted",
337 message: "You have no credits remaining.",
338 param: null
339 },
340 sequence_number: 2
341 })
342 assert.deepStrictEqual(nested, {
343 type: "error",
344 code: "credit_balance_exhausted",
345 message: "You have no credits remaining.",
346 param: null,
347 sequence_number: 2
348 })
349
350 const nestedWithStatus = Schema.decodeUnknownSync(OpenAiSchema.ResponseStreamEvent)({
351 type: "error",
352 error: {
353 code: "rate_limited",
354 message: "Too many requests.",
355 param: null
356 },
357 sequence_number: 3,
358 status: 429
359 })
360 assert.deepStrictEqual(nestedWithStatus, {
361 type: "error",
362 code: "rate_limited",
363 message: "Too many requests.",
364 param: null,
365 sequence_number: 3,
366 status: 429
367 })
368 })
369
370 it.effect("rejects error events missing spec-required fields", () =>
371 Effect.gen(function*() {
372 const malformed = [
373 { type: "error" },
374 { type: "error", error: {}, sequence_number: 3 },
375 { type: "error", error: { code: "x" }, sequence_number: 3 }
376 ]
377 for (const event of malformed) {
378 const failure = yield* Schema.decodeUnknownEffect(OpenAiSchema.ResponseStreamEvent)(event).pipe(Effect.flip)
379 assert.isDefined(failure)
380 }
381 }))
382
383 it.effect("surfaces nested error events in SSE decoding instead of aborting the stream", () =>
384 Effect.gen(function*() {
385 const sseBody = [
386 {
387 type: "response.created",
388 sequence_number: 1,
389 response: makeResponse({ status: "in_progress" })
390 },
391 {
392 type: "error",
393 error: {
394 type: "insufficient_quota",
395 code: "credit_balance_exhausted",
396 message: "You have no credits remaining.",
397 param: null
398 },
399 sequence_number: 2
400 }
401 ].map((event) => `data: ${JSON.stringify(event)}\n\n`).join("")
402
403 const events = yield* Stream.fromIterable([sseBody]).pipe(
404 Stream.pipeThroughChannel(Sse.decodeDataSchema(OpenAiSchema.ResponseStreamEvent)),
405 Stream.map((event) => event.data),
406 Stream.runCollect
407 )
408
409 const decoded = globalThis.Array.from(events)
410 assert.strictEqual(decoded.length, 2)
411 assert.deepStrictEqual(decoded[1], {
412 type: "error",
413 code: "credit_balance_exhausted",
414 message: "You have no credits remaining.",
415 param: null,
416 sequence_number: 2
417 })
418 }))
419
420 it("decodes embedding response variants (numeric + string/base64)", () => {
421 const numeric = Schema.decodeUnknownSync(OpenAiSchema.CreateEmbeddingResponse)({
422 object: "list",
423 model: "text-embedding-3-small",
424 data: [{ object: "embedding", index: 0, embedding: [0.1, 0.2] }],
425 usage: { prompt_tokens: 2, total_tokens: 2 }
426 })
427 const base64 = Schema.decodeUnknownSync(OpenAiSchema.CreateEmbeddingResponse)({
428 object: "list",
429 model: "text-embedding-3-small",
430 data: [{ object: "embedding", index: 0, embedding: "AQID" }],
431 usage: { prompt_tokens: 2, total_tokens: 2 }
432 })
433
434 assert.strictEqual(numeric.data[0].embedding[0], 0.1)
435 assert.strictEqual(base64.data[0].embedding, "AQID")
436 })
437
438 it.effect("exposes the response.failed error payload", () => {
439 const response = HttpClientResponse.fromWeb(
440 HttpClientRequest.get("https://api.openai.com/v1/responses"),
441 new Response()
442 )
443 const failed = Schema.decodeUnknownSync(OpenAiSchema.ResponseStreamEvent)({
444 type: "response.failed",
445 sequence_number: 1,
446 response: {
447 id: "resp_1",
448 object: "response",
449 created_at: 1,
450 model: "gpt-4o-mini",
451 status: "failed",
452 output: [],
453 error: { code: "server_error", message: "provider exploded" }
454 }
455 })
456 const client = Layer.succeed(
457 OpenAiClient.OpenAiClient,
458 OpenAiClient.OpenAiClient.of({
459 client: undefined as any,
460 createResponse: () => Effect.die("unexpected"),
461 createResponseStream: () =>
462 Effect.succeed([
463 response,
464 Stream.make(failed)
465 ]),
466 createEmbedding: () => Effect.die("unexpected")
467 })
468 )
469
470 return LanguageModel.streamText({ prompt: "test" }).pipe(
471 Stream.runCollect,
472 Effect.tap((parts) =>
473 Effect.sync(() => {
474 const error = Array.from(parts).find((part) => part.type === "error")
475 const finish = Array.from(parts).find((part) => part.type === "finish")
476 assert.isDefined(error)
477 assert.deepStrictEqual(error.error, { code: "server_error", message: "provider exploded" })
478 assert.isDefined(finish)
479 assert.strictEqual(finish.reason, "error")
480 })
481 ),
482 Effect.provide(OpenAiLanguageModel.model("gpt-4o-mini")),
483 Effect.provide(client)
484 )
485 })
486})