diegosouzapw--omniroute/open-sse/services/fusion.ts

/**
 * Fusion combo strategy — parallel panel + judge synthesis.
 *
 * A fusion combo fans the prompt out to every panel model in parallel, then a
 * configurable judge model synthesizes one final answer from all panel responses.
 *
 *   - quorum-grace collection caps the straggler penalty (the slowest model
 *     otherwise dominates wall time);
 *   - anonymized sources prevent judge brand-bias ("Source N" rather than model name);
 *   - degrades to a direct answer on a single survivor, 503 on total failure.
 *
 * Per OpenRouter's Fusion design, the judge does NOT merge — it analyzes
 * (consensus / contradictions / partial coverage / unique insights / blind spots)
 * then writes one answer grounded in that analysis. Most of fusion's quality lift
 * comes from this synthesis step.
 *
 * Ported from upstream decolua/9router (Daniil Schovkunov), adapted JS → TS and
 * wired through OmniRoute's existing combo schema (combo.config.judgeModel /
 * combo.config.fusionTuning).
 */
import { errorResponse, sanitizeErrorMessage } from "../utils/error.ts";
import { extractTextContent } from "../translator/helpers/geminiHelper.ts";
import type { ComboLogger, HandleSingleModel } from "./combo/types.ts";

// Fusion tuning. Overridable per-combo via combo.config.fusionTuning.
export const FUSION_DEFAULTS = {
  minPanel: 2, // answers needed before stragglers get a grace window
  stragglerGraceMs: 8000, // wait this long for laggards once quorum is reached
  panelHardTimeoutMs: 90000, // absolute cap so one hung model can't stall forever
} as const;

export type FusionTuning = {
  minPanel?: number;
  stragglerGraceMs?: number;
  panelHardTimeoutMs?: number;
};

type Body = Record<string, unknown>;

/**
 * Extract assistant text from a non-stream completion across formats
 * (OpenAI chat, Claude messages, Gemini, OpenAI Responses). Returns "" if none.
 * Panel responses are already translated to the client format by chatCore, so the
 * leaf content → string step reuses the translator's own extractTextContent.
 */
export function extractPanelText(json: unknown): string {
  if (!json || typeof json !== "object") return "";
  const j = json as Record<string, unknown>;

  // OpenAI chat completion
  const choices = j.choices as Array<Record<string, unknown>> | undefined;
  const choice = choices?.[0];
  if (choice) {
    const msg = (choice.message ?? choice.delta ?? {}) as Record<string, unknown>;
    const t = extractTextContent(msg.content);
    if (t.trim()) return t;
    if (typeof choice.text === "string" && choice.text.trim()) return choice.text;
  }

  // Claude messages (text blocks share OpenAI's {type:"text"} shape)
  const claudeText = extractTextContent(j.content);
  if (claudeText.trim()) return claudeText;

  // Gemini (parts carry .text without a type discriminator)
  const candidates = j.candidates as Array<Record<string, unknown>> | undefined;
  const parts = (candidates?.[0]?.content as Record<string, unknown> | undefined)?.parts as
    | Array<{ text?: unknown }>
    | undefined;
  if (Array.isArray(parts)) {
    const t = parts.map((p) => (typeof p?.text === "string" ? p.text : "")).join("");
    if (t.trim()) return t;
  }

  // OpenAI Responses API
  const output = j.output as Array<Record<string, unknown>> | undefined;
  if (Array.isArray(output)) {
    const t = output
      .flatMap((o) =>
        Array.isArray(o.content)
          ? (o.content as Array<{ text?: unknown }>).map((c) =>
              typeof c?.text === "string" ? c.text : ""
            )
          : []
      )
      .join("");
    if (t.trim()) return t;
  }

  return "";
}

/**
 * Append a synthesized user turn to whichever message array the request format uses.
 * Preserves the original conversation + system prompt so the judge has full context.
 */
export function appendUserTurn(body: Body, text: string): Body {
  const next: Body = { ...body };
  if (Array.isArray(body.messages)) {
    next.messages = [...(body.messages as unknown[]), { role: "user", content: text }];
  } else if (Array.isArray(body.input)) {
    next.input = [...(body.input as unknown[]), { role: "user", content: text }];
  } else if (Array.isArray(body.contents)) {
    next.contents = [
      ...(body.contents as unknown[]),
      { role: "user", parts: [{ text }] },
    ];
  } else {
    next.messages = [{ role: "user", content: text }];
  }
  return next;
}

