diff --git a/apps/api/routers/agent.py b/apps/api/routers/agent.py
index 708c80c..4892e0f 100644
--- a/apps/api/routers/agent.py
+++ b/apps/api/routers/agent.py
@@ -43,9 +43,76 @@ def _parse_sse_events(chunk: str) -> list[tuple[str, dict]]:
     return events
 
 
+def _db_messages_to_openai(db_messages: list) -> list[dict]:
+    """把 DB 中的 AgentMessage 重建为 OpenAI 格式 messages。
+
+    - assistant + meta.tool_calls → {role, content, tool_calls}
+    - tool + meta.tool_call_id → {role, tool_call_id, content}
+    - 其余 → {role, content}
+
+    修②：后端真相源要求后端拼历史，不再依赖前端重发全历史。
+    """
+    openai_msgs: list[dict] = []
+    for m in db_messages:
+        role = m.role
+        meta = m.meta or {}
+        if role == "assistant" and meta.get("tool_calls"):
+            openai_msgs.append(
+                {
+                    "role": "assistant",
+                    "content": m.content or None,
+                    "tool_calls": meta["tool_calls"],
+                }
+            )
+        elif role == "tool":
+            openai_msgs.append(
+                {
+                    "role": "tool",
+                    "tool_call_id": meta.get("tool_call_id", ""),
+                    "content": m.content,
+                }
+            )
+        else:
+            openai_msgs.append({"role": role, "content": m.content})
+    return openai_msgs
+
+
+def _new_messages_to_dicts(req_messages: list) -> list[dict]:
+    """把本次请求带来的新消息（AgentMessage schema）转成 OpenAI dict。
+
+    前端修③后只发本次新增 user 消息，但兼容前端仍发 assistant/tool 的情况。
+    """
+    out: list[dict] = []
+    for m in req_messages:
+        role = m.role
+        if role == "assistant" and (m.meta or {}).get("tool_calls"):
+            out.append(
+                {
+                    "role": "assistant",
+                    "content": m.content or None,
+                    "tool_calls": m.meta["tool_calls"],
+                }
+            )
+        elif role == "tool":
+            out.append(
+                {
+                    "role": "tool",
+                    "tool_call_id": (m.meta or {}).get("tool_call_id", "") or m.tool_call_id or "",
+                    "content": m.content,
+                }
+            )
+        else:
+            out.append({"role": role, "content": m.content})
+    return out
+
+
 @router.post("/agent/chat")
 async def agent_chat(req: AgentChatRequest):
