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 { reportRoundtableError } from "./diagnostics-report";
import { seatModeToRunParams } from "../lib/seat-mode";
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(() => "");
    reportRoundtableError({
      stage: "job", event: "job_api_http_error",
      message: `后台作业接口「${ctx}」失败 (HTTP ${res.status}): ${text.slice(0, 500)}`,
      httpStatus: res.status,
    });
    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) {
  // 席位模式 → thinking/reasoning/subagent，与前端实时路径(/run/stream)同一份映射；
  // 缺省(未选模式)时不带这三字段 → 后端按 policy 默认(≈pro)，向后兼容。
  const seat = p.seatMode ? seatModeToRunParams(p.seatMode) : null;
  return {
    draftId: p.draftId ?? null,
    // taskId 非空 → task 作业（按 task 共享、不分权，报告写回独立 task-draft 存储）。
    ...(p.taskId ? { taskId: p.taskId } : {}),
    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",
    // 每席位技能识别强化映射 {agentId: bool}：沿用所选业务链条里各席位的开关；缺省 {}。
    // 后端 run_seat 按 agentId 查表，与各席位 agent 自身开关取 OR。
    seat_skill_directives: p.seatSkillDirectives ?? {},
    // 每席位推理深度覆盖映射 {agentId: SeatMode|null}：沿用所选业务链条里各席位单独配的推理深度；
    // 后端 run_seat 按 agentId 查表，覆盖优先、否则跟随作业级 seatMode（thinking/reasoning/subagent）。
    seat_modes: Object.fromEntries(
      Object.entries(p.seatModes ?? {}).filter(([, v]) => v != null) as [string, string][],
    ),
    // 研讨前并行取数开关（来源链条 gatherFirst）：true = 后台研讨前先并行取数再研讨。缺省 false。
    gather_first: p.gatherFirst ?? false,
    // 深链接业务码：后台总结报告据此做业务链完整性核对（写实业务如 rwfx 才生效）。空 → 不注入。
    business_code: p.businessCode ?? null,
    ...(seat
      ? {
          thinking_enabled: seat.thinking_enabled,
          reasoning_effort: seat.reasoning_effort,
          subagent_enabled: seat.subagent_enabled,
        }
      : {}),
  };
}

// ── 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 {
    /* 卸载途中同步异常也不能抛 */
  }
}

/** taskId 非空 → 追加 `task_id` 查询参（task 作业按 task 共享、不分权读取/控制）。 */
function taskQuery(taskId: string | null | undefined, sep: "?" | "&"): string {
  return taskId ? `${sep}task_id=${encodeURIComponent(taskId)}` : "";
}

/** 查作业快照。404 → null。`taskId` 非空走 task 共享读取（不分权）。 */
export async function getJob(jobId: string, taskId?: string | null): Promise<RoundtableJob | null> {
  const res = await apiFetch(`${BASE()}/${encodeURIComponent(jobId)}${taskQuery(taskId, "?")}`, { method: "GET" });
  if (res.status === 404) return null;
  return mapJob(await readJson<Record<string, unknown>>(res, "get job"));
}

/** 查某草稿最近一个作业。无 → null。`taskId` 非空走 task 共享读取（不分权）。 */
export async function getJobByDraft(draftId: string, taskId?: string | null): Promise<RoundtableJob | null> {
  const res = await apiFetch(
    `${BASE()}?draft_id=${encodeURIComponent(draftId)}${taskQuery(taskId, "&")}`,
    { method: "GET" },
  );
  const data = await readJson<Record<string, unknown> | null>(res, "get job by draft");
  return data ? mapJob(data) : null;
}

/** 回答澄清后续跑。`taskId` 非空 → task 共享续跑（不分权）。 */
export async function resumeJob(jobId: string, answer: string, taskId?: string | null): Promise<RoundtableJob> {
  const res = await apiFetch(`${BASE()}/${encodeURIComponent(jobId)}/resume${taskQuery(taskId, "?")}`, {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({ answer }),
  });
  return mapJob(await readJson<Record<string, unknown>>(res, "resume job"));
}

/** 取消作业。`taskId` 非空 → task 共享取消（不分权）。 */
export async function cancelJob(jobId: string, taskId?: string | null): Promise<RoundtableJob> {
  const res = await apiFetch(`${BASE()}/${encodeURIComponent(jobId)}/cancel${taskQuery(taskId, "?")}`, { 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, taskId?: string | null): () => void {
  const controller = new AbortController();

  void (async () => {
    try {
      const res = await apiFetch(`${BASE()}/${encodeURIComponent(jobId)}/stream${taskQuery(taskId, "?")}`, {
        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") {
            const detail = String(frame.detail ?? "unknown stream error");
            reportRoundtableError({
              stage: "job", event: "job_stream_error_frame",
              message: `后台作业进度流报错：${detail.slice(0, 500)}`,
              jobId, detail: { taskId },
            });
            handlers.onError?.(detail);
            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();
}