/**
 * Build the judge directive. Sources are anonymized ("Source N") so the judge
 * weighs substance, not the reputation of a model brand.
 */
export function buildJudgePrompt(answers: Array<{ text: string }>): string {
  const panel = answers.map((a, i) => `[Source ${i + 1}]\n${a.text}`).join("\n\n");

  return [
    `You are the JUDGE in a model-fusion panel. ${answers.length} expert models independently answered the user's most recent request. Their responses are below, anonymized by source.`,
    "",
    "Do NOT mention that multiple models were used, and do NOT refer to the sources. Produce ONE authoritative final answer addressed directly to the user.",
    "",
    "First, internally analyze the panel along these dimensions: consensus (points most sources agree on — treat as higher-confidence), contradictions (where they disagree — resolve with your own judgment), partial coverage, unique insights only one source surfaced, and blind spots every source missed. Then write the best possible final answer grounded in that analysis — more complete and correct than any single response, with no filler.",
    "",
    "=== PANEL RESPONSES ===",
    panel,
    "=== END PANEL RESPONSES ===",
    "",
    "Now write the final answer to the user's original request.",
  ].join("\n");
}

type Sentinel = { __timeout?: true; __error?: unknown };

// Resolve a Response (or sentinel) within ms; the loser keeps running but is ignored.
function withTimeout(
  promise: Promise<Response>,
  ms: number
): Promise<Response | Sentinel> {
  return new Promise((resolve) => {
    const t = setTimeout(() => resolve({ __timeout: true }), ms);
    Promise.resolve(promise)
      .then((v) => {
        clearTimeout(t);
        resolve(v);
      })
      .catch((e) => {
        clearTimeout(t);
        resolve({ __error: e });
      });
  });
}

/**
 * Collect panel responses with quorum-grace: as soon as `minPanel` calls succeed,
 * start a short grace timer for the rest, then proceed with whatever arrived. This
 * caps the straggler penalty while still preferring a full panel when everyone is
 * fast. Bounded by a hard timeout.
 *
 * Returns a sparse array aligned to `calls` (undefined = not yet / dropped).
 */
export function collectPanel(
  calls: Array<Promise<Response | Sentinel>>,
  cfg: { minPanel: number; stragglerGraceMs: number; panelHardTimeoutMs: number }
): Promise<Array<Response | Sentinel | undefined>> {
  return new Promise((resolve) => {
    const out: Array<Response | Sentinel | undefined> = new Array(calls.length);
    let settled = 0;
    let ok = 0;
    let finished = false;
    let graceTimer: ReturnType<typeof setTimeout> | null = null;
    const finish = () => {
      if (finished) return;
      finished = true;
      clearTimeout(hardTimer);
      if (graceTimer) clearTimeout(graceTimer);
      resolve(out);
    };
    const hardTimer = setTimeout(finish, cfg.panelHardTimeoutMs);
    calls.forEach((p, i) => {
      Promise.resolve(p)
        .then((v) => {
          out[i] = v;
        })
        .catch((e) => {
          out[i] = { __error: e };
        })
        .finally(() => {
          settled++;
          const slot = out[i] as Response | undefined;
          if (slot && (slot as Response).ok) ok++;
          if (settled === calls.length) return finish();
          if (ok >= cfg.minPanel && !graceTimer) {
            graceTimer = setTimeout(finish, cfg.stragglerGraceMs);
          }
        });
    });
  });
}

export type HandleFusionChatOptions = {
  body: Body;
  models: string[];
  handleSingleModel: HandleSingleModel;
  log: ComboLogger;
  comboName?: string;
  judgeModel?: string | null;
  tuning?: FusionTuning | null;
};

/**
 * Handle a fusion combo: fan the prompt out to every panel model in parallel,
 * then a judge model synthesizes one final answer from all panel responses.
 *
 * Panel calls are forced non-streaming with tools stripped (the judge needs
 * complete prose to synthesize). The judge call keeps the client's original
 * stream flag + tools, so streaming and downstream tool use still work.
 *
 * Speed: quorum-grace collection caps the straggler penalty. Quality: the judge
 * runs the consensus/contradiction/blind-spot analysis before writing.
 *
 * Degrades gracefully: 0 panel answers → 503, exactly 1 → return it directly.
 */
