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

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

import type { IntentSummary } from "./intent";

/**
 * Agent-recommendation API. Sits between Step 1 (intent clarification) and
 * Step 2 (multi-agent roundtable): given the IntentSummary and a candidate
 * agent pool, the built-in `roundtable-recommender` picks 2-6 agents most
 * relevant to the task and explains why.
 *
 * One-shot, stateless: no /init, no thread_id, each call spins up a fresh
 * backend-managed thread. Use {@link filterRecommendCandidates} to clean the
 * `/api/agents` list before passing it as `candidates`.
 */

/** Single agent pick returned by the recommender. */
export interface RecommendPick {
  agent_id: string;
  reason: string;
}

/** Final status frame the backend appends after the upstream stream finishes. */
export interface StreamRecommendResult {
  /**
   * `done` → picks parsed and validated, ready to populate the dialog as
   * default-checked selections.
   * `asking` → recommender failed to emit a usable [RECOMMEND_READY] block,
   * caller should apply its fallback list (typically the 6 roundtable-* seats).
   * `error` → upstream / network failure.
   */
  status: "done" | "asking" | "error";
  /** Full final AI text (includes the natural-language rationale). */
  content: string;
  /** Populated only when status === "done"; null otherwise. */
  picks: RecommendPick[] | null;
}

/** Phase mirror of streamIntent so the UI can reuse one loading-label mapper. */
export type RecommendStreamPhase = "connecting" | "connected" | "streaming";

export interface StreamRecommendRequest {
  /** Step 1 output. `objective` is required by the backend. */
  intent: IntentSummary;
  /** Pre-filtered candidate pool (use {@link filterRecommendCandidates}). */
  candidates: RecommendCandidate[];
  model?: string;
  signal?: AbortSignal;
  /** Fires whenever parsed natural-language rationale changes (excludes [RECOMMEND_READY] + JSON). */
  onRationaleUpdate?: (rationale: string) => void;
  /** Fires whenever newly complete pick objects are parsed from the streaming JSON block. */
  onPicksUpdate?: (picks: RecommendPick[]) => void;
  /** Fires for every raw AI text chunk (includes JSON); prefer onRationaleUpdate/onPicksUpdate in UI. */
  onTextDelta?: (text: string, messageId: string) => void;
  /** Fires whenever the inferred SSE phase changes. */
  onPhaseChange?: (phase: RecommendStreamPhase) => void;
}

/** Shape the backend expects in `candidates[]`. */
export interface RecommendCandidate {
  agent_id: string;
  name: string;
  description: string;
}

/**
 * Agent ids that must never enter the candidate pool. Centralized here so
 * every caller (recommend dialog, future tools) applies the same rule.
 *
 * - `roundtable-intent`     — Step 1 clarification specialist, no role in research.
 * - `roundtable-recommender`— the recommender itself, can't recommend itself.
 * - `roundtable-coordinator-*` — disposable Step 2 coordinator shells (DB
 *   accumulates one per session); meaningless as roundtable participants.
 */
const EXCLUDED_AGENT_ID_PREFIX = "roundtable-coordinator-";
const EXCLUDED_AGENT_IDS = new Set<string>(["roundtable-intent", "roundtable-recommender"]);

/**
 * Drop excluded ids from a raw `GET /api/agents` list and project each entry
 * to the slim shape the backend expects. Returns a fresh array; original
 * objects are not mutated.
 */
export function filterRecommendCandidates(agents: Agent[]): RecommendCandidate[] {
  const out: RecommendCandidate[] = [];
  for (const a of agents) {
    if (!a || !a.id) continue;
    if (EXCLUDED_AGENT_IDS.has(a.id)) continue;
    if (a.id.startsWith(EXCLUDED_AGENT_ID_PREFIX)) continue;
    out.push({
      agent_id: a.id,
      name: a.name ?? a.id,
      description: a.description ?? "",
    });
  }
  return out;
}

function isFinalRecommendFrame(value: unknown): value is {
  status: "done" | "asking" | "error";
  content?: string;
  picks?: RecommendPick[] | null;
} {
  if (!value || typeof value !== "object") return false;
  if (!("status" in value) || !("content" in value)) return false;
  const status = (value as { status: unknown }).status;
  return status === "done" || status === "asking" || status === "error";
}

const RECOMMEND_READY_MARKER = "[RECOMMEND_READY]";

