mini-openclaw-v2/src/lib/chat-stream.ts

/** Stateful SSE decoder: TCP chunks are not event boundaries. */
export interface ChatEvent { event: string; data: Record<string, unknown> }

export class ChatEventDecoder {
  private buffer = "";
  private event = "message";
  private data: string[] = [];
  push(text: string, flush = false): ChatEvent[] {
    this.buffer += text;
    const result: ChatEvent[] = [];
    const lines = this.buffer.split("\n");
    this.buffer = lines.pop() || "";
    if (flush) { if (this.buffer) lines.push(this.buffer); lines.push(""); this.buffer = ""; }
    for (const raw of lines) {
      const line = raw.endsWith("\r") ? raw.slice(0, -1) : raw;
      if (!line) {
        if (this.data.length) {
          let parsed: unknown;
          try { parsed = JSON.parse(this.data.join("\n")); }
          catch { throw new Error("回答数据格式异常，请重试；已接收的内容会保留。"); }
          if (!parsed || typeof parsed !== "object" || Array.isArray(parsed)) throw new Error("回答事件格式不受支持");
          result.push({event: this.event, data: parsed as Record<string, unknown>});
        }
        this.event = "message"; this.data = [];
      } else if (line.startsWith("event:")) {
        this.event = line.slice(6).trim();
      } else if (line.startsWith("data:")) {
        this.data.push(line.slice(5).replace(/^ /, ""));
      }
    }
    return result;
  }
}

/** No automatic request replay: a tool may already have executed. */
export async function* readChatStream(url: string, body: unknown, options: {
  fetcher?: typeof fetch; idleMs?: number; totalMs?: number;
} = {}): AsyncGenerator<ChatEvent> {
  const controller = new AbortController();
  const fetcher = options.fetcher || fetch;
  const idleMs = options.idleMs ?? 90_000;
  let timedOut = false, receivedDone = false, receivedError = false, answerSeen = false;
  let reader: ReadableStreamDefaultReader<Uint8Array> | undefined;
  const expire = () => { timedOut = true; controller.abort(); };
  let idle = setTimeout(expire, idleMs);
  const total = setTimeout(expire, options.totalMs ?? 180_000);
  const touch = () => { clearTimeout(idle); idle = setTimeout(expire, idleMs); };
  const decoder = new TextDecoder();
  const parser = new ChatEventDecoder();
  const annotate = (events: ChatEvent[]) => {
    const normalized: ChatEvent[] = [];
    for (const event of events) {
      if (event.event === "token" && typeof event.data.content === "string" && event.data.content.trim()) answerSeen = true;
      if (event.event === "done") {
        if (!answerSeen) {
          if (typeof event.data.content !== "string" || !event.data.content.trim()) throw new Error("模型没有返回回答正文，请检查聊天模型配置后重试。");
          normalized.push({event:"token",data:{content:event.data.content}});
          answerSeen = true;
        }
        receivedDone = true;
      }
      if (event.event === "error") receivedError = true;
      normalized.push(event);
    }
    return normalized;
  };
  try {
    const response = await fetcher(url, {method:"POST",headers:{"Content-Type":"application/json"},body:JSON.stringify(body),signal:controller.signal});
    if (!response.ok) throw new Error(`聊天服务请求失败（HTTP ${response.status}），请检查当前后端连接。`);
    if (!response.headers.get("content-type")?.includes("text/event-stream")) throw new Error("聊天服务未返回事件流，请检查后端地址。");
    reader = response.body?.getReader();
    if (!reader) throw new Error("聊天服务未返回数据流。");
    touch();
    while (true) {
      const {done,value} = await reader.read();
      if (done) break;
      for (const event of annotate(parser.push(decoder.decode(value,{stream:true})))) { touch(); yield event; }
      if (receivedError) return;
    }
    for (const event of annotate(parser.push(decoder.decode(),true))) { touch(); yield event; }
    if (!receivedDone && !receivedError) throw new Error("回答连接提前中断；已接收的内容已保留，请检查后重试。");
  } catch (error) {
    if (timedOut) {
      if (receivedDone) return; // An optional title request must not invalidate a delivered answer.
      throw new Error("回答等待超时；请检查模型服务后重试。不会自动重复执行工具。");
    }
    throw error;
  } finally {
    clearTimeout(idle);clearTimeout(total);controller.abort();
    if (reader) {try {await reader.cancel();}catch {} reader.releaseLock();}
  }
}