deerflow-code/frontend-web/src/roundtable-planning/api/roundtable-jobs.ts

/**
 * 圆桌后台作业 API 客户端（Phase 4）。
 *
 * 对接后端 `/api/roundtable-jobs`（Phase 3a）：启动「后台挂起」作业、查快照、订阅
 * SSE 进度、回答澄清续跑、取消。后端响应已是 camelCase（`RoundtableJobResponse`），
 * 故无需 snake_case 转换；唯一加工是给 `dispatchChain` 节点补 `avatarType`（后端不下发，
 * 由前端 `getAvatarTypeFor(agentId, order)` 派生，避免把配色表移植到后端而漂移）。
 *
 * SSE 走「后端轮询 job 表后下发整份快照」的模型（见后端注释）：每帧 `data: {event:
 * "progress", ...job}`，前端拿到的就是最新整份 `RoundtableJob`，直接渲染即可，无需做
 * 增量事件归并。
 */

import { apiFetch } from "@/core/api/fetch-client";
import { getBackendBaseURL } from "@/core/config";

import { getAvatarTypeFor } from "../lib/roundtable-constants";
import {
  isTerminalJobStatus,
  type DispatchChainNode,
  type JobDialogue,
  type RoundtableJob,
  type StartJobPayload,
} from "../lib/job-types";

const BASE = () => `${getBackendBaseURL()}/api/roundtable-jobs`;

async function readJson<T>(res: Response, ctx: string): Promise<T> {
  if (!res.ok) {
    const text = await res.text().catch(() => "");
    throw new Error(`${ctx} failed (${res.status}): ${text}`);
  }
  return (await res.json()) as T;
}

/** 给链路节点补 avatarType（后端不下发）。 */
function withAvatar(nodes: unknown): DispatchChainNode[] {
  if (!Array.isArray(nodes)) return [];
  return nodes.map((n, i) => {
    const node = (n ?? {}) as Partial<DispatchChainNode> & { agentId?: string; order?: number };
    const agentId = String(node.agentId ?? "");
    const order = typeof node.order === "number" ? node.order : i + 1;
    return {
      agentId,
      name: String(node.name ?? agentId),
      order,
      state: (node.state as DispatchChainNode["state"]) ?? "pending",
      avatarType: node.avatarType || getAvatarTypeFor(agentId, order - 1),
      ...(node.lastSpokeCycle != null ? { lastSpokeCycle: node.lastSpokeCycle } : {}),
    };
  });
}

/** 后端快照 → 前端 RoundtableJob（容错 + 补 avatarType）。 */
function mapJob(raw: Record<string, unknown>): RoundtableJob {
  return {
    id: String(raw.id ?? ""),
    draftId: (raw.draftId as string | null) ?? null,
    status: (raw.status as RoundtableJob["status"]) ?? "queued",
    phase: (raw.phase as RoundtableJob["phase"]) ?? "initializing",
    cycle: typeof raw.cycle === "number" ? raw.cycle : 0,
    dispatchChain: withAvatar(raw.dispatchChain),
    activeAgentId: (raw.activeAgentId as string | null) ?? null,
    consensusPercentage: typeof raw.consensusPercentage === "number" ? raw.consensusPercentage : 0,
    pendingClarification: (raw.pendingClarification as string | null) ?? null,
    orchestrationMode: (raw.orchestrationMode as RoundtableJob["orchestrationMode"]) ?? "recommend",
    chain: (raw.chain as RoundtableJob["chain"]) ?? null,
    orchestrationPlan: (raw.orchestrationPlan as RoundtableJob["orchestrationPlan"]) ?? null,
    error: (raw.error as string | null) ?? null,
    dialogues: Array.isArray(raw.dialogues) ? (raw.dialogues as JobDialogue[]) : [],
    step3: (raw.step3 as RoundtableJob["step3"]) ?? null,
    createdAt: (raw.createdAt as string | null) ?? null,
    updatedAt: (raw.updatedAt as string | null) ?? null,
  };
}

function toStartBody(p: StartJobPayload) {
  return {
    draftId: p.draftId ?? null,
    intent: p.intent,
    agents: p.agents,
    model: p.model,
    orchestrationMode: p.orchestrationMode ?? "recommend",
    chain: p.chain ?? null,
    orchestrationPlan: p.orchestrationPlan ?? null,
    seedMessage: p.seedMessage,
    threadIds: p.threadIds ?? null,
    coordinatorName: p.coordinatorName ?? "roundtable-coordinator",
  };
}

// ── CRUD / 控制 ───────────────────────────────────────────────────────────────