/** Match one complete pick object inside a partially streamed JSON block. */
const PICK_OBJECT_PATTERN =
  /\{\s*"agent_id"\s*:\s*"((?:\\.|[^"\\])*)"\s*,\s*"reason"\s*:\s*"((?:\\.|[^"\\])*)"\s*\}/g;

function unescapeJsonString(raw: string): string {
  try {
    return JSON.parse(`"${raw}"`) as string;
  } catch {
    return raw.replace(/\\"/g, '"').replace(/\\\\/g, "\\");
  }
}

/**
 * Split accumulated recommender output into visible rationale text and any
 * pick objects that have fully arrived in the JSON section.
 */
export function parseRecommendStreamContent(raw: string): {
  rationale: string;
  picks: RecommendPick[];
} {
  const markerIdx = raw.indexOf(RECOMMEND_READY_MARKER);
  let rationale = markerIdx >= 0 ? raw.slice(0, markerIdx) : raw;
  rationale = rationale.replace(/\[(?:RECOMMEND(?:_(?:READY)?)?)?\]?$/i, "").trimEnd();

  const jsonSection =
    markerIdx >= 0 ? raw.slice(markerIdx + RECOMMEND_READY_MARKER.length) : "";

  const picks: RecommendPick[] = [];
  const seen = new Set<string>();
  PICK_OBJECT_PATTERN.lastIndex = 0;
  let match: RegExpExecArray | null;
  while ((match = PICK_OBJECT_PATTERN.exec(jsonSection)) !== null) {
    const agent_id = unescapeJsonString(match[1]).trim();
    const reason = unescapeJsonString(match[2]).trim();
    if (!agent_id || seen.has(agent_id)) continue;
    seen.add(agent_id);
    picks.push({ agent_id, reason });
  }

  return { rationale, picks };
}

/**
 * Extract an AI text delta from one LangGraph `messages-tuple` frame. Same
 * shape as the helper in intent.ts; duplicated rather than imported to keep
 * each helper independently readable.
 */
function extractAiTextDelta(parsed: unknown): { text: string; id: string } | null {
  if (!Array.isArray(parsed) || parsed.length < 1) return null;

  if (parsed.length >= 2 && parsed[1] && typeof parsed[1] === "object") {
    const meta = parsed[1] as Record<string, unknown>;
    const node = String(meta.langgraph_node ?? "");
    if (node.includes("TitleMiddleware")) return null;
    const tags = meta.tags;
    if (Array.isArray(tags) && tags.includes("middleware:title")) 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();
  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/recommend/stream` and resolve once the
 * upstream stream finishes. Each call is one recommendation round — the
 * recommender is stateless, so calling twice in a row produces two
 * independent runs.
 */
export async function streamRecommend(req: StreamRecommendRequest): Promise<StreamRecommendResult> {
  const res = await apiFetch(`${getBackendBaseURL()}/api/recommend/stream`, {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({
      intent: req.intent,
      candidates: req.candidates,
      model: req.model,
    }),
    signal: req.signal,
  });

  if (!res.ok || !res.body) {
    const text = await res.text().catch(() => "");
    throw new Error(`recommend/stream failed (${res.status}): ${text}`);
  }

  const reader = res.body.getReader();
  const decoder = new TextDecoder();
  let buffer = "";
  let lastEvent = "";

  let currentPhase: RecommendStreamPhase = "connecting";
  req.onPhaseChange?.("connecting");
  const setPhase = (next: RecommendStreamPhase) => {
    if (next === currentPhase) return;
    currentPhase = next;
    req.onPhaseChange?.(next);
  };

  const result: StreamRecommendResult = { status: "asking", content: "", picks: null };
  let rawContent = "";
  let lastRationale = "";
  let lastPickCount = 0;

  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 rawLine of lines) {
      const line = rawLine.trimEnd();
      if (!line) continue;
      if (line.startsWith("event:")) {
        lastEvent = line.slice(6).trim();
        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 {
        continue;
      }

      if (isFinalRecommendFrame(parsed)) {
        result.status = parsed.status;
        if (typeof parsed.content === "string") result.content = parsed.content;
        if (Array.isArray(parsed.picks)) {
          result.picks = parsed.picks as RecommendPick[];
        }
        continue;
      }

      if (lastEvent === "messages") {
        const delta = extractAiTextDelta(parsed);
        if (delta) {
          setPhase("streaming");
          req.onTextDelta?.(delta.text, delta.id);
          rawContent += delta.text;
          const { rationale, picks } = parseRecommendStreamContent(rawContent);
          if (rationale !== lastRationale) {
            lastRationale = rationale;
            req.onRationaleUpdate?.(rationale);
          }
          if (picks.length > lastPickCount) {
            lastPickCount = picks.length;
            req.onPicksUpdate?.(picks);
          }
        }
      }
    }
  }

  if (result.content) {
    const { rationale, picks } = parseRecommendStreamContent(result.content);
    if (rationale !== lastRationale) {
      req.onRationaleUpdate?.(rationale);
    }
    if (picks.length > lastPickCount) {
      req.onPicksUpdate?.(picks);
    }
  }

  return result;
}