用户引导沙箱优化

CLAUDE.md

@@ -88,7 +88,7 @@ The frontend (`frontend-web/`) is a React 19 + Vite SPA:
 - **技能地址管理 (Skill Address Management, admin-only)**: 扫描运行中的技能（`skills/{public,custom}` 下所有 `.md`/`.py`）里的 **http(s) URL 与 IP/IP:端口**，按唯一地址聚合全部出现位置（跨 .md/.py），支持**批量替换**（如把 UAT 地址一次性改成生产）。主体抽成可复用组件 `SkillAddressesPanel`（`pages/SkillAddressesPage.tsx` 导出，默认导出 `SkillAddressesPage` 仍保留旧路由 `/page/strategy/admin/skill-addresses` 作直链回退），**作为「技能地址管理」标签页嵌入管理员「技能管理」页 `pages/AdminSkillsPage.tsx`**（该页加 Tabs：「技能列表」+「技能地址管理」；入口在「设置和更多」下拉的「技能管理」`workspace-nav-menu.tsx` → `/page/workspace/admin/skills`，不再单列于 `components/page-sidebar.tsx` 左下角设置下拉）：打开自动扫描并带**进度条**（扫描逐技能 / 应用逐文件 X/N，应对 150+ 技能），URL/IP 类型过滤 + 搜索 + 「片段批量套用」（如所有 `uat.4.cn` → `prod.4.cn`）、每地址「替换为」输入框（可展开看 `file:line` 片段）、内联**预览**、**应用**、**多关键词检索**（空格分隔全部命中），以及**修改记录**面板——每次应用生成一条记录（显示时间/改了哪些地址 from→to/文件数/处数），可**针对单次记录回退**（二次确认）。client + NDJSON 流式解析在 `strategy-components/api/skill-addresses.ts`。后端 `/api/skill-addresses`（**仅管理员**）：所有文件 IO 走 `asyncio.to_thread` 不阻塞事件循环；`GET /scan`(+`/scan/stream`)、`POST /preview`、`POST /apply`(+`/apply/stream`)、`GET /backups`、`POST /rollback/{id}`；**子串安全替换**（基于 token span，不会误伤子串）、`expected_total` 乐观锁（409）、写前自动磁盘备份到 `{skills_root}/.address-edits/`（同时写一份 `{id}.manifest.json` 记录本次替换明细，供修改记录展示）、应用后刷新技能系统提示缓存。纯引擎 `deerflow/skills/address_scan.py`，router `app/gateway/routers/skill_addresses.py`，tests `tests/test_skill_addresses.py`。范围仅运行中技能，不动 `deploy/minimal/skills/`。
 - **按钮管理 (Task Button Management, admin-only)**: configures the **可配置跳转按钮** rendered in the left task panel of the task deep-link workspaces (rwfx / 3qfx / xdfx, see `TaskFxWorkspace`). Each button = `{business, label, linkType, target, appendTaskId, idParam, idKind, openMode, loginParams, appendAuth, authTokenParam, authNameParam, enabled, sortOrder}`; `linkType` is `url` (外链 `window.open`, optional append of `?{idParam}={id}` — **`idParam` is the editable query-param name**, e.g. `id`/`task_id`), `business` (内部切换到另一业务, reuses `resolveTaskDeeplinkTarget`, always task id), or `purpose` (弹出当前任务的「任务目的」表格 via the shared `TaskPurposeDialog`, no target — configurable for any page incl. xdfx). **`host`（宿主路由）** asks the embedding parent to navigate: `runTaskButton` sends `window.parent.postMessage({ type:'HOST_NAVIGATE', path, queryParams:{ [idParam]: Number(taskId) } }, hostOrigin)` — `target` is the host route (e.g. `/task/item-aiceh/step-analysis-audience`), `appendTaskId`/`idParam` become numeric `queryParams` (task id only), origin from `runtime-config.js` `VITE_HOST_NAVIGATE_ORIGIN` (else ancestorOrigins/referrer; never `*`). Independent-tab clicks toast and do nothing. **`url` buttons support two independent, coexisting login-append mechanisms**: (1) **`loginParams` (追加登录参数, like 应用管理/Light App Center)** — a list of `{key, source, customValue}` resolved at click time: `runTaskButton` reads `userInfo` (via `readUserInfo` from `strategy-components/lib/light-app`, embed-isolated through `authStorage`) and appends `?{key}={value}` for each (`source` = a userInfo field such as `accessToken`/`yUserId`, or `other` → literal `customValue`); empty list = nothing appended. Reuses `LOGIN_SOURCE_OPTIONS` from light-app; stored as JSON column `task_buttons.login_params_json` (`PortableJSON`, migration `20260625_01`). (2) **`appendAuth` (追加登录态)** — appends `?{authTokenParam}={token1}&{authNameParam}={username}` (default param names `token`/`username`) so the跳转目标外部系统 can identify the current user — e.g. `…/action/sentiment?token=<JWT>&username=lizhaoxing`. **Gated on token login**: `token1` = the `?authToken=` used by token-exchange login; `username` = the raw login name the backend resolved from `getTokenInfo`. Both are stashed at login by `persistLoginCredentials` into `login.token1`/`login.username` (`core/auth/login-credentials.ts`, scoped via [[scoped-storage]] so嵌入/独立会话隔离 + 随独立会话迁移). Only token login writes them → a non-token-login session has no `token1`, so `runTaskButton` silently skips the auth append (the jump still works, just without those two params). The frontend reads `username` straight from the new `UsernameLoginResponse.username` field (the stored email local-part is lossy — lowercased/normalized — so not used), **not** by decoding the JWT. Stored as columns `task_buttons.{append_auth, auth_token_param, auth_name_param}` (migration `20260625_02`). The admin page renders both editors for `url` buttons (loginParams 虚线框 + 追加登录态 switch). **`idKind` (`task`/`action`) only matters for xdfx**: its deep-link `?taskId=` is an 行动id and the 任务id is resolved at login (`fetchTaskIdFromAction`) — `TaskContext` now also stashes the raw `actionId`, so a url button can append either (default `task`). Page `TaskButtonsPage.tsx` (route `/page/strategy/admin/task-buttons`, **admin-gated**, entry「按钮管理」in the bottom-left settings dropdown of `components/page-sidebar.tsx`, both full + collapsed modes) has per-business tabs, edits name/type/target/启停/排序/增删, saves the **whole set atomically** (replace-all), plus 「导入默认」 to merge code-seed defaults. **Seed/fallback** (like 敏感词管理): only when the whole table is empty (fresh deploy, never saved) do the workspace + page fall back to `defaultTaskButtons(cfg)` — rwfx's url buttons (编辑/上一步/下一步/3q详情) built from the **live** `/api/public/config/task-deeplink` plus cross-business switch buttons for **all three** businesses (each switches to the other two — rwfx→三情/行动, 3qfx→任务/行动, xdfx→任务/三情). Once saved, the store is the sole source of truth. **Bottom 快捷跳转 bar groups by prev/next**: `TaskJumpBar` renders only「上一步/下一步」-type buttons in the bottom bar — a `prev` group (id 含 `prev` 或文案「上一步」, 描边次要样式 + ←图标) rendered **左侧**, then a `next` group (id 含 `next` 或文案「下一步」, 实心主色) on the **右侧** (container `justify-end`); everything else goes to the right-edge floating menu. The default 上一步 button (`rwfx-prev`, `url`, target = `DASHBOARD_AGENT_ROUTE` `/page/strategy/dashboard-agent`, `appendTaskId` + `idParam=taskId` → `?taskId=任务id`, `openMode=self`) 默认指向**应用内**「大屏绘制」Line truncated
 - **实时并发监控 (Concurrency / System-Pressure Monitor, admin-only)**: a live panel mounted at the **top of the admin 排行榜 page** (`components/workspace/admin/concurrency-monitor.tsx` `ConcurrencyMonitorPanel`, rendered in `pages/AdminUserLeaderboardPage.tsx`) showing **当前并发的大模型调用数 + 列表（用户/智能体/类型/已运行时长）**, each row with a 「停止」button (and a 「全部停止」) that **truly aborts** that run's in-flight model stream via `RunManager.cancel` (`POST /api/admin/active-runs/{run_id}/cancel`) — to relieve pressure when 使用人数过多. Two line charts (minute/hour toggle): **并发量折线图** (peak/avg, `GET /api/admin/active-runs/concurrency`, fed by a lightweight background sampler into `concurrency_samples`) and **大模型出字速度折线图** (tokens/sec, `GET /api/admin/active-runs/token-speed`, aggregated from `llm_call_metrics`) so a dip pinpoints where the bottleneck is. A **「后台采样」Switch** (`system_settings.json → concurrency_monitor.enabled`, **default off / opt-in**, `GET|PUT /api/admin/active-runs/settings`) turns the background sampling on/off **at runtime, no restart** (the live concurrency count + 「停止」 work even while sampling is off — only the 并发量 history chart needs it on); everything is designed to **never impact production** (one tiny INSERT/15s, retention-purged, best-effort). Client/hooks: `core/admin/active-runs.ts` + `core/admin/hooks.ts` (`useActiveRuns`/`useConcurrencySeries`/`useTokenSpeedSeries`/`useCancel*`/`useMonitorSettings`). See the backend CLAUDE.md「Admin Concurrency Monitor」.
