增强 Deep Research 可靠性，并接入 8BF 任务卡片与会商上下文。

修复 MySQL 读写分离下的持久化与 SSE 重连问题，报告完成前先落库；8BF 嵌入页支持 taskId 任务卡与独立意图 metadata；前端统一 createClientId 并更新 runtime 配置。

Co-authored-by: Cursor <cursoragent@cursor.com>

.gitignore

@@ -93,6 +93,12 @@ offline-backend-20260512/backend/tests/*
 # Keep the model configuration regression coverage versioned even though
 # this deployment repository otherwise ignores newly added backend tests.
 !offline-backend-20260512/backend/tests/test_model_management.py
+# Deep-research reliability regression coverage.
+!offline-backend-20260512/backend/tests/test_deep_research_persistence.py
+!offline-backend-20260512/backend/tests/test_deep_research_event_sink.py
+!offline-backend-20260512/backend/tests/test_deep_research_executor_reliability.py
+!offline-backend-20260512/backend/tests/test_deep_research_start_job.py
+!offline-backend-20260512/backend/tests/test_summarization_human_message.py
 deploy
 .history
 

frontend-web/public/runtime-config.js

@@ -32,10 +32,10 @@ window.__CM_RUNTIME_CONFIG__ = {
   VITE_BRAND_WELCOME_DARK_URL: "",
 
   // DeerFlow Gateway 地址：前端自有后端接口，例如 /api/agents、/api/auth、/api/threads 等。
-  // VITE_BACKEND_BASE_URL: "http://10.19.200.5:2026",
+  // VITE_BACKEND_BASE_URL: "https://ch1.b.uat.4.cn/deerflow",
 
   // LangGraph 流式对话接口地址：普通问答、智能体对话等流式能力使用。
-  // VITE_LANGGRAPH_BASE_URL: "http://10.19.200.5:2026/api",
+  // VITE_LANGGRAPH_BASE_URL: "https://ch1.b.uat.4.cn/deerflow/api",
 
   // Canvas/AI 写作独立 LangGraph 地址：为空时前端会回退使用 VITE_LANGGRAPH_BASE_URL。
   VITE_CANVAS_LANGGRAPH_BASE_URL: "",
@@ -50,6 +50,10 @@ window.__CM_RUNTIME_CONFIG__ = {
   // CMZS 后端接口：任务管理、任务分析、三情分析、行动分析等业务接口。
   VITE_API_BASE_URL: "https://ch1.b.uat.4.cn/zncm",
 
+  // 8BF 嵌入页专用任务系统接口。部署到其他系统时只需修改这一项，
+  // 不影响原多智能体会商使用的任务详情接口。
+  VITE_8BF_TASK_API_BASE_URL: "https://ch1.b.uat.4.cn/zncm",
+
   // 佳来模型对话接口。
   VITE_AIPI_CHAT_URL: "https://ch1.b.uat.4.cn/gpt",
 
@@ -90,7 +94,7 @@ window.__CM_RUNTIME_CONFIG__ = {
   VITE_AIPI_FRONTEND_URL: "http://47.88.25.99:3000",
 
   // Consumer 接口：深链任务详情 cop-task-detail、研判情报报告 selectSqReport 等使用。
-  VITE_AIPI_CONSUMER_URL: "http://10.19.200.5:8002/consumer",
+  VITE_AIPI_CONSUMER_URL: "https://ch1.b.uat.4.cn/consumer",
 
   // A 系统跳转地址：人物、组织、事件等列表项打开链接使用。
   VITE_AIPI_PRESENT_URL: "http://present.a.uat.4.cn/present/domain",

frontend-web/src/components/workspace/settings/appearance-settings-page.tsx

@@ -54,6 +54,7 @@ import type {
   CompactShortcutRouteParam,
 } from "@/core/system-settings/api";
 import { cn } from "@/lib/utils";
+import { createClientId } from "@/lib/client-id";
 import {
   APP_THEME_COLORS,
   LOGIN_SOURCE_OPTIONS,
@@ -77,10 +78,7 @@ const languageOptions: { value: Locale; label: string }[] = [
 const DEFAULT_SHORTCUT_THEME_MAP = "red=red;green=green;blue=blue;dark-blue=dark-blue";
 
 function makeShortcutId() {
-  if (typeof crypto !== "undefined" && "randomUUID" in crypto) {
-    return crypto.randomUUID();
-  }
-  return `shortcut-${Date.now()}-${Math.random().toString(16).slice(2)}`;
+  return createClientId("shortcut");
 }
 
 function newShortcutParam(): CompactShortcutRouteParam {

frontend-web/src/components/workspace/settings/custom-theme-library-settings.tsx

@@ -43,6 +43,7 @@ import {
   customThemeTextContrast,
   type CustomThemeAdvancedKey,
 } from "@/core/theme/custom-theme";
+import { createClientId } from "@/lib/client-id";
 import { cn } from "@/lib/utils";
 
 const MAX_CUSTOM_THEMES = 12;
@@ -127,10 +128,7 @@ const ADVANCED_COLOR_FIELDS: Array<{
 ];
 
 function makeThemeId(): `custom-theme-${string}` {
-  const suffix =
-    typeof crypto !== "undefined" && "randomUUID" in crypto
-      ? crypto.randomUUID().replace(/[^a-z0-9-]/gi, "").toLowerCase()
-      : `${Date.now().toString(36)}${Math.random().toString(36).slice(2, 10)}`;
+  const suffix = createClientId().replace(/[^a-z0-9-]/gi, "").toLowerCase();
   return `custom-theme-${suffix}`;
 }
 

frontend-web/src/core/document-rewrite/context.tsx

@@ -6,6 +6,7 @@ import type {
   DocumentRewriteStage,
 } from "@/open-canvas/hooks/useDocumentRewrite";
 import type { DocumentRewriteSummary } from "@/open-canvas/api/rewrite";
+import { createClientId } from "@/lib/client-id";
 
 export type DocumentRewritePresentation = {
   id: string;
@@ -74,10 +75,7 @@ const DocumentRewritePresentationContext =
   createContext<DocumentRewritePresentationContextValue | null>(null);
 
 function createId() {
-  if (typeof crypto !== "undefined" && typeof crypto.randomUUID === "function") {
-    return "drw-ui-" + crypto.randomUUID();
-  }
-  return "drw-ui-" + Date.now().toString(36) + "-" + Math.random().toString(36).slice(2);
+  return createClientId("drw-ui");
 }
 
 export function DocumentRewritePresentationProvider({

frontend-web/src/core/sentiment-agent/use-sentiment-agent-chat.ts

@@ -2,6 +2,7 @@ import type { Message } from "@langchain/langgraph-sdk";
 import { useCallback, useEffect, useRef, useState } from "react";
 
 import { runSentimentAgent, toAgUiMessages } from "./api";
+import { createClientId } from "@/lib/client-id";
 
 const STORAGE_PREFIX = "workspace.sentiment-agent.";
 
@@ -53,9 +54,9 @@ export function useSentimentAgentChat(threadId: string) {
       const question = query.trim();
       if (!question) return;
 
-      const assistantId = `sentiment-ai-${crypto.randomUUID()}`;
+      const assistantId = createClientId("sentiment-ai");
       const userMessage: Message = {
-        id: `sentiment-human-${crypto.randomUUID()}`,
+        id: createClientId("sentiment-human"),
         type: "human",
         content: [{ type: "text", text: question }],
         additional_kwargs: {
@@ -115,7 +116,7 @@ export function useSentimentAgentChat(threadId: string) {
       try {
         await runSentimentAgent({
           threadId,
-          runId: `run-${crypto.randomUUID()}`,
+          runId: createClientId("run"),
           messages: history,
           signal: controller.signal,
           handlers: {

frontend-web/src/core/threads/use-multi-round-rag.ts

@@ -2,6 +2,7 @@ import type { Message } from "@langchain/langgraph-sdk";
 import { useCallback, useEffect, useRef, useState } from "react";
 
 import { readRuntimeEnv } from "@/env";
+import { createClientId } from "@/lib/client-id";
 
 const STORAGE_PREFIX = "workspace.multi-round-rag.";
 
@@ -77,7 +78,7 @@ function writeStoredMessages(threadId: string, messages: Message[]) {
 }
 
 function createId(prefix: string) {
-  return `${prefix}-${crypto.randomUUID()}`;
+  return createClientId(prefix);
 }
 
 function isAbortError(error: unknown) {

frontend-web/src/core/threads/use-single-round-rag.ts

@@ -2,6 +2,7 @@ import type { Message } from "@langchain/langgraph-sdk";
 import { useCallback, useEffect, useRef, useState } from "react";
 
 import { readRuntimeEnv } from "@/env";
+import { createClientId } from "@/lib/client-id";
 
 const STORAGE_PREFIX = "workspace.single-round-rag.";
 
@@ -277,7 +278,7 @@ function mergeSources(current: RagSource[], incoming: RagSource[]): RagSource[]
 }
 
 function createId(prefix: string) {
-  return `${prefix}-${crypto.randomUUID()}`;
+  return createClientId(prefix);
 }
 
 function isAbortError(error: unknown) {

frontend-web/src/eightbf-roundtable/api/task-context.ts

@@ -0,0 +1,289 @@
+import { getBackendBaseURL } from "@/core/config";
+import { readRuntimeEnv } from "@/env";
+
+export interface EightBfTask {
+  id: string;
+  externalTaskId: string;
+  title: string;
+  description: string;
+  direction: string;
+  overview: string;
+  priority: string;
+  taskPriority: string;
+  targetDirection: string;
+  taskStatus: string;
+  source: string;
+  raw: Record<string, unknown> | null;
+}
+
+export interface EightBfDictionaryOption {
+  label: string;
+  value: string;
+  alias?: string;
+  statusTone?: string;
+}
+
+export interface EightBfTaskDictionaries {
+  priority: EightBfDictionaryOption[];
+  direction: EightBfDictionaryOption[];
+  status: EightBfDictionaryOption[];
+}
+
+export interface EightBfTaskContextPayload {
+  kind: "eightbf_roundtable_task";
+  usage: string;
+  task_id: string;
+  task: {
+    id: string;
+    title: string;
+    description: string;
+    direction: string;
+    overview: string;
+    priority: string;
+    task_priority: string;
+    target_direction: string;
+    task_status: string;
+  };
+  task_tags: {
+    priority: string;
+    direction: string;
+    status: string;
+  };
+}
+
+const TASK_DETAIL_PATH = "/task/{taskId}";
+const DICTIONARY_TYPES = {
+  priority: "task_priority",
+  direction: "task_direction",
+  status: "task_status",
+} as const;
+
+function externalApiBase(): string {
+  // The position-roundtable Vue app calls this task system through
+  // VITE_API_BASE_URL (usually the /zncm gateway), not DeerFlow's own API.
+  return (
+    readRuntimeEnv("VITE_8BF_TASK_API_BASE_URL")?.replace(/\/+$/, "") ||
+    readRuntimeEnv("VITE_API_BASE_URL")?.replace(/\/+$/, "") ||
+    getBackendBaseURL().replace(/\/+$/, "")
+  );
+}
+
+function externalUrl(path: string): string {
+  if (/^https?:\/\//i.test(path)) return path;
+  return `${externalApiBase()}${path.startsWith("/") ? path : `/${path}`}`;
+}
+
+/**
+ * 当前 DeerFlow 的外部业务请求适配器沿用任务系统的默认约定：使用
+ * `Authorization: Bearer admin`，而不是 DeerFlow 登录接口返回的 token。
+ * 两套 token 的用途不同，不能直接复用 `apiFetch` 的默认鉴权头；否则
+ * 任务详情/字典接口可能返回 401，页面只会看到空任务卡。
+ *
+ * 这里仍保留 include credentials，与原 Vue 请求保持一致；同时不走
+ * `apiFetch` 的 401 自动清理逻辑，避免外部任务系统的鉴权失败误登出
+ * DeerFlow 当前用户。
+ */
+async function fetchExternalTaskSystem(
+  path: string,
+  init: RequestInit = {},
+): Promise<Response> {
+  const headers = new Headers(init.headers);
+  if (!headers.has("Authorization")) {
+    headers.set("Authorization", "Bearer admin");
+  }
+  if (!headers.has("Accept")) {
+    headers.set("Accept", "application/json");
+  }
+  return fetch(externalUrl(path), {
+    ...init,
+    credentials: init.credentials ?? "include",
+    headers,
+  });
+}
+
+function firstText(...values: unknown[]): string {
+  for (const value of values) {
+    if (value === undefined || value === null) continue;
+    const text = String(value).trim();
+    if (text) return text;
+  }
+  return "";
+}
+
+function unwrapData(payload: unknown): Record<string, unknown> {
+  let current: unknown = payload;
+  // Task services in different deployments wrap the object as either
+  // { data: task } or { data: { data: task } }.
+  for (let index = 0; index < 3; index += 1) {
+    if (!current || typeof current !== "object" || Array.isArray(current)) break;
+    const next = (current as { data?: unknown }).data;
+    if (!next || typeof next !== "object" || Array.isArray(next)) break;
+    current = next;
+  }
+  return current && typeof current === "object" && !Array.isArray(current)
+    ? (current as Record<string, unknown>)
+    : {};
+}
+
+export function normalizeEightBfTask(
+  raw: Record<string, unknown> | null | undefined,
+  fallbackTaskId = "",
+): EightBfTask {
+  const source = raw ?? {};
+  const id = firstText(
+    source.id,
+    source.taskId,
+    source.externalTaskId,
+    source.external_task_id,
+    fallbackTaskId,
+  );
+  return {
+    id,
+    externalTaskId: firstText(source.externalTaskId, source.external_task_id, id),
+    title: firstText(source.title, source.taskName, source.task_name, source.name),
+    description: firstText(
+      source.description,
+      source.taskDescription,
+      source.task_description,
+      source.taskContent,
+      source.task_content,
+      source.content,
+    ),
+    direction: firstText(
+      source.direction,
+      source.taskDirection,
+      source.task_direction,
+    ),
+    overview: firstText(source.overview, source.summary),
+    priority: firstText(source.priority, source.urgency),
+    taskPriority: firstText(
+      source.taskPriority,
+      source.task_priority,
+      source.priority,
+      source.urgency,
+    ),
+    targetDirection: firstText(
+      source.targetDirection,
+      source.target_direction,
+      source.direction,
+      source.taskDirection,
+      source.task_direction,
+    ),
+    taskStatus: firstText(source.taskStatus, source.task_status, source.status),
+    source: "测试分析岗任务详情接口",
+    raw: source,
+  };
+}
+
+function dictionaryRows(payload: unknown): unknown[] {
+  if (Array.isArray(payload)) return payload;
+  if (payload && typeof payload === "object") {
+    const directData = (payload as { data?: unknown }).data;
+    if (Array.isArray(directData)) return directData;
+    if (directData && typeof directData === "object") {
+      const nested = directData as { rows?: unknown; list?: unknown };
+      if (Array.isArray(nested.rows)) return nested.rows;
+      if (Array.isArray(nested.list)) return nested.list;
+    }
+  }
+  const data = unwrapData(payload);
+  if (Array.isArray(data.rows)) return data.rows;
+  if (Array.isArray(data.list)) return data.list;
+  return [];
+}
+
+function normalizeDictionary(payload: unknown): EightBfDictionaryOption[] {
+  return dictionaryRows(payload)
+    .map((item) => {
+      if (!item || typeof item !== "object") return null;
+      const row = item as Record<string, unknown>;
+      const label = firstText(row.dictLabel, row.label);
+      const value = firstText(row.dictValue, row.dict_value, row.value);
+      if (!label || !value) return null;
+      return {
+        label,
+        value,
+        alias: firstText(row.alias) || undefined,
+        statusTone: firstText(row.statusTone, row.status_tone) || undefined,
+      } as EightBfDictionaryOption;
+    })
+    .filter((item): item is EightBfDictionaryOption => item !== null);
+}
+
+export async function fetchEightBfTask(taskId: string): Promise<EightBfTask> {
+  const encodedId = encodeURIComponent(taskId);
+  const path = TASK_DETAIL_PATH.replaceAll("{taskId}", encodedId);
+  const response = await fetchExternalTaskSystem(path, { method: "GET" });
+  if (!response.ok) {
+    throw new Error(`任务详情接口请求失败（${response.status}）`);
+  }
+  const payload = (await response.json()) as unknown;
+  const task = normalizeEightBfTask(unwrapData(payload), taskId);
+  // Keep the external route id as the stable scope even if a deployment's
+  // response includes a legacy/internal numeric id.
+  return { ...task, id: taskId, externalTaskId: taskId };
+}
+
+export async function fetchEightBfTaskDictionaries(): Promise<EightBfTaskDictionaries> {
+  const entries = await Promise.all(
+    (Object.entries(DICTIONARY_TYPES) as Array<[
+      keyof EightBfTaskDictionaries,
+      string,
+    ]>).map(async ([key, path]) => {
+      try {
+        const response = await fetchExternalTaskSystem(
+          `/system/dict/data/type/${path}`,
+          { method: "GET" },
+        );
+        if (!response.ok) throw new Error(`字典接口请求失败（${response.status}）`);
+        return [key, normalizeDictionary(await response.json())] as const;
+      } catch (error) {
+        // A dictionary failure must not hide the task card; the card falls back
+        // to the raw task codes and the detail request remains usable.
+        console.warn(`[8bf] ${key} 字典加载失败`, error);
+        return [key, []] as const;
+      }
+    }),
+  );
+  return Object.fromEntries(entries) as EightBfTaskDictionaries;
+}
+
+export function dictionaryText(
+  options: EightBfDictionaryOption[] | undefined,
+  value: unknown,
+  preferredField: "label" | "alias" = "label",
+): string {
+  const raw = firstText(value);
+  if (!raw) return "";
+  const match = options?.find((option) => option.value === raw);
+  return firstText(match?.[preferredField], match?.label, raw);
+}
+
+export function buildEightBfTaskContext(
+  task: EightBfTask | null,
+  dictionaries: EightBfTaskDictionaries,
+): EightBfTaskContextPayload | null {
+  if (!task?.id) return null;
+  return {
+    kind: "eightbf_roundtable_task",
+    usage:
+      "这是 8BF 页面左上角任务卡片中的任务信息。请直接基于这些信息理解用户问题，不要要求用户重复提供已经存在的任务名称、任务描述、任务方向或任务状态。",
+    task_id: task.id,
+    task: {
+      id: task.id,
+      title: task.title,
+      description: task.description,
+      direction: task.direction,
+      overview: task.overview,
+      priority: task.priority,
+      task_priority: task.taskPriority,
+      target_direction: task.targetDirection,
+      task_status: task.taskStatus,
+    },
+    task_tags: {
+      priority: dictionaryText(dictionaries.priority, task.taskPriority),
+      direction: dictionaryText(dictionaries.direction, task.targetDirection, "alias"),
+      status: dictionaryText(dictionaries.status, task.taskStatus),
+    },
+  };
+}

