import { Effect, Exit, Fiber, Layer, Option, Schema, Stream } from "effect";import { HttpClient, HttpClientRequest, HttpClientResponse } from "effect/unstable/http";import { endpointForTelemetry, type ToolFileValue } from "@executor-js/sdk/core"; import { OpenApiInvocationError } from "./errors";import { isNdjsonMediaType, NDJSON_MEDIA_TYPES, resolveServerUrl } from "./openapi-utils";import { type EncodingObject, type OperationFileHint, type OperationBinding, InvocationResult, type MediaBinding, type OperationParameter, type ServerInfo,} from "./types"; // ---------------------------------------------------------------------------// Parameter reading// --------------------------------------------------------------------------- const CONTAINER_KEYS: Record<string, readonly string[]> = { path: ["path", "pathParams", "params"], query: ["query", "queryParams", "params"], header: ["headers", "header"], cookie: ["cookies", "cookie"],}; const readParamValue = (args: Record<string, unknown>, param: OperationParameter): unknown => { const direct = args[param.name]; if (direct !== undefined) return direct; for (const key of CONTAINER_KEYS[param.location] ?? []) { const container = args[key]; if (typeof container === "object" && container !== null && !Array.isArray(container)) { const nested = (container as Record<string, unknown>)[param.name]; if (nested !== undefined) return nested; } } return undefined;}; const primitiveToString = (value: unknown): string => typeof value === "object" && value !== null ? JSON.stringify(value) : String(value); // RFC 3986 §2.2 reserved chars. `allowReserved: true` leaves these// unencoded; default OAS behavior encodes everything non-unreserved.const RESERVED_UNENCODED_RE = /[A-Za-z0-9\-._~:/?#[\]@!$&'()*+,;=]/; const encodeReservedAware = (raw: string, allowReserved: boolean): string => { if (!allowReserved) return encodeURIComponent(raw); // Walk char-by-char so the reserved set passes through as-is. let out = ""; for (const ch of raw) { out += RESERVED_UNENCODED_RE.test(ch) ? ch : encodeURIComponent(ch); } return out;}; const queryParamValues = (value: unknown, param: OperationParameter): string[] => { if (value === undefined || value === null) return []; if (!Array.isArray(value)) return [primitiveToString(value)]; const style = Option.getOrUndefined(param.style) ?? "form"; const explode = Option.getOrElse(param.explode, () => true); if (explode) return value.map(primitiveToString); const separator = style === "spaceDelimited" ? " " : style === "pipeDelimited" ? "|" : ","; return [value.map(primitiveToString).join(separator)];}; // ---------------------------------------------------------------------------// Path resolution// --------------------------------------------------------------------------- const resolvePath = Effect.fn("OpenApi.resolvePath")(function* ( pathTemplate: string, args: Record<string, unknown>, parameters: readonly OperationParameter[],) { let resolved = pathTemplate; for (const param of parameters) { if (param.location !== "path") continue; const value = readParamValue(args, param); if (value === undefined || value === null) { if (param.required) { return yield* new OpenApiInvocationError({ message: `Missing required path parameter: ${param.name}`, statusCode: Option.none(), }); } continue; } const encoded = encodeReservedAware( String(value), Option.getOrElse(param.allowReserved, () => false), ); resolved = resolved.replaceAll(`{${param.name}}`, encoded); resolved = resolved.replaceAll(`{+${param.name}}`, encoded); } const remaining = [...resolved.matchAll(/\{([^{}]+)\}/g)] .map((m) => m[1]) .filter((v): v is string => typeof v === "string"); for (const name of remaining) { const value = args[name]; if (value !== undefined && value !== null) { resolved = resolved.replaceAll(`{${name}}`, encodeURIComponent(String(value))); } } const unresolved = [...resolved.matchAll(/\{([^{}]+)\}/g)] .map((m) => m[1]) .filter((v): v is string => typeof v === "string"); if (unresolved.length > 0) { return yield* new OpenApiInvocationError({ message: `Unresolved path parameters: ${[...new Set(unresolved)].join(", ")}`, statusCode: Option.none(), }); } return resolved;}); // GitHub (and some other upstreams) reject requests that lack a User-Agent// header with a 403 ("Request forbidden by administrative rules"), which is// indistinguishable from a credential rejection downstream. Send a default so// those calls succeed. It is applied before operation header params and// resolved auth headers, so a spec- or connection-provided User-Agent wins.const DEFAULT_USER_AGENT = "executor"; const applyHeaders = ( request: HttpClientRequest.HttpClientRequest, headers: Record<string, string>,): HttpClientRequest.HttpClientRequest => { let req = request; for (const [name, value] of Object.entries(headers)) { req = HttpClientRequest.setHeader(req, name, value); } return req;}; // ---------------------------------------------------------------------------// Response helpers// --------------------------------------------------------------------------- const normalizeContentType = (ct: string | null | undefined): string => ct?.split(";")[0]?.trim().toLowerCase() ?? ""; export const STREAM_MAX_BYTES = 1_000_000;export const STREAM_MAX_MS = 10_000;// Buffered bodies should normally finish soon after headers arrive. This is// deliberately generous while still providing plain Node runtimes a backstop.export const RESPONSE_BODY_TIMEOUT_MS = 60_000;// Below Cloudflare's ~125s subrequest limit so cloud fails first with an// actionable error, but above the slowest legitimate buffered upstreams// observed in production (30d telemetry: successful calls cluster under// 110s; a 60s cap would have broken ~40 real calls).export const RESPONSE_HEADERS_TIMEOUT_MS = 110_000; class StreamCapReached extends Error { readonly _tag = "StreamCapReached";} export interface StreamingResponseCaps { readonly maxBytes?: number; readonly maxMs?: number;} export interface InvokeOptions { readonly responseHeadersTimeoutMs?: number; readonly responseBodyTimeoutMs?: number;} const formatTimeout = (timeoutMs: number): string => timeoutMs % 1_000 === 0 ? `${timeoutMs / 1_000}s` : `${timeoutMs}ms`; const responseHeadersTimeoutMessage = (timeoutMs: number): string => `Upstream returned no response headers within ${formatTimeout(timeoutMs)}. The endpoint may be a live stream with no data to send (for example, runtime logs of an idle deployment); the request was aborted. Retry when the resource has activity, or use a non-streaming endpoint.`; const responseBodyTimeoutMessage = (timeoutMs: number): string => `Upstream response body did not finish within ${formatTimeout(timeoutMs)}. The request was aborted. Retry the operation; if it times out again, check the upstream service health or use an endpoint with a bounded response.`; const STREAMING_RESPONSE_CONTENT_TYPES = new Set([...NDJSON_MEDIA_TYPES, "text/event-stream"]); const isStreamingResponseContentType = (ct: string | null | undefined): boolean => STREAMING_RESPONSE_CONTENT_TYPES.has(normalizeContentType(ct)); const isJsonContentType = (ct: string | null | undefined): boolean => { const normalized = normalizeContentType(ct); if (!normalized) return false; return ( normalized === "application/json" || normalized.includes("+json") || normalized.includes("json") );}; const isFormUrlEncoded = (ct: string | null | undefined): boolean => normalizeContentType(ct) === "application/x-www-form-urlencoded"; const isMultipartFormData = (ct: string | null | undefined): boolean => normalizeContentType(ct).startsWith("multipart/form-data"); const isXmlContentType = (ct: string | null | undefined): boolean => { const normalized = normalizeContentType(ct); if (!normalized) return false; return ( normalized === "application/xml" || normalized === "text/xml" || normalized.endsWith("+xml") );}; const isTextContentType = (ct: string | null | undefined): boolean => normalizeContentType(ct).startsWith("text/"); const isOctetStream = (ct: string | null | undefined): boolean => normalizeContentType(ct) === "application/octet-stream"; const decodeJsonLine = Schema.decodeUnknownExit(Schema.fromJsonString(Schema.Unknown)); const parseStreamingJsonLines = (text: string, truncated: boolean): unknown => { const lines = text.split(/\r?\n/); let lastNonEmptyIndex = -1; for (let index = lines.length - 1; index >= 0; index--) { const line = lines[index]; if (line !== undefined && line.trim() !== "") { lastNonEmptyIndex = index; break; } } const rows: unknown[] = []; for (let index = 0; index < lines.length; index++) { const line = lines[index]; if (line === undefined || line.trim() === "") continue; const parsed = decodeJsonLine(line); if (Exit.isSuccess(parsed)) { rows.push(parsed.value); continue; } // A trailing fragment cut mid-line by the byte/time cap is expected; any // other unparseable line means the body is not NDJSON after all. if (truncated && index === lastNonEmptyIndex) continue; return text; } return rows;}; const decodeStreamingResponseBody = ( contentType: string | null, text: string, truncated: boolean,): unknown => (isNdjsonMediaType(contentType) ? parseStreamingJsonLines(text, truncated) : text); export const collectStreamingBody = ( stream: Stream.Stream<Uint8Array, unknown, never>, contentType: string | null, caps: StreamingResponseCaps = {},) => Effect.gen(function* () { const maxBytes = caps.maxBytes ?? STREAM_MAX_BYTES; const maxMs = caps.maxMs ?? STREAM_MAX_MS; const chunks: Uint8Array[] = []; let bytes = 0; let truncated = false; const startedAt = Date.now(); const runFork = Effect.runForkWith(yield* Effect.context<never>()); const completedExitOption = yield* Effect.callback<Option.Option<Exit.Exit<void, unknown>>>( (resume, signal) => { let settled = false; const collectEffect = stream.pipe( Stream.timeout(maxMs), Stream.runForEach((chunk) => Effect.gen(function* () { const remaining = maxBytes - bytes; if (remaining <= 0) { truncated = true; return yield* Effect.fail(new StreamCapReached()); } if (chunk.byteLength > remaining) { chunks.push(chunk.subarray(0, remaining)); bytes += remaining; truncated = true; return yield* Effect.fail(new StreamCapReached()); } chunks.push(chunk); bytes += chunk.byteLength; if (bytes >= maxBytes) { truncated = true; return yield* Effect.fail(new StreamCapReached()); } }), ), Effect.catch((error: unknown) => error instanceof StreamCapReached ? Effect.void : Effect.fail(error), ), Effect.exit, Effect.tap((exit) => Effect.sync(() => { if (settled) return; settled = true; clearTimeout(timer); resume(Effect.succeed(Option.some(exit))); }), ), ); const fiber = runFork(collectEffect); const interrupt = () => { runFork(Fiber.interrupt(fiber)); }; const timer = setTimeout(() => { if (settled) return; settled = true; truncated = true; interrupt(); resume(Effect.succeed(Option.none())); }, maxMs + 1_000); signal.addEventListener("abort", interrupt, { once: true }); return Effect.sync(() => { clearTimeout(timer); signal.removeEventListener("abort", interrupt); if (!settled) interrupt(); }); }, ); if (Option.isSome(completedExitOption) && Exit.isFailure(completedExitOption.value)) { return yield* Effect.failCause(completedExitOption.value.cause); } if (Option.isNone(completedExitOption)) truncated = true; const durationMs = Date.now() - startedAt; if (durationMs >= maxMs) truncated = true; const collected = new Uint8Array(bytes); let offset = 0; for (const chunk of chunks) { collected.set(chunk, offset); offset += chunk.byteLength; } const text = new TextDecoder("utf-8").decode(collected); return { data: decodeStreamingResponseBody(contentType, text, truncated), headers: { "x-executor-stream": truncated ? "truncated" : "complete", "x-executor-stream-bytes": String(bytes), "x-executor-stream-duration-ms": String(durationMs), }, }; }); const bytesToBase64 = (bytes: Uint8Array): string => { let binary = ""; const chunkSize = 0x8000; for (let i = 0; i < bytes.length; i += chunkSize) { binary += String.fromCharCode(...bytes.subarray(i, i + chunkSize)); } return btoa(binary);}; const normalizeBase64 = (value: string, encoding: "base64" | "base64url"): string => { const compact = value.replace(/\s/g, ""); const alphabet = encoding === "base64url" ? compact.replace(/-/g, "+").replace(/_/g, "/") : compact; const remainder = alphabet.length % 4; return remainder === 0 ? alphabet : `${alphabet}${"=".repeat(4 - remainder)}`;}; const byteLengthFromBase64 = (base64: string): number => { const compact = base64.replace(/\s/g, ""); const padding = compact.endsWith("==") ? 2 : compact.endsWith("=") ? 1 : 0; return Math.max(0, Math.floor((compact.length * 3) / 4) - padding);}; const isGenericMimeType = (mimeType: string): boolean => normalizeContentType(mimeType) === "application/octet-stream"; const startsWithBytes = (bytes: Uint8Array, prefix: readonly number[]): boolean => prefix.every((byte, index) => bytes[index] === byte); const isLikelyUtf8Text = (bytes: Uint8Array): boolean => { if (bytes.length === 0) return false; let text: string; // oxlint-disable-next-line executor/no-try-catch-or-throw -- boundary: TextDecoder throws while probing arbitrary binary content try { text = new TextDecoder("utf-8", { fatal: true }).decode(bytes); } catch { return false; } let suspicious = 0; for (let index = 0; index < text.length; index += 1) { const code = text.charCodeAt(index); const allowedControl = code === 0x09 || code === 0x0a || code === 0x0c || code === 0x0d; if (code === 0x00) return false; if (code < 0x20 && !allowedControl) suspicious += 1; } return suspicious / Math.max(1, text.length) <= 0.02;}; const sniffMimeType = (bytes: Uint8Array): string | null => { if (startsWithBytes(bytes, [0xff, 0xd8, 0xff])) return "image/jpeg"; if (startsWithBytes(bytes, [0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a])) { return "image/png"; } if ( startsWithBytes(bytes, [0x47, 0x49, 0x46, 0x38, 0x37, 0x61]) || startsWithBytes(bytes, [0x47, 0x49, 0x46, 0x38, 0x39, 0x61]) ) { return "image/gif"; } if ( startsWithBytes(bytes, [0x52, 0x49, 0x46, 0x46]) && bytes[8] === 0x57 && bytes[9] === 0x45 && bytes[10] === 0x42 && bytes[11] === 0x50 ) { return "image/webp"; } if (startsWithBytes(bytes, [0x25, 0x50, 0x44, 0x46, 0x2d])) return "application/pdf"; if ( startsWithBytes(bytes, [0x50, 0x4b, 0x03, 0x04]) || startsWithBytes(bytes, [0x50, 0x4b, 0x05, 0x06]) || startsWithBytes(bytes, [0x50, 0x4b, 0x07, 0x08]) ) { return "application/zip"; } if (isLikelyUtf8Text(bytes)) return "text/plain"; return null;}; const bytesFromBase64Prefix = (base64: string): Uint8Array => { const prefix = base64.slice(0, Math.min(base64.length, 64)); const binary = atob(prefix); const bytes = new Uint8Array(binary.length); for (let index = 0; index < binary.length; index += 1) { bytes[index] = binary.charCodeAt(index); } return bytes;}; const sniffMimeTypeFromBase64 = (base64: string): string | null => sniffMimeType(bytesFromBase64Prefix(base64)); type DecodedBase64Body = { readonly ok: true; readonly bytes: Uint8Array } | { readonly ok: false }; const base64ToUint8Array = (value: string): Uint8Array | null => { let binary = ""; // oxlint-disable-next-line executor/no-try-catch-or-throw -- boundary: atob throws for invalid base64; invalid shapes are treated as non-byte input try { binary = atob(normalizeBase64(value, "base64")); } catch { return null; } const bytes = new Uint8Array(binary.length); for (let index = 0; index < binary.length; index += 1) { bytes[index] = binary.charCodeAt(index); } return bytes;}; const decodeBase64Body = (value: string): DecodedBase64Body => { const bytes = base64ToUint8Array(value); return bytes ? { ok: true, bytes } : { ok: false };}; const toUint8Array = (value: unknown): Uint8Array | null => { if (value instanceof Uint8Array) return value; if (value instanceof ArrayBuffer) return new Uint8Array(value); if (ArrayBuffer.isView(value)) { const view = value as ArrayBufferView; return new Uint8Array(view.buffer, view.byteOffset, view.byteLength); } if (Array.isArray(value) && value.every((v) => typeof v === "number")) { return new Uint8Array(value as readonly number[]); } return null;}; const readNestedBodyBase64 = (value: unknown): unknown => typeof value === "object" && value !== null && !Array.isArray(value) && Object.prototype.hasOwnProperty.call(value, "bodyBase64") ? (value as Record<string, unknown>).bodyBase64 : undefined; const readHintString = (option: OperationFileHint["dataField"], fallback: string): string => Option.getOrElse(option, () => fallback); const readHintMimeType = (hint: OperationFileHint, fallback: string): string => Option.getOrElse(hint.mimeType, () => fallback); const readHintEncoding = (hint: OperationFileHint): "base64" | "base64url" => Option.getOrElse(hint.encoding, () => "base64");Lines 1–500 of 1321. Load more
import { Effect, Exit, Fiber, Layer, Option, Schema, Stream } from "effect"; import { HttpClient, HttpClientRequest, HttpClientResponse } from "effect/unstable/http"; import { endpointForTelemetry, type ToolFileValue } from "@executor-js/sdk/core"; import { OpenApiInvocationError } from "./errors"; import { isNdjsonMediaType, NDJSON_MEDIA_TYPES, resolveServerUrl } from "./openapi-utils"; import { type EncodingObject, type OperationFileHint, type OperationBinding, InvocationResult, type MediaBinding, type OperationParameter, type ServerInfo, } from "./types"; // --------------------------------------------------------------------------- // Parameter reading // --------------------------------------------------------------------------- const CONTAINER_KEYS: Record<string, readonly string[]> = { path: ["path", "pathParams", "params"], query: ["query", "queryParams", "params"], header: ["headers", "header"], cookie: ["cookies", "cookie"], }; const readParamValue = (args: Record<string, unknown>, param: OperationParameter): unknown => { const direct = args[param.name]; if (direct !== undefined) return direct; for (const key of CONTAINER_KEYS[param.location] ?? []) { const container = args[key]; if (typeof container === "object" && container !== null && !Array.isArray(container)) { const nested = (container as Record<string, unknown>)[param.name]; if (nested !== undefined) return nested; } } return undefined; }; const primitiveToString = (value: unknown): string => typeof value === "object" && value !== null ? JSON.stringify(value) : String(value); // RFC 3986 §2.2 reserved chars. `allowReserved: true` leaves these // unencoded; default OAS behavior encodes everything non-unreserved. const RESERVED_UNENCODED_RE = /[A-Za-z0-9\-._~:/?#[\]@!$&'()*+,;=]/; const encodeReservedAware = (raw: string, allowReserved: boolean): string => { if (!allowReserved) return encodeURIComponent(raw); // Walk char-by-char so the reserved set passes through as-is. let out = ""; for (const ch of raw) { out += RESERVED_UNENCODED_RE.test(ch) ? ch : encodeURIComponent(ch); } return out; }; const queryParamValues = (value: unknown, param: OperationParameter): string[] => { if (value === undefined || value === null) return []; if (!Array.isArray(value)) return [primitiveToString(value)]; const style = Option.getOrUndefined(param.style) ?? "form"; const explode = Option.getOrElse(param.explode, () => true); if (explode) return value.map(primitiveToString); const separator = style === "spaceDelimited" ? " " : style === "pipeDelimited" ? "|" : ","; return [value.map(primitiveToString).join(separator)]; }; // --------------------------------------------------------------------------- // Path resolution // --------------------------------------------------------------------------- const resolvePath = Effect.fn("OpenApi.resolvePath")(function* ( pathTemplate: string, args: Record<string, unknown>, parameters: readonly OperationParameter[], ) { let resolved = pathTemplate; for (const param of parameters) { if (param.location !== "path") continue; const value = readParamValue(args, param); if (value === undefined || value === null) { if (param.required) { return yield* new OpenApiInvocationError({ message: `Missing required path parameter: ${param.name}`, statusCode: Option.none(), }); } continue; } const encoded = encodeReservedAware( String(value), Option.getOrElse(param.allowReserved, () => false), ); resolved = resolved.replaceAll(`{${param.name}}`, encoded); resolved = resolved.replaceAll(`{+${param.name}}`, encoded); } const remaining = [...resolved.matchAll(/\{([^{}]+)\}/g)] .map((m) => m[1]) .filter((v): v is string => typeof v === "string"); for (const name of remaining) { const value = args[name]; if (value !== undefined && value !== null) { resolved = resolved.replaceAll(`{${name}}`, encodeURIComponent(String(value))); } } const unresolved = [...resolved.matchAll(/\{([^{}]+)\}/g)] .map((m) => m[1]) .filter((v): v is string => typeof v === "string"); if (unresolved.length > 0) { return yield* new OpenApiInvocationError({ message: `Unresolved path parameters: ${[...new Set(unresolved)].join(", ")}`, statusCode: Option.none(), }); } return resolved; }); // GitHub (and some other upstreams) reject requests that lack a User-Agent // header with a 403 ("Request forbidden by administrative rules"), which is // indistinguishable from a credential rejection downstream. Send a default so // those calls succeed. It is applied before operation header params and // resolved auth headers, so a spec- or connection-provided User-Agent wins. const DEFAULT_USER_AGENT = "executor"; const applyHeaders = ( request: HttpClientRequest.HttpClientRequest, headers: Record<string, string>, ): HttpClientRequest.HttpClientRequest => { let req = request; for (const [name, value] of Object.entries(headers)) { req = HttpClientRequest.setHeader(req, name, value); } return req; }; // --------------------------------------------------------------------------- // Response helpers // --------------------------------------------------------------------------- const normalizeContentType = (ct: string | null | undefined): string => ct?.split(";")[0]?.trim().toLowerCase() ?? ""; export const STREAM_MAX_BYTES = 1_000_000; export const STREAM_MAX_MS = 10_000; // Buffered bodies should normally finish soon after headers arrive. This is // deliberately generous while still providing plain Node runtimes a backstop. export const RESPONSE_BODY_TIMEOUT_MS = 60_000; // Below Cloudflare's ~125s subrequest limit so cloud fails first with an // actionable error, but above the slowest legitimate buffered upstreams // observed in production (30d telemetry: successful calls cluster under // 110s; a 60s cap would have broken ~40 real calls). export const RESPONSE_HEADERS_TIMEOUT_MS = 110_000; class StreamCapReached extends Error { readonly _tag = "StreamCapReached"; } export interface StreamingResponseCaps { readonly maxBytes?: number; readonly maxMs?: number; } export interface InvokeOptions { readonly responseHeadersTimeoutMs?: number; readonly responseBodyTimeoutMs?: number; } const formatTimeout = (timeoutMs: number): string => timeoutMs % 1_000 === 0 ? `${timeoutMs / 1_000}s` : `${timeoutMs}ms`; const responseHeadersTimeoutMessage = (timeoutMs: number): string => `Upstream returned no response headers within ${formatTimeout(timeoutMs)}. The endpoint may be a live stream with no data to send (for example, runtime logs of an idle deployment); the request was aborted. Retry when the resource has activity, or use a non-streaming endpoint.`; const responseBodyTimeoutMessage = (timeoutMs: number): string => `Upstream response body did not finish within ${formatTimeout(timeoutMs)}. The request was aborted. Retry the operation; if it times out again, check the upstream service health or use an endpoint with a bounded response.`; const STREAMING_RESPONSE_CONTENT_TYPES = new Set([...NDJSON_MEDIA_TYPES, "text/event-stream"]); const isStreamingResponseContentType = (ct: string | null | undefined): boolean => STREAMING_RESPONSE_CONTENT_TYPES.has(normalizeContentType(ct)); const isJsonContentType = (ct: string | null | undefined): boolean => { const normalized = normalizeContentType(ct); if (!normalized) return false; return ( normalized === "application/json" || normalized.includes("+json") || normalized.includes("json") ); }; const isFormUrlEncoded = (ct: string | null | undefined): boolean => normalizeContentType(ct) === "application/x-www-form-urlencoded"; const isMultipartFormData = (ct: string | null | undefined): boolean => normalizeContentType(ct).startsWith("multipart/form-data"); const isXmlContentType = (ct: string | null | undefined): boolean => { const normalized = normalizeContentType(ct); if (!normalized) return false; return ( normalized === "application/xml" || normalized === "text/xml" || normalized.endsWith("+xml") ); }; const isTextContentType = (ct: string | null | undefined): boolean => normalizeContentType(ct).startsWith("text/"); const isOctetStream = (ct: string | null | undefined): boolean => normalizeContentType(ct) === "application/octet-stream"; const decodeJsonLine = Schema.decodeUnknownExit(Schema.fromJsonString(Schema.Unknown)); const parseStreamingJsonLines = (text: string, truncated: boolean): unknown => { const lines = text.split(/\r?\n/); let lastNonEmptyIndex = -1; for (let index = lines.length - 1; index >= 0; index--) { const line = lines[index]; if (line !== undefined && line.trim() !== "") { lastNonEmptyIndex = index; break; } } const rows: unknown[] = []; for (let index = 0; index < lines.length; index++) { const line = lines[index]; if (line === undefined || line.trim() === "") continue; const parsed = decodeJsonLine(line); if (Exit.isSuccess(parsed)) { rows.push(parsed.value); continue; } // A trailing fragment cut mid-line by the byte/time cap is expected; any // other unparseable line means the body is not NDJSON after all. if (truncated && index === lastNonEmptyIndex) continue; return text; } return rows; }; const decodeStreamingResponseBody = ( contentType: string | null, text: string, truncated: boolean, ): unknown => (isNdjsonMediaType(contentType) ? parseStreamingJsonLines(text, truncated) : text); export const collectStreamingBody = ( stream: Stream.Stream<Uint8Array, unknown, never>, contentType: string | null, caps: StreamingResponseCaps = {}, ) => Effect.gen(function* () { const maxBytes = caps.maxBytes ?? STREAM_MAX_BYTES; const maxMs = caps.maxMs ?? STREAM_MAX_MS; const chunks: Uint8Array[] = []; let bytes = 0; let truncated = false; const startedAt = Date.now(); const runFork = Effect.runForkWith(yield* Effect.context<never>()); const completedExitOption = yield* Effect.callback<Option.Option<Exit.Exit<void, unknown>>>( (resume, signal) => { let settled = false; const collectEffect = stream.pipe( Stream.timeout(maxMs), Stream.runForEach((chunk) => Effect.gen(function* () { const remaining = maxBytes - bytes; if (remaining <= 0) { truncated = true; return yield* Effect.fail(new StreamCapReached()); } if (chunk.byteLength > remaining) { chunks.push(chunk.subarray(0, remaining)); bytes += remaining; truncated = true; return yield* Effect.fail(new StreamCapReached()); } chunks.push(chunk); bytes += chunk.byteLength; if (bytes >= maxBytes) { truncated = true; return yield* Effect.fail(new StreamCapReached()); } }), ), Effect.catch((error: unknown) => error instanceof StreamCapReached ? Effect.void : Effect.fail(error), ), Effect.exit, Effect.tap((exit) => Effect.sync(() => { if (settled) return; settled = true; clearTimeout(timer); resume(Effect.succeed(Option.some(exit))); }), ), ); const fiber = runFork(collectEffect); const interrupt = () => { runFork(Fiber.interrupt(fiber)); }; const timer = setTimeout(() => { if (settled) return; settled = true; truncated = true; interrupt(); resume(Effect.succeed(Option.none())); }, maxMs + 1_000); signal.addEventListener("abort", interrupt, { once: true }); return Effect.sync(() => { clearTimeout(timer); signal.removeEventListener("abort", interrupt); if (!settled) interrupt(); }); }, ); if (Option.isSome(completedExitOption) && Exit.isFailure(completedExitOption.value)) { return yield* Effect.failCause(completedExitOption.value.cause); } if (Option.isNone(completedExitOption)) truncated = true; const durationMs = Date.now() - startedAt; if (durationMs >= maxMs) truncated = true; const collected = new Uint8Array(bytes); let offset = 0; for (const chunk of chunks) { collected.set(chunk, offset); offset += chunk.byteLength; } const text = new TextDecoder("utf-8").decode(collected); return { data: decodeStreamingResponseBody(contentType, text, truncated), headers: { "x-executor-stream": truncated ? "truncated" : "complete", "x-executor-stream-bytes": String(bytes), "x-executor-stream-duration-ms": String(durationMs), }, }; }); const bytesToBase64 = (bytes: Uint8Array): string => { let binary = ""; const chunkSize = 0x8000; for (let i = 0; i < bytes.length; i += chunkSize) { binary += String.fromCharCode(...bytes.subarray(i, i + chunkSize)); } return btoa(binary); }; const normalizeBase64 = (value: string, encoding: "base64" | "base64url"): string => { const compact = value.replace(/\s/g, ""); const alphabet = encoding === "base64url" ? compact.replace(/-/g, "+").replace(/_/g, "/") : compact; const remainder = alphabet.length % 4; return remainder === 0 ? alphabet : `${alphabet}${"=".repeat(4 - remainder)}`; }; const byteLengthFromBase64 = (base64: string): number => { const compact = base64.replace(/\s/g, ""); const padding = compact.endsWith("==") ? 2 : compact.endsWith("=") ? 1 : 0; return Math.max(0, Math.floor((compact.length * 3) / 4) - padding); }; const isGenericMimeType = (mimeType: string): boolean => normalizeContentType(mimeType) === "application/octet-stream"; const startsWithBytes = (bytes: Uint8Array, prefix: readonly number[]): boolean => prefix.every((byte, index) => bytes[index] === byte); const isLikelyUtf8Text = (bytes: Uint8Array): boolean => { if (bytes.length === 0) return false; let text: string; // oxlint-disable-next-line executor/no-try-catch-or-throw -- boundary: TextDecoder throws while probing arbitrary binary content try { text = new TextDecoder("utf-8", { fatal: true }).decode(bytes); } catch { return false; } let suspicious = 0; for (let index = 0; index < text.length; index += 1) { const code = text.charCodeAt(index); const allowedControl = code === 0x09 || code === 0x0a || code === 0x0c || code === 0x0d; if (code === 0x00) return false; if (code < 0x20 && !allowedControl) suspicious += 1; } return suspicious / Math.max(1, text.length) <= 0.02; }; const sniffMimeType = (bytes: Uint8Array): string | null => { if (startsWithBytes(bytes, [0xff, 0xd8, 0xff])) return "image/jpeg"; if (startsWithBytes(bytes, [0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a])) { return "image/png"; } if ( startsWithBytes(bytes, [0x47, 0x49, 0x46, 0x38, 0x37, 0x61]) || startsWithBytes(bytes, [0x47, 0x49, 0x46, 0x38, 0x39, 0x61]) ) { return "image/gif"; } if ( startsWithBytes(bytes, [0x52, 0x49, 0x46, 0x46]) && bytes[8] === 0x57 && bytes[9] === 0x45 && bytes[10] === 0x42 && bytes[11] === 0x50 ) { return "image/webp"; } if (startsWithBytes(bytes, [0x25, 0x50, 0x44, 0x46, 0x2d])) return "application/pdf"; if ( startsWithBytes(bytes, [0x50, 0x4b, 0x03, 0x04]) || startsWithBytes(bytes, [0x50, 0x4b, 0x05, 0x06]) || startsWithBytes(bytes, [0x50, 0x4b, 0x07, 0x08]) ) { return "application/zip"; } if (isLikelyUtf8Text(bytes)) return "text/plain"; return null; }; const bytesFromBase64Prefix = (base64: string): Uint8Array => { const prefix = base64.slice(0, Math.min(base64.length, 64)); const binary = atob(prefix); const bytes = new Uint8Array(binary.length); for (let index = 0; index < binary.length; index += 1) { bytes[index] = binary.charCodeAt(index); } return bytes; }; const sniffMimeTypeFromBase64 = (base64: string): string | null => sniffMimeType(bytesFromBase64Prefix(base64)); type DecodedBase64Body = { readonly ok: true; readonly bytes: Uint8Array } | { readonly ok: false }; const base64ToUint8Array = (value: string): Uint8Array | null => { let binary = ""; // oxlint-disable-next-line executor/no-try-catch-or-throw -- boundary: atob throws for invalid base64; invalid shapes are treated as non-byte input try { binary = atob(normalizeBase64(value, "base64")); } catch { return null; } const bytes = new Uint8Array(binary.length); for (let index = 0; index < binary.length; index += 1) { bytes[index] = binary.charCodeAt(index); } return bytes; }; const decodeBase64Body = (value: string): DecodedBase64Body => { const bytes = base64ToUint8Array(value); return bytes ? { ok: true, bytes } : { ok: false }; }; const toUint8Array = (value: unknown): Uint8Array | null => { if (value instanceof Uint8Array) return value; if (value instanceof ArrayBuffer) return new Uint8Array(value); if (ArrayBuffer.isView(value)) { const view = value as ArrayBufferView; return new Uint8Array(view.buffer, view.byteOffset, view.byteLength); } if (Array.isArray(value) && value.every((v) => typeof v === "number")) { return new Uint8Array(value as readonly number[]); } return null; }; const readNestedBodyBase64 = (value: unknown): unknown => typeof value === "object" && value !== null && !Array.isArray(value) && Object.prototype.hasOwnProperty.call(value, "bodyBase64") ? (value as Record<string, unknown>).bodyBase64 : undefined; const readHintString = (option: OperationFileHint["dataField"], fallback: string): string => Option.getOrElse(option, () => fallback); const readHintMimeType = (hint: OperationFileHint, fallback: string): string => Option.getOrElse(hint.mimeType, () => fallback); const readHintEncoding = (hint: OperationFileHint): "base64" | "base64url" => Option.getOrElse(hint.encoding, () => "base64");