fix(chat): cross-tab isolation, stream/history consistency, and reasoning replay

src/backend/api/routes/v1/artifacts.py

@@ -70,6 +70,16 @@ def sanitize_chat_preview(content: Optional[str], max_len: int = 200) -> str:
         return ""
 
     text = str(content)
+    # assistant \u6d88\u606f\u7684 content \u5728\u5e93\u91cc\u662f\u300c\u601d\u80031</think>\u601d\u80032</think>\u6b63\u6587\u300d\u7684\u539f\u59cb\u4e32\uff0c
+    # \u5361\u7247\u6458\u8981\u53ea\u8981\u6700\u7ec8\u6b63\u6587\u2014\u2014\u53d6\u6700\u540e\u4e00\u4e2a </think> \u4e4b\u540e\u7684\u90e8\u5206\uff0c\u5e76\u5265\u6389\u53ef\u80fd\u6b8b\u7559\u7684
+    # \u672a\u95ed\u5408 <think> \u8d77\u59cb\u6bb5\uff08\u622a\u65ad\u573a\u666f\uff09\uff0c\u907f\u514d\u6536\u85cf\u5361\u7247\u5c55\u793a\u601d\u8003\u8fc7\u7a0b/\u6807\u7b7e\uff08\u95ee\u98988\uff09\u3002
+    if "</think>" in text:
+        tail = text.rsplit("</think>", 1)[-1]
+        # \u5168\u662f\u601d\u8003\u6ca1\u6709\u6b63\u6587\u7684\u6781\u7aef\u60c5\u51b5\uff1a\u9000\u56de\u53bb\u6807\u7b7e\u540e\u7684\u539f\u6587\uff0c\u522b\u8ba9\u6458\u8981\u53d8\u6210\u7a7a\u767d
+        text = tail if tail.strip() else re.sub(r"</?think>", " ", text)
+    if "<think>" in text:
+        head = text.split("<think>", 1)[0]
+        text = head if head.strip() else re.sub(r"</?think>", " ", text)
     text = text.replace("\ufeff", "").replace("\u200b", "")
     text = re.sub(r"[\x00-\x08\x0b-\x1f\x7f]", "", text)
     text = re.sub(r"\s+", " ", text).strip()

src/backend/api/routes/v1/chats.py

@@ -2,6 +2,7 @@
 
 import asyncio
 import json
+import time
 import uuid
 from datetime import datetime
 from typing import Any, Dict, List, Optional
