ui/src/lib/flow-api.ts - main - alexefan136/flowstack

ui/src/lib/flow-api.ts
501 строка14 KB
Alexander Efanov
upd fix
04 авг 2026, 14:09
04 авг 2026, 14:09f63ab86
// src/lib/flow-api.ts
//
// Flows API — CRUD + execution через Gateway → Engine.
//
// Два типа flows:
//   1. DB Flows (UUID) — CRUD + run через /api/v1/flows/...
//   2. Predefined Flows (string ID) — run через /api/v1/runs
//
// DB Flows endpoints (Gateway):
//   GET    /api/v1/flows              — список flows
//   POST   /api/v1/flows              — создать flow
//   GET    /api/v1/flows/{id}         — получить flow
//   PATCH  /api/v1/flows/{id}         — обновить flow
//   DELETE /api/v1/flows/{id}         — удалить flow
//   GET    /api/v1/flows/stats        — статистика
//   POST   /api/v1/flows/validate     — валидировать граф
//   POST   /api/v1/flows/{id}/run     — запустить DB flow (SSE при stream=true)
//
// Predefined Flows endpoints (Gateway):
//   GET    /api/v1/runs/flows         — список predefined flows
//   POST   /api/v1/runs               — запустить predefined flow (SSE при stream=true)

import type { AgentEvent } from "./api";
import type {
  Flow,
  PredefinedFlow,
  PredefinedFlowRunResult,
} from "./flow-types";
import { PREDEFINED_FLOWS_CONFIG } from "./flow-types";
import { getAuthHeaders } from "./stores/auth-store";

// ============================================================================
// Constants
// ============================================================================

const API_BASE = "http://localhost:8080";

// ============================================================================
// Helpers
// ============================================================================

function getFlowHeaders(
  extra?: Record<string, string>,
): Record<string, string> {
  return {
    "Content-Type": "application/json",
    ...getAuthHeaders(),
    ...extra,
  };
}

async function extractError(res: Response): Promise<string> {
  try {
    const text = await res.text();
    const data = JSON.parse(text);
    if (typeof data === "string") return data;
    if (data.detail) {
      if (typeof data.detail === "string") return data.detail;
      return (
        data.detail.message || data.detail.error || JSON.stringify(data.detail)
      );
    }
    if (data.error) {
      if (typeof data.error === "string") return data.error;
      return data.error.message || JSON.stringify(data.error);
    }
    return data.message || JSON.stringify(data);
  } catch {
    return `Ошибка ${res.status}`;
  }
}

// ============================================================================
// Types
// ============================================================================

export interface CreateFlowPayload {
  name: string;
  description?: string;
  nodes?: unknown[];
  edges?: unknown[];
}

export interface UpdateFlowPayload {
  name?: string;
  description?: string;
  nodes?: unknown[];
  edges?: unknown[];
}

export interface FlowRunResponse {
  flow_id: string;
  status: string;
  output: Record<string, unknown>;
  steps?: unknown[];
  tokens_used?: number;
  duration_ms?: number;
  error?: string;
}

// ============================================================================
// DB Flows CRUD (Gateway → Engine, UUID-based)
// ============================================================================

/**
 * Список всех DB flows в workspace.
 */
export async function listUserFlows(): Promise<Flow[]> {
  const res = await fetch(`${API_BASE}/api/v1/flows`, {
    headers: getFlowHeaders(),
  });
  if (!res.ok) {
    throw new Error(`Failed to fetch flows: ${await extractError(res)}`);
  }
  return res.json();
}

/**
 * Получить DB flow по UUID.
 */
export async function getUserFlow(id: string): Promise<Flow> {
  const res = await fetch(`${API_BASE}/api/v1/flows/${id}`, {
    headers: getFlowHeaders(),
  });
  if (!res.ok) {
    throw new Error(`Failed to fetch flow: ${await extractError(res)}`);
  }
  return res.json();
}

/**
 * Создать новый DB flow.
 */
export async function createUserFlow(
  payload: CreateFlowPayload,
): Promise<Flow> {
  const res = await fetch(`${API_BASE}/api/v1/flows`, {
    method: "POST",
    headers: getFlowHeaders(),
    body: JSON.stringify(payload),
  });
  if (!res.ok) {
    throw new Error(`Failed to create flow: ${await extractError(res)}`);
  }
  return res.json();
}

/**
 * Обновить DB flow (частичное обновление).
 * PATCH — gateway поддерживает только PATCH.
 */
