137 lines
11 KiB
TypeScript
137 lines
11 KiB
TypeScript
/**
|
|
* 旧画布树的AI任务跟踪Hook。发现根及后代画布的活动任务，按事件游标增量读取，协调结果刷新与本地冲突显示。
|
|
* 此路径使用内存Map/Set和定时轮询，不具备新版runtime ledger的持久请求恢复；卸载会停止本Hook跟踪。
|
|
*/
|
|
"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";
|
|
/** 服务端模型目录条目：输入/输出模态、可用性与可选参数schema；本Hook只传递模型选择。 */
|
|
export type CanvasModel = { id: string; label?: string; display_name?: string; output_modalities: string[]; input_modalities: string[]; available: boolean; unavailable_reason?: string; parameters_schema?: unknown };
|
|
/** 旧AI接口的任务分页结果；next_seq/has_more跟踪事件页，steps/outputs/events供面板和活动连线显示。 */
|
|
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>;
|
|
};
|
|
/** 只有queued/running视为执行中；终态任务仍可能因has_more继续拉取剩余事件。 */
|
|
const active = (status: string) => status === "queued" || status === "running";
|
|
/** enabled仅控制画布AI模式显示；live才启用服务请求。tree通过ref保持最新，轮询effect按作用域重建。 */
|
|
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 wake = useRef<() => void>(() => {});
|
|
// cache是跟踪结果缓存，runs是发布给React的副本；每个run有自己的活动边集合，合并后供连线动画使用。
|
|
// 这些 ref/state 不因 workspace/root/live 改变而清空；宿主应按作用域重建 Hook 所在视图，避免沿用旧跟踪表。
|
|
const cache = useRef(new Map<string, AIRun>());
|
|
const edgeStates = useRef(new Map<string, Set<string>>());
|
|
// 每个board只保留最高待刷版本，避免同一批step.applied触发多次重复整画布读取。
|
|
const pendingRefresh = useRef(new Map<string, number>());
|
|
/** 发布新的Map/Set引用触发React更新；不把可变ref集合原样暴露成受控state。 */
|
|
const changed = () => { setRuns(new Map(cache.current)); setActiveEdges(new Set([...edgeStates.current.values()].flatMap((values) => [...values]))); };
|
|
// 单个effect内polling防并发轮询，dirty合并UI发布；AbortController统一中止这次作用域的读取。
|
|
useEffect(() => {
|
|
if (!live) return;
|
|
let alive = true, timer: ReturnType<typeof setTimeout> | undefined;
|
|
let polling = false, dirty = false, nextDiscovery = 0;
|
|
const controller = new AbortController();
|
|
/** 所有发现/分页GET共享取消信号并协商旧v2响应格式。 */
|
|
const request = <T,>(path: string) => api<T>(path, { signal: controller.signal, headers: { Accept: CANVAS_ACCEPT } });
|
|
/** 分页发现根和后代活动run，每页50；已跟踪run不覆盖，重复cursor判为协议异常避免无限循环。 */
|
|
// alive 只控制继续分页/发布等交接点；await 返回后仍可能先写共享 ref，不是每次赋值都有代次护栏。
|
|
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 }); dirty = true; }
|
|
cursor = page.next_cursor || ""; if (seen.has(cursor)) throw new Error("invalid_run_cursor"); seen.add(cursor);
|
|
} while (cursor && alive);
|
|
}
|
|
/** 是否仍需较快跟踪，包含终态但事件尚未读完的任务。 */
|
|
const hasActive = () => [...cache.current.values()].some((run) => active(run.status) || run.has_more);
|
|
/** 隐藏页面延后30秒；可见时串行发现、读事件、刷新已应用画布，再按活跃程度安排下一次。 */
|
|
async function poll() {
|
|
if (!alive || polling) return;
|
|
if (document.visibilityState === "hidden") { timer = setTimeout(poll, 30000); return; }
|
|
polling = true;
|
|
try {
|
|
if (Date.now() >= nextDiscovery) { await discover(); nextDiscovery = Date.now() + (hasActive() ? 6000 : 30000); }
|
|
for (const [id, previous] of cache.current) {
|
|
if (!active(previous.status) && !previous.has_more) continue;
|
|
let current = previous;
|
|
// 每页100事件，游标必须安全且不倒退，has_more时必须前进；每run最多保留最近1000条事件。
|
|
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");
|
|
// edge.active加入动画集合，completed/failed/skipped移除；任务终态清空剩余活动边。
|
|
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) };
|
|
// dirty 只比较该 run 原游标和状态；同游标/同状态下其它响应字段变化不一定立即发布给 React。
|
|
if (current.next_seq !== previous.next_seq || current.status !== previous.status) dirty = true;
|
|
cache.current.set(id, current);
|
|
} while (current.has_more && alive);
|
|
}
|
|
// 等树不忙且不在本地事务中才刷新；版本冲突记录到conflicts，不直接用生成快照覆盖本地草稿。
|
|
for (const [board, version] of pendingRefresh.current) {
|
|
if (reference.current.busy || reference.current.transactionBusy) 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) { if (dirty) { changed(); dirty = false; } setError(""); }
|
|
} catch (error) { if (alive && !controller.signal.aborted) { changed(); setError(error instanceof Error ? error.message : "request_failed"); } }
|
|
polling = false;
|
|
if (alive) timer = setTimeout(poll, hasActive() || pendingRefresh.current.size ? 1200 : 30000);
|
|
}
|
|
/** 被唤醒时清计时器并强制下一次重新发现；已有poll运行时由其收尾继续调度。 */
|
|
const resume = () => { if (timer) clearTimeout(timer); nextDiscovery = 0; void poll(); };
|
|
wake.current = resume;
|
|
/** 页面重新可见时立即唤醒，避免继续等待后台长间隔。 */
|
|
const visible = () => { if (document.visibilityState === "visible") resume(); };
|
|
document.addEventListener("visibilitychange", visible);
|
|
void poll();
|
|
// effect清理中止GET、定时器和可见性监听，wake退为空实现；此处不清除缓存或自动取消服务端任务。
|
|
return () => { alive = false; wake.current = () => {}; controller.abort(); if (timer) clearTimeout(timer); document.removeEventListener("visibilitychange", visible); };
|
|
}, [workspace, root, live]);
|
|
/** 真实画布经tree.command启动任务，先插queued占位供UI显示，再唤醒跟踪；演示模式明确拒绝。 */
|
|
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(); wake.current();
|
|
}
|
|
/** 以新幂等键发送取消命令后继续查询终态；暂记queued用于保持轮询，不表示服务端重新执行任务。 */
|
|
// 此写请求没有轮询 effect 的 signal，也未检查 live；按键是否可用由调用方和服务端权限共同约束。
|
|
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"; } wake.current(); // Poll canonical terminal state, including already applied steps.
|
|
}
|
|
/** 先prepareApply协调目标本地状态，再发显式采用命令；成功后移除冲突提示并继续跟踪服务端最终状态。 */
|
|
// run/apply 同样不由 effect 清理自动取消；卸载只停跟踪，不撤销已经发起的命令副作用。
|
|
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(); } wake.current();
|
|
}
|
|
// 对外只暴露显示状态和三个动作；调用者不直接修改cache、分页游标或刷新等待表。
|
|
return { enabled, setEnabled, runs, conflicts, activeEdges, error, run, cancel, apply, live };
|
|
}
|