fix(ai): 修复本地检索进度事件导致 ai.sqlite3 无限膨胀

frontend/components/LocalSearchSettings.vue

@@ -378,7 +378,8 @@ function subscribe(){
     if(event?.kind==='local_search_index' && event.account===current){
       const incoming=event.body, existing=state.value.jobs?.find(j=>j.id===incoming?.id)
       if(existing && incoming.updated>=existing.updated){
-        state.value.jobs=state.value.jobs.map(j=>j.id===incoming.id ? incoming : j)
+        // 进度事件只带高频字段；合并保留首次加载时拿到的 config/segments 等快照。
+        state.value.jobs=state.value.jobs.map(j=>j.id===incoming.id ? {...j, ...incoming} : j)
         now.value=Date.now()/1000
         if(incoming.status==='done')refresh()
       }else if(!existing){refresh()}else{api.diagnostic?.('response.stale',{task_id:incoming.id,component:'search'})}

src/wechat_decrypt_tool/ai/lifecycle.py

@@ -1,10 +1,21 @@
 """AI 服务统一启停诊断，独立于其他后台服务的生命周期。"""
 import logging
+import threading
 from importlib.metadata import PackageNotFoundError, version
 
 from .diagnostics import observed, event
 
 
+def _maintain_store(store, name):
+    """在后台回收过期事件并压缩数据库，避免大型遗留库拖慢启动。"""
+    try:
+        deduplicated, removed, freed = store.maintain()
+        event('storage.maintenance.finished', component=name, events_deduplicated=deduplicated,
+              events_removed=removed, bytes_freed=freed)
+    except Exception as error:
+        event('storage.maintenance.failed', level=logging.WARNING, component=name, error=error)
+
+
 @observed('lifecycle.start')
 async def start_services():
     for package in ('deepagents', 'langchain-openai', 'langchain-anthropic', 'langgraph', 'onnxruntime', 'sqlite-vec', 'tokenizers'):
@@ -19,6 +30,8 @@ async def start_services():
     get_ai_service().start()
     await get_agent_service().start()
     await get_local_search().start()
+    for name, store in (('summary', get_ai_service().store), ('search', get_local_search().store)):
+        threading.Thread(target=_maintain_store, args=(store, name), name=f'ai-store-maintenance-{name}', daemon=True).start()
 
 
 @observed('lifecycle.stop')

src/wechat_decrypt_tool/ai/storage.py

@@ -3,6 +3,7 @@
 import logging
 
 import json
+import os
 import sqlite3
 import threading
 import time
@@ -13,6 +14,28 @@
 from ..app_paths import get_output_dir
 
 
+# 事件默认保留窗口；超过该窗口且无需重放的事件会被回收。
+EVENT_RETENTION_SECONDS = 24 * 3600
+# 事件表空闲页超过该阈值才执行 VACUUM，避免频繁全库重写。
+COMPACT_MINIMUM_BYTES = 64 * 1024 * 1024
+# 超过该体积的库不做原地删除/VACUUM（对遗留巨型库会放大 WAL），改为重建：
+# 仅保留 records 与未投递提醒，丢弃可再生的 events。
+MAINTENANCE_MAX_DATABASE_BYTES = 512 * 1024 * 1024
+
+SCHEMA_SQL = """
+    CREATE TABLE IF NOT EXISTS records (
+        kind TEXT NOT NULL, id TEXT NOT NULL, account TEXT NOT NULL DEFAULT '',
+        body TEXT NOT NULL, updated REAL NOT NULL, PRIMARY KEY(kind,id));
+    CREATE INDEX IF NOT EXISTS records_account ON records(kind,account,updated);
+    CREATE INDEX IF NOT EXISTS records_status ON records(kind,json_extract(body,'$.status'),updated);
+    CREATE INDEX IF NOT EXISTS records_task_usage ON records(kind,account,json_extract(body,'$.task_id'));
+    CREATE TABLE IF NOT EXISTS events (
+        id INTEGER PRIMARY KEY AUTOINCREMENT, account TEXT NOT NULL,
+        kind TEXT NOT NULL, body TEXT NOT NULL, unique_key TEXT UNIQUE,
+        delivered INTEGER NOT NULL DEFAULT 0, created REAL NOT NULL);
+"""
+
+
 class AIStore:
     """短事务业务存储；与工作流检查点分开，避免模型请求持有数据库锁。"""
 
@@ -27,18 +50,7 @@ def __init__(self, root: Path | None = None):
         self._event_condition = threading.Condition()
         self._event_revisions = {}
         with self.connection() as db:
-            db.executescript("""
-                CREATE TABLE IF NOT EXISTS records (
-                    kind TEXT NOT NULL, id TEXT NOT NULL, account TEXT NOT NULL DEFAULT '',
-                    body TEXT NOT NULL, updated REAL NOT NULL, PRIMARY KEY(kind,id));
-                CREATE INDEX IF NOT EXISTS records_account ON records(kind,account,updated);
-                CREATE INDEX IF NOT EXISTS records_status ON records(kind,json_extract(body,'$.status'),updated);
-                CREATE INDEX IF NOT EXISTS records_task_usage ON records(kind,account,json_extract(body,'$.task_id'));
-                CREATE TABLE IF NOT EXISTS events (
-                    id INTEGER PRIMARY KEY AUTOINCREMENT, account TEXT NOT NULL,
-                    kind TEXT NOT NULL, body TEXT NOT NULL, unique_key TEXT UNIQUE,
-                    delivered INTEGER NOT NULL DEFAULT 0, created REAL NOT NULL);
-            """)
+            db.executescript(SCHEMA_SQL)
 
     @contextmanager
     def connection(self):
@@ -108,13 +120,30 @@ def delete(self, kind, id):
         with self.connection() as db:
             db.execute("DELETE FROM records WHERE kind=? AND id=?", (kind, id))
 
-    def event(self, account, kind, body, unique_key=None):
-        inserted = False
+    def event(self, account, kind, body, unique_key=None, replace=False):
+        """写入事件供 SSE 重放。
+
+        默认行为保持不变：提供 `unique_key` 时按去重语义写入（同 key 已存在则忽略），
+        用于提醒等只应投递一次的事件。`replace=True` 时改为用最新快照替换旧行，
+        让高频进度事件每个逻辑任务只保留一行，同时因 INSERT OR REPLACE 会删除旧行、
+        新行仍获得递增的自增 id，断线重连的 EventSource 依然能收到最新状态。
+        """
         with self.connection() as db:
             if account in self.revoked_accounts:
                 return
-            cursor = db.execute("INSERT OR IGNORE INTO events(account,kind,body,unique_key,created) VALUES(?,?,?,?,?)",
-                                (account, kind, json.dumps(body, ensure_ascii=False), unique_key, time.time()))
+            payload = json.dumps(body, ensure_ascii=False)
+            if unique_key is None:
+                cursor = db.execute(
+                    "INSERT INTO events(account,kind,body,created) VALUES(?,?,?,?)",
+                    (account, kind, payload, time.time()))
+            elif replace:
+                cursor = db.execute(
+                    "INSERT OR REPLACE INTO events(account,kind,body,unique_key,created) VALUES(?,?,?,?,?)",
+                    (account, kind, payload, unique_key, time.time()))
+            else:
+                cursor = db.execute(
+                    "INSERT OR IGNORE INTO events(account,kind,body,unique_key,created) VALUES(?,?,?,?,?)",
+                    (account, kind, payload, unique_key, time.time()))
             inserted = cursor.rowcount > 0
         if inserted:
             with self._event_condition:
@@ -151,6 +180,177 @@ def acknowledge(self, id):
         with self.connection() as db:
             db.execute("UPDATE events SET delivered=1 WHERE id=?", (id,))
 
+    @observed('storage.prune_duplicates')
+    def prune_duplicate_events(self, batch=2000):
+        """一次性折叠旧版追加式进度事件：每个逻辑任务只保留最新快照。
+
+        新写入的进度事件已带 unique_key、本身只保留一行；这里主要清理升级前
+        历史遗留的、同一任务多次追加的整份快照，避免巨型库只能等 TTL 慢慢过期。
+        """
+        total = 0
+        for kind, field in (('local_search_index', '$.id'), ('local_search_download', '$.id'),
+                            ('local_search_total', '$.job_id')):
+            with self.connection() as db:
+                db.execute("CREATE TEMP TABLE IF NOT EXISTS keep_event_ids(id INTEGER PRIMARY KEY)")
+                db.execute("DELETE FROM keep_event_ids")
+                db.execute(
+                    f"INSERT INTO keep_event_ids SELECT max(id) FROM events "
+                    f"WHERE kind=? AND unique_key IS NULL GROUP BY json_extract(body,'{field}')", (kind,))
+                while True:
+                    removed = db.execute(
+                        "DELETE FROM events WHERE id IN (SELECT id FROM events "
+                        "WHERE kind=? AND unique_key IS NULL AND id NOT IN (SELECT id FROM keep_event_ids) LIMIT ?)",
+                        (kind, batch)).rowcount
+                    total += removed
+                    db.commit()
+                    if removed < batch:
+                        break
+                db.execute("DELETE FROM keep_event_ids")
+        for kind in ('local_search_device', 'local_search_gpu'):
+            with self.connection() as db:
+                total += db.execute(
+                    "DELETE FROM events WHERE kind=? AND unique_key IS NULL AND id < (SELECT max(id) FROM events WHERE kind=?)",
+                    (kind, kind)).rowcount
+        if total:
+            diagnostic_event('storage.events.deduplicated', count=total)
+        return total
+
+    @observed('storage.prune_events')
+    def prune_events(self, max_age=EVENT_RETENTION_SECONDS, batch=2000):
+        """按 TTL 批量回收事件：非通知事件直接过期；已投递通知同样回收，未投递通知保留。"""
+        cutoff = time.time() - max_age
+        total = 0
+        while True:
+            with self.connection() as db:
+                removed = db.execute(
+                    "DELETE FROM events WHERE id IN (SELECT id FROM events "
+                    "WHERE created<? AND (kind!='notification' OR delivered=1) LIMIT ?)",
+                    (cutoff, batch)).rowcount
+            total += removed
+            if removed < batch:
+                break
+        if total:
+            diagnostic_event('storage.events.pruned', count=total, retention_seconds=max_age)
+        return total
+
+    @observed('storage.compact')
+    def compact(self, minimum_bytes=COMPACT_MINIMUM_BYTES):
+        """回收已删除行遗留的空闲页；空闲空间不多时不做全库重写。"""
+        with self.lock:
+            probe = sqlite3.connect(self.path, timeout=30)
+            try:
+                page_size = probe.execute('PRAGMA page_size').fetchone()[0]
+                free = probe.execute('PRAGMA freelist_count').fetchone()[0] * page_size
+            finally:
+                probe.close()
+            if free < minimum_bytes:
+                return 0
+            # VACUUM 不能在事务内执行，使用独立连接并先截断 WAL。
+            db = sqlite3.connect(self.path, timeout=30, isolation_level=None)
+            try:
+                db.execute('PRAGMA journal_mode=WAL')
+                db.execute('PRAGMA wal_checkpoint(TRUNCATE)')
+                db.execute('VACUUM')
+            finally:
+                db.close()
+        diagnostic_event('storage.compacted', freed_bytes=free)
+        return free
+
+    @observed('storage.repair')
+    def repair_oversized(self, max_database_bytes=MAINTENANCE_MAX_DATABASE_BYTES):
+        """重建过大的库：保留 records 与未投递提醒，丢弃可再生的 events。
+
+        对遗留巨型库，原地 DELETE + VACUUM 会把 WAL 放大到库体积且长时间占锁；
+        这里改为把少量存活数据复制到新库再原子替换，耗时与库体积无关。
+        返回重建前的库大小（字节），未触发或失败返回 0。
+        """
+        with self.lock:
+            try:
+                probe = sqlite3.connect(self.path, timeout=30)
+                try:
+                    database_bytes = self._database_bytes(probe)
+                finally:
+                    probe.close()
+                if database_bytes <= max_database_bytes:
+                    return 0
+
+                temporary = self.path.with_name(self.path.name + '.repair')
+                for suffix in ('', '-wal', '-shm'):
+                    try:
+                        os.remove(str(temporary) + suffix)
+                    except FileNotFoundError:
+                        pass
+
+                source = sqlite3.connect(self.path, timeout=30)
+                target = sqlite3.connect(temporary, timeout=30)
+                try:
+                    source.execute('PRAGMA wal_checkpoint(TRUNCATE)')
+                    target.executescript(SCHEMA_SQL)
+                    target.executemany(
+                        'INSERT INTO records(kind,id,account,body,updated) VALUES(?,?,?,?,?)',
+                        source.execute('SELECT kind,id,account,body,updated FROM records'))
+                    # 未投递提醒不可再生，随 records 一起保留；其余 events 只是进度快照。
+                    target.executemany(
+                        'INSERT INTO events(account,kind,body,unique_key,delivered,created) VALUES(?,?,?,?,?,?)',
+                        source.execute("SELECT account,kind,body,unique_key,delivered,created FROM events "
+                                       "WHERE kind='notification' AND delivered=0"))
+                    sequence = source.execute("SELECT seq FROM sqlite_sequence WHERE name='events'").fetchone()
+                    if sequence:
+                        target.execute("INSERT INTO sqlite_sequence(name,seq) VALUES('events',?)", (sequence[0],))
+                    target.commit()
+                finally:
+                    source.close()
+                    target.close()
+
+                # 先原子替换主库，再清理旧的 WAL/SHM：即使替换失败，原库仍完整。
+                for attempt in range(4):
+                    try:
+                        os.replace(temporary, self.path)
+                        break
+                    except OSError as error:
+                        if attempt == 3:
+                            raise
+                        diagnostic_event('storage.repair.retry', level=logging.WARNING, error=error)
+                        time.sleep(0.3 * (attempt + 1))
+                for suffix in ('-wal', '-shm'):
+                    try:
+                        os.remove(str(self.path) + suffix)
+                    except FileNotFoundError:
+                        pass
+                diagnostic_event('storage.repaired', bytes_before=database_bytes)
+                return database_bytes
+            except Exception as error:
+                for suffix in ('', '-wal', '-shm'):
+                    try:
+                        os.remove(str(self.path.with_name(self.path.name + '.repair')) + suffix)
+                    except FileNotFoundError:
+                        pass
+                diagnostic_event('storage.repair.failed', level=logging.WARNING, error=error)
+                return 0
+
+    def maintain(self, max_age=EVENT_RETENTION_SECONDS, minimum_bytes=COMPACT_MINIMUM_BYTES,
+                 max_database_bytes=MAINTENANCE_MAX_DATABASE_BYTES):
+        """启动维护：折叠遗留重复事件、回收过期事件，再按需压缩数据库文件。
+
+        超过 `max_database_bytes` 的遗留巨型库改为重建（保留 records 与未投递提醒），
+        避免原地删除/VACUUM 长时间占锁并放大 WAL。
+        """
+        with self.connection() as db:
+            database_bytes = self._database_bytes(db)
+        if database_bytes > max_database_bytes:
+            repaired = self.repair_oversized(max_database_bytes)
+            return 0, 0, repaired
+        deduplicated = self.prune_duplicate_events()
+        removed = self.prune_events(max_age)
+        freed = self.compact(minimum_bytes)
+        return deduplicated, removed, freed
+
+    @staticmethod
+    def _database_bytes(db):
+        page_size = db.execute('PRAGMA page_size').fetchone()[0]
+        page_count = db.execute('PRAGMA page_count').fetchone()[0]
+        return page_size * page_count
+
     @observed('storage.purge_account')
     def purge_account(self, account):
         with self.connection() as db:

src/wechat_decrypt_tool/local_search/downloads.py

@@ -131,7 +131,7 @@ def update(self, job, **values):
         job.update(values)
         job['updated'] = time.time()
         self.store.put('download', job, id=job['id'])
-        self.store.event('', 'local_search_download', job)
+        self.store.event('', 'local_search_download', job, unique_key=f'download:{job["id"]}', replace=True)
         if changed:
             diagnostic_event('download.state', level=logging.ERROR if job.get('status')=='error' else logging.INFO,
                              model=job['id'], status=job.get('status'), phase=job.get('stage'), bytes=job.get('bytes'), attempt=job.get('attempt'))

src/wechat_decrypt_tool/local_search/gpu.py

@@ -56,7 +56,7 @@ def status(self):
     def update(self, **values):
         job={**(self.store.get('gpu_component','global') or {}),**values,'updated':time.time()}
         self.store.put('gpu_component',job,id='global')
-        self.store.event('','local_search_gpu',job)
+        self.store.event('','local_search_gpu',job, unique_key='local_search_gpu', replace=True)
 
     @observed('gpu.start')
     async def start(self, source=None):

src/wechat_decrypt_tool/local_search/service.py

@@ -24,12 +24,15 @@
             'start': None, 'end': None, 'device': 'auto', 'device_id': 0, 'auto_update': True,
             'read_batch_size': 0, 'agent_global': False}
 
+# 进度事件里体积大且很少变化、或可从权威记录重建的字段，不随每次进度写入事件表。
+EVENT_OMITTED_FIELDS = frozenset({'config', 'coverage', 'segments', 'read_starts'})
+
 
 class LocalSearch(ProgressiveIndex, MessageTotals):
     def __init__(self, root=None, model_root=None, reader=None, engine=None):
         self.root = Path(root or get_output_dir() / 'local_search')
         self.store = AIStore(self.root)
-        self.engine = engine or LocalInference(callback=lambda status: self.store.event('', 'local_search_device', status))
+        self.engine = engine or LocalInference(callback=lambda status: self.store.event('', 'local_search_device', status, unique_key='local_search_device', replace=True))
         self.downloads = ModelDownloads(model_root or get_data_dir() / 'local_search_models', self.store, self.engine)
         from .gpu import GPUComponent
         self.gpu = GPUComponent(self.downloads.root.parent / 'local_search_gpu', self.store, self.engine)
@@ -103,7 +106,9 @@ def update(self, job, **changes):
         if job['id'] in self.restarting: job['resume_on_start']=True
         if job['account'] in self.revoked: return
         self.store.put('index_job', job, id=job['id'], account=job['account'])
-        self.store.event(job['account'], 'local_search_index', job)
+        # 事件表只保留每个任务的最新进度；完整快照以 records 表为准。
+        progress = {key: value for key, value in job.items() if key not in EVENT_OMITTED_FIELDS}
+        self.store.event(job['account'], 'local_search_index', progress, unique_key=f'index_job:{job["id"]}', replace=True)
 
     def enrichment_version(self, account):
         """只检查本地提取缓存，不触发媒体分析或网络访问。"""

src/wechat_decrypt_tool/local_search/totals.py

@@ -11,7 +11,7 @@ class MessageTotals:
     def save_message_total(self, job, **values):
         total = {'job_id': job['id'], 'updated': time.time(), **values}
         self.store.put('index_message_total', total, id=job['id'], account=job['account'])
-        self.store.event(job['account'], 'local_search_total', total)
+        self.store.event(job['account'], 'local_search_total', total, unique_key=f'index_total:{job["id"]}', replace=True)
 
     def message_plan(self, job):
         account_key = hashlib.sha256(job['account'].encode()).hexdigest()