import { HttpChatTransport, type UIMessage, type UIMessageChunk } from "ai";

const record = (value: unknown): value is Record<string, unknown> => typeof value === "object" && value !== null && !Array.isArray(value);
export class AgentStreamDisconnectedError extends Error {}

export function createResponseRequest(messages: UIMessage[], options: {
  conversation?: string; model?: string; approvalMode?: string; capabilities?: string[];
  interactionId?: string; triggerContext?: unknown; taskContext?: unknown; topicId?: string;
}) {
  const latest = [...messages].reverse().find((message) => message.role === "user");
  const text = latest?.parts.filter((part) => part.type === "text").map((part) => part.text).join("\n").trim();
  if (!text) throw new Error("Responses 请求需要用户文本");
  return {
    input: [{ role: "user", content: [{ type: "input_text", text }] }], stream: true,
    ...(options.model ? { model: options.model } : {}),
    ...(options.conversation ? { conversation: options.conversation } : {}),
    tjwater: { ...(options.interactionId ? { interaction_id: options.interactionId } : {}), ...(options.topicId ? { topic_id: options.topicId } : {}), ...(latest?.id ? { client_message_id: latest.id } : {}), approval_mode: options.approvalMode ?? "request", capabilities: options.capabilities ?? [],
      ...(options.triggerContext ? { trigger_context: options.triggerContext } : {}),
      ...(options.taskContext ? { task_context: options.taskContext } : {}),
    },
  };
}

/** Keep UI rendering independent of the public Responses wire protocol. */
export function responsesToUiStream(stream: ReadableStream<Uint8Array>): ReadableStream<UIMessageChunk> {
  const decoder = new TextDecoder();
  let buffer = "";
  let responseId: string | undefined;
  let messageStarted = false;
  let messageId: string | undefined;
  let terminal = false;
  let sequence = -1;
  const texts = new Set<string>();
  const parse = (block: string, controller: TransformStreamDefaultController<UIMessageChunk>) => {
    if (block.length > 1024 * 1024) throw new Error("Responses 单个流事件过大");
    const lines = block.split(/\r?\n/);
    const data = lines.filter((line) => line.startsWith("data:")).map((line) => line.slice(5).trimStart()).join("\n");
    if (!data) return;
    if (terminal) throw new Error("Responses 流在结束后仍返回事件");
    const value: unknown = JSON.parse(data);
    if (!record(value) || typeof value.type !== "string" || !Number.isInteger(value.sequence_number) || (value.sequence_number as number) <= sequence) throw new Error("Responses 流事件格式不正确");
    sequence = value.sequence_number as number;
    const named = lines.find((line) => line.startsWith("event:"))?.slice(6).trim();
    if (named && named !== value.type) throw new Error("Responses 流事件类型不一致");
    if (value.type === "response.created") {
      if (responseId || !record(value.response) || typeof value.response.id !== "string") throw new Error("Responses 创建事件不正确");
      responseId = value.response.id;
      if (record(value.response.conversation) && typeof value.response.conversation.id === "string") {
        controller.enqueue({ type: "data-session_created", data: { session_id: value.response.conversation.id }, transient: true });
      }
      return;
    }
    if (!responseId) throw new Error("Responses 流缺少创建事件");
    const textId = () => {
      if (value.item_id !== messageId || !Number.isInteger(value.content_index) || (value.content_index as number) < 0) throw new Error("Responses 文本归属不正确");
      return `${messageId}:${String(value.content_index)}`;
    };
    if (value.type === "response.output_item.added" && record(value.item) && value.item.type === "message") {
      if (messageStarted || typeof value.item.id !== "string") throw new Error("Responses 回答消息不正确");
      messageStarted = true;
      messageId = value.item.id;
      controller.enqueue({ type: "start", messageId: value.item.id });
    } else if (value.type === "response.content_part.added" && record(value.part) && value.part.type === "output_text") {
      if (!messageStarted) throw new Error("Responses 流缺少回答消息");
      texts.add(textId());
      controller.enqueue({ type: "text-start", id: textId() });
    } else if (value.type === "response.output_text.delta") {
      if (!texts.has(textId()) || typeof value.delta !== "string") throw new Error("Responses 文本事件顺序不正确");
      controller.enqueue({ type: "text-delta", id: textId(), delta: value.delta });
    } else if (value.type === "response.output_text.done") {
      if (!texts.delete(textId())) throw new Error("Responses 文本结束事件不正确");
      controller.enqueue({ type: "text-end", id: textId() });
    } else if (value.type.startsWith("tjwater.")) {
      if (value.response_id !== responseId || !/^tjwater\.[a-z][a-z0-9_]*$/.test(value.type)) throw new Error("业务流事件归属不正确");
      controller.enqueue({ type: `data-${value.type.slice(8)}`, data: value.data,
        ...(typeof value.id === "string" ? { id: value.id } : {}), ...(value.transient === true ? { transient: true } : {}) } as UIMessageChunk);
    } else if (["response.completed", "response.failed", "response.incomplete"].includes(value.type)) {
      if (!record(value.response) || value.response.id !== responseId || texts.size) throw new Error("Responses 结束事件不正确");
      terminal = true;
      if (!messageStarted) controller.enqueue({ type: "start", messageId: responseId });
      if (value.type !== "response.completed") {
        const failure = value.response.error;
        controller.enqueue({ type: "error", errorText: record(failure) && typeof failure.message === "string" ? failure.message : "请求未完成或已中止" });
      }
      controller.enqueue({ type: "finish", finishReason: value.type === "response.completed" ? "stop" : "error" });
    }
  };
  const flushBlocks = (controller: TransformStreamDefaultController<UIMessageChunk>) => {
    let match: RegExpExecArray | null;
    while ((match = /\r?\n\r?\n/.exec(buffer))) {
      const block = buffer.slice(0, match.index);
      buffer = buffer.slice(match.index + match[0].length);
      parse(block, controller);
    }
    if (buffer.length > 1024 * 1024) throw new Error("Responses 单个流事件过大");
  };
  return stream.pipeThrough(new TransformStream<Uint8Array, UIMessageChunk>({
    transform(chunk, controller) { buffer += decoder.decode(chunk, { stream: true }); flushBlocks(controller); },
    flush(controller) {
      buffer += decoder.decode();
      flushBlocks(controller);
      if (buffer.trim()) parse(buffer, controller);
      if (!terminal) throw new AgentStreamDisconnectedError("Responses 流在完成前断开，正在恢复会话连接");
    },
  }));
}

export class ResponsesChatTransport<Message extends UIMessage> extends HttpChatTransport<Message> {
  protected processResponseStream(stream: ReadableStream<Uint8Array>) {
    return responsesToUiStream(stream);
  }
}