@@ -1306,6 +1307,17 @@ async def _stream_sse_response(
         pending_message_id = f"msg_{uuid.uuid4().hex[:16]}"
         metadata: dict = {}
         tool_calls_log: list = []
+        # 本轮回答总耗时起点 —— 持久化进 extra_data.duration_ms
+        _stream_started_monotonic = time.monotonic()
+        # 结构化 reasoning 思考增量：按到达顺序以 <think>…</think> 块交错并入
+        # full_response 落库（与内联思考模型一致），刷新后思考过程可回放。
+        _thinking_parts: list = []
+
+        def _flush_thinking() -> None:
+            nonlocal full_response
+            if _thinking_parts:
+                full_response += "<think>" + "".join(_thinking_parts) + "</think>"
+                _thinking_parts.clear()
         # Per-run workspace state — pin_to_workspace tool reads/writes this.
         _workspace_mod.init_state()
 
@@ -1316,18 +1328,28 @@ async def _stream_sse_response(
         ):
             chunk_type = chunk.get("type")
             if chunk_type == "thinking":
-                yield f"data: {json.dumps(build_thinking_event(chunk, chat_id), ensure_ascii=False)}\n\n"
+                _thinking_evt = build_thinking_event(chunk, chat_id)
+                if _thinking_evt.get("delta"):
+                    _thinking_parts.append(str(_thinking_evt["delta"]))
+                yield f"data: {json.dumps(_thinking_evt, ensure_ascii=False)}\n\n"
             elif chunk_type in {"ai_message", "content"}:
                 delta = chunk.get("delta", "")
                 if delta:
+                    _flush_thinking()
                     full_response += delta
                     yield f"data: {json.dumps({'type': 'content', 'event': 'ai_message', 'delta': delta, 'chat_id': chat_id}, ensure_ascii=False)}\n\n"
             elif chunk_type == "content_replace":
+                _thinking_parts.clear()
                 full_response = str(chunk.get("content") or "")
                 event = {**chunk, "chat_id": chat_id}
                 yield f"data: {json.dumps(event, ensure_ascii=False)}\n\n"
             elif chunk_type == "tool_call":
+                _flush_thinking()
                 _tc_evt = build_tool_call_event(chunk, chat_id, tool_calls_log)
+                # 记录该工具卡片出现时正文的累计长度：历史重建按此偏移把
+                # 「文本 ↔ 工具卡片」按流式原顺序交错（问题15：刷新后内容与实时不一致）。
+                for _tc in tool_calls_log:
+                    _tc.setdefault("content_offset", len(full_response))
                 yield f"data: {json.dumps(_tc_evt, ensure_ascii=False)}\n\n"
             elif chunk_type == "tool_result":
                 _tr_evt = build_tool_result_event(chunk, chat_id, tool_calls_log)
@@ -1366,6 +1388,7 @@ async def _stream_sse_response(
                 event = {**chunk, "chat_id": chat_id}
                 yield f"data: {json.dumps(event, ensure_ascii=False)}\n\n"
             elif chunk_type == "meta":
+                _flush_thinking()
                 pending_message_id = f"msg_{uuid.uuid4().hex[:16]}"
                 # Strict workspace gate: the agent's pin_to_workspace calls
                 # are the SOLE source of user-visible artifacts. We always
@@ -1402,6 +1425,7 @@ async def _stream_sse_response(
                     "warnings": metadata.get("warnings", []),
                     "citations": metadata.get("citations", []),
                     "workspace_files": metadata.get("workspace_files", []),
+                    "duration_ms": int((time.monotonic() - _stream_started_monotonic) * 1000),
                 }
                 if metadata.get("ontology_governance"):
                     _persist_extra["ontology_governance"] = metadata["ontology_governance"]

src/backend/api/routes/v1/kb.py

@@ -403,6 +403,7 @@ async def list_documents(
     kb_id: str = Path(..., description="Local or external knowledge collection ID"),
     page: int = Query(1, ge=1, description="Page number"),
     page_size: int = Query(20, ge=1, le=100, description="Items per page"),
+    keyword: Optional[str] = Query(None, description="按标题/文件名搜索（全库范围，再分页）"),
     user: UserContext = Depends(get_current_user),
     db: Session = Depends(get_db),
 ):
@@ -416,7 +417,7 @@ async def list_documents(
         from core.kb.external_provider import is_enabled, list_documents
 
         if is_enabled():
-            result = list_documents(kb_id, page=page, limit=page_size)
+            result = list_documents(kb_id, page=page, limit=page_size, keyword=keyword or "")
             return paginated_response(
                 items=result.get("items", []),
                 page=result.get("page", page),
@@ -432,7 +433,7 @@ async def list_documents(
             message="Access denied", reason="Only the KB space owner can list documents"
         )
 
-    documents, total = kb_repo.list_documents(kb_id, page, page_size)
+    documents, total = kb_repo.list_documents(kb_id, page, page_size, keyword=keyword)
     items = [
         {
             "id": d.document_id,

src/backend/core/db/repository/chat.py

@@ -15,6 +15,30 @@
 from sqlalchemy.orm import Session
 
 
+def _visible_message_text(content: str) -> str:
+    """assistant 消息 content 的可见正文（剥离 <think> 思考段）。
+
+    存储格式（多轮工具调用）：``思考1</think>可见文本<think>思考2</think>…最终正文``；
+    也可能只有裸 ``</think>`` 分隔（开标签被服务栈吞掉）。规则与前端
+    ``utils/segments.ts`` 的历史重建一致：每个 ``</think>`` 之前、上一个
+    ``<think>`` 之后的内容是思考；其余是可见正文。
+    """
+    if "</think>" not in content and "<think>" not in content:
+        return content
+    parts = content.split("</think>")
+    out: List[str] = []
+    for i, part in enumerate(parts):
+        if i == len(parts) - 1:
+            # 最后一段：若有未闭合的 <think>，其后是（被截断的）思考
+            out.append(part.split("<think>", 1)[0])
+        else:
+            idx = part.find("<think>")
+            if idx >= 0:
+                out.append(part[:idx])
+            # 无开标签 → 整段是思考，丢弃
+    return " ".join(x for x in out if x.strip()).strip()
+
+
 class ChatSessionRepository:
     """Repository for chat session operations."""
 
@@ -135,9 +159,13 @@ def search(
         }
 
         if scope == "all":
-            # Content-only matching chat_ids (exclude ones already matched by title)
+            # Content-only matching chat_ids (exclude ones already matched by title).
+            # SQL ilike 只做粗筛：assistant 消息的 content 里混有 <think> 思考文本
+            # （代码字段/英文/路径等），不应参与搜索（问题11）。粗筛命中后在
+            # Python 层剥掉思考、只对可见正文复核，摘要同样基于可见正文生成。
             content_id_set: set[str] = set()
-            content_rows = (
+            content_snippet_source: Dict[str, str] = {}
+            candidate_rows = (
                 self.db.query(ChatMessage.chat_id)
                 .join(ChatSession, ChatMessage.chat_id == ChatSession.chat_id)
                 .filter(
@@ -148,10 +176,31 @@ def search(
                 .distinct()
                 .all()
             )
-            content_id_set = {row[0] for row in content_rows} - title_id_set
+            q_lower = query.lower()
+            for (cid,) in candidate_rows:
+                if cid in title_id_set:
+                    continue
+                msgs = (
+                    self.db.query(ChatMessage.content)
+                    .filter(
+                        ChatMessage.chat_id == cid,
+                        ChatMessage.role.in_(["user", "assistant"]),
+                        ChatMessage.content.ilike(like_pattern),
+                    )
+                    .order_by(ChatMessage.created_at)
+                    .limit(20)
+                    .all()
+                )
+                for (raw,) in msgs:
+                    visible = _visible_message_text(raw or "")
+                    if q_lower in visible.lower():
+                        content_id_set.add(cid)
+                        content_snippet_source[cid] = visible
+                        break
             all_ids = title_id_set | content_id_set
         else:
             content_id_set = set()
+            content_snippet_source = {}
             all_ids = title_id_set
 
         total = len(all_ids)
@@ -192,19 +241,10 @@ def search(
             matched_snippet: Optional[str] = None
 
             if match_type == "content":
-                msg = (
-                    self.db.query(ChatMessage)
-                    .filter(
-                        ChatMessage.chat_id == s.chat_id,
-                        ChatMessage.role.in_(["user", "assistant"]),
-                        ChatMessage.content.ilike(like_pattern),
-                    )
-                    .order_by(ChatMessage.created_at)
-                    .first()
-                )
-                if msg and msg.content:
-                    # Center the snippet around the keyword
-                    content = msg.content.replace("\n", " ")
+                snippet_source = content_snippet_source.get(s.chat_id)
+                if snippet_source:
+                    # Center the snippet around the keyword（基于剥离思考后的可见正文）
+                    content = snippet_source.replace("\n", " ")
                     lower_content = content.lower()
                     idx = lower_content.find(query.lower())
                     if idx == -1:

src/backend/core/db/repository/kb.py

@@ -115,12 +115,17 @@ def get_document(self, document_id: str) -> Optional[KBDocument]:
         )
 
     def list_documents(
-        self, kb_id: str, page: int = 1, page_size: int = 20
+        self, kb_id: str, page: int = 1, page_size: int = 20, keyword: Optional[str] = None
     ) -> tuple[List[KBDocument], int]:
-        """List documents in a KB space."""
+        """List documents in a KB space, optionally filtered by title/filename keyword."""
         query = self.db.query(KBDocument).filter(
             KBDocument.kb_id == kb_id, KBDocument.deleted_at.is_(None)
         )
+        if keyword and keyword.strip():
+            like = f"%{keyword.strip()}%"
+            query = query.filter(
+                sa.or_(KBDocument.title.ilike(like), KBDocument.filename.ilike(like))
+            )
 
         total = query.count()
         documents = (

src/backend/orchestration/chat_run_executor.py

@@ -21,6 +21,7 @@
 import contextlib
 import json
 import os
+import time
 import uuid
 from datetime import datetime, timedelta, timezone
 from typing import Any, AsyncIterator, Awaitable, Callable, Dict, List, Literal, Optional
@@ -350,6 +351,8 @@ async def _run_workflow(
     from core.services.artifact_service import persist_artifacts as _persist_artifacts
 
     _update_run_status(run_id, status="running", started_at=_utcnow())
+    # 本轮回答总耗时起点 —— 持久化进 extra_data.duration_ms，历史加载后「用时」不再消失
+    _run_started_monotonic = time.monotonic()
 
     offset_counter = 0
 
@@ -364,6 +367,17 @@ async def _emit(event: Dict[str, Any]) -> None:
     metadata: Dict[str, Any] = {}
     tool_calls_log: list = []
     _workspace_mod.init_state()
+    # 结构化 reasoning 通道（deepseek 系 reasoning_content/reasoning）的思考增量。
+    # 过去这些只走 SSE thinking 事件、不落库 → 刷新后思考过程无法回放。现在按
+    # 到达顺序以 <think>…</think> 块交错并入 full_response（与内联思考模型的存储
+    # 格式一致），历史重建（buildHistorySegments）即可原位还原思考块。
+    _thinking_parts: List[str] = []
+
+    def _flush_thinking() -> None:
+        nonlocal full_response
+        if _thinking_parts:
+            full_response += "<think>" + "".join(_thinking_parts) + "</think>"
+            _thinking_parts.clear()
 
     try:
         # First frame: run_started — carries run_id / message_id; the frontend uses these to resume / cancel
@@ -441,11 +455,17 @@ async def _emit(event: Dict[str, Any]) -> None:
             chunk_type = chunk.get("type")
 
             if chunk_type == "thinking":
-                await _emit(build_thinking_event(chunk, chat_id))
+                _thinking_evt = build_thinking_event(chunk, chat_id)
+                # 只累积真实思考增量；进度提示（message）与 structured_reasoning
+                # 协议标记不落库
+                if _thinking_evt.get("delta"):
+                    _thinking_parts.append(str(_thinking_evt["delta"]))
+                await _emit(_thinking_evt)
 
             elif chunk_type in {"ai_message", "content"}:
                 delta = chunk.get("delta", "")
                 if delta:
+                    _flush_thinking()
                     full_response += delta
                     await _emit(
                         {
@@ -461,6 +481,7 @@ async def _emit(event: Dict[str, Any]) -> None:
                 # committee revised it, replace the visible/persisted answer
                 # atomically instead of appending a second full answer.
                 replacement = str(chunk.get("content") or "")
+                _thinking_parts.clear()
                 full_response = replacement
                 await _emit(
                     {
@@ -472,7 +493,13 @@ async def _emit(event: Dict[str, Any]) -> None:
                 )
 
             elif chunk_type == "tool_call":
-                await _emit(build_tool_call_event(chunk, chat_id, tool_calls_log))
+                _flush_thinking()
+                _tc_evt = build_tool_call_event(chunk, chat_id, tool_calls_log)
+                # 记录该工具卡片出现时正文的累计长度：历史重建按此偏移把
+                # 「文本 ↔ 工具卡片」按流式原顺序交错（问题15：刷新后内容与实时不一致）。
+                for _tc in tool_calls_log:
+                    _tc.setdefault("content_offset", len(full_response))
+                await _emit(_tc_evt)
 
             elif chunk_type == "tool_result":
                 await _emit(build_tool_result_event(chunk, chat_id, tool_calls_log))
@@ -599,6 +626,7 @@ async def _emit(event: Dict[str, Any]) -> None:
                 await _emit(ontology_event)
 
             elif chunk_type == "meta":
+                _flush_thinking()
                 # Strict workspace gate: pinned list is the sole source of
                 # user-visible artifacts. See chats.py:_stream_sse_response.
                 _ws_pinned = _workspace_mod.get_pinned()
@@ -632,6 +660,7 @@ async def _emit(event: Dict[str, Any]) -> None:
                     "citations": metadata.get("citations", []),
                     "message_id": message_id,
                     "workspace_files": _ws_files,
+                    "duration_ms": int((time.monotonic() - _run_started_monotonic) * 1000),
                 }
                 if metadata.get("ontology_governance"):
                     _persist_extra["ontology_governance"] = metadata["ontology_governance"]

src/backend/orchestration/streaming.py

@@ -133,6 +133,12 @@ def __init__(
         # Accumulated answer text (for dedup in the <think>-suppression scenario)
         self._raw_text = ""
         self._emitted_answer = ""
+        # 首轮正文累计（enable_thinking 下）：轮次结束时若整轮没有 </think>，
+        # 说明该模型不是"内联思考"形态（结构化 reasoning 通道、或本轮无思考）——
+        # 补发 structured_reasoning 标记，让前端把误缓冲成"思考"的正文流回正文区。
+        self._first_round_text = ""
+        self._round_index = 0
+        self._structured_marker_sent = False
         self._in_thinking = False
         self._reasoning_protocol_emitted = False
         # True between ModelCallStartEvent and ModelCallEndEvent. While a call is
@@ -384,6 +390,8 @@ async def _map_event(self, ev: Any) -> AsyncIterator[Tuple[str, Any]]:
             # directly, no accumulate+recompute needed (the frontend parses <think> itself),
             # avoiding an O(n) scan of the full answer on every delta.
             if self._enable_thinking:
+                if self._round_index == 0:
+                    self._first_round_text += delta
                 yield ("text_delta", delta)
                 return
             # Suppression state: <think> may span multiple deltas — accumulate, then strip out the answer after the closing tag.
@@ -492,6 +500,21 @@ async def _map_event(self, ev: Any) -> AsyncIterator[Tuple[str, Any]]:
                 "prompt_tokens": int(getattr(ev, "input_tokens", 0) or 0),
                 "completion_tokens": int(getattr(ev, "output_tokens", 0) or 0),
             })
+            # 首轮权威判定：思考模式下整轮正文没有出现 </think> → 该模型不内联思考
+            # （结构化 reasoning 通道，或本轮确实没思考）。补发协议标记，前端据此
+            # 把误当思考缓冲/展示的正文重归正文区（bug：无思考时正文进思考块）。
+            # 只看首轮——混合形态模型工具后省略 <think> 属既有启发式管辖，不在此误判。
+            if (
+                self._enable_thinking
+                and self._round_index == 0
+                and not self._structured_marker_sent
+                and self._first_round_text
+                and "</think>" not in self._first_round_text
+            ):
+                self._structured_marker_sent = True
+                yield ("reasoning_protocol", {"structured_reasoning": True})
+            self._round_index += 1
+            self._first_round_text = ""
             # New model call round → reset answer accumulation (the next text segment computes deltas from scratch)
             self._raw_text = ""
             self._emitted_answer = ""