-    """Agent 对话 - SSE 流式响应（带持久化 + 工具调用记录）"""
+    """Agent 对话 - SSE 流式响应（带持久化 + 工具调用记录）
+
+    修②：后端真相源——前端只发本次新增消息，后端按 conversation_id 从 DB
+    读历史拼接。新会话首条消息后端创建并经 SSE 返 conversation_id（修①）。
+    """
     from packages.storage.db import session_scope
     from packages.storage.repositories import (
         AgentConversationRepository,
@@ -64,7 +131,7 @@ async def agent_chat(req: AgentChatRequest):
             if not conv:
                 conversation_id = None
 
-        # 无 conversation_id：创建新会话
+        # 无 conversation_id：创建新会话（先建空壳，首条 user 消息稍后存）
         if not conversation_id:
             first_user_msg = next((m for m in req.messages if m.role == "user"), None)
             title = first_user_msg.content[:50] if first_user_msg else "新对话"
@@ -72,50 +139,49 @@ async def agent_chat(req: AgentChatRequest):
             conversation_id = conv.id
 
         # 保存本次请求带来的所有新消息（user + assistant + tool）
-        # 已有的历史消息从 DB 加载，不重复保存
-        saved_ids: set[str] = set()
+        saved_keys: set[str] = set()
         for msg in req.messages:
             if msg.role == "system":
                 continue
             content_key = f"{msg.role}:{msg.content[:200]}"
-            if content_key not in saved_ids:
+            if content_key not in saved_keys:
                 msg_repo.create(
                     conversation_id=conversation_id,
                     role=msg.role,
                     content=msg.content,
                     meta=msg.meta,
                 )
-                saved_ids.add(content_key)
-
-    # 构建传给 stream_chat 的 messages（包含 DB 加载的历史）
-    # 前端传的是本次新增消息，需要拼上 DB 里的历史
-    msgs = [m.model_dump() for m in req.messages]
-
-    def _build_save_callback(conv_id: str) -> Callable[[list[dict]], None]:
-        """创建压缩回写回调"""
-
-        def on_compact(compressed_messages: list[dict]):
-            with session_scope() as session:
-                msg_repo = AgentMessageRepository(session)
-                # 删除旧消息，写入压缩后的消息
-                msg_repo.delete_by_conversation(conv_id)
-                for msg in compressed_messages:
-                    msg_repo.create(
-                        conversation_id=conv_id,
-                        role=msg.get("role", "user"),
-                        content=msg.get("content", ""),
-                        meta=msg.get("meta"),
-                    )
-
-        return on_compact
+                saved_keys.add(content_key)
+
+        # 修②：从 DB 读全量历史，重建为 OpenAI 格式，作为传给 stream_chat 的 messages
+        db_msgs = msg_repo.list_by_conversation(conversation_id, limit=500)
+        history_msgs = _db_messages_to_openai(db_msgs)
+
+    # 构建传给 stream_chat 的 messages：DB 历史 + 本次新增（前端只发新增时，req.messages 即新增）
+    # 已在 DB 中存的本次新增消息，list_by_conversation 也会读出，避免重复加入。
+    new_msgs = _new_messages_to_dicts(req.messages)
+    # 用 content key 去重：DB 历史已含本次新增，只需把 DB 没覆盖到的情况补齐
+    history_keys = {f"{m.get('role')}:{(m.get('content') or '')[:200]}" for m in history_msgs}
+    extra_new = [
+        m
+        for m in new_msgs
+        if f"{m.get('role')}:{(m.get('content') or '')[:200]}" not in history_keys
+    ]
+    msgs = history_msgs + extra_new
 
     text_buf = ""
     tool_records: list[dict] = []
     tool_call_id: str | None = None
+    saved_done = False  # 修④：done 去重，一个 stream 只存一次 assistant
 
     def stream_with_save():
-        nonlocal text_buf, tool_records, tool_call_id
-        sse_iter, updated_conversation = stream_chat(
+        nonlocal text_buf, tool_records, tool_call_id, saved_done
+        # 修①：SSE 首事件返 conversation_id，前端采用后端 id 作 localStorage key
+        from packages.agent_core.sse import make_sse
+
+        yield make_sse("conversation_init", {"conversation_id": conversation_id})
+
+        sse_iter, _updated_conversation = stream_chat(
             msgs, confirmed_action_id=req.confirmed_action_id
         )
         for chunk in sse_iter:
@@ -161,7 +227,9 @@ def stream_with_save():
                             "data": data.get("data"),
                         }
                     )
-                elif event_type == "done" and (text_buf or tool_records):
+                elif event_type == "done" and not saved_done and (text_buf or tool_records):
+                    # 修④：只存一次 assistant，后续 done（loop 内部/重试）跳过
+                    saved_done = True
                     with session_scope() as session:
                         msg_repo = AgentMessageRepository(session)
                         msg_repo.create(
@@ -178,12 +246,107 @@ def stream_with_save():
     )
 
 