frontend-web/src/eightbf-roundtable/components/EightBfTaskCard.tsx

@@ -0,0 +1,58 @@
+import { Clipboard, Loader2 } from "lucide-react";
+
+import { cn } from "@/lib/utils";
+import {
+  dictionaryText,
+  type EightBfDictionaryOption,
+  type EightBfTask,
+} from "@/eightbf-roundtable/api/task-context";
+
+interface EightBfTaskCardProps {
+  task: EightBfTask | null;
+  dictionaries: {
+    priority: EightBfDictionaryOption[];
+    direction: EightBfDictionaryOption[];
+    status: EightBfDictionaryOption[];
+  };
+  loading?: boolean;
+  error?: string | null;
+}
+
+export function EightBfTaskCard({
+  task,
+  dictionaries,
+  loading = false,
+  error,
+}: EightBfTaskCardProps) {
+  const priority = dictionaryText(dictionaries.priority, task?.taskPriority);
+  const direction = dictionaryText(dictionaries.direction, task?.targetDirection, "alias");
+  const status = dictionaryText(dictionaries.status, task?.taskStatus);
+  const tags = [priority, direction, status].filter(Boolean);
+
+  return (
+    <section className="eightbf-roundtable__task-card" aria-labelledby="eightbf-task-card-title">
+      <div className="eightbf-roundtable__task-card-heading">
+        <span className="eightbf-roundtable__task-card-icon" aria-hidden="true">
+          <Clipboard className="h-3.5 w-3.5" strokeWidth={2.5} />
+        </span>
+        <h2 id="eightbf-task-card-title">当前任务</h2>
+        {loading && <Loader2 className="h-3.5 w-3.5 animate-spin" aria-label="正在加载任务" />}
+      </div>
+
+      <div className="eightbf-roundtable__task-card-body">
+        <p className="eightbf-roundtable__task-card-title">
+          {task?.title || (task?.id ? `任务 ${task.id}` : "等待任务信息")}
+        </p>
+        {tags.length > 0 && (
+          <div className="eightbf-roundtable__task-card-tags" aria-label="任务标签">
+            {tags.map((tag) => <span key={tag}>{tag}</span>)}
+          </div>
+        )}
+        <p className={cn("eightbf-roundtable__task-card-description", !task?.description && "is-empty")}>
+          {task?.description || task?.overview || (loading ? "正在读取任务详情…" : "暂无任务描述")}
+        </p>
+        {error && <p className="eightbf-roundtable__task-card-error" role="status">{error}</p>}
+      </div>
+    </section>
+  );
+}

frontend-web/src/eightbf-roundtable/pages/EightBfRoundtablePage.tsx

@@ -1,7 +1,17 @@
 import { useCallback, useEffect, useMemo, useState } from "react";
+import { useSearchParams } from "react-router-dom";
 import { Loader2, Save, Settings2, X } from "lucide-react";
 
 import { getAccountDisplayName, getStoredAuth } from "@/core/auth";
+import { EightBfTaskCard } from "@/eightbf-roundtable/components/EightBfTaskCard";
+import {
+  buildEightBfTaskContext,
+  fetchEightBfTask,
+  fetchEightBfTaskDictionaries,
+  normalizeEightBfTask,
+  type EightBfTask,
+  type EightBfTaskDictionaries,
+} from "@/eightbf-roundtable/api/task-context";
 import { PositionBusinessChainFlow } from "@/position-roundtable/components/PositionBusinessChainFlow";
 import RoundtablePlanningPage from "@/roundtable-planning/pages/RoundtablePlanningPage";
 import { listChains } from "@/roundtable-planning/api/chains";
