import type { AIMessage, Message } from "@langchain/langgraph-sdk";
import type { ThreadsClient } from "@langchain/langgraph-sdk/client";
import { useStream } from "@langchain/langgraph-sdk/react";
import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query";
import type { QueryClient } from "@tanstack/react-query";
import { useCallback, useEffect, useRef, useState } from "react";
import { useNavigate } from "react-router-dom";
import { toast } from "sonner";

import type { PromptInputMessage } from "@/components/ai-elements/prompt-input";

import { getAPIClient } from "../api";
import { apiFetch } from "../api";
import { getStoredAuth } from "../auth";
import { rememberAskTime } from "../chat-timestamps";
import { getBackendBaseURL, getLangGraphBaseURL } from "../config";
import { useI18n } from "../i18n/hooks";
import { OptType, writeYLog } from "../log/y-log";
import type { FileInMessage } from "../messages/utils";
import type { LocalSettings } from "../settings";
import { isThreadRagModeLocked, lockThreadRagMode, useLocalSettings } from "../settings";
import { useUpdateSubtask } from "../tasks/context";
import type { UploadedFileInfo } from "../uploads";
import { promptInputFilePartToFile, uploadFiles } from "../uploads";

import type { AgentThread, AgentThreadState } from "./types";
import {
  collectHumanMessageIds,
  hasServerEchoedOptimisticHuman,
  mergeStreamDisplayValues,
  shouldDiscardTerminalOverlay,
} from "./stream-display";
import {
  isAbortLikeError,
  isTransientStreamTransportError,
} from "./thread-access-error";
import { isHiddenFromRecentChats, shouldHideSystemThreadsFromList } from "./utils";

export type ToolEndEvent = {
  name: string;
  data: unknown;
  toolCallId?: string;
};

export type ToolStartEvent = ToolEndEvent;

export type ThreadStreamOptions = {
  threadId?: string | null | undefined;
  context: LocalSettings["context"];
  isMock?: boolean;
  assistantId?: string;
  /**
   * Per-page run behaviour.  Most chats keep the SDK defaults; long-running
   * workspace flows can opt into a resumable stream without changing global
   * chat semantics.
   */
  runOptions?: {
    onDisconnect?: "cancel" | "continue";
    multitaskStrategy?: "reject" | "rollback" | "interrupt" | "enqueue";
  };
  /** 额外写到新建 thread 的 metadata（与 agent_id 合并）。
   *  任务深链用 `{ taskId }` 把对话挂到任务下（1:N），列表按 metadata.taskId 过滤。 */
  threadMetadata?: Record<string, unknown>;
  onStart?: (threadId: string) => void;
  onFinish?: (state: AgentThreadState) => void;
  onToolStart?: (event: ToolStartEvent) => void;
  onToolEnd?: (event: ToolEndEvent) => void;
};

type SendMessageOptions = {
  additionalKwargs?: Record<string, unknown>;
  /**
   * Extra human-context messages submitted before the visible user message.
   * They are persisted with `hide_from_ui` so the model receives rich business
   * context without dumping long prompts into the conversation surface.
   */
  hiddenContext?:
    | string
    | {
        text: string;
        name?: string;
        additionalKwargs?: Record<string, unknown>;
      }
    | Array<{
        text: string;
        name?: string;
        additionalKwargs?: Record<string, unknown>;
      }>;
};

type ThreadTitleRefreshState = Pick<AgentThreadState, "title"> & {
  title_provisional?: boolean;
};

function normalizeHiddenContextMessages(options?: SendMessageOptions) {
  const value = options?.hiddenContext;
  if (!value) return [];
  const items =
    typeof value === "string"
      ? [{ text: value }]
      : Array.isArray(value)
        ? value
        : [value];
  return items
    .map((item) => ({
      ...item,
      text: item.text.trim(),
    }))
    .filter((item) => item.text.length > 0);
}

