import {
  type AssistantMessage,
  type AssistantMessageEventStream,
  type Context,
  type Model,
  type SimpleStreamOptions,
  type Message,
  type ToolCall,
  createAssistantMessageEventStream,
  calculateCost,
} from "@earendil-works/pi-ai";
import { createHash } from "node:crypto";
import { existsSync, realpathSync } from "node:fs";
import http from "node:http";
import https from "node:https";
import { dirname, join } from "node:path";
import { pathToFileURL } from "node:url";
import { resolveParleyApiKey, resolveParleyBaseUrl } from "./keys.js";
import { resolveRequestContext, toConverterContext } from "./context.js";
import { usesAdaptiveThinking } from "./models.js";

const MISSING_KEY_MESSAGE =
  "Missing Parley API key. Set PARLEY_API_KEY (or MIT_PARLEY_API_KEY / MY_MIT_PARLEY_API_KEY), " +
  "or PARLEY_<NICK>_API_KEY and pick it with /parley-key.";

type ResponsesShared = typeof import("@earendil-works/pi-ai/api/openai-responses-shared");

let responsesSharedPromise: Promise<ResponsesShared | undefined> | undefined;

// Pi's extension loader only aliases a few @earendil-works/pi-ai entrypoints
// (root, /compat, /oauth, /providers/all). The Responses helpers live under
// /api/*, which is not aliased, and an npm-installed extension has no pi-ai
// copy of its own to resolve it against. Try a normal import first (works in a
// dev checkout with node_modules), then locate the helper inside the running
// Pi installation. Return undefined when neither works so callers can fall
// back to Chat Completions instead of failing to load the extension.
async function loadResponsesShared(): Promise<ResponsesShared | undefined> {
  if (!responsesSharedPromise) {
    responsesSharedPromise = (async () => {
      try {
        return (await import("@earendil-works/pi-ai/api/openai-responses-shared")) as ResponsesShared;
      } catch {
        // Fall through to host lookup.
      }

      const candidates = ["@earendil-works/pi-ai", "@mariozechner/pi-ai"];
      let dir: string | undefined;
      try {
        dir = dirname(realpathSync(process.argv[1] ?? ""));
      } catch {
        return undefined;
      }

      while (dir) {
        for (const pkg of candidates) {
          const file = join(dir, "node_modules", pkg, "dist", "api", "openai-responses-shared.js");
          if (existsSync(file)) {
            try {
              return (await import(pathToFileURL(file).href)) as ResponsesShared;
            } catch {
              return undefined;
            }
          }
        }
        const parent = dirname(dir);
        if (parent === dir) break;
        dir = parent;
      }
      return undefined;
    })();
  }
  return responsesSharedPromise;
}

function getParleyRequestBaseUrl(modelBaseUrl: string): string {
  // Resolve this lazily at request time so the active key profile's base URL
  // (under nono, its proxy route) and any override set after the provider was
  // registered still win.
  const runtimeBaseUrl = resolveParleyBaseUrl();
  return runtimeBaseUrl || modelBaseUrl;
}

function getParleyApiKey(options: SimpleStreamOptions | undefined): string | undefined {
  // The active key profile wins over the key Pi resolved from the provider
  // registration, which is fixed at startup and cannot follow /parley-key.
  const profileKey = resolveParleyApiKey();
  if (profileKey) return profileKey;
  if (options?.apiKey?.trim()) return options.apiKey.trim();

  return undefined;
}

function responsesBaseUrl(baseUrl: string): string {
  const url = new URL(baseUrl.trim().replace(/\/+$/, ""));
  const pathname = url.pathname.replace(/\/+$/, "");

  if (!pathname) {
    url.pathname = "/v1";
  } else if (pathname.endsWith("/v1/chat/completions")) {
    url.pathname = pathname.slice(0, -"/chat/completions".length);
  } else if (pathname.endsWith("/v1/responses")) {
    url.pathname = pathname.slice(0, -"/responses".length);
  } else if (pathname.endsWith("/v1")) {
    url.pathname = pathname;
  } else {
    url.pathname = `${pathname}/v1`;
  }

  url.search = "";
  return url.toString().replace(/\/+$/, "");
}

function responsesUrl(baseUrl: string): URL {
  return new URL(`${responsesBaseUrl(baseUrl)}/responses`);
}

