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('\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 };

  return { event: JSON.parse(dataLine) as StreamEvent, seq };
}

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

  const frames = buffer.split('\n\n');
  const hasTrailingDelimiter = buffer.endsWith('\n\n');
  let processCount = hasTrailingDelimiter ? frames.length : Math.max(0, frames.length - 1);

  if (options?.flush && !hasTrailingDelimiter && frames.length > 0) {
    processCount = frames.length;
  }

  let lastSeq: number | undefined;
  for (let i = 0; i < processCount; i++) {
    const parsed = parseFrame(frames[i]);
    if (parsed.seq !== undefined) lastSeq = parsed.seq;
    if (parsed.event) events.push(parsed.event);
  }

  const rest = hasTrailingDelimiter || options?.flush ? '' : (frames[frames.length - 1] ?? '');
  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,
  };
}