95 lines
6.9 KiB
TypeScript
95 lines
6.9 KiB
TypeScript
"use client";
|
|
import { useEffect, useRef, useState } from "react";
|
|
import { api } from "@/lib/api";
|
|
import { CANVAS_ACCEPT } from "@/lib/canvas-v2";
|
|
import { canvasPath } from "@/lib/canvas-sync";
|
|
import type { useCanvasTree } from "./use-canvas-tree";
|
|
export type CanvasModel = { id: string; label?: string; display_name?: string; output_modalities: string[]; input_modalities: string[]; available: boolean; unavailable_reason?: string; parameters_schema?: unknown };
|
|
export type AIRun = {
|
|
run_id: string; board_id: string; status: string; next_seq: number; has_more?: boolean; error?: string;
|
|
steps?: { step_id: string; board_id: string; node_id: string; status: string; error?: string }[];
|
|
outputs?: { step_id: string; output: { output_id: string; kind: string; src?: string } }[];
|
|
events?: { seq: number; type: string; board_id?: string; node_id?: string; logical_edge_id?: string; canvas_version?: number; output?: { output_id: string; text?: string; src?: string } }[];
|
|
applied_versions?: Record<string, number>;
|
|
};
|
|
const active = (status: string) => status === "queued" || status === "running";
|
|
export function useCanvasAI(workspace: string, root: string, live: boolean, tree: ReturnType<typeof useCanvasTree>) {
|
|
const [enabled, setEnabled] = useState(false);
|
|
const [runs, setRuns] = useState(new Map<string, AIRun>());
|
|
const [error, setError] = useState("");
|
|
const [conflicts, setConflicts] = useState(new Set<string>());
|
|
const [activeEdges, setActiveEdges] = useState(new Set<string>());
|
|
const reference = useRef(tree); reference.current = tree;
|
|
const cache = useRef(new Map<string, AIRun>());
|
|
const edgeStates = useRef(new Map<string, Set<string>>());
|
|
const pendingRefresh = useRef(new Map<string, number>());
|
|
const changed = () => { setRuns(new Map(cache.current)); setActiveEdges(new Set([...edgeStates.current.values()].flatMap((values) => [...values]))); };
|
|
useEffect(() => {
|
|
if (!live) return;
|
|
let alive = true, timer: ReturnType<typeof setTimeout> | undefined;
|
|
const controller = new AbortController();
|
|
const request = <T,>(path: string) => api<T>(path, { signal: controller.signal, headers: { Accept: CANVAS_ACCEPT } });
|
|
async function discover() {
|
|
let cursor = ""; const seen = new Set<string>();
|
|
do {
|
|
const page = await request<{ runs: { run_id: string; board_id: string; status: string }[]; next_cursor?: string }>(canvasPath(workspace, root) + "/ai/runs?status=active&include_descendants=true&limit=50" + (cursor ? "&cursor=" + encodeURIComponent(cursor) : ""));
|
|
for (const run of page.runs || []) if (!cache.current.has(run.run_id)) cache.current.set(run.run_id, { ...run, next_seq: 0 });
|
|
cursor = page.next_cursor || ""; if (seen.has(cursor)) throw new Error("invalid_run_cursor"); seen.add(cursor);
|
|
} while (cursor && alive);
|
|
}
|
|
let polls = 0;
|
|
async function poll() {
|
|
try {
|
|
if (++polls % 4 === 0) await discover();
|
|
for (const [id, previous] of cache.current) {
|
|
if (!active(previous.status) && !previous.has_more) continue;
|
|
let current = previous;
|
|
do {
|
|
const page = await request<Omit<AIRun, "board_id">>(canvasPath(workspace, previous.board_id) + `/ai/runs/${id}?after_seq=${current.next_seq}&limit=100`);
|
|
if (!Number.isSafeInteger(page.next_seq) || page.next_seq < current.next_seq || page.has_more && page.next_seq === current.next_seq) throw new Error("invalid_run_cursor");
|
|
const edges = edgeStates.current.get(id) || new Set<string>();
|
|
for (const event of page.events || []) {
|
|
if (event.logical_edge_id) { if (event.type === "edge.active") edges.add(event.logical_edge_id); else if (["edge.completed", "edge.failed", "edge.skipped"].includes(event.type)) edges.delete(event.logical_edge_id); }
|
|
if (event.type === "step.applied" && event.board_id && event.canvas_version) pendingRefresh.current.set(event.board_id, Math.max(pendingRefresh.current.get(event.board_id) || 0, event.canvas_version));
|
|
}
|
|
if (!active(page.status)) edges.clear(); edgeStates.current.set(id, edges);
|
|
current = { ...page, board_id: previous.board_id, events: [...(current.events || []), ...(page.events || [])].slice(-1000) };
|
|
cache.current.set(id, current);
|
|
} while (current.has_more && alive);
|
|
}
|
|
for (const [board, version] of pendingRefresh.current) {
|
|
if (reference.current.busy) break;
|
|
const writer = reference.current.writers.current.get(board);
|
|
if (writer && writer.version < version) {
|
|
try { await reference.current.refresh(board); } catch (error) {
|
|
if (error instanceof Error && error.message === "version_conflict") { pendingRefresh.current.delete(board); setConflicts((current) => new Set(current).add(board)); }
|
|
throw error;
|
|
}
|
|
}
|
|
pendingRefresh.current.delete(board);
|
|
}
|
|
if (alive) { changed(); setError(""); }
|
|
} catch (error) { if (alive && !controller.signal.aborted) { changed(); setError(error instanceof Error ? error.message : "request_failed"); } }
|
|
if (alive) timer = setTimeout(poll, 1200);
|
|
}
|
|
void discover().catch((error) => { if (alive) setError(error.message); }).finally(() => { if (alive) void poll(); });
|
|
return () => { alive = false; controller.abort(); if (timer) clearTimeout(timer); };
|
|
}, [workspace, root, live]);
|
|
async function run(boardId: string, node: string, model: string, prompt: string, propagate: boolean) {
|
|
if (!live) throw new Error("demo_ai_unavailable");
|
|
const result = await reference.current.command(boardId, "ai/runs", { node_id: node, model_id: model, prompt, parameters: {}, propagate }, undefined, false);
|
|
const id = result.run_id as string; cache.current.set(id, { run_id: id, board_id: boardId, status: "queued", next_seq: 0, steps: [{ step_id: "pending", board_id: boardId, node_id: node, status: "queued" }] }); changed();
|
|
}
|
|
async function cancel(run: AIRun) {
|
|
await api(canvasPath(workspace, run.board_id) + `/ai/runs/${run.run_id}/cancel`, { method: "POST", headers: { Accept: CANVAS_ACCEPT, "Idempotency-Key": crypto.randomUUID() }, body: "{}" });
|
|
const next = cache.current.get(run.run_id); if (next) { next.status = "queued"; } // Poll canonical terminal state, including already applied steps.
|
|
}
|
|
async function apply(run: AIRun, node: string, output: string, targetBoard: string) {
|
|
await reference.current.prepareApply(targetBoard, node);
|
|
await reference.current.command(run.board_id, `ai/runs/${run.run_id}/apply`, { node_id: node, output_id: output }, undefined, false, targetBoard);
|
|
setConflicts((current) => { const next = new Set(current); next.delete(targetBoard); return next; });
|
|
const previous = cache.current.get(run.run_id); if (previous) { previous.status = "queued"; changed(); }
|
|
}
|
|
return { enabled, setEnabled, runs, conflicts, activeEdges, error, run, cancel, apply, live };
|
|
}
|