/**
 * 文件作用：
 * 封装 SuperAgent 页面与 useStream 的唯一协议连接。
 *
 * 职责：
 * 1. 暴露标准 messages、toolCalls、interrupts、values 投影和运行控制方法。
 * 2. 提交用户消息、响应 interrupt、停止运行和仅断开订阅。
 * 3. 订阅业务 custom channel，并通知容器刷新 workspace/sections。
 *
 * 不负责：
 * 1. 创建业务 session、加载 workspace/sections 或管理业务会话。
 * 2. 手写 SSE、调用旧聊天协议或维护独立恢复接口。
 *
 * 维护说明：
 * Agent v2 不可用时只暴露错误，调用方不得回退 /api/chat 或 runs/stream。
 */
"use strict";

import { useCallback, useEffect, useMemo, useRef, useState } from "react";
import { useStream } from "@langchain/react";
import { ProtocolSseTransportAdapter } from "@langchain/langgraph-sdk";
import { ROOT_PUMP_CHANNELS } from "@langchain/langgraph-sdk/stream";
import { buildSuperAgentServiceUrl } from "../actions/api";
import {
   getCompatibleToken,
   getCompatibleUserId,
} from "../../../utils/authCompat";

const BUSINESS_CHANNELS = [
   "custom",
   "custom:generation",
   "custom:section",
   "custom:workspace",
];

const TERMINAL_INTERRUPT_STATUSES = new Set([
   "confirmed",
   "edited",
   "completed",
   "done",
   "success",
   "cancelled",
   "canceled",
]);

const ACTIVE_TASK_STATUSES = new Set([
   "queued",
   "starting",
   "started",
   "running",
   "pending",
   "processing",
   "generating",
   "writing",
   "active",
   "streaming",
   "resuming",
   "awaiting_interrupt",
]);

const getInterruptStatus = (item = {}) => {
   const payload = item?.value || item?.payload || {};
   return String(
      item.status ||
         item.checkpointStatus ||
         payload.checkpoint_status ||
         payload.checkpointStatus ||
         payload.status ||
         ""
   ).toLowerCase();
};

const isTerminalInterrupt = (item = {}) =>
   TERMINAL_INTERRUPT_STATUSES.has(getInterruptStatus(item));

const getInterruptInteractionKey = (item = {}) => {
   const payload = item?.value || item?.payload || item || {};
   const interactionId = String(
      payload.interaction_id ||
         payload.interactionId ||
         item.interaction_id ||
         item.interactionId ||
         ""
   ).trim();
   if (!interactionId) return "";

   const revision = payload.revision ?? item.revision ?? "";
   return `${interactionId}:${String(revision)}`;
};

const parseMessageContent = (content) => {
   if (content && typeof content === "object") return content;
   if (typeof content !== "string") return null;

   try {
      return JSON.parse(content);
   } catch (_) {
      return null;
   }
};

const collectResolvedInterruptInteractions = (messages = []) => {
   const resolvedInteractions = new Set();
   (Array.isArray(messages) ? messages : []).forEach((message) => {
      if (String(message?.type || "").toLowerCase() !== "tool") return;

      const result = parseMessageContent(message.content);
      const decisionType = String(result?.type || "").toLowerCase();
      const interactionId = String(result?.interaction_id || "").trim();
      const revision = result?.revision ?? "";
      if (
         interactionId &&
         ["approve", "edit", "reject"].includes(decisionType)
      ) {
         resolvedInteractions.add(`${interactionId}:${String(revision)}`);
      }
   });
   return resolvedInteractions;
};

const isSuperAgentEventRequest = (input) => {
   const requestUrl = String(input?.url || input || "");
   return /\/threads\/[^/]+\/stream\/events(?:\?|$)/.test(requestUrl);
};

const isUseStreamHistoryDiscoveryRequest = (input) => {
   const requestUrl = String(input?.url || input || "");
   return /\/(?:threads|tasks)\/[^/]+\/history(?:\?|$)/.test(requestUrl);
};

const isLifecycleWatcherRequest = (input, init = {}) => {
   if (!isSuperAgentEventRequest(input) || typeof init.body !== "string") {
      return false;
   }

   try {
      const requestBody = JSON.parse(init.body);
      const channels = Array.isArray(requestBody?.channels)
         ? requestBody.channels
         : [];
      return (
         channels.length === 2 &&
         channels.includes("lifecycle") &&
         channels.includes("input") &&
         !Object.prototype.hasOwnProperty.call(requestBody, "namespaces") &&
         !Object.prototype.hasOwnProperty.call(requestBody, "depth")
      );
   } catch {
      return false;
   }
};

const isLifecycleWatcherFilter = (params = {}) => {
   const channels = Array.isArray(params.channels) ? params.channels : [];
   return (
      channels.length === 2 &&
      channels.includes("lifecycle") &&
      channels.includes("input") &&
      !Object.prototype.hasOwnProperty.call(params, "namespaces") &&
      !Object.prototype.hasOwnProperty.call(params, "depth")
   );
};

const getEventRetryDelay = (response, attempt) => {
   const retryAfter = response.headers.get("retry-after");
   if (retryAfter) {
      const seconds = Number(retryAfter);
      if (Number.isFinite(seconds)) {
         return Math.min(Math.max(seconds * 1000, 100), 10000);
      }

      const retryAt = Date.parse(retryAfter);
      if (Number.isFinite(retryAt)) {
         return Math.min(Math.max(retryAt - Date.now(), 100), 10000);
      }
   }

   return Math.min(250 * 2 ** Math.max(attempt - 1, 0), 4000);
};

const waitForEventRetry = (delay, signal) => {
   if (signal?.aborted) {
      return Promise.reject(signal.reason || new Error("事件流连接已中止"));
   }

   return new Promise((resolve, reject) => {
      let timer = null;
      const handleAbort = () => {
         if (timer) clearTimeout(timer);
         signal?.removeEventListener("abort", handleAbort);
         reject(signal.reason || new Error("事件流连接已中止"));
      };

      timer = setTimeout(() => {
         signal?.removeEventListener("abort", handleAbort);
         resolve();
      }, delay);
      signal?.addEventListener("abort", handleAbort, { once: true });
   });
};