export async function updateUserFlow(
  id: string,
  payload: UpdateFlowPayload,
): Promise<Flow> {
  const res = await fetch(`${API_BASE}/api/v1/flows/${id}`, {
    method: "PATCH",
    headers: getFlowHeaders(),
    body: JSON.stringify(payload),
  });
  if (!res.ok) {
    throw new Error(`Failed to update flow: ${await extractError(res)}`);
  }
  return res.json();
}

/**
 * Удалить DB flow.
 */
export async function deleteUserFlow(id: string): Promise<void> {
  const res = await fetch(`${API_BASE}/api/v1/flows/${id}`, {
    method: "DELETE",
    headers: getFlowHeaders(),
  });
  if (!res.ok) {
    throw new Error(`Failed to delete flow: ${await extractError(res)}`);
  }
}

// ============================================================================
// DB Flows — Validation & Stats
// ============================================================================

/**
 * Валидировать граф flow без создания.
 */
export async function validateFlow(payload: CreateFlowPayload): Promise<{
  valid: boolean;
  nodes_count: number;
  edges_count: number;
}> {
  const res = await fetch(`${API_BASE}/api/v1/flows/validate`, {
    method: "POST",
    headers: getFlowHeaders(),
    body: JSON.stringify(payload),
  });
  if (!res.ok) {
    throw new Error(`Failed to validate flow: ${await extractError(res)}`);
  }
  return res.json();
}

/**
 * Статистика по DB flows.
 */
export async function fetchFlowStats(): Promise<{
  total_flows: number;
  by_status: Record<string, number>;
}> {
  const res = await fetch(`${API_BASE}/api/v1/flows/stats`, {
    headers: getFlowHeaders(),
  });
  if (!res.ok) {
    throw new Error(`Failed to fetch flow stats: ${await extractError(res)}`);
  }
  return res.json();
}

// ============================================================================
// DB Flows — Execution (POST /api/v1/flows/{uuid}/run)
// ============================================================================

/**
 * Запустить DB flow (non-streaming).
 */
export async function runUserFlow(
  flow: Flow,
  input: Record<string, unknown>,
): Promise<FlowRunResponse> {
  const res = await fetch(`${API_BASE}/api/v1/flows/${flow.id}/run`, {
    method: "POST",
    headers: getFlowHeaders(),
    body: JSON.stringify({ input, stream: false }),
  });

  if (!res.ok) {
    throw new Error(`Run failed: ${await extractError(res)}`);
  }

  return res.json();
}

/**
 * Запустить DB flow (SSE streaming).
 */
export async function* streamUserFlowRun(
  flow: Flow,
  input: Record<string, unknown>,
  signal?: AbortSignal,
): AsyncGenerator<AgentEvent, void, unknown> {
  const res = await fetch(`${API_BASE}/api/v1/flows/${flow.id}/run`, {
    method: "POST",
    headers: getFlowHeaders({ Accept: "text/event-stream" }),
    body: JSON.stringify({ input, stream: true }),
    signal,
  });

  if (!res.ok) {
    throw new Error(`Run failed: ${await extractError(res)}`);
  }

  if (!res.body) {
    throw new Error("No response body for streaming");
  }

  yield* parseSSEStream(res.body);
}

// ============================================================================
// Predefined Flows (POST /api/v1/runs, string ID из FLOW_REGISTRY)
// ============================================================================

/**
 * Получить список predefined flows.
 *
 * Сначала пытается загрузить с бэкенда GET /api/v1/runs/flows,
 * при недоступности — fallback на локальный PREDEFINED_FLOWS_CONFIG.
 */
export async function listPredefinedFlows(): Promise<PredefinedFlow[]> {
  try {
    const res = await fetch(`${API_BASE}/api/v1/runs/flows`, {
      headers: getFlowHeaders(),
    });
    if (res.ok) {
      const data = await res.json();
      // Обогащаем локальными inputFields/features из PREDEFINED_FLOWS_CONFIG
      return (data as PredefinedFlow[]).map((flow) => ({
        ...flow,
        inputFields: PREDEFINED_FLOWS_CONFIG[flow.id]?.inputFields,
        features: PREDEFINED_FLOWS_CONFIG[flow.id]?.features,
      }));
    }
  } catch (err) {
    console.warn("[PredefinedFlows] API unavailable, using local config:", err);
  }
  // Fallback: локальная конфигурация
  return Object.values(PREDEFINED_FLOWS_CONFIG);
}

/**
 * Запустить predefined flow (non-streaming).
 * ✅ POST /api/v1/runs — flow_id строковый ("research"), НЕ UUID.
 */
