From 7e179e72ec6f0b9103c21690ce07fbcf140c1feb Mon Sep 17 00:00:00 2001
From: dolphin <dolphin061888@gmail.com>
Date: Wed, 12 Aug 2026 19:32:49 +0800
Subject: [PATCH] =?UTF-8?q?fix(assistant):=20=E5=B7=A5=E5=85=B7=E5=8D=A1?=
 =?UTF-8?q?=E7=89=87=E4=B8=8D=E5=86=8D=E6=B0=B8=E4=B9=85=E8=BD=AC=E5=9C=88?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

三处叠加导致助手会话结束后仍有卡片在转：

1. 知识库工具内部通过 invoke 调检索器，等于又开了一次工具调用。
   聊天里每次检索多出一张卡，名字取自检索器而非知识库，显示
   「知识库已被删除」。改为直接调用，检索回到工具内部步骤。
2. 那张多出来的卡永远不闭合：检索器返回的是文档对象，结束帧
   序列化直接抛异常，帧发不出去也不入库。改为对任何工具返回值
   都能序列化，并容忍结束回调拿不到工具名。
3. 会话结束时前端不兜底。两个前端收到 close 只翻运行标记，
   漏掉的帧没人收尾。现在会把未闭合的卡片收掉，并标记为
   「工具调用未返回结果」——不给它一个没挣到的成功勾。

test/api + test/tool 45 项通过；两端 typecheck / lint / check-i18n 通过。
---
 src/backend/bisheng/api/v1/callback.py        | 544 +++++++++---------
 .../tool/domain/langchain/knowledge.py        |  47 +-
 .../test/api/test_assistant_runlog_frames.py  |  61 ++
 .../client/src/locales/en/translation.json    |   1 +
 .../client/src/locales/ja/translation.json    |   1 +
 .../src/locales/zh-Hans/translation.json      |   1 +
 .../src/pages/appChat/appUtils/skillMethod.ts |   2 +-
 .../appChat/components/MessageRunlog.tsx      |   3 +
 .../src/pages/appChat/useChatHelpers.ts       |  16 +-
 .../platform/public/locales/en-US/bs.json     |   3 +-
 .../platform/public/locales/ja/bs.json        |   3 +-
 .../platform/public/locales/zh-Hans/bs.json   |   3 +-
 .../bs-comp/chatComponent/ChatInput.tsx       |   3 +-
 .../bs-comp/chatComponent/RunLog.tsx          |   3 +
 .../bs-comp/chatComponent/messageStore.ts     |  19 +
 .../src/test/assistantRunLogClosing.test.ts   |  58 ++
 16 files changed, 458 insertions(+), 310 deletions(-)
 create mode 100644 src/backend/test/api/test_assistant_runlog_frames.py
 create mode 100644 src/frontend/platform/src/test/assistantRunLogClosing.test.ts

diff --git a/src/backend/bisheng/api/v1/callback.py b/src/backend/bisheng/api/v1/callback.py
index 265e5588e1..77c0631eca 100644
--- a/src/backend/bisheng/api/v1/callback.py
+++ b/src/backend/bisheng/api/v1/callback.py
@@ -2,7 +2,7 @@
 import copy
 import json
 from queue import Queue
-from typing import Any, Dict, List, Union
+from typing import Any, Union
 
 from fastapi import WebSocket
 from langchain_classic.callbacks.base import AsyncCallbackHandler, BaseCallbackHandler
@@ -11,24 +11,19 @@
 from langchain_classic.schema.document import Document
 from langchain_classic.schema.messages import BaseMessage
 from langchain_core.messages import ToolMessage
+from loguru import logger
 
 from bisheng.api.v1.schemas import ChatResponse
 from bisheng.database.models.message import ChatMessage as ChatMessageModel
 from bisheng.database.models.message import ChatMessageDao
 from bisheng.llm.domain.utils import extract_reasoning_content
-from loguru import logger
 
 
 # https://github.com/hwchase17/chat-langchain/blob/master/callback.py
 class AsyncStreamingLLMCallbackHandler(AsyncCallbackHandler):
     """Callback handler for streaming LLM responses."""
 
-    def __init__(self,
-                 websocket: WebSocket,
-                 flow_id: str,
-                 chat_id: str,
-                 user_id: int = None,
-                 **kwargs: Any):
+    def __init__(self, websocket: WebSocket, flow_id: str, chat_id: str, user_id: int = None, **kwargs: Any):
         self.websocket = websocket
         self.flow_id = flow_id
         self.chat_id = chat_id