const EVENT_STREAM_RELEASE_WAIT_MS = 300;
const activeEventStreamsByThread = new Map();
const eventStreamCloseAtByThread = new Map();

const waitForEventStreamRelease = async (threadId, signal) => {
   const closedAt = eventStreamCloseAtByThread.get(threadId);
   if (!closedAt) return;

   const elapsed = Date.now() - closedAt;
   const remaining = EVENT_STREAM_RELEASE_WAIT_MS - elapsed;
   if (remaining > 0) {
      await waitForEventRetry(remaining, signal);
   }
   if (eventStreamCloseAtByThread.get(threadId) === closedAt) {
      eventStreamCloseAtByThread.delete(threadId);
   }
};

// 服务端释放旧 subscriber 需要一点时间。SDK 默认会在 429 后直接进入
// 固定次数的重连，耗尽后把 ThreadStream 标记为失败；这里按 Retry-After
// 在 fetch 层恢复，避免一次连接竞争让整个会话进入不可恢复状态。
const fetchEventStreamWithRecovery = async (input, init) => {
   const max429Retries = 10;
   let attempt = 0;

   while (true) {
      const response = await fetch(input, init);
      if (response.status !== 429 || !isSuperAgentEventRequest(input)) {
         return response;
      }

      if (attempt >= max429Retries || init?.signal?.aborted) {
         return response;
      }

      attempt += 1;
      if (response.body) {
         await response.body.cancel().catch(() => undefined);
      }
      await waitForEventRetry(getEventRetryDelay(response, attempt), init?.signal);
   }
};

//[单订阅 SSE]当前 Agent 服务同一 thread 只允许一个 /events subscriber。
//SDK 默认 rotation 是先开新连接、再关旧连接，会在服务端触发 429；这里改为先关旧再开新。
class SingleSubscriberSseTransportAdapter extends ProtocolSseTransportAdapter {
   activeEventStream = null;
   eventHandles = new Set();
   latestState = null;
   onStateChange = null;
   stateRequestVersion = 0;

   async getState() {
      const requestVersion = ++this.stateRequestVersion;
      const requestedThreadId = String(this.threadId || "").trim();
      const state = await super.getState();
      // getState 可能在切换 thread 后才返回；旧请求不能覆盖新 thread 的快照。
      if (
         requestVersion === this.stateRequestVersion &&
         String(this.threadId || "").trim() === requestedThreadId
      ) {
         this.latestState = state;
         this.onStateChange?.();
      }
      return state;
   }

   openEventStream(params) {
      const isLifecycleWatcher = isLifecycleWatcherFilter(params);
      const threadId = String(this.threadId || "").trim();

      if (isLifecycleWatcher) {
         // useStream 的 root pump 已经覆盖 lifecycle/input，额外 watcher
         // 不参与主事件流的连接排他和释放等待。
         return super.openEventStream(params);
      }

      const previousEventStream = activeEventStreamsByThread.get(threadId);
      previousEventStream?.close();
      this.activeEventStream?.close();
      this.activeEventStream = null;

      let innerHandle = null;
      let isClosed = false;
      let eventIterator = null;
      const createStartPromise = () => (async () => {
          await waitForEventStreamRelease(threadId);
          if (isClosed) throw new Error("事件流连接已取消");
          innerHandle = super.openEventStream(params);
          return innerHandle;
      })();
      let startPromise = createStartPromise();

      const handle = {
         ready: startPromise.then((nextHandle) => nextHandle.ready),
         events: {
            [Symbol.asyncIterator]: () => ({
               next: async () => {
                   while (!isClosed) {
                      if (!eventIterator) {
                         const nextHandle = await startPromise;
                         eventIterator = nextHandle.events[Symbol.asyncIterator]();
                      }
                      const nextEvent = await eventIterator.next();
                      if (!nextEvent.done) return nextEvent;

                      // 服务端可能在一轮运行结束后主动关闭 SSE。保持外层
                      // handle 不结束，下一轮提交时重新建立同一 thread 的事件流。
                      innerHandle?.close();
                      innerHandle = null;
                      eventIterator = null;
                      if (!isClosed) {
                         await waitForEventRetry(100);
                         startPromise = createStartPromise();
                      }
                   }
                   return { done: true, value: undefined };
               },
               return: async () => {
                  handle.close();
                  if (eventIterator?.return) await eventIterator.return();
                  return { done: true, value: undefined };
               },
            }),
         },
         close: () => {
            if (isClosed) return;
            isClosed = true;
            eventStreamCloseAtByThread.set(threadId, Date.now());
            if (activeEventStreamsByThread.get(threadId) === handle) {
               activeEventStreamsByThread.delete(threadId);
            }
            if (this.activeEventStream === handle) this.activeEventStream = null;
            this.eventHandles.delete(handle);
            innerHandle?.close();
         },
      };

      this.eventHandles.add(handle);
      this.activeEventStream = handle;
      activeEventStreamsByThread.set(threadId, handle);
      return handle;
   }

   // ThreadStream 会在切换 thread 时调用 close，但同一个 adapter 会被 SDK 复用；
   // 这里只关闭连接，不调用父类 close（父类会把 transport 永久标记为 closed）。
   close() {
      for (const handle of this.eventHandles) handle.close();
      this.eventHandles.clear();
      this.activeEventStream = null;
   }
}

const mergeAbortSignals = (signals = []) => {
   const validSignals = signals.filter(Boolean);
   if (!validSignals.length) return undefined;
   if (validSignals.length === 1) return validSignals[0];
   if (typeof AbortSignal !== "undefined" && typeof AbortSignal.any === "function") {
      return AbortSignal.any(validSignals);
   }

   const controller = new AbortController();
   const abort = () => controller.abort();
   validSignals.forEach((signal) => {
      if (signal.aborted) {
         abort();
         return;
      }
      signal.addEventListener("abort", abort, { once: true });
   });
   return controller.signal;
};

// 业务事件复用 useStream 已有的 root pump，不再额外建立页面级事件连接。
// 将业务 channel 纳入 root pump，避免 useChannelEffect 为 custom channel
// 触发 SSE rotation，导致新旧 /events 短暂重叠并被服务端返回 429。
BUSINESS_CHANNELS.forEach((channel) => {
   if (!ROOT_PUMP_CHANNELS.includes(channel)) ROOT_PUMP_CHANNELS.push(channel);
});

const createClientMessageId = () => {
   if (typeof crypto !== "undefined" && typeof crypto.randomUUID === "function") {
      return crypto.randomUUID();
   }
   return `client-${Date.now()}-${Math.random().toString(16).slice(2)}`;
};

const getEventPayload = (event) =>
   event?.data ??
   event?.payload ??
   event?.params?.data?.payload ??
   event?.params?.data ??
   event;

const getEventDedupeKey = (event) => {
   if (event?.event_id) return String(event.event_id);
   const payload = getEventPayload(event) || {};
   return [
      payload.task_id || payload.taskId || "",
      payload.node_id || payload.nodeId || "",
      payload.version || "",
      payload.type || event?.type || event?.event || "",
      payload.progress ?? payload.task?.progress ?? "",
      payload.done ?? payload.task?.done ?? "",
      payload.section_id || payload.sectionId || payload.task?.section_id || "",
   ].join(":");
};

const normalizeStreamError = (error) => {
   const status = error?.status || error?.response?.status;
   if (status === 401 || status === 403) return new Error("登录状态已失效，请重新登录");
   if (status === 404) return new Error("智能体服务暂不可用");
   const message = String(error?.message || "");
   if (status === 429 || /(?:^|\D)429(?:\D|$)|too many requests|rate limit/i.test(message)) {
      return new Error("事件流连接繁忙，正在自动恢复，请稍候");
   }
   if (/sse|stream|network|fetch/i.test(message)) return new Error("智能体流式连接异常");
   return error instanceof Error ? error : new Error("智能体协议响应错误");
};

const getSuperAgentApiUrl = () => buildSuperAgentServiceUrl("/api/v1");

