Agent-System/web/src/api/chatStream.ts

import type { ChatErrorInfo } from '../types/workbench';
import type { MessageExportCapability } from '../types/replyExport';
import { fallbackErrorClass } from './chatRetry';
import { mergeAuthHeaders } from './authHeaders';

export type StreamEvent = {
  id?: string;
  type: string;
  phase?: {
    phase_id?: string;
    message?: string;
    step_index?: number;
    total_steps?: number;
    progress?: number;
  };
  status?: { message?: string; intent?: string };
  delta?: string;
  error?: string;
  deliverable?: {
    task_key?: string;
    title?: string;
    file_path?: string;
    word_count?: number;
    quality_score?: number;
  };
  governance?: {
    doc_id?: string;
    action?: string;
    current_state?: string;
    is_locked?: boolean;
    author_claimed?: boolean;
    notification_title?: string;
    notification_msg?: string;
  };
  error_info?: {
    user_message?: string;
    code?: string;
    class?: string;
    phase_id?: string;
    upstream?: string;
    request_id?: string;
    session_id?: string;
    workspace_id?: string;
    actions?: string[];
  };
  turn?: {
    client_turn_id?: string;
    user_message_id?: number;
    status?: string;
    duplicate?: boolean;
  };
  message_id?: number;
  user_message_id?: number;
  content_revision?: number;
  content_sha256?: string;
  export_capability?: MessageExportCapability;
  done?: {
    session_id?: string;
    duration_ms?: number;
    total_events?: number;
    workspace_id?: string;
    message_id?: number;
    user_message_id?: number;
    content_revision?: number;
    content_sha256?: string;
    export_capability?: MessageExportCapability;
  };
};

type ErrorAction = ChatErrorInfo['actions'][number];

const VALID_ACTIONS: readonly ErrorAction[] = ['retry', 'continue', 'restart'];

function isErrorClass(value: unknown): value is ChatErrorInfo['class'] {
  return (
    value === 'transient' ||
    value === 'resumable' ||
    value === 'fatal' ||
    value === 'cancelled'
  );
}

function defaultActionsForClass(cls: ChatErrorInfo['class']): ChatErrorInfo['actions'] {
  switch (cls) {
    case 'transient':
      return ['retry'];
    case 'resumable':
      return ['continue', 'restart'];
    case 'fatal':
      return ['restart'];
    case 'cancelled':
      return ['retry'];
    default: {
      const _exhaustive: never = cls;
      return _exhaustive;
    }
  }
}

function parseActions(raw?: string[]): ChatErrorInfo['actions'] {
  if (!raw?.length) return [];
  return raw.filter((a): a is ErrorAction => (VALID_ACTIONS as readonly string[]).includes(a));
}

function parseFrame(frame: string): { event: StreamEvent | null; seq?: number } {
  if (!frame.trim()) return { event: null };

  let dataLine: string | undefined;
  let seq: number | undefined;
  for (const line of frame.split(/\r?\n/)) {
    if (line.startsWith('id:')) {
      const n = Number(line.slice(3).trim());
      if (Number.isFinite(n)) seq = n;
    }
    if (line.startsWith('data:')) {
      dataLine = line.slice(5).trim();
    }
  }
  // SSE comment / heartbeat frames (": ...") have no data — ignore.
  if (!dataLine) return { event: null, seq };

  try {
    return { event: JSON.parse(dataLine) as StreamEvent, seq };
  } catch (err) {
    console.warn('[sse] 忽略无法解析的帧:', dataLine.slice(0, 200), err);
    return { event: null, seq };
  }
}

/** Split complete SSE frames; accept both LF and CRLF blank-line delimiters. */
function takeSSEFrames(
  buffer: string,
  flush?: boolean,
): { frames: string[]; rest: string } {
  const frames: string[] = [];
  let start = 0;
  while (start < buffer.length) {
    const lfIdx = buffer.indexOf('\n\n', start);
    const crlfIdx = buffer.indexOf('\r\n\r\n', start);
    let delimAt = -1;
    let delimLen = 0;
    if (lfIdx === -1 && crlfIdx === -1) {
      break;
    }
    if (crlfIdx !== -1 && (lfIdx === -1 || crlfIdx <= lfIdx)) {
      delimAt = crlfIdx;
      delimLen = 4;
    } else {
      delimAt = lfIdx;
      delimLen = 2;
    }
    frames.push(buffer.slice(start, delimAt));
    start = delimAt + delimLen;
  }
  if (flush && start < buffer.length) {
    frames.push(buffer.slice(start));
    return { frames, rest: '' };
  }
  return { frames, rest: buffer.slice(start) };
}

export function parseSSEChunk(
  buffer: string,
  options?: { flush?: boolean },
): { events: StreamEvent[]; rest: string; lastSeq?: number } {
  const events: StreamEvent[] = [];
  if (!buffer) return { events, rest: '' };

  const { frames, rest } = takeSSEFrames(buffer, options?.flush);
  let lastSeq: number | undefined;
  for (const frame of frames) {
    const parsed = parseFrame(frame);
    if (parsed.seq !== undefined) lastSeq = parsed.seq;
    if (parsed.event) events.push(parsed.event);
  }
  return { events, rest, lastSeq };
}

export async function streamChat(
  body: Record<string, unknown>,
  handlers: {
    onEvent: (ev: StreamEvent) => void;
    onRequestId?: (id: string) => void;
  },
  signal?: AbortSignal,
  options?: { idempotencyKey?: string },
): Promise<void> {
  const headers = mergeAuthHeaders(
    {
      Accept: 'text/event-stream',
      ...(options?.idempotencyKey ? { 'Idempotency-Key': options.idempotencyKey } : {}),
    },
    'application/json',
  );
  const resp = await fetch('/api/v1/chat/stream', {
    method: 'POST',
    headers,
    body: JSON.stringify(body),
    signal,
  });
  const rid = resp.headers.get('X-Request-Id') || resp.headers.get('x-request-id');
  if (rid) handlers.onRequestId?.(rid);
  if (!resp.ok || !resp.body) {
    throw Object.assign(new Error(`HTTP ${resp.status}`), {
      httpStatus: resp.status,
      requestId: rid,
    });
  }
  const reader = resp.body.getReader();
  const decoder = new TextDecoder();
  let buf = '';
  while (true) {
    const { done, value } = await reader.read();
    if (done) break;
    buf += decoder.decode(value, { stream: true });
    const parsed = parseSSEChunk(buf);
    buf = parsed.rest;
    for (const ev of parsed.events) handlers.onEvent(ev);
  }
  if (buf) {
    const flushed = parseSSEChunk(buf, { flush: true });
    for (const ev of flushed.events) handlers.onEvent(ev);
  }
}

export function mapErrorInfo(ev: StreamEvent): ChatErrorInfo {
  const ei = ev.error_info ?? {};
  const userMessage = ei.user_message ?? ev.error ?? '发生未知错误';
  const code = ei.code ?? 'UNKNOWN_ERROR';

  const cls: ChatErrorInfo['class'] = isErrorClass(ei.class)
    ? ei.class
    : fallbackErrorClass({ workspaceId: ei.workspace_id });

  let actions = parseActions(ei.actions);
  if (!Array.isArray(ei.actions)) {
    actions = defaultActionsForClass(cls);
  }

  return {
    userMessage,
    code,
    class: cls,
    phaseId: ei.phase_id,
    upstream: ei.upstream,
    requestId: ei.request_id,
    sessionId: ei.session_id,
    workspaceId: ei.workspace_id,
    actions,
  };
}