-- **会话冲突 409 中文化**: when a user submits a new question while the previous turn is still streaming, the backend's `RunManager.create_or_reject` rejects with HTTP 409 (`services.py` now returns a **Chinese** detail「当前对话还在生成回答…」). `core/threads/hooks.ts` `onError` detects the conflict (`isActiveRunConflictError`: 409 / "already has an active run" / the Chinese text) and shows a friendly `toast.info` instead of a raw error.
+- **会话冲突 409 中文化**: when a user submits a new question while the previous turn is still streaming, the backend's `RunManager.create_or_reject` rejects with HTTP 409 (`services.py` now returns a **Chinese** detail「当前对话还在生成回答…」). `core/threads/hooks.ts` `onError` detects the conflict (`isActiveRunConflictError`: 409 / "already has an active run" / the Chinese text). A premature SSE close (SDK treats it as success and restores the send button while the backend run is still occupying the thread) is recovered by `confirmOrRejoin`: hold the stop button, `joinStream` the live run, and on 409 auto-reconnect instead of a fake AI bubble. `sse_consumer` no longer ends a resumable (`on_disconnect=continue`) stream on a false-positive `is_disconnected()`.
 - **State**: TanStack Query for server state, context providers (`SubtasksProvider`, `ArtifactsProvider`, `PromptInputProvider`) scoped to `WorkspaceRoutes`
 - **LLM streaming**: `@langchain/langgraph-sdk` streams messages from the backend via SSE
 - **UI**: Radix UI primitives + TailwindCSS v4 + shadcn-style components in `src/components/ui/`