const useSuperAgentStream = ({
   activeSessionId,
   userId: sessionUserId,
   onBusinessInvalidation,
   onStreamError,
}) => {
   const handledEventKeysRef = useRef(new Set());
   const interruptResumeRef = useRef({ active: false, running: false });
   const pendingInterruptRef = useRef(false);
   const awaitingInterruptRef = useRef(false);
   const [runPhase, setRunPhase] = useState("idle");
   const [isLocalRunActive, setIsLocalRunActive] = useState(false);
   const [isRespondingInterrupt, setIsRespondingInterrupt] = useState(false);
   const [isBusinessRunning, setIsBusinessRunning] = useState(false);
   const [runtimeCheckpointDeliveries, setRuntimeCheckpointDeliveries] = useState([]);
   const [runtimeInterrupts, setRuntimeInterrupts] = useState([]);
   const [dismissedInterruptIds, setDismissedInterruptIds] = useState(new Set());
   const resolvedInterruptIdsRef = useRef(new Set());
   const resolvedInterruptInteractionsRef = useRef(new Set());
   const [currentInterruptId, setCurrentInterruptId] = useState("");
   const [hydratedThreadId, setHydratedThreadId] = useState("");
   const [transportStateRevision, setTransportStateRevision] = useState(0);
   const eventRequestControllersRef = useRef(new Map());
   const pendingSubmitThreadIdRef = useRef("");
   const hydrationSessionRef = useRef({
      sessionId: "",
      hasSeenLoading: false,
   });
   const localRunActiveRef = useRef(false);
   const localRunHasSeenLoadingRef = useRef(false);
   const activeSessionIdRef = useRef("");
   activeSessionIdRef.current = activeSessionId;
   const token = getCompatibleToken();
   const userId = sessionUserId || getCompatibleUserId();
   // 保持身份头稳定；会话切换由 callerOptions 的会话级 fetch 变化触发
   // 新 controller，同时先中止旧会话的 /events 请求。
   const defaultHeaders = useMemo(
      () => ({
         ...(token ? { Authorization: `Bearer ${token}` } : {}),
         ...(token ? { "X-Search-Token": token } : {}),
         ...(userId ? { "X-User-Id": String(userId) } : {}),
      }),
      [token, userId]
   );
   const fetchUseStreamTransport = useCallback(async (...args) => {
      const [input, init = {}] = args;
      if (isUseStreamHistoryDiscoveryRequest(input)) {
         // 当前 SDK hydrate 会为 namespace discovery 自动读取 history。
         // SuperAgent 只以 thread state 和 event stream 为准，禁止该
         // 非业务读取落到服务端，避免 history 再次参与会话恢复。
         return new Response("[]", {
            status: 200,
            headers: { "content-type": "application/json" },
         });
      }
      if (isLifecycleWatcherRequest(input, init)) {
         // Agent 服务每个 thread 只允许一个 events subscriber。useStream
         // 的 root pump 已覆盖 lifecycle/input，额外 watcher 会稳定触发 429。
         return new Response(null, {
            status: 204,
            headers: { "content-type": "text/event-stream" },
         });
      }
      let requestInit = init;
      const sessionKey = String(
         pendingSubmitThreadIdRef.current || activeSessionIdRef.current || ""
      ).trim();
      if (sessionKey && isSuperAgentEventRequest(input)) {
         let sessionController = eventRequestControllersRef.current.get(sessionKey);
         if (!sessionController) {
            sessionController = new AbortController();
            eventRequestControllersRef.current.set(sessionKey, sessionController);
         }
         requestInit = {
            ...init,
            signal: mergeAbortSignals([init.signal, sessionController.signal]),
         };
      }

      return fetchEventStreamWithRecovery(input, requestInit);
   }, []);
   const eventTransport = useMemo(
      () => {
         const transport = new SingleSubscriberSseTransportAdapter({
            apiUrl: getSuperAgentApiUrl(),
            defaultHeaders,
            fetch: fetchUseStreamTransport,
            maxReconnectAttempts: 5,
         });
         transport.onStateChange = () => {
            setTransportStateRevision((revision) => revision + 1);
         };
         return transport;
      },
      [activeSessionId, defaultHeaders, fetchUseStreamTransport]
   );
   // StreamController 在 hydrate 时会先通过 transport.getState() 取新 thread；
   // 自定义 adapter 需在该 effect 运行前同步绑定目标 thread，避免读到旧会话。
   eventTransport.setThreadId(
      activeSessionId || pendingSubmitThreadIdRef.current || ""
   );
   const callerOptions = useMemo(
      () => ({ fetch: fetchUseStreamTransport }),
      [fetchUseStreamTransport]
   );
   const stream = useStream({
      assistantId: "freesun_agent",
      apiUrl: getSuperAgentApiUrl(),
      threadId: activeSessionId || null,
      defaultHeaders,
      callerOptions,
      messagesKey: "messages",
      optimistic: true,
      transport: eventTransport,
   });

   const handleBusinessEvent = useCallback((event) => {
      const eventKey = getEventDedupeKey(event);
      if (handledEventKeysRef.current.has(eventKey)) return;
      handledEventKeysRef.current.add(eventKey);
      const payload = getEventPayload(event) || {};
      const eventType = String(payload.type || "");
      if ([
         "generation.started",
         "generation.progress",
         "section.started",
         "section.progress",
         "section.updated",
         "sections.updated",
         "workspace.updated",
         "sections.rewrite.requested",
      ].includes(eventType)) {
         setIsBusinessRunning(true);
         awaitingInterruptRef.current = false;
         setRunPhase("running");
      }
      if ([
         "generation.completed",
         "generation.failed",
         "generation.cancelled",
         "generation.error",
      ].includes(eventType)) {
         setIsBusinessRunning(false);
      }
      const envelope = payload.envelope;
      if (payload.type === "workspace.snapshot" && envelope?.checkpoint_id) {
         setRuntimeCheckpointDeliveries((current) => {
            const next = current.filter(
               (item) => item?.checkpoint_id !== envelope.checkpoint_id
            );
            return [...next, envelope];
         });
      }
      onBusinessInvalidation?.(payload, event);
   }, [onBusinessInvalidation]);

   const handleStreamError = useCallback((error) => {
      localRunActiveRef.current = false;
      setIsLocalRunActive(false);
      localRunHasSeenLoadingRef.current = false;
      interruptResumeRef.current = { active: false, running: false };
      pendingInterruptRef.current = false;
      awaitingInterruptRef.current = false;
      setRunPhase("failed");
      setIsRespondingInterrupt(false);
      setIsBusinessRunning(false);
      onStreamError?.(normalizeStreamError(error));
   }, [onStreamError]);

   useEffect(() => {
      const sessionKey = String(activeSessionId || "").trim();
      return () => {
         if (sessionKey) {
            const sessionController = eventRequestControllersRef.current.get(sessionKey);
            sessionController?.abort();
            eventRequestControllersRef.current.delete(sessionKey);
         }
         // SDK 的 stop/disconnect 不一定会销毁自定义 transport 的旧 handle，
         // 卸载或切换会话时必须主动关闭，确保后端释放对应 subscriber。
         eventTransport.close();
      };
   }, [activeSessionId, eventTransport]);

   useEffect(() => {
      if (pendingSubmitThreadIdRef.current === String(activeSessionId || "")) {
         pendingSubmitThreadIdRef.current = "";
      }
      hydrationSessionRef.current = {
         sessionId: String(activeSessionId || ""),
         hasSeenLoading: false,
      };
      handledEventKeysRef.current.clear();
      localRunActiveRef.current = false;
      setIsLocalRunActive(false);
      localRunHasSeenLoadingRef.current = false;
      interruptResumeRef.current = { active: false, running: false };
      pendingInterruptRef.current = false;
      awaitingInterruptRef.current = false;
      setRunPhase("idle");
      setIsRespondingInterrupt(false);
      setIsBusinessRunning(false);
      setRuntimeCheckpointDeliveries([]);
      setRuntimeInterrupts([]);
      setDismissedInterruptIds(new Set());
      resolvedInterruptIdsRef.current.clear();
      resolvedInterruptInteractionsRef.current.clear();
      setCurrentInterruptId("");
      setHydratedThreadId("");
   }, [activeSessionId]);

   useEffect(() => {
      const sessionId = String(activeSessionId || "");
      const threadId = String(stream.threadId || "");
      if (!sessionId || sessionId !== threadId) return undefined;

      let cancelled = false;
      // isThreadLoading 在 useStream hydrate 切换的中间阶段可能仍是 false，
      // 不能用它作为快照完成标志；hydrationPromise 才代表当前 thread 的 state
      // 已经写入 root store。
      Promise.resolve(stream.hydrationPromise)
         .then(() => {
            if (cancelled || activeSessionIdRef.current !== sessionId) return;
            setHydratedThreadId(sessionId);
         })
         .catch(() => undefined);

      return () => {
         cancelled = true;
      };
   }, [
      activeSessionId,
      stream.hydrationPromise,
      stream.threadId,
      transportStateRevision,
   ]);

   const checkpointDeliveries = useMemo(() => {
      const deliveries = new Map();
      const hydratedDeliveries = [
         ...(Array.isArray(stream.values?.checkpoint_deliveries)
            ? stream.values.checkpoint_deliveries
            : []),
         ...(Array.isArray(eventTransport.latestState?.values?.checkpoint_deliveries)
            ? eventTransport.latestState.values.checkpoint_deliveries
            : []),
      ];
      [...hydratedDeliveries, ...runtimeCheckpointDeliveries].forEach((item) => {
         if (!item?.checkpoint_id) return;
         deliveries.set(item.checkpoint_id, item);
      });
      return [...deliveries.values()];
   }, [
      eventTransport.latestState,
      runtimeCheckpointDeliveries,
      stream.values?.checkpoint_deliveries,
      transportStateRevision,
   ]);

   useEffect(() => {
      if (stream.error) handleStreamError(stream.error);
   }, [handleStreamError, stream.error]);

   useEffect(() => {
      if (!localRunActiveRef.current) return;
      if (stream.isLoading) {
         localRunHasSeenLoadingRef.current = true;
         return;
      }
      if (!localRunHasSeenLoadingRef.current) return;

      localRunActiveRef.current = false;
      localRunHasSeenLoadingRef.current = false;
      setIsLocalRunActive(false);
      setRunPhase((currentPhase) =>
         ["running", "resuming"].includes(currentPhase) ? "completed" : currentPhase
      );
      setIsBusinessRunning(false);
      setIsRespondingInterrupt(false);
   }, [stream.isLoading]);

   useEffect(() => {
      const isCurrentThread =
         Boolean(activeSessionId) &&
         String(stream.threadId || "") === String(activeSessionId || "");
      if (!isCurrentThread) return undefined;

      const listenerSessionId = String(activeSessionId || "");
      const currentThread = stream.getThread?.();
      if (!currentThread) return undefined;

      return currentThread.onEvent((event) => {
         // 切换会话后旧 Thread 的监听器可能在清理前再收到一个事件，
         // 不能把旧会话的事件写入新会话的流式消息状态。
         if (activeSessionIdRef.current !== listenerSessionId) return;
         if (event?.method === "input.requested") {
            const data = event.params?.data || {};
            const interruptId = String(data.interrupt_id || "").trim();
            const interactionKey = getInterruptInteractionKey(data.payload);
            const isResolvedInterrupt =
               resolvedInterruptIdsRef.current.has(interruptId) ||
               (interactionKey &&
                  resolvedInterruptInteractionsRef.current.has(interactionKey));
            if (interruptId && !isResolvedInterrupt) {
               localRunActiveRef.current = false;
               setIsLocalRunActive(false);
               setCurrentInterruptId(interruptId);
               setRuntimeInterrupts((currentInterrupts) => {
                  const nextInterrupts = currentInterrupts.filter(
                     (item) => String(item?.id || "") !== interruptId
                  );
                  return [
                     ...nextInterrupts,
                     {
                        id: interruptId,
                        value: data.payload,
                        namespace: Array.isArray(event.params?.namespace)
                           ? event.params.namespace
                           : [],
                     },
                  ];
               });
               pendingInterruptRef.current = true;
               awaitingInterruptRef.current = false;
               setRunPhase("waiting_interrupt");
            }
         }
         if (event?.method === "lifecycle" && Array.isArray(event?.params?.namespace) && event.params.namespace.length === 0) {
            const lifecycleEvent = event.params.data?.event;
            if (["started", "running"].includes(lifecycleEvent)) {
               pendingInterruptRef.current = false;
               awaitingInterruptRef.current = false;
               setRunPhase("running");
            }
            if (lifecycleEvent === "running" && interruptResumeRef.current.active) {
               interruptResumeRef.current.running = true;
            }
            if (lifecycleEvent === "interrupted") {
               const threadInterrupts = currentThread.interrupts || [];
               const hasRealInterrupt =
                  pendingInterruptRef.current ||
                  stream.interrupts.length > 0 ||
                  threadInterrupts.length > 0;
               if (hasRealInterrupt) {
                  localRunActiveRef.current = false;
                  setIsLocalRunActive(false);
                  awaitingInterruptRef.current = false;
                  setRunPhase("waiting_interrupt");
               } else {
                  // interrupted 可能早于 input.requested 到达，不能将本轮误判为完成。
                  awaitingInterruptRef.current = true;
                  setRunPhase("awaiting_interrupt");
               }
            }
            if (["completed", "failed", "aborted"].includes(lifecycleEvent)) {
               if (pendingInterruptRef.current) {
                  setRunPhase("waiting_interrupt");
               } else {
                  awaitingInterruptRef.current = false;
                  localRunActiveRef.current = false;
                  setIsLocalRunActive(false);
                  // aborted 表示本轮已结束，不能让 started 留下的 running 状态继续生效。
                  setRunPhase(lifecycleEvent === "aborted" ? "completed" : lifecycleEvent);
               }
               setIsBusinessRunning(false);
               if (interruptResumeRef.current.active) {
                  interruptResumeRef.current = { active: false, running: false };
                  setIsRespondingInterrupt(false);
               }
            }
         }
         if (!BUSINESS_CHANNELS.includes(event?.method)) return;
         handleBusinessEvent(event);
      });
   }, [activeSessionId, handleBusinessEvent, stream.threadId]);

   const prepareForSessionSwitch = useCallback(() => {
      // 先让旧 thread 的事件监听失效。activeSessionId 在 React 下一次渲染
      // 前仍可能是旧值，单靠监听器里的 sessionId 判断不够。
      activeSessionIdRef.current = "";
      // pendingSubmitThreadId 只服务于“创建会话后首次发送”这一瞬间；
      // 切换会话时必须清掉，否则新 transport 会继续绑定旧 thread。
      pendingSubmitThreadIdRef.current = "";
      setHydratedThreadId("");
      localRunActiveRef.current = false;
      setIsLocalRunActive(false);
      localRunHasSeenLoadingRef.current = false;
      interruptResumeRef.current = { active: false, running: false };
      pendingInterruptRef.current = false;
      awaitingInterruptRef.current = false;
      setRunPhase("idle");
      setIsRespondingInterrupt(false);
      setIsBusinessRunning(false);
      setRuntimeInterrupts([]);
      resolvedInterruptIdsRef.current.clear();
      resolvedInterruptInteractionsRef.current.clear();

      // 会话切换只断开当前浏览器订阅，不能发送 stop/cancel 命令。
      // stopRun 只由用户点击停止按钮触发，否则后端会把原会话改为 suspended。
      // useStream 切换 threadId 时会等待旧 root pump 清理；直接关闭当前
      // ThreadStream 可以立即中止旧 /events 请求，避免阻塞新会话接管。
      const currentThread = stream.getThread?.();
      eventTransport.close();

      return Promise.allSettled([
         stream.disconnect(),
         currentThread ? currentThread.close() : Promise.resolve(),
      ]);
   }, [eventTransport, stream]);

   const submitMessage = useCallback(async ({
      content,
      displayContent,
      attachmentIds = [],
      threadId,
      //[中断恢复后的直接输入允许进入 SDK 队列，等待当前恢复运行结束后再发送]
      allowDuringInterruptResponse = false,
      multitaskStrategy = "reject",
   }) => {
      const targetThreadId = String(threadId || activeSessionId || "").trim();
      if (!targetThreadId) throw new Error("请先创建任务");
       const isBlockedByInterrupt = isRespondingInterrupt && !allowDuringInterruptResponse;
       const isSwitchingThread = String(stream.threadId || "") !== targetThreadId;
       // isThreadLoading 只表示当前 thread 的快照还在 hydrate，不能阻止首条消息提交。
       // 真正表示本轮任务仍在运行的是 isLoading；否则新建会话会先创建成功，
       // 但首条消息被这里提前拒绝，页面最终只剩空会话。
       if ((stream.isLoading && !isSwitchingThread) || isBlockedByInterrupt) {
          throw new Error("任务正在处理，请稍后再试");
       }
       if (stream.isThreadLoading && !isSwitchingThread) {
          // 新建会话的 state 可能还未写入 root store。等待 hydrate 完成后再提交，
          // 避免 optimistic 消息被空快照或旧事件回放覆盖。
          await Promise.resolve(stream.hydrationPromise).catch(() => undefined);
       }
       // activeSessionId 的 React 状态更新可能晚于首次提交；先绑定 transport
      // 和事件请求所属会话，避免首轮响应仍落到旧 thread。
      if (hydrationSessionRef.current.sessionId === targetThreadId) {
         hydrationSessionRef.current.hasSeenLoading = true;
      }
      pendingSubmitThreadIdRef.current = targetThreadId;
      eventTransport.setThreadId(targetThreadId);
      localRunActiveRef.current = true;
      setIsLocalRunActive(true);
      localRunHasSeenLoadingRef.current = false;
      setRunPhase("running");
      try {
         await stream.submit({
            messages: [{
               type: "human",
               id: createClientMessageId(),
               content: String(content || ""),
            }],
            display_content: String(displayContent ?? content ?? ""),
            attachment_ids: Array.isArray(attachmentIds) ? attachmentIds : [],
         }, {
            threadId: targetThreadId,
            multitaskStrategy,
            metadata: { source: "super-agent-web" },
         });
      } catch (error) {
         localRunActiveRef.current = false;
         setIsLocalRunActive(false);
         throw error;
      }
   }, [activeSessionId, eventTransport, isRespondingInterrupt, stream]);

   const respondInterrupt = useCallback((response, options = {}) => {
      const interruptId = String(options.interruptId || "").trim();
      localRunActiveRef.current = true;
      setIsLocalRunActive(true);
      localRunHasSeenLoadingRef.current = false;
      interruptResumeRef.current = { active: true, running: false };
      pendingInterruptRef.current = false;
      awaitingInterruptRef.current = false;
      setRunPhase("resuming");
      setIsRespondingInterrupt(true);

      if (interruptId) {
         // 提交恢复请求后先隐藏旧卡片，避免响应流期间旧 interrupt 被重复渲染。
         setDismissedInterruptIds((current) => {
            const next = new Set(current);
            next.add(interruptId);
            return next;
         });
         setCurrentInterruptId((currentId) =>
            currentId === interruptId ? "" : currentId
         );
      }

      return stream.respond(response, {
         interruptId: options.interruptId,
         namespace: options.namespace,
      }).then((result) => {
         if (interruptId) {
            resolvedInterruptIdsRef.current.add(interruptId);
            const interactionKey = getInterruptInteractionKey(
               response?.value || response?.payload || response
            );
            if (interactionKey) {
               resolvedInterruptInteractionsRef.current.add(interactionKey);
            }
            setRuntimeInterrupts((currentInterrupts) =>
               currentInterrupts.filter(
                  (item) => String(item?.id || "") !== interruptId
               )
            );
         }
         // respond 只代表恢复命令已提交，不代表后端本轮运行完成。
         // 处理中状态由 lifecycle completed/failed/interrupted 统一释放。
         return result;
      }).catch((error) => {
         if (interruptId) {
            // 恢复失败时撤销临时隐藏，让用户可以重新提交。
            resolvedInterruptIdsRef.current.delete(interruptId);
            setDismissedInterruptIds((current) => {
               const next = new Set(current);
               next.delete(interruptId);
               return next;
            });
            setCurrentInterruptId((currentId) => currentId || interruptId);
         }
         interruptResumeRef.current = { active: false, running: false };
         localRunActiveRef.current = false;
         setIsLocalRunActive(false);
         pendingInterruptRef.current = true;
         setRunPhase("waiting_interrupt");
         setIsRespondingInterrupt(false);
         throw error;
      });
   }, [stream]);

   const stopRun = useCallback(async () => {
      localRunActiveRef.current = false;
      setIsLocalRunActive(false);
      localRunHasSeenLoadingRef.current = false;
      interruptResumeRef.current = { active: false, running: false };
      pendingInterruptRef.current = false;
      awaitingInterruptRef.current = false;
      setRunPhase("idle");
      setIsRespondingInterrupt(false);
      setIsBusinessRunning(false);
      await stream.stop();
   }, [stream]);

   const currentSessionId = String(activeSessionId || "");
   const currentThreadId = String(stream.threadId || "");
   const projectedThreadId = String(
      pendingSubmitThreadIdRef.current || currentThreadId
   );
   const isThreadBound =
      Boolean(currentSessionId) && currentSessionId === currentThreadId;
   const isProjectedThreadBound =
      Boolean(currentSessionId) && currentSessionId === projectedThreadId;
   if (hydrationSessionRef.current.sessionId !== currentSessionId) {
      hydrationSessionRef.current = {
         sessionId: currentSessionId,
         hasSeenLoading: false,
      };
   }
   if (isThreadBound && stream.isThreadLoading) {
      hydrationSessionRef.current.hasSeenLoading = true;
   }
   const isThreadReady =
      isThreadBound &&
      !stream.isThreadLoading &&
      hydratedThreadId === currentSessionId;
   const isPendingSubmitThread =
      isProjectedThreadBound &&
      pendingSubmitThreadIdRef.current === currentSessionId;
   const isStreamProjectionReady =
      isThreadReady || isPendingSubmitThread;
   const isInterruptProjectionReady =
      isThreadBound && runtimeInterrupts.length > 0;
   const isCurrentThread = isThreadReady;
   const hydratedStateThreadId = String(
      eventTransport.latestState?.checkpoint?.thread_id ||
      eventTransport.latestState?.checkpoint?.threadId ||
      eventTransport.latestState?.values?.task?.task_id ||
      eventTransport.latestState?.values?.task?.taskId ||
      ""
   );
   const hydratedState =
      isThreadReady &&
      hydratedStateThreadId === currentSessionId
         ? eventTransport.latestState
         : null;
    const rawStreamMessages = isStreamProjectionReady ? stream.messages : [];
    const rawStreamValues = isStreamProjectionReady ? stream.values || {} : {};
    const hydratedMessages = Array.isArray(hydratedState?.values?.messages)
       ? hydratedState.values.messages
       : [];
    const historicalResolvedInteractionKeys = useMemo(
       () =>
          collectResolvedInterruptInteractions([
             ...hydratedMessages,
             ...rawStreamMessages,
          ]),
       [hydratedMessages, rawStreamMessages]
    );
    useEffect(() => {
       historicalResolvedInteractionKeys.forEach((interactionKey) => {
          resolvedInterruptInteractionsRef.current.add(interactionKey);
       });
    }, [historicalResolvedInteractionKeys]);
     // 只有 hydrate 真正拿到消息后，才能解除首次提交的 thread 投影保护。
    // 仅有 task 快照不代表消息已完成落库，否则 task 先到时会让当前消息再次被隐藏。
    const hasHydratedSessionContent = hydratedMessages.length > 0;
   useEffect(() => {
      if (!isThreadReady || !isPendingSubmitThread || !hasHydratedSessionContent) return;
      pendingSubmitThreadIdRef.current = "";
   }, [
      hasHydratedSessionContent,
      isPendingSubmitThread,
      isThreadReady,
   ]);

   const hydratedTailIds = hydratedMessages
      .slice(-3)
      .map((item) => String(item?.id || "").trim())
      .filter(Boolean);
   const rawStreamMessageIds = new Set(
      rawStreamMessages
         .map((item) => String(item?.id || "").trim())
         .filter(Boolean)
   );
   const hydratedTaskStatus = String(
      hydratedState?.values?.status ||
      hydratedState?.values?.session_status ||
      hydratedState?.values?.sessionStatus ||
      hydratedState?.values?.task?.session_status ||
      hydratedState?.values?.task?.sessionStatus ||
      hydratedState?.values?.task?.status ||
      ""
   ).toLowerCase();
   const rawTaskStatus = String(
      rawStreamValues.status ||
      rawStreamValues.session_status ||
      rawStreamValues.sessionStatus ||
      rawStreamValues.task?.session_status ||
      rawStreamValues.task?.sessionStatus ||
      rawStreamValues.task?.status ||
      ""
   ).toLowerCase();
   const hydratedCheckpointDeliveryCount = Array.isArray(
      hydratedState?.values?.checkpoint_deliveries
   )
      ? hydratedState.values.checkpoint_deliveries.length
      : 0;
   const rawCheckpointDeliveryCount = Array.isArray(rawStreamValues.checkpoint_deliveries)
      ? rawStreamValues.checkpoint_deliveries.length
      : 0;
   const shouldPreferHydratedRuntimeState =
      Boolean(hydratedState) &&
      Boolean(hydratedTaskStatus || hydratedCheckpointDeliveryCount) &&
      (!rawTaskStatus && !rawCheckpointDeliveryCount ||
         hydratedTaskStatus !== rawTaskStatus ||
         hydratedCheckpointDeliveryCount > rawCheckpointDeliveryCount);
   //[state 是当前 thread 的最新快照，SSE 重连可能先重放更早的 values]
    // 消息和 task/value 的权威来源不能共用一个切换条件。
    // hydrate 返回空 messages 时，仍可能带有较新的 task 状态；此时只能更新
    // task/value，不能让空快照覆盖事件流已经收到的消息。
    const shouldPreferHydratedMessages =
       hydratedMessages.length > 0 &&
       (rawStreamMessages.length < hydratedMessages.length ||
          (hydratedTailIds.length > 0 &&
             !hydratedTailIds.every((messageId) => rawStreamMessageIds.has(messageId))));
    const shouldPreferHydratedState =
       shouldPreferHydratedRuntimeState || shouldPreferHydratedMessages;
    const streamMessages = shouldPreferHydratedMessages
       ? hydratedMessages
       : rawStreamMessages;
   const streamToolCalls = isStreamProjectionReady ? stream.toolCalls : [];
   const streamValues = shouldPreferHydratedState
      ? hydratedState?.values || {}
      : rawStreamValues;
   const streamSessionStatus = String(
      streamValues.status ||
      streamValues.session_status ||
      streamValues.sessionStatus ||
      streamValues.task?.session_status ||
      streamValues.task?.sessionStatus ||
      ""
   ).toLowerCase();
   const streamTask = streamValues?.task
      ? {
           ...streamValues.task,
           ...(streamSessionStatus
              ? { session_status: streamSessionStatus }
              : {}),
        }
      : null;
   const persistedTaskStatus = String(
      streamSessionStatus || streamTask?.status || ""
   ).toLowerCase();
   const isPersistedTaskActive = ACTIVE_TASK_STATUSES.has(persistedTaskStatus);
   const hasPersistedNextNodes =
      Array.isArray(hydratedState?.next) && hydratedState.next.length > 0;
    const hydratedInterrupts = [
       ...(Array.isArray(hydratedState?.interrupts)
          ? hydratedState.interrupts
         : []),
      ...(Array.isArray(hydratedState?.values?.__interrupt__)
         ? hydratedState.values.__interrupt__.map((item) => ({
              ...item,
              id: item?.id || item?.interruptId,
              value: item?.value ?? item?.payload,
              namespace: Array.isArray(item?.namespace) ? item.namespace : [],
           }))
            : []),
    ];
    const pendingHydratedInterrupts = hydratedInterrupts.filter((item) => {
       const interruptId = String(item?.id || item?.interruptId || "").trim();
       const interactionKey = getInterruptInteractionKey(item);
       return (
          !resolvedInterruptIdsRef.current.has(interruptId) &&
          !historicalResolvedInteractionKeys.has(interactionKey)
       );
    });
    const hasPersistedInterrupts = pendingHydratedInterrupts.length > 0;
    const persistedInterrupts =
       hydratedState
          ? pendingHydratedInterrupts
          : isStreamProjectionReady
             ? stream.interrupts
             : [];
   const streamInterrupts = [
      ...(Array.isArray(persistedInterrupts) ? persistedInterrupts : []),
      ...runtimeInterrupts,
   ];
   const shouldResetStaleWaitingPhase =
      Boolean(hydratedState) &&
      !hasPersistedInterrupts &&
      ["waiting_interrupt", "awaiting_interrupt"].includes(runPhase);
   const isPersistedThreadActive =
      isPersistedTaskActive || hasPersistedNextNodes || hasPersistedInterrupts;
   const isPersistedExecutionActive =
      isPersistedThreadActive &&
      ![
         "suspended",
         "waiting_user",
         "awaiting_interrupt",
         "waiting_interrupt",
      ].includes(persistedTaskStatus) &&
      !hasPersistedInterrupts;
    const hasRuntimeExecutionSignal =
       localRunActiveRef.current ||
       isLocalRunActive ||
       isBusinessRunning ||
       ["running", "resuming"].includes(runPhase);
    const isPersistedSuspended =
       persistedTaskStatus === "suspended" &&
       !hasPersistedInterrupts &&
       !hasRuntimeExecutionSignal;
   const isTerminalRunPhase = [
      "completed",
      "aborted",
      "failed",
      "cancelled",
      "canceled",
   ].includes(runPhase);
   const effectiveRunPhase =
      isPersistedSuspended
         ? "idle"
         : shouldResetStaleWaitingPhase
            ? "running"
         : isPersistedExecutionActive && (runPhase === "idle" || isTerminalRunPhase)
            ? "running"
            : runPhase;
   const effectiveIsLocalRunActive = isPersistedSuspended ? false : isLocalRunActive;
   const effectiveIsBusinessRunning = isPersistedSuspended ? false : isBusinessRunning;
   const visibleInterrupts = useMemo(() => {
       if (!isStreamProjectionReady && !isInterruptProjectionReady) return [];

      const merged = new Map();
      const appendInterrupt = (item, fallback = {}) => {
         const id = String(item?.id || item?.interruptId || "").trim();
         if (!id) return;
         const previous = merged.get(id) || {};
         merged.set(id, {
            ...previous,
            ...fallback,
            ...item,
            id,
            value: item?.value ?? item?.payload ?? previous.value,
            namespace: Array.isArray(item?.namespace)
               ? item.namespace
               : previous.namespace || [],
         });
      };

       streamInterrupts.forEach((item) => appendInterrupt(item));
       return [...merged.values()].filter(
          (item) => {
             const interruptId = String(item.id || "");
             const interactionKey = getInterruptInteractionKey(item);
            const isHistoricalResolved =
                historicalResolvedInteractionKeys.has(interactionKey) ||
                resolvedInterruptInteractionsRef.current.has(interactionKey);
             return (
                !dismissedInterruptIds.has(interruptId) &&
                !resolvedInterruptIdsRef.current.has(interruptId) &&
                !isHistoricalResolved &&
                !isTerminalInterrupt(item)
             );
          }
       );
    }, [
       dismissedInterruptIds,
       isInterruptProjectionReady,
       isStreamProjectionReady,
       historicalResolvedInteractionKeys,
       streamInterrupts,
    ]);

   const isRunPhaseActive = ["running", "awaiting_interrupt", "resuming"].includes(
      effectiveRunPhase
   );
   const effectiveIsLoading =
      isPersistedSuspended
         ? isRespondingInterrupt
         : stream.isLoading ||
            isRespondingInterrupt ||
            isBusinessRunning ||
            isRunPhaseActive ||
            isPersistedExecutionActive;

   return {
      messages: streamMessages,
      toolCalls: streamToolCalls,
      interrupt: visibleInterrupts[0] || null,
      interrupts: visibleInterrupts,
      values: streamValues,
      checkpointDeliveries: isStreamProjectionReady ? checkpointDeliveries : [],
      task: streamTask,
      sessionStatus: streamSessionStatus,
      isLoading: effectiveIsLoading,
      isLocalRunActive: effectiveIsLocalRunActive,
      runPhase: effectiveRunPhase,
      isPersistedThreadActive,
       currentInterruptId:
          hydratedState &&
          pendingHydratedInterrupts.length === 0 &&
         runtimeInterrupts.length === 0
            ? ""
            : currentInterruptId,
      isWaitingForInterrupt:
         visibleInterrupts.length > 0 ||
         (effectiveRunPhase === "waiting_interrupt" && !isPersistedThreadActive),
      isAwaitingInterrupt: effectiveRunPhase === "awaiting_interrupt",
      isBusinessRunning: effectiveIsBusinessRunning,
      isThreadLoading: stream.isThreadLoading,
      isThreadReady,
      isThreadProjectionReady:
         isStreamProjectionReady || isInterruptProjectionReady,
      error: isCurrentThread && stream.error ? normalizeStreamError(stream.error) : null,
      threadId: projectedThreadId,
      submitMessage,
      respondInterrupt,
      respondAllInterrupts: stream.respondAll,
      stopRun,
      disconnect: stream.disconnect,
      prepareForSessionSwitch,
   };
};

export default useSuperAgentStream;