@@ -44,94 +39,90 @@ def __init__(self,
         # }
 
         # Queue for Streaming Output
-        self.stream_queue: Queue = kwargs.get('stream_queue')
+        self.stream_queue: Queue = kwargs.get("stream_queue")
 
     async def on_llm_new_token(self, token: str, **kwargs: Any) -> None:
-        chunk = kwargs.get('chunk')
+        chunk = kwargs.get("chunk")
         # azureOccasionally returns aNone
         if token is None and chunk is None:
             return
         reasoning_content = extract_reasoning_content(chunk)
         if token is None:
-            token = ''
-        resp = ChatResponse(message={
-            'content': token,
-            'reasoning_content': reasoning_content
-        },
-            type='stream',
+            token = ""
+        resp = ChatResponse(
+            message={"content": token, "reasoning_content": reasoning_content},
+            type="stream",
             flow_id=self.flow_id,
-            chat_id=self.chat_id)
+            chat_id=self.chat_id,
+        )
         # Streaming output is placed in a queue to facilitate recording of content to a database after interrupting the streaming output
         await self.websocket.send_json(resp.dict())
         if self.stream_queue:
             if reasoning_content:
-                self.stream_queue.put({'type': 'reasoning', 'content': reasoning_content})
+                self.stream_queue.put({"type": "reasoning", "content": reasoning_content})
             if token:
-                self.stream_queue.put({'type': 'answer', 'content': token})
+                self.stream_queue.put({"type": "answer", "content": token})
 
-    async def on_llm_start(self, serialized: Dict[str, Any], prompts: List[str],
-                           **kwargs: Any) -> Any:
+    async def on_llm_start(self, serialized: dict[str, Any], prompts: list[str], **kwargs: Any) -> Any:
         """Run when LLM starts running."""
-        logger.debug(f'llm_start prompts={prompts}')
+        logger.debug(f"llm_start prompts={prompts}")
 
     async def on_llm_end(self, response: LLMResult, **kwargs: Any) -> Any:
         """Run when LLM ends running."""
-        logger.debug(f'llm_end response={response}')
+        logger.debug(f"llm_end response={response}")
 
     async def on_llm_error(self, error: Union[Exception, KeyboardInterrupt], **kwargs: Any) -> Any:
         """Run when LLM errors."""
-        logger.debug(f'on_llm_error error={error} kwargs={kwargs}')
+        logger.debug(f"on_llm_error error={error} kwargs={kwargs}")
 
-    async def on_chain_start(self, serialized: Dict[str, Any], inputs: Dict[str, Any],
-                             **kwargs: Any) -> Any:
+    async def on_chain_start(self, serialized: dict[str, Any], inputs: dict[str, Any], **kwargs: Any) -> Any:
         """Run when chain starts running."""
-        logger.debug(f'on_chain_start inputs={inputs} kwargs={kwargs}')
-        logger.info('k=s act=on_chain_start flow_id={} input_dict={}', self.flow_id, inputs)
+        logger.debug(f"on_chain_start inputs={inputs} kwargs={kwargs}")
+        logger.info("k=s act=on_chain_start flow_id={} input_dict={}", self.flow_id, inputs)
 
-    async def on_chain_end(self, outputs: Dict[str, Any], **kwargs: Any) -> Any:
+    async def on_chain_end(self, outputs: dict[str, Any], **kwargs: Any) -> Any:
         """Run when chain ends running."""
-        logger.debug(f'on_chain_end outputs={outputs} kwargs={kwargs}')
+        logger.debug(f"on_chain_end outputs={outputs} kwargs={kwargs}")
         tmp_output = copy.deepcopy(outputs)
         if isinstance(tmp_output, dict):
-            tmp_output.pop('source_documents', '')
-        logger.info('k=s act=on_chain_end flow_id={} output_dict={}', self.flow_id, tmp_output)
+            tmp_output.pop("source_documents", "")
+        logger.info("k=s act=on_chain_end flow_id={} output_dict={}", self.flow_id, tmp_output)
 
-    async def on_chain_error(self, error: Union[Exception, KeyboardInterrupt],
-                             **kwargs: Any) -> Any:
+    async def on_chain_error(self, error: Union[Exception, KeyboardInterrupt], **kwargs: Any) -> Any:
         """Run when chain errors."""
-        logger.debug(f'on_chain_error error={error} kwargs={kwargs}')
+        logger.debug(f"on_chain_error error={error} kwargs={kwargs}")
 
-    async def on_tool_start(self, serialized: Dict[str, Any], input_str: str,
-                            **kwargs: Any) -> Any:
+    async def on_tool_start(self, serialized: dict[str, Any], input_str: str, **kwargs: Any) -> Any:
         """Run when tool starts running."""
-        logger.debug(
-            f'on_tool_start  serialized={serialized} input_str={input_str} kwargs={kwargs}')
-        logger.info('k=s act=on_tool_start flow_id={} tool_name={} input_str={}', self.flow_id,
-                    serialized.get('name'), input_str)
-
-        resp = ChatResponse(type='stream',
-                            intermediate_steps=f'Tool input: {input_str}',
-                            flow_id=self.flow_id,
-                            chat_id=self.chat_id)
+        logger.debug(f"on_tool_start  serialized={serialized} input_str={input_str} kwargs={kwargs}")
+        logger.info(
+            "k=s act=on_tool_start flow_id={} tool_name={} input_str={}",
+            self.flow_id,
+            serialized.get("name"),
+            input_str,
+        )
+
+        resp = ChatResponse(
+            type="stream", intermediate_steps=f"Tool input: {input_str}", flow_id=self.flow_id, chat_id=self.chat_id
+        )
         await self.websocket.send_json(resp.dict())
 
     async def on_tool_end(self, output: str, **kwargs: Any) -> Any:
         """Run when tool ends running."""
-        logger.debug(f'on_tool_end  output={output} kwargs={kwargs}')
+        logger.debug(f"on_tool_end  output={output} kwargs={kwargs}")
         logger.info("k=s act=on_tool_end flow_id={} output='{}'", self.flow_id, output)
-        observation_prefix = kwargs.get('observation_prefix', 'Tool output: ')
-        # from langchain_classic.docstore.document import Document # noqa
+        observation_prefix = kwargs.get("observation_prefix", "Tool output: ")
+        # from langchain_classic.docstore.document import Document
         # result = eval(output).get('result')
-        result = output if isinstance(output, str) else getattr(output, 'content', output)
+        result = output if isinstance(output, str) else getattr(output, "content", output)
 
         # Create a formatted message.
-        intermediate_steps = f'{observation_prefix}{result[:100]}'
+        intermediate_steps = f"{observation_prefix}{result[:100]}"
 
         # Create a ChatResponse instance.
-        resp = ChatResponse(type='stream',
-                            intermediate_steps=intermediate_steps,
-                            flow_id=self.flow_id,
-                            chat_id=self.chat_id)
+        resp = ChatResponse(
+            type="stream", intermediate_steps=intermediate_steps, flow_id=self.flow_id, chat_id=self.chat_id
+        )
 
         try:
             # This is to emulate the stream of tokens
@@ -139,147 +130,126 @@ async def on_tool_end(self, output: str, **kwargs: Any) -> Any:
         except Exception as e:
             logger.error(e)
 
-    async def on_tool_error(self, error: Union[Exception, KeyboardInterrupt],
-                            **kwargs: Any) -> Any:
+    async def on_tool_error(self, error: Union[Exception, KeyboardInterrupt], **kwargs: Any) -> Any:
         """Run when tool errors."""
-        logger.debug(f'on_tool_error error={error} kwargs={kwargs}')
+        logger.debug(f"on_tool_error error={error} kwargs={kwargs}")
 
     async def on_text(self, text: str, **kwargs: Any) -> Any:
         """Run on arbitrary text."""
         # This runs when first sending the prompt
         # to the LLM, adding it will send the final prompt
         # to the frontend
-        logger.debug(f'on_text text={text} kwargs={kwargs}')
-        if 'Prompt after formatting:' in text:
+        logger.debug(f"on_text text={text} kwargs={kwargs}")
+        if "Prompt after formatting:" in text:
             prompt_str = text[24:]
             logger.info(
                 "k=s act=on_text prompt='{}'",
                 prompt_str,
             )
-        sender = kwargs.get('sender')
-        receiver = kwargs.get('receiver')
-        if kwargs.get('sender'):
-            log = ChatResponse(message=text,
-                               type='end',
-                               sender=sender,
-                               receiver=receiver,
-                               flow_id=self.flow_id,
-                               chat_id=self.chat_id)
-            start = ChatResponse(type='start',
-                                 sender=sender,
-                                 receiver=receiver,
-                                 flow_id=self.flow_id,
-                                 chat_id=self.chat_id)
-
-            if receiver and receiver.get('is_self'):
+        sender = kwargs.get("sender")
+        receiver = kwargs.get("receiver")
+        if kwargs.get("sender"):
+            log = ChatResponse(
+                message=text, type="end", sender=sender, receiver=receiver, flow_id=self.flow_id, chat_id=self.chat_id
+            )
+            start = ChatResponse(
+                type="start", sender=sender, receiver=receiver, flow_id=self.flow_id, chat_id=self.chat_id
+            )
+
+            if receiver and receiver.get("is_self"):
                 await self.websocket.send_json(log.dict())
             else:
                 await self.websocket.send_json(log.dict())
                 await self.websocket.send_json(start.dict())
-        elif 'category' in kwargs:
-            if 'autogen' == kwargs['category']:
-                log = ChatResponse(message=text,
-                                   type='stream',
-                                   flow_id=self.flow_id,
-                                   chat_id=self.chat_id)
+        elif "category" in kwargs:
+            if "autogen" == kwargs["category"]:
+                log = ChatResponse(message=text, type="stream", flow_id=self.flow_id, chat_id=self.chat_id)
                 await self.websocket.send_json(log.dict())
-                if kwargs.get('type'):
+                if kwargs.get("type"):
                     # Under compatibility
-                    start = ChatResponse(type='start',
-                                         category=kwargs.get('type'),
-                                         flow_id=self.flow_id,
-                                         chat_id=self.chat_id)
-                    end = ChatResponse(type='end',
-                                       intermediate_steps=text,
-                                       category=kwargs.get('type'),
-                                       flow_id=self.flow_id,
-                                       chat_id=self.chat_id)
+                    start = ChatResponse(
+                        type="start", category=kwargs.get("type"), flow_id=self.flow_id, chat_id=self.chat_id
+                    )
+                    end = ChatResponse(
+                        type="end",
+                        intermediate_steps=text,
+                        category=kwargs.get("type"),
+                        flow_id=self.flow_id,
+                        chat_id=self.chat_id,
+                    )
                     await self.websocket.send_json(start.dict())
                     await self.websocket.send_json(end.dict())
             else:
-                log = ChatResponse(message=text,
-                                   intermediate_steps=kwargs['log'],
-                                   type=kwargs['type'],
-                                   category=kwargs['category'],
-                                   flow_id=self.flow_id,
-                                   chat_id=self.chat_id)
+                log = ChatResponse(
+                    message=text,
+                    intermediate_steps=kwargs["log"],
+                    type=kwargs["type"],
+                    category=kwargs["category"],
+                    flow_id=self.flow_id,
+                    chat_id=self.chat_id,
+                )
                 await self.websocket.send_json(log.dict())
 
     async def on_agent_action(self, action: AgentAction, **kwargs: Any):
-        logger.debug(f'on_agent_action action={action} kwargs={kwargs}')
-        logger.info('k=s act=on_agent_action {}', action)
-        log = f'\nThought: {action.log}'
+        logger.debug(f"on_agent_action action={action} kwargs={kwargs}")
+        logger.info("k=s act=on_agent_action {}", action)
+        log = f"\nThought: {action.log}"
         # if there are line breaks, split them and send them
         # as separate messages
-        log = log.replace('\n', '\n\n')
-        resp = ChatResponse(type='stream',
-                            intermediate_steps=log,
-                            flow_id=self.flow_id,
-                            chat_id=self.chat_id)
+        log = log.replace("\n", "\n\n")
+        resp = ChatResponse(type="stream", intermediate_steps=log, flow_id=self.flow_id, chat_id=self.chat_id)
         await self.websocket.send_json(resp.dict())
 
     async def on_agent_finish(self, finish: AgentFinish, **kwargs: Any) -> Any:
         """Run on agent end."""
-        logger.debug(f'on_agent_finish finish={finish} kwargs={kwargs}')
-        logger.info('k=s act=on_agent_finish {}', finish)
-        resp = ChatResponse(flow_id=self.flow_id,
-                            chat_id=self.chat_id,
-                            type='stream',
-                            intermediate_steps=finish.log)
+        logger.debug(f"on_agent_finish finish={finish} kwargs={kwargs}")
+        logger.info("k=s act=on_agent_finish {}", finish)
+        resp = ChatResponse(flow_id=self.flow_id, chat_id=self.chat_id, type="stream", intermediate_steps=finish.log)
         await self.websocket.send_json(resp.dict())
 
-    async def on_retriever_start(self, serialized: Dict[str, Any], query: str,
-                                 **kwargs: Any) -> Any:
+    async def on_retriever_start(self, serialized: dict[str, Any], query: str, **kwargs: Any) -> Any:
         """Run when retriever start running."""
-        logger.debug(f'on_retriever_start serialized={serialized} query={query} kwargs={kwargs}')
-        logger.info('k=s act=on_retriever_start flow_id={} query={} meta={}', self.flow_id, query,
-                    serialized.get('repr'))
+        logger.debug(f"on_retriever_start serialized={serialized} query={query} kwargs={kwargs}")
+        logger.info(
+            "k=s act=on_retriever_start flow_id={} query={} meta={}", self.flow_id, query, serialized.get("repr")
+        )
 
-    async def on_retriever_end(self, result: List[Document], **kwargs: Any) -> Any:
+    async def on_retriever_end(self, result: list[Document], **kwargs: Any) -> Any:
         """Run when retriever end running."""
         # todo Determine skill permissions
-        logger.debug(f'on_retriever_end result={result} kwargs={kwargs}')
+        logger.debug(f"on_retriever_end result={result} kwargs={kwargs}")
         if result:
             tmp_result = copy.deepcopy(result)
-            [doc.metadata.pop('bbox', '') for doc in tmp_result]
-            logger.info('k=s act=on_retriever_end flow_id={} result_without_bbox={}', self.flow_id,
-                        tmp_result)
+            [doc.metadata.pop("bbox", "") for doc in tmp_result]
+            logger.info("k=s act=on_retriever_end flow_id={} result_without_bbox={}", self.flow_id, tmp_result)
 
-    async def on_chat_model_start(self, serialized: Dict[str, Any],
-                                  messages: List[List[BaseMessage]], **kwargs: Any) -> Any:
+    async def on_chat_model_start(
+        self, serialized: dict[str, Any], messages: list[list[BaseMessage]], **kwargs: Any
+    ) -> Any:
         # """Run when retriever end running."""
         # content = messages[0][0] if isinstance(messages[0][0], str) else messages[0][0].get('content')
         # stream = ChatResponse(message=f'{content}', type='stream')
         # await self.websocket.send_json(stream.dict())
-        logger.debug(
-            f'on_chat_model_start serialized={serialized} messages={messages} kwargs={kwargs}')
-        logger.info('k=s act=on_chat_model_start messages={}', messages)
+        logger.debug(f"on_chat_model_start serialized={serialized} messages={messages} kwargs={kwargs}")
+        logger.info("k=s act=on_chat_model_start messages={}", messages)
 
 
 class StreamingLLMCallbackHandler(BaseCallbackHandler):
     """Callback handler for streaming LLM responses."""
 
-    def __init__(self,
-                 websocket: WebSocket,
-                 flow_id: str,
-                 chat_id: str,
-                 user_id: int = None,
-                 **kwargs: Any):
+    def __init__(self, websocket: WebSocket, flow_id: str, chat_id: str, user_id: int = None, **kwargs: Any):
         self.websocket = websocket
         self.flow_id = flow_id
         self.chat_id = chat_id
         self.user_id = user_id
 
-        self.stream_queue: Queue = kwargs.get('stream_queue')
+        self.stream_queue: Queue = kwargs.get("stream_queue")
 
     def on_llm_new_token(self, token: str, **kwargs: Any) -> None:
         # azureOccasionally returns aNone
         if token is None:
             return
-        resp = ChatResponse(message=token,
-                            type='stream',
-                            flow_id=self.flow_id,
-                            chat_id=self.chat_id)
+        resp = ChatResponse(message=token, type="stream", flow_id=self.flow_id, chat_id=self.chat_id)
         if self.websocket:
             loop = asyncio.get_event_loop()
             coroutine = self.websocket.send_json(resp.dict())
@@ -289,60 +259,56 @@ def on_llm_new_token(self, token: str, **kwargs: Any) -> None:
             self.stream_queue.put(token)
 
     def on_agent_action(self, action: AgentAction, **kwargs: Any) -> Any:
-        log = f'\nThought: {action.log}'
+        log = f"\nThought: {action.log}"
         # if there are line breaks, split them and send them
         # as separate messages
-        log = log.replace('\n', '\n\n')
-        resp = ChatResponse(type='stream',
-                            intermediate_steps=log,
-                            flow_id=self.flow_id,
-                            chat_id=self.chat_id)
+        log = log.replace("\n", "\n\n")
+        resp = ChatResponse(type="stream", intermediate_steps=log, flow_id=self.flow_id, chat_id=self.chat_id)
         if self.websocket:
             loop = asyncio.get_event_loop()
             coroutine = self.websocket.send_json(resp.dict())
             asyncio.run_coroutine_threadsafe(coroutine, loop)
-        logger.info('k=s act=on_agent_action {}', action)
+        logger.info("k=s act=on_agent_action {}", action)
 
     def on_agent_finish(self, finish: AgentFinish, **kwargs: Any) -> Any:
         """Run on agent end."""
-        resp = ChatResponse(type='stream',
-                            intermediate_steps=finish.log,
-                            flow_id=self.flow_id,
-                            chat_id=self.chat_id)
+        resp = ChatResponse(type="stream", intermediate_steps=finish.log, flow_id=self.flow_id, chat_id=self.chat_id)
         if self.websocket:
             loop = asyncio.get_event_loop()
             coroutine = self.websocket.send_json(resp.dict())
             asyncio.run_coroutine_threadsafe(coroutine, loop)
-        logger.info('k=s act=on_agent_finish {}', finish)
+        logger.info("k=s act=on_agent_finish {}", finish)
 
-    def on_tool_start(self, serialized: Dict[str, Any], input_str: str, **kwargs: Any) -> Any:
+    def on_tool_start(self, serialized: dict[str, Any], input_str: str, **kwargs: Any) -> Any:
         """Run when tool starts running."""
-        resp = ChatResponse(type='stream',
-                            intermediate_steps=f'Tool input: {input_str}',
-                            flow_id=self.flow_id,
-                            chat_id=self.chat_id)
+        resp = ChatResponse(
+            type="stream", intermediate_steps=f"Tool input: {input_str}", flow_id=self.flow_id, chat_id=self.chat_id
+        )
         if self.websocket:
             loop = asyncio.get_event_loop()
             coroutine = self.websocket.send_json(resp.dict())
             asyncio.run_coroutine_threadsafe(coroutine, loop)
-        logger.info('k=s act=on_tool_start flow_id={} tool_name={} input_str={}', self.flow_id,
-                    serialized.get('name'), input_str)
+        logger.info(
+            "k=s act=on_tool_start flow_id={} tool_name={} input_str={}",
+            self.flow_id,
+            serialized.get("name"),
+            input_str,
+        )
 
     def on_tool_end(self, output: str, **kwargs: Any) -> Any:
         """Run when tool ends running."""
-        observation_prefix = kwargs.get('observation_prefix', 'Tool output: ')
+        observation_prefix = kwargs.get("observation_prefix", "Tool output: ")
 
-        # from langchain_classic.docstore.document import Document # noqa
+        # from langchain_classic.docstore.document import Document
         # result = eval(output).get('result')
-        result = output if isinstance(output, str) else getattr(output, 'content', output)
+        result = output if isinstance(output, str) else getattr(output, "content", output)
         # Create a formatted message.
-        intermediate_steps = f'{observation_prefix}{result}'
+        intermediate_steps = f"{observation_prefix}{result}"
 
         # Create a ChatResponse instance.
-        resp = ChatResponse(type='stream',
-                            intermediate_steps=intermediate_steps,
-                            flow_id=self.flow_id,
-                            chat_id=self.chat_id)
+        resp = ChatResponse(
+            type="stream", intermediate_steps=intermediate_steps, flow_id=self.flow_id, chat_id=self.chat_id
+        )
 
         # Try to send the response, handle potential errors.
         try:
@@ -354,37 +320,35 @@ def on_tool_end(self, output: str, **kwargs: Any) -> Any:
             logger.error(e)
         logger.info("k=s act=on_tool_end flow_id={} output='{}'", self.flow_id, output)
 
-    def on_retriever_start(self, serialized: Dict[str, Any], query: str, **kwargs: Any) -> Any:
+    def on_retriever_start(self, serialized: dict[str, Any], query: str, **kwargs: Any) -> Any:
         """Run when retriever start running."""
-        logger.info('k=s act=on_retriever_start flow_id={} query={} meta={}', self.flow_id, query,
-                    serialized.get('repr'))
+        logger.info(
+            "k=s act=on_retriever_start flow_id={} query={} meta={}", self.flow_id, query, serialized.get("repr")
+        )
 
-    def on_retriever_end(self, result: List[Document], **kwargs: Any) -> Any:
+    def on_retriever_end(self, result: list[Document], **kwargs: Any) -> Any:
         """Run when retriever end running."""
         # todo Determine skill permissions
-        logger.debug(f'retriver_result result={result}')
+        logger.debug(f"retriver_result result={result}")
         if result:
             tmp_result = copy.deepcopy(result)
-            [doc.metadata.pop('bbox', '') for doc in tmp_result]
-            logger.info('k=s act=on_retriever_end flow_id={} result_without_bbox={}', self.flow_id,
-                        tmp_result)
+            [doc.metadata.pop("bbox", "") for doc in tmp_result]
+            logger.info("k=s act=on_retriever_end flow_id={} result_without_bbox={}", self.flow_id, tmp_result)
 
-    def on_chain_start(self, serialized: Dict[str, Any], inputs: Dict[str, Any],
-                       **kwargs: Any) -> Any:
+    def on_chain_start(self, serialized: dict[str, Any], inputs: dict[str, Any], **kwargs: Any) -> Any:
         """Run when chain starts running."""
-        logger.debug(f'on_chain_start inputs={inputs}')
-        logger.info('k=s act=on_chain_start flow_id={} input_dict={}', self.flow_id, inputs)
+        logger.debug(f"on_chain_start inputs={inputs}")
+        logger.info("k=s act=on_chain_start flow_id={} input_dict={}", self.flow_id, inputs)
 
-    def on_chain_end(self, outputs: Dict[str, Any], **kwargs: Any) -> Any:
+    def on_chain_end(self, outputs: dict[str, Any], **kwargs: Any) -> Any:
         """Run when chain ends running."""
-        logger.debug(f'on_chain_end outputs={outputs}')
+        logger.debug(f"on_chain_end outputs={outputs}")
         tmp_output = copy.deepcopy(outputs)
         if isinstance(tmp_output, dict):
-            tmp_output.pop('source_documents', '')
-        logger.info('k=s act=on_chain_end flow_id={} output_dict={}', self.flow_id, tmp_output)
+            tmp_output.pop("source_documents", "")
+        logger.info("k=s act=on_chain_end flow_id={} output_dict={}", self.flow_id, tmp_output)
 
-    def on_chat_model_start(self, serialized: Dict[str, Any], messages: List[List[BaseMessage]],
-                            **kwargs: Any) -> Any:
+    def on_chat_model_start(self, serialized: dict[str, Any], messages: list[list[BaseMessage]], **kwargs: Any) -> Any:
         """Run when retriever end running."""
         # sender = kwargs['sender']
         # receiver = kwargs['receiver']
@@ -396,12 +360,12 @@ def on_chat_model_start(self, serialized: Dict[str, Any], messages: List[List[Ba
         # coroutine3 = self.websocket.send_json(start.dict())
         # asyncio.run_coroutine_threadsafe(coroutine2, loop)
         # asyncio.run_coroutine_threadsafe(coroutine3, loop)
-        logger.debug(f'on_chat result={messages}')
-        logger.info('k=s act=on_chat_model_start messages={}', messages)
+        logger.debug(f"on_chat result={messages}")
+        logger.info("k=s act=on_chat_model_start messages={}", messages)
 
     def on_text(self, text: str, **kwargs) -> Any:
         logger.info(text)
-        if 'Prompt after formatting:' in text:
+        if "Prompt after formatting:" in text:
             prompt_str = text[24:]
             logger.info(
                 "k=s act=on_text prompt='{}'",
@@ -410,168 +374,180 @@ def on_text(self, text: str, **kwargs) -> Any:
 
 
 class AsyncGptsLLMCallbackHandler(AsyncStreamingLLMCallbackHandler):
-
-    async def on_tool_start(self, serialized: Dict[str, Any], input_str: str,
-                            **kwargs: Any) -> Any:
+    async def on_tool_start(self, serialized: dict[str, Any], input_str: str, **kwargs: Any) -> Any:
         """Run when tool starts running."""
-        logger.debug(
-            f'on_tool_start serialized={serialized} input_str={input_str} kwargs={kwargs}')
+        logger.debug(f"on_tool_start serialized={serialized} input_str={input_str} kwargs={kwargs}")
         pass
 
     async def on_tool_end(self, output: str, **kwargs: Any) -> Any:
         """Run when tool ends running."""
-        logger.debug(f'on_tool_end output={output} kwargs={kwargs}')
+        logger.debug(f"on_tool_end output={output} kwargs={kwargs}")
         pass
 
 
-class AsyncGptsDebugCallbackHandler(AsyncGptsLLMCallbackHandler):
+def _dump_run_log(payload: dict) -> str:
+    """Serialize a run-log frame so an exotic tool result cannot suppress it.
+
+    A tool is free to answer with anything — the knowledge retriever returns
+    ``Document`` objects. Plain ``json.dumps`` raised on those, the end frame was
+    never sent, and the chat card spun forever with no way to recover.
+    """
 
+    return json.dumps(payload, ensure_ascii=False, default=str)
+
+
+class AsyncGptsDebugCallbackHandler(AsyncGptsLLMCallbackHandler):
     @staticmethod
     def parse_tool_category(tool_name) -> (str, str):
         """
         will betool_nameResolve totool_categoryand the realtool_name
         """
-        tool_category = 'tool'
-        if tool_name.startswith('flow_'):
+        tool_category = "tool"
+        if not tool_name:
+            # `on_tool_end` reads the name out of kwargs, which is not always
+            # populated; without this the end frame would never be sent and the
+            # chat card would spin forever.
+            return "", tool_category
+        if tool_name.startswith("flow_"):
             # Description is a skill call
-            tool_category = 'flow'
-            tool_name = tool_name.replace('flow_', '')
-        elif tool_name.startswith('knowledge_'):
+            tool_category = "flow"
+            tool_name = tool_name.replace("flow_", "")
+        elif tool_name.startswith("knowledge_"):
             # Description is a knowledge base call
-            tool_category = 'knowledge'
-            tool_name = tool_name.replace('knowledge_', '')
+            tool_category = "knowledge"
+            tool_name = tool_name.replace("knowledge_", "")
         return tool_name, tool_category
 
-    async def on_chat_model_start(self, serialized: Dict[str, Any],
-                                  messages: List[List[BaseMessage]], **kwargs: Any) -> Any:
+    async def on_chat_model_start(
+        self, serialized: dict[str, Any], messages: list[list[BaseMessage]], **kwargs: Any
+    ) -> Any:
         # """Run when retriever end running."""
         # content = messages[0][0] if isinstance(messages[0][0], str) else messages[0][0].get('content')
         # stream = ChatResponse(message=f'{content}', type='stream')
         # await self.websocket.send_json(stream.dict())
-        logger.debug(
-            f'on_chat_model_start serialized={serialized} messages={messages} kwargs={kwargs}')
-        resp = ChatResponse(type='start',
-                            category='processing',
-                            flow_id=self.flow_id,
-                            chat_id=self.chat_id)
+        logger.debug(f"on_chat_model_start serialized={serialized} messages={messages} kwargs={kwargs}")
+        resp = ChatResponse(type="start", category="processing", flow_id=self.flow_id, chat_id=self.chat_id)
         await self.websocket.send_json(resp.dict())
 
     async def on_llm_end(self, response: LLMResult, **kwargs: Any) -> Any:
         """Run when LLM ends running."""
-        logger.debug(f'llm_end response={response}')
-        resp = ChatResponse(type='end',
-                            category='processing',
-                            flow_id=self.flow_id,
-                            chat_id=self.chat_id)
+        logger.debug(f"llm_end response={response}")
+        resp = ChatResponse(type="end", category="processing", flow_id=self.flow_id, chat_id=self.chat_id)
         await self.websocket.send_json(resp.dict())
 
     async def on_llm_error(self, error: Union[Exception, KeyboardInterrupt], **kwargs: Any) -> Any:
         """Run when LLM errors."""
-        logger.debug(f'on_llm_error error={error} kwargs={kwargs}')
-        resp = ChatResponse(type='end',
-                            category='processing',
-                            flow_id=self.flow_id,
-                            chat_id=self.chat_id)
+        logger.debug(f"on_llm_error error={error} kwargs={kwargs}")
+        resp = ChatResponse(type="end", category="processing", flow_id=self.flow_id, chat_id=self.chat_id)
         await self.websocket.send_json(resp.dict())
 
-    async def on_tool_start(self, serialized: Dict[str, Any], input_str: str,
-                            **kwargs: Any) -> Any:
+    async def on_tool_start(self, serialized: dict[str, Any], input_str: str, **kwargs: Any) -> Any:
         """Run when tool starts running."""
-        logger.debug(
-            f'on_tool_start serialized={serialized} input_str={input_str} kwargs={kwargs}')
+        logger.debug(f"on_tool_start serialized={serialized} input_str={input_str} kwargs={kwargs}")
 
         input_str = input_str
-        tool_name, tool_category = self.parse_tool_category(serialized['name'])
-        input_info = {'tool_key': tool_name, 'serialized': serialized, 'input_str': input_str}
-        self.tool_cache[kwargs.get('run_id').hex] = {
-            'input': input_info,
-            'category': tool_category,
-            'steps': f'Tool input: \n\n{input_str}\n\n',
+        tool_name, tool_category = self.parse_tool_category(serialized["name"])
+        input_info = {"tool_key": tool_name, "serialized": serialized, "input_str": input_str}
+        self.tool_cache[kwargs.get("run_id").hex] = {
+            "input": input_info,
+            "category": tool_category,
+            "steps": f"Tool input: \n\n{input_str}\n\n",
         }
-        resp = ChatResponse(type='start',
-                            category=tool_category,
-                            intermediate_steps=self.tool_cache[kwargs.get('run_id').hex]['steps'],
-                            message=json.dumps(input_info, ensure_ascii=False),
-                            flow_id=self.flow_id,
-                            chat_id=self.chat_id,
-                            extra=json.dumps({'run_id': kwargs.get('run_id').hex}))
+        resp = ChatResponse(
+            type="start",
+            category=tool_category,
+            intermediate_steps=self.tool_cache[kwargs.get("run_id").hex]["steps"],
+            message=json.dumps(input_info, ensure_ascii=False),
+            flow_id=self.flow_id,
+            chat_id=self.chat_id,
+            extra=json.dumps({"run_id": kwargs.get("run_id").hex}),
+        )
         await self.websocket.send_json(resp.dict())
 
     async def on_tool_end(self, output: ToolMessage, **kwargs: Any) -> Any:
         """Run when tool ends running."""
-        logger.debug(f'on_tool_end output={output} kwargs={kwargs}')
-        observation_prefix = kwargs.get('observation_prefix', 'Tool output: ')
+        logger.debug(f"on_tool_end output={output} kwargs={kwargs}")
+        observation_prefix = kwargs.get("observation_prefix", "Tool output: ")
 
-        result = output if isinstance(output, str) else getattr(output, 'content', output)
+        result = output if isinstance(output, str) else getattr(output, "content", output)
+        if not isinstance(result, str):
+            result = str(result)
         # Create a formatted message.
-        intermediate_steps = f'{observation_prefix}\n\n{result}'
-        tool_name, tool_category = self.parse_tool_category(kwargs.get('name'))
+        intermediate_steps = f"{observation_prefix}\n\n{result}"
+        tool_name, tool_category = self.parse_tool_category(kwargs.get("name"))
 
         # Create a ChatResponse instance.
-        output_info = {'tool_key': tool_name, 'output': result}
-        resp = ChatResponse(type='end',
-                            category=tool_category,
-                            intermediate_steps=intermediate_steps,
-                            message=json.dumps(output_info, ensure_ascii=False),
-                            flow_id=self.flow_id,
-                            chat_id=self.chat_id,
-                            extra=json.dumps({'run_id': kwargs.get('run_id').hex}))
+        output_info = {"tool_key": tool_name, "output": result}
+        resp = ChatResponse(
+            type="end",
+            category=tool_category,
+            intermediate_steps=intermediate_steps,
+            message=_dump_run_log(output_info),
+            flow_id=self.flow_id,
+            chat_id=self.chat_id,
+            extra=json.dumps({"run_id": kwargs.get("run_id").hex}),
+        )
 
         await self.websocket.send_json(resp.dict())
         # FROMtool cacheGet ininputMessage
-        input_info = self.tool_cache.get(kwargs.get('run_id').hex)
+        input_info = self.tool_cache.get(kwargs.get("run_id").hex)
         if input_info:
             if not self.chat_id:
                 # Explain that it is a debugging interface and does not need to persist data
-                self.tool_cache.pop(kwargs.get('run_id').hex)
+                self.tool_cache.pop(kwargs.get("run_id").hex)
                 return
-            output_info.update(input_info['input'])
-            intermediate_steps = f'{input_info["steps"]}\n\n{intermediate_steps}'
+            output_info.update(input_info["input"])
+            intermediate_steps = f"{input_info['steps']}\n\n{intermediate_steps}"
             ChatMessageDao.insert_one(
-                ChatMessageModel(is_bot=1,
-                                 message=json.dumps(output_info),
-                                 intermediate_steps=intermediate_steps,
-                                 category=tool_category,
-                                 type='end',
-                                 flow_id=self.flow_id,
-                                 chat_id=self.chat_id,
-                                 user_id=self.user_id,
-                                 extra=json.dumps({'run_id': kwargs.get('run_id').hex})))
-            self.tool_cache.pop(kwargs.get('run_id').hex)
-
-    async def on_tool_error(self, error: Union[Exception, KeyboardInterrupt],
-                            **kwargs: Any) -> Any:
+                ChatMessageModel(
+                    is_bot=1,
+                    message=_dump_run_log(output_info),
+                    intermediate_steps=intermediate_steps,
+                    category=tool_category,
+                    type="end",
+                    flow_id=self.flow_id,
+                    chat_id=self.chat_id,
+                    user_id=self.user_id,
+                    extra=json.dumps({"run_id": kwargs.get("run_id").hex}),
+                )
+            )
+            self.tool_cache.pop(kwargs.get("run_id").hex)
+
+    async def on_tool_error(self, error: Union[Exception, KeyboardInterrupt], **kwargs: Any) -> Any:
         """Run when tool errors."""
-        logger.debug(f'on_tool_error error={error} kwargs={kwargs}')
-        input_info = self.tool_cache.get(kwargs.get('run_id').hex)
+        logger.debug(f"on_tool_error error={error} kwargs={kwargs}")
+        input_info = self.tool_cache.get(kwargs.get("run_id").hex)
         if input_info:
-            output_info = {'output': 'Error: ' + str(error)}
-            output_info.update(input_info['input'])
-            resp = ChatResponse(type='end',
-                                category=input_info['category'],
-                                intermediate_steps='\n\nTool output:\n\n  Error: ' + str(error),
-                                message=json.dumps(output_info, ensure_ascii=False),
-                                flow_id=self.flow_id,
-                                chat_id=self.chat_id,
-                                extra=json.dumps({'run_id': kwargs.get('run_id').hex}))
+            output_info = {"output": "Error: " + str(error)}
+            output_info.update(input_info["input"])
+            resp = ChatResponse(
+                type="end",
+                category=input_info["category"],
+                intermediate_steps="\n\nTool output:\n\n  Error: " + str(error),
+                message=_dump_run_log(output_info),
+                flow_id=self.flow_id,
+                chat_id=self.chat_id,
+                extra=json.dumps({"run_id": kwargs.get("run_id").hex}),
+            )
             await self.websocket.send_json(resp.dict())
 
             # Save tool call history
             if not self.chat_id:
                 # Explain that it is a debugging interface and does not need to persist data
-                self.tool_cache.pop(kwargs.get('run_id').hex)
+                self.tool_cache.pop(kwargs.get("run_id").hex)
                 return
-            tool_name, tool_category = self.parse_tool_category(kwargs.get('name'))
-            self.tool_cache.pop(kwargs.get('run_id').hex)
+            tool_name, tool_category = self.parse_tool_category(kwargs.get("name"))
+            self.tool_cache.pop(kwargs.get("run_id").hex)
             ChatMessageDao.insert_one(
                 ChatMessageModel(
                     is_bot=1,
-                    message=json.dumps(output_info),
-                    intermediate_steps=f'{input_info["steps"]}\n\nTool output:\n\n  Error: ' +
-                                       str(error),
+                    message=_dump_run_log(output_info),
+                    intermediate_steps=f"{input_info['steps']}\n\nTool output:\n\n  Error: " + str(error),
                     category=tool_category,
-                    type='end',
+                    type="end",
                     flow_id=self.flow_id,
                     chat_id=self.chat_id,
                     user_id=self.user_id,
-                    extra=json.dumps({'run_id': kwargs.get('run_id').hex})))
+                    extra=json.dumps({"run_id": kwargs.get("run_id").hex}),
+                )
+            )
diff --git a/src/backend/bisheng/tool/domain/langchain/knowledge.py b/src/backend/bisheng/tool/domain/langchain/knowledge.py
index bacb041967..57de2e40c6 100644
--- a/src/backend/bisheng/tool/domain/langchain/knowledge.py
+++ b/src/backend/bisheng/tool/domain/langchain/knowledge.py
@@ -1,9 +1,9 @@
-from typing import Any, Type, List, Optional
+from typing import Any
 
 from langchain_classic.chains.combine_documents import create_stuff_documents_chain
-from langchain_core.documents import Document, BaseDocumentCompressor
+from langchain_core.documents import BaseDocumentCompressor, Document
 from langchain_core.language_models import BaseChatModel
-from langchain_core.prompts import SystemMessagePromptTemplate, HumanMessagePromptTemplate, ChatPromptTemplate
+from langchain_core.prompts import ChatPromptTemplate, HumanMessagePromptTemplate, SystemMessagePromptTemplate
 from langchain_core.retrievers import BaseRetriever
 from langchain_core.tools import BaseTool
 from pydantic import BaseModel, Field
@@ -34,17 +34,17 @@ class ToolInputSchema(BaseModel):
 class KnowledgeRetrieverTool(BaseTool):
     name: str = "knowledge_retriever_tool"
     description: str = "在知识库中检索与查询相关的文档内容。"
-    args_schema: Type[BaseModel] = ToolInputSchema
+    args_schema: type[BaseModel] = ToolInputSchema
 
-    vector_retriever: Optional[BaseRetriever] = None
-    elastic_retriever: Optional[BaseRetriever] = None
-    rerank: Optional[BaseDocumentCompressor] = None
+    vector_retriever: BaseRetriever | None = None
+    elastic_retriever: BaseRetriever | None = None
+    rerank: BaseDocumentCompressor | None = None
     max_content: int = Field(default=15000, description="The max length of the combined document content.")
     sort_by_source_and_index: bool = Field(default=False, description="Sort by document name & chunk index.")
-    rrf_weights: List[float] = Field(default=None)
+    rrf_weights: list[float] = Field(default=None)
     rrf_remove_zero_score: bool = Field(default=False)
 
-    def _run(self, query: str, **kwargs: Any) -> List[Document]:
+    def _run(self, query: str, **kwargs: Any) -> list[Document]:
         milvus_docs, es_docs = [], []
         if self.vector_retriever:
             milvus_docs = self.vector_retriever.invoke(query)
@@ -57,7 +57,7 @@ def _run(self, query: str, **kwargs: Any) -> List[Document]:
             finally_docs = self.rerank.compress_documents(finally_docs, query)
         return finally_docs
 
-    async def _arun(self, query: str, **kwargs: Any) -> List[Document]:
+    async def _arun(self, query: str, **kwargs: Any) -> list[Document]:
         milvus_docs, es_docs = [], []
         if self.vector_retriever:
             milvus_docs = await self.vector_retriever.ainvoke(query)
@@ -70,7 +70,7 @@ async def _arun(self, query: str, **kwargs: Any) -> List[Document]:
             finally_docs = await self.rerank.acompress_documents(finally_docs, query)
         return finally_docs
 
-    def _rrf_rerank(self, milvus_docs: List[Document], es_docs: List[Document], query: str) -> List[Document]:
+    def _rrf_rerank(self, milvus_docs: list[Document], es_docs: list[Document], query: str) -> list[Document]:
         if not milvus_docs and not es_docs:
             return []
         rrf_rerank = RRFRerank(
@@ -103,16 +103,16 @@ def _rrf_rerank(self, milvus_docs: List[Document], es_docs: List[Document], quer
 class KnowledgeRagTool(BaseTool):
     name: str
     description: str
-    args_schema: Type[BaseModel] = ToolInputSchema
+    args_schema: type[BaseModel] = ToolInputSchema
 
     llm: BaseChatModel
-    chat_prompt: Optional[ChatPromptTemplate] = CHAT_PROMPT
+    chat_prompt: ChatPromptTemplate | None = CHAT_PROMPT
 
-    vector_retriever: Optional[BaseRetriever] = None
-    elastic_retriever: Optional[BaseRetriever] = None
+    vector_retriever: BaseRetriever | None = None
+    elastic_retriever: BaseRetriever | None = None
     max_content: int = Field(default=15000, description="The max length of the combined document content.")
     sort_by_source_and_index: bool = Field(default=False, description="Sort by document name & chunk index.")
-    rrf_weights: List[float] = Field(default=None)
+    rrf_weights: list[float] = Field(default=None)
     rrf_remove_zero_score: bool = Field(default=False)
 
     knowledge_retriever_tool: KnowledgeRetrieverTool = None
@@ -121,7 +121,10 @@ class KnowledgeRagTool(BaseTool):
     def init_knowledge_rag_tool(cls, name: str, description: str, **kwargs) -> BaseTool:
         llm = kwargs.pop("llm")
         chat_prompt = kwargs.pop("chat_prompt", CHAT_PROMPT)
-        # cancel assistant deep callback
+        # The retriever is an internal step of this tool, not a tool call of its
+        # own — it must stay invisible to the caller's callbacks. Dropping them
+        # here is not enough on its own (LangChain also inherits handlers from
+        # the ambient run context), so `_run`/`_arun` bypass `invoke` as well.
         kwargs.pop("callbacks", None)
         knowledge_retriever_tool = KnowledgeRetrieverTool(**kwargs)
         return cls(
@@ -134,14 +137,18 @@ def init_knowledge_rag_tool(cls, name: str, description: str, **kwargs) -> BaseT
         )
 
     def _run(self, query: str) -> Any:
-        # 1. retrieve documents
-        retrieval_result = self.knowledge_retriever_tool.invoke({"query": query})
+        # 1. retrieve documents — called directly rather than through `invoke`,
+        # which would open a nested tool run. The chat then showed a second card
+        # per search, named after the retriever instead of the knowledge base
+        # ("知识库已被删除"), and it never closed because the retriever answers
+        # with Document objects that the run-log frame could not serialize.
+        retrieval_result = self.knowledge_retriever_tool._run(query)
         llm_inputs = self._get_llm_inputs(query, retrieval_result)
         qa_chain = create_stuff_documents_chain(llm=self.llm, prompt=self.chat_prompt)
         return qa_chain.invoke(llm_inputs)
 
     async def _arun(self, query: str) -> Any:
-        retrieval_result = await self.knowledge_retriever_tool.ainvoke({"query": query})
+        retrieval_result = await self.knowledge_retriever_tool._arun(query)
         llm_inputs = self._get_llm_inputs(query, retrieval_result)
         qa_chain = create_stuff_documents_chain(llm=self.llm, prompt=self.chat_prompt)
         return await qa_chain.ainvoke(llm_inputs)
diff --git a/src/backend/test/api/test_assistant_runlog_frames.py b/src/backend/test/api/test_assistant_runlog_frames.py
new file mode 100644
index 0000000000..b2d030fd4a
--- /dev/null
+++ b/src/backend/test/api/test_assistant_runlog_frames.py
@@ -0,0 +1,61 @@
+"""A tool card in the chat closes only when its end frame arrives.
+
+Two ways it never did. The knowledge tool ran its retriever through `invoke`,
+which opened a *nested* tool run: a second card per search, named after the
+retriever rather than the knowledge base ("知识库已被删除"), whose result is a
+list of ``Document`` objects that plain ``json.dumps`` refuses — so the end frame
+raised inside the callback and the card spun forever, past the end of the
+session, with nothing persisted to recover it from.
+"""
+
+from __future__ import annotations
+
+import inspect
+import json
+
+from langchain_core.documents import Document
+
+from bisheng.api.v1.callback import AsyncGptsDebugCallbackHandler, _dump_run_log
+from bisheng.tool.domain.langchain.knowledge import KnowledgeRagTool
+
+
+def test_a_document_answer_still_serializes() -> None:
+    """The retriever's own answer shape must not be able to suppress a frame."""
+
+    payload = {"tool_key": "4138", "output": [Document(page_content="hello")]}
+
+    decoded = json.loads(_dump_run_log(payload))
+
+    assert decoded["tool_key"] == "4138"
+    assert "hello" in decoded["output"][0]
+
+
+def test_chinese_survives_the_frame() -> None:
+    # ensure_ascii stays off: the card shows this text verbatim.
+    assert "知识库" in _dump_run_log({"output": "知识库内容"})
+
+
+def test_a_nameless_end_callback_is_not_fatal() -> None:
+    """`on_tool_end` reads the name from kwargs, which is not always populated."""
+
+    assert AsyncGptsDebugCallbackHandler.parse_tool_category(None) == ("", "tool")
+    assert AsyncGptsDebugCallbackHandler.parse_tool_category("") == ("", "tool")
+
+
+def test_a_knowledge_tool_is_still_recognised_by_its_id() -> None:
+    name, category = AsyncGptsDebugCallbackHandler.parse_tool_category("knowledge_4138")
+    assert (name, category) == ("4138", "knowledge")
+
+
+def test_the_retriever_is_an_internal_step_not_a_tool_call() -> None:
+    """Going through `invoke` re-opens the callback machinery for the inner tool.
+
+    That is what produced the duplicate, wrongly-named card; the direct call
+    keeps the retrieval invisible to whoever is watching the outer tool.
+    """
+
+    for source in (inspect.getsource(KnowledgeRagTool._run), inspect.getsource(KnowledgeRagTool._arun)):
+        assert "knowledge_retriever_tool.invoke" not in source
+        assert "knowledge_retriever_tool.ainvoke" not in source
+    assert "knowledge_retriever_tool._run(query)" in inspect.getsource(KnowledgeRagTool._run)
+    assert "knowledge_retriever_tool._arun(query)" in inspect.getsource(KnowledgeRagTool._arun)
diff --git a/src/frontend/client/src/locales/en/translation.json b/src/frontend/client/src/locales/en/translation.json
index 9cf354537e..a0c276cf8b 100644
--- a/src/frontend/client/src/locales/en/translation.json
+++ b/src/frontend/client/src/locales/en/translation.json
@@ -698,6 +698,7 @@
   "com_runlog_done": "Done",
   "com_runlog_flow_deleted": "The skill has been deleted and the name cannot be retrieved",
   "com_runlog_knowledge_deleted": "The knowledge base has been deleted and the name cannot be retrieved",
+  "com_runlog_interrupted": "The tool call returned no result (the session ended)",
   "com_runlog_offline": "Offline",
   "com_runlog_searched": "Searched",
   "com_runlog_searching": "Searching",
diff --git a/src/frontend/client/src/locales/ja/translation.json b/src/frontend/client/src/locales/ja/translation.json
index 06e896f60f..de8b16c5a4 100644
--- a/src/frontend/client/src/locales/ja/translation.json
+++ b/src/frontend/client/src/locales/ja/translation.json
@@ -679,6 +679,7 @@
   "com_runlog_done": "完了",
   "com_runlog_flow_deleted": "スキルが削除されたため、スキル名を取得できません",
   "com_runlog_knowledge_deleted": "ナレッジベースが削除されたため、名称を取得できません",
+  "com_runlog_interrupted": "ツール呼び出しは結果を返しませんでした（セッション終了）",
   "com_runlog_offline": "オフライン",
   "com_runlog_searched": "検索済み",
   "com_runlog_searching": "検索中",
diff --git a/src/frontend/client/src/locales/zh-Hans/translation.json b/src/frontend/client/src/locales/zh-Hans/translation.json
index 2f2e5498e1..b7ee16bdd0 100644
--- a/src/frontend/client/src/locales/zh-Hans/translation.json
+++ b/src/frontend/client/src/locales/zh-Hans/translation.json
@@ -682,6 +682,7 @@
   "com_runlog_done": "完成",
   "com_runlog_flow_deleted": "技能已被删除，无法获取技能名",
   "com_runlog_knowledge_deleted": "知识库已被删除，无法获取知识库名",
+  "com_runlog_interrupted": "工具调用未返回结果（会话已结束）",
   "com_runlog_offline": "已下线",
   "com_runlog_searched": "已搜索",
   "com_runlog_searching": "正在搜索",
diff --git a/src/frontend/client/src/pages/appChat/appUtils/skillMethod.ts b/src/frontend/client/src/pages/appChat/appUtils/skillMethod.ts
index bc10911388..72e34f0ed6 100644
--- a/src/frontend/client/src/pages/appChat/appUtils/skillMethod.ts
+++ b/src/frontend/client/src/pages/appChat/appUtils/skillMethod.ts
@@ -1,6 +1,6 @@
 import { formatDate } from "~/utils";
 
-const runLogsTypes = ['tool', 'flow', 'knowledge']
+export const runLogsTypes = ['tool', 'flow', 'knowledge']
 // 兼容处理技能和助手
 export const SkillMethod = {
     /** 获取input发送参数 */
diff --git a/src/frontend/client/src/pages/appChat/components/MessageRunlog.tsx b/src/frontend/client/src/pages/appChat/components/MessageRunlog.tsx
index 131e6888e4..0aec2a212a 100644
--- a/src/frontend/client/src/pages/appChat/components/MessageRunlog.tsx
+++ b/src/frontend/client/src/pages/appChat/components/MessageRunlog.tsx
@@ -16,6 +16,9 @@ export default function MessageRunlog({ data }) {
     }, [_chatsState, data])
 
     const [title, lost] = useMemo(() => {
+        // Settled by the session closing rather than by its own end frame — say so
+        // instead of showing a success tick the call never earned.
+        if (data.interrupted) return [t('com_runlog_interrupted'), true]
         let lost = false
         let title = ''
         const status = data.end ? t('com_runlog_used') : t('com_runlog_using')
diff --git a/src/frontend/client/src/pages/appChat/useChatHelpers.ts b/src/frontend/client/src/pages/appChat/useChatHelpers.ts
index 205603ac61..5e87358618 100644
--- a/src/frontend/client/src/pages/appChat/useChatHelpers.ts
+++ b/src/frontend/client/src/pages/appChat/useChatHelpers.ts
@@ -8,7 +8,7 @@ import { Chat } from "~/@types/chat"
 import { baseMsgItem } from "~/api/apps"
 import { formatDate, generateUUID } from "~/utils"
 import { FLOW_TYPES } from "."
-import { SkillMethod } from "./appUtils/skillMethod"
+import { runLogsTypes, SkillMethod } from "./appUtils/skillMethod"
 import { bishengConfState, chatApiVersionState, chatIdState, chatsState, currentChatState, currentRunningState, runningState } from "./store/atoms"
 import { emitAreaTextEvent, EVENT_TYPE } from "./useAreaText"
 
@@ -341,6 +341,20 @@ export default function useChatHelpers() {
             )
         },
         skillCloseMsg: () => {
+            // A tool card only closes when its `end` frame arrives. Lose one — a
+            // serialization failure, a dropped socket — and it spins forever with
+            // nothing to recover it: the round is over and nothing was persisted.
+            // Settle what is still open, marked interrupted rather than wearing a
+            // success tick it never earned.
+            setChats((prev) =>
+                updateChatMessages(prev, chatId, (messages) =>
+                    messages.map((msg) =>
+                        runLogsTypes.includes(msg.category) && !msg.end
+                            ? { ...msg, end: true, interrupted: true }
+                            : msg
+                    )
+                )
+            )
             setRunningState((prev) => {
                 return {
                     ...prev,
diff --git a/src/frontend/platform/public/locales/en-US/bs.json b/src/frontend/platform/public/locales/en-US/bs.json
index 3ad3f74302..062028f0db 100644
--- a/src/frontend/platform/public/locales/en-US/bs.json
+++ b/src/frontend/platform/public/locales/en-US/bs.json
@@ -557,7 +557,8 @@
       "flowOffline": "{{name}} is offline",
       "flowDeleted": "The skill has been deleted, its name is unavailable",
       "toolDeleted": "The tool has been deleted, its name is unavailable",
-      "knowledgeDeleted": "The knowledge base has been deleted, its name is unavailable"
+      "knowledgeDeleted": "The knowledge base has been deleted, its name is unavailable",
+      "interrupted": "The tool call returned no result (the session ended)"
     }
   },
   "importLinsight": {
diff --git a/src/frontend/platform/public/locales/ja/bs.json b/src/frontend/platform/public/locales/ja/bs.json
index 40a4e9abff..c5f04153dd 100644
--- a/src/frontend/platform/public/locales/ja/bs.json
+++ b/src/frontend/platform/public/locales/ja/bs.json
@@ -547,7 +547,8 @@
       "flowOffline": "{{name}} はオフラインです",
       "flowDeleted": "スキルが削除されているため、スキル名を取得できません",
       "toolDeleted": "ツールが削除されているため、ツール名を取得できません",
-      "knowledgeDeleted": "ナレッジベースが削除されているため、名称を取得できません"
+      "knowledgeDeleted": "ナレッジベースが削除されているため、名称を取得できません",
+      "interrupted": "ツール呼び出しは結果を返しませんでした（セッション終了）"
     }
   },
   "model": {
diff --git a/src/frontend/platform/public/locales/zh-Hans/bs.json b/src/frontend/platform/public/locales/zh-Hans/bs.json
index 98f55c3e99..095dddfa66 100644
--- a/src/frontend/platform/public/locales/zh-Hans/bs.json
+++ b/src/frontend/platform/public/locales/zh-Hans/bs.json
@@ -552,7 +552,8 @@
       "flowOffline": "{{name}} 已下线",
       "flowDeleted": "技能已被删除，无法获取技能名",
       "toolDeleted": "工具已被删除，无法获取工具名",
-      "knowledgeDeleted": "知识库已被删除，无法获取知识库名"
+      "knowledgeDeleted": "知识库已被删除，无法获取知识库名",
+      "interrupted": "工具调用未返回结果（会话已结束）"
     }
   },
   "model": {
diff --git a/src/frontend/platform/src/components/bs-comp/chatComponent/ChatInput.tsx b/src/frontend/platform/src/components/bs-comp/chatComponent/ChatInput.tsx
index ca0100d1f8..7c5f4aa50d 100644
--- a/src/frontend/platform/src/components/bs-comp/chatComponent/ChatInput.tsx
+++ b/src/frontend/platform/src/components/bs-comp/chatComponent/ChatInput.tsx
@@ -29,7 +29,7 @@ export default function ChatInput({ clear, form, questions, inputForm, wsUrl, on
 
     const { isLoading: audioOpening } = useAudioStore()
 
-    const { messages, hisMessages, chatId, createSendMsg, createWsMsg, updateCurrentMessage, destory, setShowGuideQuestion } = useMessageStore()
+    const { messages, hisMessages, chatId, createSendMsg, createWsMsg, updateCurrentMessage, closeDanglingRunLogs, destory, setShowGuideQuestion } = useMessageStore()
     const currentChatIdRef = useRef(null)
     const inputRef = useRef(null)
     const continueRef = useRef(false)
@@ -256,6 +256,7 @@ export default function ChatInput({ clear, form, questions, inputForm, wsUrl, on
 
             if (!msgClosedRef.current) msgClosedRef.current = true
         } else if (data.type === "close") {
+            closeDanglingRunLogs()
             setStop({ show: false, disable: false })
             setInputLock((prev) => (prev.reason ? prev : { locked: false, reason: '' }))
         }
diff --git a/src/frontend/platform/src/components/bs-comp/chatComponent/RunLog.tsx b/src/frontend/platform/src/components/bs-comp/chatComponent/RunLog.tsx
index 7eb3fbde73..0bf5ce1862 100644
--- a/src/frontend/platform/src/components/bs-comp/chatComponent/RunLog.tsx
+++ b/src/frontend/platform/src/components/bs-comp/chatComponent/RunLog.tsx
@@ -20,6 +20,9 @@ export default function RunLog({ data }) {
     const assistantState = useAssistantStore(state => state.assistantState)
 
     const [title, lost] = useMemo(() => {
+        // Settled by the session closing rather than by its own end frame — say so
+        // instead of showing a success tick the call never earned.
+        if (data.interrupted) return [t('chat.runLog.interrupted'), true]
         let lost = false
         let title = ''
         const status = data.end ? t('chat.runLog.used') : t('chat.runLog.using')
diff --git a/src/frontend/platform/src/components/bs-comp/chatComponent/messageStore.ts b/src/frontend/platform/src/components/bs-comp/chatComponent/messageStore.ts
index 5892bf684c..dcac369cd7 100644
--- a/src/frontend/platform/src/components/bs-comp/chatComponent/messageStore.ts
+++ b/src/frontend/platform/src/components/bs-comp/chatComponent/messageStore.ts
@@ -41,6 +41,7 @@ type Actions = {
     insetSystemMsg: (text: string) => void;
     insetBsMsg: (text: string) => void;
     setShowGuideQuestion: (text: boolean) => void;
+    closeDanglingRunLogs: () => void;
     clearMsgs: () => void;
 }
 
@@ -152,6 +153,24 @@ export const useMessageStore = create<State & Actions>((set, get) => ({
             set({ historyEnd: true })
         }
     },
+    /**
+     * A tool card only closes when its `end` frame arrives. Lose one — a
+     * serialization failure, a dropped socket — and the card spins forever, with
+     * nothing to recover it: the round is over and nothing was persisted. So on
+     * session close, settle whatever is still open and mark it interrupted
+     * rather than leaving a success tick it never earned.
+     */
+    closeDanglingRunLogs() {
+        const messages = get().messages
+        if (!messages.some(msg => runLogsTypes.includes(msg.category) && !msg.end)) return
+        set({
+            messages: messages.map(msg =>
+                runLogsTypes.includes(msg.category) && !msg.end
+                    ? { ...msg, end: true, interrupted: true }
+                    : msg
+            )
+        })
+    },
     clearMsgs() {
         setTimeout(() => {
             set({ hisMessages: [], messages: [], historyEnd: true })
diff --git a/src/frontend/platform/src/test/assistantRunLogClosing.test.ts b/src/frontend/platform/src/test/assistantRunLogClosing.test.ts
new file mode 100644
index 0000000000..16ed9e5506
--- /dev/null
+++ b/src/frontend/platform/src/test/assistantRunLogClosing.test.ts
@@ -0,0 +1,58 @@
+import { useMessageStore } from "@/components/bs-comp/chatComponent/messageStore";
+import { ChatMessageType } from "@/types/chat";
+import { beforeEach, describe, expect, it } from "vitest";
+
+type RunLogCard = ChatMessageType & { interrupted?: boolean };
+
+const asMessages = (rows: unknown[]) => rows as ChatMessageType[];
+const readMessages = () => useMessageStore.getState().messages as RunLogCard[];
+
+/**
+ * A tool card closes only when its own end frame arrives. One got lost — the
+ * knowledge retriever answered with objects the frame could not serialize — and
+ * the card kept spinning after the session had ended, with nothing persisted to
+ * recover it from. Closing the session must settle whatever is still open.
+ */
+describe("dangling run-log cards at session close", () => {
+  const runLog = (id: string, end: boolean) => ({
+    id,
+    category: "knowledge",
+    end,
+    message: { tool_key: "4138" },
+    thought: "",
+  });
+
+  beforeEach(() => {
+    useMessageStore.setState({ messages: [], hisMessages: [] });
+  });
+
+  it("settles a tool card whose end frame never arrived", () => {
+    useMessageStore.setState({ messages: asMessages([runLog("a", true), runLog("b", false)]) });
+
+    useMessageStore.getState().closeDanglingRunLogs();
+
+    const [settled, interrupted] = readMessages();
+    expect(settled.interrupted).toBeUndefined();
+    // Marked, not silently ticked: it never earned a success icon.
+    expect(interrupted.end).toBe(true);
+    expect(interrupted.interrupted).toBe(true);
+  });
+
+  it("leaves the streaming answer alone", () => {
+    const answer = { id: "answer", category: "answer", end: false, message: "", thought: "" };
+    useMessageStore.setState({ messages: asMessages([answer]) });
+
+    useMessageStore.getState().closeDanglingRunLogs();
+
+    expect(readMessages()[0].end).toBe(false);
+  });
+
+  it("does not touch the list when every card is already closed", () => {
+    const messages = asMessages([runLog("a", true)]);
+    useMessageStore.setState({ messages });
+
+    useMessageStore.getState().closeDanglingRunLogs();
+
+    expect(useMessageStore.getState().messages).toBe(messages);
+  });
+});