export async function handleFusionChat({
  body,
  models,
  handleSingleModel,
  log,
  comboName,
  judgeModel,
  tuning,
}: HandleFusionChatOptions): Promise<Response> {
  const panel = Array.isArray(models) ? models.filter(Boolean) : [];
  if (panel.length === 0) {
    return errorResponse(400, "Fusion combo has no models");
  }

  // A single-model fusion has nothing to fuse — just answer directly.
  if (panel.length === 1) {
    return handleSingleModel(body, panel[0]);
  }

  const cfg = {
    minPanel: tuning?.minPanel ?? FUSION_DEFAULTS.minPanel,
    stragglerGraceMs: tuning?.stragglerGraceMs ?? FUSION_DEFAULTS.stragglerGraceMs,
    panelHardTimeoutMs: tuning?.panelHardTimeoutMs ?? FUSION_DEFAULTS.panelHardTimeoutMs,
  };
  const minPanel = Math.min(Math.max(2, cfg.minPanel), panel.length);
  const judge = judgeModel && judgeModel.trim() ? judgeModel.trim() : panel[0];
  log.info(
    "FUSION",
    `Combo "${comboName ?? ""}" | panel=${panel.length} [${panel.join(", ")}] | judge=${judge} | quorum=${minPanel}`
  );

  // 1. Fan out to the panel in parallel: non-streaming, tools stripped (we want prose).
  const { tools: _tools, tool_choice: _tc, ...rest } = body;
  void _tools;
  void _tc;
  const panelBody: Body = { ...rest, stream: false };
  const t0 = Date.now();
  const calls = panel.map((m) =>
    withTimeout(handleSingleModel(panelBody, m), cfg.panelHardTimeoutMs)
  );
  const settled = await collectPanel(calls, { ...cfg, minPanel });
  log.info("FUSION", `fan-out collected in ${Date.now() - t0}ms`);

  // 2. Collect successful answers.
  const answers: Array<{ model: string; text: string }> = [];
  const rateLimited: string[] = [];
  for (let i = 0; i < settled.length; i++) {
    const res = settled[i];
    const model = panel[i];
    if (!res) {
      log.warn("FUSION", `Panel ${model} dropped (straggler/timeout)`);
      continue;
    }
    const sentinel = res as Sentinel;
    if (sentinel.__timeout) {
      log.warn("FUSION", `Panel ${model} timed out`);
      continue;
    }
    if (sentinel.__error) {
      log.warn("FUSION", `Panel ${model} threw`, {
        error: sanitizeErrorMessage(sentinel.__error as Error),
      });
      continue;
    }
    const resp = res as Response;
    if (!resp.ok) {
      if (resp.status === 429) {
        rateLimited.push(model);
        log.warn("FUSION", `Panel ${model} rate-limited`, { status: resp.status });
      } else {
        log.warn("FUSION", `Panel ${model} failed`, { status: resp.status });
      }
      continue;
    }
    try {
      const json = await resp.clone().json();
      const text = extractPanelText(json);
      if (text) {
        answers.push({ model, text });
        log.info("FUSION", `Panel ${model} ok (${text.length} chars)`);
      } else {
        log.warn("FUSION", `Panel ${model} returned empty content`);
      }
    } catch (e) {
      log.warn("FUSION", `Panel ${model} unparseable`, {
        error: sanitizeErrorMessage(e as Error),
      });
    }
  }

  // 3. Degrade gracefully when the panel is too thin to fuse.
  if (answers.length === 0) {
    const detail =
      rateLimited.length > 0
        ? `${rateLimited.length} models rate-limited, ${panel.length - rateLimited.length} failed`
        : `all ${panel.length} models failed`;
    log.warn("FUSION", `No live models: ${detail}`);
    return errorResponse(503, `All fusion panel models failed (${detail})`);
  }
  if (answers.length === 1) {
    log.info(
      "FUSION",
      `Only ${answers[0].model} succeeded — answering directly (no fusion)`
    );
    return handleSingleModel(body, answers[0].model);
  }

  // 4. Judge analyzes + writes one final answer (streams to client if requested).
  const judgeBody = appendUserTurn(body, buildJudgePrompt(answers));
  log.info("FUSION", `Judging ${answers.length} answers with ${judge}`);
  return handleSingleModel(judgeBody, judge);
}