frontend-web/src/components/workspace/artifacts/artifact-file-detail.tsx

@@ -77,6 +77,10 @@ import {
   useUpdateArtifactContent,
 } from "@/core/artifacts/hooks";
 import { resolveArtifactURL, urlOfArtifact } from "@/core/artifacts/utils";
+import {
+  isSandboxStreamPerfEnabled,
+  useSandboxFollowScroll,
+} from "@/core/artifacts/sandbox-stream-perf";
 import { isArtifactApiPath } from "@/core/artifacts/streaming-write-preview";
 import { useI18n } from "@/core/i18n/hooks";
 import { downloadMarkdownAsDocx } from "@/core/scheduled-tasks/markdown-docx";
@@ -401,6 +405,14 @@ export function ArtifactFileDetail({
     : (canEditArtifact || (canUseAiEditor && hasLocalEdits)
         ? draftContent
         : (content ?? ""));
+  const streamPerfEnabled = isSandboxStreamPerfEnabled();
+  const isLiveStreaming = thread.isLoading || documentRewriteActive;
+  const sandboxDisplayText = useMemo(() => {
+    const raw = displayContent ?? "";
+    if (disableDesensitize) return raw;
+    if (streamPerfEnabled && isLiveStreaming) return raw;
+    return desensitize(raw);
+  }, [disableDesensitize, displayContent, isLiveStreaming, streamPerfEnabled]);
   const artifactContentPending =
     isCodeFile &&
     !useWriteFileStreaming &&
@@ -924,39 +936,12 @@ export function ArtifactFileDetail({
     [],
   );
 
-  useEffect(() => {
-    if (!(thread.isLoading || documentRewriteActive) || !autoScrollEnabled) {
-      return;
-    }
-
-    const tick = () => {
-      scrollArtifactToBottom("auto");
-    };
-
-    tick();
-    const intervalId = window.setInterval(tick, 180);
-    return () => window.clearInterval(intervalId);
-  }, [
-    autoScrollEnabled,
-    documentRewriteActive,
-    scrollArtifactToBottom,
-    thread.isLoading,
-  ]);
-
-  useEffect(() => {
-    if (!(thread.isLoading || documentRewriteActive) || !autoScrollEnabled) {
-      return;
-    }
-
-    scrollArtifactToBottom("auto");
-  }, [
-    autoScrollEnabled,
-    displayContent,
-    documentRewriteActive,
-    scrollArtifactToBottom,
-    thread.isLoading,
+  useSandboxFollowScroll({
+    follow: isLiveStreaming && autoScrollEnabled,
+    scrollToBottom: scrollArtifactToBottom,
+    contentKey: sandboxDisplayText,
     viewMode,
-  ]);
+  });
 
   useEffect(() => {
     const handleResumeEvent = () => {
@@ -1741,7 +1726,7 @@ export function ArtifactFileDetail({
             language === "html" ||
             language === "mermaid") && (
               <ArtifactFilePreview
-                content={disableDesensitize ? (displayContent ?? "") : desensitize(displayContent ?? "")}
+                content={sandboxDisplayText}
                 language={language ?? "text"}
                 threadId={threadId}
                 publicShareCode={publicShareCode}
@@ -1778,7 +1763,7 @@ export function ArtifactFileDetail({
               value={
                 canEditArtifact || disableDesensitize
                   ? (displayContent ?? "")
-                  : desensitize(displayContent ?? "")
+                  : sandboxDisplayText
               }
               readonly={!canEditArtifact || documentRewriteActive}
               onChange={
@@ -1789,6 +1774,7 @@ export function ArtifactFileDetail({
                     }
                   : undefined
               }
+              isStreaming={thread.isLoading}
               streamingReadonlyFallback={!canEditArtifact}
             />
           </div>

frontend-web/src/components/workspace/code-editor.tsx

@@ -11,12 +11,11 @@ import { basicLightInit } from "@uiw/codemirror-theme-basic";
 import { monokaiInit } from "@uiw/codemirror-theme-monokai";
 import CodeMirror from "@uiw/react-codemirror";
 import { useTheme } from "next-themes";
-import { useMemo } from "react";
+import { memo, useMemo } from "react";
 
 import { Textarea } from "@/components/ui/textarea";
 import { cn } from "@/lib/utils";
 
-import { useThread } from "./messages/context";
 const customDarkTheme = monokaiInit({
   settings: {
     background: "transparent",
@@ -34,7 +33,7 @@ const customLightTheme = basicLightInit({
   },
 });
 
-export function CodeEditor({
+export const CodeEditor = memo(function CodeEditor({
   className,
   placeholder,
   value,
@@ -43,6 +42,7 @@ export function CodeEditor({
   disabled,
   autoFocus,
   settings,
+  isStreaming = false,
   streamingReadonlyFallback = true,
 }: {
   className?: string;
@@ -53,11 +53,10 @@ export function CodeEditor({
   disabled?: boolean;
   autoFocus?: boolean;
   settings?: unknown;
+  /** Parent-owned streaming flag so this editor does not subscribe to `thread`. */
+  isStreaming?: boolean;
   streamingReadonlyFallback?: boolean;
 }) {
-  const {
-    thread: { isLoading },
-  } = useThread();
   const { resolvedTheme } = useTheme();
 
   const extensions = useMemo(() => {
@@ -81,7 +80,7 @@ export function CodeEditor({
         className,
       )}
     >
-      {isLoading && streamingReadonlyFallback ? (
+      {isStreaming && streamingReadonlyFallback ? (
         <Textarea
           className={cn(
             "h-full overflow-auto font-mono [&_.cm-editor]:h-full [&_.cm-focused]:outline-none!",
@@ -116,4 +115,4 @@ export function CodeEditor({
       )}
     </div>
   );
-}
+});

frontend-web/src/core/artifacts/hooks.ts

@@ -1,5 +1,5 @@
 import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query";
-import { useMemo } from "react";
+import { useEffect, useMemo, useRef, useState } from "react";
 
 import { useThread } from "@/components/workspace/messages/context";
 
@@ -10,6 +10,10 @@ import {
   loadArtifactContentFromLatestWriteFile,
   loadArtifactContentFromToolCall,
 } from "./loader";
+import {
+  SANDBOX_STREAM_FLUSH_MS,
+  isSandboxStreamPerfEnabled,
+} from "./sandbox-stream-perf";
 import { isArtifactApiPath } from "./streaming-write-preview";
 import { updateArtifactContent } from "./api";
 
@@ -81,6 +85,48 @@ function applyPendingTextEdit({
   return null;
 }
 
+/**
+ * Keep the latest write_file text in a ref (always current for later readers)
+ * but only publish a display snapshot on a timer while the run is in flight.
+ * Display therefore does not follow `thread` object identity / every SSE chunk.
+ */
+function useThrottledWriteFileDisplay({
+  filepath,
+  liveContent,
+  enabled,
+  streaming,
+}: {
+  filepath: string;
+  liveContent: string | undefined;
+  enabled: boolean;
+  streaming: boolean;
+}) {
+  const latestRef = useRef(liveContent);
+  latestRef.current = liveContent;
+  const [displayed, setDisplayed] = useState(liveContent);
+
+  useEffect(() => {
+    if (!enabled) {
+      setDisplayed(latestRef.current);
+      return;
+    }
+    const flush = () => {
+      setDisplayed((previous) =>
+        Object.is(previous, latestRef.current) ? previous : latestRef.current,
+      );
+    };
+    if (!streaming) {
+      flush();
+      return;
+    }
+    flush();
+    const intervalId = window.setInterval(flush, SANDBOX_STREAM_FLUSH_MS);
+    return () => window.clearInterval(intervalId);
+  }, [enabled, filepath, streaming]);
+
+  return enabled && streaming ? displayed : liveContent;
+}
+
 export function useArtifactContent({
   filepath,
   threadId,
@@ -94,12 +140,17 @@ export function useArtifactContent({
     return isWriteFileStreamingUrl(filepath);
   }, [filepath]);
   const { thread, isMock, publicShareCode } = useThread();
-  const content = useMemo(() => {
-    if (useWriteFileStreaming) {
-      return loadArtifactContentFromToolCall({ url: filepath, thread });
-    }
-    return null;
-  }, [filepath, useWriteFileStreaming, thread]);
+  const streamPerfEnabled = isSandboxStreamPerfEnabled();
+  const liveWriteFileContent = useWriteFileStreaming
+    ? loadArtifactContentFromToolCall({ url: filepath, thread })
+    : undefined;
+  const displayedWriteFileContent = useThrottledWriteFileDisplay({
+    filepath,
+    liveContent: liveWriteFileContent,
+    enabled: streamPerfEnabled && useWriteFileStreaming,
+    streaming: thread.isLoading,
+  });
+  const content = useWriteFileStreaming ? displayedWriteFileContent : null;
   const toolCallFallbackContent = useMemo(() => {
     if (useWriteFileStreaming || !isArtifactApiPath(filepath)) {
       return undefined;

frontend-web/src/core/artifacts/sandbox-stream-perf.test.ts

@@ -0,0 +1,49 @@
+import assert from "node:assert/strict";
+import test from "node:test";
+
+import {
+  SANDBOX_STREAM_PERF_QUERY,
+  resolveSandboxStreamPerfEnabled,
+} from "./sandbox-stream-perf";
+
+test("sandbox stream perf defaults to on", () => {
+  assert.equal(resolveSandboxStreamPerfEnabled({}), true);
+  assert.equal(resolveSandboxStreamPerfEnabled({ storageValue: null }), true);
+});
+
+test("sandbox stream perf can be rolled back from hash or search query", () => {
+  assert.equal(
+    resolveSandboxStreamPerfEnabled({
+      hash: "#/page/workspace/chats/new?sandboxStreamPerf=0",
+    }),
+    false,
+  );
+  assert.equal(
+    resolveSandboxStreamPerfEnabled({
+      search: `?${SANDBOX_STREAM_PERF_QUERY}=off`,
+    }),
+    false,
+  );
+  assert.equal(
+    resolveSandboxStreamPerfEnabled({
+      hash: "#/page/workspace/agents/abc/chats/def?foo=1&sandboxStreamPerf=false",
+    }),
+    false,
+  );
+});
+
+test("sandbox stream perf query wins over localStorage", () => {
+  assert.equal(
+    resolveSandboxStreamPerfEnabled({
+      hash: "#/page/workspace/chats/new?sandboxStreamPerf=1",
+      storageValue: "0",
+    }),
+    true,
+  );
+});
+
+test("sandbox stream perf can be rolled back from localStorage", () => {
+  assert.equal(resolveSandboxStreamPerfEnabled({ storageValue: "0" }), false);
+  assert.equal(resolveSandboxStreamPerfEnabled({ storageValue: "off" }), false);
+  assert.equal(resolveSandboxStreamPerfEnabled({ storageValue: "1" }), true);
+});

frontend-web/src/core/artifacts/sandbox-stream-perf.ts

@@ -0,0 +1,115 @@
+import { useEffect } from "react";
+
+/**
+ * Sandbox streaming performance guard (write_file live preview).
+ *
+ * Default ON. Rollback without a rebuild:
+ *   1. Append `?sandboxStreamPerf=0` to the page URL and refresh
+ *      e.g. `/#/page/workspace/chats/new?sandboxStreamPerf=0`
+ *   2. Or in the console:
+ *      localStorage.setItem("deerflow.sandbox-stream-perf", "0"); location.reload();
+ *
+ * Restore the optimized path:
+ *   localStorage.removeItem("deerflow.sandbox-stream-perf") and drop the query, then refresh.
+ */
+export const SANDBOX_STREAM_PERF_STORAGE_KEY = "deerflow.sandbox-stream-perf";
+export const SANDBOX_STREAM_PERF_QUERY = "sandboxStreamPerf";
+/** Aligns with roundtable write_file preview (~12fps). */
+export const SANDBOX_STREAM_FLUSH_MS = 80;
+export const SANDBOX_LEGACY_SCROLL_INTERVAL_MS = 180;
+
+function parseFlag(raw: string | null | undefined): boolean | null {
+  if (raw == null) return null;
+  const value = raw.trim().toLowerCase();
+  if (value === "0" || value === "off" || value === "false" || value === "no") {
+    return false;
+  }
+  if (value === "1" || value === "on" || value === "true" || value === "yes") {
+    return true;
+  }
+  return null;
+}
+
+function queryValue(search: string, key: string): string | null {
+  if (!search) return null;
+  const normalized = search.startsWith("?") ? search.slice(1) : search;
+  return new URLSearchParams(normalized).get(key);
+}
+
+export function resolveSandboxStreamPerfEnabled(input: {
+  search?: string;
+  hash?: string;
+  storageValue?: string | null;
+}): boolean {
+  const searchFlag = parseFlag(queryValue(input.search ?? "", SANDBOX_STREAM_PERF_QUERY));
+  if (searchFlag != null) return searchFlag;
+
+  const hash = input.hash ?? "";
+  const hashQueryIndex = hash.indexOf("?");
+  if (hashQueryIndex >= 0) {
+    const hashFlag = parseFlag(
+      queryValue(hash.slice(hashQueryIndex), SANDBOX_STREAM_PERF_QUERY),
+    );
+    if (hashFlag != null) return hashFlag;
+  }
+
+  const storedFlag = parseFlag(input.storageValue);
+  if (storedFlag != null) return storedFlag;
+  return true;
+}
+
+export function isSandboxStreamPerfEnabled(): boolean {
+  if (typeof window === "undefined") return true;
+  let storageValue: string | null = null;
+  try {
+    storageValue = window.localStorage.getItem(SANDBOX_STREAM_PERF_STORAGE_KEY);
+  } catch {
+    storageValue = null;
+  }
+  return resolveSandboxStreamPerfEnabled({
+    search: window.location.search,
+    hash: window.location.hash,
+    storageValue,
+  });
+}
+
+export function useSandboxFollowScroll({
+  follow,
+  scrollToBottom,
+  contentKey,
+  viewMode,
+}: {
+  follow: boolean;
+  scrollToBottom: (behavior?: ScrollBehavior) => boolean;
+  contentKey: string;
+  viewMode: string;
+}) {
+  const perfEnabled = isSandboxStreamPerfEnabled();
+
+  useEffect(() => {
+    if (!perfEnabled || !follow) return;
+    let raf = 0;
+    raf = requestAnimationFrame(() => {
+      raf = 0;
+      scrollToBottom("auto");
+    });
+    return () => {
+      if (raf) cancelAnimationFrame(raf);
+    };
+  }, [perfEnabled, follow, contentKey, viewMode, scrollToBottom]);
+
+  useEffect(() => {
+    if (perfEnabled || !follow) return;
+    const tick = () => {
+      scrollToBottom("auto");
+    };
+    tick();
+    const intervalId = window.setInterval(tick, SANDBOX_LEGACY_SCROLL_INTERVAL_MS);
+    return () => window.clearInterval(intervalId);
+  }, [perfEnabled, follow, scrollToBottom]);
+
+  useEffect(() => {
+    if (perfEnabled || !follow) return;
+    scrollToBottom("auto");
+  }, [perfEnabled, follow, contentKey, viewMode, scrollToBottom]);
+}

frontend-web/src/core/threads/hooks.ts

@@ -178,6 +178,61 @@ function getRunMetadataStorage(): {
   };
 }
 
+/**
+ * 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。
  *
@@ -192,9 +247,10 @@ export async function stopThreadRun(
   if (!threadId || typeof window === "undefined") return;
   let runId: string | null = null;
   try {
-    runId = normalizeStoredRunId(
-      window.sessionStorage.getItem(`lg:stream:${threadId}`),
-    );
+    runId =
+      normalizeStoredRunId(
+        window.sessionStorage.getItem(`lg:stream:${threadId}`),
+      ) || readActiveRunId(threadId);
   } catch {
     runId = null;
   }
@@ -204,11 +260,7 @@ export async function stopThreadRun(
   } catch {
     /* 已结束 / 不存在 / 网络错误等忽略 */
   }
-  try {
-    window.sessionStorage.removeItem(`lg:stream:${threadId}`);
-  } catch {
-    /* noop */
-  }
+  clearActiveRunId(threadId);
 }
 
 /**
@@ -222,9 +274,10 @@ export async function stopActiveThreadBeforeClarificationReply(
   let storedRunId: string | null = null;
   if (threadId && typeof window !== "undefined") {
     try {
-      storedRunId = normalizeStoredRunId(
-        window.sessionStorage.getItem(`lg:stream:${threadId}`),
-      );
+      storedRunId =
+        normalizeStoredRunId(
+          window.sessionStorage.getItem(`lg:stream:${threadId}`),
+        ) || readActiveRunId(threadId);
     } catch {
       storedRunId = null;
     }
@@ -511,6 +564,22 @@ export function useThreadStream({
   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 markStreamActivity = useCallback(() => {
     lastStreamActivityAtRef.current = Date.now();
@@ -564,9 +633,6 @@ export function useThreadStream({
   const runMetadataStorageRef = useRef<
     ReturnType<typeof getRunMetadataStorage> | undefined
   >(undefined);
-  const joinStreamRef = useRef<(runId: string) => Promise<unknown>>(
-    async () => undefined,
-  );
 
   if (
     typeof window !== "undefined" &&
@@ -586,6 +652,11 @@ export function useThreadStream({
     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 : {};
@@ -731,21 +802,9 @@ export function useThreadStream({
       // 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)) {
-        const staleThreadId = threadIdRef.current ?? onStreamThreadId ?? null;
-        const runId = staleThreadId
-          ? runMetadataStorageRef.current?.getItem(`lg:stream:${staleThreadId}`)
-          : null;
-        if (
-          runId &&
-          streamReconnectAttemptsRef.current < STREAM_RECONNECT_MAX_ATTEMPTS
-        ) {
-          streamReconnectAttemptsRef.current += 1;
-          lastStreamActivityAtRef.current = Date.now();
-          void joinStreamRef.current(runId).catch(() => {
-            // Reconciliation will retry while the run stays non-terminal.
-          });
-          return;
-        }
+        setStreamHoldOpen(true);
+        void confirmOrRejoinRef.current("error");
+        return;
       }
       setOptimisticMessages([]);
       // A stale auto-reconnect ("Run <id> not found") after a backend restart:
@@ -754,22 +813,22 @@ export function useThreadStream({
       // silently — there is no live run to rejoin.
       if (isRunNotFoundError(message)) {
         const staleThreadId = threadIdRef.current ?? onStreamThreadId ?? null;
-        if (staleThreadId && runMetadataStorageRef.current) {
-          runMetadataStorageRef.current.removeItem(
-            `lg:stream:${staleThreadId}`,
-          );
+        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}，原对话不存在或无权访问，已为你打开新对话。`,
-        );
+      // 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 : "";
@@ -780,22 +839,43 @@ export function useThreadStream({
           : iframeMatch
             ? "/page/workspace/chats_iframe/new"
             : "/page/workspace/chats/new";
-        navigate(target, { replace: true });
+        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) {
-      markStreamActivity();
-      listeners.current.onFinish?.(state.values);
-      void queryClient.invalidateQueries({ queryKey: ["threads", "search"] });
+      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.
+      setStreamHoldOpen(true);
+      void confirmOrRejoinRef.current("finish").then((result) => {
+        if (result === "rejoined") {
+          return;
+        }
+        setStreamHoldOpen(false);
+        listeners.current.onFinish?.(state.values);
+        void queryClient.invalidateQueries({ queryKey: ["threads", "search"] });
+      });
     },
   });
 
@@ -821,7 +901,158 @@ export function useThreadStream({
   // new `thread` identity.
   const threadStreamRef = useRef(thread);
   threadStreamRef.current = thread;
-  joinStreamRef.current = (runId: string) => thread.joinStream(runId);
+
+  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";
+      }
+      streamReconciliationInFlightRef.current = true;
+      try {
+        const client = getAPIClient(isMock);
+        let runId = activeRunIdRef.current ?? readActiveRunId(tid);
+
+        const applyTerminalState = async (status: string) => {
+          clearActiveRunId(tid);
+          activeRunIdRef.current = null;
+          setStreamHoldOpen(false);
+          streamReconnectAttemptsRef.current = 0;
+          streamReconnectFailureToastShownRef.current = false;
+          const finalState = await client.threads.getState(tid);
+          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;
+            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;
+        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;
@@ -830,121 +1061,24 @@ export function useThreadStream({
     }
 
     let disposed = false;
-    let refreshTimer: number | null = null;
-
     const reconcileRun = async () => {
-      if (
-        disposed ||
-        streamReconciliationInFlightRef.current ||
-        Date.now() - lastStreamActivityAtRef.current < STREAM_RECONCILIATION_IDLE_MS
-      ) {
+      if (disposed) return;
+      const dropped = !threadStreamRef.current.isLoading;
+      const idleFor = Date.now() - lastStreamActivityAtRef.current;
+      if (!dropped && idleFor < STREAM_RECONCILIATION_IDLE_MS) {
         return;
       }
-
-      const storage = runMetadataStorageRef.current;
-      const runId = storage?.getItem(`lg:stream:${activeThreadId}`);
-      if (!runId) return;
-
-      streamReconciliationInFlightRef.current = true;
-      try {
-        const client = getAPIClient(isMock);
-        const run = await client.runs.get(activeThreadId, runId);
-        if (disposed) {
-          return;
-        }
-
-        if (!TERMINAL_RUN_STATUSES.has(run.status)) {
-          // ``useStream`` retries one failed connection on its own.  If that
-          // reader quietly stalls, however, its internal queue never gets a
-          // terminal error and the page remains frozen despite the backend
-          // continuing the run.
-          //
-          // SDK ``stop()`` is *not* a local-only abort: its onStop hook calls
-          // ``runs.cancel`` and deletes ``lg:stream:{threadId}``, which is
-          // exactly "答一半就不答了". Hide the run id first so stop() only
-          // aborts the browser reader, then join the still-running server run.
-          if (Date.now() - lastStreamActivityAtRef.current < STREAM_RECONNECT_IDLE_MS) {
-            return;
-          }
-
-          if (streamReconnectAttemptsRef.current >= STREAM_RECONNECT_MAX_ATTEMPTS) {
-            if (!streamReconnectFailureToastShownRef.current) {
-              streamReconnectFailureToastShownRef.current = true;
-              toast.error("回答仍在后台处理中，但页面连接暂未恢复；请稍后返回此会话查看结果。");
-            }
-            return;
-          }
-
-          streamReconnectAttemptsRef.current += 1;
-          // Avoid enqueueing another reconnect before this new reader has had
-          // enough time to receive its first event.  A genuine event resets
-          // the attempt counter through ``markStreamActivity``.
-          lastStreamActivityAtRef.current = Date.now();
-
-          const stream = threadStreamRef.current;
-          storage?.removeItem(`lg:stream:${activeThreadId}`);
-          void Promise.resolve(stream.stop())
-            .catch(() => {
-              // Stopping a stale local reader is best-effort; joining is still
-              // useful if the SDK already considers the stream stopped.
-            })
-            .finally(() => {
-              if (disposed || threadIdRef.current !== activeThreadId) return;
-              try {
-                storage?.setItem(`lg:stream:${activeThreadId}`, runId);
-              } catch {
-                /* noop */
-              }
-              void stream.joinStream(runId).catch(() => {
-                // The status poll will either reconcile a terminal run or
-                // schedule the next bounded reconnect attempt.
-              });
-            });
-          return;
-        }
-
-        // The server is already terminal, so the existing SSE listener cannot
-        // receive any more useful events.  Load the checkpoint directly before
-        // resetting the SDK stream; this preserves the normal onFinish side
-        // effects (e.g. marking a position delivery complete) even when the
-        // browser missed the terminal event or already reported a transport
-        // error and dropped ``isLoading``.
-        const finalState = await client.threads.getState(activeThreadId);
-        if (disposed) {
-          return;
-        }
-
-        storage?.removeItem(`lg:stream:${activeThreadId}`);
-        streamReconnectAttemptsRef.current = 0;
-        streamReconnectFailureToastShownRef.current = false;
-        if (run.status === "success") {
-          setOptimisticMessages([]);
-          setThreadValueOverrides(finalState.values as Partial<AgentThreadState>);
-          listeners.current.onFinish?.(finalState.values as AgentThreadState);
-          void queryClient.invalidateQueries({ queryKey: ["threads", "search"] });
-        } else {
-          setThreadValueOverrides(finalState.values as Partial<AgentThreadState>);
-          toast.error("本轮生成未能完成，已同步当前对话内容，可重新发起请求。");
-        }
-
-        // Changing the controlled thread id clears only the local SDK stream
-        // (it does not cancel the already-terminal backend run). Switching it
-        // back immediately makes useStream fetch the persisted messages, so
-        // the page exits the false loading state without a manual refresh.
-        setOnStreamThreadId(null);
-        refreshTimer = window.setTimeout(() => {
-          if (!disposed && threadIdRef.current === activeThreadId) {
-            setOnStreamThreadId(activeThreadId);
-          }
-        }, 0);
-      } catch {
-        // A transient status lookup failure must not disrupt a real stream;
-        // the next polling interval will try again if it is still quiet.
-      } finally {
-        if (!disposed) {
-          streamReconciliationInFlightRef.current = false;
-        }
+      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(
@@ -955,11 +1089,8 @@ export function useThreadStream({
     return () => {
       disposed = true;
       window.clearInterval(interval);
-      if (refreshTimer !== null) {
-        window.clearTimeout(refreshTimer);
-      }
     };
-  }, [isMock, onStreamThreadId, queryClient, thread.isLoading]);
+  }, [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
@@ -990,6 +1121,8 @@ export function useThreadStream({
     startedRef.current = false;
     sendInFlightRef.current = false;
     prevMsgCountRef.current = 0;
+    activeRunIdRef.current = null;
+    setStreamHoldOpen(false);
     setOptimisticMessages([]);
     setIsUploading(false);
     setThreadValueOverrides({});
@@ -1355,6 +1488,18 @@ export function useThreadStream({
           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] : [];
@@ -1368,7 +1513,6 @@ export function useThreadStream({
               } as Message,
             ];
           });
-          setIsUploading(false);
           return;
         }
         setOptimisticMessages([]);
@@ -1378,9 +1522,33 @@ export function useThreadStream({
         sendInFlightRef.current = false;
       }
     },
-    [thread, _handleOnStart, t.uploads.uploadingFiles, context, localSettings.display.ragModeEnabled, queryClient, runOptions],
+    [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
   const mergedValues =
     Object.keys(threadValueOverrides).length > 0
@@ -1395,14 +1563,16 @@ export function useThreadStream({
       ? (mergedValues.messages as Message[])
       : thread.messages;
 
-  const mergedThread =
-    optimisticMessages.length > 0 || mergedValues !== thread.values
-      ? ({
-          ...thread,
-          values: mergedValues,
-          messages: [...mergedMessages, ...optimisticMessages],
-        } as typeof thread)
-      : thread;
+  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）。
   // 调用方据此判断「流是否已切到全新线程」，避免在切换会话的那一帧把消息发进上一个会话

offline-backend-20260512/backend/app/gateway/services.py

@@ -1243,12 +1243,19 @@ async def sse_consumer(
 
     The ``finally`` block implements ``on_disconnect`` semantics:
     - ``cancel``: abort the background task on client disconnect.
-    - ``continue``: let the task run; events are discarded.
+    - ``continue``: let the task run so the client can ``joinStream``.
+
+    ``is_disconnected()`` is only consulted in cancel mode. In continue
+    (resumable) mode a false-positive disconnect would *cleanly* end the
+    HTTP stream; the LangGraph SDK treats that as success, clears
+    ``lg:stream:{threadId}``, and flips ``isLoading`` off — while the
+    backend run keeps the thread occupied, so the next send 409s.
+    A truly gone client fails on the next ``yield`` instead.
     """
     last_event_id = request.headers.get("Last-Event-ID")
     try:
         async for entry in bridge.subscribe(record.run_id, last_event_id=last_event_id):
-            if await request.is_disconnected():
+            if record.on_disconnect == DisconnectMode.cancel and await request.is_disconnected():
                 break
 
             if entry is HEARTBEAT_SENTINEL:
@@ -1261,6 +1268,9 @@ async def sse_consumer(
 
             yield format_sse(entry.event, entry.data, event_id=entry.id or None)
 
+    except (ConnectionError, BrokenPipeError, ConnectionResetError, ConnectionAbortedError):
+        logger.info("SSE client connection lost run_id=%s thread_id=%s", record.run_id, record.thread_id)
+
     finally:
         if record.status in (RunStatus.pending, RunStatus.running):
             if record.on_disconnect == DisconnectMode.cancel: