569 lines
21 KiB
TypeScript
569 lines
21 KiB
TypeScript
import { apiFetch } from "@/core/api/fetch-client";
|
|
import { getBackendBaseURL } from "@/core/config";
|
|
|
|
import { resolveAllowMultiple } from "@/roundtable-planning/lib/clarification";
|
|
|
|
export type ThreadIdMap = Record<string, string>;
|
|
|
|
export interface SeatReadyEvent {
|
|
agentName: string;
|
|
threadId: string;
|
|
displayName: string;
|
|
description: string;
|
|
}
|
|
|
|
export interface CoordinatorReadyEvent {
|
|
mainAgentName: string;
|
|
threadId: string;
|
|
}
|
|
|
|
export interface InitMultiAgentRequest {
|
|
agentNames: string[];
|
|
mainAgentName: string;
|
|
model?: string;
|
|
/** Fires the moment each sub-agent's thread is ready (real completion order). */
|
|
onSeatReady?: (event: SeatReadyEvent) => void;
|
|
/** Fires when the coordinator's thread + agent have been created. */
|
|
onCoordinatorReady?: (event: CoordinatorReadyEvent) => void;
|
|
signal?: AbortSignal;
|
|
}
|
|
|
|
export interface InitMultiAgentResponse {
|
|
status: "success";
|
|
main_agent_name: string;
|
|
thread_ids: ThreadIdMap;
|
|
}
|
|
|
|
/**
|
|
* Open the /api/multi-agent/init SSE stream. Fires per-seat / coordinator
|
|
* callbacks as events arrive so the UI can light up roles individually, and
|
|
* resolves with the final {main_agent_name, thread_ids} once init_done lands.
|
|
* If the server emits an "error" frame mid-stream, the promise rejects.
|
|
*/
|
|
export async function initMultiAgent(
|
|
payload: InitMultiAgentRequest,
|
|
): Promise<InitMultiAgentResponse> {
|
|
// Timing telemetry for /api/multi-agent/init — enabled by default so we can
|
|
// see where time goes when users report "Step 2 init feels slow". Disable
|
|
// with `window.__MULTI_AGENT_INIT_TIMING__ = false` in devtools.
|
|
const timingEnabled =
|
|
typeof window === "undefined"
|
|
? false
|
|
: (window as unknown as { __MULTI_AGENT_INIT_TIMING__?: boolean }).__MULTI_AGENT_INIT_TIMING__ !== false;
|
|
const t0 = performance.now();
|
|
let tFirstByte = 0;
|
|
let tCoord = 0;
|
|
let seatCount = 0;
|
|
const tlog = (label: string) => {
|
|
if (!timingEnabled) return;
|
|
const elapsed = (performance.now() - t0).toFixed(0);
|
|
console.debug(`[init-timing] +${elapsed}ms ${label}`);
|
|
};
|
|
tlog("fetch.start");
|
|
|
|
const res = await apiFetch(`${getBackendBaseURL()}/api/multi-agent/init`, {
|
|
method: "POST",
|
|
headers: { "Content-Type": "application/json" },
|
|
body: JSON.stringify({
|
|
agent_names: payload.agentNames,
|
|
main_agent_name: payload.mainAgentName,
|
|
model: payload.model,
|
|
}),
|
|
signal: payload.signal,
|
|
});
|
|
tlog(`fetch.headers status=${res.status}`);
|
|
if (!res.ok || !res.body) {
|
|
const text = await res.text().catch(() => "");
|
|
throw new Error(`init multi-agent failed (${res.status}): ${text}`);
|
|
}
|
|
|
|
const reader = res.body.getReader();
|
|
const decoder = new TextDecoder();
|
|
let buffer = "";
|
|
let mainAgentName = "";
|
|
let threadIds: ThreadIdMap | null = null;
|
|
let aborted = false;
|
|
let errorDetail: string | null = null;
|
|
|
|
try {
|
|
while (true) {
|
|
const { done, value } = await reader.read();
|
|
if (done) break;
|
|
if (tFirstByte === 0) {
|
|
tFirstByte = performance.now();
|
|
tlog(`first-byte (TTFB-after-headers=${(tFirstByte - t0).toFixed(0)}ms)`);
|
|
}
|
|
buffer += decoder.decode(value, { stream: true });
|
|
const lines = buffer.split("\n");
|
|
buffer = lines.pop() ?? "";
|
|
|
|
for (const rawLine of lines) {
|
|
const line = rawLine.trimEnd();
|
|
if (!line.startsWith("data:")) continue;
|
|
const data = line.slice(5).trim();
|
|
if (!data) continue;
|
|
let parsed: unknown;
|
|
try {
|
|
parsed = JSON.parse(data);
|
|
} catch {
|
|
continue;
|
|
}
|
|
if (!parsed || typeof parsed !== "object") continue;
|
|
const obj = parsed as Record<string, unknown>;
|
|
const event = typeof obj.event === "string" ? obj.event : "";
|
|
switch (event) {
|
|
case "seat_ready": {
|
|
seatCount += 1;
|
|
tlog(`seat_ready #${seatCount} ${String(obj.agent_name ?? "?")}`);
|
|
payload.onSeatReady?.({
|
|
agentName: String(obj.agent_name ?? ""),
|
|
threadId: String(obj.thread_id ?? ""),
|
|
displayName: String(obj.display_name ?? ""),
|
|
description: String(obj.description ?? ""),
|
|
});
|
|
break;
|
|
}
|
|
case "coordinator_ready": {
|
|
tCoord = performance.now();
|
|
tlog(`coordinator_ready (since-first-seat=${(tCoord - tFirstByte).toFixed(0)}ms)`);
|
|
payload.onCoordinatorReady?.({
|
|
mainAgentName: String(obj.main_agent_name ?? ""),
|
|
threadId: String(obj.thread_id ?? ""),
|
|
});
|
|
break;
|
|
}
|
|
case "init_done": {
|
|
tlog(`init_done seats=${seatCount}`);
|
|
mainAgentName = String(obj.main_agent_name ?? "");
|
|
threadIds = (obj.thread_ids ?? {}) as ThreadIdMap;
|
|
break;
|
|
}
|
|
case "error": {
|
|
errorDetail = String(obj.detail ?? "unknown init error");
|
|
break;
|
|
}
|
|
default:
|
|
// Unknown event — ignore for forward-compat.
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
} catch (err) {
|
|
aborted = (err as Error)?.name === "AbortError" || payload.signal?.aborted === true;
|
|
if (!aborted) throw err;
|
|
}
|
|
|
|
tlog(`total ${(performance.now() - t0).toFixed(0)}ms (seats=${seatCount})`);
|
|
|
|
if (errorDetail) {
|
|
throw new Error(`init multi-agent failed: ${errorDetail}`);
|
|
}
|
|
if (aborted) {
|
|
throw new DOMException("init aborted", "AbortError");
|
|
}
|
|
if (!threadIds || !mainAgentName) {
|
|
throw new Error("init multi-agent stream ended without init_done");
|
|
}
|
|
return {
|
|
status: "success",
|
|
main_agent_name: mainAgentName,
|
|
thread_ids: threadIds,
|
|
};
|
|
}
|
|
|
|
export type StreamAgentType = "leader" | "special";
|
|
|
|
export type LeaderDispatch = [string, string];
|
|
|
|
/**
|
|
* High-level phases the SSE stream goes through. Used by the UI to show
|
|
* "what the backend is doing" while the user waits — most of the time before
|
|
* a first token is upstream/LLM latency, which feels like a stall otherwise.
|
|
*
|
|
* - `connecting` — fetch sent, no frames yet
|
|
* - `connected` — got the `event: metadata` frame (run_id assigned)
|
|
* - `streaming` — first AI text chunk arrived; tokens flowing
|
|
* - `tool_calling` — model emitted a tool_call chunk (e.g. agent_orchestration,
|
|
* present_files); usually means the run is about to interrupt
|
|
* - `tool_result` — a tool result message came back; brief, may go back to streaming
|
|
*/
|
|
export type StreamPhase = "connecting" | "connected" | "streaming" | "tool_calling" | "tool_result";
|
|
|
|
export interface StreamMultiAgentRequest {
|
|
agentType: StreamAgentType;
|
|
dAgentThreadId: ThreadIdMap;
|
|
agentName: string;
|
|
newMessage: string;
|
|
model?: string;
|
|
skillStopNames?: string[];
|
|
signal?: AbortSignal;
|
|
/** Fires for every non-status data frame the upstream emits. */
|
|
onUpstreamFrame?: (frame: unknown) => void;
|
|
/**
|
|
* Fires for every AI text chunk parsed out of LangGraph messages-tuple frames.
|
|
* `text` is the incremental delta (concatenate to rebuild the message);
|
|
* `messageId` is stable per AI message so callers can group deltas.
|
|
*/
|
|
onTextDelta?: (text: string, messageId: string) => void;
|
|
/**
|
|
* Fires whenever the inferred SSE phase changes. The UI can use this to
|
|
* show progressive status messages during the inevitable first-token wait.
|
|
* `name` is the tool name when phase is `tool_calling` (or empty otherwise).
|
|
*/
|
|
onPhaseChange?: (phase: StreamPhase, name?: string) => void;
|
|
}
|
|
|
|
export interface ClarificationPayload {
|
|
/** Raw question text from the model's ask_clarification args. */
|
|
question: string;
|
|
/** Optional category hint from the model (e.g. "missing_info"). */
|
|
clarificationType: string;
|
|
/** Free-text reason the model needs clarification. */
|
|
context: string;
|
|
/** Suggested answer chips (may be empty). */
|
|
options: unknown[];
|
|
/** Whether the user is allowed to type a custom answer. */
|
|
allowCustom: boolean;
|
|
/** When false, option cards are single-select and send on click. Omitted → inferred from clarificationType. */
|
|
allowMultiple?: boolean;
|
|
}
|
|
|
|
export interface StreamMultiAgentResult {
|
|
/** For leader runs: dispatched [agentName, task] tuples. Empty = consensus. */
|
|
dispatched: LeaderDispatch[];
|
|
/** For special runs: the completion status string. */
|
|
message: string | null;
|
|
/** Final assistant text emitted by the underlying agent. */
|
|
content: string;
|
|
/** Echoed agent_name of the agent that produced this run. */
|
|
agentName: string | null;
|
|
/**
|
|
* Populated when the leader called ask_clarification — the run ended without
|
|
* dispatching, and the model is asking the user a question instead. When
|
|
* present, callers should render `content` (or `clarification.question` as
|
|
* a fallback) as a question and collect a user reply rather than treating
|
|
* an empty `dispatched` as consensus.
|
|
*/
|
|
clarification: ClarificationPayload | null;
|
|
}
|
|
|
|
/**
|
|
* Identify the final status frame our backend appends after the LangGraph
|
|
* stream. Required keys are `status` (array for leader, string for special)
|
|
* and `content` (final assistant text); `agent_name` is also expected.
|
|
*/
|
|
function isFinalStatusFrame(value: unknown): value is {
|
|
status: LeaderDispatch[] | string;
|
|
content?: string;
|
|
agent_name?: string;
|
|
question?: string;
|
|
clarification_type?: string;
|
|
clarification_context?: string;
|
|
options?: unknown[];
|
|
allow_custom?: boolean;
|
|
allow_multiple?: boolean;
|
|
} {
|
|
if (!value || typeof value !== "object") return false;
|
|
if (!("status" in value) || !("content" in value)) return false;
|
|
const status = (value as { status: unknown }).status;
|
|
if (typeof status === "string") return true;
|
|
if (Array.isArray(status)) {
|
|
return status.every(
|
|
(item) =>
|
|
Array.isArray(item) && item.length === 2 && typeof item[0] === "string" && typeof item[1] === "string",
|
|
);
|
|
}
|
|
return false;
|
|
}
|
|
|
|
/**
|
|
* Extract an AI text delta from a LangGraph `messages-tuple` data frame.
|
|
* Frames look like `[chunk, metadata]`; an AI text chunk carries `type` in
|
|
* {"ai","AIMessageChunk"} and a `content` that is either a string or a list of
|
|
* structured content blocks. Returns null when the frame is not an AI text
|
|
* chunk (e.g., tool calls, tool results, value snapshots).
|
|
*/
|
|
function extractAiTextDelta(parsed: unknown): { text: string; id: string } | null {
|
|
if (!Array.isArray(parsed) || parsed.length < 1) return null;
|
|
const chunk = parsed[0];
|
|
if (!chunk || typeof chunk !== "object") return null;
|
|
const obj = chunk as Record<string, unknown>;
|
|
const type = String(obj.type ?? "").toLowerCase();
|
|
// Accept any AI-shaped message chunk LangChain might emit across versions:
|
|
// "ai" (older), "AIMessageChunk" (newer streaming), or plain "AIMessage".
|
|
if (type !== "ai" && type !== "aimessagechunk" && type !== "aimessage") return null;
|
|
const content = obj.content;
|
|
let text = "";
|
|
if (typeof content === "string") {
|
|
text = content;
|
|
} else if (Array.isArray(content)) {
|
|
for (const item of content) {
|
|
if (typeof item === "string") text += item;
|
|
else if (item && typeof item === "object") {
|
|
const block = item as Record<string, unknown>;
|
|
const blockText = block.text;
|
|
if (typeof blockText === "string") text += blockText;
|
|
}
|
|
}
|
|
}
|
|
if (!text) return null;
|
|
const id = typeof obj.id === "string" ? obj.id : "";
|
|
return { text, id };
|
|
}
|
|
|
|
/**
|
|
* Open an SSE connection to /api/multi-agent/run/stream and resolve once the
|
|
* stream completes. The final frame (`{status: ...}`) is interpreted; every
|
|
* other `data:` frame is forwarded to `onUpstreamFrame` so the caller can do
|
|
* its own rendering (we don't try to mimic LangGraph SDK semantics here).
|
|
*/
|
|
export async function streamMultiAgent(
|
|
req: StreamMultiAgentRequest,
|
|
): Promise<StreamMultiAgentResult> {
|
|
const res = await apiFetch(`${getBackendBaseURL()}/api/multi-agent/run/stream`, {
|
|
method: "POST",
|
|
headers: { "Content-Type": "application/json" },
|
|
body: JSON.stringify({
|
|
agent_type: req.agentType,
|
|
d_agent_thread_id: req.dAgentThreadId,
|
|
agent_name: req.agentName,
|
|
new_message: req.newMessage,
|
|
model: req.model,
|
|
skill_stop_names: req.skillStopNames ?? [],
|
|
}),
|
|
signal: req.signal,
|
|
});
|
|
|
|
if (!res.ok || !res.body) {
|
|
const text = await res.text().catch(() => "");
|
|
throw new Error(`run/stream failed (${res.status}): ${text}`);
|
|
}
|
|
|
|
const reader = res.body.getReader();
|
|
const decoder = new TextDecoder();
|
|
let buffer = "";
|
|
// Track the last `event:` line so logging can show which event a `data:`
|
|
// belongs to. The frontend doesn't actually dispatch by event name — every
|
|
// data frame goes through the same parser — but the label is invaluable
|
|
// for diagnosing why a sub-agent stream "looks empty".
|
|
let lastEvent = "";
|
|
let frameSeq = 0;
|
|
|
|
const debugLabel = `[stream:${req.agentType}:${req.agentName}]`;
|
|
const debugEnabled =
|
|
typeof window !== "undefined" &&
|
|
(window as unknown as { __MULTI_AGENT_DEBUG__?: boolean }).__MULTI_AGENT_DEBUG__ !== false;
|
|
|
|
const result: StreamMultiAgentResult = {
|
|
dispatched: [],
|
|
message: null,
|
|
content: "",
|
|
agentName: null,
|
|
clarification: null,
|
|
};
|
|
|
|
// Inferred SSE phase + a tiny helper so we only fire onPhaseChange when the
|
|
// value actually changes (the reader loop sees many frames per phase).
|
|
let currentPhase: StreamPhase = "connecting";
|
|
req.onPhaseChange?.("connecting");
|
|
const setPhase = (next: StreamPhase, name?: string) => {
|
|
if (next === currentPhase && !name) return;
|
|
currentPhase = next;
|
|
req.onPhaseChange?.(next, name);
|
|
};
|
|
|
|
while (true) {
|
|
const { done, value } = await reader.read();
|
|
if (done) break;
|
|
const chunk = decoder.decode(value, { stream: true });
|
|
buffer += chunk;
|
|
|
|
const lines = buffer.split("\n");
|
|
buffer = lines.pop() ?? "";
|
|
|
|
for (const rawLine of lines) {
|
|
const line = rawLine.trimEnd();
|
|
if (!line) continue;
|
|
if (line.startsWith("event:")) {
|
|
lastEvent = line.slice(6).trim();
|
|
// First metadata frame = run accepted by upstream. Move phase out
|
|
// of "connecting" so the UI can switch from "调度中" to "已连接".
|
|
if (lastEvent === "metadata" && currentPhase === "connecting") {
|
|
setPhase("connected");
|
|
}
|
|
continue;
|
|
}
|
|
if (!line.startsWith("data:")) continue;
|
|
const payload = line.slice(5).trim();
|
|
if (!payload) continue;
|
|
|
|
let parsed: unknown;
|
|
try {
|
|
parsed = JSON.parse(payload);
|
|
} catch {
|
|
if (debugEnabled) console.debug(debugLabel, `bad JSON in event=${lastEvent}:`, payload.slice(0, 200));
|
|
continue;
|
|
}
|
|
|
|
frameSeq += 1;
|
|
if (debugEnabled) console.debug(debugLabel, `frame#${frameSeq} event=${lastEvent}`, parsed);
|
|
|
|
// Phase inference from LangGraph messages-tuple frames. We only flip
|
|
// to tool_calling/tool_result on chunks that clearly indicate them;
|
|
// everything else with non-empty AI text means we're streaming.
|
|
if (lastEvent === "messages" && Array.isArray(parsed) && parsed.length > 0) {
|
|
const chunk = parsed[0];
|
|
if (chunk && typeof chunk === "object") {
|
|
const chunkObj = chunk as Record<string, unknown>;
|
|
const chunkType = String(chunkObj.type ?? "").toLowerCase();
|
|
const toolCalls = chunkObj.tool_calls;
|
|
const toolCallChunks = chunkObj.tool_call_chunks;
|
|
const content = chunkObj.content;
|
|
const hasToolCall =
|
|
(Array.isArray(toolCalls) && toolCalls.length > 0) ||
|
|
(Array.isArray(toolCallChunks) && toolCallChunks.length > 0);
|
|
if (chunkType === "tool") {
|
|
setPhase("tool_result");
|
|
} else if (hasToolCall) {
|
|
// Pick the most informative name we can find in either field.
|
|
const firstCall = Array.isArray(toolCalls) && toolCalls.length > 0
|
|
? (toolCalls[0] as Record<string, unknown>)
|
|
: Array.isArray(toolCallChunks) && toolCallChunks.length > 0
|
|
? (toolCallChunks[0] as Record<string, unknown>)
|
|
: null;
|
|
const toolName = firstCall && typeof firstCall.name === "string" ? firstCall.name : "";
|
|
setPhase("tool_calling", toolName);
|
|
} else if (
|
|
(chunkType === "ai" || chunkType === "aimessagechunk" || chunkType === "aimessage") &&
|
|
((typeof content === "string" && content.length > 0) || (Array.isArray(content) && content.length > 0))
|
|
) {
|
|
setPhase("streaming");
|
|
}
|
|
}
|
|
}
|
|
|
|
if (isFinalStatusFrame(parsed)) {
|
|
if (Array.isArray(parsed.status)) {
|
|
result.dispatched = parsed.status;
|
|
} else {
|
|
result.message = parsed.status;
|
|
// "clarification" status carries the leader's ask_clarification
|
|
// payload. Extract the structured fields so the page can render
|
|
// a question (or chip choices) instead of mistaking the run for
|
|
// a consensus.
|
|
if (parsed.status === "clarification") {
|
|
const clarificationType =
|
|
typeof parsed.clarification_type === "string" ? parsed.clarification_type : "";
|
|
result.clarification = {
|
|
question: typeof parsed.question === "string" ? parsed.question : "",
|
|
clarificationType,
|
|
context:
|
|
typeof parsed.clarification_context === "string" ? parsed.clarification_context : "",
|
|
options: Array.isArray(parsed.options) ? parsed.options : [],
|
|
allowCustom: parsed.allow_custom !== false,
|
|
allowMultiple: resolveAllowMultiple(
|
|
typeof parsed.allow_multiple === "boolean" ? parsed.allow_multiple : undefined,
|
|
clarificationType,
|
|
),
|
|
};
|
|
}
|
|
}
|
|
if (typeof parsed.content === "string") {
|
|
result.content = parsed.content;
|
|
}
|
|
if (typeof parsed.agent_name === "string") {
|
|
result.agentName = parsed.agent_name;
|
|
}
|
|
continue;
|
|
}
|
|
|
|
const delta = extractAiTextDelta(parsed);
|
|
if (delta) {
|
|
if (debugEnabled) console.debug(debugLabel, `→ delta (${delta.text.length} chars)`);
|
|
req.onTextDelta?.(delta.text, delta.id);
|
|
}
|
|
|
|
req.onUpstreamFrame?.(parsed);
|
|
}
|
|
}
|
|
|
|
if (debugEnabled) {
|
|
console.debug(debugLabel, `stream ended after ${frameSeq} frames; final content length=${result.content.length}`);
|
|
}
|
|
return result;
|
|
}
|
|
|
|
export interface SessionArtifactFile {
|
|
/** Virtual path under the sandbox, e.g. `mnt/user-data/outputs/foo.md`. */
|
|
path: string;
|
|
/** Bare filename (basename of `path`). */
|
|
name: string;
|
|
sizeBytes: number;
|
|
/** MIME type guessed from the extension, or null when unknown. */
|
|
mimeType: string | null;
|
|
/** ISO 8601 UTC timestamp of last modification. */
|
|
modifiedAt: string;
|
|
}
|
|
|
|
export interface SessionArtifactsResponse {
|
|
threadId: string;
|
|
files: SessionArtifactFile[];
|
|
}
|
|
|
|
/**
|
|
* List every file written to `/mnt/user-data/outputs/` for a thread. Used by
|
|
* the Step 3 final-report view to surface a thread's produced artifacts so the
|
|
* user can preview/download them via {@link buildArtifactDownloadUrl}.
|
|
*
|
|
* Returns an empty `files` array when the thread produced no artifacts (the
|
|
* backend treats a missing outputs directory as empty, not 404).
|
|
*/
|
|
export async function listSessionArtifacts(
|
|
threadId: string,
|
|
options: { signal?: AbortSignal } = {},
|
|
): Promise<SessionArtifactsResponse> {
|
|
const res = await apiFetch(
|
|
`${getBackendBaseURL()}/api/threads/${encodeURIComponent(threadId)}/artifacts`,
|
|
{ method: "GET", signal: options.signal },
|
|
);
|
|
if (!res.ok) {
|
|
const text = await res.text().catch(() => "");
|
|
throw new Error(`list artifacts failed (${res.status}): ${text}`);
|
|
}
|
|
const json = (await res.json()) as {
|
|
thread_id: string;
|
|
files: Array<{
|
|
path: string;
|
|
name: string;
|
|
size_bytes: number;
|
|
mime_type: string | null;
|
|
modified_at: string;
|
|
}>;
|
|
};
|
|
return {
|
|
threadId: json.thread_id,
|
|
files: (json.files ?? []).map((f) => ({
|
|
path: f.path,
|
|
name: f.name,
|
|
sizeBytes: f.size_bytes,
|
|
mimeType: f.mime_type,
|
|
modifiedAt: f.modified_at,
|
|
})),
|
|
};
|
|
}
|
|
|
|
/**
|
|
* Build a URL that previews (inline) or downloads (attachment) an artifact
|
|
* file. The `path` must be a virtual path returned by
|
|
* {@link listSessionArtifacts} (e.g. `mnt/user-data/outputs/foo.md`).
|
|
*/
|
|
export function buildArtifactDownloadUrl(
|
|
threadId: string,
|
|
virtualPath: string,
|
|
options: { download?: boolean } = {},
|
|
): string {
|
|
const segments = virtualPath.split("/").map((s) => encodeURIComponent(s)).join("/");
|
|
const base = `${getBackendBaseURL()}/api/threads/${encodeURIComponent(threadId)}/artifacts/${segments}`;
|
|
return options.download ? `${base}?download=true` : base;
|
|
}
|