290 lines
13 KiB
TypeScript
290 lines
13 KiB
TypeScript
/**
|
||
* 圆桌后台作业 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 { OptType, writeYLog } from "@/core/log/y-log";
|
||
|
||
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)),
|
||
});
|
||
const job = mapJob(await readJson<Record<string, unknown>>(res, "start job"));
|
||
writeYLog({
|
||
optType: OptType.other,
|
||
optDetail: "B1-CM助手-会商-运行",
|
||
url: BASE(),
|
||
reqParam: { draftId: payload.draftId ?? null, taskId: payload.taskId ?? null, model: payload.model },
|
||
returnParam: { id: job.id, status: job.status },
|
||
});
|
||
return 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 }),
|
||
});
|
||
const job = mapJob(await readJson<Record<string, unknown>>(res, "resume job"));
|
||
writeYLog({
|
||
optType: OptType.other,
|
||
optDetail: "B1-CM助手-会商-回答澄清续跑",
|
||
url: `${BASE()}/${jobId}/resume`,
|
||
reqParam: { jobId, taskId: taskId ?? null },
|
||
returnParam: { id: job.id, status: job.status },
|
||
});
|
||
return 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" });
|
||
const job = mapJob(await readJson<Record<string, unknown>>(res, "cancel job"));
|
||
writeYLog({
|
||
optType: OptType.edit,
|
||
optDetail: "B1-CM助手-会商-取消",
|
||
url: `${BASE()}/${jobId}/cancel`,
|
||
reqParam: { jobId, taskId: taskId ?? null },
|
||
returnParam: { id: job.id, status: job.status },
|
||
});
|
||
return 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();
|
||
}
|