function normalizeStoredRunId(runId: string | null): string | null {
  if (!runId) {
    return null;
  }

  const trimmed = runId.trim();
  if (!trimmed) {
    return null;
  }

  const queryIndex = trimmed.indexOf("?");
  if (queryIndex >= 0) {
    const params = new URLSearchParams(trimmed.slice(queryIndex + 1));
    const queryRunId = params.get("run_id")?.trim();
    if (queryRunId) {
      return queryRunId;
    }
  }

  const pathWithoutQueryOrHash = trimmed.split(/[?#]/, 1)[0]?.trim() ?? "";
  if (!pathWithoutQueryOrHash) {
    return null;
  }

  const runsMarker = "/runs/";
  const runsIndex = pathWithoutQueryOrHash.lastIndexOf(runsMarker);
  if (runsIndex >= 0) {
    const runIdAfterMarker = pathWithoutQueryOrHash
      .slice(runsIndex + runsMarker.length)
      .split("/", 1)[0]
      ?.trim();
    if (runIdAfterMarker) {
      return runIdAfterMarker;
    }
    return null;
  }

  const segments = pathWithoutQueryOrHash
    .split("/")
    .map((segment) => segment.trim())
    .filter(Boolean);
  return segments.at(-1) ?? null;
}

function getRunMetadataStorage(): {
  getItem(key: `lg:stream:${string}`): string | null;
  setItem(key: `lg:stream:${string}`, value: string): void;
  removeItem(key: `lg:stream:${string}`): void;
} {
  return {
    getItem(key) {
      const normalized = normalizeStoredRunId(
        window.sessionStorage.getItem(key),
      );
      if (normalized) {
        window.sessionStorage.setItem(key, normalized);
        return normalized;
      }
      window.sessionStorage.removeItem(key);
      return null;
    },
    setItem(key, value) {
      const normalized = normalizeStoredRunId(value);
      if (normalized) {
        window.sessionStorage.setItem(key, normalized);
        return;
      }
      window.sessionStorage.removeItem(key);
    },
    removeItem(key) {
      window.sessionStorage.removeItem(key);
    },
  };
}

/**
 * SDK ``useStream`` stores the in-flight run id at ``lg:stream:{threadId}``
 * but **deletes that key on stream onSuccess**. A premature SSE close
 * (proxy timeout / false-positive disconnect) looks like success, so the
 * reconnect key vanishes while the backend run is still occupying the
 * thread. We keep a second key the SDK never touches and only clear it
 * after ``runs.get`` confirms a terminal status.
 */
const ACTIVE_RUN_STORAGE_PREFIX = "df:active-run:";

function persistActiveRunId(threadId: string, runId: string): void {
  if (typeof window === "undefined") return;
  const normalized = normalizeStoredRunId(runId);
  if (!normalized) return;
  try {
    window.sessionStorage.setItem(`lg:stream:${threadId}`, normalized);
    window.sessionStorage.setItem(
      `${ACTIVE_RUN_STORAGE_PREFIX}${threadId}`,
      normalized,
    );
  } catch {
    /* private-mode / quota */
  }
}

function readActiveRunId(threadId: string): string | null {
  if (typeof window === "undefined") return null;
  try {
    return (
      normalizeStoredRunId(
        window.sessionStorage.getItem(`${ACTIVE_RUN_STORAGE_PREFIX}${threadId}`),
      ) ||
      normalizeStoredRunId(window.sessionStorage.getItem(`lg:stream:${threadId}`))
    );
  } catch {
    return null;
  }
}

function clearActiveRunId(threadId: string): void {
  if (typeof window === "undefined") return;
  try {
    window.sessionStorage.removeItem(`lg:stream:${threadId}`);
    window.sessionStorage.removeItem(
      `${ACTIVE_RUN_STORAGE_PREFIX}${threadId}`,
    );
  } catch {
    /* noop */
  }
}

function isInflightRunStatus(status: string | undefined): boolean {
  return status === "pending" || status === "running";
}

/**
 * 显式停止某会话「服务端正在跑」的 run。
 *
 * 注意：useStream / `thread.stop()` 只中断本地 SSE 监听，服务端 run 仍会继续跑。
 * 任务流程要求「开启新对话时老对话要停止」，所以这里读出 SDK 存在 sessionStorage
 * （``lg:stream:{threadId}``）里的 run id，调用 ``runs.cancel`` 真正取消后端 run，
 * 再清掉本地 reconnect 记录。已结束 / 不存在的 run 取消失败时静默忽略。
 */
export async function stopThreadRun(
  threadId: string | null | undefined,
): Promise<void> {
  if (!threadId || typeof window === "undefined") return;
  let runId: string | null = null;
  try {
    runId =
      normalizeStoredRunId(
        window.sessionStorage.getItem(`lg:stream:${threadId}`),
      ) || readActiveRunId(threadId);
  } catch {
    runId = null;
  }
  if (!runId) return;
  try {
    await getAPIClient().runs.cancel(threadId, runId);
  } catch {
    /* 已结束 / 不存在 / 网络错误等忽略 */
  }
  clearActiveRunId(threadId);
}

/**
 * 协助卡提交前：若上一轮还在流式（弱模型出卡后偶发未停干净），先取消服务端 run
 * 并断开本地 SSE，避免 409「已有活跃 run」或用户点了选项却发不出去。
 */
export async function stopActiveThreadBeforeClarificationReply(
  threadId: string | null | undefined,
  thread: { isLoading?: boolean; stop: () => void | Promise<void> },
): Promise<void> {
  let storedRunId: string | null = null;
  if (threadId && typeof window !== "undefined") {
    try {
      storedRunId =
        normalizeStoredRunId(
          window.sessionStorage.getItem(`lg:stream:${threadId}`),
        ) || readActiveRunId(threadId);
    } catch {
      storedRunId = null;
    }
  }
  // 协助卡出现时图已经 jump_to=end，run 多半已是 terminal。再 stop()/cancel
  // 会再触发 onFinish → getState overlay，把刚发出的用户回复冻住、整段攒着出。
  if (storedRunId && threadId) {
    try {
      const run = await getAPIClient().runs.get(threadId, storedRunId);
      if (run && !isInflightRunStatus(run.status)) {
        clearActiveRunId(threadId);
        return;
      }
    } catch {
      /* 404 / 网络：按下面的 best-effort stop 走 */
    }
  }
  if (!thread.isLoading && !storedRunId) return;
  try {
    await stopThreadRun(threadId);
  } catch {
    /* best-effort */
  }
  try {
    await Promise.resolve(thread.stop());
  } catch {
    /* best-effort */
  }
  // cancel 落地需要一拍，立刻 submit 仍可能 409。
  await new Promise((resolve) => setTimeout(resolve, 120));
}

function getStreamErrorMessage(error: unknown): string {
  if (typeof error === "string" && error.trim()) {
    return error;
  }
  if (error instanceof Error && error.message.trim()) {
    return error.message;
  }
  if (typeof error === "object" && error !== null) {
    const message = Reflect.get(error, "message");
    if (typeof message === "string" && message.trim()) {
      return message;
    }
    const nestedError = Reflect.get(error, "error");
    if (nestedError instanceof Error && nestedError.message.trim()) {
      return nestedError.message;
    }
    if (typeof nestedError === "string" && nestedError.trim()) {
      return nestedError;
    }
  }
  return "Request failed.";
}

const TERMINAL_RUN_STATUSES = new Set([
  "success",
  "error",
  "timeout",
  "interrupted",
]);

// A completed server run normally sends a terminal SSE event.  In local
// development, however, a Vite hot update, a proxy disconnect, or a brief
// network interruption can make the browser miss that final event while the
// backend continues (and successfully completes) the run.  Do not probe on
// every render: only treat the stream as suspicious after it has been quiet
// for a while, then reconcile it from the authoritative run/state endpoints.
const STREAM_RECONCILIATION_IDLE_MS = 12_000;
const STREAM_RECONCILIATION_POLL_MS = 5_000;
// A run can legitimately be quiet while a long tool call / slow model write is
// in progress. Reconnect only after a long stall; replacing the local reader
// must NOT cancel the backend run (see reconnect below).
const STREAM_RECONNECT_IDLE_MS = 90_000;
const STREAM_RECONNECT_MAX_ATTEMPTS = 3;

async function fetchThreadTitleRefreshState(
  threadId: string,
  signal?: AbortSignal,
): Promise<ThreadTitleRefreshState | null> {
  const response = await apiFetch(
    `${getBackendBaseURL()}/api/threads/${encodeURIComponent(threadId)}/state`,
    { signal },
  );
  if (!response.ok) {
    return null;
  }
  const json = (await response.json()) as {
    values?: { title?: string; title_provisional?: boolean };
  };
  if (!json.values) {
    return null;
  }
  return {
    title: json.values.title ?? "",
    title_provisional: json.values.title_provisional,
  };
}

/**
 * Detect "thread not found" / forbidden errors from the LangGraph SDK.
 *
 * Backend returns 404 both when a thread truly doesn't exist and when the
 * currently authenticated user doesn't own it — the SDK surfaces both as
 * ``HTTP 404: {"detail":"Thread <uuid> not found"}``. The most common cause
 * we see in practice is a stale tab whose URL still points at the previous
 * login session's thread. Raw-toasting the 404 just confuses the user, so
 * we re-frame it around the current login identity.
 */
function isThreadNotAccessibleError(message: string): boolean {
  const lowered = message.toLowerCase();
  // Backend now returns "not found" (no row + no checkpoint) and
  // "not accessible" (row owned by another user / stale login session) as
  // distinct details — both mean the current user can't use this thread.
  if (/thread\s+[\w-]{8,}\s+not\s+(found|accessible)/i.test(message)) {
    return true;
  }
  // Fallback: a bare 404/403 from the threads endpoint when the SDK has
  // wrapped the detail differently.
  return (
    (lowered.includes("http 404") || lowered.includes("http 403")) &&
    lowered.includes("thread")
  );
}

/**
 * Detect a "Run <id> not found" 404 from the LangGraph SDK.
 *
 * The SDK persists the in-flight run id under ``lg:stream:{threadId}`` in
 * sessionStorage and, on mount, auto-``joinStream``s it to resume a dropped
 * SSE. After a backend restart the in-memory run registry is gone, so the join
 * 404s — and the SDK only clears that key on reconnect *success*, never on
 * failure. So every switch back into the conversation re-fires the same 404.
 * We detect it here to clear the stale key ourselves and stay silent (the
 * conversation renders from history; there is nothing live to reconnect to).
 */
function isRunNotFoundError(message: string): boolean {
  return /run\s+[\w-]{8,}\s+not\s+found/i.test(message);
}

/**
 * 同一会话上一轮还没生成完，又提交了新问题 → 后端返回 409
 * 「already has an active run」。匹配 HTTP 409、英文兜底文案，或后端已中文化
 * 的提示，统一映射成一句友好的中文。
 */
function isActiveRunConflictError(message: string): boolean {
  return (
    /\b409\b/.test(message) ||
    /already has an active run/i.test(message) ||
    /还在生成回答/.test(message)
  );
}

function isModelConcurrencyLimitError(message: string): boolean {
  return (
    /\b429\b/.test(message) ||
    /并发已达到上限/.test(message) ||
    /并发调用已达到上限/.test(message) ||
    /使用人数较多/.test(message) ||
    /too many requests/i.test(message)
  );
}

function friendlySendLimitMessage(message: string): string | null {
  if (isActiveRunConflictError(message)) {
    return "当前对话还在生成回答，请等待本轮回复结束，或先点“停止”再发送新的问题。";
  }
  if (isModelConcurrencyLimitError(message)) {
    return /当前账号对该模型/.test(message)
      ? "当前账号对该模型的并发调用已达到上限，请等待已有回答结束后再试。"
      : "当前使用人数较多，请稍等一下再试。";
  }
  return null;
}

function buildCurrentUserLabel(): string {
  const auth = getStoredAuth();
  if (!auth) {
    return "未登录";
  }
  const email = auth.email?.trim();
  if (email) {
    return email;
  }
  const userId = auth.user_id?.trim();
  return userId ? userId : "未知用户";
}

/**
 * 从 `on_chat_model_end` 事件的 output 里尽力取出模型输出文本。
 *
 * output 形态可能是 AIMessage（`{ content: string | parts[] }`），也可能是
 * LangChain 的 `{ generations: [[{ text, message }]] }`。全部容错，取不到返回空串。
 */
function textFromChatModelOutput(output: unknown): string {
  if (output == null) return "";
  if (typeof output === "string") return output;
  if (typeof output !== "object") return String(output);
  const o = output as Record<string, unknown>;
  if ("content" in o) {
    const c = o.content;
    if (typeof c === "string") return c;
    if (Array.isArray(c)) {
      const parts = c
        .map((p) =>
          p && typeof p === "object" && "text" in p
            ? String((p as { text?: unknown }).text ?? "")
            : "",
        )
        .filter(Boolean);
      if (parts.length) return parts.join("");
    }
  }
  if (Array.isArray(o.generations)) {
    const texts: string[] = [];
    const collect = (gen: unknown) => {
      if (Array.isArray(gen)) {
        gen.forEach(collect);
        return;
      }
      if (gen && typeof gen === "object") {
        const gg = gen as Record<string, unknown>;
        if (typeof gg.text === "string" && gg.text) texts.push(gg.text);
        else if (gg.message) {
          const t = textFromChatModelOutput(gg.message);
          if (t) texts.push(t);
        }
      }
    };
    collect(o.generations);
    if (texts.length) return texts.join("");
  }
  return "";
}

/**
 * 从 `on_chat_model_*` 事件的 input 里尽力取出最后一条 user/human 文本，
 * 作为本次大模型调用的「问题」。input 可能是 `{ messages: BaseMessage[] | BaseMessage[][] }`。
 */
function lastHumanFromChatModelInput(input: unknown): string {
  try {
    const raw =
      input && typeof input === "object" && "messages" in input
        ? (input as { messages?: unknown }).messages
        : input;
    const flat: unknown[] = [];
    const push = (m: unknown) => {
      if (Array.isArray(m)) m.forEach(push);
      else if (m != null) flat.push(m);
    };
    push(raw);
    for (let i = flat.length - 1; i >= 0; i--) {
      const m = flat[i];
      if (!m || typeof m !== "object") continue;
      const obj = m as Record<string, unknown>;
      const type = obj.type ?? obj.role;
      if (type === "human" || type === "user") {
        const kwargs = obj.additional_kwargs;
        if (
          kwargs &&
          typeof kwargs === "object" &&
          (kwargs as Record<string, unknown>).hide_from_ui === true
        ) {
          continue;
        }
        const text = textFromChatModelOutput(m);
        if (text) return text;
      }
    }
  } catch {
    /* 取不到忽略 */
  }
  return "";
}

export function useThreadStream({
  threadId,
  context,
  isMock,
  assistantId,
  runOptions,
  threadMetadata,
  onStart,
  onFinish,
  onToolStart,
  onToolEnd,
}: ThreadStreamOptions) {
  const { t } = useI18n();
  const [localSettings] = useLocalSettings();
  // Track the thread ID that is currently streaming to handle thread changes during streaming
  const [onStreamThreadId, setOnStreamThreadId] = useState(() => threadId);
  // Ref to track current thread ID across async callbacks without causing re-renders,
  // and to allow access to the current thread id in onUpdateEvent
  const threadIdRef = useRef<string | null>(threadId ?? null);
  const startedRef = useRef(false);
  const lastStreamActivityAtRef = useRef(0);
  const streamReconciliationInFlightRef = useRef(false);
  const streamReconnectAttemptsRef = useRef(0);
  const streamReconnectFailureToastShownRef = useRef(false);
  const activeRunIdRef = useRef<string | null>(
    threadId && typeof window !== "undefined" ? readActiveRunId(threadId) : null,
  );
  const confirmOrRejoinRef = useRef<
    (reason: "finish" | "error" | "poll" | "conflict") => Promise<
      "terminal" | "rejoined" | "unknown"
    >
  >(async () => "unknown");
  // Keep the send button in "stop" until we confirm the backend run is
  // terminal. A premature SSE close must not look like the turn ended.
  const [streamHoldOpen, setStreamHoldOpen] = useState(
    () =>
      Boolean(
        threadId && typeof window !== "undefined" && readActiveRunId(threadId),
      ),
  );
  const sendInFlightRef = useRef(false);
  const sendEpochRef = useRef(0);
  // Filled after useStream; onFinish/onError may read it asynchronously.
  const threadStreamRef = useRef<{
    isLoading: boolean;
    stop: () => void | Promise<void>;
    joinStream: (runId: string) => Promise<unknown>;
    messages: Message[];
    values?: AgentThreadState;
  } | null>(null);

  const markStreamActivity = useCallback(() => {
    lastStreamActivityAtRef.current = Date.now();
    streamReconnectAttemptsRef.current = 0;
    streamReconnectFailureToastShownRef.current = false;
  }, []);

  const listeners = useRef({
    onStart,
    onFinish,
    onToolStart,
    onToolEnd,
  });

  // Keep listeners ref updated with latest callbacks
  useEffect(() => {
    listeners.current = { onStart, onFinish, onToolStart, onToolEnd };
  }, [onStart, onFinish, onToolStart, onToolEnd]);

  useEffect(() => {
    const normalizedThreadId = threadId ?? null;
    if (!normalizedThreadId) {
      // Reset when the UI moves back to a brand new unsaved thread.
      startedRef.current = false;
      setOnStreamThreadId(normalizedThreadId);
    } else {
      setOnStreamThreadId(normalizedThreadId);
    }
    threadIdRef.current = normalizedThreadId;
  }, [threadId]);

  const _handleOnStart = useCallback((id: string) => {
    if (!startedRef.current) {
      listeners.current.onStart?.(id);
      startedRef.current = true;
    }
  }, []);

  const handleStreamStart = useCallback(
    (_threadId: string) => {
      threadIdRef.current = _threadId;
      markStreamActivity();
      _handleOnStart(_threadId);
    },
    [_handleOnStart, markStreamActivity],
  );

  const queryClient = useQueryClient();
  const updateSubtask = useUpdateSubtask();
  const navigate = useNavigate();
  const runMetadataStorageRef = useRef<
    ReturnType<typeof getRunMetadataStorage> | undefined
  >(undefined);

  if (
    typeof window !== "undefined" &&
    runMetadataStorageRef.current === undefined
  ) {
    runMetadataStorageRef.current = getRunMetadataStorage();
  }

  const thread = useStream<AgentThreadState>({
    client: getAPIClient(isMock),
    assistantId: assistantId ?? "lead_agent",
    threadId: onStreamThreadId,
    reconnectOnMount: runMetadataStorageRef.current
      ? () => runMetadataStorageRef.current!
      : false,
    fetchStateHistory: { limit: 1 },
    onCreated(meta) {
      handleStreamStart(meta.thread_id);
      setOnStreamThreadId(meta.thread_id);
      if (meta.run_id) {
        activeRunIdRef.current = meta.run_id;
        persistActiveRunId(meta.thread_id, meta.run_id);
        setStreamHoldOpen(true);
      }
      const metadataAgentId = context.agent_id ?? context.agent_name;
      const extraMetadata =
        threadMetadata && typeof threadMetadata === "object" ? threadMetadata : {};
      const metadata: Record<string, unknown> = {
        ...(metadataAgentId ? { agent_id: metadataAgentId } : {}),
        ...extraMetadata,
      };
      if (Object.keys(metadata).length > 0 && !isMock) {
        void getAPIClient()
          .threads.update(meta.thread_id, { metadata })
          .catch(() => ({}));
      }
    },
    onLangChainEvent(event) {
      markStreamActivity();
      // 操作日志(Y-Log)：每一次大模型调用结束都各上报一条（一次 LLM 调用 = 一条），
      // 让问答日志尽量「多、全」——一个问答轮次里 lead agent / 子代理的每次模型调用
      // 都会落一条。门控与失败静默都在 writeYLog 内部处理，绝不影响问答。
      if (event.event === "on_chat_model_end") {
        try {
          const data = (event.data ?? {}) as {
            input?: unknown;
            output?: unknown;
          };
          const answer = textFromChatModelOutput(data.output);
          const question = lastHumanFromChatModelInput(data.input);
          writeYLog({
            optType: OptType.other,
            optDetail: "CM助手-全量-智能问答",
            url: `/api/threads/${threadIdRef.current ?? ""}/runs/stream`,
            reqParam: { model: event.name ?? "", question },
            returnParam: { data: answer },
          });
        } catch {
          /* 上报绝不影响问答 */
        }
      }
    },
    // The SDK emits UI tool lifecycle events on the dedicated `tools` stream
    // mode. `events` is a separate LangChain trace channel; using it for UI
    // reactions can miss a tool that the conversation has already rendered.
    onToolEvent:
      onToolStart || onToolEnd
        ? (event) => {
            markStreamActivity();
            if (event.event === "on_tool_start") {
              listeners.current.onToolStart?.({
                name: event.name,
                data: event.input,
                toolCallId: event.toolCallId,
              });
              return;
            }
            if (event.event === "on_tool_end") {
              listeners.current.onToolEnd?.({
                name: event.name,
                data: event.output,
                toolCallId: event.toolCallId,
              });
            }
          }
        : undefined,
    onUpdateEvent(data) {
      markStreamActivity();
      const updates: Array<Partial<AgentThreadState> | null> = Object.values(
        data || {},
      );
      for (const update of updates) {
        if (update && "title" in update && update.title) {
          const nextTitle = update.title;
          updateCachedThreadLists(queryClient, (threads) =>
            threads.map((t) =>
              t.thread_id === threadIdRef.current
                ? {
                    ...t,
                    values: {
                      ...t.values,
                      title: nextTitle,
                    },
                  }
                : t,
            ),
          );
        }
      }
    },
    onCustomEvent(event: unknown) {
      markStreamActivity();
      if (
        typeof event === "object" &&
        event !== null &&
        "type" in event &&
        event.type === "task_running"
      ) {
        const e = event as {
          type: "task_running";
          task_id: string;
          message: AIMessage;
        };
        updateSubtask({ id: e.task_id, latestMessage: e.message });
        return;
      }

      if (
        typeof event === "object" &&
        event !== null &&
        "type" in event &&
        event.type === "llm_retry" &&
        "message" in event &&
        typeof event.message === "string" &&
        event.message.trim()
      ) {
        // The backend is already handling this with bounded retry/backoff.
        // Surfacing every transient model retry as a toast made normal model
        // stalls look like repeated "network fluctuation" failures.
        return;
      }

      // Context compression about to run a (blocking) summarization LLM call —
      // surface a transient hint so the wait before the answer streams isn't a
      // silent dead air. Emitted by DeerFlowSummarizationMiddleware.
      if (
        typeof event === "object" &&
        event !== null &&
        "type" in event &&
        event.type === "context_compacting" &&
        "message" in event &&
        typeof event.message === "string" &&
        event.message.trim()
      ) {
        const e = event as { type: "context_compacting"; message: string };
        toast(e.message);
      }
    },
    onError(error) {
      markStreamActivity();
      const message = getStreamErrorMessage(error);
      // Replacing a stalled local SSE reader aborts the in-flight fetch. That
      // is expected and must not look like a user-facing "network" failure.
      if (isAbortLikeError(error)) {
        return;
      }
      // A dropped SSE while the backend run is still alive: rejoin instead of
      // toasting "Failed to fetch" / 网络波动 and leaving the answer half-done.
      if (isTransientStreamTransportError(message)) {
        setStreamHoldOpen(true);
        void confirmOrRejoinRef.current("error");
        return;
      }
      setOptimisticMessages([]);
      // A stale auto-reconnect ("Run <id> not found") after a backend restart:
      // drop the dirty ``lg:stream:{threadId}`` key so switching back into this
      // conversation stops re-firing the same 404, and render from history
      // silently — there is no live run to rejoin.
      if (isRunNotFoundError(message)) {
        const staleThreadId = threadIdRef.current ?? onStreamThreadId ?? null;
        if (staleThreadId) {
          clearActiveRunId(staleThreadId);
        }
        activeRunIdRef.current = null;
        setStreamHoldOpen(false);
        return;
      }
      // A "thread not found" 404 means this route cannot accept another turn.
      // Move to a fresh chat without a full browser reload; the old reload path
      // made every click look like a crash and could loop indefinitely when a
      // different tab's cookie kept overriding this tab's Bearer token.
      if (isThreadNotAccessibleError(message)) {
        const userLabel = buildCurrentUserLabel();
        toast.info(
          `当前登录用户为 ${userLabel}，原对话不存在或无权访问，已为你打开新对话。`,
        );
        // Prefer staying inside an agent chat context when the user is on an
        // agent thread URL — otherwise fall back to the global new chat.
        const hash = typeof window !== "undefined" ? window.location.hash : "";
        const agentMatch = hash.match(/#\/page\/workspace\/agents\/([^/]+)\//);
        const iframeMatch = hash.match(/#\/page\/workspace\/chats_iframe\//);
        const target = agentMatch
          ? `/page/workspace/agents/${agentMatch[1]}/chats/new`
          : iframeMatch
            ? "/page/workspace/chats_iframe/new"
            : "/page/workspace/chats/new";
        navigate(target, { replace: true });
        return;
      }
      // 上一轮还没结束又发了新问题（409）或并发达到上限（429）：
      // 给一句明确的中文提示，而不是把原始 HTTP 错误抛给用户。
      const friendly = friendlySendLimitMessage(message);
      if (friendly) {
        if (isActiveRunConflictError(message)) {
          setStreamHoldOpen(true);
          void confirmOrRejoinRef.current("conflict").then((result) => {
            toast.info(
              result === "rejoined"
                ? "上一轮回答仍在生成，已重新连接当前回复。请等待结束或点“停止”后再提问。"
                : friendly,
            );
          });
          return;
        }
        toast.info(friendly);
        return;
      }
      toast.error(message);
    },
    onFinish(state) {
      lastStreamActivityAtRef.current = Date.now();
      // Do not trust a clean SSE close as "the run is done". streamResumable
      // chats keep the backend run alive on disconnect; the SDK still fires
      // onFinish and would otherwise restore the send button.
      //
      // If a newer turn is already streaming (common: 用户协助卡提交后，上一轮
      // interrupt 的迟到 onFinish 才到), do not freeze/rejoin — that would
      // steal the live SSE and dump tokens only at the end.
      if (threadStreamRef.current?.isLoading || sendInFlightRef.current) {
        return;
      }
      setStreamHoldOpen(true);
      void confirmOrRejoinRef.current("finish").then((result) => {
        if (result === "rejoined") {
          return;
        }
        if (threadStreamRef.current?.isLoading || sendInFlightRef.current) {
          return;
        }
        setStreamHoldOpen(false);
        listeners.current.onFinish?.(state.values);
        void queryClient.invalidateQueries({ queryKey: ["threads", "search"] });
      });
    },
  });

  // Optimistic messages shown before the server stream responds
  const [optimisticMessages, setOptimisticMessages] = useState<Message[]>([]);
  const [isUploading, setIsUploading] = useState(false);
  const [threadValueOverrides, setThreadValueOverrides] = useState<
    Partial<AgentThreadState>
  >({});
  // Carries the ask time of an in-flight send until the server-assigned human
  // message id shows up, so we can bind the timestamp to the real id (see the
  // reconciliation effect below). ``known`` is the set of human ids that
  // already existed at send time, used to spot the newly-arrived one.
  const pendingAskTimeRef = useRef<{ ts: string; known: Set<string> } | null>(
    null,
  );
  // Human ids known at optimistic-send time; used to clear optimistic only when
  // a *new* server human appears (not when length bumps from tool/checkpoint).
  const optimisticKnownHumanIdsRef = useRef<Set<string>>(new Set());
  // Track message count before sending so we know when server has responded
  const prevMsgCountRef = useRef(thread.messages.length);

  threadStreamRef.current = thread;

  const confirmOrRejoin = useCallback(
    async (
      reason: "finish" | "error" | "poll" | "conflict",
    ): Promise<"terminal" | "rejoined" | "unknown"> => {
      const tid = threadIdRef.current ?? onStreamThreadId ?? null;
      if (!tid) {
        setStreamHoldOpen(false);
        return "unknown";
      }

      if (streamReconciliationInFlightRef.current && reason === "poll") {
        return "unknown";
      }
      // Stale finish/error from a prior interrupt must not joinStream over a
      // turn the user already started (协助卡提交后的实时流会被打断并攒批)。
      if (
        (reason === "finish" || reason === "error") &&
        (sendInFlightRef.current || threadStreamRef.current?.isLoading)
      ) {
        return "rejoined";
      }
      streamReconciliationInFlightRef.current = true;
      const epochAtStart = sendEpochRef.current;
      try {
        const client = getAPIClient(isMock);
        let runId = activeRunIdRef.current ?? readActiveRunId(tid);
        const terminalRunId = runId;

        const applyTerminalState = async (status: string) => {
          const finalState = await client.threads.getState(tid);
          if (
            shouldDiscardTerminalOverlay({
              epochAtStart,
              currentEpoch: sendEpochRef.current,
              sendInFlight: sendInFlightRef.current,
              terminalRunId,
              currentRunId: activeRunIdRef.current,
            })
          ) {
            return;
          }
          clearActiveRunId(tid);
          activeRunIdRef.current = null;
          setStreamHoldOpen(false);
          streamReconnectAttemptsRef.current = 0;
          streamReconnectFailureToastShownRef.current = false;
          setThreadValueOverrides(
            finalState.values as Partial<AgentThreadState>,
          );
          if (status === "success") {
            setOptimisticMessages([]);
            if (reason !== "finish") {
              listeners.current.onFinish?.(
                finalState.values as AgentThreadState,
              );
              void queryClient.invalidateQueries({
                queryKey: ["threads", "search"],
              });
            }
          } else if (reason !== "finish") {
            toast.error(
              "本轮生成未能完成，已同步当前对话内容，可重新发起请求。",
            );
          }
        };

        if (runId) {
          try {
            const run = await client.runs.get(tid, runId);
            if (TERMINAL_RUN_STATUSES.has(run.status)) {
              await applyTerminalState(run.status);
              return "terminal";
            }
          } catch {
            runId = null;
          }
        }

        if (!runId) {
          try {
            const runs = await client.runs.list(tid, { limit: 20 });
            const inflight = runs.find((item) =>
              isInflightRunStatus(item.status),
            );
            runId = inflight?.run_id ?? null;
            if (runId) {
              activeRunIdRef.current = runId;
            }
          } catch {
            runId = null;
          }
        }

        if (!runId) {
          setStreamHoldOpen(false);
          activeRunIdRef.current = null;
          return "unknown";
        }

        persistActiveRunId(tid, runId);
        setStreamHoldOpen(true);

        const streamAboutToEnd = reason === "finish" || reason === "error";
        if (
          threadStreamRef.current?.isLoading &&
          !streamAboutToEnd &&
          reason !== "conflict"
        ) {
          if (
            Date.now() - lastStreamActivityAtRef.current <
            STREAM_RECONNECT_IDLE_MS
          ) {
            return "rejoined";
          }
          // Local SSE reader stalled while the backend run is still alive.
          // SDK stop() cancels the server run if lg:stream is set — hide it
          // first, abort only the browser reader, then join the live run.
          if (reason === "poll") {
            if (
              streamReconnectAttemptsRef.current >= STREAM_RECONNECT_MAX_ATTEMPTS
            ) {
              if (!streamReconnectFailureToastShownRef.current) {
                streamReconnectFailureToastShownRef.current = true;
                toast.error(
                  "回答仍在后台处理中，但页面连接暂未恢复；请稍后返回此会话查看结果，或点“停止”后重新提问。",
                );
              }
              return "rejoined";
            }
            streamReconnectAttemptsRef.current += 1;
            lastStreamActivityAtRef.current = Date.now();
            const stream = threadStreamRef.current;
            if (!stream) {
              return "unknown";
            }
            runMetadataStorageRef.current?.removeItem(`lg:stream:${tid}`);
            void Promise.resolve(stream.stop())
              .catch(() => undefined)
              .finally(() => {
                if (threadIdRef.current !== tid) return;
                persistActiveRunId(tid, runId);
                void stream.joinStream(runId).catch(() => undefined);
              });
            return "rejoined";
          }
        }

        if (reason !== "conflict") {
          if (
            streamReconnectAttemptsRef.current >= STREAM_RECONNECT_MAX_ATTEMPTS
          ) {
            if (!streamReconnectFailureToastShownRef.current) {
              streamReconnectFailureToastShownRef.current = true;
              toast.error(
                "回答仍在后台处理中，但页面连接暂未恢复；请稍后返回此会话查看结果，或点“停止”后重新提问。",
              );
            }
            return "rejoined";
          }
          streamReconnectAttemptsRef.current += 1;
        }
        lastStreamActivityAtRef.current = Date.now();

        const stream = threadStreamRef.current;
        if (!stream) {
          return "unknown";
        }
        void stream.joinStream(runId).catch(() => {
          /* poll retries while the run stays non-terminal */
        });
        return "rejoined";
      } finally {
        streamReconciliationInFlightRef.current = false;
      }
    },
    [isMock, onStreamThreadId, queryClient],
  );
  confirmOrRejoinRef.current = confirmOrRejoin;

  useEffect(() => {
    const activeThreadId = threadIdRef.current ?? onStreamThreadId ?? null;
    if (!activeThreadId || typeof window === "undefined") {
      return;
    }

    let disposed = false;
    const reconcileRun = async () => {
      if (disposed) return;
      const dropped = !threadStreamRef.current?.isLoading;
      const idleFor = Date.now() - lastStreamActivityAtRef.current;
      if (!dropped && idleFor < STREAM_RECONCILIATION_IDLE_MS) {
        return;
      }
      if (dropped && idleFor < 800) {
        return;
      }
      const stored = activeRunIdRef.current ?? readActiveRunId(activeThreadId);
      if (!stored && !streamHoldOpen && threadStreamRef.current?.isLoading) {
        return;
      }
      if (!stored && !streamHoldOpen) {
        return;
      }
      await confirmOrRejoin("poll");
    };

    const interval = window.setInterval(
      () => void reconcileRun(),
      STREAM_RECONCILIATION_POLL_MS,
    );

    return () => {
      disposed = true;
      window.clearInterval(interval);
    };
  }, [confirmOrRejoin, onStreamThreadId, streamHoldOpen, thread.isLoading]);

  // Reset thread-local pending UI state when the caller actually switches AWAY
  // from a thread (threadId becomes nullish). This is the "click 新对话 / open
  // a fresh chat" signal — abort the in-flight SSE and clear optimistic state.
  //
  // We deliberately do NOT reset on every threadId change. During a normal
  // "user just sent the first message on /chats/new" flow, useThreadStream
  // sees threadId transition from undefined (the landing state) to the
  // server-assigned thread id once onCreated fires. Stopping the stream and
  // clearing optimistic messages in that case is exactly what made submitted
  // questions visually disappear — the live run that just started would be
  // aborted by the very effect intended to reset stale state.
  useEffect(() => {
    if (threadId != null) {
      // Forward direction: starting or continuing with a real thread.
      // No reset needed; the upstream useStream will load history for this id.
      setThreadValueOverrides({});
      return;
    }
    const prevThread = threadStreamRef.current;
    if (prevThread?.isLoading) {
      try {
        prevThread.stop();
      } catch {
        // best-effort; nothing useful to do if the SDK fails to abort
      }
    }
    startedRef.current = false;
    sendInFlightRef.current = false;
    sendEpochRef.current += 1;
    prevMsgCountRef.current = 0;
    optimisticKnownHumanIdsRef.current = new Set();
    activeRunIdRef.current = null;
    setStreamHoldOpen(false);
    setOptimisticMessages([]);
    setIsUploading(false);
    setThreadValueOverrides({});
  }, [threadId]);

  useEffect(() => {
    const activeThreadId = threadIdRef.current ?? onStreamThreadId ?? null;
    const effectiveTitleProvisional =
      (threadValueOverrides.title_provisional ??
        thread.values?.title_provisional) === true;

    if (!activeThreadId || thread.isLoading || !effectiveTitleProvisional) {
      return;
    }

    let cancelled = false;
    let timeoutId: ReturnType<typeof setTimeout> | null = null;
    const controller = new AbortController();

    const poll = async () => {
      while (!cancelled) {
        try {
          const nextState = await fetchThreadTitleRefreshState(
            activeThreadId,
            controller.signal,
          );
          if (cancelled || !nextState) {
            return;
          }

          setThreadValueOverrides((prev) => ({
            ...prev,
            title: nextState.title,
            title_provisional: nextState.title_provisional,
          }));

          if (nextState.title) {
            updateCachedThreadLists(queryClient, (threads) =>
              threads.map((t) =>
                t.thread_id === activeThreadId
                  ? {
                      ...t,
                      values: {
                        ...t.values,
                        title: nextState.title,
                        title_provisional: nextState.title_provisional,
                      },
                    }
                  : t,
              ),
            );
          }

          if (nextState.title_provisional !== true) {
            void queryClient.invalidateQueries({ queryKey: ["threads", "search"] });
            return;
          }
        } catch {
          return;
        }

        await new Promise<void>((resolve) => {
          timeoutId = setTimeout(resolve, 2000);
        });
      }
    };

    void poll();

    return () => {
      cancelled = true;
      controller.abort();
      if (timeoutId) {
        clearTimeout(timeoutId);
      }
    };
  }, [
    onStreamThreadId,
    queryClient,
    thread.isLoading,
    thread.values?.title_provisional,
    threadValueOverrides.title_provisional,
  ]);

  // Clear optimistic only when the server echoes a *new* human message.
  // After 用户协助, a late checkpoint / tool result can bump length without
  // the new answer yet — clearing then makes the bubble flash and vanish.
  useEffect(() => {
    if (optimisticMessages.length === 0) {
      return;
    }
    if (
      hasServerEchoedOptimisticHuman(
        thread.messages,
        optimisticKnownHumanIdsRef.current,
      )
    ) {
      setOptimisticMessages([]);
    }
  }, [thread.messages, optimisticMessages.length]);

  // Bind a just-sent question's ask time to the server-assigned human message
  // id once it appears. Persists to localStorage so the time shows on every
  // question and survives reloads, independent of the backend echoing it back.
  useEffect(() => {
    const pending = pendingAskTimeRef.current;
    if (!pending) return;
    for (const message of thread.messages) {
      if (
        message.type === "human" &&
        message.id &&
        !message.id.startsWith("opt-") &&
        !pending.known.has(message.id)
      ) {
        rememberAskTime(message.id, pending.ts);
        pendingAskTimeRef.current = null;
        break;
      }
    }
  }, [thread.messages]);

  const sendMessage = useCallback(
    async (
      threadId: string,
      message: PromptInputMessage,
      extraContext?: Record<string, unknown>,
      options?: SendMessageOptions,
    ) => {
      if (sendInFlightRef.current) {
        return;
      }
      sendInFlightRef.current = true;
      sendEpochRef.current += 1;
      // Drop frozen checkpoint messages so the new SSE can paint live tokens.
      // Keep title / title_provisional overlays from the prior reconcile.
      setThreadValueOverrides((prev) => {
        if (!("messages" in prev)) {
          return prev;
        }
        const { messages: _dropped, ...rest } = prev;
        return rest;
      });
      setStreamHoldOpen(false);

      const text = message.text.trim();
      // Wall-clock time the question was asked. Stored on the human message so
      // the chat can show "几点提问" and have it survive a reload (it round-trips
      // through the checkpoint inside additional_kwargs, same as `files`).
      const clientTs = new Date().toISOString();
      const knownHumanIds = collectHumanMessageIds(
        thread.messages,
        Array.isArray(thread.values?.messages)
          ? (thread.values.messages as Message[])
          : undefined,
      );
      // Remember which human messages already exist so the reconciliation
      // effect can bind ``clientTs`` to the new one once the server returns it.
      pendingAskTimeRef.current = {
        ts: clientTs,
        known: knownHumanIds,
      };
      optimisticKnownHumanIdsRef.current = knownHumanIds;

      // Capture current count before showing optimistic messages
      prevMsgCountRef.current = thread.messages.length;

      // Build optimistic files list with uploading status
      const optimisticFiles: FileInMessage[] = (message.files ?? []).map(
        (f) => ({
          filename: f.filename ?? "",
          size: 0,
          status: "uploading" as const,
        }),
      );

      const hideFromUI = options?.additionalKwargs?.hide_from_ui === true;
      const optimisticAdditionalKwargs = {
        ...options?.additionalKwargs,
        client_ts: clientTs,
        ...(optimisticFiles.length > 0 ? { files: optimisticFiles } : {}),
      };

      const newOptimistic: Message[] = [];
      if (!hideFromUI) {
        newOptimistic.push({
          type: "human",
          id: `opt-human-${Date.now()}`,
          content: text ? [{ type: "text", text }] : "",
          additional_kwargs: optimisticAdditionalKwargs,
        });
      }

      if (optimisticFiles.length > 0 && !hideFromUI) {
        // Mock AI message while files are being uploaded
        newOptimistic.push({
          type: "ai",
          id: `opt-ai-${Date.now()}`,
          content: t.uploads.uploadingFiles,
          additional_kwargs: { element: "task" },
        });
      }
      setOptimisticMessages(newOptimistic);

      // Only fire onStart immediately for an existing persisted thread.
      // Brand-new chats should wait for onCreated(meta.thread_id) so URL sync
      // uses the real server-generated thread id.
      if (threadIdRef.current) {
        _handleOnStart(threadId);
      }

      let uploadedFileInfo: UploadedFileInfo[] = [];

      try {
        // Upload files first if any
        if (message.files && message.files.length > 0) {
          setIsUploading(true);
          try {
            const filePromises = message.files.map((fileUIPart) =>
              promptInputFilePartToFile(fileUIPart),
            );

            const conversionResults = await Promise.all(filePromises);
            const files = conversionResults.filter(
              (file): file is File => file !== null,
            );
            const failedConversions = conversionResults.length - files.length;

            if (failedConversions > 0) {
              throw new Error(
                `Failed to prepare ${failedConversions} attachment(s) for upload. Please retry.`,
              );
            }

            if (!threadId) {
              throw new Error("Thread is not ready for file upload.");
            }

            if (files.length > 0) {
              const uploadResponse = await uploadFiles(threadId, files);
              uploadedFileInfo = uploadResponse.files;

              // Update optimistic human message with uploaded status + paths
              const uploadedFiles: FileInMessage[] = uploadedFileInfo.map(
                (info) => ({
                  filename: info.filename,
                  size: info.size,
                  path: info.virtual_path,
                  status: "uploaded" as const,
                }),
              );
              setOptimisticMessages((messages) => {
                if (messages.length > 1 && messages[0]) {
                  const humanMessage: Message = messages[0];
                  return [
                    {
                      ...humanMessage,
                      additional_kwargs: {
                        ...humanMessage.additional_kwargs,
                        files: uploadedFiles,
                      },
                    },
                    ...messages.slice(1),
                  ];
                }
                return messages;
              });
            }
          } catch (error) {
            const errorMessage =
              error instanceof Error
                ? error.message
                : "Failed to upload files.";
            toast.error(errorMessage);
            setOptimisticMessages([]);
            throw error;
          } finally {
            setIsUploading(false);
          }
        }

        // Build files metadata for submission (included in additional_kwargs)
        const filesForSubmit: FileInMessage[] = uploadedFileInfo.map(
          (info) => ({
            filename: info.filename,
            size: info.size,
            path: info.virtual_path,
            status: "uploaded" as const,
          }),
        );

        const mergedContext = {
          ...context,
          ...extraContext,
        };
        const selectedLlmWikiIds = Array.isArray(
          mergedContext.llmwiki_knowledge_base_ids,
        )
          ? mergedContext.llmwiki_knowledge_base_ids
          : [];
        const ragModeEnabled =
          localSettings.display.ragModeEnabled ||
          isThreadRagModeLocked(threadId) ||
          selectedLlmWikiIds.length > 0;
        if (ragModeEnabled) {
          lockThreadRagMode(threadId);
        }

        const contextCompressionEnabled =
          localSettings.display.contextCompressionEnabled ?? false;

        const hiddenContextMessages = normalizeHiddenContextMessages(options);
        const submittedMessages: Message[] = [
          ...hiddenContextMessages.map((item) => ({
            type: "human" as const,
            ...(item.name ? { name: item.name } : {}),
            content: [
              {
                type: "text" as const,
                text: item.text,
              },
            ],
            additional_kwargs: {
              ...item.additionalKwargs,
              hide_from_ui: true,
            },
          })),
          {
            type: "human" as const,
            content: [
              {
                type: "text" as const,
                text,
              },
            ],
            additional_kwargs: {
              ...options?.additionalKwargs,
              client_ts: clientTs,
              ...(filesForSubmit.length > 0
                ? { files: filesForSubmit }
                : {}),
            },
          },
        ];

        await thread.submit(
          {
            messages: submittedMessages,
          },
          {
            threadId: threadId,
            streamSubgraphs: true,
            streamResumable: true,
            onDisconnect: runOptions?.onDisconnect,
            multitaskStrategy: runOptions?.multitaskStrategy,
            config: {
              recursion_limit: 1000,
            },
            context: {
              ...mergedContext,
              thinking_enabled: mergedContext.mode !== "flash",
              is_plan_mode:
                mergedContext.mode === "pro" || mergedContext.mode === "ultra",
              subagent_enabled: mergedContext.mode === "ultra",
              reasoning_effort:
                mergedContext.reasoning_effort ??
                (mergedContext.mode === "ultra"
                  ? "high"
                  : mergedContext.mode === "pro"
                    ? "medium"
                    : mergedContext.mode === "thinking"
                      ? "low"
                      : undefined),
              memory_injection_enabled:
                mergedContext.memory_injection_enabled ?? true,
              rag_mode_enabled: ragModeEnabled,
              summarization_enabled: contextCompressionEnabled,
              thread_id: threadId,
            },
          },
        );
        void queryClient.invalidateQueries({ queryKey: ["threads", "search"] });
      } catch (error) {
        const message =
          error instanceof Error ? error.message : String(error);
        const friendly = friendlySendLimitMessage(message);
        if (friendly) {
          setIsUploading(false);
          if (isActiveRunConflictError(message)) {
            setOptimisticMessages([]);
            setStreamHoldOpen(true);
            const result = await confirmOrRejoin("conflict");
            toast.info(
              result === "rejoined"
                ? "上一轮回答仍在生成，已重新连接当前回复。请等待结束或点“停止”后再提问。"
                : friendly,
            );
            return;
          }
          setOptimisticMessages((messages) => {
            const human = messages.find((item) => item.type === "human");
            const kept = human ? [human] : [];
            return [
              ...kept,
              {
                type: "ai",
                id: `opt-ai-limit-${Date.now()}`,
                content: friendly,
                additional_kwargs: { element: "task" },
              } as Message,
            ];
          });
          return;
        }
        setOptimisticMessages([]);
        setIsUploading(false);
        throw error;
      } finally {
        sendInFlightRef.current = false;
      }
    },
    [thread, _handleOnStart, t.uploads.uploadingFiles, context, localSettings.display.ragModeEnabled, queryClient, runOptions, confirmOrRejoin],
  );

  const handleThreadStop = useCallback(async () => {
    const tid = threadIdRef.current;
    const runId =
      activeRunIdRef.current ?? (tid ? readActiveRunId(tid) : null);
    setStreamHoldOpen(false);
    activeRunIdRef.current = null;
    if (tid && runId) {
      persistActiveRunId(tid, runId);
    }
    try {
      await Promise.resolve(threadStreamRef.current?.stop());
    } catch {
      /* local abort is best-effort */
    }
    if (tid && runId) {
      try {
        await getAPIClient(isMock).runs.cancel(tid, runId);
      } catch {
        /* already stopped / missing */
      }
      clearActiveRunId(tid);
    }
  }, [isMock]);

  // Merge thread with optimistic messages for display. While a live SSE is
  // open, never let a stale getState snapshot's `messages` win — that is what
  // made 协助卡提交后「发送中」攒到结束才一口气出字。
  const mergedValues =
    mergeStreamDisplayValues(
      thread.values as AgentThreadState | undefined,
      threadValueOverrides,
      thread.isLoading,
    ) ?? thread.values;

  const mergedMessages =
    Array.isArray(mergedValues?.messages)
      ? (mergedValues.messages as Message[])
      : thread.messages;

  const mergedThread = {
    ...thread,
    values: mergedValues,
    messages:
      optimisticMessages.length > 0
        ? [...mergedMessages, ...optimisticMessages]
        : mergedMessages,
    isLoading: thread.isLoading || streamHoldOpen,
    stop: handleThreadStop,
  } as typeof thread;

  // 第 4 个返回值 `onStreamThreadId` = 底层 useStream 当前实际绑定的线程 id（新建/未保存时为 null）。
  // 调用方据此判断「流是否已切到全新线程」，避免在切换会话的那一帧把消息发进上一个会话
  // （submit 的 threadId 选项仅在 hook 线程为 null/undefined 时才生效）。
  return [mergedThread, sendMessage, isUploading, onStreamThreadId] as const;
}

export function useThreads(
  params: Parameters<ThreadsClient["search"]>[0] = {
    limit: 50,
    sortBy: "updated_at",
    sortOrder: "desc",
    select: ["thread_id", "updated_at", "values", "metadata"],
  },
) {
  const apiClient = getAPIClient();
  return useQuery<AgentThread[]>({
    queryKey: ["threads", "search", params],
    queryFn: async () => {
      const maxResults = params.limit;
      const initialOffset = params.offset ?? 0;
      const DEFAULT_PAGE_SIZE = 50;

      // Preserve prior semantics: if a non-positive limit is explicitly provided,
      // delegate to a single search call with the original parameters.
      if (maxResults !== undefined && maxResults <= 0) {
        const response =
          await apiClient.threads.search<AgentThreadState>(params);
        return response as AgentThread[];
      }

      const pageSize =
        typeof maxResults === "number" && maxResults > 0
          ? Math.min(DEFAULT_PAGE_SIZE, maxResults)
          : DEFAULT_PAGE_SIZE;

      const threads: AgentThread[] = [];
      let offset = initialOffset;

      while (true) {
        if (typeof maxResults === "number" && threads.length >= maxResults) {
          break;
        }

        const currentLimit =
          typeof maxResults === "number"
            ? Math.min(pageSize, maxResults - threads.length)
            : pageSize;

        if (typeof maxResults === "number" && currentLimit <= 0) {
          break;
        }

        const response = (await apiClient.threads.search<AgentThreadState>({
          ...params,
          limit: currentLimit,
          offset,
        })) as AgentThread[];

        threads.push(...response);

        if (response.length < currentLimit) {
          break;
        }

        offset += response.length;
      }

      // Safety net on top of the backend's system-thread exclusion: drop
      // machine-created / wizard threads (scheduled tasks, report-structure
      // setup, writing setup, notebooks, …) in case an older backend still
      // returns them.
      return threads.filter((t) => !isHiddenFromRecentChats(t));
    },
    refetchOnWindowFocus: false,
  });
}

export interface ThreadsPageResult {
  threads: AgentThread[];
  hasMore: boolean;
}

/**
 * Apply ``transform`` to the thread array inside every cached
 * ``["threads", "search"]`` entry. The sidebar queries cache a plain
 * ``AgentThread[]`` while the paged chats list (`useThreadsPage`) caches a
 * ``ThreadsPageResult`` object under the same key prefix, so fuzzy
 * ``setQueriesData`` updaters must handle both shapes.
 */
function updateCachedThreadLists(
  queryClient: QueryClient,
  transform: (threads: AgentThread[]) => AgentThread[],
) {
  queryClient.setQueriesData<AgentThread[] | ThreadsPageResult | undefined>(
    { queryKey: ["threads", "search"], exact: false },
    (oldData) => {
      if (Array.isArray(oldData)) {
        return transform(oldData);
      }
      if (oldData && Array.isArray(oldData.threads)) {
        return { ...oldData, threads: transform(oldData.threads) };
      }
      return oldData;
    },
  );
}

/**
 * Server-side paginated conversation list for the chats page. Unlike
 * `useThreads` (which loads a bounded recent window for the sidebar), this
 * fetches exactly one page per request so a long history is never pulled in
 * one go. `search` matches thread titles on the backend; one extra row is
 * over-fetched only to detect whether a next page exists. The query key stays
 * under ["threads", "search"] so existing invalidations refresh pages too.
 */
export function useThreadsPage(options: {
  page: number;
  pageSize: number;
  search?: string;
  metadata?: Record<string, unknown>;
}) {
  const { page, pageSize, search, metadata } = options;
  const normalizedSearch = search?.trim() ?? "";
  return useQuery<ThreadsPageResult>({
    queryKey: [
      "threads",
      "search",
      "paged",
      page,
      pageSize,
      normalizedSearch,
      metadata ?? null,
    ],
    queryFn: async () => {
      // Direct fetch instead of the LangGraph SDK: the SDK's threads.search
      // drops fields it doesn't know about, and `query` (title search) is a
      // gateway extension.
      const response = await apiFetch(
        `${getLangGraphBaseURL()}/threads/search`,
        {
          method: "POST",
          headers: { "Content-Type": "application/json" },
          body: JSON.stringify({
            limit: pageSize + 1,
            offset: (page - 1) * pageSize,
            metadata: metadata ?? {},
            query: normalizedSearch || undefined,
          }),
        },
      );
      if (!response.ok) {
        throw new Error("Failed to load conversations");
      }
      const rows = (await response.json()) as AgentThread[];
      // 与后端 exclude_system 对齐：普通列表再兜一层。调用方显式按
      // thread_type / system 取数（笔记本等）时不滤，避免把目标会话藏掉。
      const visible = shouldHideSystemThreadsFromList(metadata)
        ? rows.filter((thread) => !isHiddenFromRecentChats(thread))
        : rows;
      return {
        threads: visible.slice(0, pageSize),
        hasMore: rows.length > pageSize,
      };
    },
    placeholderData: (previous) => previous,
    refetchOnWindowFocus: false,
  });
}

/**
 * Total number of conversations matching the same filters as
 * `useThreadsPage`, via the lightweight `/threads/count` endpoint — powers
 * the "共 N 条 / X 页" display without fetching any rows. Depends only on the
 * filters (not the page), so paging doesn't refire it.
 */
export function useThreadsCount(options: {
  search?: string;
  metadata?: Record<string, unknown>;
}) {
  const normalizedSearch = options.search?.trim() ?? "";
  return useQuery<number>({
    queryKey: [
      "threads",
      "search",
      "count",
      normalizedSearch,
      options.metadata ?? null,
    ],
    queryFn: async () => {
      const response = await apiFetch(
        `${getLangGraphBaseURL()}/threads/count`,
        {
          method: "POST",
          headers: { "Content-Type": "application/json" },
          body: JSON.stringify({
            metadata: options.metadata ?? {},
            query: normalizedSearch || undefined,
          }),
        },
      );
      if (!response.ok) {
        throw new Error("Failed to count conversations");
      }
      const data = (await response.json()) as { total: number };
      return data.total;
    },
    refetchOnWindowFocus: false,
  });
}

export function useDeleteThread() {
  const queryClient = useQueryClient();
  const apiClient = getAPIClient();
  return useMutation({
    mutationFn: async ({ threadId }: { threadId: string }) => {
      await apiClient.threads.delete(threadId);

      const response = await apiFetch(
        `${getBackendBaseURL()}/api/threads/${encodeURIComponent(threadId)}`,
        {
          method: "DELETE",
        },
      );

      if (!response.ok) {
        const error = await response
          .json()
          .catch(() => ({ detail: "Failed to delete local thread data." }));
        throw new Error(error.detail ?? "Failed to delete local thread data.");
      }
    },
    onSuccess(_, { threadId }) {
      updateCachedThreadLists(queryClient, (threads) =>
        threads.filter((t) => t.thread_id !== threadId),
      );
    },
    onSettled() {
      void queryClient.invalidateQueries({ queryKey: ["threads", "search"] });
    },
  });
}

/**
 * Delete a conversation turn (a user question together with its answer) from a
 * thread. Hits the Gateway endpoint that removes the question and every
 * following message up to the next question from the latest checkpoint, and
 * returns the ids that were actually removed. The live `useStream` cache is not
 * auto-refreshed by this call — callers hide the returned ids locally (see
 * `MessageList`) so the removal shows immediately; a reload then reads the
 * persisted, already-trimmed state.
 */
export function useDeleteMessage() {
  return useMutation({
    mutationFn: async ({
      threadId,
      messageId,
    }: {
      threadId: string;
      messageId: string;
    }): Promise<string[]> => {
      const response = await apiFetch(
        `${getBackendBaseURL()}/api/threads/${encodeURIComponent(
          threadId,
        )}/messages/${encodeURIComponent(messageId)}`,
        {
          method: "DELETE",
        },
      );

      if (!response.ok) {
        const error = await response
          .json()
          .catch(() => ({ detail: "删除消息失败。" }));
        throw new Error(error.detail ?? "删除消息失败。");
      }

      const body = (await response.json().catch(() => ({}))) as {
        deleted_ids?: string[];
      };
      // Fall back to the requested id so the caller can always hide something.
      return body.deleted_ids?.length ? body.deleted_ids : [messageId];
    },
  });
}

/**
 * Clear all messages from a thread checkpoint. The caller should immediately
 * hide the returned ids locally because the active stream cache may still hold
 * the old message objects until a reload/refetch reads the trimmed checkpoint.
 */
export function useClearThreadMessages() {
  return useMutation({
    mutationFn: async ({
      threadId,
    }: {
      threadId: string;
    }): Promise<string[]> => {
      const response = await apiFetch(
        `${getBackendBaseURL()}/api/threads/${encodeURIComponent(
          threadId,
        )}/messages`,
        {
          method: "DELETE",
        },
      );

      if (!response.ok) {
        const error = await response
          .json()
          .catch(() => ({ detail: "清空对话失败。" }));
        throw new Error(error.detail ?? "清空对话失败。");
      }

      const body = (await response.json().catch(() => ({}))) as {
        deleted_ids?: string[];
      };
      return body.deleted_ids ?? [];
    },
  });
}

/**
 * Persist an in-place message edit in the latest thread checkpoint.
 *
 * This is opt-in UI behaviour: most chat pages keep their existing
 * "edit question and resubmit" affordance, while position roundtable can pass
 * this mutation down to save the original message content itself.
 */
export function useEditMessage() {
  return useMutation({
    mutationFn: async ({
      threadId,
      messageId,
      content,
    }: {
      threadId: string;
      messageId: string;
      content: string;
    }): Promise<Message> => {
      const response = await apiFetch(
        `${getBackendBaseURL()}/api/threads/${encodeURIComponent(
          threadId,
        )}/messages/${encodeURIComponent(messageId)}`,
        {
          method: "PATCH",
          headers: { "Content-Type": "application/json" },
          body: JSON.stringify({ content }),
        },
      );

      if (!response.ok) {
        const error = await response
          .json()
          .catch(() => ({ detail: "消息保存失败。" }));
        throw new Error(error.detail ?? "消息保存失败。");
      }

      const body = (await response.json().catch(() => ({}))) as {
        message?: Message;
      };
      return body.message ?? ({ id: messageId, type: "human", content } as Message);
    },
  });
}

/**
 * Merge metadata into a conversation's record via the Gateway
 * `PATCH /api/threads/{id}` route. Used to persist per-conversation flags such
 * as `is_pinned` (置顶) and `is_favorite` (收藏) — a conversation belongs to a
 * single user, so these live directly on the thread metadata (no join table).
 * Updates the `["threads", "search"]` cache optimistically so the UI reflects
 * the toggle immediately, mirroring `useRenameThread`.
 */
export function useUpdateThreadMetadata() {
  const queryClient = useQueryClient();
  return useMutation({
    mutationFn: async ({
      threadId,
      metadata,
    }: {
      threadId: string;
      metadata: Record<string, unknown>;
    }) => {
      const response = await apiFetch(
        `${getBackendBaseURL()}/api/threads/${encodeURIComponent(threadId)}`,
        {
          method: "PATCH",
          headers: { "Content-Type": "application/json" },
          body: JSON.stringify({ metadata }),
        },
      );
      if (!response.ok) {
        const error = await response
          .json()
          .catch(() => ({ detail: "更新对话失败。" }));
        throw new Error(error.detail ?? "更新对话失败。");
      }
    },
    onSuccess(_, { threadId, metadata }) {
      updateCachedThreadLists(queryClient, (threads) =>
        threads.map((t) =>
          t.thread_id === threadId
            ? {
                ...t,
                metadata: {
                  ...(t.metadata as Record<string, unknown> | null),
                  ...metadata,
                },
              }
            : t,
        ),
      );
    },
  });
}

/** Toggle a conversation's pinned (置顶) flag. */
export function useToggleThreadPin() {
  const { mutate, ...rest } = useUpdateThreadMetadata();
  return {
    ...rest,
    toggle: (threadId: string, pinned: boolean) =>
      mutate({ threadId, metadata: { is_pinned: pinned } }),
  };
}

/** Toggle a conversation's favorite (收藏) flag. */
export function useToggleThreadFavorite() {
  const { mutate, ...rest } = useUpdateThreadMetadata();
  return {
    ...rest,
    toggle: (threadId: string, favorite: boolean) =>
      mutate({ threadId, metadata: { is_favorite: favorite } }),
  };
}

export function useRenameThread() {
  const queryClient = useQueryClient();
  const apiClient = getAPIClient();
  return useMutation({
    mutationFn: async ({
      threadId,
      title,
    }: {
      threadId: string;
      title: string;
    }) => {
      await apiClient.threads.updateState(threadId, {
        values: { title, title_provisional: false },
      });
    },
    onSuccess(_, { threadId, title }) {
      updateCachedThreadLists(queryClient, (threads) =>
        threads.map((t) =>
          t.thread_id === threadId
            ? {
                ...t,
                values: {
                  ...t.values,
                  title,
                  title_provisional: false,
                },
              }
            : t,
        ),
      );
    },
  });
}