async function* parseSseJson(response: Response): AsyncIterable<any> {
  const reader = response.body?.getReader();
  if (!reader) return;

  const decoder = new TextDecoder();
  let buffer = "";

  while (true) {
    const { value, done } = await reader.read();
    if (done) break;
    buffer += decoder.decode(value, { stream: true });
    const lines = buffer.split("\n");
    buffer = lines.pop() || "";

    for (const line of lines) {
      if (!line.trim() || !line.startsWith("data: ")) continue;
      const dataStr = line.slice(6).trim();
      if (dataStr === "[DONE]") continue;
      yield JSON.parse(dataStr);
    }
  }

  buffer += decoder.decode();
  for (const line of buffer.split("\n")) {
    if (!line.trim() || !line.startsWith("data: ")) continue;
    const dataStr = line.slice(6).trim();
    if (dataStr === "[DONE]") continue;
    yield JSON.parse(dataStr);
  }
}

function chatCompletionsUrl(baseUrl: string): URL {
  const url = new URL(baseUrl.trim().replace(/\/+$/, ""));
  const pathname = url.pathname.replace(/\/+$/, "");

  if (!pathname) {
    url.pathname = "/v1/chat/completions";
  } else if (pathname.endsWith("/v1/chat/completions")) {
    url.pathname = pathname;
  } else if (pathname.endsWith("/v1")) {
    url.pathname = `${pathname}/chat/completions`;
  } else {
    url.pathname = `${pathname}/v1/chat/completions`;
  }

  url.search = "";
  return url;
}

function requestModuleForUrl(url: URL): typeof http | typeof https {
  if (url.protocol === "http:") return http;
  if (url.protocol === "https:") return https;
  throw new Error(`Unsupported Parley URL protocol: ${url.protocol}`);
}

function isLoopbackHostname(hostname: string): boolean {
  return hostname === "localhost" || hostname === "127.0.0.1" || hostname === "::1" || hostname === "[::1]";
}

function shouldBypassEnvProxy(url: URL): boolean {
  // A local credential proxy is reached at http://127.0.0.1:<port>/... . Node's
  // NODE_USE_ENV_PROXY could otherwise route that request back through the same
  // proxy, which hangs. Keep normal env-proxy behavior for real upstream HTTPS URLs.
  return url.protocol === "http:" && isLoopbackHostname(url.hostname);
}

function isBlankTextBlock(block: any): boolean {
  return block?.type === "text" && typeof block.text === "string" && block.text.trim().length === 0;
}

// Gemini's function_declarations accept only an OpenAPI subset of JSON Schema
// and reject the whole request on unknown keywords (e.g. exclusiveMinimum,
// deprecated), which tools from other extensions commonly emit.
const GEMINI_UNSUPPORTED_SCHEMA_KEYS = new Set([
  "$schema",
  "$id",
  "$comment",
  "exclusiveMinimum",
  "exclusiveMaximum",
  "deprecated",
  "readOnly",
  "writeOnly",
  "examples",
]);

export function sanitizeSchemaForGemini(schema: any, isPropertyMap = false): any {
  if (Array.isArray(schema)) return schema.map((item) => sanitizeSchemaForGemini(item));
  if (!schema || typeof schema !== "object") return schema;
  const out: Record<string, any> = {};
  for (const [key, value] of Object.entries(schema)) {
    // Keys of a `properties` map are user-defined names, not schema keywords.
    if (!isPropertyMap && GEMINI_UNSUPPORTED_SCHEMA_KEYS.has(key)) continue;
    out[key] = sanitizeSchemaForGemini(value, !isPropertyMap && (key === "properties" || key === "patternProperties"));
  }
  return out;
}

function mapToolCallIdForParley(id: string, mappedIds: Map<string, string>): string {
  const existing = mappedIds.get(id);
  if (existing) return existing;

  // Parley's Chat Completions endpoint validates OpenAI-compatible tool call
  // ids, which are limited to 64 characters. Pi/provider-generated ids can be
  // longer, so shorten them deterministically and use the same mapped id for
  // the corresponding tool result later in this request.
  const mapped =
    id.length <= 64
      ? id
      : `call_${createHash("sha256").update(id).digest("base64url").slice(0, 43)}`;
  mappedIds.set(id, mapped);
  return mapped;
}