export async function runPredefinedFlow(
  flowId: string,
  input: Record<string, unknown>,
): Promise<PredefinedFlowRunResult> {
  const res = await fetch(`${API_BASE}/api/v1/runs`, {
    method: "POST",
    headers: getFlowHeaders(),
    body: JSON.stringify({ flow_id: flowId, input, stream: false }),
  });

  if (!res.ok) {
    throw new Error(`Run failed: ${await extractError(res)}`);
  }

  return res.json();
}

/**
 * Запустить predefined flow (SSE streaming).
 * ✅ POST /api/v1/runs — flow_id строковый ("research"), НЕ UUID.
 *
 * События SSE от Engine:
 *   agent_start, agent_message, agent_done, ideas_generated, flow_done, error
 */
export async function* streamPredefinedFlowRun(
  flowId: string,
  input: Record<string, unknown>,
  signal?: AbortSignal,
): AsyncGenerator<AgentEvent, void, unknown> {
  const res = await fetch(`${API_BASE}/api/v1/runs`, {
    method: "POST",
    headers: getFlowHeaders({ Accept: "text/event-stream" }),
    body: JSON.stringify({ flow_id: flowId, input, stream: true }),
    signal,
  });

  if (!res.ok) {
    throw new Error(`Run failed: ${await extractError(res)}`);
  }

  if (!res.body) {
    throw new Error("No response body for streaming");
  }

  yield* parseSSEStream(res.body);
}

// ============================================================================
// SSE Stream Parser (shared)
// ============================================================================

/**
 * Универсальный парсер SSE потока из ReadableStream.
 * Yield-ает AgentEvent для каждого data-блока.
 */
async function* parseSSEStream(
  body: ReadableStream<Uint8Array>,
): AsyncGenerator<AgentEvent, void, unknown> {
  const reader = body.getReader();
  const decoder = new TextDecoder("utf-8");
  let buffer = "";

  try {
    while (true) {
      const { done, value } = await reader.read();

      if (done) {
        if (buffer.trim()) {
          yield* parseSSEBlock(buffer);
        }
        break;
      }

      buffer += decoder.decode(value, { stream: true });

      const parts = buffer.split(/\r?\n\r?\n/);
      buffer = parts.pop() || "";

      for (const block of parts) {
        if (!block.trim()) continue;
        yield* parseSSEBlock(block);
      }
    }
  } finally {
    reader.releaseLock();
  }
}

/**
 * Парсит один SSE блок в AgentEvent.
 *
 * Поддерживает форматы Engine:
 * - `event: <type>\ndata: <json>` — именованные события
 * - `data: <json>` — без имени (type из JSON)
 */
function* parseSSEBlock(block: string): Generator<AgentEvent, void, unknown> {
  const lines = block.split(/\r?\n/);
  let eventName = "";
  let eventData = "";

  for (const line of lines) {
    if (line.startsWith("event:")) {
      eventName = line.slice(6).trim();
    } else if (line.startsWith("data:")) {
      eventData += line.slice(5).trimStart();
    }
  }

  if (!eventData || eventData === "[DONE]") return;

  try {
    const data = JSON.parse(eventData) as Record<string, unknown>;
    const type = eventName || (data.type as string) || "content";

    switch (type) {
      case "flow_start":
      case "agent_start":
      case "run_start":
        yield {
          type: "agent_start",
          agent: (data.agent as string) || (data.node as string) || "Flow",
        };
        break;

      case "node_start":
        yield {
          type: "agent_start",
          agent: (data.node as string) || (data.agent as string) || "Node",
        };
        break;

      case "node_done":
        break;

      case "agent_message":
      case "content":
      case "reasoning":
        yield {
          type: "content",
          content: (data.content as string) || "",
          agent: (data.agent as string) || (data.node as string),
          model: data.model as string,
        };
        break;

      case "tool_call":
        yield {
          type: "tool_call",
          tool: (data.tool_name as string) || (data.tool as string),
        };
        break;

      case "tool_result":
        break;

      case "ideas_generated":
      case "experts_done":
      case "options_researched":
        // Мета-события predefined flows — пробрасываем как есть
        yield data as unknown as AgentEvent;
        break;

      case "flow_done":
      case "agent_done":
      case "run_complete":
      case "done":
        yield {
          type: "done",
          model: data.model as string,
          tokens_total:
            (data.tokens as number) || (data.tokens_total as number) || 0,
          output: data.output as string | undefined,
        };
        break;

      case "error":
        yield {
          type: "error",
          error:
            (data.error as string) ||
            (data.message as string) ||
            "Unknown error",
        };
        break;

      default:
        if (data.content) {
          yield {
            type: "content",
            content: data.content as string,
          };
        }
        break;
    }
  } catch {
    // Skip malformed JSON
  }
}