Unified · Split
| @@ -1,4 +1,4 @@ | |||
| 1 | import { Effect, Layer, Option, Stream } from "effect"; | 1 | import { Effect, Exit, Layer, Option, Schema, Stream } from "effect"; |
| 2 | import { HttpClient, HttpClientRequest } from "effect/unstable/http"; | 2 | import { HttpClient, HttpClientRequest } from "effect/unstable/http"; |
| 3 | import type { ToolFileValue } from "@executor-js/sdk/core"; | 3 | import type { ToolFileValue } from "@executor-js/sdk/core"; |
| 4 | 4 | ||
| @@ -201,6 +201,8 @@ const isTextContentType = (ct: string | null | undefined): boolean => · 196 unmodified lines | |||
| 201 | const isOctetStream = (ct: string | null | undefined): boolean => | 201 | const isOctetStream = (ct: string | null | undefined): boolean => |
| 202 | normalizeContentType(ct) === "application/octet-stream"; | 202 | normalizeContentType(ct) === "application/octet-stream"; |
| 203 | 203 | ||
| 204 | const decodeJsonLine = Schema.decodeUnknownExit(Schema.fromJsonString(Schema.Unknown)); | ||
| 205 | |||
| 204 | const parseStreamingJsonLines = (text: string, truncated: boolean): unknown => { | 206 | const parseStreamingJsonLines = (text: string, truncated: boolean): unknown => { |
| 205 | const lines = text.split(/\r?\n/); | 207 | const lines = text.split(/\r?\n/); |
| 206 | let lastNonEmptyIndex = -1; | 208 | let lastNonEmptyIndex = -1; |
| @@ -216,32 +218,34 @@ const parseStreamingJsonLines = (text: string, truncated: boolean): unknown => { · 9 unmodified lines | |||
| 216 | for (let index = 0; index < lines.length; index++) { | 218 | for (let index = 0; index < lines.length; index++) { |
| 217 | const line = lines[index]; | 219 | const line = lines[index]; |
| 218 | if (line === undefined || line.trim() === "") continue; | 220 | if (line === undefined || line.trim() === "") continue; |
| 219 | try { | 221 | const parsed = decodeJsonLine(line); |
| 220 | rows.push(JSON.parse(line)); | 222 | if (Exit.isSuccess(parsed)) { |
| 221 | } catch { | 223 | rows.push(parsed.value); |
| 222 | if (truncated && index === lastNonEmptyIndex) continue; | 224 | continue; |
| 223 | return text; | ||
| 224 | } | 225 | } |
| 226 | // A trailing fragment cut mid-line by the byte/time cap is expected; any | ||
| 227 | // other unparseable line means the body is not NDJSON after all. | ||
| 228 | if (truncated && index === lastNonEmptyIndex) continue; | ||
| 229 | return text; | ||
| 225 | } | 230 | } |
| 226 | 231 | ||
| 227 | return rows; | 232 | return rows; |
| 228 | }; | 233 | }; |
| 229 | 234 | ||
| 235 | const NDJSON_CONTENT_TYPES = new Set([ | ||
| 236 | "application/stream+json", | ||
| 237 | "application/x-ndjson", | ||
| 238 | "application/jsonl", | ||
| 239 | ]); | ||
| 240 | |||
| 230 | const decodeStreamingResponseBody = ( | 241 | const decodeStreamingResponseBody = ( |
| 231 | contentType: string | null, | 242 | contentType: string | null, |
| 232 | text: string, | 243 | text: string, |
| 233 | truncated: boolean, | 244 | truncated: boolean, |
| 234 | ): unknown => { | 245 | ): unknown => |
| 235 | switch (normalizeContentType(contentType)) { | 246 | NDJSON_CONTENT_TYPES.has(normalizeContentType(contentType)) |
| 236 | case "application/stream+json": | 247 | ? parseStreamingJsonLines(text, truncated) |
| 237 | case "application/x-ndjson": | 248 | : text; |
| 238 | case "application/jsonl": | ||
| 239 | return parseStreamingJsonLines(text, truncated); | ||
| 240 | case "text/event-stream": | ||
| 241 | default: | ||
| 242 | return text; | ||
| 243 | } | ||
| 244 | }; | ||
| 245 | 249 | ||
| 246 | export const collectStreamingBody = ( | 250 | export const collectStreamingBody = ( |
| 247 | stream: Stream.Stream<Uint8Array, unknown, never>, | 251 | stream: Stream.Stream<Uint8Array, unknown, never>, |