fix(agent): 会话真相源后端化 + loop 健壮性 + usage 修复

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,58 +131,57 @@ 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 "新对话"
             conv = conv_repo.create(title=title)
             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,23 +246,118 @@ 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,
     )
 
 
 @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,
     )

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;

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);