function sanitizeMessagesForParley(messages: any[]): any[] {
  const result: any[] = [];

  for (const message of messages) {
    const sanitized = { ...message };

    if (typeof sanitized.content === "string") {
      if (sanitized.content.trim().length === 0) {
        if (sanitized.role === "tool") sanitized.content = "Tool returned no text.";
        else if (sanitized.role === "assistant") sanitized.content = null;
        else continue;
      }
    } else if (Array.isArray(sanitized.content)) {
      sanitized.content = sanitized.content.filter((block: any) => !isBlankTextBlock(block));
      if (sanitized.content.length === 0) {
        if (sanitized.role === "tool") sanitized.content = "Tool returned no text.";
        else if (sanitized.role === "assistant") sanitized.content = null;
        else continue;
      }
    } else if (
      sanitized.content == null &&
      sanitized.role === "assistant" &&
      !sanitized.tool_calls &&
      !sanitized.thinking
    ) {
      continue;
    }

    result.push(sanitized);
  }

  return result;
}

function mapMessages(messages: Message[]) {
  const result: any[] = [];
  const mappedToolCallIds = new Map<string, string>();

  for (const msg of messages) {
    if (msg.role === "user") {
      if (typeof msg.content === "string") {
        if (msg.content.trim().length > 0) result.push({ role: "user", content: msg.content });
      } else {
        const content: any[] = [];
        for (const part of msg.content) {
          if (part.type === "text") {
            if (part.text.trim().length > 0) content.push({ type: "text", text: part.text });
          } else if (part.type === "image") {
            content.push({
              type: "image_url",
              image_url: { url: `data:${part.mimeType};base64,${part.data}` },
            });
          }
        }
        if (content.length > 0) result.push({ role: "user", content });
      }
    } else if (msg.role === "assistant") {
      let contentStr = "";
      let thinkingStr = "";
      let thinkingSignature: string | undefined;

      for (const part of msg.content) {
        if (part.type === "text") contentStr += part.text;
        if (part.type === "thinking") {
          thinkingStr += part.thinking;
          if (part.thinkingSignature) thinkingSignature = part.thinkingSignature;
        }
      }

      const toolCalls = msg.content.filter((p) => p.type === "toolCall") as ToolCall[];
      if (contentStr.trim().length === 0 && thinkingStr.trim().length === 0 && toolCalls.length === 0) {
        continue;
      }

      const apiMsg: any = { role: "assistant", content: contentStr.trim().length > 0 ? contentStr : null };

      if (thinkingStr.trim().length > 0) apiMsg.thinking = thinkingStr;
      if (thinkingSignature) apiMsg.thinking_signature = thinkingSignature;

      if (toolCalls.length > 0) {
        apiMsg.tool_calls = toolCalls.map((tc) => ({
          id: mapToolCallIdForParley(tc.id, mappedToolCallIds),
          type: "function",
          function: { name: tc.name, arguments: JSON.stringify(tc.arguments) },
        }));
      }

      result.push(apiMsg);
    } else if (msg.role === "toolResult") {
      let text = "";
      for (const part of msg.content) {
        if (part.type === "text") text += part.text + "\n";
        if (part.type === "image") text += `[image: ${part.mimeType}]\n`;
      }
      result.push({ role: "tool", tool_call_id: mapToolCallIdForParley(msg.toolCallId, mappedToolCallIds), content: text.trim().length > 0 ? text : "Tool returned no text." });
    }
  }

  return result;
}

function stopReasonFromFinishReason(finishReason: string): AssistantMessage["stopReason"] {
  if (finishReason === "length") return "length";
  if (finishReason === "tool_calls" || finishReason === "tool_use") return "toolUse";
  return "stop";
}

function headersToRecord(headers: import("node:http").IncomingHttpHeaders): Record<string, string> {
  const result: Record<string, string> = {};
  for (const [key, value] of Object.entries(headers)) {
    if (Array.isArray(value)) result[key] = value.join(", ");
    else if (value !== undefined) result[key] = String(value);
  }
  return result;
}

function isOpenAIResponsesModel(model: Model<any>): boolean {
  // GPT-5.6 needs Responses for reasoning plus tools. GPT-5 / 5-mini / 5-nano
  // need it because Parley's Chat Completions route forces reasoning_effort to
  // "none" whenever tools are present, which those models reject. Codex is
  // Responses-only upstream; Astra takes the same route as GPT-5.6.
  return (
    /^openai\/gpt-(?:5\.(?:[6-9]|\d{2,})(?:-|$)|(?:[6-9]|\d{2,})(?:[.-]|$))/.test(model.id) ||
    model.id.includes("-codex") ||
    isLegacyGpt5Model(model)
  );
}

