修复同一 Agent 多并发会话记忆混乱问题 (#458)

### 问题现象
同一 agent 开多个并发会话时（如 A1=查南京天气、A2=查北京天气），A2 在多轮 ReAct 执行中会"突然去查南京天气"，表现为 A1 会话的上下文泄漏到 A2 会话。

### 根因4：Agent 实例共享 + state 覆盖（确认，仅状态显示问题）
确认点：
- AgentService.java:83-90 agentInstances 按 (agentId, modelKey) 缓存，不含 conversationId
- AgentService.java:598-624 getOrBuildAgentForConversation 只按 (agentId, provider, model) 解析，不按 conversationId
- AgentService.java:530-545 withLifecycleFlux 无锁 ，A/B/C 完全并发
影响： A 完成设 IDLE → B 仍在运行但显示 IDLE → 状态显示错乱。 不会直接导致记忆串台 ，但对用户可见。

## 二、根因与症状匹配度总结
根因 匹配度 触发条件 串台通道 1. SessionSearchTool ★★★★★ LLM 多轮遇到困难时主动调用 session_search 返回并发兄弟会话消息 2. 审批重放无过滤 ★★★☆☆ Plan-Execute + 审批 + 并发 awaiting_approval 误取兄弟会话计划 3. 结构化记忆共享 ★★★☆☆ A 会话 LLM 主动 remember_structured 写入 prefetch 注入到 B 会话 4. Agent 实例共享 ★★☆☆☆ 任意并发 state 显示错乱（非记忆串台）

用户描述的"B突然去查南京天气"最可能是根因1 ——因为 system prompt 明确引导 LLM 在遇到困难时用 session_search 回忆历史，而 SQL 会返回并发兄弟会话的"南京天气"内容。

## 三、修复方案（按优先级排序）
### 方案1：修复 SessionSearchTool（最优先，直接命中症状）
改动点 A — SessionSearchTool 增加 ToolContext 参数，强制读取真实 conversationId：

SessionSearchTool.java:37-44

改动点 B — SessionSearchService 增加运行状态过滤，排除并发兄弟会话：

SessionSearchService.java:60-73 SQL 增加：

或更保守：只返回 status = 'completed' 的会话，排除 running / awaiting_approval 的并发会话。

风险评估： 改动 SQL 查询条件，不影响写入逻辑。 completed 会话才是真正的"历史对话"， running 会话是"正在进行"不应被搜索。功能上合理。

### 方案2：修复审批重放跨会话取计划（确凿 bug，必须修）
改动点 A — PlanningService.findAwaitingApprovalContext 增加 conversationId 参数：

PlanningService.java:228-232

改动点 B — 调用方传入 conversationId：

StateGraphPlanExecuteAgent.java:100

风险评估： 需确认 PlanEntity 有 conversationId 字段（从之前排查看应存在）。改动最小，仅加查询过滤，不影响其他逻辑。

### 方案3：结构化记忆引入会话级隔离（改动较大，需评估）
问题： 当前 ownerKey = user:<requesterId> ，3 会话共享。如果改为 conversation:<conversationId> ，会破坏"用户长期记忆跨会话共享"的设计意图（用户画像、偏好等应跨会话）。

建议方案： 不改 ownerKey 机制，而是在 StructuredMemoryTool.remember_structured 的 system prompt 说明中 明确限制 只记住"长期有效的事实"，临时任务结果（如天气查询）不应写入。或在 type 枚举中新增 transient 类型，该类型按 conversationId 隔离，会话结束自动清除。

风险评估： 改动较大，涉及记忆分层设计。建议作为中长期优化，本次先修方案1和2。

### 方案4：Agent 实例 state 按 conversationId 隔离（可选）
改动点： BaseAgent.java:33 AtomicReference<AgentState> state 改为 Map<String, AtomicReference<AgentState>> （按 conversationId）。

风险评估： 影响所有 getState() / setState() 调用点，改动面广。且这只是状态显示问题，不影响记忆串台。建议暂不修，或在前端按 conversationId 单独查询状态。

--------------------------------------------
本次完成bug1、2修复；3、4未动

mateclaw-server/src/main/java/vip/mate/agent/AgentGraphBuilder.java

@@ -1530,10 +1530,11 @@ public class AgentGraphBuilder {
                 adopting a KB article as the user's project.
 
                 ## Session Search
-                - `session_search(agentId, currentConversationId, mode, query, limit)` — search conversation history
+                - `session_search(agentId, mode, query, limit)` — search conversation history
                 - mode="recent": list recent conversations (titles, times, message counts)
                 - mode="search": keyword full-text search across past messages
                 - Use this to recall previous discussions, look up past decisions, or find context from earlier conversations
+                - Only completed sessions (not currently running) are included in results
 
                 ## Tool Usage Guidelines
                 When you have available tools, use them to access local system information, files, or execute commands.

mateclaw-server/src/main/java/vip/mate/agent/graph/plan/StateGraphPlanExecuteAgent.java

@@ -97,8 +97,8 @@ public class StateGraphPlanExecuteAgent extends BaseAgent implements StructuredS
             log.info("[{}] Plan-Execute replay stream: conversationId={}", agentName, conversationId);
             Map<String, Object> inputs = buildInitialState(userMessage, conversationId);
 
-            // 从 DB 恢复 awaiting_approval 状态的计划上下文
+            // 从 DB 恢复 awaiting_approval 状态的计划上下文（按 conversationId 过滤，避免并发会话误取）
-            PlanningService.PlanResumeContext ctx = planningService.findAwaitingApprovalContext();
+            PlanningService.PlanResumeContext ctx = planningService.findAwaitingApprovalContext(conversationId);
             if (ctx != null) {
                 inputs.put(PlanStateKeys.PLAN_ID, ctx.planId());
                 inputs.put(PlanStateKeys.PLAN_STEPS, ctx.steps());

mateclaw-server/src/main/java/vip/mate/memory/search/SessionSearchService.java

@@ -58,13 +58,17 @@ public class SessionSearchService {
 
     /**
      * List recent conversations for the given agent.
+     * Excludes the current conversation and any still-running sessions to prevent
+     * cross-conversation context leakage in concurrent multi-conversation scenarios.
      */
-    public List<Map<String, Object>> listRecent(Long agentId, int limit) {
+    public List<Map<String, Object>> listRecent(Long agentId, String currentConversationId, int limit) {
         int effectiveLimit = Math.min(Math.max(limit, 1), 50);
         String sql = """
                 SELECT conversation_id, title, message_count, last_active_time, create_time
                 FROM mate_conversation
                 WHERE agent_id = ? AND deleted = 0
+                  AND conversation_id != ?
+                  AND (stream_status IS NULL OR stream_status != 'running')
                 ORDER BY last_active_time DESC
                 LIMIT ?
                 """;
@@ -77,7 +81,7 @@ public class SessionSearchService {
             row.put("lastActiveTime", toLocalDateTime(rs.getTimestamp("last_active_time")));
             row.put("createTime", toLocalDateTime(rs.getTimestamp("create_time")));
             return row;
-        }, agentId, effectiveLimit);
+        }, agentId, currentConversationId != null ? currentConversationId : "", effectiveLimit);
     }
 
     // ==================== MySQL FULLTEXT ====================
@@ -93,6 +97,7 @@ public class SessionSearchService {
                 WHERE c.agent_id = ? AND m.conversation_id != ?
                   AND m.role IN ('user', 'assistant')
                   AND m.deleted = 0 AND c.deleted = 0
+                  AND (c.stream_status IS NULL OR c.stream_status != 'running')
                   AND MATCH(m.content) AGAINST(? IN NATURAL LANGUAGE MODE)
                 ORDER BY relevance DESC
                 LIMIT ?
@@ -116,6 +121,7 @@ public class SessionSearchService {
                 WHERE c.agent_id = ? AND m.conversation_id != ?
                   AND m.role IN ('user', 'assistant')
                   AND m.deleted = 0 AND c.deleted = 0
+                  AND (c.stream_status IS NULL OR c.stream_status != 'running')
                   AND LOWER(m.content) LIKE LOWER(CONCAT('%', ?, '%'))
                 ORDER BY m.create_time DESC
                 LIMIT ?

mateclaw-server/src/main/java/vip/mate/memory/search/SessionSearchTool.java

@@ -4,9 +4,11 @@ import cn.hutool.json.JSONObject;
 import cn.hutool.json.JSONUtil;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.ai.chat.model.ToolContext;
 import org.springframework.ai.tool.annotation.Tool;
 import org.springframework.ai.tool.annotation.ToolParam;
 import org.springframework.stereotype.Component;
+import vip.mate.agent.context.ChatOrigin;
 
 import java.util.List;
 import java.util.Map;
@@ -33,13 +35,14 @@ public class SessionSearchTool {
             - "recent"：列出最近的会话（标题、时间、消息数），不需要 query 参数
             - "search"：按关键词全文搜索消息内容，返回匹配的消息片段
             适用于回忆之前讨论过的话题、查找历史决策、检索之前的上下文。
+            注意：只会搜索已完成的会话，不会返回当前正在运行中的其他会话内容。
             """)
     public String session_search(
             @ToolParam(description = "当前 Agent 的 ID") Long agentId,
-            @ToolParam(description = "当前会话 ID（用于排除当前会话）") String currentConversationId,
             @ToolParam(description = "搜索模式：recent 或 search") String mode,
             @ToolParam(description = "搜索关键词（mode=search 时必填）", required = false) String query,
-            @ToolParam(description = "返回结果数量上限，默认 10", required = false) Integer limit) {
+            @ToolParam(description = "返回结果数量上限，默认 10", required = false) Integer limit,
+            ToolContext toolContext) {
 
         if (agentId == null) {
             return error("agentId 不能为空");
@@ -48,11 +51,20 @@ public class SessionSearchTool {
             mode = "recent";
         }
 
+        // 从 ToolContext 强制读取当前会话 ID（替代 LLM 自传，防止并发会话记忆混乱）
+        String currentConversationId = "";
+        if (toolContext != null) {
+            String fromOrigin = ChatOrigin.from(toolContext).conversationId();
+            if (fromOrigin != null && !fromOrigin.isEmpty()) {
+                currentConversationId = fromOrigin;
+            }
+        }
+
         int effectiveLimit = limit != null && limit > 0 ? limit : 10;
 
         try {
             if ("recent".equalsIgnoreCase(mode.trim())) {
-                return handleRecent(agentId, effectiveLimit);
+                return handleRecent(agentId, currentConversationId, effectiveLimit);
             } else if ("search".equalsIgnoreCase(mode.trim())) {
                 if (query == null || query.isBlank()) {
                     return error("mode=search 时 query 不能为空");
@@ -67,8 +79,8 @@ public class SessionSearchTool {
         }
     }
 
-    private String handleRecent(Long agentId, int limit) {
+    private String handleRecent(Long agentId, String currentConversationId, int limit) {
-        List<Map<String, Object>> sessions = sessionSearchService.listRecent(agentId, limit);
+        List<Map<String, Object>> sessions = sessionSearchService.listRecent(agentId, currentConversationId, limit);
 
         JSONObject result = new JSONObject();
         result.set("mode", "recent");

mateclaw-server/src/main/java/vip/mate/planning/service/PlanningService.java

@@ -224,12 +224,22 @@ public class PlanningService {
     /**
      * 审批 replay 上下文：找到最近一条 running 且含 awaiting_approval 步骤的计划，
      * 返回恢复图执行所需的全部状态。
+     * <p>
+     * 按 conversationId 过滤，防止并发会话误取兄弟会话的计划。
      */
-    public PlanResumeContext findAwaitingApprovalContext() {
+    public PlanResumeContext findAwaitingApprovalContext(String conversationId) {
-        PlanEntity plan = planMapper.selectOne(new LambdaQueryWrapper<PlanEntity>()
+        if (conversationId == null || conversationId.isBlank()) {
+            log.warn("[PlanningService] findAwaitingApprovalContext called without conversationId, " +
+                    "risk of cross-conversation plan pickup in concurrent scenarios");
+        }
+        LambdaQueryWrapper<PlanEntity> wrapper = new LambdaQueryWrapper<PlanEntity>()
                 .eq(PlanEntity::getStatus, "running")
                 .orderByDesc(PlanEntity::getCreateTime)
-                .last("LIMIT 1"));
+                .last("LIMIT 1");
+        if (conversationId != null && !conversationId.isBlank()) {
+            wrapper.eq(PlanEntity::getConversationId, conversationId);
+        }
+        PlanEntity plan = planMapper.selectOne(wrapper);
         if (plan == null) return null;
 
         List<SubPlanEntity> subPlans = subPlanMapper.selectList(