feat: 日志持久化(6h归档) + purge增强(source_type/source_has_status过滤)

- ActivityRecorder 新增 LogPersister 后台线程，每10s增量写入logs/operations.*.log

- 日志每6小时自动gzip压缩归档

- memory_purge 新增 source_type / target_type / source_has_status 过滤条件

- 支持精准清理已归档任务的残留 HAS_STATE 关系

core/activity_recorder.py

@@ -1,6 +1,26 @@
+"""
+活动记录器 - 记录 AI 对图数据库的操作
+使用 SQLite :memory: 供 WebUI 实时渲染，同时后台线程持久化到日志文件
+日志每 6 小时自动压缩归档
+"""
+
 import sqlite3
 import time
+import os
+import gzip
+import json
+import threading
+import shutil
 from typing import List, Dict, Optional
+from datetime import datetime, timedelta
+
+
+# 日志目录
+LOG_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "logs")
+# 归档间隔（秒）
+ARCHIVE_INTERVAL = 6 * 3600  # 6 小时
+# 轮询间隔（秒）
+POLL_INTERVAL = 10
 
 
 class ActivityRecorder:
@@ -8,18 +28,46 @@ class ActivityRecorder:
 
     def __init__(self):
         self.conn = sqlite3.connect(":memory:", check_same_thread=False)
-        self.conn.execute("CREATE TABLE activities (id INTEGER PRIMARY KEY AUTOINCREMENT, timestamp REAL, action TEXT, tool_name TEXT, entity TEXT, detail TEXT)")
+        self.conn.execute(
+            "CREATE TABLE activities (id INTEGER PRIMARY KEY AUTOINCREMENT, "
+            "timestamp REAL, action TEXT, tool_name TEXT, entity TEXT, detail TEXT)"
+        )
         self.conn.commit()
 
     def record(self, action: str, tool_name: str, entity: str, detail: str = "") -> None:
-        self.conn.execute("INSERT INTO activities (timestamp, action, tool_name, entity, detail) VALUES (?, ?, ?, ?, ?)",
-                          (time.time(), action, tool_name, entity, detail))
+        self.conn.execute(
+            "INSERT INTO activities (timestamp, action, tool_name, entity, detail) VALUES (?, ?, ?, ?, ?)",
+            (time.time(), action, tool_name, entity, detail)
+        )
         self.conn.commit()
 
     def get_all(self) -> List[Dict]:
-        cursor = self.conn.execute("SELECT id, timestamp, action, tool_name, entity, detail FROM activities ORDER BY id")
+        cursor = self.conn.execute(
+            "SELECT id, timestamp, action, tool_name, entity, detail FROM activities ORDER BY id"
+        )
         rows = cursor.fetchall()
-        return [{"id": r[0], "timestamp": r[1], "action": r[2], "tool_name": r[3], "entity": r[4], "detail": r[5]} for r in rows]
+        return [
+            {"id": r[0], "timestamp": r[1], "action": r[2],
+             "tool_name": r[3], "entity": r[4], "detail": r[5]}
+            for r in rows
+        ]
+
+    def get_since_id(self, last_id: int) -> List[Dict]:
+        """获取自 last_id 之后的新记录"""
+        cursor = self.conn.execute(
+            "SELECT id, timestamp, action, tool_name, entity, detail FROM activities WHERE id > ? ORDER BY id",
+            (last_id,)
+        )
+        rows = cursor.fetchall()
+        return [
+            {"id": r[0], "timestamp": r[1], "action": r[2],
+             "tool_name": r[3], "entity": r[4], "detail": r[5]}
+            for r in rows
+        ]
+
+    def get_max_id(self) -> int:
+        cursor = self.conn.execute("SELECT COALESCE(MAX(id), 0) FROM activities")
+        return cursor.fetchone()[0]
 
     def clear(self) -> None:
         self.conn.execute("DELETE FROM activities")
@@ -31,11 +79,107 @@ class ActivityRecorder:
         return {r[0]: r[1] for r in rows}
 
 