@@ -26,10 +36,28 @@ const BUSINESS_CODE = "8BF";
  */
 export default function EightBfRoundtablePage({ compactChrome = false }: { compactChrome?: boolean }) {
   const auth = getStoredAuth();
+  const [searchParams] = useSearchParams();
+  const taskId = useMemo(
+    () => searchParams.get("taskId")?.trim() || "",
+    [searchParams],
+  );
   // 8BF 公共业务链由指定账号维护，不依赖外部透传的管理员角色。
   // 与敏感词管理使用同一可信账户展示名规则（登录邮箱 @ 前缀）。
   const canConfigureChain = getAccountDisplayName(auth).toLowerCase() === "lqq";
-  const storageTaskId = useMemo(() => `8bf:${auth?.user_id || "anonymous"}`, [auth?.user_id]);
+  const storageTaskId = useMemo(
+    () => `8bf:${auth?.user_id || "anonymous"}:${taskId || "anonymous"}`,
+    [auth?.user_id, taskId],
+  );
+  const [task, setTask] = useState<EightBfTask | null>(() => (
+    taskId ? normalizeEightBfTask({}, taskId) : null
+  ));
+  const [taskDictionaries, setTaskDictionaries] = useState<EightBfTaskDictionaries>({
+    priority: [],
+    direction: [],
+    status: [],
+  });
+  const [taskLoading, setTaskLoading] = useState(Boolean(taskId));
+  const [taskError, setTaskError] = useState<string | null>(null);
   const [mapping, setMapping] = useState<BusinessMapping | null>(null);
   const [chains, setChains] = useState<BusinessChain[]>([]);
   const [isLoading, setIsLoading] = useState(true);
@@ -41,6 +69,53 @@ export default function EightBfRoundtablePage({ compactChrome = false }: { compa
     () => chains.find((chain) => chain.id === mapping?.chainId) ?? null,
     [chains, mapping?.chainId],
   );
+  const taskContext = useMemo(
+    () => buildEightBfTaskContext(task, taskDictionaries),
+    [task, taskDictionaries],
+  );
+
+  useEffect(() => {
+    if (!taskId) {
+      setTask(null);
+      setTaskError(null);
+      setTaskLoading(false);
+      setTaskDictionaries({ priority: [], direction: [], status: [] });
+      return;
+    }
+
+    let cancelled = false;
+    setTaskLoading(true);
+    setTaskError(null);
+    setTask(normalizeEightBfTask({}, taskId));
+    let pendingRequests = 2;
+    const markRequestFinished = () => {
+      pendingRequests -= 1;
+      if (!cancelled && pendingRequests === 0) setTaskLoading(false);
+    };
+    // Keep detail and dictionary requests independent: a missing dictionary
+    // must not prevent the task card or task context from being available.
+    void fetchEightBfTask(taskId)
+      .then((nextTask) => {
+        if (!cancelled) setTask(nextTask);
+      })
+      .catch((error) => {
+        if (cancelled) return;
+        setTaskError(error instanceof Error ? error.message : "任务详情加载失败");
+      })
+      .finally(markRequestFinished);
+    void fetchEightBfTaskDictionaries()
+      .then((nextDictionaries) => {
+        if (!cancelled) setTaskDictionaries(nextDictionaries);
+      })
+      .catch(() => {
+        // fetchEightBfTaskDictionaries already degrades each dictionary to an
+        // empty list; this guard protects future adapter changes.
+      })
+      .finally(markRequestFinished);
+    return () => {
+      cancelled = true;
+    };
+  }, [taskId]);
 
   const loadConfiguration = useCallback(async () => {
     setIsLoading(true);
@@ -162,6 +237,16 @@ export default function EightBfRoundtablePage({ compactChrome = false }: { compa
       lockedChainId={mapping?.chainId ?? null}
       lockBusinessChain
       storageTaskId={storageTaskId}
+      taskContext={taskContext as Record<string, unknown> | null}
+      entryRailTaskId={taskId || null}
+      entryRailTaskCardSlot={taskId ? (
+        <EightBfTaskCard
+          task={task}
+          dictionaries={taskDictionaries}
+          loading={taskLoading}
+          error={taskError}
+        />
+      ) : undefined}
       toolbarSlot={toolbarSlot}
       entryRailBusinessChainSlot={
         <PositionBusinessChainFlow

frontend-web/src/eightbf-roundtable/styles/step1.css

@@ -27,4 +27,100 @@
   min-height: 460px;
 }
 
+/* 8BF 外部 taskId：左上任务卡。仅挂在 8BF 根节点下，避免改动原会商任务卡。 */
+.eightbf-roundtable .eightbf-roundtable__task-card {
+  flex: 0 0 auto;
+  border: 1px solid #d9e5ef;
+  border-radius: 16px;
+  background: linear-gradient(145deg, #ffffff 0%, #f8fbff 100%);
+  padding: 14px;
+  box-shadow: 0 4px 20px rgb(15 23 42 / 3%);
+}
+
+.eightbf-roundtable .eightbf-roundtable__task-card-heading {
+  display: flex;
+  align-items: center;
+  gap: 8px;
+  color: #1e293b;
+  font-size: 13px;
+  font-weight: 700;
+}
+
+.eightbf-roundtable .eightbf-roundtable__task-card-heading > svg:last-child {
+  margin-left: auto;
+  color: #0f766e;
+}
+
+.eightbf-roundtable .eightbf-roundtable__task-card-icon {
+  display: inline-flex;
+  align-items: center;
+  justify-content: center;
+  width: 22px;
+  height: 22px;
+  border-radius: 7px;
+  background: #0f766e;
+  color: #fff;
+}
+
+.eightbf-roundtable .eightbf-roundtable__task-card-body {
+  min-width: 0;
+  margin-top: 11px;
+}
+
+.eightbf-roundtable .eightbf-roundtable__task-card-title {
+  margin: 0;
+  color: #0f172a;
+  font-size: 14px;
+  font-weight: 700;
+  line-height: 1.55;
+  overflow-wrap: anywhere;
+}
+
+.eightbf-roundtable .eightbf-roundtable__task-card-tags {
+  display: flex;
+  flex-wrap: wrap;
+  gap: 5px;
+  margin-top: 8px;
+}
+
+.eightbf-roundtable .eightbf-roundtable__task-card-tags span {
+  border: 1px solid #c7e6df;
+  border-radius: 999px;
+  background: #ecfdf5;
+  color: #0f766e;
+  padding: 2px 7px;
+  font-size: 10px;
+  line-height: 1.4;
+}
+
+.eightbf-roundtable .eightbf-roundtable__task-card-description {
+  display: -webkit-box;
+  margin: 9px 0 0;
+  color: #475569;
+  font-size: 11px;
+  line-height: 1.65;
+  overflow: hidden;
+  -webkit-box-orient: vertical;
+  -webkit-line-clamp: 5;
+  overflow-wrap: anywhere;
+  white-space: pre-wrap;
+}
+
+.eightbf-roundtable .eightbf-roundtable__task-card-description.is-empty { color: #94a3b8; }
+.eightbf-roundtable .eightbf-roundtable__task-card-error { margin: 8px 0 0; color: #b91c1c; font-size: 10px; line-height: 1.5; }
+
+.dark .eightbf-roundtable .eightbf-roundtable__task-card {
+  border-color: rgb(92 156 219 / 45%);
+  background: linear-gradient(145deg, #0b315d 0%, #08264b 100%);
+  box-shadow: 0 8px 24px rgb(0 0 0 / 18%);
+}
+
+.dark .eightbf-roundtable .eightbf-roundtable__task-card-heading,
+.dark .eightbf-roundtable .eightbf-roundtable__task-card-title { color: #f1f7ff; }
+.dark .eightbf-roundtable .eightbf-roundtable__task-card-heading > svg:last-child { color: #65d6c6; }
+.dark .eightbf-roundtable .eightbf-roundtable__task-card-tags span { border-color: rgb(101 214 198 / 40%); background: rgb(15 118 110 / 24%); color: #9ef0df; }
+.dark .eightbf-roundtable .eightbf-roundtable__task-card-description { color: #c3d4e8; }
+.dark .eightbf-roundtable .eightbf-roundtable__task-card-description.is-empty { color: #8fa8c5; }
+.dark .eightbf-roundtable .eightbf-roundtable__task-card-error { color: #fca5a5; }
+
 /* 可在此继续覆盖第 1 步的任务理解面板与业务链流程图样式。 */

frontend-web/src/lib/client-id.ts

@@ -0,0 +1,27 @@
+/**
+ * Generate a client-only id without assuming a secure browser context.
+ *
+ * `crypto.randomUUID()` is unavailable in several browsers when the app is
+ * opened directly through an `http://<public-ip>:<port>` address.  Request and
+ * UI ids need uniqueness, not a security token, so a timestamp/random fallback
+ * keeps those flows usable until the deployment is served over HTTPS.
+ */
+export function createClientId(prefix?: string): string {
+  const cryptoApi = typeof globalThis === "object" ? globalThis.crypto : undefined;
+  let id: string | undefined;
+
+  if (typeof cryptoApi?.randomUUID === "function") {
+    id = cryptoApi.randomUUID();
+  } else if (typeof cryptoApi?.getRandomValues === "function") {
+    try {
+      const values = new Uint32Array(4);
+      cryptoApi.getRandomValues(values);
+      id = Array.from(values, (value) => value.toString(16).padStart(8, "0")).join("");
+    } catch {
+      // Some restricted browser contexts expose crypto but reject random data.
+    }
+  }
+
+  id ??= `${Date.now().toString(36)}-${Math.random().toString(36).slice(2, 12)}`;
+  return prefix ? `${prefix}-${id}` : id;
+}

frontend-web/src/pages/PluginManagementPage.tsx

@@ -8,6 +8,7 @@ import { Button } from "@/components/ui/button";
 import { Input } from "@/components/ui/input";
 import { Switch } from "@/components/ui/switch";
 import { useAgentSelector } from "@/core/agents";
+import { createClientId } from "@/lib/client-id";
 import {
   type DomainAgentMapping,
   normalizeDomainInput,
@@ -20,10 +21,7 @@ import {
 
 function makeRow(): DomainAgentMapping {
   return {
-    id:
-      typeof crypto !== "undefined" && typeof crypto.randomUUID === "function"
-        ? crypto.randomUUID()
-        : `${Date.now()}-${Math.random().toString(36).slice(2, 10)}`,
+    id: createClientId(),
     domain: "",
     agentId: "",
     enabled: true,

frontend-web/src/position-roundtable/api/sessions.ts

@@ -6,6 +6,7 @@
  */
 import { apiFetch } from "@/core/api/fetch-client";
 import { getBackendBaseURL } from "@/core/config";
+import { createClientId } from "@/lib/client-id";
 
 /**
  * 节点状态由岗位会商服务推进；前端只据此决定可否发送、展示进度，
@@ -147,10 +148,7 @@ const BASE = () => `${getBackendBaseURL()}${PRIMARY_API_PATH}`;
  * （不会重复 +revision / 重复解锁下游 / 重复驳回）。每次新的用户操作则取新键。
  */
 function newCommandId(): string {
-  if (typeof crypto !== "undefined" && typeof crypto.randomUUID === "function") {
-    return crypto.randomUUID().replace(/-/g, "");
-  }
-  return `${Date.now().toString(36)}${Math.random().toString(36).slice(2, 12)}`;
+  return createClientId().replace(/-/g, "");
 }
 
 /**

frontend-web/src/roundtable-planning/hooks/useStep1Intent.ts

@@ -24,6 +24,7 @@ import {
   type IntentSummary,
 } from "@/roundtable-planning/api/intent";
 import { uploadFiles } from "@/core/uploads/api";
+import { createClientId } from "@/lib/client-id";
 import {
   clarificationKey,
   formatIntentClarificationAnswer,
@@ -449,9 +450,7 @@ export function useStep1Intent(options: UseStep1IntentOptions): UseStep1IntentRe
 
         // Phase 4 单飞幂等键：每个回合一个稳定 id。后端据此对 (threadId, clientTurnId)
         // 跨 worker 单飞——重复 / 双击命中既有回合即幂等返回，杜绝并发下双发意图。
-        const clientTurnId =
-          (typeof crypto !== "undefined" && crypto.randomUUID && crypto.randomUUID()) ||
-          `${Date.now().toString(36)}-${Math.random().toString(36).slice(2)}`;
+        const clientTurnId = createClientId();
 
         const res = await streamIntent({
           threadId: intentThreadId,

frontend-web/src/roundtable-planning/pages/RoundtablePlanningPage.tsx

@@ -298,6 +298,12 @@ export default function RoundtablePlanningPage({
   toolbarSlot,
   /** 独立业务页在第一步左上角替换默认业务链选择区的内容。 */
   entryRailBusinessChainSlot,
+  /** 独立业务页在第一步左上角提供自己的任务卡片；不触发原深链任务接口。 */
+  entryRailTaskCardSlot,
+  /** 独立业务页传给意图智能体的隐藏任务上下文。 */
+  taskContext,
+  /** 独立业务页的任务作用域，仅用于左栏记录筛选，不改变原深链逻辑。 */
+  entryRailTaskId,
   /** 独立业务页的根节点样式类，避免修改原会商样式。 */
   rootClassName,
 }: {
@@ -307,6 +313,9 @@ export default function RoundtablePlanningPage({
   storageTaskId?: string | null;
   toolbarSlot?: ReactNode;
   entryRailBusinessChainSlot?: ReactNode;
+  entryRailTaskCardSlot?: ReactNode;
+  taskContext?: Record<string, unknown> | null;
+  entryRailTaskId?: string | null;
   rootClassName?: string;
 }) {
   const { theme } = useTheme();
@@ -315,6 +324,10 @@ export default function RoundtablePlanningPage({
   const isDark = theme === "dark";
   const navigate = useNavigate();
   const location = useLocation();
+  const taskContextRef = useRef<Record<string, unknown> | null>(taskContext ?? null);
+  useEffect(() => {
+    taskContextRef.current = taskContext ?? null;
+  }, [taskContext]);
   /** 无界(wujie)嵌入抽屉：隐藏业务链条入口，设置按钮改作手动回传交付。 */
   const inWujieEmbed = isWujieEnv();
 
@@ -542,6 +555,38 @@ export default function RoundtablePlanningPage({
   }>({ intentReady: null, interactiveMessages: [] });
 
   const buildInitialIntent = useCallback((): string => {
+    const task = taskContextRef.current?.task as Record<string, unknown> | null | undefined;
+    const taskTags = taskContextRef.current?.task_tags as
+      | Record<string, unknown>
+      | undefined;
+    const taskValue = (value: unknown) =>
+      value === undefined || value === null ? "" : String(value).trim();
+    const taskLines = task
+      ? [
+          ["任务 ID", taskValue(task.id ?? taskContextRef.current?.task_id)],
+          ["任务名称", taskValue(task.title)],
+          ["任务描述", taskValue(task.description)],
+          ["任务概览", taskValue(task.overview)],
+          ["任务方向", taskValue(task.direction ?? task.target_direction)],
+          ["优先级", taskValue(task.task_priority ?? task.priority)],
+          ["任务状态", taskValue(task.task_status)],
+          ["优先级标签", taskValue(taskTags?.priority)],
+          ["方向标签", taskValue(taskTags?.direction)],
+          ["状态标签", taskValue(taskTags?.status)],
+        ]
+          .filter(([, value]) => value)
+          .map(([label, value]) => `${label}: ${value}`)
+      : [];
+    const taskCardContext = taskLines.length
+      ? [
+          "【8BF 任务卡片上下文】",
+          ...taskLines,
+          "请把以上任务卡片信息作为本轮研讨的既有背景，不要让用户重复提供。",
+          "",
+        ].join("\n")
+      : "";
+    const withTaskCardContext = (value: string) =>
+      taskCardContext ? `${taskCardContext}${value}` : value;
     const { intentReady, interactiveMessages } = step1StateRef.current;
     if (intentReady) {
       const parts: string[] = [intentReady.objective];
@@ -559,17 +604,17 @@ export default function RoundtablePlanningPage({
       parts.push(
         "用户已经过任务理解智能体澄清,以上信息可以直接作为研讨依据。请基于上述目标与约束,在圆桌内调度合适的子智能体完成研讨,不要再反问用户。",
       );
-      return parts.join("\n");
+      return withTaskCardContext(parts.join("\n"));
     }
     const userTurns = interactiveMessages
       .filter((m) => m.sender === "user")
       .map((m) => m.content?.trim())
       .filter((s): s is string => !!s);
     if (userTurns.length === 0) {
-      return "用户尚未在第一步明确任务意图。请先调用 ask_clarification,简要询问目标、范围与硬性约束,再进入研讨。";
+      return withTaskCardContext("用户尚未在第一步明确任务意图。请先调用 ask_clarification,简要询问目标、范围与硬性约束,再进入研讨。");
     }
     const conversation = userTurns.map((t, i) => `用户#${i + 1}: ${t}`).join("\n");
-    return `${conversation}\n\n注意:用户尚未完成结构化的任务理解流程,以上是原始输入。若信息不足以开展研讨,请先调用 ask_clarification 询问关键约束,再开始派活。`;
+    return withTaskCardContext(`${conversation}\n\n注意:用户尚未完成结构化的任务理解流程,以上是原始输入。若信息不足以开展研讨,请先调用 ask_clarification 询问关键约束,再开始派活。`);
   }, []);
 
   const step2 = useStep2Orchestration({
@@ -593,6 +638,7 @@ export default function RoundtablePlanningPage({
     currentStep,
     selectedModel,
     triggerToast,
+    taskContext,
   });
 
   // 同步 step1StateRef,让 buildInitialIntent / Persona / Step 3 都能读到最新值。
@@ -2176,12 +2222,13 @@ export default function RoundtablePlanningPage({
           onSelectChain={handleSelectChainFromSidebar}
           onOpenDraft={drafts.handleLoadDraft}
           activeDraftId={drafts.currentDraftId}
-          entryRailTaskId={recordTaskIdOf(prefillRef.current)}
+          entryRailTaskId={entryRailTaskId ?? recordTaskIdOf(prefillRef.current)}
           historyTaskId={draftStorageTaskId}
           entryRailBusinessCode={prefillRef.current?.businessCode ?? null}
           lockedChainId={lockedChainId}
           lockBusinessChain={lockBusinessChain}
           entryRailBusinessChainSlot={entryRailBusinessChainSlot}
+          entryRailTaskCardSlot={entryRailTaskCardSlot}
         />
 
         {/* MIDDLE VIEWPORT */}
@@ -2743,6 +2790,8 @@ interface LeftSidebarProps {
   lockedChainId?: string | null;
   /** 第一阶段替换默认业务链选择区的自定义内容。 */
   entryRailBusinessChainSlot?: ReactNode;
+  /** 第一阶段由独立业务页提供的自定义任务卡片。 */
+  entryRailTaskCardSlot?: ReactNode;
 }
 
 function Step2SandboxSidebarSync({
@@ -2794,6 +2843,7 @@ function LeftSidebar({
   lockBusinessChain,
   lockedChainId,
   entryRailBusinessChainSlot,
+  entryRailTaskCardSlot,
 }: LeftSidebarProps) {
   // 整片脱敏：左侧面板（任务概览/参与角色/人工参与等）的所有文本——标题、动态内容、
   // 嵌套子组件文本一并脱敏。消息列表与沙箱是另外的容器，不挂此 Hook，保持原文。
@@ -2937,6 +2987,7 @@ function LeftSidebar({
           lockBusinessChain={lockBusinessChain}
           lockedChainId={lockedChainId ?? null}
           businessChainSlot={entryRailBusinessChainSlot}
+          taskCardSlot={entryRailTaskCardSlot}
         />
       )}
     </section>
@@ -2992,6 +3043,7 @@ function RoundtableEntryRail({
   lockedChainId,
   lockBusinessChain = false,
   businessChainSlot,
+  taskCardSlot,
 }: {
   selectedChainId: string | null;
   onSelectChain: (chain: BusinessChain) => void;
@@ -3007,6 +3059,7 @@ function RoundtableEntryRail({
   lockedChainId?: string | null;
   lockBusinessChain?: boolean;
   businessChainSlot?: ReactNode;
+  taskCardSlot?: ReactNode;
 }) {
   const [chains, setChains] = useState<BusinessChain[]>([]);
   const [chainSectionExpanded, setChainSectionExpanded] = useState(true);
@@ -3093,7 +3146,12 @@ function RoundtableEntryRail({
       )}
     >
       {/* 深链接（taskId）模式：左上角展示「当前任务」（rwfx/3qfx）或「当前行动」（xdfx），替代业务链选择。 */}
-      {deeplink ? (
+      {taskCardSlot ? (
+        <>
+          {taskCardSlot}
+          {lockBusinessChain && businessChainSlot}
+        </>
+      ) : deeplink ? (
         isXdfx ? (
           <div className="bg-white border border-[#ECE7E4] rounded-2xl p-4 shadow-[0_4px_20px_rgba(0,0,0,0.02)]">
             <div className="flex items-center gap-2 mb-3">

frontend-web/src/strategy-components/api/deep-research.ts

@@ -17,6 +17,7 @@ import {
 
 import { apiFetch } from "@/core/api/fetch-client";
 import { getBackendBaseURL } from "@/core/config";
+import { createClientId } from "@/lib/client-id";
 import type {
   DocumentRewriteSummary,
   DocumentRewriteStreamEvent,
@@ -355,7 +356,7 @@ export async function startJob(
     method: "POST",
     headers: { "Content-Type": "application/json" },
     body: JSON.stringify({
-      request_id: opts.requestId ?? crypto.randomUUID(),
+      request_id: opts.requestId ?? createClientId(),
       entry: opts.entry ?? "legacy",
       config: opts.config ?? null,
     }),

offline-backend-20260512/backend/CLAUDE.md

@@ -1147,6 +1147,10 @@ Automatic conversation summarization when approaching token limits:
   the config flag. `config.summarization.enabled` now defaults to **false**.
 - `config.yaml` `summarization` still supplies the **tuning**: trigger types
   (tokens, messages, or fraction of max input) and the keep policy.
+- A compressed request must retain the latest non-empty `HumanMessage`, even
+  when it has already been folded into the hidden summary. This keeps strict
+  OpenAI-compatible gateways from rejecting a tool-continuation call with
+  `No user query found in message`.
 - Keeps recent messages while summarizing older ones.
 
 Tests: `tests/test_summarization_toggle.py` (gating + whitelist),
@@ -1284,6 +1288,31 @@ The per-user Deep Research concurrency guard counts only session statuses
 consume a slot; otherwise saved drafts make `POST /sessions/{id}/jobs` return
 429 permanently. Regression: `test_saved_draft_sessions_do_not_consume_research_concurrency`.
 
+`POST /api/deep-research/sessions/{id}/jobs` is retry-safe under MySQL
+read/write splitting. Deep Research session/job/event/source/message writes
+must serialise their changed row while the write transaction is still open
+(`flush` for inserts, select-after-update **before** commit), never through a
+post-commit `session.refresh()` or a fresh pooled read. Replica lag otherwise
+turns a committed job into a misleading 500, can make dispatcher `claim_job()`
+lose its snapshot, or can fail report/follow-up persistence after the model has
+already completed. Always nudge the dispatcher for both a newly created job and
+an idempotently reused queued job; otherwise a retry after such an interrupted
+response returns the old job but leaves it waiting for the periodic scan. The
+SSE replay generator must catch temporary event/job-store reads, emit a
+recoverable warning, and keep the connection open so the next poll can repair
+the cursor instead of surfacing an ASGI `ExceptionGroup`. Regressions:
+`test_deep_research_writes_never_depend_on_post_commit_refresh`,
+`test_job_create_never_reads_back_after_commit`,
+`test_retry_of_existing_queued_job_nudges_dispatcher`, and
+`test_stream_stays_open_when_durable_replay_has_a_transient_failure`.
+
+At ordinary report completion, persist and confirm the session's
+`report_markdown` projection **before** marking the Job `completed`. Retry a
+short transient projection failure; if it cannot be confirmed, let the job
+become failed rather than reporting a report that vanishes on refresh.
+`test_report_is_persisted_before_job_is_marked_completed` and
+`test_report_projection_failure_never_marks_job_completed` cover the ordering.
+
 The chat-collection frontend must keep report-writing progress in the existing
 workspace `MessageList` / `MessageGroup` tool-step renderer: inject only
 virtual `deep_research_progress` tool calls for the durable phases, keep the

offline-backend-20260512/backend/README.md

@@ -151,7 +151,7 @@ FastAPI application providing REST endpoints for frontend integration:
 | `GET /api/threads/{id}/uploads/list` | List uploaded files |
 | `DELETE /api/threads/{id}` | Delete DeerFlow-managed local thread data after LangGraph thread deletion; unexpected failures are logged server-side and return a generic 500 detail |
 | `GET /api/threads/{id}/artifacts/{path}` | Serve generated artifacts |
-| `GET/POST /api/deep-research/*` | Durable Deep Research sessions and jobs. `quick`/`basic` run the compact report flow; `detailed` plans subtopics and writes independent sections; `deep` recursively investigates bounded follow-up branches. All modes reuse DeerFlow's configured material provider, LLM factory, hidden-thread artifacts, and SSE event log. Optional report illustrations use a separately configured OpenAI-compatible image endpoint and are served only through the owning session's authenticated artifact route. |
+| `GET/POST /api/deep-research/*` | Durable Deep Research sessions and jobs. `quick`/`basic` run the compact report flow; `detailed` plans subtopics and writes independent sections; `deep` recursively investigates bounded follow-up branches. All modes reuse DeerFlow's configured material provider, LLM factory, hidden-thread artifacts, and SSE event log. Deep Research session/job/event/source/message writes serialise their rows in the writer transaction, rather than doing a post-commit ORM refresh or readback; this prevents MySQL read/write-splitting replica lag from failing “start writing”, dispatcher claim, report persistence, or follow-up history. A report is written to the session and confirmed before its Job becomes `completed`; an unconfirmed write is retried and never exposed as a false completion. Retrying an already-created queued job re-nudges the dispatcher, recovering an interrupted HTTP response without creating a second report. If progress-event persistence, durable replay, or live fan-out is briefly unavailable, the worker/stream logs the degradation and continues or reconnects rather than failing the whole job. Optional report illustrations use a separately configured OpenAI-compatible image endpoint and are served only through the owning session's authenticated artifact route. |
 | `POST /api/ai-writing/sample/extract` | Extract text from Word/PDF/Markdown/TXT samples for AI-writing imitation mode |
 | `POST /cop/saveSuperiorTask` | TaskCOP compatibility API (DeerFlow Bearer/session authentication required): create a situation-overview task with a sequential four-digit id (`0001`, `0002`, ...) and return the legacy `{state, msg, data}` envelope |
 | `POST /api/taskcop/import/tasks` | TaskCOP compatibility API (authenticated): import tasks from a JSON array or `tasks`/`records` envelope; every row's supplied `id`/`taskId` is authoritative and a duplicate id completely replaces the stored task |

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

@@ -61,6 +61,8 @@ WORKER_ID = f"w-{os.getpid()}-{_uuid.uuid4().hex[:8]}"
 _LEASE_TTL_SECONDS = 60
 _HEARTBEAT_INTERVAL = 20.0
 _CANCEL_POLL_INTERVAL = 5.0
+_REPORT_PROJECTION_RETRY_ATTEMPTS = 3
+_REPORT_PROJECTION_RETRY_DELAY_SECONDS = 0.5
 
 # Phase → progress% mapping for the job timeline.
 _PHASE_PROGRESS: dict[str, int] = {
@@ -1372,7 +1374,17 @@ class DeepResearchJobExecutor:
     async def _mark_completed(
         self, job_id: str, session_id: str, lease_owner: str | None, result: DeepResearchResult
     ) -> None:
-        # Job row → completed.
+        # Persist the report projection before exposing a terminal completed
+        # job.  The session row is the durable report itself; marking the job
+        # complete first and suppressing a session-write failure can otherwise
+        # make a successfully generated report disappear after refresh.
+        await self._persist_completed_report_projection(
+            job_id=job_id,
+            session_id=session_id,
+            result=result,
+        )
+
+        # Job row → completed only after the report is durable.
         await self._job_store.update_progress(
             job_id,
             lease_owner=lease_owner,
@@ -1385,26 +1397,62 @@ class DeepResearchJobExecutor:
                 "plan": result.plan,
             },
         )
-        # Session row → completed + report.
-        source_count = len(result.source_ids)
-        with suppress(Exception):  # noqa: BLE001
-            await self._session_store.update(
-                session_id,
-                status="completed",
-                report_markdown=result.report_markdown,
-                report_html=result.report_html,
-                plan_snapshot=result.plan,
-                # Keep the terminal job id alongside model usage.  The
-                # frontend uses it only to replay the persisted research trace
-                # after a refresh; it does not affect billing or job state.
-                usage_snapshot={**(result.usage or {}), "lastJobId": job_id},
-                source_count=source_count,
-                active_job_id=None,
-            )
         # Emit a terminal event for SSE consumers.
         with suppress(Exception):  # noqa: BLE001
             await self._emit_terminal(job_id, session_id, "job_completed")
 
+    async def _persist_completed_report_projection(
+        self,
+        *,
+        job_id: str,
+        session_id: str,
+        result: DeepResearchResult,
+    ) -> None:
+        """Durably save the report before publishing job completion.
+
+        A write can succeed while a short-lived MySQL proxy/readback hiccup
+        makes its caller fail. Retry the full owner-scoped update a few times;
+        if it still cannot be confirmed, propagate the error so ``_run`` marks
+        the job failed instead of advertising a report that cannot be reopened.
+        """
+        last_error: Exception | None = None
+        for attempt in range(1, _REPORT_PROJECTION_RETRY_ATTEMPTS + 1):
+            try:
+                updated = await self._session_store.update(
+                    session_id,
+                    status="completed",
+                    report_markdown=result.report_markdown,
+                    report_html=result.report_html,
+                    plan_snapshot=result.plan,
+                    # Keep the terminal job id alongside model usage.  The
+                    # frontend uses it only to replay the persisted research
+                    # trace after a refresh; it does not affect billing.
+                    usage_snapshot={**(result.usage or {}), "lastJobId": job_id},
+                    source_count=len(result.source_ids),
+                    active_job_id=None,
+                )
+                if updated is not None:
+                    return
+                last_error = RuntimeError(f"Deep Research session {session_id} no longer exists")
+            except asyncio.CancelledError:
+                raise
+            except Exception as exc:  # noqa: BLE001
+                last_error = exc
+
+            if attempt < _REPORT_PROJECTION_RETRY_ATTEMPTS:
+                logger.warning(
+                    "deep-research report projection failed for %s (attempt %s/%s); retrying",
+                    job_id,
+                    attempt,
+                    _REPORT_PROJECTION_RETRY_ATTEMPTS,
+                    exc_info=True,
+                )
+                await asyncio.sleep(_REPORT_PROJECTION_RETRY_DELAY_SECONDS * attempt)
+
+        raise RuntimeError(
+            f"研究报告未能保存到会话 {session_id}，任务不会标记为完成"
+        ) from last_error
+
     async def _mark_awaiting(
         self, job_id: str, session_id: str, pause: ResearchAwaitingInput
     ) -> None:

offline-backend-20260512/backend/app/gateway/routers/deep_research.py

@@ -639,17 +639,30 @@ async def start_job(session_id: str, body: StartJobRequest, request: Request) ->
         input_snapshot=input_snapshot,
     )
 
-    # Update session status.
-    await session_store.update(
-        session_id,
-        user_id=user_id,
-        status="running",
-        active_job_id=job["id"],
-    )
+    # The durable job is the source of truth once it has been committed.  A
+    # transient failure updating the display projection must not turn that
+    # successful write into a misleading 500 response or leave it undispatched.
+    # The executor re-projects the terminal status when it completes.
+    try:
+        await session_store.update(
+            session_id,
+            user_id=user_id,
+            status="running",
+            active_job_id=job["id"],
+        )
+    except Exception:  # noqa: BLE001
+        logger.exception(
+            "deep-research job %s was queued but its session running projection could not be updated",
+            job["id"],
+        )
 
-    # Nudge the dispatcher to claim immediately.
+    # Nudge the dispatcher even for an idempotent retry.  If an earlier request
+    # committed the job but failed before this point (for example a transient
+    # MySQL read-after-write failure), the retry receives the existing queued
+    # job.  Restricting the nudge to ``created`` would leave that durable job
+    # queued until the next periodic scan and make the UI appear stuck.
     dispatcher = _get_dispatcher(request)
-    if dispatcher is not None and created:
+    if dispatcher is not None:
         dispatcher.nudge()
 
     return {
@@ -810,36 +823,68 @@ async def stream_job(
                 if loop.time() >= next_durable_poll:
                     # Durable replay repairs a missed live delta after
                     # reconnect or when another worker owned the job.
-                    events = await event_store.list_after(job_id, after=cursor, limit=200)
-                    for ev in events:
-                        payload = {
-                            "seq": ev["seq"],
-                            "sessionId": session_id,
-                            "jobId": job_id,
-                            "type": ev["event_type"],
-                            "phase": ev.get("phase", "initializing"),
-                            "timestamp": _iso(ev.get("created_at")),
-                            "payload": ev.get("payload") or {},
-                        }
-                        yield format_sse("progress", payload, event_id=str(ev["seq"]))
-                        cursor = max(cursor, ev["seq"])
-                    current = await job_store.get_unscoped(job_id)
-                    if current and current.get("status") in TERMINAL_STATUSES:
+                    try:
+                        events = await event_store.list_after(job_id, after=cursor, limit=200)
+                        for ev in events:
+                            payload = {
+                                "seq": ev["seq"],
+                                "sessionId": session_id,
+                                "jobId": job_id,
+                                "type": ev["event_type"],
+                                "phase": ev.get("phase", "initializing"),
+                                "timestamp": _iso(ev.get("created_at")),
+                                "payload": ev.get("payload") or {},
+                            }
+                            yield format_sse("progress", payload, event_id=str(ev["seq"]))
+                            cursor = max(cursor, ev["seq"])
+                        current = await job_store.get_unscoped(job_id)
+                        if current and current.get("status") in TERMINAL_STATUSES:
+                            yield format_sse(
+                                "progress",
+                                {
+                                    "seq": cursor + 1,
+                                    "sessionId": session_id,
+                                    "jobId": job_id,
+                                    "type": "job_status",
+                                    "phase": current.get("phase", "done"),
+                                    "timestamp": _iso(datetime.now(UTC)),
+                                    "payload": {"status": current["status"], "progress": current.get("progress", 0)},
+                                },
+                                event_id=str(cursor + 1),
+                            )
+                            return
+                        next_durable_poll = loop.time() + _POLL_INTERVAL_S
+                    except asyncio.CancelledError:
+                        raise
+                    except Exception:  # noqa: BLE001
+                        # Do not let one read-replica / transient DB failure
+                        # abort the ASGI generator (which surfaces as an
+                        # ExceptionGroup and leaves the page stuck). Live
+                        # frames can still reach an attached browser, and the
+                        # next durable poll repairs the cursor on recovery.
+                        logger.warning(
+                            "deep-research SSE durable replay temporarily failed for job %s; keeping stream open",
+                            job_id,
+                            exc_info=True,
+                        )
                         yield format_sse(
                             "progress",
                             {
-                                "seq": cursor + 1,
+                                "seq": 0,
+                                "liveId": f"durable-replay-retry:{job_id}:{int(loop.time() * 1000)}",
                                 "sessionId": session_id,
                                 "jobId": job_id,
-                                "type": "job_status",
-                                "phase": current.get("phase", "done"),
+                                "type": "warning",
+                                "phase": "initializing",
                                 "timestamp": _iso(datetime.now(UTC)),
-                                "payload": {"status": current["status"], "progress": current.get("progress", 0)},
+                                "payload": {
+                                    "code": "durable_replay_temporarily_unavailable",
+                                    "message": "进度同步暂时不可用，任务仍在继续；正在自动重试。",
+                                    "recoverable": True,
+                                },
                             },
-                            event_id=str(cursor + 1),
                         )
-                        return
-                    next_durable_poll = loop.time() + _POLL_INTERVAL_S
+                        next_durable_poll = loop.time() + max(_POLL_INTERVAL_S, 3.0)
                 if live_queue is None:
                     await asyncio.sleep(max(0.01, next_durable_poll - loop.time()))
                     continue

offline-backend-20260512/backend/app/gateway/routers/intent.py

@@ -195,17 +195,22 @@ def _intent_thread_metadata(
     if not task_id:
         return metadata
 
-    metadata.update(
-        {
-            "taskId": task_id,
-            "position_roundtable_task_id": task_id,
-            "position_task_id": f"{task_id}:intelligence",
-            "position_id": "intelligence",
-            "position_roundtable_phase": "intent",
-        }
-    )
-    if intent_agent_id == POSITION_INTENT_AGENT_ID:
-        metadata["position_roundtable_intent_agent_id"] = intent_agent_id
+    metadata["taskId"] = task_id
+    # 8BF 使用通用意图接口但有明确的上下文 kind；它不能被岗位会商的
+    # 会话索引误认为岗位任务，否则两个页面的历史记录会串在一起。
+    # 其它历史调用保持原有 metadata 形状，避免影响既有会商恢复逻辑。
+    is_eightbf_context = task_context.get("kind") == "eightbf_roundtable_task"
+    if intent_agent_id == POSITION_INTENT_AGENT_ID or not is_eightbf_context:
+        metadata.update(
+            {
+                "position_roundtable_task_id": task_id,
+                "position_task_id": f"{task_id}:intelligence",
+                "position_id": "intelligence",
+                "position_roundtable_phase": "intent",
+            }
+        )
+        if intent_agent_id == POSITION_INTENT_AGENT_ID:
+            metadata["position_roundtable_intent_agent_id"] = intent_agent_id
     return metadata
 
 

offline-backend-20260512/backend/packages/harness/deerflow/agents/deep_research/adapters/event_sink.py

@@ -48,6 +48,11 @@ class PersistingEventSink:
         self._session_id = session_id
         self._job_id = job_id
         self._on_event = on_event
+        # A durable event row normally owns the sequence.  When the event
+        # database is temporarily unavailable, use distinct negative sequence
+        # numbers for the live-only path so clients can render every milestone
+        # without advancing their durable ``after`` cursor.
+        self._fallback_seq = 0
 
     async def emit(
         self,
@@ -56,26 +61,66 @@ class PersistingEventSink:
         phase: ResearchPhase = "initializing",
         payload: dict | None = None,
     ) -> DeepResearchEvent:
-        row = await self._repo.append(
-            session_id=self._session_id,
-            job_id=self._job_id,
-            event_type=type,
-            phase=phase,
-            payload=payload or {},
-        )
-        # Reconstruct the envelope from the persisted row (seq is now assigned).
-        event = DeepResearchEvent(
-            seq=int(row.get("seq", 0)),
-            session_id=self._session_id,
-            job_id=self._job_id,
-            type=row.get("event_type", type),
-            phase=row.get("phase", phase),
-            payload=row.get("payload") or payload or {},
-        )
-        if self._on_event is not None:
-            await self._on_event(event.to_sse_dict())
+        event_payload = payload or {}
+        try:
+            row = await self._repo.append(
+                session_id=self._session_id,
+                job_id=self._job_id,
+                event_type=type,
+                phase=phase,
+                payload=event_payload,
+            )
+            # Reconstruct the envelope from the persisted row (seq is now assigned).
+            event = DeepResearchEvent(
+                seq=int(row.get("seq", 0)),
+                session_id=self._session_id,
+                job_id=self._job_id,
+                type=row.get("event_type", type),
+                phase=row.get("phase", phase),
+                payload=row.get("payload") or event_payload,
+            )
+        except Exception:  # noqa: BLE001
+            # Progress persistence must not turn a recoverable database blip
+            # into a failed report. The job/status stores still record terminal
+            # state; this live-only frame lets an attached UI stay responsive
+            # until durable replay becomes available again after reconnect.
+            self._fallback_seq -= 1
+            logger.exception(
+                "deep-research event persistence failed; continuing job with a live-only event (job=%s, type=%s, phase=%s)",
+                self._job_id,
+                type,
+                phase,
+            )
+            event = DeepResearchEvent(
+                seq=self._fallback_seq,
+                session_id=self._session_id,
+                job_id=self._job_id,
+                type=type,
+                phase=phase,
+                payload={**event_payload, "eventPersistenceDegraded": True},
+            )
+        await self._publish_event(event)
         return event
 
+    async def _publish_event(self, event: DeepResearchEvent) -> None:
+        """Push a persisted or live-only event without risking the worker."""
+        if self._on_event is None:
+            return
+        wire = event.to_sse_dict()
+        if event.seq < 0:
+            wire["liveId"] = f"event-persistence-fallback:{self._job_id}:{abs(event.seq)}"
+        try:
+            await self._on_event(wire)
+        except Exception:  # noqa: BLE001
+            # A browser disconnect or a live-hub issue must not abort report
+            # generation; persisted rows will still be replayable on reconnect.
+            logger.warning(
+                "deep-research live event publish failed; job continues (job=%s, seq=%s)",
+                self._job_id,
+                event.seq,
+                exc_info=True,
+            )
+
     async def publish_live(
         self,
         type: str,  # noqa: A002
@@ -85,16 +130,24 @@ class PersistingEventSink:
     ) -> None:
         if self._on_event is None:
             return
-        await self._on_event(
-            {
-                "seq": 0,
-                "sessionId": self._session_id,
-                "jobId": self._job_id,
-                "type": type,
-                "phase": phase,
-                "payload": payload or {},
-            }
-        )
+        try:
+            await self._on_event(
+                {
+                    "seq": 0,
+                    "sessionId": self._session_id,
+                    "jobId": self._job_id,
+                    "type": type,
+                    "phase": phase,
+                    "payload": payload or {},
+                }
+            )
+        except Exception:  # noqa: BLE001
+            logger.warning(
+                "deep-research live frame publish failed; job continues (job=%s, type=%s)",
+                self._job_id,
+                type,
+                exc_info=True,
+            )
 
 
 class NullEventSink:

offline-backend-20260512/backend/packages/harness/deerflow/agents/middlewares/summarization_middleware.py

@@ -27,14 +27,15 @@ from typing import Any, Protocol, override, runtime_checkable
 from langchain.agents import AgentState
 from langchain.agents.middleware import SummarizationMiddleware
 from langchain.agents.middleware.types import ModelCallResult, ModelRequest, ModelResponse
-from langchain_core.messages import (
-    AIMessage,
-    AnyMessage,
-    SystemMessage,
-    ToolMessage,
-    message_to_dict,
-    messages_from_dict,
-)
+from langchain_core.messages import (
+    AIMessage,
+    AnyMessage,
+    HumanMessage,
+    SystemMessage,
+    ToolMessage,
+    message_to_dict,
+    messages_from_dict,
+)
 from langgraph.config import get_config
 from langgraph.runtime import Runtime
 
@@ -145,8 +146,8 @@ class _SummaryCacheEntry:
 
     summarized_ids: frozenset[str]
     summary_text: str
-    preserved_ids: frozenset[str] = frozenset()
-    preserved_messages: tuple[AnyMessage, ...] = ()
+    preserved_ids: frozenset[str] = frozenset()
+    preserved_messages: tuple[AnyMessage, ...] = ()
 
 
 # Process-wide summary cache shared across every middleware instance.
@@ -160,51 +161,99 @@ class _SummaryCacheEntry:
 # so only the first threshold crossing (or a genuine re-compaction) pays for an
 # LLM call. A lock guards the non-atomic LRU eviction against concurrent
 # threads (subagents / parallel conversations share this process).
-_GLOBAL_SUMMARY_CACHE: OrderedDict[str, _SummaryCacheEntry] = OrderedDict()
-_GLOBAL_SUMMARY_CACHE_LOCK = threading.Lock()
-_STATE_CACHE_KEY = "context_summary_cache"
-
-
-def _clear_global_summary_cache() -> None:
-    """Drop every cached summary. Intended for tests/diagnostics only."""
-    with _GLOBAL_SUMMARY_CACHE_LOCK:
-        _GLOBAL_SUMMARY_CACHE.clear()
-
-
-def _serialize_cache_entry(entry: _SummaryCacheEntry) -> dict[str, Any]:
-    """Serialize a summary cache entry so it can live in LangGraph state."""
-    return {
-        "version": 1,
-        "summarized_ids": sorted(entry.summarized_ids),
-        "summary_text": entry.summary_text,
-        "preserved_ids": sorted(entry.preserved_ids),
-        "preserved_messages": [message_to_dict(message) for message in entry.preserved_messages],
-    }
-
-
-def _deserialize_cache_entry(raw: Any) -> _SummaryCacheEntry | None:
-    """Best-effort restore of a state-backed summary cache entry."""
-    if not isinstance(raw, dict):
-        return None
-    try:
-        summary_text = raw.get("summary_text")
-        if not isinstance(summary_text, str) or not summary_text:
-            return None
-        summarized_ids = raw.get("summarized_ids") or []
-        preserved_ids = raw.get("preserved_ids") or []
-        preserved_messages_raw = raw.get("preserved_messages") or []
-        if not isinstance(summarized_ids, list) or not isinstance(preserved_ids, list) or not isinstance(preserved_messages_raw, list):
-            return None
-        preserved_messages = tuple(messages_from_dict(preserved_messages_raw))
-        return _SummaryCacheEntry(
-            summarized_ids=frozenset(str(item) for item in summarized_ids if item is not None),
-            summary_text=summary_text,
-            preserved_ids=frozenset(str(item) for item in preserved_ids if item is not None),
-            preserved_messages=preserved_messages,
-        )
-    except Exception:
-        logger.debug("summarization: failed to restore state-backed summary cache", exc_info=True)
-        return None
+_GLOBAL_SUMMARY_CACHE: OrderedDict[str, _SummaryCacheEntry] = OrderedDict()
+_GLOBAL_SUMMARY_CACHE_LOCK = threading.Lock()
+_STATE_CACHE_KEY = "context_summary_cache"
+
+
+def _clear_global_summary_cache() -> None:
+    """Drop every cached summary. Intended for tests/diagnostics only."""
+    with _GLOBAL_SUMMARY_CACHE_LOCK:
+        _GLOBAL_SUMMARY_CACHE.clear()
+
+
+def _serialize_cache_entry(entry: _SummaryCacheEntry) -> dict[str, Any]:
+    """Serialize a summary cache entry so it can live in LangGraph state."""
+    return {
+        "version": 1,
+        "summarized_ids": sorted(entry.summarized_ids),
+        "summary_text": entry.summary_text,
+        "preserved_ids": sorted(entry.preserved_ids),
+        "preserved_messages": [message_to_dict(message) for message in entry.preserved_messages],
+    }
+
+
+def _deserialize_cache_entry(raw: Any) -> _SummaryCacheEntry | None:
+    """Best-effort restore of a state-backed summary cache entry."""
+    if not isinstance(raw, dict):
+        return None
+    try:
+        summary_text = raw.get("summary_text")
+        if not isinstance(summary_text, str) or not summary_text:
+            return None
+        summarized_ids = raw.get("summarized_ids") or []
+        preserved_ids = raw.get("preserved_ids") or []
+        preserved_messages_raw = raw.get("preserved_messages") or []
+        if not isinstance(summarized_ids, list) or not isinstance(preserved_ids, list) or not isinstance(preserved_messages_raw, list):
+            return None
+        preserved_messages = tuple(messages_from_dict(preserved_messages_raw))
+        return _SummaryCacheEntry(
+            summarized_ids=frozenset(str(item) for item in summarized_ids if item is not None),
+            summary_text=summary_text,
+            preserved_ids=frozenset(str(item) for item in preserved_ids if item is not None),
+            preserved_messages=preserved_messages,
+        )
+    except Exception:
+        logger.debug("summarization: failed to restore state-backed summary cache", exc_info=True)
+        return None
+
+
+def _has_nonempty_human_message(message: AnyMessage) -> bool:
+    """Whether a message is a real, non-empty user turn.
+
+    Some OpenAI-compatible gateways reject a completion request that contains
+    only ``system`` / ``assistant`` / ``tool`` roles. That can legitimately
+    happen after compaction: the active model call is continuing a tool turn,
+    while the user request that started the turn has already moved into the
+    internal summary.
+    """
+    if not isinstance(message, HumanMessage):
+        return False
+    content = message.content
+    if isinstance(content, str):
+        return bool(content.strip())
+    if isinstance(content, list):
+        return any(bool(block.strip()) if isinstance(block, str) else bool((block.get("text") or block.get("content") or "").strip()) if isinstance(block, dict) else False for block in content)
+    return bool(str(content).strip())
+
+
+def _ensure_latest_human_message(
+    source_messages: list[AnyMessage],
+    compressed_messages: list[AnyMessage],
+) -> list[AnyMessage]:
+    """Keep the latest real user request visible to strict chat gateways.
+
+    The message is inserted after leading system messages, so it neither
+    exposes the internal summary to the UI nor splits an AI tool-call from its
+    following ToolMessages. When there was no user turn in the source at all,
+    do not fabricate one: that is an upstream caller bug and should remain
+    diagnosable rather than silently changing the task.
+    """
+    if any(_has_nonempty_human_message(message) for message in compressed_messages):
+        return compressed_messages
+
+    latest_human = next(
+        (message for message in reversed(source_messages) if _has_nonempty_human_message(message)),
+        None,
+    )
+    if latest_human is None:
+        logger.warning("summarization: compressed model request has no user message and source history contains none; passing it through unchanged")
+        return compressed_messages
+
+    insert_at = 0
+    while insert_at < len(compressed_messages) and isinstance(compressed_messages[insert_at], SystemMessage):
+        insert_at += 1
+    return [*compressed_messages[:insert_at], latest_human, *compressed_messages[insert_at:]]
 
 
 class DeerFlowSummarizationMiddleware(SummarizationMiddleware):
@@ -231,8 +280,8 @@ class DeerFlowSummarizationMiddleware(SummarizationMiddleware):
         # Per-thread cache of the most recent summary: thread_id -> (summarized message ids, summary text).
         # Points at the *process-wide* cache (not a fresh per-instance dict) so a
         # summary survives across runs/turns — see ``_GLOBAL_SUMMARY_CACHE``.
-        self._summary_cache: OrderedDict[str, _SummaryCacheEntry] = _GLOBAL_SUMMARY_CACHE
-        self._pending_state_cache_update: dict[str, Any] | None = None
+        self._summary_cache: OrderedDict[str, _SummaryCacheEntry] = _GLOBAL_SUMMARY_CACHE
+        self._pending_state_cache_update: dict[str, Any] | None = None
 
     @override
     def before_model(self, state: AgentState, runtime: Runtime) -> dict | None:
@@ -248,21 +297,21 @@ class DeerFlowSummarizationMiddleware(SummarizationMiddleware):
         """Async counterpart of :meth:`before_model` — also a no-op."""
         return None
 
-    @override
-    def after_model(self, state: AgentState, runtime: Runtime) -> dict | None:
-        """Persist the latest reusable summary cache into thread state."""
-        if self._pending_state_cache_update is None:
-            return None
-        update = self._pending_state_cache_update
-        self._pending_state_cache_update = None
-        return {_STATE_CACHE_KEY: update}
-
-    @override
-    async def aafter_model(self, state: AgentState, runtime: Runtime) -> dict | None:
-        """Async counterpart of :meth:`after_model`."""
-        return self.after_model(state, runtime)
-
-    # ---- wrap_model_call: transient, display-preserving compression ----------
+    @override
+    def after_model(self, state: AgentState, runtime: Runtime) -> dict | None:
+        """Persist the latest reusable summary cache into thread state."""
+        if self._pending_state_cache_update is None:
+            return None
+        update = self._pending_state_cache_update
+        self._pending_state_cache_update = None
+        return {_STATE_CACHE_KEY: update}
+
+    @override
+    async def aafter_model(self, state: AgentState, runtime: Runtime) -> dict | None:
+        """Async counterpart of :meth:`after_model`."""
+        return self.after_model(state, runtime)
+
+    # ---- wrap_model_call: transient, display-preserving compression ----------
 
     @override
     def wrap_model_call(
@@ -376,20 +425,21 @@ class DeerFlowSummarizationMiddleware(SummarizationMiddleware):
             return None
 
         thread_id = self._resolve_request_thread_id(request)
-        if thread_id is None:
-            logger.debug("summarization: thread_id unresolved; cross-turn summary cache disabled for this call")
-        cached = self._get_cache_entry(thread_id) if thread_id else None
-        if cached is None:
-            cached = _deserialize_cache_entry(getattr(request, "state", {}).get(_STATE_CACHE_KEY))
-            if cached is not None and thread_id:
-                self._store_cache_entry(thread_id, cached)
-        present_ids = {m.id for m in messages}
-        cache_usable = cached is not None and cached.summarized_ids <= present_ids and cached.preserved_ids <= present_ids
-        if cache_usable:
-            preserved = list(cached.preserved_messages)
-            excluded_ids = cached.summarized_ids | cached.preserved_ids
-            preserved.extend(m for m in messages if m.id not in excluded_ids)
-            effective = [self._summary_message(cached.summary_text), *preserved]
+        if thread_id is None:
+            logger.debug("summarization: thread_id unresolved; cross-turn summary cache disabled for this call")
+        cached = self._get_cache_entry(thread_id) if thread_id else None
+        if cached is None:
+            cached = _deserialize_cache_entry(getattr(request, "state", {}).get(_STATE_CACHE_KEY))
+            if cached is not None and thread_id:
+                self._store_cache_entry(thread_id, cached)
+        present_ids = {m.id for m in messages}
+        cache_usable = cached is not None and cached.summarized_ids <= present_ids and cached.preserved_ids <= present_ids
+        if cache_usable:
+            preserved = list(cached.preserved_messages)
+            excluded_ids = cached.summarized_ids | cached.preserved_ids
+            preserved.extend(m for m in messages if m.id not in excluded_ids)
+            effective = [self._summary_message(cached.summary_text), *preserved]
+            effective = _ensure_latest_human_message(messages, effective)
             # Decide re-summarization on the *counted* size of the compressed
             # context only — NOT ``_should_summarize`` — because that folds in the
             # last AIMessage's reported ``usage_metadata`` (system prompt + tool
@@ -400,8 +450,7 @@ class DeerFlowSummarizationMiddleware(SummarizationMiddleware):
                 # Refresh LRU recency and reuse without an LLM call.
                 self._touch_cache_entry(thread_id)
                 logger.info(
-                    "summarization: 上下文压缩(复用缓存摘要,无 LLM 调用) context compaction reusing cached summary "
-                    "thread=%s full_tokens=%d compressed_tokens=%d preserved_msgs=%d",
+                    "summarization: 上下文压缩(复用缓存摘要,无 LLM 调用) context compaction reusing cached summary thread=%s full_tokens=%d compressed_tokens=%d preserved_msgs=%d",
                     thread_id,
                     self.token_counter(messages),
                     self.token_counter(effective),
@@ -415,7 +464,7 @@ class DeerFlowSummarizationMiddleware(SummarizationMiddleware):
         # an "it compacts every turn" report is diagnosable straight from the
         # backend log: cache=miss → thread_id/key problem; unusable → message-id
         # instability; stale → preserved genuinely grew back over the trigger.
-        cache_state = "miss" if cached is None else "stale"
+        cache_state = "miss" if cached is None else "stale"
         logger.info(
             "summarization: compacting thread=%s cache=%s counted_tokens=%d msgs=%d",
             thread_id,
@@ -452,22 +501,23 @@ class DeerFlowSummarizationMiddleware(SummarizationMiddleware):
     ) -> list[AnyMessage]:
         """Assemble the compressed message list and update the per-thread cache."""
         compressed = [self._summary_message(summary), *preserved]
+        compressed = _ensure_latest_human_message([*messages_to_summarize, *preserved], compressed)
 
         if thread_id:
             summarized_ids = frozenset(m.id for m in messages_to_summarize if m.id is not None)
             preserved_ids = {m.id for m in preserved if m.id is not None}
-            self._store_cache_entry(
-                thread_id,
-                entry := _SummaryCacheEntry(
-                    summarized_ids=summarized_ids,
-                    summary_text=summary,
-                    preserved_ids=frozenset(preserved_ids),
-                    preserved_messages=tuple(preserved),
-                ),
-            )
-            self._pending_state_cache_update = _serialize_cache_entry(entry)
-
-        return compressed
+            self._store_cache_entry(
+                thread_id,
+                entry := _SummaryCacheEntry(
+                    summarized_ids=summarized_ids,
+                    summary_text=summary,
+                    preserved_ids=frozenset(preserved_ids),
+                    preserved_messages=tuple(preserved),
+                ),
+            )
+            self._pending_state_cache_update = _serialize_cache_entry(entry)
+
+        return compressed
 
     def _get_cache_entry(self, thread_id: str) -> _SummaryCacheEntry | None:
         """Return the cached entry for ``thread_id`` (thread-safe read)."""
@@ -571,7 +621,7 @@ class DeerFlowSummarizationMiddleware(SummarizationMiddleware):
             return None
         return _resolve_thread_id(runtime)
 
-    def _summary_message(self, summary: str) -> SystemMessage:
+    def _summary_message(self, summary: str) -> SystemMessage:
         """Build the summary message handed to the model.
 
         It is named ``summary`` and flagged ``hide_from_ui`` defensively. Because
@@ -579,21 +629,21 @@ class DeerFlowSummarizationMiddleware(SummarizationMiddleware):
         state), the frontend never receives it — the user keeps seeing the
         original conversation.
         """
-        return SystemMessage(
-            content=(
-                "Internal conversation summary for continuity only. Use it silently "
-                "to answer the user's latest request. Never reveal, quote, mention, "
-                "or describe this summary or the fact that context was compacted.\n\n"
-                "<internal_context_summary>\n"
-                f"{summary}\n"
-                "</internal_context_summary>"
-            ),
+        return SystemMessage(
+            content=(
+                "Internal conversation summary for continuity only. Use it silently "
+                "to answer the user's latest request. Never reveal, quote, mention, "
+                "or describe this summary or the fact that context was compacted.\n\n"
+                "<internal_context_summary>\n"
+                f"{summary}\n"
+                "</internal_context_summary>"
+            ),
             name="summary",
             additional_kwargs={"hide_from_ui": True, "lc_source": "summarization"},
         )
 
     @override
-    def _build_new_messages(self, summary: str) -> list[AnyMessage]:
+    def _build_new_messages(self, summary: str) -> list[AnyMessage]:
         """Kept for backwards compatibility; delegates to :meth:`_summary_message`."""
         return [self._summary_message(summary)]
 

offline-backend-20260512/backend/packages/harness/deerflow/persistence/deep_research_events/sql.py

@@ -68,13 +68,7 @@ class DeepResearchEventRepository:
         """
         for attempt in range(max_retries):
             async with self._sf() as session:
-                current_max = (
-                    await session.execute(
-                        select(func.max(DeepResearchEventRow.seq)).where(
-                            DeepResearchEventRow.job_id == job_id
-                        )
-                    )
-                ).scalar() or 0
+                current_max = (await session.execute(select(func.max(DeepResearchEventRow.seq)).where(DeepResearchEventRow.job_id == job_id))).scalar() or 0
                 next_seq = int(current_max) + 1
                 row = DeepResearchEventRow(
                     job_id=job_id,
@@ -86,49 +80,38 @@ class DeepResearchEventRepository:
                 )
                 session.add(row)
                 try:
+                    # Do not issue ``session.refresh(row)`` after commit. On a
+                    # MySQL read/write-splitting deployment that read can land
+                    # on a lagging replica, which cannot see the just-committed
+                    # event and raises ``Could not refresh instance``. Flush
+                    # assigns generated primary-key/default values on the writer
+                    # connection, so the plain dict is complete before commit.
+                    await session.flush()
+                    result = self._row_to_dict(row)
                     await session.commit()
-                    await session.refresh(row)
-                    return self._row_to_dict(row)
+                    return result
                 except IntegrityError:
                     await session.rollback()
                     continue  # retry with recomputed max
-        raise RuntimeError(
-            f"deep_research_events.append: seq collision persisted after {max_retries} retries (job={job_id})"
-        )
+        raise RuntimeError(f"deep_research_events.append: seq collision persisted after {max_retries} retries (job={job_id})")
 
-    async def list_after(
-        self, job_id: str, *, after: int = 0, limit: int = 200
-    ) -> list[dict[str, Any]]:
+    async def list_after(self, job_id: str, *, after: int = 0, limit: int = 200) -> list[dict[str, Any]]:
         """Events with ``seq > after`` in ascending order (SSE replay, §15.3)."""
         async with self._sf() as session:
-            stmt = (
-                select(DeepResearchEventRow)
-                .where(DeepResearchEventRow.job_id == job_id)
-                .where(DeepResearchEventRow.seq > after)
-                .order_by(DeepResearchEventRow.seq.asc())
-                .limit(limit)
-            )
+            stmt = select(DeepResearchEventRow).where(DeepResearchEventRow.job_id == job_id).where(DeepResearchEventRow.seq > after).order_by(DeepResearchEventRow.seq.asc()).limit(limit)
             rows = (await session.execute(stmt)).scalars().all()
             return [self._row_to_dict(r) for r in rows]
 
     async def last_seq(self, job_id: str) -> int:
         async with self._sf() as session:
-            mx = (
-                await session.execute(
-                    select(func.max(DeepResearchEventRow.seq)).where(
-                        DeepResearchEventRow.job_id == job_id
-                    )
-                )
-            ).scalar()
+            mx = (await session.execute(select(func.max(DeepResearchEventRow.seq)).where(DeepResearchEventRow.job_id == job_id))).scalar()
             return int(mx or 0)
 
     async def delete_by_job(self, job_id: str) -> int:
         from sqlalchemy import delete
 
         async with self._sf() as session:
-            result = await session.execute(
-                delete(DeepResearchEventRow).where(DeepResearchEventRow.job_id == job_id)
-            )
+            result = await session.execute(delete(DeepResearchEventRow).where(DeepResearchEventRow.job_id == job_id))
             await session.commit()
             return result.rowcount
 
@@ -136,8 +119,6 @@ class DeepResearchEventRepository:
         from sqlalchemy import delete
 
         async with self._sf() as session:
-            result = await session.execute(
-                delete(DeepResearchEventRow).where(DeepResearchEventRow.session_id == session_id)
-            )
+            result = await session.execute(delete(DeepResearchEventRow).where(DeepResearchEventRow.session_id == session_id))
             await session.commit()
             return result.rowcount

offline-backend-20260512/backend/packages/harness/deerflow/persistence/deep_research_jobs/sql.py

@@ -113,9 +113,18 @@ class DeepResearchJobRepository:
                         phase=phase,
                     )
                     session.add(row)
+                    # Do not issue ``session.refresh(row)`` after commit.  A
+                    # MySQL read/write-splitting proxy can route that read to
+                    # a lagging replica, even though the INSERT has already
+                    # committed on the writer.  The caller would receive 500
+                    # while an invisible queued job remains in the database.
+                    # ``flush`` materializes ORM defaults on the writer, so we
+                    # can return a complete, durable snapshot without a
+                    # replica-dependent readback.
+                    await session.flush()
+                    result = self._row_to_dict(row)
                     await session.commit()
-                    await session.refresh(row)
-                    return self._row_to_dict(row), True
+                    return result, True
             except IntegrityError:
                 # Unique-index conflict → an active job already exists.
                 pass
@@ -230,13 +239,14 @@ class DeepResearchJobRepository:
             )
             if result.rowcount == 0:
                 return None
-            await session.commit()
             row = (
                 await session.execute(
                     select(DeepResearchJobRow).where(DeepResearchJobRow.id == job_id)
                 )
             ).scalar_one_or_none()
-            return self._row_to_dict(row) if row else None
+            snapshot = self._row_to_dict(row) if row else None
+            await session.commit()
+            return snapshot
 
     # ── heartbeat ───────────────────────────────────────────────────────────
 
@@ -351,13 +361,14 @@ class DeepResearchJobRepository:
             )
             if result.rowcount == 0:
                 return None
-            await session.commit()
             row = (
                 await session.execute(
                     select(DeepResearchJobRow).where(DeepResearchJobRow.id == job_id)
                 )
             ).scalar_one_or_none()
-            return self._row_to_dict(row) if row else None
+            snapshot = self._row_to_dict(row) if row else None
+            await session.commit()
+            return snapshot
 
     # ── two-phase cancel ────────────────────────────────────────────────────
 
@@ -385,13 +396,14 @@ class DeepResearchJobRepository:
             )
             if result.rowcount == 0:
                 return None
-            await session.commit()
             row = (
                 await session.execute(
                     select(DeepResearchJobRow).where(DeepResearchJobRow.id == job_id)
                 )
             ).scalar_one_or_none()
-            return self._row_to_dict(row) if row else None
+            snapshot = self._row_to_dict(row) if row else None
+            await session.commit()
+            return snapshot
 
     async def finalize_cancel(
         self, job_id: str, *, lease_owner: str | None = None
@@ -419,13 +431,14 @@ class DeepResearchJobRepository:
             )
             if result.rowcount == 0:
                 return None
-            await session.commit()
             row = (
                 await session.execute(
                     select(DeepResearchJobRow).where(DeepResearchJobRow.id == job_id)
                 )
             ).scalar_one_or_none()
-            return self._row_to_dict(row) if row else None
+            snapshot = self._row_to_dict(row) if row else None
+            await session.commit()
+            return snapshot
 
     # ── resume after human-review interrupt ─────────────────────────────────
 
@@ -460,13 +473,14 @@ class DeepResearchJobRepository:
             )
             if result.rowcount == 0:
                 return None
-            await session.commit()
             row = (
                 await session.execute(
                     select(DeepResearchJobRow).where(DeepResearchJobRow.id == job_id)
                 )
             ).scalar_one_or_none()
-            return self._row_to_dict(row) if row else None
+            snapshot = self._row_to_dict(row) if row else None
+            await session.commit()
+            return snapshot
 
     # ── cleanup ─────────────────────────────────────────────────────────────
 

offline-backend-20260512/backend/packages/harness/deerflow/persistence/deep_research_messages/sql.py

@@ -74,9 +74,13 @@ class DeepResearchMessageRepository:
                 seq=int(current_max) + 1,
             )
             session.add(row)
+            # Keep the successful append independent of a later replica read.
+            # The explicit id/seq and Python-side defaults are available after
+            # flush, before the transaction releases its writer connection.
+            await session.flush()
+            snapshot = self._row_to_dict(row)
             await session.commit()
-            await session.refresh(row)
-            return self._row_to_dict(row)
+            return snapshot
 
     async def list_by_session(
         self, session_id: str, *, user_id: str | _AutoSentinel = AUTO, limit: int = 100
@@ -139,10 +143,11 @@ class DeepResearchMessageRepository:
             if not result.rowcount:
                 await session.rollback()
                 return None
-            await session.commit()
             stmt = select(DeepResearchMessageRow).where(*where)
             row = (await session.execute(stmt)).scalar_one_or_none()
-            return self._row_to_dict(row) if row is not None else None
+            snapshot = self._row_to_dict(row) if row is not None else None
+            await session.commit()
+            return snapshot
 
     async def delete_by_session(self, session_id: str) -> int:
         from sqlalchemy import delete

offline-backend-20260512/backend/packages/harness/deerflow/persistence/deep_research_sessions/sql.py

@@ -85,9 +85,13 @@ class DeepResearchSessionRepository:
                 runtime_thread_id=runtime_thread_id,
             )
             session.add(row)
+            # New sessions are a control-plane write. Avoid a post-commit
+            # refresh because MySQL read/write-splitting can make a just
+            # created session look absent on a lagging replica.
+            await session.flush()
+            snapshot = self._row_to_dict(row)
             await session.commit()
-            await session.refresh(row)
-            return self._row_to_dict(row)
+            return snapshot
 
     # ── read ────────────────────────────────────────────────────────────────
 
@@ -187,13 +191,14 @@ class DeepResearchSessionRepository:
             )
             if result.rowcount == 0:
                 return None
-            await session.commit()
             row = (
                 await session.execute(
                     select(DeepResearchSessionRow).where(DeepResearchSessionRow.id == session_id)
                 )
             ).scalar_one_or_none()
-            return self._row_to_dict(row) if row else None
+            snapshot = self._row_to_dict(row) if row else None
+            await session.commit()
+            return snapshot
 
     async def replace_report_if_unchanged(
         self,
@@ -231,13 +236,14 @@ class DeepResearchSessionRepository:
             )
             if result.rowcount == 0:
                 return None
-            await session.commit()
             row = (
                 await session.execute(
                     select(DeepResearchSessionRow).where(DeepResearchSessionRow.id == session_id)
                 )
             ).scalar_one_or_none()
-            return self._row_to_dict(row) if row else None
+            snapshot = self._row_to_dict(row) if row else None
+            await session.commit()
+            return snapshot
 
     # ── delete ──────────────────────────────────────────────────────────────
 

offline-backend-20260512/backend/packages/harness/deerflow/persistence/deep_research_sources/model.py

@@ -23,6 +23,13 @@ from deerflow.persistence.types import BeijingDateTime, PortableLongText
 
 class DeepResearchSourceRow(Base):
     __tablename__ = "deep_research_sources"
+    __table_args__ = (
+        # MySQL requires a prefix length when indexing TEXT columns.  The
+        # column remains unbounded so long URLs are preserved; the prefix is
+        # only an index implementation detail and equality filters still
+        # compare the complete URL value.
+        sa.Index("ix_deep_research_sources_url", "url", mysql_length=512),
+    )
 
     id: Mapped[str] = mapped_column(String(64), primary_key=True)
     session_id: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
@@ -32,7 +39,7 @@ class DeepResearchSourceRow(Base):
     # The query that first surfaced this material.
     first_seen_query: Mapped[str | None] = mapped_column(Text, nullable=True)
     title: Mapped[str] = mapped_column(Text, nullable=False, default="")
-    url: Mapped[str | None] = mapped_column(Text, nullable=True, index=True)
+    url: Mapped[str | None] = mapped_column(Text, nullable=True)
     raw_content: Mapped[str] = mapped_column(PortableLongText, nullable=False, default="")
     snippet: Mapped[str | None] = mapped_column(Text, nullable=True)
     source: Mapped[str] = mapped_column(String(256), nullable=False, default="")

offline-backend-20260512/backend/packages/harness/deerflow/persistence/deep_research_sources/sql.py

@@ -37,9 +37,14 @@ class DeepResearchSourceRepository:
             row = DeepResearchSourceRow(session_id=session_id, job_id=job_id, **fields)
             session.add(row)
             try:
+                # The source was produced on the worker's write path. Build
+                # its response on that same connection so a replica-lag read
+                # cannot turn a successfully collected source into a failed
+                # research run.
+                await session.flush()
+                snapshot = self._row_to_dict(row)
                 await session.commit()
-                await session.refresh(row)
-                return self._row_to_dict(row)
+                return snapshot
             except Exception:
                 await session.rollback()
         # Fallback: find existing by rec_uuid or url.
@@ -147,13 +152,14 @@ class DeepResearchSourceRepository:
             )
             if result.rowcount == 0:
                 return None
-            await session.commit()
             row = (
                 await session.execute(
                     select(DeepResearchSourceRow).where(DeepResearchSourceRow.id == source_id)
                 )
             ).scalar_one_or_none()
-            return self._row_to_dict(row) if row else None
+            snapshot = self._row_to_dict(row) if row else None
+            await session.commit()
+            return snapshot
 
     async def delete_by_session(self, session_id: str) -> int:
         from sqlalchemy import delete

offline-backend-20260512/backend/tests/test_deep_research_event_sink.py

@@ -0,0 +1,83 @@
+"""Fault-tolerance coverage for Deep Research progress events."""
+
+import pytest
+
+from deerflow.agents.deep_research.adapters.event_sink import PersistingEventSink
+
+
+class _UnavailableEventRepository:
+    async def append(self, **kwargs):  # noqa: ANN003
+        raise RuntimeError("database temporarily unavailable")
+
+
+@pytest.mark.asyncio
+async def test_event_persistence_failure_keeps_job_live_and_emits_unique_fallback_event():
+    published: list[dict] = []
+
+    async def publish(event: dict) -> None:
+        published.append(event)
+
+    sink = PersistingEventSink(
+        _UnavailableEventRepository(),
+        session_id="drs_1",
+        job_id="drj_1",
+        on_event=publish,
+    )
+
+    first = await sink.emit("phase_changed", phase="planning", payload={"to": "planning"})
+    second = await sink.emit("queries_planned", phase="collecting", payload={"queries": ["AI"]})
+
+    assert (first.seq, second.seq) == (-1, -2)
+    assert first.payload["eventPersistenceDegraded"] is True
+    assert published[0]["liveId"] == "event-persistence-fallback:drj_1:1"
+    assert published[1]["liveId"] == "event-persistence-fallback:drj_1:2"
+
+
+@pytest.mark.asyncio
+async def test_live_publish_failure_does_not_fail_event_delivery():
+    class _EventRepository:
+        async def append(self, **kwargs):  # noqa: ANN003
+            return {
+                "seq": 1,
+                "event_type": kwargs["event_type"],
+                "phase": kwargs["phase"],
+                "payload": kwargs["payload"],
+            }
+
+    async def broken_publisher(event: dict) -> None:  # noqa: ARG001
+        raise RuntimeError("browser disconnected")
+
+    sink = PersistingEventSink(
+        _EventRepository(),
+        session_id="drs_1",
+        job_id="drj_1",
+        on_event=broken_publisher,
+    )
+
+    event = await sink.emit("phase_changed", phase="planning")
+
+    assert event.seq == 1
+
+
+@pytest.mark.asyncio
+async def test_document_rewrite_milestone_is_a_supported_durable_event():
+    """Guard the artifact-rewrite job path against EventType regressions."""
+
+    class _EventRepository:
+        async def append(self, **kwargs):  # noqa: ANN003
+            return {
+                "seq": 1,
+                "event_type": kwargs["event_type"],
+                "phase": kwargs["phase"],
+                "payload": kwargs["payload"],
+            }
+
+    sink = PersistingEventSink(_EventRepository(), session_id="drs_1", job_id="drj_1")
+
+    event = await sink.emit(
+        "document_rewrite",
+        phase="planning",
+        payload={"type": "requirements_started"},
+    )
+
+    assert event.type == "document_rewrite"

offline-backend-20260512/backend/tests/test_deep_research_executor_reliability.py

@@ -0,0 +1,66 @@
+"""Failure-ordering regression coverage for final Deep Research report writes."""
+
+import pytest
+
+from app.gateway import deep_research_job_executor as executor_module
+from app.gateway.deep_research_job_executor import DeepResearchJobExecutor
+from deerflow.agents.deep_research.types import DeepResearchResult
+
+
+class _JobStore:
+    def __init__(self, session_store) -> None:
+        self._session_store = session_store
+        self.progress_calls: list[dict] = []
+
+    async def update_progress(self, job_id: str, **fields):  # noqa: ARG002
+        assert self._session_store.report_updates, "report must be persisted before job completion"
+        self.progress_calls.append(fields)
+        return {"id": job_id, **fields}
+
+
+class _SessionStore:
+    def __init__(self, *, available: bool = True) -> None:
+        self.available = available
+        self.report_updates: list[dict] = []
+
+    async def update(self, session_id: str, **fields):  # noqa: ARG002
+        if not self.available:
+            raise RuntimeError("temporary session database failure")
+        self.report_updates.append(fields)
+        return {"id": session_id, **fields}
+
+
+@pytest.mark.asyncio
+async def test_report_is_persisted_before_job_is_marked_completed():
+    sessions = _SessionStore()
+    jobs = _JobStore(sessions)
+    runner = DeepResearchJobExecutor(jobs, sessions, event_store=None)
+
+    await runner._mark_completed(
+        "drj_1",
+        "drs_1",
+        "worker_1",
+        DeepResearchResult(report_markdown="# 报告", source_ids=["src_1"]),
+    )
+
+    assert sessions.report_updates[0]["report_markdown"] == "# 报告"
+    assert jobs.progress_calls[0]["status"] == "completed"
+
+
+@pytest.mark.asyncio
+async def test_report_projection_failure_never_marks_job_completed(monkeypatch):
+    monkeypatch.setattr(executor_module, "_REPORT_PROJECTION_RETRY_ATTEMPTS", 2)
+    monkeypatch.setattr(executor_module, "_REPORT_PROJECTION_RETRY_DELAY_SECONDS", 0)
+    sessions = _SessionStore(available=False)
+    jobs = _JobStore(sessions)
+    runner = DeepResearchJobExecutor(jobs, sessions, event_store=None)
+
+    with pytest.raises(RuntimeError, match="研究报告未能保存"):
+        await runner._mark_completed(
+            "drj_1",
+            "drs_1",
+            "worker_1",
+            DeepResearchResult(report_markdown="# 报告"),
+        )
+
+    assert jobs.progress_calls == []

offline-backend-20260512/backend/tests/test_deep_research_persistence.py

@@ -0,0 +1,538 @@
+"""Persistence tests for deep-research sessions, jobs, events, sources, messages.
+
+Uses in-memory SQLite + aiosqlite, mirroring ``test_ai_writing_session_repo``.
+Focuses on the durable-job mechanics that are load-bearing for correctness:
+
+- Session CRUD + user isolation.
+- Job ``try_create_or_get_active`` idempotency (unique dedupe key).
+- Job lease claim / renew / release (atomic CAS).
+- Two-phase cancel (``request_cancel`` → ``finalize_cancel``).
+- Event seq monotonicity + ``list_after`` replay (SSE reconnect correctness).
+
+Pattern: a plain ``async def _make_repos()`` helper (not a fixture) — matches
+the project convention for async DB setup in pytest-asyncio strict mode.
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime, timedelta
+
+import pytest
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+
+
+async def _make_repos():
+    """Create an in-memory SQLite repo set with all deep-research tables."""
+    from deerflow.persistence.base import Base
+    from deerflow.persistence.deep_research_events.model import DeepResearchEventRow  # noqa: F401
+    from deerflow.persistence.deep_research_events.sql import DeepResearchEventRepository
+    from deerflow.persistence.deep_research_jobs.model import DeepResearchJobRow  # noqa: F401
+    from deerflow.persistence.deep_research_jobs.sql import DeepResearchJobRepository
+    from deerflow.persistence.deep_research_messages.model import DeepResearchMessageRow  # noqa: F401
+    from deerflow.persistence.deep_research_messages.sql import DeepResearchMessageRepository
+    from deerflow.persistence.deep_research_sessions.model import DeepResearchSessionRow  # noqa: F401
+    from deerflow.persistence.deep_research_sessions.sql import DeepResearchSessionRepository
+    from deerflow.persistence.deep_research_sources.model import DeepResearchSourceRow  # noqa: F401
+    from deerflow.persistence.deep_research_sources.sql import DeepResearchSourceRepository
+
+    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
+    async with engine.begin() as conn:
+        await conn.run_sync(Base.metadata.create_all)
+    sf = async_sessionmaker(engine, expire_on_commit=False)
+    return {
+        "sessions": DeepResearchSessionRepository(sf),
+        "jobs": DeepResearchJobRepository(sf),
+        "events": DeepResearchEventRepository(sf),
+        "sources": DeepResearchSourceRepository(sf),
+        "messages": DeepResearchMessageRepository(sf),
+        "_engine": engine,
+    }
+
+
+# ── sessions ────────────────────────────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_session_create_and_get():
+    repos = await _make_repos()
+    try:
+        created = await repos["sessions"].create(id="drs_1", user_id="u1", title="Test", query="AI research", mode="deep")
+        assert created["id"] == "drs_1"
+        assert created["status"] == "draft"
+
+        got = await repos["sessions"].get("drs_1", user_id="u1")
+        assert got is not None
+        assert got["query"] == "AI research"
+    finally:
+        await repos["_engine"].dispose()
+
+
+@pytest.mark.asyncio
+async def test_deep_research_writes_never_depend_on_post_commit_refresh(monkeypatch):
+    """Control-plane writes must survive a MySQL replica-lag readback."""
+    from sqlalchemy.ext.asyncio import AsyncSession
+
+    async def fail_refresh(self, instance, *args, **kwargs):  # noqa: ANN001, ARG001
+        raise AssertionError("Deep Research writes must not call session.refresh")
+
+    monkeypatch.setattr(AsyncSession, "refresh", fail_refresh)
+    repos = await _make_repos()
+    try:
+        session = await repos["sessions"].create(
+            id="drs_1", user_id="u1", title="Test", query="AI research"
+        )
+        assert session["id"] == "drs_1"
+        assert await repos["sessions"].update("drs_1", user_id="u1", status="running")
+
+        job, created = await repos["jobs"].try_create_or_get_active(
+            id="drj_1", session_id="drs_1", user_id="u1"
+        )
+        assert created is True
+        assert job["id"] == "drj_1"
+
+        source = await repos["sources"].upsert(
+            session_id="drs_1",
+            id="src_1",
+            user_id="u1",
+            title="A",
+            raw_content="source content",
+            source="web",
+            content_hash="hash_1",
+        )
+        assert source["id"] == "src_1"
+
+        message = await repos["messages"].append(
+            id="msg_1",
+            session_id="drs_1",
+            user_id="u1",
+            role="assistant",
+            content="已开始撰写",
+        )
+        assert message["id"] == "msg_1"
+    finally:
+        await repos["_engine"].dispose()
+
+
+@pytest.mark.asyncio
+async def test_session_user_isolation():
+    repos = await _make_repos()
+    try:
+        await repos["sessions"].create(id="drs_1", user_id="u1", query="q1")
+        assert await repos["sessions"].get("drs_1", user_id="u2") is None
+    finally:
+        await repos["_engine"].dispose()
+
+
+@pytest.mark.asyncio
+async def test_session_list_strips_heavy_columns():
+    repos = await _make_repos()
+    try:
+        await repos["sessions"].create(id="drs_1", user_id="u1", query="q", config_snapshot={"mode": "deep"})
+        items = await repos["sessions"].list_by_user(user_id="u1")
+        assert len(items) == 1
+        assert "report_markdown" not in items[0]
+        assert "config_snapshot" not in items[0]
+    finally:
+        await repos["_engine"].dispose()
+
+
+@pytest.mark.asyncio
+async def test_session_update_and_delete():
+    repos = await _make_repos()
+    try:
+        await repos["sessions"].create(id="drs_1", user_id="u1", query="q")
+        updated = await repos["sessions"].update("drs_1", user_id="u1", title="New Title")
+        assert updated["title"] == "New Title"
+        assert await repos["sessions"].delete("drs_1", user_id="u1") is True
+        assert await repos["sessions"].get("drs_1", user_id="u1") is None
+    finally:
+        await repos["_engine"].dispose()
+
+
+@pytest.mark.asyncio
+async def test_count_active_by_user():
+    repos = await _make_repos()
+    try:
+        await repos["sessions"].create(id="drs_1", user_id="u1", query="q", status="running")
+        await repos["sessions"].create(id="drs_2", user_id="u1", query="q", status="completed")
+        count = await repos["sessions"].count_active_by_user(user_id="u1")
+        assert count == 1
+    finally:
+        await repos["_engine"].dispose()
+
+
+# ── jobs: idempotent create ─────────────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_job_try_create_returns_created_true():
+    repos = await _make_repos()
+    try:
+        job, created = await repos["jobs"].try_create_or_get_active(id="drj_1", session_id="drs_1", user_id="u1", request_id="req_1")
+        assert created is True
+        assert job["status"] == "queued"
+        assert job["active_dedupe_key"] is not None
+    finally:
+        await repos["_engine"].dispose()
+
+
+@pytest.mark.asyncio
+async def test_job_create_never_reads_back_after_commit(monkeypatch):
+    """A committed job must not fail when a MySQL replica briefly lags."""
+    from sqlalchemy.ext.asyncio import AsyncSession
+
+    async def fail_refresh(self, instance, *args, **kwargs):  # noqa: ANN001, ARG001
+        raise AssertionError("DeepResearchJobRepository must not call session.refresh")
+
+    monkeypatch.setattr(AsyncSession, "refresh", fail_refresh)
+    repos = await _make_repos()
+    try:
+        job, created = await repos["jobs"].try_create_or_get_active(
+            id="drj_1",
+            session_id="drs_1",
+            user_id="u1",
+            request_id="req_1",
+        )
+        assert created is True
+        assert job["id"] == "drj_1"
+        assert job["status"] == "queued"
+    finally:
+        await repos["_engine"].dispose()
+
+
+@pytest.mark.asyncio
+async def test_job_try_create_idempotent_returns_existing():
+    repos = await _make_repos()
+    try:
+        job1, created1 = await repos["jobs"].try_create_or_get_active(id="drj_1", session_id="drs_1", user_id="u1", request_id="req_1")
+        job2, created2 = await repos["jobs"].try_create_or_get_active(id="drj_2", session_id="drs_1", user_id="u1", request_id="req_2")
+        assert created1 is True
+        assert created2 is False
+        assert job2["id"] == job1["id"]
+    finally:
+        await repos["_engine"].dispose()
+
+
+@pytest.mark.asyncio
+async def test_job_dedupe_released_on_terminal():
+    repos = await _make_repos()
+    try:
+        job, _ = await repos["jobs"].try_create_or_get_active(id="drj_1", session_id="drs_1", user_id="u1")
+        assert job["active_dedupe_key"] is not None
+        await repos["jobs"].update_progress("drj_1", lease_owner=None, user_id="u1", status="completed")
+        job2, created2 = await repos["jobs"].try_create_or_get_active(id="drj_2", session_id="drs_1", user_id="u1")
+        assert created2 is True
+        assert job2["id"] == "drj_2"
+    finally:
+        await repos["_engine"].dispose()
+
+
+# ── jobs: lease mechanics ───────────────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_job_claim_succeeds_for_queued():
+    repos = await _make_repos()
+    try:
+        await repos["jobs"].try_create_or_get_active(id="drj_1", session_id="drs_1", user_id="u1")
+        lease_until = datetime.now(UTC) + timedelta(seconds=60)
+        claimed = await repos["jobs"].claim_job("drj_1", lease_owner="w1", lease_until=lease_until)
+        assert claimed is not None
+        assert claimed["status"] == "running"
+        assert claimed["lease_owner"] == "w1"
+        assert claimed["attempt"] == 1
+    finally:
+        await repos["_engine"].dispose()
+
+
+@pytest.mark.asyncio
+async def test_dispatcher_claims_queued_job_and_starts_report_worker():
+    """The claimed snapshot must reach the executor before the job can run."""
+    from app.gateway.deep_research_job_dispatcher import DeepResearchJobDispatcher
+
+    class _Executor:
+        def __init__(self) -> None:
+            self.started: list[tuple[dict, str, str | None]] = []
+
+        def start_job(self, snapshot: dict, *, job_id: str, lease_owner: str | None) -> bool:
+            self.started.append((snapshot, job_id, lease_owner))
+            return True
+
+    repos = await _make_repos()
+    try:
+        await repos["jobs"].try_create_or_get_active(
+            id="drj_1",
+            session_id="drs_1",
+            user_id="u1",
+            input_snapshot={"query": "测试报告"},
+        )
+        runner = _Executor()
+        dispatcher = DeepResearchJobDispatcher(repos["jobs"], runner, worker_id="worker_1")
+
+        started = await dispatcher.dispatch_once()
+
+        assert started == 1
+        assert runner.started == [({"query": "测试报告"}, "drj_1", "worker_1")]
+    finally:
+        await repos["_engine"].dispose()
+
+
+@pytest.mark.asyncio
+async def test_job_claim_fails_if_already_claimed():
+    repos = await _make_repos()
+    try:
+        await repos["jobs"].try_create_or_get_active(id="drj_1", session_id="drs_1", user_id="u1")
+        lease_until = datetime.now(UTC) + timedelta(seconds=60)
+        first = await repos["jobs"].claim_job("drj_1", lease_owner="w1", lease_until=lease_until)
+        assert first is not None
+        second = await repos["jobs"].claim_job("drj_1", lease_owner="w2", lease_until=lease_until)
+        assert second is None
+    finally:
+        await repos["_engine"].dispose()
+
+
+@pytest.mark.asyncio
+async def test_job_renew_lease_only_by_owner():
+    repos = await _make_repos()
+    try:
+        await repos["jobs"].try_create_or_get_active(id="drj_1", session_id="drs_1", user_id="u1")
+        lease_until = datetime.now(UTC) + timedelta(seconds=60)
+        await repos["jobs"].claim_job("drj_1", lease_owner="w1", lease_until=lease_until)
+        assert await repos["jobs"].renew_lease("drj_1", lease_owner="w1", lease_until=lease_until) is True
+        assert await repos["jobs"].renew_lease("drj_1", lease_owner="w2", lease_until=lease_until) is False
+    finally:
+        await repos["_engine"].dispose()
+
+
+@pytest.mark.asyncio
+async def test_job_expired_lease_reclaimable():
+    repos = await _make_repos()
+    try:
+        await repos["jobs"].try_create_or_get_active(id="drj_1", session_id="drs_1", user_id="u1")
+        expired = datetime.now(UTC) - timedelta(seconds=10)
+        await repos["jobs"].claim_job("drj_1", lease_owner="w1", lease_until=expired)
+        now = datetime.now(UTC)
+        claimable = await repos["jobs"].list_claimable(now=now)
+        assert "drj_1" in claimable
+        reclaimed = await repos["jobs"].claim_job("drj_1", lease_owner="w2", lease_until=now + timedelta(seconds=60))
+        assert reclaimed is not None
+        assert reclaimed["lease_owner"] == "w2"
+    finally:
+        await repos["_engine"].dispose()
+
+
+# ── jobs: two-phase cancel ──────────────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_job_cancel_two_phase():
+    repos = await _make_repos()
+    try:
+        await repos["jobs"].try_create_or_get_active(id="drj_1", session_id="drs_1", user_id="u1")
+        lease_until = datetime.now(UTC) + timedelta(seconds=60)
+        await repos["jobs"].claim_job("drj_1", lease_owner="w1", lease_until=lease_until)
+
+        requested = await repos["jobs"].request_cancel("drj_1", user_id="u1")
+        assert requested is not None
+        assert requested["cancel_requested"] is True
+
+        finalized = await repos["jobs"].finalize_cancel("drj_1")
+        assert finalized is not None
+        assert finalized["status"] == "cancelled"
+        assert finalized["active_dedupe_key"] is None
+    finally:
+        await repos["_engine"].dispose()
+
+
+@pytest.mark.asyncio
+async def test_job_cancel_idempotent():
+    repos = await _make_repos()
+    try:
+        await repos["jobs"].try_create_or_get_active(id="drj_1", session_id="drs_1", user_id="u1")
+        first = await repos["jobs"].request_cancel("drj_1", user_id="u1")
+        assert first is not None
+        await repos["jobs"].finalize_cancel("drj_1")
+        second = await repos["jobs"].request_cancel("drj_1", user_id="u1")
+        assert second is None
+    finally:
+        await repos["_engine"].dispose()
+
+
+# ── jobs: progress update with lease gate ───────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_job_update_progress_lease_gated():
+    repos = await _make_repos()
+    try:
+        await repos["jobs"].try_create_or_get_active(id="drj_1", session_id="drs_1", user_id="u1")
+        lease_until = datetime.now(UTC) + timedelta(seconds=60)
+        await repos["jobs"].claim_job("drj_1", lease_owner="w1", lease_until=lease_until)
+
+        updated = await repos["jobs"].update_progress("drj_1", lease_owner="w1", phase="collecting", progress=50)
+        assert updated is not None
+        assert updated["phase"] == "collecting"
+        assert updated["progress"] == 50
+
+        rejected = await repos["jobs"].update_progress("drj_1", lease_owner="w2", phase="writing")
+        assert rejected is None
+    finally:
+        await repos["_engine"].dispose()
+
+
+# ── events: monotonic seq + replay ──────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_event_seq_monotonic():
+    repos = await _make_repos()
+    try:
+        repo = repos["events"]
+        e1 = await repo.append(session_id="drs_1", job_id="drj_1", event_type="phase_changed", payload={"to": "planning"})
+        e2 = await repo.append(session_id="drs_1", job_id="drj_1", event_type="source_added", payload={"id": "s1"})
+        e3 = await repo.append(session_id="drs_1", job_id="drj_1", event_type="source_added", payload={"id": "s2"})
+        assert e1["seq"] == 1
+        assert e2["seq"] == 2
+        assert e3["seq"] == 3
+        assert await repo.last_seq("drj_1") == 3
+    finally:
+        await repos["_engine"].dispose()
+
+
+@pytest.mark.asyncio
+async def test_event_append_never_reads_back_after_commit(monkeypatch):
+    """Event persistence must work when a post-commit read hits a lagging replica."""
+    from sqlalchemy.ext.asyncio import AsyncSession
+
+    async def fail_refresh(self, instance, *args, **kwargs):  # noqa: ANN001, ARG001
+        raise AssertionError("DeepResearchEventRepository must not call session.refresh")
+
+    monkeypatch.setattr(AsyncSession, "refresh", fail_refresh)
+    repos = await _make_repos()
+    try:
+        event = await repos["events"].append(
+            session_id="drs_1",
+            job_id="drj_1",
+            event_type="phase_changed",
+            payload={"to": "planning"},
+        )
+        assert event["id"] is not None
+        assert event["seq"] == 1
+    finally:
+        await repos["_engine"].dispose()
+
+
+@pytest.mark.asyncio
+async def test_event_list_after_replay():
+    repos = await _make_repos()
+    try:
+        repo = repos["events"]
+        for i in range(5):
+            await repo.append(session_id="drs_1", job_id="drj_1", event_type="heartbeat", payload={"i": i})
+        replayed = await repo.list_after("drj_1", after=2)
+        assert len(replayed) == 3
+        assert [e["seq"] for e in replayed] == [3, 4, 5]
+    finally:
+        await repos["_engine"].dispose()
+
+
+@pytest.mark.asyncio
+async def test_event_payload_round_trip():
+    repos = await _make_repos()
+    try:
+        repo = repos["events"]
+        await repo.append(
+            session_id="drs_1",
+            job_id="drj_1",
+            event_type="plan_created",
+            payload={"subtopics": ["AI", "ML"], "nested": {"a": 1}},
+        )
+        events = await repo.list_after("drj_1", after=0)
+        assert events[0]["payload"]["subtopics"] == ["AI", "ML"]
+        assert events[0]["payload"]["nested"]["a"] == 1
+    finally:
+        await repos["_engine"].dispose()
+
+
+@pytest.mark.asyncio
+async def test_event_seq_independent_per_job():
+    repos = await _make_repos()
+    try:
+        repo = repos["events"]
+        await repo.append(session_id="drs_1", job_id="drj_1", event_type="heartbeat")
+        await repo.append(session_id="drs_1", job_id="drj_1", event_type="heartbeat")
+        await repo.append(session_id="drs_1", job_id="drj_2", event_type="heartbeat")
+        assert await repo.last_seq("drj_1") == 2
+        assert await repo.last_seq("drj_2") == 1
+    finally:
+        await repos["_engine"].dispose()
+
+
+# ── sources ─────────────────────────────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_source_upsert_and_list():
+    repos = await _make_repos()
+    try:
+        await repos["sources"].upsert(
+            session_id="drs_1",
+            id="src_1",
+            user_id="u1",
+            title="Source A",
+            url="https://example.com/a",
+            raw_content="content A",
+            source="web",
+            source_type="web",
+            content_hash="hash_a",
+        )
+        items = await repos["sources"].list_by_session("drs_1", user_id="u1")
+        assert len(items) == 1
+        assert items[0]["title"] == "Source A"
+        assert "raw_content" not in items[0]
+    finally:
+        await repos["_engine"].dispose()
+
+
+@pytest.mark.asyncio
+async def test_source_set_selected():
+    repos = await _make_repos()
+    try:
+        await repos["sources"].upsert(
+            session_id="drs_1",
+            id="src_1",
+            user_id="u1",
+            title="A",
+            raw_content="c",
+            source="web",
+            content_hash="h",
+        )
+        updated = await repos["sources"].set_selected("src_1", selected=True, reason="relevant", citation_key="1", user_id="u1")
+        assert updated["selected"] is True
+        assert updated["citation_key"] == "1"
+    finally:
+        await repos["_engine"].dispose()
+
+
+# ── messages ────────────────────────────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_message_append_and_list():
+    repos = await _make_repos()
+    try:
+        await repos["messages"].append(id="msg_1", session_id="drs_1", user_id="u1", role="user", content="question?")
+        await repos["messages"].append(
+            id="msg_2",
+            session_id="drs_1",
+            user_id="u1",
+            role="assistant",
+            content="answer",
+            citation_source_ids=["src_1"],
+        )
+        msgs = await repos["messages"].list_by_session("drs_1", user_id="u1")
+        assert len(msgs) == 2
+        assert msgs[0]["role"] == "user"
+        assert msgs[1]["citation_source_ids"] == ["src_1"]
+    finally:
+        await repos["_engine"].dispose()

offline-backend-20260512/backend/tests/test_deep_research_start_job.py

@@ -0,0 +1,156 @@
+"""Regression coverage for idempotent Deep Research job submission."""
+
+from types import SimpleNamespace
+
+import pytest
+
+from app.gateway.routers import deep_research
+
+
+class _SessionStore:
+    def __init__(self) -> None:
+        self.updates: list[dict] = []
+
+    async def get(self, session_id: str, *, user_id: str):  # noqa: ARG002
+        return {
+            "id": session_id,
+            "query": "测试主题",
+            "title": "测试主题",
+            "config_snapshot": {"mode": "basic"},
+        }
+
+    async def count_active_by_user(self, *, user_id: str) -> int:  # noqa: ARG002
+        return 0
+
+    async def update(self, session_id: str, *, user_id: str, **fields):  # noqa: ARG002
+        self.updates.append(fields)
+        return {"id": session_id, **fields}
+
+
+class _JobStore:
+    async def get_active_for_session(self, session_id: str, *, user_id: str):  # noqa: ARG002
+        return None
+
+    async def try_create_or_get_active(self, **kwargs):  # noqa: ANN003
+        # Simulate a retry after an earlier request committed the job but did
+        # not finish responding to the browser.
+        return {"id": "drj_existing", "status": "queued"}, False
+
+
+class _Dispatcher:
+    def __init__(self) -> None:
+        self.nudges = 0
+
+    def nudge(self) -> None:
+        self.nudges += 1
+
+
+def _request(*, sessions, jobs, dispatcher):
+    return SimpleNamespace(
+        app=SimpleNamespace(
+            state=SimpleNamespace(
+                deep_research_session_store=sessions,
+                deep_research_job_store=jobs,
+                deep_research_dispatcher=dispatcher,
+            )
+        )
+    )
+
+
+@pytest.mark.asyncio
+async def test_retry_of_existing_queued_job_nudges_dispatcher(monkeypatch):
+    """An idempotent retry must recover a queued-but-not-yet-dispatched job."""
+
+    async def fake_require_user(request):  # noqa: ANN001
+        return "u1"
+
+    monkeypatch.setattr(deep_research, "_require_user", fake_require_user)
+    sessions = _SessionStore()
+    dispatcher = _Dispatcher()
+    request = _request(sessions=sessions, jobs=_JobStore(), dispatcher=dispatcher)
+
+    response = await deep_research.start_job(
+        "drs_1",
+        deep_research.StartJobRequest(entry="legacy"),
+        request,
+    )
+
+    assert response == {
+        "jobId": "drj_existing",
+        "sessionId": "drs_1",
+        "status": "queued",
+        "reused": True,
+    }
+    assert dispatcher.nudges == 1
+    assert sessions.updates == [{"status": "running", "active_job_id": "drj_existing"}]
+
+
+@pytest.mark.asyncio
+async def test_queued_job_still_starts_when_session_projection_temporarily_fails(monkeypatch):
+    """A committed job is more important than its non-authoritative UI projection."""
+
+    async def fake_require_user(request):  # noqa: ANN001
+        return "u1"
+
+    class _ProjectionFailureSessionStore(_SessionStore):
+        async def update(self, session_id: str, *, user_id: str, **fields):  # noqa: ARG002
+            raise RuntimeError("temporary database readback failure")
+
+    class _NewJobStore(_JobStore):
+        async def try_create_or_get_active(self, **kwargs):  # noqa: ANN003
+            return {"id": "drj_new", "status": "queued"}, True
+
+    monkeypatch.setattr(deep_research, "_require_user", fake_require_user)
+    dispatcher = _Dispatcher()
+    response = await deep_research.start_job(
+        "drs_1",
+        deep_research.StartJobRequest(entry="legacy"),
+        _request(
+            sessions=_ProjectionFailureSessionStore(),
+            jobs=_NewJobStore(),
+            dispatcher=dispatcher,
+        ),
+    )
+
+    assert response["jobId"] == "drj_new"
+    assert response["reused"] is False
+    assert dispatcher.nudges == 1
+
+
+@pytest.mark.asyncio
+async def test_stream_stays_open_when_durable_replay_has_a_transient_failure(monkeypatch):
+    """A stream DB blip must yield a recoverable warning, not an ASGI error group."""
+
+    async def fake_require_user(request):  # noqa: ANN001
+        return "u1"
+
+    class _StreamJobStore:
+        async def get(self, job_id: str, *, user_id: str):  # noqa: ARG002
+            return {"id": job_id, "session_id": "drs_1"}
+
+        async def get_unscoped(self, job_id: str):  # noqa: ARG002
+            return None
+
+    class _UnavailableEventStore:
+        async def list_after(self, *args, **kwargs):  # noqa: ANN002, ANN003
+            raise RuntimeError("temporary replica failure")
+
+    async def is_connected() -> bool:
+        return False
+
+    monkeypatch.setattr(deep_research, "_require_user", fake_require_user)
+    request = SimpleNamespace(
+        app=SimpleNamespace(
+            state=SimpleNamespace(
+                deep_research_job_store=_StreamJobStore(),
+                deep_research_event_store=_UnavailableEventStore(),
+            )
+        ),
+        is_disconnected=is_connected,
+    )
+
+    response = await deep_research.stream_job("drj_1", request)
+    chunk = await anext(response.body_iterator)
+    await response.body_iterator.aclose()
+
+    assert "durable_replay_temporarily_unavailable" in chunk

offline-backend-20260512/backend/tests/test_summarization_human_message.py

@@ -0,0 +1,30 @@
+"""Regression coverage for provider-safe transient context compression."""
+
+from langchain_core.messages import AIMessage, HumanMessage, SystemMessage, ToolMessage
+
+from deerflow.agents.middlewares.summarization_middleware import _ensure_latest_human_message
+
+
+def test_compression_restores_latest_human_turn_when_only_tool_tail_remains():
+    user = HumanMessage(content="请分析这份材料", id="user-1")
+    source = [
+        user,
+        AIMessage(content="", id="ai-1"),
+        ToolMessage(content="检索结果", tool_call_id="call-1", id="tool-1"),
+    ]
+    compressed = [SystemMessage(content="internal summary"), *source[1:]]
+
+    result = _ensure_latest_human_message(source, compressed)
+
+    assert result[0].type == "system"
+    assert result[1] is user
+    assert [message.type for message in result[2:]] == ["ai", "tool"]
+
+
+def test_compression_does_not_duplicate_human_turn_already_in_tail():
+    user = HumanMessage(content="继续", id="user-1")
+    compressed = [SystemMessage(content="internal summary"), user]
+
+    result = _ensure_latest_human_message([user], compressed)
+
+    assert result == compressed