function isLegacyGpt5Model(model: Model<any>): boolean {
  return ["openai/gpt-5", "openai/gpt-5-mini", "openai/gpt-5-nano"].includes(model.id);
}

// Map a Pi thinking level to a `reasoning.effort` value the model accepts on
// Parley's Responses route. `undefined` level means thinking is off.
function responsesEffort(model: Model<any>, level: string | undefined): string {
  const mapped = level ? (model.thinkingLevelMap?.[level as keyof typeof model.thinkingLevelMap] ?? level) : "none";
  if (model.id.endsWith("-astra")) {
    // Astra: low | medium | high | xhigh | max. No "none" or "minimal".
    return mapped === "none" || mapped === "minimal" ? "low" : mapped;
  }
  if (isLegacyGpt5Model(model)) {
    // minimal | low | medium | high. No "none".
    if (mapped === "none") return "minimal";
    return mapped === "xhigh" || mapped === "max" ? "high" : mapped;
  }
  // GPT-5.6 and Codex: none | minimal | low | medium | high.
  return mapped === "xhigh" || mapped === "max" ? "high" : mapped;
}

async function streamParleyResponses(
  shared: ResponsesShared,
  model: Model<any>,
  context: Context,
  options: SimpleStreamOptions | undefined,
  stream: AssistantMessageEventStream,
): Promise<void> {
  await (async () => {
    const output: AssistantMessage = {
      role: "assistant",
      content: [],
      api: model.api,
      provider: model.provider,
      model: model.id,
      usage: {
        input: 0,
        output: 0,
        cacheRead: 0,
        cacheWrite: 0,
        totalTokens: 0,
        cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
      },
      stopReason: "stop",
      timestamp: Date.now(),
    };

    try {
      const url = responsesUrl(getParleyRequestBaseUrl(model.baseUrl));
      const apiKey = getParleyApiKey(options);
      if (!apiKey) {
        throw new Error(MISSING_KEY_MESSAGE);
      }

      const maxTokens = options?.maxTokens || model.maxTokens;
      // See toConverterContext: the loaded converter may predate or postdate
      // the Pi that built `context`, so give it a shape both versions read.
      const converterContext = toConverterContext(context);
      const tools = converterContext.tools ?? [];
      const params: any = {
        model: model.id,
        input: shared.convertResponsesMessages(
          model,
          converterContext as Parameters<ResponsesShared["convertResponsesMessages"]>[1],
          new Set([model.provider]),
        ),
        stream: true,
        store: false,
      };

      if (maxTokens) params.max_output_tokens = Math.max(maxTokens, 16);
      if (options?.temperature !== undefined) params.temperature = options.temperature;
      if (tools.length > 0) params.tools = shared.convertResponsesTools(tools);

      if (options?.reasoning && model.reasoning) {
        params.reasoning = { effort: responsesEffort(model, options.reasoning), summary: "auto" };
        params.include = ["reasoning.encrypted_content"];
      } else if (model.reasoning) {
        params.reasoning = { effort: responsesEffort(model, undefined) };
      }

      const payload = options?.onPayload ? (await options.onPayload(params, model)) ?? params : params;
      const controller = options?.timeoutMs ? new AbortController() : undefined;
      const timeout = controller && options?.timeoutMs ? setTimeout(() => controller.abort(), options.timeoutMs) : undefined;
      const signal = options?.signal ?? controller?.signal;

      const response = await fetch(url, {
        method: "POST",
        headers: {
          "Content-Type": "application/json",
          Accept: "text/event-stream",
          ...(options?.headers ?? {}),
          Authorization: `Bearer ${apiKey}`,
        },
        body: JSON.stringify(payload),
        signal,
      });
      if (timeout) clearTimeout(timeout);

      await options?.onResponse?.({ status: response.status, headers: Object.fromEntries(response.headers.entries()) }, model);
      if (!response.ok) {
        throw new Error(`HTTP ${response.status}: ${await response.text()}`);
      }

      stream.push({ type: "start", partial: output });
      await shared.processResponsesStream(parseSseJson(response) as any, output, stream, model);

      if (options?.signal?.aborted) throw new Error("Request was aborted");
      if (output.stopReason === "aborted" || output.stopReason === "error") {
        throw new Error(output.errorMessage || "An unknown error occurred");
      }

      stream.push({ type: "done", reason: output.stopReason as "stop" | "length" | "toolUse", message: output });
      stream.end();
    } catch (error) {
      output.stopReason = options?.signal?.aborted ? "aborted" : "error";
      output.errorMessage = error instanceof Error ? error.message : String(error);
      stream.push({ type: "error", reason: output.stopReason, error: output });
      stream.end();
    }
  })();
}

export function streamSimple(
  model: Model<any>,
  context: Context,
  options?: SimpleStreamOptions,
): AssistantMessageEventStream {
  const stream = createAssistantMessageEventStream();

  (async () => {
    if (isOpenAIResponsesModel(model)) {
      const shared = await loadResponsesShared();
      if (shared) {
        await streamParleyResponses(shared, model, context, options, stream);
        return;
      }
    }
    await streamParleyChatCompletions(model, context, options, stream);
  })();

  return stream;
}

async function streamParleyChatCompletions(
  model: Model<any>,
  context: Context,
  options: SimpleStreamOptions | undefined,
  stream: AssistantMessageEventStream,
): Promise<void> {
  await (async () => {
    const output: AssistantMessage = {
      role: "assistant",
      content: [],
      api: model.api,
      provider: model.provider,
      model: model.id,
      usage: {
        input: 0,
        output: 0,
        cacheRead: 0,
        cacheWrite: 0,
        totalTokens: 0,
        cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
      },
      stopReason: "stop",
      timestamp: Date.now(),
    };

    let closed = false;
    const fail = (reason: "error" | "aborted", message?: string) => {
      if (closed) return;
      closed = true;
      output.stopReason = reason;
      if (message) output.errorMessage = message;
      stream.push({ type: "error", reason, error: output });
      stream.end();
    };

    try {
      if (options?.signal?.aborted) {
        fail("aborted", "Request was aborted");
        return;
      }

      stream.push({ type: "start", partial: output });

      const request = resolveRequestContext(context);
      const messages = [];
      if (request.systemPrompt.trim()) {
        messages.push({ role: "system", content: request.systemPrompt });
      }
      messages.push(...mapMessages(request.messages));
      const sanitizedMessages = sanitizeMessagesForParley(messages);

      const tools = request.tools;
      const hasTools = tools.length > 0;
      const reqBody: any = {
        model: model.id,
        messages: sanitizedMessages,
        stream: true,
      };

      if (hasTools) {
        reqBody.tools = tools.map((t) => ({
          type: "function",
          function: {
            name: t.name,
            description: t.description,
            parameters: model.id.startsWith("google/") ? sanitizeSchemaForGemini(t.parameters) : t.parameters,
          },
        }));
      }

      let maxTokens = options?.maxTokens || model.maxTokens;

      if (options?.reasoning && model.reasoning) {
        const isOpenAIModel = model.id.startsWith("openai/");
        const isAdaptiveModel = usesAdaptiveThinking(model.id);

        if (isOpenAIModel) {
          // Parley's documented `thinking` field currently fails on OpenAI chat
          // routes with `Unknown parameter: 'reasoning'`. Use the OpenAI-style
          // effort knob for OpenAI models instead; OpenAI does not return visible
          // thinking text on Chat Completions.
          //
          // Parley GPT-5.6 chat completions reject function tools unless
          // reasoning_effort is explicitly disabled. Pi relies heavily on tools,
          // so prefer working tool calls over hidden reasoning for these models.
          if (hasTools && model.id.startsWith("openai/gpt-5.6-")) {
            reqBody.reasoning_effort = "none";
          } else {
            const mappedEffort = model.thinkingLevelMap?.[options.reasoning] ?? options.reasoning;
            reqBody.reasoning_effort = mappedEffort === "xhigh" ? "high" : mappedEffort;
          }
        } else if (isAdaptiveModel) {
          reqBody.thinking = { type: "adaptive" };
        } else {
          // Older Claude models and Gemini use token-budget thinking.
          // Opus 4.7+ and Sonnet 5+ use adaptive thinking instead.
          const defaultBudgets: Record<string, number> = {
            minimal: 1024,
            low: 2048,
            medium: 4096,
            high: 8192,
            xhigh: 16384,
          };
          const budgets: Record<string, number> = { ...defaultBudgets, ...(options.thinkingBudgets as any) };
          const budgetTokens = budgets[options.reasoning] || defaultBudgets.high;
          reqBody.thinking = { type: "enabled", budget_tokens: budgetTokens };
          if (maxTokens <= budgetTokens) {
            maxTokens = budgetTokens + 4096;
          }
        }
      }

      if (maxTokens) {
        reqBody.max_completion_tokens = maxTokens;
      }
      if (options?.temperature !== undefined) reqBody.temperature = options.temperature;

      const payload = options?.onPayload ? (await options.onPayload(reqBody, model)) ?? reqBody : reqBody;
      const reqData = JSON.stringify(payload);

      const url = chatCompletionsUrl(getParleyRequestBaseUrl(model.baseUrl));
      const apiKey = getParleyApiKey(options);
      if (!apiKey) {
        throw new Error(MISSING_KEY_MESSAGE);
      }

      const requestModule = requestModuleForUrl(url);

      const req = requestModule.request(
        url,
        {
          method: "POST",
          headers: {
            "Content-Type": "application/json",
            "Content-Length": Buffer.byteLength(reqData),
            Accept: "text/event-stream",
            ...(options?.headers ?? {}),
            Authorization: `Bearer ${apiKey}`,
          },
          ...(shouldBypassEnvProxy(url) ? { agent: false } : {}),
        },
        async (res) => {
          await options?.onResponse?.({ status: res.statusCode ?? 0, headers: headersToRecord(res.headers) }, model);

          if (res.statusCode && res.statusCode >= 400) {
            let errBody = "";
            res.on("data", (chunk) => (errBody += chunk.toString()));
            res.on("end", () => fail("error", `HTTP ${res.statusCode}: ${errBody}`));
            return;
          }

          let buffer = "";
          let contentIndex = -1;
          let currentType: "text" | "thinking" | "toolCall" | null = null;
          let currentToolKey: string | null = null;
          const toolStates = new Map<string, { contentIndex: number; args: string }>();

          const parseToolArgs = (toolIndex: number, args: string) => {
            try {
              (output.content[toolIndex] as any).arguments = JSON.parse(args || "{}");
            } catch {
              // Keep the last successfully parsed arguments object until the JSON is complete.
            }
          };

          const finishCurrentBlock = () => {
            if (!currentType || contentIndex < 0) return;

            if (currentType === "toolCall") {
              const state = currentToolKey ? toolStates.get(currentToolKey) : undefined;
              if (state) parseToolArgs(state.contentIndex, state.args);
              stream.push({
                type: "toolcall_end",
                contentIndex,
                toolCall: output.content[contentIndex] as any,
                partial: output,
              });
            } else if (currentType === "text") {
              stream.push({
                type: "text_end",
                contentIndex,
                content: (output.content[contentIndex] as any).text,
                partial: output,
              });
            } else if (currentType === "thinking") {
              stream.push({
                type: "thinking_end",
                contentIndex,
                content: (output.content[contentIndex] as any).thinking,
                partial: output,
              });
            }

            currentType = null;
            currentToolKey = null;
            contentIndex = -1;
          };

          const startText = (delta: string) => {
            if (currentType !== "text") {
              finishCurrentBlock();
              output.content.push({ type: "text", text: "" });
              contentIndex = output.content.length - 1;
              currentType = "text";
              stream.push({ type: "text_start", contentIndex, partial: output });
            }
            (output.content[contentIndex] as any).text += delta;
            if (delta) stream.push({ type: "text_delta", contentIndex, delta, partial: output });
          };

          const startThinking = (delta: string) => {
            if (currentType !== "thinking") {
              finishCurrentBlock();
              output.content.push({ type: "thinking", thinking: "" });
              contentIndex = output.content.length - 1;
              currentType = "thinking";
              stream.push({ type: "thinking_start", contentIndex, partial: output });
            }
            (output.content[contentIndex] as any).thinking += delta;
            if (delta) stream.push({ type: "thinking_delta", contentIndex, delta, partial: output });
          };

          const startTool = (tc: any) => {
            const key = String(tc.index ?? tc.id ?? toolStates.size);
            if (currentType === "toolCall" && currentToolKey !== key) {
              finishCurrentBlock();
            } else if (currentType && currentType !== "toolCall") {
              finishCurrentBlock();
            }

            const id = tc.id || `tool_call_${toolStates.size}`;
            const name = tc.function?.name || "";
            const args = tc.function?.arguments || "";
            output.content.push({ type: "toolCall", id, name, arguments: {} });
            const toolContentIndex = output.content.length - 1;
            toolStates.set(key, { contentIndex: toolContentIndex, args });
            currentType = "toolCall";
            currentToolKey = key;
            contentIndex = toolContentIndex;
            stream.push({ type: "toolcall_start", contentIndex, partial: output });
            if (args) {
              parseToolArgs(contentIndex, args);
              stream.push({ type: "toolcall_delta", contentIndex, delta: args, partial: output });
            }
          };

          const appendToolArgs = (tc: any) => {
            const key = String(tc.index ?? currentToolKey ?? "0");
            let state = toolStates.get(key);
            if (!state || tc.id) {
              startTool(tc);
              if (tc.id) return;
              state = toolStates.get(key);
              if (!state) return;
            }

            if (currentType !== "toolCall" || currentToolKey !== key) {
              finishCurrentBlock();
              currentType = "toolCall";
              currentToolKey = key;
              contentIndex = state.contentIndex;
            }

            const name = tc.function?.name;
            if (name) (output.content[state.contentIndex] as any).name = name;
            const argsDelta = tc.function?.arguments || "";
            if (argsDelta) {
              state.args += argsDelta;
              parseToolArgs(state.contentIndex, state.args);
              stream.push({ type: "toolcall_delta", contentIndex: state.contentIndex, delta: argsDelta, partial: output });
            }
          };

          res.on("data", (chunk) => {
            if (closed) return;
            buffer += chunk.toString();
            const lines = buffer.split("\n");
            buffer = lines.pop() || "";

            for (const line of lines) {
              if (closed) return;
              if (!line.trim() || !line.startsWith("data: ")) continue;
              const dataStr = line.slice(6).trim();
              if (dataStr === "[DONE]") continue;

              try {
                const parsed = JSON.parse(dataStr);

                if (parsed.error) {
                  fail("error", parsed.error.message || JSON.stringify(parsed.error));
                  return;
                }

                const choice = parsed.choices?.[0];
                if (choice?.delta) {
                  const delta = choice.delta;

                  if (delta.tool_calls) {
                    for (const tc of delta.tool_calls) appendToolArgs(tc);
                  }

                  if (delta.thinking !== undefined) startThinking(delta.thinking || "");

                  if (delta.thinking_signature && contentIndex >= 0) {
                    (output.content[contentIndex] as any).thinkingSignature = delta.thinking_signature;
                  }

                  if (delta.content !== undefined && delta.content !== null) startText(delta.content || "");
                }

                if (parsed.usage) {
                  output.usage.input = parsed.usage.prompt_tokens || 0;
                  output.usage.output = parsed.usage.completion_tokens || 0;
                  output.usage.cacheRead = parsed.usage.prompt_tokens_details?.cache_read_input_tokens || 0;
                  output.usage.cacheWrite = parsed.usage.prompt_tokens_details?.cache_write_input_tokens || 0;
                  output.usage.totalTokens =
                    parsed.usage.total_tokens ||
                    output.usage.input + output.usage.output;
                  output.usage.cost = calculateCost(model, output.usage);
                }

                if (choice?.finish_reason) {
                  finishCurrentBlock();
                  output.stopReason = stopReasonFromFinishReason(choice.finish_reason);
                }
              } catch {
                // Ignore invalid JSON chunks. Parley streams valid JSON under normal operation;
                // this guard protects the session from malformed intermediary output.
              }
            }
          });

          res.on("end", () => {
            if (closed) return;
            finishCurrentBlock();
            if (output.usage.totalTokens === 0) {
              output.usage.cost = calculateCost(model, output.usage);
            }
            closed = true;
            stream.push({ type: "done", reason: output.stopReason as "stop" | "length" | "toolUse", message: output });
            stream.end();
          });

          res.on("error", (err) => fail("error", err.message));
        },
      );

      if (options?.timeoutMs && options.timeoutMs > 0) {
        req.setTimeout(options.timeoutMs, () => {
          req.destroy(new Error(`Request timed out after ${options.timeoutMs}ms`));
        });
      }

      req.on("error", (err) => fail(options?.signal?.aborted ? "aborted" : "error", err.message));

      if (options?.signal) {
        options.signal.addEventListener(
          "abort",
          () => {
            req.destroy();
            fail("aborted", "Request was aborted");
          },
          { once: true },
        );
      }

      req.write(reqData);
      req.end();
    } catch (error) {
      fail("error", error instanceof Error ? error.message : String(error));
    }
  })();
}