+# ── 日志文件管理 ──
+
+def _current_log_path() -> str:
+    """返回当前日志文件路径（按日期命名）"""
+    os.makedirs(LOG_DIR, exist_ok=True)
+    date_str = datetime.now().strftime("%Y%m%d")
+    return os.path.join(LOG_DIR, f"operations.{date_str}.log")
+
+
+def _archive_log(filepath: str) -> str:
+    """压缩归档日志文件，返回归档文件路径"""
+    if not os.path.exists(filepath) or os.path.getsize(filepath) == 0:
+        return ""
+    archive_path = filepath + ".gz"
+    try:
+        with open(filepath, "rb") as f_in:
+            with gzip.open(archive_path, "wb") as f_out:
+                shutil.copyfileobj(f_in, f_out)
+        os.remove(filepath)
+        return archive_path
+    except Exception:
+        return ""
+
+
+class LogPersister:
+    """后台日志持久化线程 - 定期将内存记录写入日志文件并自动归档"""
+
+    def __init__(self, recorder: ActivityRecorder):
+        self.recorder = recorder
+        self._last_persisted_id = 0
+        self._last_archive_time = time.time()
+        self._running = True
+        self._thread = threading.Thread(target=self._run, daemon=True, name="log-persister")
+        self._thread.start()
+
+    def _run(self):
+        """主循环"""
+        while self._running:
+            try:
+                self._persist_new()
+                self._check_archive()
+            except Exception:
+                pass  # 不因日志异常影响主进程
+            time.sleep(POLL_INTERVAL)
+
+    def _persist_new(self):
+        """增量写入新记录到日志文件"""
+        records = self.recorder.get_since_id(self._last_persisted_id)
+        if not records:
+            return
+
+        log_path = _current_log_path()
+        with open(log_path, "a", encoding="utf-8") as f:
+            for r in records:
+                line = json.dumps(r, ensure_ascii=False)
+                f.write(line + "\n")
+
+        # 更新水位
+        if records:
+            self._last_persisted_id = records[-1]["id"]
+
+    def _check_archive(self):
+        """检查是否需要归档"""
+        elapsed = time.time() - self._last_archive_time
+        if elapsed < ARCHIVE_INTERVAL:
+            return
+
+        log_path = _current_log_path()
+        archived = _archive_log(log_path)
+        if archived:
+            dt = datetime.fromtimestamp(self._last_archive_time)
+            print(f"[日志归档] {dt.strftime('%H:%M')} → {os.path.basename(archived)} ({_fmt_size(archived)})")
+        self._last_archive_time = time.time()
+
+    def stop(self):
+        self._running = False
+
+
+def _fmt_size(path: str) -> str:
+    size = os.path.getsize(path)
+    for unit in ("B", "KB", "MB"):
+        if size < 1024:
+            return f"{size:.1f}{unit}"
+        size /= 1024
+    return f"{size:.1f}GB"
+
+
+# ── 单例 ──
+
 _recorder: Optional[ActivityRecorder] = None
+_persister: Optional[LogPersister] = None
 
 
 def get_recorder() -> ActivityRecorder:
-    global _recorder
+    """获取全局 ActivityRecorder（首次调用时自动启动日志持久化线程）"""
+    global _recorder, _persister
     if _recorder is None:
         _recorder = ActivityRecorder()
+        _persister = LogPersister(_recorder)
     return _recorder
+
+
+def get_persister() -> Optional[LogPersister]:
+    return _persister

core/embedded_db.py

@@ -443,6 +443,16 @@ class EmbeddedGraphDB:
         
         Args:
             criteria: 删除条件
+                支持:
+                - source: 源实体名（精确匹配）
+                - target: 目标实体名（精确匹配）
+                - relation: 关系类型
+                - subject_contains: 源实体名包含（模糊匹配）
+                - target_contains: 目标实体名包含（模糊匹配）
+                - relation_type: 关系类型（同 relation）
+                - source_type: 源实体类型过滤
+                - target_type: 目标实体类型过滤
+                - source_has_status: 源实体 mentions_count 状态（支持 type 字段）
             mode: 删除模式 (soft/hard)
             new_relation: 替代关系
         
@@ -454,39 +464,94 @@ class EmbeddedGraphDB:
         # 构建查询条件
         conditions = []
         params = []
+        joins = []
+        
+        relation_type = criteria.get('relation') or criteria.get('relation_type', '')
         
         if criteria.get('source'):
             cursor.execute("SELECT id FROM entities WHERE name = ?", (criteria['source'],))
             row = cursor.fetchone()
             if row:
-                conditions.append("source_id = ?")
+                conditions.append("r.source_id = ?")
                 params.append(row['id'])
         
         if criteria.get('target'):
             cursor.execute("SELECT id FROM entities WHERE name = ?", (criteria['target'],))
             row = cursor.fetchone()
             if row:
-                conditions.append("target_id = ?")
+                conditions.append("r.target_id = ?")
                 params.append(row['id'])
         
