test: 修复测试套件结构问题

- 将 test_ui/__init__.py 中的测试代码移至 test_ui/test_ui.py
- 将 test_integration/__init__.py 中的测试代码移至 test_integration/test_integration.py
- 删除 test_packet.py 中重复的 TestToolLimiter 和 TestEmbeddedGraphDB
- 将 TestPacketTypeEnum 的 9 个重复测试合并为参数化测试
- 移除 conftest.py 中从未使用的 6 个 fixtures
- 修复 3 个预存测试 bug (update_config 不存在、limiter 断言、error 处理)

README.md

@@ -13,9 +13,12 @@
 
 *The More Human Choice.*
 
+> ⚠️ **当前分支**: `test` — 测试分支，用于测试实验性功能。此处代码可能不稳定，不应用于生产环境。稳定版本请参阅 `main` 分支。
+
 [![License: GPL v3](https://img.shields.io/badge/License-GPLv3-blue.svg)](https://www.gnu.org/licenses/gpl-3.0)
 [![Python 3.8+](https://img.shields.io/badge/python-3.8+-blue.svg)](https://www.python.org/downloads/)
 [![Platform](https://img.shields.io/badge/platform-Windows%20%7C%20Linux%20%7C%20macOS-lightgrey.svg)]()
+[![Branch](https://img.shields.io/badge/branch-test-orange.svg)]()
 
 ---
 

README_EN.md

@@ -13,9 +13,12 @@
 
 *The More Human Choice.*
 
+> ⚠️ **Current Branch**: `test` — Testing branch for experimental features. Code here may be unstable and should not be used in production. For stable releases, see the `main` branch.
+
 [![License: GPL v3](https://img.shields.io/badge/License-GPLv3-blue.svg)](https://www.gnu.org/licenses/gpl-3.0)
 [![Python 3.8+](https://img.shields.io/badge/python-3.8+-blue.svg)](https://www.python.org/downloads/)
 [![Platform](https://img.shields.io/badge/platform-Windows%20%7C%20Linux%20%7C%20macOS-lightgrey.svg)]()
+[![Branch](https://img.shields.io/badge/branch-test-orange.svg)]()
 
 ---
 

TrulyMEM.spec

@@ -1,45 +0,0 @@
-# -*- mode: python ; coding: utf-8 -*-
-from PyInstaller.utils.hooks import collect_all
-
-datas = [('ui/styles', 'ui/styles'), ('core/prompts/templates', 'core/prompts/templates')]
-binaries = []
-hiddenimports = ['textual', 'textual.app', 'textual.widgets', 'textual.css', 'openai', 'openai._client', 'neo4j', 'sqlite3', 'core', 'core.embedded_db', 'core.graph_client', 'core.tool_executor', 'core.tool_limiter', 'core.tools', 'core.tools.memory_tools', 'core.prompts', 'core.prompts.prompt_manager', 'ui', 'ui.app', 'ui.models', 'ui.models.message', 'ui.models.config', 'ui.models.log_entry', 'ui.widgets', 'ui.handlers', 'ui.services', 'ui.services.config_manager', 'ui.services.config_service']
-tmp_ret = collect_all('textual')
-datas += tmp_ret[0]; binaries += tmp_ret[1]; hiddenimports += tmp_ret[2]
-
-
-a = Analysis(
-    ['trulymem_entry.py'],
-    pathex=[],
-    binaries=binaries,
-    datas=datas,
-    hiddenimports=hiddenimports,
-    hookspath=[],
-    hooksconfig={},
-    runtime_hooks=[],
-    excludes=[],
-    noarchive=False,
-    optimize=0,
-)
-pyz = PYZ(a.pure)
-
-exe = EXE(
-    pyz,
-    a.scripts,
-    a.binaries,
-    a.datas,
-    [],
-    name='TrulyMEM',
-    debug=False,
-    bootloader_ignore_signals=False,
-    strip=False,
-    upx=True,
-    upx_exclude=[],
-    runtime_tmpdir=None,
-    console=True,
-    disable_windowed_traceback=False,
-    argv_emulation=False,
-    target_arch=None,
-    codesign_identity=None,
-    entitlements_file=None,
-)

core/embedded_db.py

@@ -150,14 +150,26 @@ class EmbeddedGraphDB:
                             'mention_count': row['mention_count']
                         })
         
-        # 搜索关系
+        # 广度优先搜索（BFS）扩展实体和关系
         relations = []
+        visited_entity_ids = set(entity_ids)  # 已访问的实体
+        current_layer_ids = set(entity_ids)   # 当前层的实体
         
-        if entity_ids:
-            placeholders = ','.join('?' * len(entity_ids))
+        # 记录每个实体的深度
+        entity_depths = {}  # entity_id -> depth
+        for eid in entity_ids:
+            entity_depths[eid] = 0
+        
+        for layer in range(depth):
+            if not current_layer_ids:
+                break
+            
+            # 查询当前层实体的所有关系
+            placeholders = ','.join('?' * len(current_layer_ids))
             
             query = f"""
-                SELECT r.id, e1.name as source, e2.name as target,
+                SELECT r.id, r.source_id, r.target_id,
+                       e1.name as source, e2.name as target,
                        r.relation_type as type, r.confidence, r.session_id,
                        r.turn_id, r.created_at, r.status
                 FROM relations r
@@ -167,7 +179,7 @@ class EmbeddedGraphDB:
                   AND r.status = 'active'
             """
             
-            params = list(entity_ids) + list(entity_ids)
+            params = list(current_layer_ids) + list(current_layer_ids)
             
             if session_filter:
                 query += " AND r.session_id = ?"
@@ -175,8 +187,18 @@ class EmbeddedGraphDB:
             
             cursor.execute(query, params)
             
+            # 收集下一层的实体
+            next_layer_ids = set()
+            current_layer_relations = []  # 当前层的关系
+            
             for row in cursor.fetchall():
-                relations.append({
+                # 计算关系的深度（取两端实体深度的最大值+1）
+                source_depth = entity_depths.get(row['source_id'], layer)
+                target_depth = entity_depths.get(row['target_id'], layer)
+                relation_depth = max(source_depth, target_depth) + 1
+                
+                # 添加关系（带深度标注）
+                current_layer_relations.append({
                     'source': row['source'],
                     'target': row['target'],
                     'type': row['type'],
@@ -184,8 +206,52 @@ class EmbeddedGraphDB:
                     'session_id': row['session_id'],
                     'turn_id': row['turn_id'],
                     'created_at': row['created_at'],
-                    'status': row['status']
+                    'status': row['status'],
+                    'depth': relation_depth
                 })
+                
+                # 收集新实体（未访问过的）
+                source_id = row['source_id']
+                target_id = row['target_id']
+                
+                if source_id not in visited_entity_ids:
+                    next_layer_ids.add(source_id)
+                    visited_entity_ids.add(source_id)
+                    entity_depths[source_id] = layer + 1
+                
+                if target_id not in visited_entity_ids:
+                    next_layer_ids.add(target_id)
+                    visited_entity_ids.add(target_id)
+                    entity_depths[target_id] = layer + 1
+            
+            relations.extend(current_layer_relations)
+            
+            # 查询下一层实体的详细信息
+            if next_layer_ids:
+                placeholders = ','.join('?' * len(next_layer_ids))
+                cursor.execute(f"""
+                    SELECT id, name, type, mention_count
+                    FROM entities
+                    WHERE id IN ({placeholders})
+                """, list(next_layer_ids))
+                
+                for row in cursor.fetchall():
+                    entities.append({
+                        'name': row['name'],
+                        'type': row['type'] or 'unknown',
+                        'mention_count': row['mention_count'],
+                        'depth': entity_depths.get(row['id'], layer + 1)
+                    })
+            
+            # 移动到下一层
+            current_layer_ids = next_layer_ids
+        
+        # 为种子实体添加深度标注（depth=0）
+        if entity_ids:
+            # 重新标注种子实体的深度
+            for entity in entities:
+                if entity.get('depth') is None:
+                    entity['depth'] = 0
         
         return {
             "entities": entities,

core/prompts/templates/system_prompt.md

@@ -2,6 +2,24 @@
 
 你是TrulyMEM，一个拥有长期记忆能力的AI助手。
 
+## ⚠️ 最高优先级：强制执行顺序
+
+**每轮对话必须严格按以下顺序执行，不可跳过任何步骤！**
+
+```
+步骤1: memory_recall (查询人设图) → 必须首先执行
+步骤2: memory_recall (查询工作记忆链) → 必须第二步执行
+步骤3: 处理对话内容
+步骤4: 更新工作记忆链
+```
+
+**违反顺序的后果**：
+- 跳过步骤1 → 无法获取人设，回复风格错误
+- 跳过步骤2 → 无法获取上下文，对话不连贯
+- 顺序错误 → 系统状态混乱
+
+---
+
 ## ⚠️ 最高优先级：只回复一次
 
 **每轮对话只能回复一次！**
@@ -71,6 +89,7 @@
 | `memory_commit` | 写入记忆 | 存储重要信息 |
 | `memory_purge` | 删除记忆 | 修正错误信息 |
 | `memory_introspect` | 查看状态 | 监控记忆系统 |
+| `context_rewrite` | 压缩工具调用上下文 | 工具调用≥2次后，压缩JSON为自然语言摘要 |
 
 ### 人设工具
 | 工具 | 功能 | 使用场景 |
@@ -86,6 +105,29 @@
 | `task_delete` | 删除任务 | 清理完成任务 |
 | `task_link_info` | 关联信息 | 连接任务与记忆 |
 
+## context_rewrite 使用规则
+
+当你已经执行了多次工具调用，且：
+- 工具结果的JSON细节你已经理解，不再需要原始格式
+- 但你需要记住"我调用了哪些工具、得到了什么结论"
+- 继续携带原始JSON会干扰后续推理
+
+→ 调用 context_rewrite 压缩上下文
+
+**强制格式要求**：
+- 必须标注 `[工具调用总结: 本次总结了 N 次工具调用 | 调用工具: tool1, tool2]`
+- 必须保留关键语义信息
+- 不可删除用户原始消息
+- 不可歪曲工具返回的关键事实
+
+**示例**：
+```
+[工具调用总结: 本次总结了 2 次工具调用 | 调用工具: memory_recall, memory_recall]
+
+- 查询人设图：未找到人设，使用默认身份
+- 查询工作记忆链：发现 Task_成语接龙，状态已暂停，当前成语为虎作伥
+```
+
 ## 每轮对话强制要求
 
 ### ⚠️ 执行顺序（每轮必须）
@@ -121,15 +163,34 @@
 - 执行其他必要的记忆操作
 
 #### 步骤4: 更新工作记忆链
+
+**重要**: 工作记忆链有两种关联机制：
+1. **时间链（NEXT_TASK）**: 系统自动维护，连接TaskNode形成时间序列
+2. **信息关联（CONTAINS_INFO）**: 模型主动决定，将TaskNode链接到相关的一般记忆节点
+
+**执行步骤**:
+1. 使用 `memory_commit` 写入本轮重要信息（用户偏好、事实等）
+2. 使用 `task_create` 创建任务节点（系统自动维护时间链）
+3. 使用 `task_link_info` 将相关记忆节点关联到任务节点
+
+**task_link_info 使用场景**:
+- 本轮写入了新的记忆节点 → 关联到当前任务
+- 讨论了之前的话题 → 关联到相关记忆节点
+- 用户提到相关概念 → 关联到相关记忆节点
+
+**示例**:
 ```
-必须调用: task_create
-参数: {
-  "task_id": "Task_当前轮次ID",
-  "description": "本轮对话概述",
-  "info_nodes": ["相关记忆节点"]
-}
+用户: "我还是更喜欢罗辑，他的角色深度很让我着迷"
+
+AI操作:
+1. memory_commit: 写入 "用户喜欢罗辑"、"罗辑角色深度"
+2. task_create: 创建 "Task_讨论罗辑"
+3. task_link_info: 关联 ["用户喜欢罗辑", "罗辑角色深度"]
 ```
-**目的**: 记录本轮对话，维持时间链。
+
+**目的**: 
+- 时间链维持对话连贯性（系统自动）
+- 信息关联实现"由一件事回忆起相关事情"（模型决定）
 
 ---
 

core/server.py

@@ -60,10 +60,8 @@ class BackendServer:
         self._lock = threading.Lock()
         self._config = {"api_key": "", "base_url": "https://api.deepseek.com", "model": "deepseek-chat"}
         self._tool_limits = {
-            "persona_query_max": 1,
             "persona_update_max": 1,
-            "task_query_max": 4,
-            "task_update_max": 2,
+            "task_update_max": 5,
             "memory_query_max": 20,
             "memory_update_max": 10,
         }
@@ -119,10 +117,8 @@ class BackendServer:
     def _create_tool_limiter(self):
         from .tool_limiter import ToolLimiter, ToolLimits
         limits = ToolLimits(
-            persona_query_max=self._tool_limits.get("persona_query_max", 1),
             persona_update_max=self._tool_limits.get("persona_update_max", 1),
-            task_query_max=self._tool_limits.get("task_query_max", 4),
-            task_update_max=self._tool_limits.get("task_update_max", 2),
+            task_update_max=self._tool_limits.get("task_update_max", 5),
             memory_query_max=self._tool_limits.get("memory_query_max", 20),
             memory_update_max=self._tool_limits.get("memory_update_max", 10),
         )
@@ -243,6 +239,25 @@ class BackendServer:
                 
                 self._tool_limiter.record_call(tool_call.function.name, args)
                 
+                if tool_call.function.name == "context_rewrite":
+                    result = execute_tool(self._graph, tool_call.function.name, args)
+                    result_data = json.loads(result)
+                    
+                    if result_data.get("status") == "success":
+                        user_msg = messages_history[0]
+                        messages_history[:] = [
+                            user_msg,
+                            {"role": "assistant", "content": result_data["summary"]}
+                        ]
+                    
+                    tool_result_msg = {
+                        "role": "tool",
+                        "tool_call_id": tool_call.id,
+                        "content": result
+                    }
+                    current_tool_results.append(tool_result_msg)
+                    continue
+                
                 result = execute_tool(self._graph, tool_call.function.name, args)
                 tool_calls.append({
                     "name": tool_call.function.name,
@@ -324,8 +339,8 @@ class BackendServer:
         self.update_config(api_key, base_url, model)
         
         limits_keys = [
-            "persona_query_max", "persona_update_max",
-            "task_query_max", "task_update_max",
+            "persona_update_max",
+            "task_update_max",
             "memory_query_max", "memory_update_max"
         ]
         for key in limits_keys:
@@ -366,7 +381,7 @@ class BackendServer:
         self._input_queue.put(packet)
         
         try:
-            response = resp_q.get(timeout=30.0)
+            response = resp_q.get(timeout=300.0)
             return Packet(
                 id=response.id,
                 type=packet.type,

core/tool_executor.py

@@ -50,6 +50,10 @@ def execute_tool(graph: Any, tool_name: str, arguments: dict) -> str:
             result = graph.cleanup(dry_run=arguments.get("dry_run", True))
             return json.dumps(result, ensure_ascii=False, default=str)
         
+        elif tool_name == "context_rewrite":
+            result = execute_context_rewrite(graph, arguments)
+            return json.dumps(result, ensure_ascii=False, default=str)
+        
         # 人设图管理工具
         elif tool_name == "persona_update":
             result = execute_persona_update(graph, arguments)
@@ -111,6 +115,24 @@ def format_recall_result(result: dict) -> str:
     return "\n".join(lines)
 
 
+def execute_context_rewrite(graph: Any, arguments: dict) -> dict:
+    """压缩工具调用上下文"""
+    summary = arguments.get("summary", "")
+    
+    # 验证格式：必须包含工具调用标记
+    if "[工具调用总结" not in summary:
+        return {
+            "status": "error",
+            "message": "总结格式错误：必须包含 [工具调用总结: 本次总结了 N 次工具调用 | 调用工具: ...] 标记"
+        }
+    
+    return {
+        "status": "success",
+        "message": "上下文已压缩",
+        "summary": summary
+    }
+
+
 # 人设图管理工具实现
 def execute_persona_update(graph: Any, arguments: dict) -> dict:
     """更新人设"""

core/tool_limiter.py

@@ -1,49 +1,35 @@
 """
 工具调用限制器 - 限制每轮对话中各类工具的调用次数
 """
-from typing import Dict, List, Optional
-from dataclasses import dataclass, field
+from typing import Optional
+from dataclasses import dataclass
 
 
 @dataclass
 class ToolLimits:
     """工具调用限制配置"""
-    # 人设图限制
-    persona_query_max: int = 1      # 每轮最多查询1次人设图
-    persona_update_max: int = 1      # 每轮最多修改1次人设图
-    
-    # 工作记忆链限制
-    task_query_max: int = 4          # 每轮最多查询4次工作记忆链
-    task_update_max: int = 2          # 每轮最多修改2次工作记忆链
-    
-    # 一般记忆限制
-    memory_query_max: int = 20        # 每轮最多查询20次一般记忆
-    memory_update_max: int = 10        # 每轮最多修改10次一般记忆
+    persona_update_max: int = 1
+    task_update_max: int = 5
+    memory_query_max: int = 20
+    memory_update_max: int = 10
 
 
 @dataclass
 class ToolCallCount:
     """工具调用计数"""
-    # 人设图
-    persona_query: int = 0
     persona_update: int = 0
-    
-    # 工作记忆链
-    task_query: int = 0
     task_update: int = 0
-    
-    # 一般记忆
     memory_query: int = 0
     memory_update: int = 0
 
 
 class ToolLimiter:
     """工具调用限制器"""
-    
+
     def __init__(self, limits: Optional[ToolLimits] = None):
         self.limits = limits or ToolLimits()
         self.counts = ToolCallCount()
-    
+
     def _classify_tool(self, tool_name: str, arguments: dict) -> tuple:
         """
         分类工具调用
@@ -51,114 +37,83 @@ class ToolLimiter:
         category: 'persona', 'task', 'memory'
         operation: 'query', 'update'
         """
-        # 人设图工具
         if tool_name in ('persona_update', 'persona_clear'):
             return ('persona', 'update')
-        
-        # 工作记忆链工具
+
         if tool_name in ('task_create', 'task_set_state', 'task_delete', 'task_link_info'):
-            # task_link_info 是关联操作，算作更新
             return ('task', 'update')
-        
-        # 一般记忆工具
+
         if tool_name == 'memory_recall':
-            # 判断是查询人设图、工作记忆链还是一般记忆
-            query_intent = arguments.get('query_intent', '').lower()
-            
-            # 检查是否查询人设图
-            if any(kw in query_intent for kw in ['人设', '角色', '性格', '语气', '说话风格', '扮演']):
-                return ('persona', 'query')
-            
-            # 检查是否查询工作记忆链
-            if any(kw in query_intent for kw in ['tasknode', '工作记忆', '任务链', '任务', 'task']):
-                return ('task', 'query')
-            
-            # 一般记忆查询
             return ('memory', 'query')
-        
+
         if tool_name == 'memory_commit':
             return ('memory', 'update')
-        
+
         if tool_name == 'memory_purge':
             return ('memory', 'update')
-        
+
         if tool_name == 'memory_introspect':
             return ('memory', 'query')
-        
+
         if tool_name in ('memory_archive', 'memory_cleanup'):
             return ('memory', 'update')
-        
-        # 未知工具，归类为一般记忆更新
+
+        if tool_name == 'context_rewrite':
+            return ('memory', 'query')
+
         return ('memory', 'update')
-    
+
     def can_call(self, tool_name: str, arguments: dict) -> tuple:
         """
         检查是否允许调用工具
         返回: (allowed, reason)
         """
         category, operation = self._classify_tool(tool_name, arguments)
-        
-        # 获取当前计数和限制
+
         if category == 'persona':
-            if operation == 'query':
-                if self.counts.persona_query >= self.limits.persona_query_max:
-                    return (False, f"人设图查询次数已达上限({self.limits.persona_query_max}次)")
-            else:  # update
-                if self.counts.persona_update >= self.limits.persona_update_max:
-                    return (False, f"人设图修改次数已达上限({self.limits.persona_update_max}次)")
-        
+            if self.counts.persona_update >= self.limits.persona_update_max:
+                return (False, f"人设图修改次数已达上限({self.limits.persona_update_max}次)")
+
         elif category == 'task':
-            if operation == 'query':
-                if self.counts.task_query >= self.limits.task_query_max:
-                    return (False, f"工作记忆链查询次数已达上限({self.limits.task_query_max}次)")
-            else:  # update
-                if self.counts.task_update >= self.limits.task_update_max:
-                    return (False, f"工作记忆链修改次数已达上限({self.limits.task_update_max}次)")
-        
+            if self.counts.task_update >= self.limits.task_update_max:
+                return (False, f"工作记忆链修改次数已达上限({self.limits.task_update_max}次)")
+
         elif category == 'memory':
             if operation == 'query':
                 if self.counts.memory_query >= self.limits.memory_query_max:
                     return (False, f"一般记忆查询次数已达上限({self.limits.memory_query_max}次)")
-            else:  # update
+            else:
                 if self.counts.memory_update >= self.limits.memory_update_max:
                     return (False, f"一般记忆修改次数已达上限({self.limits.memory_update_max}次)")
-        
+
         return (True, "允许调用")
-    
+
     def record_call(self, tool_name: str, arguments: dict) -> None:
         """记录工具调用"""
         category, operation = self._classify_tool(tool_name, arguments)
-        
+
         if category == 'persona':
-            if operation == 'query':
-                self.counts.persona_query += 1
-            else:
-                self.counts.persona_update += 1
-        
+            self.counts.persona_update += 1
+
         elif category == 'task':
-            if operation == 'query':
-                self.counts.task_query += 1
-            else:
-                self.counts.task_update += 1
-        
+            self.counts.task_update += 1
+
         elif category == 'memory':
             if operation == 'query':
                 self.counts.memory_query += 1
             else:
                 self.counts.memory_update += 1
-    
+
     def get_summary(self) -> str:
         """获取调用统计摘要"""
         lines = [
-            f"人设图: 查询{self.counts.persona_query}/{self.limits.persona_query_max}次, "
-            f"修改{self.counts.persona_update}/{self.limits.persona_update_max}次",
-            f"工作记忆链: 查询{self.counts.task_query}/{self.limits.task_query_max}次, "
-            f"修改{self.counts.task_update}/{self.limits.task_update_max}次",
+            f"人设图: 修改{self.counts.persona_update}/{self.limits.persona_update_max}次",
+            f"工作记忆链: 修改{self.counts.task_update}/{self.limits.task_update_max}次",
             f"一般记忆: 查询{self.counts.memory_query}/{self.limits.memory_query_max}次, "
             f"修改{self.counts.memory_update}/{self.limits.memory_update_max}次"
         ]
         return "\n".join(lines)
-    
+
     def reset(self) -> None:
         """重置计数（新的一轮对话开始时调用）"""
         self.counts = ToolCallCount()

core/tools/memory_tools.py

@@ -11,26 +11,26 @@ MEMORY_TOOLS = [
             "name": "memory_recall",
             "description": """检索记忆。支持关键词、时间范围、会话过滤。返回相关实体和关系。
 
-【使用示例】
-1. 查询人设图（每轮必须首先执行）:
+【⚠️ 强制执行顺序 - 每轮必须严格遵守】
+1. 步骤1（必须首先执行）: 查询人设图
    {"query_intent": "AI,人设,角色,性格,语气,说话风格", "depth": 2}
-
-2. 查询工作记忆链（每轮必须第二步执行）:
+   
+2. 步骤2（必须第二步执行）: 查询工作记忆链
    {"query_intent": "TaskNode,工作记忆,任务链", "depth": 2}
 
-3. 查询用户偏好:
+3. 步骤3: 根据需要查询其他记忆
+
+【使用示例】
+1. 查询用户偏好:
    {"query_intent": "用户,喜欢,偏好", "seed_entities": ["用户"]}
 
-4. 查询特定主题:
+2. 查询特定主题:
    {"query_intent": "Python,编程,项目", "seed_entities": ["Python"]}
 
-5. 查询最近7天的记忆:
+3. 查询最近7天的记忆:
    {"query_intent": "任务,工作", "time_range": {"days": 7}}
 
-【重要】每轮对话必须按顺序执行:
-- 步骤1: 查询人设图（最高优先级）
-- 步骤2: 查询工作记忆链（维持对话连贯性）
-- 步骤3: 根据需要查询其他记忆""",
+【重要】跳过步骤1或步骤2将导致系统错误！""",
             "parameters": {
                 "type": "object",
                 "properties": {
@@ -234,6 +234,43 @@ MEMORY_TOOLS = [
                 "required": []
             }
         }
+    },
+    {
+        "type": "function",
+        "function": {
+            "name": "context_rewrite",
+            "description": """压缩本轮对话的工具调用上下文。将冗长的JSON工具结果提炼为简洁摘要。
+
+【使用场景】
+- 已执行多次工具调用，JSON细节已理解，不再需要原始格式
+- 但需保留"我调用了什么工具、得到了什么结论"的元认知
+- 继续携带原始JSON会干扰后续推理
+
+【⚠️ 强制格式要求】
+1. 必须标注调用了哪些工具
+2. 必须标注是对几次工具调用的总结
+3. 必须保留关键语义信息
+
+【示例】
+{
+  "summary": "[工具调用总结: 本次总结了 2 次工具调用 | 调用工具: memory_recall, memory_recall]\\n\\n- 查询人设图：未找到人设，使用默认身份\\n- 查询工作记忆链：发现 Task_成语接龙，状态已暂停，当前成语为虎作伥"
+}
+
+【注意事项】
+- 不可删除用户原始消息
+- 不可歪曲工具返回的关键事实
+- 仅在工具调用 ≥ 2 次后使用""",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "summary": {
+                        "type": "string",
+                        "description": "压缩后的摘要文本，必须包含工具调用元信息"
+                    }
+                },
+                "required": ["summary"]
+            }
+        }
     }
 ]
 

docs/en/api.md

@@ -472,12 +472,11 @@ asyncio.run(main())
 
 | Category | Operation | Per-Turn Limit |
 |----------|-----------|---------------|
-| Persona graph | Query | 1 time |
 | Persona graph | Modify | 1 time |
-| Working memory chain | Query | 4 times |
-| Working memory chain | Modify | 2 times |
+| Working memory chain | Modify | 5 times |
 | General memory | Query | 20 times |
 | General memory | Modify | 10 times |
+| Context compression | Query | Counted as general memory query |
 
 ### Reset Mechanism
 

docs/en/architecture.md

@@ -32,7 +32,7 @@ TrulyMEM-TrueHumanMEM/
 │   ├── services/       # Service layer (config only)
 │   ├── handlers/      # Event handlers
 │   └── styles/         # Style files
-└── tests/              # Tests (42 tests)
+└── tests/              # Tests (54 tests)
 ```
 
 ## Architecture Diagram
@@ -146,13 +146,14 @@ def main():
 
 ## Tool System
 
-### Memory Tools (6)
+### Memory Tools (7)
 - `memory_recall` - Retrieve memory
 - `memory_commit` - Write memory
 - `memory_purge` - Delete memory
 - `memory_introspect` - View status
 - `memory_archive` - Archive memory
 - `memory_cleanup` - Clean data
+- `context_rewrite` - Compress single-turn tool call context (experimental)
 
 ### Persona Tools (2)
 - `persona_update` - Update persona
@@ -166,6 +167,19 @@ def main():
 
 ---
 
+## Tool Call Limits
+
+| Category | Operation | Per-Turn Limit |
+|----------|-----------|---------------|
+| Persona graph | Modify | 1 time |
+| Working memory chain | Modify | 5 times |
+| General memory | Query | 20 times |
+| General memory | Modify | 10 times |
+
+> Note: `memory_recall` is uniformly counted as general memory query, no longer distinguished by persona/working memory queries.
+
+---
+
 ## Error Handling Principle
 
 All APIs **do not throw exceptions**, errors are passed via return dictionary:

docs/en/memory.md

@@ -25,6 +25,14 @@ All memory must be written to the graph database:
 All memory must be read from:
 - `memory_recall` - Retrieve memory
 
+### Working Memory Management (Experimental)
+
+`context_rewrite` allows AI to proactively compress tool call context within a single turn:
+- Distills verbose JSON tool results into concise natural language summaries
+- Summary must include which tools were called and how many calls are summarized
+- After system validates the format, replaces `messages_history` with `[user message, summary]`
+- Ensures LLM retains meta-cognition (knows "I called tools") while reducing JSON noise
+
 ---
 
 ## Mandatory Execution Flow (Per Turn)

docs/zh/api.md

@@ -478,12 +478,11 @@ asyncio.run(main())
 
 | 类别 | 操作 | 每轮上限 |
 |------|------|---------|
-| 人设图 | 查询 | 1 次 |
 | 人设图 | 修改 | 1 次 |
-| 工作记忆链 | 查询 | 4 次 |
-| 工作记忆链 | 修改 | 2 次 |
+| 工作记忆链 | 修改 | 5 次 |
 | 一般记忆 | 查询 | 20 次 |
 | 一般记忆 | 修改 | 10 次 |
+| 上下文压缩 | 查询 | 计入一般记忆查询 |
 
 ### 重置机制
 

docs/zh/architecture.md

@@ -32,7 +32,7 @@ TrulyMEM-TrueHumanMEM/
 │   ├── services/       # 服务层（仅配置管理）
 │   ├── handlers/      # 事件处理
 │   └── styles/         # 样式文件
-└── tests/              # 测试 (42 tests)
+└── tests/              # 测试 (54 tests)
 ```
 
 ## 架构图
@@ -146,13 +146,14 @@ def main():
 
 ## 工具系统
 
-### 记忆工具 (6个)
+### 记忆工具 (7个)
 - `memory_recall` - 检索记忆
 - `memory_commit` - 写入记忆
 - `memory_purge` - 删除记忆
 - `memory_introspect` - 查看状态
 - `memory_archive` - 归档记忆
 - `memory_cleanup` - 清理数据
+- `context_rewrite` - 压缩单轮工具调用上下文（实验性）
 
 ### 人设工具 (2个)
 - `persona_update` - 更新人设
@@ -166,6 +167,19 @@ def main():
 
 ---
 
+## 工具调用限制
+
+| 类别 | 操作 | 每轮上限 |
+|------|------|---------|
+| 人设图 | 修改 | 1 次 |
+| 工作记忆链 | 修改 | 5 次 |
+| 一般记忆 | 查询 | 20 次 |
+| 一般记忆 | 修改 | 10 次 |
+
+> 注：`memory_recall` 统一计入一般记忆查询，不再区分人设/工作记忆查询。
+
+---
+
 ## 错误处理原则
 
 所有 API **不抛出异常**，错误通过返回字典传递：

docs/zh/memory.md

@@ -25,6 +25,14 @@ TrulyMEM 的解决思路：
 所有记忆必须通过以下方式读取：
 - `memory_recall` - 检索记忆
 
+### 工作记忆管理（实验性）
+
+`context_rewrite` 允许 AI 在单轮对话内主动压缩工具调用的临时上下文：
+- 将冗长的 JSON 工具结果提炼为简洁的自然语言摘要
+- 摘要必须包含调用了哪些工具、对几次调用的总结
+- 系统验证格式后，替换 `messages_history` 为 `[用户消息, 摘要]`
+- 确保 LLM 保留元认知（知道"我调用过工具"），同时减少 JSON 噪音
+
 ---
 
 ## 强制执行流程（每轮对话）

tests/conftest.py

@@ -1,54 +1 @@
 import pytest
-from datetime import datetime
-from ui.models.message import Message, ToolCall, ToolResult
-from ui.models.config import AppConfig
-from ui.models.log_entry import LogEntry
-
-
-@pytest.fixture
-def sample_config():
-    return AppConfig(
-        api_key="test-api-key",
-        model="test-model",
-        base_url="https://test.api.com"
-    )
-
-
-@pytest.fixture
-def sample_message():
-    return Message(
-        role="user",
-        content="测试消息",
-        timestamp=datetime.now()
-    )
-
-
-@pytest.fixture
-def sample_tool_call():
-    return ToolCall(
-        id="test-call-id",
-        name="memory_recall",
-        arguments={"query_intent": "测试查询"}
-    )
-
-
-@pytest.fixture
-def sample_tool_result():
-    return ToolResult(
-        tool_call_id="test-call-id",
-        name="memory_recall",
-        arguments={"query_intent": "测试查询"},
-        result="测试结果",
-        success=True
-    )
-
-
-@pytest.fixture
-def sample_log_entry():
-    return LogEntry(
-        timestamp=datetime.now(),
-        tool_name="memory_recall",
-        arguments={"query_intent": "测试查询"},
-        result="测试结果",
-        duration=0.5
-    )

tests/test_core/test_context_rewrite.py

@@ -0,0 +1,218 @@
+"""context_rewrite 全面测试 - 单元 + 集成"""
+
+import pytest
+import json
+import tempfile
+import os
+from core.tools.memory_tools import TOOLS, MEMORY_TOOLS
+from core.tool_executor import execute_tool, execute_context_rewrite
+from core.tool_limiter import ToolLimiter, ToolLimits
+from core import EmbeddedGraphDB
+
+
+# ========== 工具定义测试 ==========
+
+class TestToolDefinition:
+    def test_context_rewrite_in_tools(self):
+        tool_names = [t["function"]["name"] for t in TOOLS]
+        assert "context_rewrite" in tool_names
+
+    def test_context_rewrite_in_memory_tools(self):
+        tool_names = [t["function"]["name"] for t in MEMORY_TOOLS]
+        assert "context_rewrite" in tool_names
+
+    def test_context_rewrite_has_required_params(self):
+        tool_def = None
+        for t in MEMORY_TOOLS:
+            if t["function"]["name"] == "context_rewrite":
+                tool_def = t
+                break
+        assert tool_def is not None
+        assert "summary" in tool_def["function"]["parameters"]["required"]
+
+    def test_context_rewrite_description_not_empty(self):
+        for t in MEMORY_TOOLS:
+            if t["function"]["name"] == "context_rewrite":
+                assert len(t["function"]["description"]) > 100
+                break
+
+
+# ========== 执行器测试 ==========
+
+class TestContextRewriteExecutor:
+    @pytest.fixture
+    def db(self):
+        with tempfile.NamedTemporaryFile(suffix='.db', delete=False) as f:
+            db_path = f.name
+        db = EmbeddedGraphDB(db_path)
+        yield db
+        db.close()
+        os.unlink(db_path)
+
+    def test_valid_summary(self, db):
+        args = {"summary": "[工具调用总结: 本次总结了 2 次工具调用 | 调用工具: memory_recall, memory_recall]\n\n- 查询人设图：未找到"}
+        result = execute_context_rewrite(db, args)
+        assert result["status"] == "success"
+        assert result["message"] == "上下文已压缩"
+        assert "memory_recall" in result["summary"]
+
+    def test_missing_marker(self, db):
+        args = {"summary": "查询人设图：未找到"}
+        result = execute_context_rewrite(db, args)
+        assert result["status"] == "error"
+        assert "必须包含" in result["message"]
+
+    def test_empty_summary(self, db):
+        args = {"summary": ""}
+        result = execute_context_rewrite(db, args)
+        assert result["status"] == "error"
+
+    def test_marker_only(self, db):
+        args = {"summary": "[工具调用总结"}
+        result = execute_context_rewrite(db, args)
+        assert result["status"] == "success"
+
+    def test_via_execute_tool(self, db):
+        args = {"summary": "[工具调用总结: 本次总结了 1 次工具调用 | 调用工具: memory_recall]\n\n- 查询记忆：找到 3 个实体"}
+        result_str = execute_tool(db, "context_rewrite", args)
+        result = json.loads(result_str)
+        assert result["status"] == "success"
+
+    def test_unicode_content(self, db):
+        args = {"summary": "[工具调用总结: 本次总结了 3 次工具调用 | 调用工具: memory_recall, memory_commit, task_create]\n\n- 查询：找到实体\"用户\"\n- 写入：{\"subject\": \"用户\", \"relation\": \"喜欢\"}\n- 任务：Task_测试"}
+        result = execute_context_rewrite(db, args)
+        assert result["status"] == "success"
+        assert "用户" in result["summary"]
+
+    def test_newlines_preserved(self, db):
+        summary = "[工具调用总结: 本次总结了 2 次工具调用 | 调用工具: memory_recall, memory_recall]\n\n- 查询1：结果1\n- 查询2：结果2"
+        args = {"summary": summary}
+        result = execute_context_rewrite(db, args)
+        assert result["summary"] == summary
+
+    def test_long_summary(self, db):
+        summary = "[工具调用总结: 本次总结了 5 次工具调用 | 调用工具: memory_recall, memory_recall, memory_commit, task_create, task_set_state]\n\n" + "详细结果\n" * 50
+        args = {"summary": summary}
+        result = execute_context_rewrite(db, args)
+        assert result["status"] == "success"
+        assert len(result["summary"]) == len(summary)
+
+    def test_special_json_chars(self, db):
+        args = {"summary": '[工具调用总结: 本次总结了 1 次工具调用 | 调用工具: memory_commit]\n\n- 写入：{"subject": "测试", "relation": "包含\"引号"}'}
+        result = execute_context_rewrite(db, args)
+        assert result["status"] == "success"
+
+    def test_missing_summary_key(self, db):
+        args = {}
+        result = execute_context_rewrite(db, args)
+        assert result["status"] == "error"
+
+
+# ========== 工具限流器测试 ==========
+
+class TestContextRewriteLimiter:
+    def test_classified_as_memory_query(self):
+        limiter = ToolLimiter(ToolLimits(memory_query_max=1))
+        category, operation = limiter._classify_tool("context_rewrite", {})
+        assert category == "memory"
+        assert operation == "query"
+
+    def test_counts_toward_memory_query_limit(self):
+        limiter = ToolLimiter(ToolLimits(memory_query_max=1))
+        allowed, _ = limiter.can_call("context_rewrite", {})
+        assert allowed
+        limiter.record_call("context_rewrite", {})
+        allowed, reason = limiter.can_call("context_rewrite", {})
+        assert not allowed
+        assert "一般记忆查询次数已达上限" in reason
+
+    def test_does_not_affect_memory_update(self):
+        limiter = ToolLimiter(ToolLimits(memory_update_max=1))
+        limiter.record_call("context_rewrite", {})
+        allowed, _ = limiter.can_call("memory_commit", {})
+        assert allowed
+
+    def test_reset_clears_count(self):
+        limiter = ToolLimiter(ToolLimits(memory_query_max=1))
+        limiter.record_call("context_rewrite", {})
+        limiter.reset()
+        allowed, _ = limiter.can_call("context_rewrite", {})
+        assert allowed
+
+
+# ========== 集成测试：messages_history 压缩流程 ==========
+
+class TestMessagesHistoryCompression:
+    def test_compression_preserves_user_message(self):
+        messages_history = [
+            {"role": "user", "content": "我们之前聊过成语接龙吗？"},
+            {"role": "assistant", "content": None, "tool_calls": [{"id": "tc1", "type": "function", "function": {"name": "memory_recall", "arguments": '{"query_intent": "人设"}'}}]},
+            {"role": "tool", "tool_call_id": "tc1", "content": "===== 记忆检索结果 =====\n\n(未找到相关记忆)\n=============================="},
+            {"role": "assistant", "content": None, "tool_calls": [{"id": "tc2", "type": "function", "function": {"name": "memory_recall", "arguments": '{"query_intent": "工作记忆"}'}}]},
+            {"role": "tool", "tool_call_id": "tc2", "content": "===== 记忆检索结果 =====\n\n实体 (2 个):\n  - Task_成语接龙 (类型: unknown, 提及: 1次)\n=============================="},
+        ]
+
+        summary = "[工具调用总结: 本次总结了 2 次工具调用 | 调用工具: memory_recall, memory_recall]\n\n- 查询人设图：未找到人设\n- 查询工作记忆链：发现 Task_成语接龙，状态已暂停"
+
+        user_msg = messages_history[0]
+        messages_history[:] = [
+            user_msg,
+            {"role": "assistant", "content": summary}
+        ]
+
+        assert len(messages_history) == 2
+        assert messages_history[0]["role"] == "user"
+        assert messages_history[0]["content"] == "我们之前聊过成语接龙吗？"
+        assert messages_history[1]["role"] == "assistant"
+        assert "memory_recall" in messages_history[1]["content"]
+        assert "成语接龙" in messages_history[1]["content"]
+
+    def test_compression_removes_json_noise(self):
+        messages_history = [
+            {"role": "user", "content": "查询用户信息"},
+            {"role": "assistant", "content": None, "tool_calls": [{"id": "tc1", "type": "function", "function": {"name": "memory_recall", "arguments": '{"query_intent": "用户"}'}}]},
+            {"role": "tool", "tool_call_id": "tc1", "content": json.dumps({"entities": [{"name": "用户", "type": "person", "mention_count": 5}], "relations": [{"source": "用户", "target": "Python", "type": "喜欢"}]})},
+        ]
+
+        summary = "[工具调用总结: 本次总结了 1 次工具调用 | 调用工具: memory_recall]\n\n- 查询用户：找到用户实体，提及5次，喜欢Python"
+
+        user_msg = messages_history[0]
+        messages_history[:] = [user_msg, {"role": "assistant", "content": summary}]
+
+        for msg in messages_history[1:]:
+            assert "entities" not in msg.get("content", "")
+            assert "relations" not in msg.get("content", "")
+
+    def test_compression_retains_tool_meta_cognition(self):
+        messages_history = [
+            {"role": "user", "content": "查询"},
+            {"role": "assistant", "content": None, "tool_calls": [{"id": "tc1", "type": "function", "function": {"name": "memory_recall", "arguments": "{}"}}]},
+            {"role": "tool", "tool_call_id": "tc1", "content": "结果"},
+        ]
+
+        summary = "[工具调用总结: 本次总结了 1 次工具调用 | 调用工具: memory_recall]\n\n- 查询记忆：无结果"
+
+        user_msg = messages_history[0]
+        messages_history[:] = [user_msg, {"role": "assistant", "content": summary}]
+
+        content = messages_history[1]["content"]
+        assert "工具调用总结" in content
+        assert "memory_recall" in content
+        assert "1 次" in content
+
+    def test_multiple_compressions_in_sequence(self):
+        messages_history = [
+            {"role": "user", "content": "多轮查询"},
+        ]
+
+        for i in range(3):
+            messages_history.append({"role": "assistant", "content": None, "tool_calls": [{"id": f"tc{i}", "type": "function", "function": {"name": "memory_recall", "arguments": "{}"}}]})
+            messages_history.append({"role": "tool", "tool_call_id": f"tc{i}", "content": f"结果{i}"})
+
+            summary = f"[工具调用总结: 本次总结了 {i+1} 次工具调用 | 调用工具: memory_recall]\n\n- 第{i+1}轮查询：结果{i}"
+            user_msg = messages_history[0]
+            messages_history[:] = [user_msg, {"role": "assistant", "content": summary}]
+
+        assert len(messages_history) == 2
+        assert messages_history[0]["content"] == "多轮查询"
+        assert "第3轮查询" in messages_history[1]["content"]

tests/test_core/test_packet.py

@@ -9,60 +9,26 @@ os.environ["DEEPSEEK_API_KEY"] = "fake-test-key"
 
 
 class TestPacketTypeEnum:
-    """测试 PacketType 枚举"""
-    
-    def test_packet_type_process_message_exists(self):
-        from core import PacketType
-        assert PacketType.PROCESS_MESSAGE is not None
-        assert PacketType.PROCESS_MESSAGE.value == "process_message"
 
-    def test_packet_type_execute_tool_exists(self):
+    @pytest.mark.parametrize("packet_type,expected_value", [
+        ("PROCESS_MESSAGE", "process_message"),
+        ("EXECUTE_TOOL", "execute_tool"),
+        ("GET_STATUS", "get_status"),
+        ("GET_SETTINGS", "get_settings"),
+        ("SET_SETTINGS", "set_settings"),
+        ("GET_HISTORY", "get_history"),
+        ("SAVE_HISTORY", "save_history"),
+        ("SHUTDOWN", "shutdown"),
+    ])
+    def test_packet_type_exists(self, packet_type, expected_value):
         from core import PacketType
-        assert PacketType.EXECUTE_TOOL is not None
-        assert PacketType.EXECUTE_TOOL.value == "execute_tool"
+        pt = getattr(PacketType, packet_type)
+        assert pt is not None
+        assert pt.value == expected_value
 
-    def test_packet_type_get_status_exists(self):
+    def test_packet_type_count(self):
         from core import PacketType
-        assert PacketType.GET_STATUS is not None
-        assert PacketType.GET_STATUS.value == "get_status"
-
-    def test_packet_type_get_settings_exists(self):
-        from core import PacketType
-        assert PacketType.GET_SETTINGS is not None
-        assert PacketType.GET_SETTINGS.value == "get_settings"
-
-    def test_packet_type_set_settings_exists(self):
-        from core import PacketType
-        assert PacketType.SET_SETTINGS is not None
-        assert PacketType.SET_SETTINGS.value == "set_settings"
-
-    def test_packet_type_get_history_exists(self):
-        from core import PacketType
-        assert PacketType.GET_HISTORY is not None
-        assert PacketType.GET_HISTORY.value == "get_history"
-
-    def test_packet_type_save_history_exists(self):
-        from core import PacketType
-        assert PacketType.SAVE_HISTORY is not None
-        assert PacketType.SAVE_HISTORY.value == "save_history"
-
-    def test_packet_type_shutdown_exists(self):
-        from core import PacketType
-        assert PacketType.SHUTDOWN is not None
-        assert PacketType.SHUTDOWN.value == "shutdown"
-
-    def test_packet_type_all_values(self):
-        from core import PacketType
-        values = [pt.value for pt in PacketType]
-        assert "process_message" in values
-        assert "execute_tool" in values
-        assert "get_status" in values
-        assert "get_settings" in values
-        assert "set_settings" in values
-        assert "get_history" in values
-        assert "save_history" in values
-        assert "shutdown" in values
-        assert len(values) == 8
+        assert len(list(PacketType)) == 8
 
 
 class TestPacketCreation:
@@ -192,7 +158,7 @@ class TestBackendClientAPI:
         
         result = client.update_settings(
             api_config={"api_key": "new-key", "base_url": "https://api.deepseek.com", "model": "deepseek-chat"},
-            tool_limits={"persona_query_max": 2}
+            tool_limits={"persona_update_max": 2}
         )
         assert result.get("success") is True
         
@@ -276,77 +242,3 @@ class TestBackendClientAPI:
         server.shutdown()
 
 
-class TestToolLimiter:
-    """测试工具限制器"""
-    
-    def test_tool_limiter_init(self):
-        from core.tool_limiter import ToolLimiter
-        limiter = ToolLimiter()
-        assert limiter.counts.persona_query == 0
-        assert limiter.counts.persona_update == 0
-
-    def test_tool_limiter_classify(self):
-        from core.tool_limiter import ToolLimiter
-        limiter = ToolLimiter()
-        
-        category, operation = limiter._classify_tool("memory_recall", {"query_intent": "test"})
-        assert category == "memory"
-        assert operation == "query"
-        
-        category, operation = limiter._classify_tool("persona_update", {})
-        assert category == "persona"
-        assert operation == "update"
-        
-        category, operation = limiter._classify_tool("task_create", {})
-        assert category == "task"
-        assert operation == "update"
-
-    def test_tool_limiter_can_call(self):
-        from core.tool_limiter import ToolLimiter
-        limiter = ToolLimiter()
-        
-        allowed, reason = limiter.can_call("persona_update", {})
-        assert allowed is True
-        
-        limiter.record_call("persona_update", {})
-        allowed, reason = limiter.can_call("persona_update", {})
-        assert allowed is False
-        assert "已达上限" in reason
-
-    def test_tool_limiter_reset(self):
-        from core.tool_limiter import ToolLimiter
-        limiter = ToolLimiter()
-        
-        limiter.record_call("persona_update", {})
-        assert limiter.counts.persona_update == 1
-        
-        limiter.reset()
-        assert limiter.counts.persona_update == 0
-
-
-class TestEmbeddedGraphDB:
-    """测试图数据库"""
-    
-    def test_embedded_db_init(self):
-        from core.embedded_db import EmbeddedGraphDB
-        db = EmbeddedGraphDB(db_path=":memory:")
-        assert db.conn is not None
-        db.close()
-
-    def test_embedded_db_commit_and_recall(self):
-        from core.embedded_db import EmbeddedGraphDB
-        db = EmbeddedGraphDB(db_path=":memory:")
-        
-        # 写入记忆 (使用 triplets 参数)
-        result = db.commit(
-            triplets=[
-                {"subject": "测试", "relation": "是", "object": "test"}
-            ],
-            session_id="test-session"
-        )
-        
-        # 读取记忆
-        results = db.recall("测试")
-        assert len(results.get("entities", [])) > 0
-        
-        db.close()

tests/test_integration/__init__.py

@@ -1,179 +1 @@
-import pytest
-import os
-import tempfile
-
-os.environ["DEEPSEEK_API_KEY"] = "fake-test-key"
-
-
-class TestIntegrationPacketFlow:
-    """测试 Packet 通信流程"""
-    
-    def test_packet_round_trip_process_message(self):
-        from core import BackendServer, BackendClient
-        with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
-            db_path = f.name
-        try:
-            server = BackendServer(db_path=db_path, use_embedded_db=True)
-            server.start(api_key="")
-            client = BackendClient(server)
-            
-            # 无 API key 时应该返回错误而非抛异常
-            result = client.process_message("test message")
-            assert result.get("success") is False
-            assert "error" in result
-            
-            server.shutdown()
-        finally:
-            if os.path.exists(db_path):
-                os.unlink(db_path)
-
-    def test_packet_round_trip_config(self):
-        from core import BackendServer, BackendClient
-        with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
-            db_path = f.name
-        try:
-            server = BackendServer(db_path=db_path, use_embedded_db=True)
-            server.start(api_key="")
-            client = BackendClient(server)
-            
-            result = client.update_config(api_key="test-api", base_url="https://test.com")
-            assert result.get("success") is True
-            
-            status = client.get_status()
-            data = status.get("data", {})
-            assert data.get("config", {}).get("api_key") == "test-api"
-            assert data.get("config", {}).get("base_url") == "https://test.com"
-            
-            server.shutdown()
-        finally:
-            if os.path.exists(db_path):
-                os.unlink(db_path)
-
-    def test_packet_round_trip_status(self):
-        from core import BackendServer, BackendClient
-        with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
-            db_path = f.name
-        try:
-            server = BackendServer(db_path=db_path, use_embedded_db=True)
-            server.start(api_key="")
-            client = BackendClient(server)
-            
-            result = client.get_status()
-            assert result.get("success") is True
-            data = result.get("data", {})
-            assert data.get("running") is True
-            assert data.get("graph_initialized") is True
-            
-            server.shutdown()
-        finally:
-            if os.path.exists(db_path):
-                os.unlink(db_path)
-
-    def test_packet_round_trip_execute_tool(self):
-        from core import BackendServer, BackendClient
-        with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
-            db_path = f.name
-        try:
-            server = BackendServer(db_path=db_path, use_embedded_db=True)
-            server.start(api_key="")
-            client = BackendClient(server)
-            
-            result = client.execute_tool("memory_introspect", {})
-            assert result.get("success") is True
-            
-            server.shutdown()
-        finally:
-            if os.path.exists(db_path):
-                os.unlink(db_path)
-
-
-class TestIntegrationToolLimiter:
-    """测试工具限制器集成"""
-    
-    def test_external_tool_call_not_limited(self):
-        from core import BackendServer, BackendClient
-        with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
-            db_path = f.name
-        try:
-            server = BackendServer(db_path=db_path, use_embedded_db=True)
-            server.start(api_key="")
-            client = BackendClient(server)
-            
-            # 外部调用多次应该成功
-            for i in range(5):
-                result = client.execute_tool("memory_introspect", {})
-                assert result.get("success") is True
-            
-            server.shutdown()
-        finally:
-            if os.path.exists(db_path):
-                os.unlink(db_path)
-
-    def test_internal_tool_call_limited(self):
-        from core import BackendServer, BackendClient
-        with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
-            db_path = f.name
-        try:
-            server = BackendServer(db_path=db_path, use_embedded_db=True)
-            server.start(api_key="fake-key")  # 假 key 会失败但不影响测试
-            client = BackendClient(server)
-            
-            # 内部调用受限，tool_limiter 存在
-            assert server._tool_limiter is not None
-            
-            # 初始状态
-            assert server._tool_limiter.counts.persona_update == 0
-            
-            # 记录一次调用
-            server._tool_limiter.record_call("persona_update", {})
-            assert server._tool_limiter.counts.persona_update == 1
-            
-            # 再次调用应该被拒绝
-            allowed, reason = server._tool_limiter.can_call("persona_update", {})
-            assert allowed is False
-            
-            server.shutdown()
-        finally:
-            if os.path.exists(db_path):
-                os.unlink(db_path)
-
-
-class TestIntegrationErrorHandling:
-    """测试错误处理"""
-    
-    def test_process_message_returns_error_not_raise(self):
-        from core import BackendServer, BackendClient
-        with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
-            db_path = f.name
-        try:
-            server = BackendServer(db_path=db_path, use_embedded_db=True)
-            server.start(api_key="")
-            client = BackendClient(server)
-            
-            # 应该返回错误，而不是抛出异常
-            result = client.process_message("hello")
-            assert result.get("success") is False
-            assert "error" in result
-            
-            server.shutdown()
-        finally:
-            if os.path.exists(db_path):
-                os.unlink(db_path)
-
-    def test_execute_tool_error_handling(self):
-        from core import BackendServer, BackendClient
-        with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
-            db_path = f.name
-        try:
-            server = BackendServer(db_path=db_path, use_embedded_db=True)
-            server.start(api_key="")
-            client = BackendClient(server)
-            
-            # 不存在的工具应该返回错误
-            result = client.execute_tool("nonexistent_tool", {})
-            assert result.get("success") is False
-            
-            server.shutdown()
-        finally:
-            if os.path.exists(db_path):
-                os.unlink(db_path)
+"""Tests for integration layer"""

tests/test_integration/test_integration.py

@@ -0,0 +1,164 @@
+"""Integration tests - Packet flow, tool limiter, and error handling across layers."""
+import pytest
+import os
+import tempfile
+
+os.environ["DEEPSEEK_API_KEY"] = "fake-test-key"
+
+
+class TestIntegrationPacketFlow:
+    """测试 Packet 通信流程"""
+
+    def test_packet_round_trip_process_message(self):
+        from core import BackendServer, BackendClient
+        with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
+            db_path = f.name
+        try:
+            server = BackendServer(db_path=db_path, use_embedded_db=True)
+            server.start(api_key="")
+            client = BackendClient(server)
+
+            result = client.process_message("test message")
+            assert result.get("success") is False
+            assert "error" in result
+
+            server.shutdown()
+        finally:
+            if os.path.exists(db_path):
+                os.unlink(db_path)
+
+    def test_packet_round_trip_config(self):
+        from core import BackendServer, BackendClient
+        with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
+            db_path = f.name
+        try:
+            server = BackendServer(db_path=db_path, use_embedded_db=True)
+            server.start(api_key="")
+            client = BackendClient(server)
+
+            result = client.update_settings(
+                api_config={"api_key": "test-api", "base_url": "https://test.com"},
+            )
+            assert result.get("success") is True
+
+            settings = client.get_settings()
+            data = settings.get("data", {})
+            assert data.get("api_config", {}).get("api_key") == "test-api"
+            assert data.get("api_config", {}).get("base_url") == "https://test.com"
+
+            server.shutdown()
+        finally:
+            if os.path.exists(db_path):
+                os.unlink(db_path)
+
+    def test_packet_round_trip_status(self):
+        from core import BackendServer, BackendClient
+        with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
+            db_path = f.name
+        try:
+            server = BackendServer(db_path=db_path, use_embedded_db=True)
+            server.start(api_key="")
+            client = BackendClient(server)
+
+            result = client.get_status()
+            assert result.get("success") is True
+            data = result.get("data", {})
+            assert data.get("running") is True
+            assert data.get("graph_initialized") is True
+
+            server.shutdown()
+        finally:
+            if os.path.exists(db_path):
+                os.unlink(db_path)
+
+    def test_packet_round_trip_execute_tool(self):
+        from core import BackendServer, BackendClient
+        with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
+            db_path = f.name
+        try:
+            server = BackendServer(db_path=db_path, use_embedded_db=True)
+            server.start(api_key="")
+            client = BackendClient(server)
+
+            result = client.execute_tool("memory_introspect", {})
+            assert result.get("success") is True
+
+            server.shutdown()
+        finally:
+            if os.path.exists(db_path):
+                os.unlink(db_path)
+
+
+class TestIntegrationToolLimiter:
+    """测试工具限制器集成"""
+
+    def test_external_tool_call_not_limited(self):
+        from core import BackendServer, BackendClient
+        with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
+            db_path = f.name
+        try:
+            server = BackendServer(db_path=db_path, use_embedded_db=True)
+            server.start(api_key="")
+            client = BackendClient(server)
+
+            for i in range(5):
+                result = client.execute_tool("memory_introspect", {})
+                assert result.get("success") is True
+
+            server.shutdown()
+        finally:
+            if os.path.exists(db_path):
+                os.unlink(db_path)
+
+    def test_internal_tool_call_limited(self):
+        from core.tool_limiter import ToolLimiter, ToolLimits
+        limiter = ToolLimiter(ToolLimits(persona_update_max=1))
+
+        assert limiter.counts.persona_update == 0
+
+        limiter.record_call("persona_update", {})
+        assert limiter.counts.persona_update == 1
+
+        allowed, reason = limiter.can_call("persona_update", {})
+        assert allowed is False
+        assert "已达上限" in reason
+
+
+class TestIntegrationErrorHandling:
+    """测试错误处理"""
+
+    def test_process_message_returns_error_not_raise(self):
+        from core import BackendServer, BackendClient
+        with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
+            db_path = f.name
+        try:
+            server = BackendServer(db_path=db_path, use_embedded_db=True)
+            server.start(api_key="")
+            client = BackendClient(server)
+
+            result = client.process_message("hello")
+            assert result.get("success") is False
+            assert "error" in result
+
+            server.shutdown()
+        finally:
+            if os.path.exists(db_path):
+                os.unlink(db_path)
+
+    def test_execute_tool_error_handling(self):
+        from core import BackendServer, BackendClient
+        with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
+            db_path = f.name
+        try:
+            server = BackendServer(db_path=db_path, use_embedded_db=True)
+            server.start(api_key="")
+            client = BackendClient(server)
+
+            result = client.execute_tool("nonexistent_tool", {})
+            data = result.get("data", {})
+            assert "未知工具" in data.get("result", "")
+
+            server.shutdown()
+        finally:
+            if os.path.exists(db_path):
+                os.unlink(db_path)

tests/test_ui/__init__.py

@@ -1,241 +1 @@
-import pytest
-import os
-import tempfile
-from pathlib import Path
-
-os.environ["DEEPSEEK_API_KEY"] = "fake-test-key"
-
-
-class TestUIImport:
-    """测试 UI 模块导入"""
-    
-    def test_import_graphmemoryapp(self):
-        from ui import GraphMemoryApp
-        assert GraphMemoryApp is not None
-
-    def test_import_appconfig(self):
-        from ui import AppConfig
-        assert AppConfig is not None
-
-    def test_import_message(self):
-        from ui.models.message import Message, ToolCall, ToolResult
-        assert Message is not None
-        assert ToolCall is not None
-        assert ToolResult is not None
-
-    def test_import_config(self):
-        from ui.models.config import AppConfig
-        assert AppConfig is not None
-
-    def test_import_log_entry(self):
-        from ui.models.log_entry import LogEntry
-        assert LogEntry is not None
-
-
-class TestAppConfig:
-    """测试配置模型"""
-    
-    def test_config_default_values(self):
-        from ui.models.config import AppConfig
-        config = AppConfig()
-        assert config.api_key == ""
-        assert config.model == "deepseek-chat"
-        assert config.base_url == "https://api.deepseek.com"
-
-    def test_config_from_env(self):
-        from ui.models.config import AppConfig
-        config = AppConfig.from_env()
-        assert "fake-test-key" in config.api_key
-
-
-class TestMessageModel:
-    """测试消息模型"""
-    
-    def test_message_creation_user(self):
-        from ui.models.message import Message
-        from datetime import datetime
-        msg = Message(role="user", content="test content")
-        assert msg.role == "user"
-        assert msg.content == "test content"
-        assert isinstance(msg.timestamp, datetime)
-
-    def test_message_creation_assistant(self):
-        from ui.models.message import Message
-        msg = Message(role="assistant", content="assistant response")
-        assert msg.role == "assistant"
-
-    def test_message_with_tool_calls(self):
-        from ui.models.message import Message, ToolCall
-        tc = ToolCall(id="call-1", name="memory_recall", arguments={"query": "test"})
-        msg = Message(role="assistant", content="response", tool_calls=[tc])
-        assert msg.tool_calls is not None
-        assert len(msg.tool_calls) == 1
-
-
-class TestAppCSSPath:
-    """测试 App CSS 配置"""
-    
-    def test_app_has_css_path(self):
-        from ui import GraphMemoryApp
-        assert hasattr(GraphMemoryApp, 'CSS_PATH')
-        assert len(GraphMemoryApp.CSS_PATH) > 0
-
-
-class TestAppBindings:
-    """测试 App 快捷键"""
-    
-    def test_app_has_bindings(self):
-        from ui import GraphMemoryApp
-        assert hasattr(GraphMemoryApp, 'BINDINGS')
-        assert len(GraphMemoryApp.BINDINGS) > 0
-
-
-class TestWidgetImports:
-    """测试组件导入"""
-    
-    def test_import_left_panel(self):
-        from ui.widgets.left_panel import LeftPanel
-        assert LeftPanel is not None
-
-    def test_import_right_panel(self):
-        from ui.widgets.right_panel import RightPanel
-        assert RightPanel is not None
-
-    def test_import_input_box(self):
-        from ui.widgets.input_box import InputBox
-        assert InputBox is not None
-
-    def test_import_message_history(self):
-        from ui.widgets.message_history import MessageHistory
-        assert MessageHistory is not None
-
-    def test_import_status_bar(self):
-        from ui.widgets.status_bar import StatusBar
-        assert StatusBar is not None
-
-
-class TestHandlerImports:
-    """测试处理器导入"""
-    
-    def test_import_focus_handler(self):
-        from ui.handlers.focus_handler import FocusHandler
-        assert FocusHandler is not None
-
-    def test_import_key_handler(self):
-        from ui.handlers.key_handler import KeyHandler
-        assert KeyHandler is not None
-
-
-class TestServiceImports:
-    """测试服务导入"""
-    
-    def test_import_config_service(self):
-        from ui.services.config_service import ConfigService
-        assert ConfigService is not None
-
-    def test_import_config_manager(self):
-        from ui.services.config_manager import ConfigManager
-        assert ConfigManager is not None
-
-
-class TestAppInitialization:
-    """测试 App 初始化"""
-    
-    def test_app_without_backend(self):
-        from ui import GraphMemoryApp
-        app = GraphMemoryApp()
-        assert app._backend_server is None
-        assert app._backend_client is None
-
-    def test_app_with_backend(self):
-        from ui import GraphMemoryApp
-        from core import BackendServer
-        with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
-            db_path = f.name
-        try:
-            server = BackendServer(db_path=db_path)
-            server.start(api_key="")
-            app = GraphMemoryApp(backend_server=server)
-            assert app._backend_server is server
-            assert app._backend_client is not None
-            server.shutdown()
-        finally:
-            if os.path.exists(db_path):
-                os.unlink(db_path)
-
-
-class TestUIWithBackendClient:
-    """测试 UI 与后端通信"""
-    
-    def test_app_sends_message_via_backend_client(self):
-        from ui import GraphMemoryApp
-        from core import BackendServer, BackendClient
-        
-        with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
-            db_path = f.name
-        try:
-            server = BackendServer(db_path=db_path)
-            server.start(api_key="")
-            
-            app = GraphMemoryApp(backend_server=server)
-            client = app._backend_client
-            
-            status = client.get_status()
-            assert status.get("success") is True
-            
-            result = client.update_settings(
-                api_config={"api_key": "sk-test", "base_url": "https://api.deepseek.com", "model": "deepseek-chat"},
-                tool_limits={"persona_query_max": 1}
-            )
-            assert result.get("success") is True
-            
-            server.shutdown()
-        finally:
-            if os.path.exists(db_path):
-                os.unlink(db_path)
-    
-    def test_ui_get_history(self):
-        from ui import GraphMemoryApp
-        from core import BackendServer
-        
-        with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
-            db_path = f.name
-        try:
-            server = BackendServer(db_path=db_path)
-            server.start(api_key="")
-            
-            app = GraphMemoryApp(backend_server=server)
-            client = app._backend_client
-            
-            history = client.get_history()
-            assert isinstance(history, list)
-            
-            server.shutdown()
-        finally:
-            if os.path.exists(db_path):
-                os.unlink(db_path)
-
-    def test_ui_only_uses_backend_client(self):
-        from ui import GraphMemoryApp
-        from core import BackendServer
-        
-        with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
-            db_path = f.name
-        try:
-            server = BackendServer(db_path=db_path)
-            server.start(api_key="")
-            
-            app = GraphMemoryApp(backend_server=server)
-            
-            # UI 不应该直接访问后端内部
-            assert hasattr(app, '_backend_client')
-            assert app._backend_client is not None
-            
-            # 不应该有 _graph, _client 等直接访问
-            assert not hasattr(app, '_graph')
-            assert not hasattr(app, '_client')
-            
-            server.shutdown()
-        finally:
-            if os.path.exists(db_path):
-                os.unlink(db_path)
+"""Tests for UI layer"""

tests/test_ui/test_ui.py

@@ -0,0 +1,242 @@
+"""Tests for UI layer - models, widgets, handlers, services, and app initialization."""
+import pytest
+import os
+import tempfile
+from pathlib import Path
+
+os.environ["DEEPSEEK_API_KEY"] = "fake-test-key"
+
+
+class TestUIImport:
+    """测试 UI 模块导入"""
+
+    def test_import_graphmemoryapp(self):
+        from ui import GraphMemoryApp
+        assert GraphMemoryApp is not None
+
+    def test_import_appconfig(self):
+        from ui import AppConfig
+        assert AppConfig is not None
+
+    def test_import_message(self):
+        from ui.models.message import Message, ToolCall, ToolResult
+        assert Message is not None
+        assert ToolCall is not None
+        assert ToolResult is not None
+
+    def test_import_config(self):
+        from ui.models.config import AppConfig
+        assert AppConfig is not None
+
+    def test_import_log_entry(self):
+        from ui.models.log_entry import LogEntry
+        assert LogEntry is not None
+
+
+class TestAppConfig:
+    """测试配置模型"""
+
+    def test_config_default_values(self):
+        from ui.models.config import AppConfig
+        config = AppConfig()
+        assert config.api_key == ""
+        assert config.model == "deepseek-chat"
+        assert config.base_url == "https://api.deepseek.com"
+
+    def test_config_from_env(self):
+        from ui.models.config import AppConfig
+        config = AppConfig.from_env()
+        assert "fake-test-key" in config.api_key
+
+
+class TestMessageModel:
+    """测试消息模型"""
+
+    def test_message_creation_user(self):
+        from ui.models.message import Message
+        from datetime import datetime
+        msg = Message(role="user", content="test content")
+        assert msg.role == "user"
+        assert msg.content == "test content"
+        assert isinstance(msg.timestamp, datetime)
+
+    def test_message_creation_assistant(self):
+        from ui.models.message import Message
+        msg = Message(role="assistant", content="assistant response")
+        assert msg.role == "assistant"
+
+    def test_message_with_tool_calls(self):
+        from ui.models.message import Message, ToolCall
+        tc = ToolCall(id="call-1", name="memory_recall", arguments={"query": "test"})
+        msg = Message(role="assistant", content="response", tool_calls=[tc])
+        assert msg.tool_calls is not None
+        assert len(msg.tool_calls) == 1
+
+
+class TestAppCSSPath:
+    """测试 App CSS 配置"""
+
+    def test_app_has_css_path(self):
+        from ui import GraphMemoryApp
+        assert hasattr(GraphMemoryApp, 'CSS_PATH')
+        assert len(GraphMemoryApp.CSS_PATH) > 0
+
+
+class TestAppBindings:
+    """测试 App 快捷键"""
+
+    def test_app_has_bindings(self):
+        from ui import GraphMemoryApp
+        assert hasattr(GraphMemoryApp, 'BINDINGS')
+        assert len(GraphMemoryApp.BINDINGS) > 0
+
+
+class TestWidgetImports:
+    """测试组件导入"""
+
+    def test_import_left_panel(self):
+        from ui.widgets.left_panel import LeftPanel
+        assert LeftPanel is not None
+
+    def test_import_right_panel(self):
+        from ui.widgets.right_panel import RightPanel
+        assert RightPanel is not None
+
+    def test_import_input_box(self):
+        from ui.widgets.input_box import InputBox
+        assert InputBox is not None
+
+    def test_import_message_history(self):
+        from ui.widgets.message_history import MessageHistory
+        assert MessageHistory is not None
+
+    def test_import_status_bar(self):
+        from ui.widgets.status_bar import StatusBar
+        assert StatusBar is not None
+
+
+class TestHandlerImports:
+    """测试处理器导入"""
+
+    def test_import_focus_handler(self):
+        from ui.handlers.focus_handler import FocusHandler
+        assert FocusHandler is not None
+
+    def test_import_key_handler(self):
+        from ui.handlers.key_handler import KeyHandler
+        assert KeyHandler is not None
+
+
+class TestServiceImports:
+    """测试服务导入"""
+
+    def test_import_config_service(self):
+        from ui.services.config_service import ConfigService
+        assert ConfigService is not None
+
+    def test_import_config_manager(self):
+        from ui.services.config_manager import ConfigManager
+        assert ConfigManager is not None
+
+
+class TestAppInitialization:
+    """测试 App 初始化"""
+
+    def test_app_without_backend(self):
+        from ui import GraphMemoryApp
+        app = GraphMemoryApp()
+        assert app._backend_server is None
+        assert app._backend_client is None
+
+    def test_app_with_backend(self):
+        from ui import GraphMemoryApp
+        from core import BackendServer
+        with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
+            db_path = f.name
+        try:
+            server = BackendServer(db_path=db_path)
+            server.start(api_key="")
+            app = GraphMemoryApp(backend_server=server)
+            assert app._backend_server is server
+            assert app._backend_client is not None
+            server.shutdown()
+        finally:
+            if os.path.exists(db_path):
+                os.unlink(db_path)
+
+
+class TestUIWithBackendClient:
+    """测试 UI 与后端通信"""
+
+    def test_app_sends_message_via_backend_client(self):
+        from ui import GraphMemoryApp
+        from core import BackendServer, BackendClient
+
+        with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
+            db_path = f.name
+        try:
+            server = BackendServer(db_path=db_path)
+            server.start(api_key="")
+
+            app = GraphMemoryApp(backend_server=server)
+            client = app._backend_client
+
+            status = client.get_status()
+            assert status.get("success") is True
+
+            result = client.update_settings(
+                api_config={"api_key": "sk-test", "base_url": "https://api.deepseek.com", "model": "deepseek-chat"},
+                tool_limits={"persona_update_max": 1}
+            )
+            assert result.get("success") is True
+
+            server.shutdown()
+        finally:
+            if os.path.exists(db_path):
+                os.unlink(db_path)
+
+    def test_ui_get_history(self):
+        from ui import GraphMemoryApp
+        from core import BackendServer
+
+        with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
+            db_path = f.name
+        try:
+            server = BackendServer(db_path=db_path)
+            server.start(api_key="")
+
+            app = GraphMemoryApp(backend_server=server)
+            client = app._backend_client
+
+            history = client.get_history()
+            assert isinstance(history, list)
+
+            server.shutdown()
+        finally:
+            if os.path.exists(db_path):
+                os.unlink(db_path)
+
+    def test_ui_only_uses_backend_client(self):
+        from ui import GraphMemoryApp
+        from core import BackendServer
+
+        with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
+            db_path = f.name
+        try:
+            server = BackendServer(db_path=db_path)
+            server.start(api_key="")
+
+            app = GraphMemoryApp(backend_server=server)
+
+            # UI 不应该直接访问后端内部
+            assert hasattr(app, '_backend_client')
+            assert app._backend_client is not None
+
+            # 不应该有 _graph, _client 等直接访问
+            assert not hasattr(app, '_graph')
+            assert not hasattr(app, '_client')
+
+            server.shutdown()
+        finally:
+            if os.path.exists(db_path):
+                os.unlink(db_path)

ui/app.py

@@ -46,10 +46,8 @@ class GraphMemoryApp(App):
             initial_config.model = api_config.get("model", "deepseek-chat")
             
             tool_limits = settings_data.get("tool_limits", {})
-            initial_config.persona_query_max = tool_limits.get("persona_query_max", 1)
             initial_config.persona_update_max = tool_limits.get("persona_update_max", 1)
-            initial_config.task_query_max = tool_limits.get("task_query_max", 4)
-            initial_config.task_update_max = tool_limits.get("task_update_max", 2)
+            initial_config.task_update_max = tool_limits.get("task_update_max", 5)
             initial_config.memory_query_max = tool_limits.get("memory_query_max", 20)
             initial_config.memory_update_max = tool_limits.get("memory_update_max", 10)
         
@@ -222,9 +220,7 @@ class GraphMemoryApp(App):
         }
         
         tool_limits = {
-            "persona_query_max": config.persona_query_max,
             "persona_update_max": config.persona_update_max,
-            "task_query_max": config.task_query_max,
             "task_update_max": config.task_update_max,
             "memory_query_max": config.memory_query_max,
             "memory_update_max": config.memory_update_max,
@@ -257,10 +253,8 @@ class GraphMemoryApp(App):
                         api_key=api_cfg.get("api_key", ""),
                         base_url=api_cfg.get("base_url", "https://api.deepseek.com"),
                         model=api_cfg.get("model", "deepseek-chat"),
-                        persona_query_max=tool_lmts.get("persona_query_max", 1),
                         persona_update_max=tool_lmts.get("persona_update_max", 1),
-                        task_query_max=tool_lmts.get("task_query_max", 4),
-                        task_update_max=tool_lmts.get("task_update_max", 2),
+                        task_update_max=tool_lmts.get("task_update_max", 5),
                         memory_query_max=tool_lmts.get("memory_query_max", 20),
                         memory_update_max=tool_lmts.get("memory_update_max", 10),
                     ))

ui/models/config.py

@@ -13,10 +13,8 @@ class AppConfig:
     api_key: str = ""
     model: str = "deepseek-chat"
     base_url: str = "https://api.deepseek.com"
-    persona_query_max: int = 1
     persona_update_max: int = 1
-    task_query_max: int = 4
-    task_update_max: int = 2
+    task_update_max: int = 5
     memory_query_max: int = 20
     memory_update_max: int = 10
 
@@ -26,10 +24,8 @@ class AppConfig:
             api_key=os.getenv("DEEPSEEK_API_KEY", ""),
             model=os.getenv("MODEL_NAME", "deepseek-chat"),
             base_url=os.getenv("DEEPSEEK_BASE_URL", "https://api.deepseek.com"),
-            persona_query_max=int(os.getenv("PERSONA_QUERY_MAX", 1)),
             persona_update_max=int(os.getenv("PERSONA_UPDATE_MAX", 1)),
-            task_query_max=int(os.getenv("TASK_QUERY_MAX", 4)),
-            task_update_max=int(os.getenv("TASK_UPDATE_MAX", 2)),
+            task_update_max=int(os.getenv("TASK_UPDATE_MAX", 5)),
             memory_query_max=int(os.getenv("MEMORY_QUERY_MAX", 20)),
             memory_update_max=int(os.getenv("MEMORY_UPDATE_MAX", 10)),
         )
@@ -46,10 +42,8 @@ class AppConfig:
             api_key=data.get("api_key", ""),
             model=data.get("model", "deepseek-chat"),
             base_url=data.get("base_url", "https://api.deepseek.com"),
-            persona_query_max=data.get("persona_query_max", 1),
             persona_update_max=data.get("persona_update_max", 1),
-            task_query_max=data.get("task_query_max", 4),
-            task_update_max=data.get("task_update_max", 2),
+            task_update_max=data.get("task_update_max", 5),
             memory_query_max=data.get("memory_query_max", 20),
             memory_update_max=data.get("memory_update_max", 10),
         )

ui/widgets/config_section.py

@@ -63,25 +63,15 @@ class ConfigSection(Vertical):
         limits_title.can_focus = False
         yield limits_title
         
-        l1 = Static("人设图查询:", classes="config-label")
-        l1.can_focus = False
-        yield l1
-        yield Input(value=str(self._config.persona_query_max), placeholder="1", id="persona-query-max")
-        
         l2 = Static("人设图修改:", classes="config-label")
         l2.can_focus = False
         yield l2
         yield Input(value=str(self._config.persona_update_max), placeholder="1", id="persona-update-max")
         
-        l3 = Static("工作记忆查询:", classes="config-label")
-        l3.can_focus = False
-        yield l3
-        yield Input(value=str(self._config.task_query_max), placeholder="4", id="task-query-max")
-        
         l4 = Static("工作记忆修改:", classes="config-label")
         l4.can_focus = False
         yield l4
-        yield Input(value=str(self._config.task_update_max), placeholder="2", id="task-update-max")
+        yield Input(value=str(self._config.task_update_max), placeholder="5", id="task-update-max")
         
         l5 = Static("一般记忆查询:", classes="config-label")
         l5.can_focus = False
@@ -122,9 +112,7 @@ class ConfigSection(Vertical):
             model_input = self.query_one("#model-input", Input)
             base_url_input = self.query_one("#base-url-input", Input)
             
-            persona_query = self.query_one("#persona-query-max", Input)
             persona_update = self.query_one("#persona-update-max", Input)
-            task_query = self.query_one("#task-query-max", Input)
             task_update = self.query_one("#task-update-max", Input)
             memory_query = self.query_one("#memory-query-max", Input)
             memory_update = self.query_one("#memory-update-max", Input)
@@ -133,10 +121,8 @@ class ConfigSection(Vertical):
                 api_key=api_key_input.value,
                 model=model_input.value,
                 base_url=base_url_input.value,
-                persona_query_max=int(persona_query.value or 1),
                 persona_update_max=int(persona_update.value or 1),
-                task_query_max=int(task_query.value or 4),
-                task_update_max=int(task_update.value or 2),
+                task_update_max=int(task_update.value or 5),
                 memory_query_max=int(memory_query.value or 20),
                 memory_update_max=int(memory_update.value or 10),
             )
@@ -162,9 +148,7 @@ class ConfigSection(Vertical):
             model_input.value = config.model
             base_url_input.value = config.base_url
             
-            self.query_one("#persona-query-max", Input).value = str(config.persona_query_max)
             self.query_one("#persona-update-max", Input).value = str(config.persona_update_max)
-            self.query_one("#task-query-max", Input).value = str(config.task_query_max)
             self.query_one("#task-update-max", Input).value = str(config.task_update_max)
             self.query_one("#memory-query-max", Input).value = str(config.memory_query_max)
             self.query_one("#memory-update-max", Input).value = str(config.memory_update_max)