Function streamRequest

packages/ai/openai/src/OpenAiClient.ts:125–164

(
      request: HttpClientRequest.HttpClientRequest,
      schema: Schema.Schema<A, I, R>
    )

    })

    const streamRequest = <A, I, R>(
      request: HttpClientRequest.HttpClientRequest,
      schema: Schema.Schema<A, I, R>
    ): Stream.Stream<A, AiError.AiError, R> => {
      const decodeEvent = Schema.decode(Schema.parseJson(schema))
      return httpClientOk.execute(request).pipe(
        Effect.map((r) => r.stream),
        Stream.unwrapScoped,
        Stream.decodeText(),
        Stream.pipeThroughChannel(Sse.makeChannel()),
        // Decode each SSE event, but don't let a single unrecognized or malformed
        // frame abort the whole stream. OpenAI emits events absent from the generated
        // schema (e.g. `keepalive` heartbeats during long Responses turns); skip what
        // we can't decode and keep the stream alive.
        Stream.mapEffect((event) =>
          decodeEvent(event.data).pipe(
            Effect.asSome,
            Effect.catchTag("ParseError", (error) =>
              Effect.logDebug("Skipping undecodable stream event", error).pipe(
                Effect.as(Option.none())
              ))
          )
        ),
        Stream.filterMap(identity),
        Stream.catchTags({
          RequestError: (error) =>
            AiError.HttpRequestError.fromRequestError({
              module: "OpenAiClient",
              method: "streamRequest",
              error
            }),
          ResponseError: (error) =>
            AiError.HttpResponseError.fromResponseError({
              module: "OpenAiClient",
              method: "streamRequest",
              error
            })
        })
      )
    }

    const createResponse = (
      options: typeof Generated.CreateResponse.Encoded