/** 启动一个后台作业。 */
export async function startJob(payload: StartJobPayload): Promise<RoundtableJob> {
  const res = await apiFetch(BASE(), {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify(toStartBody(payload)),
  });
  return mapJob(await readJson<Record<string, unknown>>(res, "start job"));
}

/**
 * 启动后台作业（``keepalive`` 版）—— 给 ``beforeunload``（刷新 / 关标签页）兜底用：
 * keepalive fetch 在页面卸载后浏览器仍会把请求发完（payload ≤64KB），保证「正在
 * 运行的研讨」在用户刷新 / 关页时也能转入后台挂起。fire-and-forget，不读响应；
 * 重复触发由后端「同草稿未终态作业幂等返回」兜住，不会重复建作业。
 */
export function startJobKeepalive(payload: StartJobPayload): void {
  try {
    void apiFetch(BASE(), {
      method: "POST",
      headers: { "Content-Type": "application/json" },
      body: JSON.stringify(toStartBody(payload)),
      keepalive: true,
    }).catch(() => {});
  } catch {
    /* 卸载途中同步异常也不能抛 */
  }
}

/** 查作业快照。404 → null。 */
export async function getJob(jobId: string): Promise<RoundtableJob | null> {
  const res = await apiFetch(`${BASE()}/${encodeURIComponent(jobId)}`, { method: "GET" });
  if (res.status === 404) return null;
  return mapJob(await readJson<Record<string, unknown>>(res, "get job"));
}

/** 查某草稿最近一个作业。无 → null。 */
export async function getJobByDraft(draftId: string): Promise<RoundtableJob | null> {
  const res = await apiFetch(`${BASE()}?draft_id=${encodeURIComponent(draftId)}`, { method: "GET" });
  const data = await readJson<Record<string, unknown> | null>(res, "get job by draft");
  return data ? mapJob(data) : null;
}

/** 回答澄清后续跑。 */
export async function resumeJob(jobId: string, answer: string): Promise<RoundtableJob> {
  const res = await apiFetch(`${BASE()}/${encodeURIComponent(jobId)}/resume`, {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({ answer }),
  });
  return mapJob(await readJson<Record<string, unknown>>(res, "resume job"));
}

/** 取消作业。 */
export async function cancelJob(jobId: string): Promise<RoundtableJob> {
  const res = await apiFetch(`${BASE()}/${encodeURIComponent(jobId)}/cancel`, { method: "POST" });
  return mapJob(await readJson<Record<string, unknown>>(res, "cancel job"));
}

// ── SSE 进度订阅 ─────────────────────────────────────────────────────────────

export interface StreamJobHandlers {
  /** 每收到一帧整份快照就回调（含首帧）。 */
  onProgress: (job: RoundtableJob) => void;
  /** 终态（done/error/cancelled）时回调一次（在对应的 onProgress 之后）。 */
  onDone?: (job: RoundtableJob) => void;
  /** 流错误 / 作业不存在。 */
  onError?: (detail: string) => void;
}

/**
 * 订阅作业 SSE 进度。返回 unsubscribe（中断连接）。
 *
 * 终态帧后后端会主动关闭流；本函数也在收到终态快照时调 onDone，调用方可据此 unsubscribe。
 */
export function streamJob(jobId: string, handlers: StreamJobHandlers): () => void {
  const controller = new AbortController();

  void (async () => {
    try {
      const res = await apiFetch(`${BASE()}/${encodeURIComponent(jobId)}/stream`, {
        method: "GET",
        signal: controller.signal,
      });
      if (!res.ok || !res.body) {
        handlers.onError?.(`stream failed (${res.status})`);
        return;
      }
      const reader = res.body.getReader();
      const decoder = new TextDecoder();
      let buffer = "";
      while (true) {
        const { done, value } = await reader.read();
        if (done) break;
        buffer += decoder.decode(value, { stream: true });
        const lines = buffer.split("\n");
        buffer = lines.pop() ?? "";
        for (const raw of lines) {
          const line = raw.trim();
          if (!line.startsWith("data:")) continue;
          const json = line.slice("data:".length).trim();
          if (!json) continue;
          let frame: Record<string, unknown>;
          try {
            frame = JSON.parse(json) as Record<string, unknown>;
          } catch {
            continue;
          }
          if (frame.event === "error") {
            handlers.onError?.(String(frame.detail ?? "unknown stream error"));
            continue;
          }
          const job = mapJob(frame);
          handlers.onProgress(job);
          if (isTerminalJobStatus(job.status)) handlers.onDone?.(job);
        }
      }
    } catch (err) {
      if (!controller.signal.aborted) {
        handlers.onError?.(err instanceof Error ? err.message : String(err));
      }
    }
  })();

  return () => controller.abort();
}