import { type AgentOptionView, type AgentPlanStep, type AgentRunEvent, type AgentUsageTotals, CAPABILITY_ACTIVITY, EventType, type KnownCapabilityId, } from "@umt/contracts"; import { useCallback, useEffect, useMemo, useRef, useState } from "react"; import { agentMockEnabled, fetchRunEvents } from "~/api/agent.ts"; import { sse } from "~/lib/sse.ts"; // Agent 运行流:先按 seq 回放 agent_run_event,再跟随 SSE(agent.* 事件只作"有新事件"的提示, // 收到后按 lastSeq 增量补读;thinking/compaction 不上全局总线,补读时一并拿到)。SSE 不可用时退回轮询。 export type ToolCallView = { toolCallId: string; capabilityId: string; inputSummary: string | null; outputSummary: string | null; status: "running" | "succeeded" | "failed" | "denied"; reason: string | null; startedAt: string; endedAt: string | null; recovery: boolean; }; export type StreamItem = | { kind: "text"; key: string; text: string; at: string } | { kind: "thinking"; key: string; text: string; at: string } | { kind: "tool"; key: string; call: ToolCallView } | { kind: "compaction"; key: string; at: string; reason: string; status: string | null; done: boolean } | { kind: "options"; key: string; at: string; view: OptionsView } | { kind: "approval"; key: string; at: string; view: ApprovalView } | { kind: "end"; key: string; at: string; ok: boolean; message: string; reason: string | null; budget: string | null; }; /** agent.propose_options 的一次提问(options_proposed + options_resolved) */ export type OptionsView = { decisionId: string; question: string; options: AgentOptionView[]; recommendedId: string | null; stepId: string | null; expiresAt: string; timeoutAction: "recommended" | "stop"; resolved: { outcome: "chosen" | "rejected" | "stop"; choice: string | null; reason: string | null; by: string; } | null; }; /** 步骤处的审批卡(approval_requested + approval_resolved;挂起期间以审批对象的实时状态为准) */ export type ApprovalView = { approvalId: string; capabilityId: string; planType: string; planId: string; payloadHash: string; expectedRevision: number; riskLevel: "low" | "medium" | "high"; summary: string; stepId: string | null; expiresAt: string; resolved: { status: string; reason: string | null; decidedBy: string | null } | null; }; export type AgentStreamState = { events: AgentRunEvent[]; items: StreamItem[]; /** 最新的步骤清单(agent.plan);没有调用过为 null */ plan: { steps: AgentPlanStep[]; revision: number } | null; usage: AgentUsageTotals | null; finished: boolean; error: string | null; connected: "sse" | "poll" | "mock" | "idle"; /** 读取失败后手动重试(从最后收到的 seq 之后补读) */ retry: () => void; }; const SSE_TYPES: string[] = [EventType.agentTextDelta, EventType.agentToolCall, EventType.agentRunStatus]; /** * 不作为工具调用出现的能力(AGUI-3):步骤清单(agent.plan)在概况条里,方案(agent.propose_options)有自己的方案卡。 * 旧会话的事件里照样有它们的 tool_call_*,回放时一并跳过。 */ export const isHiddenCapability = (id: string | null | undefined) => !!id && CAPABILITY_ACTIVITY[id as KnownCapabilityId]?.hidden === true; /** 事件 → 展示项:相邻文本增量合并;工具调用按 toolCallId 折叠成一项;步骤清单 / 方案的调用不出现 */ export function foldEvents(events: AgentRunEvent[]): StreamItem[] { const items: StreamItem[] = []; const tools = new Map(); const asks = new Map(); const approvals = new Map(); for (const e of events) { const last = items[items.length - 1]; if ( (e.type === "tool_call_start" || e.type === "tool_call_end" || e.type === "tool_call_rejected") && isHiddenCapability(e.capabilityId) ) continue; switch (e.type) { case "text_delta": if (last?.kind === "text") last.text += e.delta; else items.push({ kind: "text", key: `t${e.seq}`, text: e.delta, at: e.at }); break; case "thinking_delta": if (last?.kind === "thinking") last.text += e.delta; else items.push({ kind: "thinking", key: `th${e.seq}`, text: e.delta, at: e.at }); break; case "tool_call_start": { // 挂起的调用(方案选择 / 审批 / 等子会话)恢复时按同一 toolCallId 重放:沿用原卡片,不重复出卡 const prev = tools.get(e.toolCallId); if (prev) { Object.assign(prev, { status: "running", outputSummary: null, reason: null, endedAt: null }); prev.recovery = e.recovery; if (e.inputSummary) prev.inputSummary = e.inputSummary; break; } const call: ToolCallView = { toolCallId: e.toolCallId, capabilityId: e.capabilityId, inputSummary: e.inputSummary, outputSummary: null, status: "running", reason: null, startedAt: e.at, endedAt: null, recovery: e.recovery, }; tools.set(e.toolCallId, call); items.push({ kind: "tool", key: `c${e.toolCallId}`, call }); break; } case "tool_call_end": { const call = tools.get(e.toolCallId); if (call) { call.outputSummary = e.outputSummary; call.status = e.isError ? "failed" : "succeeded"; call.endedAt = e.at; } else { items.push({ kind: "tool", key: `c${e.toolCallId}`, call: { toolCallId: e.toolCallId, capabilityId: e.capabilityId, inputSummary: null, outputSummary: e.outputSummary, status: e.isError ? "failed" : "succeeded", reason: null, startedAt: e.at, endedAt: e.at, recovery: e.recovery, }, }); } break; } case "tool_call_rejected": { const call = tools.get(e.toolCallId); if (call) { call.status = "denied"; call.reason = e.reason; call.endedAt = e.at; } else { items.push({ kind: "tool", key: `c${e.toolCallId}`, call: { toolCallId: e.toolCallId, capabilityId: e.capabilityId ?? e.toolName, inputSummary: null, outputSummary: null, status: "denied", reason: e.reason, startedAt: e.at, endedAt: e.at, recovery: false, }, }); } break; } case "compaction": if (e.phase === "start") items.push({ kind: "compaction", key: `k${e.seq}`, at: e.at, reason: e.reason, status: null, done: false }); else { const open = [...items].reverse().find((i) => i.kind === "compaction" && !i.done); if (open && open.kind === "compaction") { open.done = true; open.status = e.status ?? "completed"; } else items.push({ kind: "compaction", key: `k${e.seq}`, at: e.at, reason: e.reason, status: e.status ?? "completed", done: true, }); } break; case "options_proposed": { const view: OptionsView = { decisionId: e.decisionId, question: e.question, options: e.options, recommendedId: e.recommendedId, stepId: e.stepId, expiresAt: e.expiresAt, timeoutAction: e.timeoutAction, resolved: null, }; asks.set(e.decisionId, view); items.push({ kind: "options", key: `o${e.decisionId}`, at: e.at, view }); break; } case "options_resolved": { const v = asks.get(e.decisionId); if (v) v.resolved = { outcome: e.outcome, choice: e.choice, reason: e.reason, by: e.by }; break; } case "approval_requested": { const view: ApprovalView = { approvalId: e.approvalId, capabilityId: e.capabilityId, planType: e.planType, planId: e.planId, payloadHash: e.payloadHash, expectedRevision: e.expectedRevision, riskLevel: e.riskLevel, summary: e.summary, stepId: e.stepId, expiresAt: e.expiresAt, resolved: null, }; approvals.set(e.approvalId, view); items.push({ kind: "approval", key: `a${e.approvalId}`, at: e.at, view }); break; } case "approval_resolved": { const v = approvals.get(e.approvalId); if (v) v.resolved = { status: e.status, reason: e.reason, decidedBy: e.decidedBy }; break; } case "run_completed": items.push({ kind: "end", key: `e${e.seq}`, at: e.at, ok: true, message: e.finalText, reason: null, budget: null, }); break; case "run_failed": items.push({ kind: "end", key: `e${e.seq}`, at: e.at, ok: false, message: e.message, reason: e.reason, budget: e.budget ?? null, }); break; default: break; } } return items; } /** * 最近看过的会话的事件(模块级,最多 30 条会话):切回看过的会话时直接用已读到的事件铺出来, * 再从最后一个 seq 之后补读;不再先空一帧、再整段补读回来。 */ const EVENT_CACHE_MAX = 30; const eventCache = new Map(); export function cachedRunEvents(key: string): AgentRunEvent[] { return eventCache.get(key) ?? []; } export function cacheRunEvents(key: string, events: AgentRunEvent[]): void { if (!key || !events.length) return; eventCache.delete(key); eventCache.set(key, events); while (eventCache.size > EVENT_CACHE_MAX) { const oldest = eventCache.keys().next().value; if (oldest === undefined) break; eventCache.delete(oldest); } } const lastSeqOf = (events: AgentRunEvent[]) => events[events.length - 1]?.seq ?? -1; export function useAgentStream(runId: string | undefined, active: boolean): AgentStreamState { // 演示数据与真实会话分开缓存 const cacheKey = runId ? `${agentMockEnabled() ? "mock" : "live"}:${runId}` : ""; const [events, setEvents] = useState(() => cachedRunEvents(cacheKey)); const [error, setError] = useState(null); const [connected, setConnected] = useState("idle"); const [nonce, setNonce] = useState(0); const lastSeq = useRef(lastSeqOf(events)); // biome-ignore lint/correctness/useExhaustiveDependencies: 换 run 时重置(有缓存就从缓存接着读) useEffect(() => { const cached = cachedRunEvents(cacheKey); setEvents(cached); setError(null); lastSeq.current = lastSeqOf(cached); }, [runId]); useEffect(() => cacheRunEvents(cacheKey, events), [cacheKey, events]); // biome-ignore lint/correctness/useExhaustiveDependencies: nonce 变化 = 手动重试,重新建立补读 useEffect(() => { if (!runId) return; let stopped = false; const ctrl = new AbortController(); // 在途/待补标记按本次 effect 独立:active 变化重建 effect 时,被中止的旧请求不会把新一轮的补读挡在门外 const inflight = { current: false }; const pending = { current: false }; const pull = async () => { if (inflight.current) { pending.current = true; return; } inflight.current = true; try { do { pending.current = false; const next = await fetchRunEvents(runId, lastSeq.current, ctrl.signal); if (stopped) return; const fresh = next.filter((e) => e.seq > lastSeq.current).sort((a, b) => a.seq - b.seq); if (fresh.length) { lastSeq.current = fresh[fresh.length - 1]!.seq; setEvents((prev) => [...prev, ...fresh]); } setError(null); } while (pending.current && !stopped); } catch (err) { if ((err as Error).name !== "AbortError" && !stopped) setError((err as Error).message); } finally { inflight.current = false; } }; void pull(); if (!active) return () => { stopped = true; ctrl.abort(); }; const mock = agentMockEnabled(); // 共用页面级 SSE 连接:agent.* 事件只当"有新事件"的提示,按 lastSeq 补读 const unsubscribe = mock ? null : sse.subscribe(SSE_TYPES, (e) => { const p = e.payload as { runId?: string } | null; if (e.type === EventType.resync || p?.runId === runId) void pull(); }); setConnected(mock ? "mock" : typeof EventSource !== "undefined" ? "sse" : "poll"); // mock 模拟流式:密集轮询;真实端点:SSE 为主,2.5 秒轮询兜底 const timer = setInterval(() => void pull(), mock ? 150 : 2500); return () => { stopped = true; ctrl.abort(); clearInterval(timer); unsubscribe?.(); setConnected("idle"); }; }, [runId, active, nonce]); const retry = useCallback(() => setNonce((n) => n + 1), []); return useMemo(() => { const items = foldEvents(events); const end = [...events].reverse().find((e) => e.type === "run_completed" || e.type === "run_failed"); const usage = end && (end.type === "run_completed" || end.type === "run_failed") ? end.usage : null; return { events, items, plan: latestPlan(events), usage, finished: !!end, error, connected, retry }; }, [events, error, connected, retry]); } /** 最新的步骤清单(最后一条 plan_updated) */ export function latestPlan(events: AgentRunEvent[]): AgentStreamState["plan"] { for (let i = events.length - 1; i >= 0; i--) { const e = events[i]!; if (e.type === "plan_updated") return { steps: e.steps, revision: e.revision }; } return null; }