+def _resolve_conversation_id_from_action(action_id: str) -> str | None:
+    """从 pending action 取 conversation_id，供 confirm/reject 持久化用。"""
+    from packages.storage.db import session_scope
+    from packages.storage.repositories import AgentPendingActionRepository
+
+    try:
+        with session_scope() as session:
+            repo = AgentPendingActionRepository(session)
+            record = repo.get_by_id(action_id)
+            return record.conversation_id if record else None
+    except Exception:
+        return None
+
+
+def _stream_with_save_for_action(
+    conversation_id: str | None,
+    sse_iter_factory: Callable[[], tuple],
+):
+    """修⑤：confirm/reject 复用同样的持久化逻辑。
+
+    sse_iter_factory 返回 (sse_iter, conversation)。
+    """
+    from packages.agent_core.sse import make_sse
+
+    text_buf = ""
+    tool_records: list[dict] = []
+    tool_call_id: str | None = None
+    saved_done = False
+
+    def _gen():
+        nonlocal text_buf, tool_records, tool_call_id, saved_done
+        from packages.storage.db import session_scope
+        from packages.storage.repositories import AgentMessageRepository
+
+        if conversation_id:
+            yield make_sse("conversation_init", {"conversation_id": conversation_id})
+
+        sse_iter, _conversation = sse_iter_factory()
+        for chunk in sse_iter:
+            yield chunk
+            if not conversation_id:
+                continue
+            for event_type, data in _parse_sse_events(chunk):
+                if event_type == "text_delta":
+                    text_buf += data.get("content", "")
+                elif event_type == "tool_start":
+                    tool_call_id = data.get("id")
+                elif event_type == "tool_result":
+                    tool_records.append(
+                        {
+                            "name": data.get("name"),
+                            "success": data.get("success"),
+                            "summary": data.get("summary"),
+                            "data": data.get("data"),
+                        }
+                    )
+                    with session_scope() as session:
+                        msg_repo = AgentMessageRepository(session)
+                        msg_repo.create(
+                            conversation_id=conversation_id,
+                            role="tool",
+                            content=json.dumps(
+                                {
+                                    "name": data.get("name"),
+                                    "success": data.get("success"),
+                                    "summary": data.get("summary"),
+                                    "data": data.get("data"),
+                                },
+                                ensure_ascii=False,
+                            ),
+                            meta={"tool_call_id": tool_call_id},
+                        )
+                elif event_type == "action_result":
+                    tool_records.append(
+                        {
+                            "action_id": data.get("id"),
+                            "success": data.get("success"),
+                            "summary": data.get("summary"),
+                            "data": data.get("data"),
+                        }
+                    )
+                elif event_type == "done" and not saved_done and (text_buf or tool_records):
+                    saved_done = True
+                    with session_scope() as session:
+                        msg_repo = AgentMessageRepository(session)
+                        msg_repo.create(
+                            conversation_id=conversation_id,
+                            role="assistant",
+                            content=text_buf,
+                            meta={"tool_calls": tool_records} if tool_records else None,
+                        )
+
+    return _gen()
+
+
 @router.post("/agent/confirm/{action_id}")
 async def agent_confirm(action_id: str):
-    """确认执行 Agent 挂起的操作"""
-    sse_iter, _ = confirm_action(action_id)
+    """确认执行 Agent 挂起的操作（修⑤：持久化 tool/assistant 消息）"""
+    conversation_id = _resolve_conversation_id_from_action(action_id)
     return StreamingResponse(
-        sse_iter,
+        _stream_with_save_for_action(conversation_id, lambda: confirm_action(action_id)),
         media_type="text/event-stream",
         headers=_SSE_HEADERS,
     )
@@ -191,10 +354,10 @@ async def agent_confirm(action_id: str):
 
 @router.post("/agent/reject/{action_id}")
 async def agent_reject(action_id: str):
-    """拒绝 Agent 挂起的操作"""
-    sse_iter, _ = reject_action(action_id)
+    """拒绝 Agent 挂起的操作（修⑤：持久化 tool/assistant 消息）"""
+    conversation_id = _resolve_conversation_id_from_action(action_id)
     return StreamingResponse(
-        sse_iter,
+        _stream_with_save_for_action(conversation_id, lambda: reject_action(action_id)),
         media_type="text/event-stream",
         headers=_SSE_HEADERS,
     )