-        if criteria.get('relation'):
-            conditions.append("relation_type = ?")
-            params.append(criteria['relation'])
+        if relation_type:
+            conditions.append("r.relation_type = ?")
+            params.append(relation_type)
+        
+        # 通过子查询支持实体属性过滤
+        if criteria.get('subject_contains'):
+            cursor.execute("SELECT id FROM entities WHERE name LIKE ?",
+                          (f'%{criteria["subject_contains"]}%',))
+            ids = [row['id'] for row in cursor.fetchall()]
+            if ids:
+                placeholders = ','.join(['?'] * len(ids))
+                conditions.append(f"r.source_id IN ({placeholders})")
+                params.extend(ids)
+        
+        if criteria.get('target_contains'):
+            cursor.execute("SELECT id FROM entities WHERE name LIKE ?",
+                          (f'%{criteria["target_contains"]}%',))
+            ids = [row['id'] for row in cursor.fetchall()]
+            if ids:
+                placeholders = ','.join(['?'] * len(ids))
+                conditions.append(f"r.target_id IN ({placeholders})")
+                params.extend(ids)
+        
+        # 源实体类型过滤
+        if criteria.get('source_type'):
+            cursor.execute("SELECT id FROM entities WHERE type = ?",
+                          (criteria['source_type'],))
+            ids = [row['id'] for row in cursor.fetchall()]
+            if ids:
+                placeholders = ','.join(['?'] * len(ids))
+                conditions.append(f"r.source_id IN ({placeholders})")
+                params.extend(ids)
+        
+        # 目标实体类型过滤
+        if criteria.get('target_type'):
+            cursor.execute("SELECT id FROM entities WHERE type = ?",
+                          (criteria['target_type'],))
+            ids = [row['id'] for row in cursor.fetchall()]
+            if ids:
+                placeholders = ','.join(['?'] * len(ids))
+                conditions.append(f"r.target_id IN ({placeholders})")
+                params.extend(ids)
+        
+        # 源实体状态过滤（检查 entity name 是否以特定后缀结尾等）
+        if criteria.get('source_has_status'):
+            status = criteria['source_has_status']
+            # 匹配 entities 表中 type 字段包含状态信息的节点
+            cursor.execute("SELECT id FROM entities WHERE type LIKE ?",
+                          (f'%{status}%',))
+            ids = [row['id'] for row in cursor.fetchall()]
+            if ids:
+                placeholders = ','.join(['?'] * len(ids))
+                conditions.append(f"r.source_id IN ({placeholders})")
+                params.extend(ids)
         
         if not conditions:
             return {"deleted": 0, "message": "无删除条件"}
         
+        conditions.append("r.status = 'active'")
         where_clause = " AND ".join(conditions)
         
         if mode == "soft":
             cursor.execute(f"""
-                UPDATE relations
+                UPDATE relations r
                 SET status = 'deleted', updated_at = CURRENT_TIMESTAMP
-                WHERE {where_clause} AND status = 'active'
+                WHERE {where_clause}
             """, params)
         else:
             cursor.execute(f"""
-                DELETE FROM relations
+                DELETE FROM relations r
                 WHERE {where_clause}
             """, params)
         

core/graph_client.py

@@ -173,6 +173,9 @@ class Neo4jGraph:
             subject_pattern = criteria.get("subject_contains", "")
             rel_type = criteria.get("relation_type", "")
             target_pattern = criteria.get("target_contains", "")
+            source_type = criteria.get("source_type", "")
+            target_type = criteria.get("target_type", "")
+            source_status = criteria.get("source_has_status", "")
             session_id = criteria.get("session_id", CURRENT_SESSION_ID)
             
             cond_parts = ["r.status = 'active'"]
@@ -187,6 +190,15 @@ class Neo4jGraph:
             if rel_type:
                 cond_parts.append("r.type = $rel_type")
                 params["rel_type"] = rel_type
+            if source_type:
+                cond_parts.append("s.entity_type = $source_type")
+                params["source_type"] = source_type
+            if target_type:
+                cond_parts.append("t.entity_type = $target_type")
+                params["target_type"] = target_type
+            if source_status:
+                cond_parts.append("s.status = $source_status")
+                params["source_status"] = source_status
             
             where_clause = " AND ".join(cond_parts)
             
@@ -208,7 +220,7 @@ class Neo4jGraph:
                 return {"deleted_count": count, "mode": "supersede"}
             else:
                 result = session.run(f"""
-                MATCH ()-[r:RELATES]->()
+                MATCH (s:Entity)-[r:RELATES]->(t:Entity)
                 WHERE {where_clause}
                 SET r.status = 'deleted', r.updated_at = datetime()
                 RETURN count(r) as deleted

core/tools/memory_tools.py

@@ -146,6 +146,12 @@ MEMORY_TOOLS = [
 4. 删除旧记忆:
    {"criteria": {"time_before": "2024-01-01"}, "mode": "soft"}
 
+5. 删除残留在已归档任务上的状态关系:
+   {"criteria": {"relation_type": "HAS_STATE", "source_type": "TaskNode", "source_has_status": "archived"}, "mode": "soft"}
+
+6. 删除特定类型的节点关系:
+   {"criteria": {"relation_type": "某种关系", "target_type": "某种类型"}, "mode": "soft"}
+
 【重要】删除原则:
 - 优先使用 supersede 模式修正错误
 - 软删除不会物理删除数据
@@ -159,6 +165,9 @@ MEMORY_TOOLS = [
                             "subject_contains": {"type": "string"},
                             "relation_type": {"type": "string"},
                             "target_contains": {"type": "string"},
+                            "source_type": {"type": "string", "description": "源实体类型过滤（如 TaskNode）"},
+                            "target_type": {"type": "string", "description": "目标实体类型过滤"},
+                            "source_has_status": {"type": "string", "description": "源实体状态过滤（如 archived）"},
                             "time_before": {"type": "string"},
                             "session_id": {"type": "string"}
                         },