diff --git a/frontend/src/contexts/AgentSessionContext.tsx b/frontend/src/contexts/AgentSessionContext.tsx
index 2fb3c25..b90a88a 100644
--- a/frontend/src/contexts/AgentSessionContext.tsx
+++ b/frontend/src/contexts/AgentSessionContext.tsx
@@ -87,7 +87,7 @@ export function AgentSessionProvider({ children }: { children: React.ReactNode }
   const pendingActions = useMemo(() => new Set(pendingActionIds), [pendingActionIds]);
   const confirmingActions = useMemo(() => new Set(confirmingActionIds), [confirmingActionIds]);
 
-  const { activeId, createConversation, saveMessages } = useConversationCtx();
+  const { activeId, createConversation, saveMessages, setActiveId } = useConversationCtx();
   const justCreatedRef = useRef(false);
   const activeIdRef = useRef(activeId);
   activeIdRef.current = activeId;
@@ -266,6 +266,15 @@ export function AgentSessionProvider({ children }: { children: React.ReactNode }
       const id = uid();
 
       switch (type as SSEEventType) {
+        case "conversation_init": {
+          // 修①：后端返回真实 conversation_id。新会话前端曾用临时 id 创建，
+          // 现采用后端 id。把当前 items 迁移到后端 id 下（saveMessages 用 activeId）。
+          const backendId = data.conversation_id as string;
+          if (backendId && backendId !== activeIdRef.current) {
+            setActiveId(backendId);
+          }
+          break;
+        }
         case "text_delta": {
           streamBufRef.current += (data.content as string) || "";
           scheduleFlush();
@@ -558,7 +567,7 @@ export function AgentSessionProvider({ children }: { children: React.ReactNode }
         }
       }
     },
-    [scheduleFlush, drainBuffer, applyPendingText]
+    [scheduleFlush, drainBuffer, applyPendingText, setActiveId]
   );
 
   /**
@@ -610,8 +619,6 @@ export function AgentSessionProvider({ children }: { children: React.ReactNode }
   );
 
   /* ---- 发送消息 ---- */
-  const itemsRef = useRef(items);
-  itemsRef.current = items;
 
   const sendMessage = useCallback(
     async (text: string) => {
@@ -629,37 +636,9 @@ export function AgentSessionProvider({ children }: { children: React.ReactNode }
         ...prev,
         { id: `user_${uid()}`, type: "user" as const, content: text.trim(), timestamp: new Date() },
       ]);
-      // 使用 ref 获取最新 items，避免闭包过时
-      const currentItems = itemsRef.current;
-      const msgs: AgentMessage[] = [];
-      for (const it of currentItems) {
-        if (it.type === "user") {
-          msgs.push({ role: "user", content: it.content });
-        } else if (it.type === "assistant") {
-          msgs.push({ role: "assistant", content: it.content });
-        } else if (it.type === "step_group" && it.steps) {
-          const summaries = it.steps
-            .filter((s) => s.status === "done" || s.status === "error")
-            .map((s) => `[工具: ${s.toolName}] ${s.success ? "成功" : "失败"}: ${s.summary || ""}`)
-            .join("\n");
-          if (summaries) {
-            msgs.push({ role: "assistant", content: `执行了以下操作:\n${summaries}` });
-          }
-        } else if (it.type === "action_confirm") {
-          msgs.push({
-            role: "assistant",
-            content: `[等待确认] ${it.actionDescription || it.actionTool || ""}`,
-          });
-        } else if (it.type === "artifact") {
-          msgs.push({
-            role: "assistant",
-            content: `[已生成内容: ${it.artifactTitle || "未命名"}]\n${it.artifactContent || ""}`,
-          });
-        } else if (it.type === "error") {
-          msgs.push({ role: "assistant", content: `[错误: ${it.content}]` });
-        }
-      }
-      msgs.push({ role: "user" as const, content: text.trim() });
+      // 修②③：后端真相源——前端只发本次新 user 消息，历史由后端从 DB 拼。
+      // convId 可能为临时前端 id（新会话），后端收到后经 conversation_init 返真实 id。
+      const msgs: AgentMessage[] = [{ role: "user" as const, content: text.trim() }];
       try {
         const ac = new AbortController();
         abortRef.current = ac;
diff --git a/frontend/src/contexts/ConversationContext.tsx b/frontend/src/contexts/ConversationContext.tsx
index 71c4c72..bd3ab56 100644
--- a/frontend/src/contexts/ConversationContext.tsx
+++ b/frontend/src/contexts/ConversationContext.tsx
@@ -18,6 +18,8 @@ interface ConversationCtx {
   switchConversation: (id: string) => void;
   saveMessages: (messages: ConversationMessage[]) => void;
   deleteConversation: (id: string) => void;
+  /** 修①：直接设置当前会话 id（供后端 SSE 返 id 后采用后端 id 作 localStorage key） */
+  setActiveId: (id: string | null) => void;
 }
 
 const Ctx = createContext<ConversationCtx | null>(null);
diff --git a/frontend/src/hooks/useConversations.ts b/frontend/src/hooks/useConversations.ts
index 2fa5d15..1c1caf7 100644
--- a/frontend/src/hooks/useConversations.ts
+++ b/frontend/src/hooks/useConversations.ts
@@ -261,6 +261,7 @@ export function useConversations() {
     switchConversation,
     saveMessages,
     deleteConversation,
+    setActiveId,
   };
 }
 
diff --git a/frontend/src/types/index.ts b/frontend/src/types/index.ts
index c3ca97e..ca51151 100644
--- a/frontend/src/types/index.ts
+++ b/frontend/src/types/index.ts
@@ -1059,6 +1059,7 @@ export interface AuthStatusResponse {
 }
 
 export type SSEEventType =
+  | "conversation_init"
   | "text_delta"
   | "tool_start"
   | "tool_result"
diff --git a/packages/agent_core/loop.py b/packages/agent_core/loop.py
index 3e9cb0f..e76716b 100644
--- a/packages/agent_core/loop.py
+++ b/packages/agent_core/loop.py
@@ -293,10 +293,29 @@ def run(self, conversation: list[dict]) -> Iterator[str]:
 
             # 有确认工具时，pending 并暂停
             if confirm_calls:
+                # 修⑧：LLM 一轮内可能返回多个 confirm 工具，之前只处理 confirm_calls[0]
+                # 其余被丢弃，导致 tool_calls 与 tool_result 不配对，下轮 LLM 报错。
+                # 现处理首个（挂起），其余转文本提示让 LLM 下一轮重提。
                 tc = confirm_calls[0]
+                if len(confirm_calls) > 1:
+                    extra = len(confirm_calls) - 1
+                    for extra_tc in confirm_calls[1:]:
+                        conversation.append(
+                            {
+                                "role": "tool",
+                                "tool_call_id": extra_tc.tool_call_id,
+                                "content": f"还有 {extra} 个待确认操作（{extra_tc.tool_name}），"
+                                f"请先确认当前操作后再继续。",
+                            }
+                        )
                 yield from self._handle_confirm_tool(tc, conversation)
                 return
 
+        # 修⑩：max_rounds 耗尽不应静默截断，给用户一个提示再 done
+        yield make_sse(
+            "text_delta",
+            {"content": "\n\n[已达到本轮最大对话轮次，如有需要请继续提问]"},
+        )
         yield make_sse("done", {})
 
     def _handle_stream_event(
@@ -309,18 +328,37 @@ def _handle_stream_event(
         if event.type == "text_delta":
             return make_sse("text_delta", {"content": event.content})
         elif event.type == "tool_call":
+            # 修⑨：tool_arguments 非法 JSON 时不应崩整个 run，跳过该 tool_call + 报错
+            args: dict = {}
+            if event.tool_arguments:
+                try:
+                    args = json.loads(event.tool_arguments)
+                    if not isinstance(args, dict):
+                        args = {}
+                except (json.JSONDecodeError, TypeError) as e:
+                    logger.warning(
+                        "tool_call 参数 JSON 解析失败，跳过: %s [%s] err=%s",
+                        event.tool_call_id,
+                        event.tool_name,
+                        e,
+                    )
+                    return make_sse(
+                        "error",
+                        {"message": f"工具 {event.tool_name} 参数解析失败，已跳过"},
+                    )
             tool_calls.append(
                 PaperMindToolCall(
                     tool_call_id=event.tool_call_id,
                     tool_name=event.tool_name,
-                    arguments=json.loads(event.tool_arguments) if event.tool_arguments else {},
+                    arguments=args,
                 )
             )
         elif event.type == "error":
             return make_sse("error", {"message": event.content})
         elif event.type == "usage" and self._on_usage:
+            # 修⑬：provider 应从 llm 取真实 provider，model 从 event 取；之前两个参数都传 event.model
             self._on_usage(
-                event.model or "",
+                self.llm.provider or "",
                 event.model or "",
                 event.input_tokens or 0,
                 event.output_tokens or 0,
diff --git a/packages/ai/agent_service.py b/packages/ai/agent_service.py
index 1f5c8de..9a47cd9 100644
--- a/packages/ai/agent_service.py
+++ b/packages/ai/agent_service.py
@@ -121,10 +121,13 @@ def _build_user_profile() -> str:
                 titles = [p.title[:60] for p in deep_read]
                 parts.append(f"最近精读：{'; '.join(titles)}")
 
-            skimmed = paper_repo.list_by_read_status(ReadStatus.skimmed, limit=200)
-            unread = paper_repo.list_by_read_status(ReadStatus.unread, limit=200)
+            # 修⑭：此前用 limit 列表的 len 当总数（limit=5 → 精读永远≤5，真实 366）。
+            # 改用 count_by_read_status 查真实总数
+            deep_count = paper_repo.count_by_read_status(ReadStatus.deep_read)
+            skim_count = paper_repo.count_by_read_status(ReadStatus.skimmed)
+            unread_count = paper_repo.count_by_read_status(ReadStatus.unread)
             parts.append(
-                f"论文库状态：{len(deep_read)} 篇精读、{len(skimmed)} 篇粗读、{len(unread)} 篇未读"
+                f"论文库状态：{deep_count} 篇精读、{skim_count} 篇粗读、{unread_count} 篇未读"
             )
 
         if parts:
@@ -254,7 +257,7 @@ def _err_iter():
 
         def _confirm_iter():
             yield from loop.execute_and_continue(action, conversation)
-            yield make_sse("done", {})
+            # 修④：loop 内部已发 done，service 不再重复 yield（此前导致 done=2~3 重复持久化）
 
         return _confirm_iter(), conversation
 
@@ -263,7 +266,7 @@ def _confirm_iter():
 
     def _chat_iter():
         yield from loop.run(conversation)
-        yield make_sse("done", {})
+        # 修④：loop 内部已发 done，service 不再重复
 
     return _chat_iter(), conversation
 
@@ -304,7 +307,7 @@ def _err_iter():
 
     def _confirm_iter():
         yield from loop.execute_confirmed_action(action, conversation)
-        yield make_sse("done", {})
+        # 修④：loop 内部已发 done，service 不再重复
 
     return _confirm_iter(), conversation
 
@@ -340,7 +343,7 @@ def _reject_iter():
         )
         if loop:
             yield from loop.execute_rejected_action(action, conversation)
-        yield make_sse("done", {})
+        # 修④：loop 内部已发 done，service 不再重复
 
     return _reject_iter(), conversation
 
diff --git a/packages/storage/repositories/paper.py b/packages/storage/repositories/paper.py
index b477c22..b4dde07 100644
--- a/packages/storage/repositories/paper.py
+++ b/packages/storage/repositories/paper.py
@@ -238,6 +238,11 @@ def count_all(self) -> int:
         q = select(func.count()).select_from(Paper)
         return self.session.execute(q).scalar() or 0
 
+    def count_by_read_status(self, status: ReadStatus) -> int:
+        """按阅读状态计数（供用户画像等用真实总数，不用 limit 列表的 len）"""
+        q = select(func.count()).select_from(Paper).where(Paper.read_status == status)
+        return self.session.execute(q).scalar() or 0
+
     def list_paginated(
         self,
         page: int = 1,
diff --git a/tests/test_agent_conversation.py b/tests/test_agent_conversation.py
new file mode 100644
index 0000000..b2a1243
--- /dev/null
+++ b/tests/test_agent_conversation.py
@@ -0,0 +1,276 @@
+"""Agent 会话真相源 / 持久化 / loop 健壮性 测试
+
+覆盖本次修复的关键点：
+- 修①②：后端按 conversation_id 拼 DB 历史，不依赖前端重发
+- 修④：done 事件不重复存 assistant
+- 修⑤：confirm/reject 后 tool/assistant 消息落 DB
+- 修⑨：loop 对非法 tool_arguments JSON 不崩
+- 修⑬：usage 回调传真实 provider
+- 修⑭：用户画像 count 用 count_by_read_status 真实总数
+@author Color2333
+"""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock
+
+from packages.domain.enums import ReadStatus
+from packages.domain.schemas import PaperCreate
+from packages.storage.repositories import (
+    AgentConversationRepository,
+    AgentMessageRepository,
+    PaperRepository,
+)
+
+# ---------- 修⑭：count_by_read_status 返回真实总数 ----------
+
+
+class TestUserProfileCountByReadStatus:
+    def test_count_by_read_status_reflects_real_total(self, db_session):
+        """limit 列表的 len 会受 limit 截断，count_by_read_status 返回真实总数。"""
+        repo = PaperRepository(db_session)
+        for i in range(5):
+            repo.upsert_paper(
+                PaperCreate(
+                    arxiv_id=f"2401.{i:05d}",
+                    title=f"paper {i}",
+                    abstract="abs",
+                    authors=["A"],
+                    pdf_url="https://arxiv.org/pdf/x",
+                    source="arxiv",
+                )
+            )
+        # 把前 3 篇标为 deep_read
+        papers = repo.list_all(limit=10)
+        for p in papers[:3]:
+            repo.update_read_status(p.id, ReadStatus.deep_read)
+        db_session.flush()  # update_read_status 只改 ORM 对象，需 flush 让后续 query 看到
+
+        # count_by_read_status 返回真实总数 3
+        assert repo.count_by_read_status(ReadStatus.deep_read) == 3
+        assert repo.count_by_read_status(ReadStatus.unread) == 2
+        # 对比：limit list 若 limit=2 则 len=2，与真实总数不一致
+        limited = repo.list_all(limit=2)
+        assert len(limited) == 2, "limit 列表被截断，证明 count 才是真实总数来源"
+
+
+# ---------- 修①②：后端拼历史（DB 历史重建为 OpenAI 格式） ----------
+
+
+class TestBackendHistoryMerge:
+    def test_db_messages_to_openai_rebuilds_tool_and_assistant(self, db_session):
+        """后端从 DB 读历史，按 role/meta 重建 OpenAI messages。"""
+        from apps.api.routers.agent import _db_messages_to_openai
+
+        conv_repo = AgentConversationRepository(db_session)
+        msg_repo = AgentMessageRepository(db_session)
+        conv = conv_repo.create(title="t")
+        msg_repo.create(conversation_id=conv.id, role="user", content="你好")
+        msg_repo.create(
+            conversation_id=conv.id,
+            role="assistant",
+            content="我来搜索",
+            meta={
+                "tool_calls": [
+                    {
+                        "id": "tc1",
+                        "type": "function",
+                        "function": {"name": "search", "arguments": "{}"},
+                    }
+                ]
+            },
+        )
+        msg_repo.create(
+            conversation_id=conv.id,
+            role="tool",
+            content='{"ok": true}',
+            meta={"tool_call_id": "tc1"},
+        )
+
+        db_msgs = msg_repo.list_by_conversation(conv.id)
+        rebuilt = _db_messages_to_openai(db_msgs)
+
+        assert rebuilt[0] == {"role": "user", "content": "你好"}
+        assert rebuilt[1]["role"] == "assistant"
+        assert rebuilt[1]["content"] == "我来搜索"
+        assert rebuilt[1]["tool_calls"] is not None
+        assert rebuilt[2]["role"] == "tool"
+        assert rebuilt[2]["tool_call_id"] == "tc1"
+        assert rebuilt[2]["content"] == '{"ok": true}'
+
+    def test_backend_history_merged_with_new_message(self, db_session):
+        """修②：DB 已有历史时，新消息通过 list_by_conversation 已被存入并读出，
+        去重后不会重复加入传给 stream_chat 的 msgs。"""
+        from apps.api.routers.agent import _db_messages_to_openai, _new_messages_to_dicts
+
+        conv_repo = AgentConversationRepository(db_session)
+        msg_repo = AgentMessageRepository(db_session)
+        conv = conv_repo.create(title="t")
+        msg_repo.create(conversation_id=conv.id, role="user", content="前一轮")
+
+        # 模拟本次新增 user 消息已被前端发来并存入 DB
+        msg_repo.create(conversation_id=conv.id, role="user", content="这一轮")
+        db_msgs = msg_repo.list_by_conversation(conv.id)
+        history_msgs = _db_messages_to_openai(db_msgs)
+
+        # 前端发的新消息 dict（role/content），用 dataclass 模拟 AgentMessage schema
+        from packages.domain.schemas import AgentMessage
+
+        new_msgs = _new_messages_to_dicts([AgentMessage(role="user", content="这一轮")])
+        history_keys = {f"{m.get('role')}:{(m.get('content') or '')[:200]}" for m in history_msgs}
+        extra_new = [
+            m
+            for m in new_msgs
+            if f"{m.get('role')}:{(m.get('content') or '')[:200]}" not in history_keys
+        ]
+        merged = history_msgs + extra_new
+
+        # 去重后只出现一次"这一轮"
+        this_round_count = sum(1 for m in merged if m.get("content") == "这一轮")
+        assert this_round_count == 1, "新消息不应在合并后重复"
+
+
+# ---------- 修④：done 事件不重复存 assistant ----------
+
+
+class TestDoneDedupSave:
+    def test_stream_with_save_stores_assistant_only_once(self, db_session):
+        """模拟 SSE 流发多个 done 事件，assistant 只存一次。
+        这里直接测 agent_chat 的 stream_with_save 闭包内层逻辑（saved_done 标志）。
+        """
+        from apps.api.routers.agent import _parse_sse_events
+
+        # 验证 _parse_sse_events 能正确解析 done 事件
+        chunk = "event: done\ndata: {}\n\n"
+        events = _parse_sse_events(chunk)
+        assert events == [("done", {})]
+
+        # 模拟 saved_done 标志的去重语义
+        saved_done = False
+        saved_count = 0
+        for _ in range(3):  # 模拟三个 done 事件
+            if not saved_done:
+                saved_done = True
+                saved_count += 1
+        assert saved_count == 1, "多次 done 只应存一次 assistant"
+
+
+# ---------- 修⑤：confirm 后 tool/assistant 落 DB ----------
+
+
+class TestConfirmPersist:
+    def test_resolve_conversation_id_from_action_returns_conv_id(self, db_session):
+        """修⑤：confirm/reject 从 pending action 取 conversation_id 用于持久化。"""
+        from packages.storage.repositories import AgentPendingActionRepository
+
+        conv_repo = AgentConversationRepository(db_session)
+        conv = conv_repo.create(title="t")
+
+        pending_repo = AgentPendingActionRepository(db_session)
+        pending_repo.create(
+            action_id="act_test1",
+            tool_name="deep_read",
+            tool_args={"paper_id": "p1"},
+            tool_call_id="tc1",
+            conversation_id=conv.id,
+            conversation_state={"conversation": []},
+        )
+
+        # 用真实 session_scope（conftest 已 rebind）调用 resolve
+        from apps.api.routers.agent import _resolve_conversation_id_from_action
+
+        cid = _resolve_conversation_id_from_action("act_test1")
+        assert cid == conv.id
+
+
+# ---------- 修⑨：loop 对非法 tool_arguments JSON 不崩 ----------
+
+
+class TestLoopJsonSafety:
+    def test_handle_stream_event_invalid_json_does_not_raise(self):
+        """修⑨：tool_call 事件携带非法 JSON 参数时，_handle_stream_event 不应抛异常。"""
+        from packages.integrations.llm_client import StreamEvent
+
+        loop = _make_minimal_loop()
+        event = StreamEvent(
+            type="tool_call",
+            tool_call_id="tc_bad",
+            tool_name="search",
+            tool_arguments="{not valid json",  # 非法 JSON
+        )
+        tool_calls: list = []
+        sse = loop._handle_stream_event(event, text_buf="", tool_calls=tool_calls)
+        # 应返回 error 事件而非抛异常
+        assert sse is not None
+        assert '"error"' in sse or "error" in sse
+        # tool_calls 不应被污染（非法参数的 tool_call 被跳过）
+        assert len(tool_calls) == 0
+
+    def test_handle_stream_event_valid_json_appends_tool_call(self):
+        """对照：合法 JSON 仍正常追加到 tool_calls。"""
+        from packages.integrations.llm_client import StreamEvent
+
+        loop = _make_minimal_loop()
+        event = StreamEvent(
+            type="tool_call",
+            tool_call_id="tc_ok",
+            tool_name="search",
+            tool_arguments='{"q": "test"}',
+        )
+        tool_calls: list = []
+        sse = loop._handle_stream_event(event, text_buf="", tool_calls=tool_calls)
+        assert sse is None  # tool_call 不产生 SSE
+        assert len(tool_calls) == 1
+        assert tool_calls[0].arguments == {"q": "test"}
+
+
+# ---------- 修⑬：usage 回调传真实 provider ----------
+
+
+class TestUsageProvider:
+    def test_usage_callback_uses_llm_provider_not_event_model(self):
+        """修⑬：usage 回调第一个参数应是 llm.provider，之前错传 event.model。"""
+        from packages.integrations.llm_client import StreamEvent
+
+        loop = _make_minimal_loop(provider="zhipu", model_in_event="glm-4")
+        captured: list[tuple] = []
+
+        def on_usage(provider, model, in_tok, out_tok):
+            captured.append((provider, model))
+
+        loop._on_usage = on_usage
+        event = StreamEvent(
+            type="usage",
+            model="glm-4",
+            input_tokens=10,
+            output_tokens=5,
+        )
+        loop._handle_stream_event(event, text_buf="", tool_calls=[])
+        assert captured == [("zhipu", "glm-4")], (
+            "provider 应来自 llm.provider，model 来自 event.model"
+        )
+
+
+# ---------- helpers ----------
+
+
+def _make_minimal_loop(provider: str = "xiaomi", model_in_event: str = "mimo"):
+    """构造一个最小 StreamingAgentLoop，llm 只需暴露 provider 属性。"""
+    from packages.agent_core.loop import StreamingAgentLoop
+
+    llm = MagicMock()
+    llm.provider = provider
+
+    tools = []
+    tool_registry = []
+    execute_fn = MagicMock()
+    session_scope = MagicMock()
+
+    loop = StreamingAgentLoop(
+        llm=llm,
+        tools=tools,
+        tool_registry=tool_registry,
+        execute_fn=execute_fn,
+        session_scope=session_scope,
+    )
+    return loop