perf(ai): 按业务边界持久化，降低回复与索引进度写盘

src/wechat_decrypt_tool/ai/agent_service.py

@@ -42,6 +42,7 @@ def __init__(self, ai, tools=None, model=None):
         self.readers = {}
         self.index_workers = {}
         self.reference_contacts = {}
+        self._draft_saved_at = {}
         from .deep_subtasks import DeepSubtasks
         self.subtasks = DeepSubtasks(self)
 
@@ -68,14 +69,20 @@ def run(self, id, account=None):
         return record
 
     @serialized
-    def update(self, id, **fields):
+    def update(self, id, *, transient=False, emit=True, **fields):
+        with self.store.lock if transient else self.store.connection():
+            return self._update(id, transient=transient, emit=emit, **fields)
+
+    def _update(self, id, *, transient, emit, **fields):
         record = self.run(id)
         values = fields.pop('evidence',None)
         if values is not None and not isinstance(values,Evidence):
             record['evidence'].replace(values)
         record.pop('evidence',None)
+        if all(record.get(key) == value for key, value in fields.items()) and not self.store.has_live_record('agent_run', id):
+            return self.run(id)
         record.update(fields, updated_at=time.time())
-        self.store.put('agent_run', record)
+        self.store.put('agent_run', record, transient=transient)
         if 'version' in fields or 'status' in fields:
             diagnostic_event('agent.run.state', run_id=id, thread_id=record['thread_id'], version=record['version'], status=record['status'])
         event = {'type': 'run_patch', 'run_id': id, 'thread_id': record['thread_id'], 'status': record['status'],
@@ -104,9 +111,24 @@ def update(self, id, **fields):
             }
         if patch:
             event['patch'] = patch
-        self.store.event(record['account'], 'agent', event)
+        if emit and (patch or any(key in fields for key in ('context_budget', 'analysis', 'status', 'version'))):
+            self.store.event(record['account'], 'agent', event, transient=transient,
+                             unique_key=f'run_patch:{id}' if transient else None)
         return self.run(id)
 
+    @serialized
+    def stream_answer(self, id, text, stage):
+        """实时正文只推送内存事件；每两秒及终态保存可恢复草稿。"""
+        run = self.guard(id)
+        now = time.monotonic()
+        key = (id, run['version'])
+        saved_at = self._draft_saved_at.setdefault(key, now)
+        persist = now - saved_at >= 2.0
+        self.timeline_item(id, 'answer', text, item_id='answer:' + id, status='running',
+                           persist=persist, transient_event=True, run_fields={'answer': text, 'stage': stage})
+        if persist:
+            self._draft_saved_at[key] = now
+
     def guard(self, id):
         run = self.run(id)
         if self.stopping or run['status'] not in ACTIVE:
@@ -190,10 +212,16 @@ def profile(self, run, vision=False):
         return snapshot | {'api_key': live.get('api_key', '')}
 
     @observed('agent.finish', id_field='run_id')
+    @serialized
     def finish(self, id, status, error='', error_info=None):
+        with self.store.connection():
+            return self._finish(id, status, error, error_info)
+
+    def _finish(self, id, status, error, error_info):
         run = self.store.get('agent_run', id)
         if not run:
             return
+        self._draft_saved_at = {key: value for key, value in self._draft_saved_at.items() if key[0] != id}
         if status == 'completed' and run.get('delegation_partial'):
             status = 'interrupted'
             error = '部分子任务未完成，当前为阶段结果，可继续分析。'

src/wechat_decrypt_tool/ai/agent_timeline.py

@@ -2,15 +2,33 @@
 import time
 import uuid
 import re
+from contextlib import nullcontext
+from .deep_synchronization import serialized
 
 
 class AgentTimeline:
-    def timeline_item(self, id, kind, text, *, item_id=None, status='completed', **fields):
+    @serialized
+    def timeline_item(self, id, kind, text, *, item_id=None, status='completed', persist=True,
+                      transient_event=False, run_fields=None, **fields):
+        # 业务时间线、任务快照和持久事件在同一事务提交，避免一个动作多次刷盘。
+        with self.store.connection() if persist else nullcontext():
+            return self._timeline_item(id, kind, text, item_id=item_id, status=status, persist=persist,
+                                       transient_event=transient_event, run_fields=run_fields, **fields)
+
+    def _timeline_item(self, id, kind, text, *, item_id, status, persist, transient_event, run_fields, **fields):
         run = self.run(id)
         items = run.get('timeline') or [dict(x, seq=i+1, revision=1, kind='status') for i,x in enumerate(run.get('activity', []))]
         now = time.time()
         item = next((x for x in items if x['id'] == item_id), None)
         previous_text = item.get('text', '') if item else ''
+        if (item and item.get('text') == text and item.get('status') == status
+                and all(item.get(k) == v for k, v in fields.items())
+                and all(run.get(k) == v for k, v in (run_fields or {}).items())):
+            if persist and self.store.has_live_record('agent_run', id):
+                self.update(id, emit=False, **(run_fields or {}))
+                if hasattr(self, 'workspace'):
+                    self.workspace.put(id, run['version'], f'timeline:{item["seq"]:012d}', 'timeline', item)
+            return item['id']
         if item is None:
             item = dict(id=item_id or uuid.uuid4().hex, seq=max(run.get('timeline_seq',0),max((x.get('seq',0) for x in items),default=0)) + 1, revision=0,
                         kind=kind, started_at=now, input_version=run['version'])
@@ -19,12 +37,18 @@ def timeline_item(self, id, kind, text, *, item_id=None, status='completed', **f
         if status not in ('running', 'received'):
             item['finished_at'] = now
         # 普通过程保留最近 200 步；压缩节点是持久的对话分隔，不能随步骤淘汰。
-        if hasattr(self,'workspace'):
+        if persist and hasattr(self,'workspace'):
             self.workspace.put(id,run['version'],f'timeline:{item["seq"]:012d}','timeline',item)
         retained = [x for i, x in enumerate(items) if i >= len(items) - 200 or
                     (x.get('kind') == 'notice' and x.get('context_job', {}).get('before') is not None)]
-        self.update(id, timeline=retained,timeline_seq=max(x.get('seq',0) for x in items))
+        self.update(id, transient=not persist, emit=False, timeline=retained,
+                    timeline_seq=max(x.get('seq',0) for x in items), **(run_fields or {}))
         event = {'type':'timeline_item', 'run_id':id, 'thread_id':run['thread_id'], 'version':run['version'], 'timeline_item':item}
+        if run_fields:
+            current = self.run(id)
+            event['updated_at'] = current['updated_at']
+            # 正文仍从时间线合并，阶段信息同一个事件送达，避免额外写入 run_patch。
+            event['patch'] = {key: current[key] for key in run_fields if key != 'answer'}
         if kind in ('answer', 'progress'):
             # 新标记与已校验身份同时送达，避免正文先出现、来源等待慢速快照。
             # 只发送本次新增标记，完整映射仍由运行快照和历史保存负责。
@@ -35,7 +59,8 @@ def timeline_item(self, id, kind, text, *, item_id=None, status='completed', **f
                 from .agent_references import cited_references
                 event['citations'] = self.citations({**run, 'answer': added, 'timeline': [], 'answer_context': {}})
                 event['references'] = cited_references(added, run.get('references', {}))
-        self.store.event(run['account'], 'agent', event)
+        self.store.event(run['account'], 'agent', event, transient=transient_event,
+                         unique_key=f'timeline:{id}:{item["id"]}' if transient_event else None)
         return item['id']
 
     def close_activity(self, id, status='completed'):

src/wechat_decrypt_tool/ai/deep_runtime.py

@@ -802,8 +802,7 @@ async def execute(self, id):
                                 text = delta.get('text', '') if delta.get('type') == 'text-delta' else ''
                                 partial += text
                                 if text and time.monotonic() - last_emit >= .08:
-                                    self.update(id, answer=partial, stage='提取局部事实' if run.get('parent_run_id') and run.get('subtask_plan_version') == REVISION else '正在回答')
-                                    self.timeline_item(id, 'answer', partial, item_id='answer:' + id, status='running')
+                                    self.stream_answer(id, partial, '提取局部事实' if run.get('parent_run_id') and run.get('subtask_plan_version') == REVISION else '正在回答')
                                     last_emit = time.monotonic()
                 final = await agent.aget_state(config)
                 answer = next((m for m in reversed(final.values.get('messages', [])) if isinstance(m, AIMessage) and not m.tool_calls), None)

src/wechat_decrypt_tool/ai/lifecycle.py

@@ -92,3 +92,6 @@ async def stop_services():
             await factory().stop()
         except Exception as error:
             event('lifecycle.service.stop_failed', level=logging.ERROR, component=name, error=error)
+    # Agent 与摘要共享业务库，全部工作线程退出后才关闭持久连接。
+    for store in (get_ai_service().store, get_local_search().store):
+        store.close()

src/wechat_decrypt_tool/ai/runtime_check.py

@@ -4,6 +4,7 @@
 import platform
 import sys
 import tempfile
+from contextlib import closing
 from pathlib import Path
 
 
@@ -40,26 +41,28 @@ def bind_tools(self, tools, **kwargs):
     from .providers import ModelService
     from .service import AIService
     from .agent_service import AgentService
-    store = AIStore(root / 'application')
-    store.put('profile', {'model': 'synthetic', 'name': '冻结验收', 'protocol': 'openai',
-        'base_url': 'http://127.0.0.1:1/v1', 'api_key': 'synthetic-unused-key',
-        'model_overrides': {'context_window': 32768}}, id='synthetic')
-    store.put('defaults', {'text': 'synthetic'}, id='global')
-    models = ModelService(store)
-    models.client = lambda profile: RuntimeModel(messages=iter([AIMessage(content='运行正常')]))
-    class NoQueries:
-        async def conversations(self, account):
-            raise AssertionError('冻结问候检查不能读取聊天目录')
-    service = AgentService(AIService(store, models), tools=NoQueries())
-    thread = await service.create_thread('synthetic', '', '运行检查')
-    run = await service.submit(thread['id'], 'synthetic', {'text': '你好', 'request_id': 'smoke'})
-    await service.workers[run['id']]
-    result = service.public_run(run['id'], 'synthetic')
-    assert result['status'] == 'completed', result.get('error')
-    assert result['answer'] == '运行正常'
-    assert result['used']['models'] == 1 and result['used']['tools'] == 0
-    assert result['engine'] == 'deepagents' and result['engine_version'] == 3
-    await service.stop()
+    with closing(AIStore(root / 'application')) as store:
+        store.put('profile', {'model': 'synthetic', 'name': '冻结验收', 'protocol': 'openai',
+            'base_url': 'http://127.0.0.1:1/v1', 'api_key': 'synthetic-unused-key',
+            'model_overrides': {'context_window': 32768}}, id='synthetic')
+        store.put('defaults', {'text': 'synthetic'}, id='global')
+        models = ModelService(store)
+        models.client = lambda profile: RuntimeModel(messages=iter([AIMessage(content='运行正常')]))
+        class NoQueries:
+            async def conversations(self, account):
+                raise AssertionError('冻结问候检查不能读取聊天目录')
+        service = AgentService(AIService(store, models), tools=NoQueries())
+        try:
+            thread = await service.create_thread('synthetic', '', '运行检查')
+            run = await service.submit(thread['id'], 'synthetic', {'text': '你好', 'request_id': 'smoke'})
+            await service.workers[run['id']]
+            result = service.public_run(run['id'], 'synthetic')
+            assert result['status'] == 'completed', result.get('error')
+            assert result['answer'] == '运行正常'
+            assert result['used']['models'] == 1 and result['used']['tools'] == 0
+            assert result['engine'] == 'deepagents' and result['engine_version'] == 3
+        finally:
+            await service.stop()
 
 
 def check_runtime(model_root=None):
@@ -90,9 +93,10 @@ def check_runtime(model_root=None):
     assert 'cl100k_base' in tiktoken.list_encoding_names()
     with tempfile.TemporaryDirectory(prefix='wechat-ai-runtime-') as directory:
         root = Path(directory)
-        store = AIStore(root / 'business')
-        store.put('runtime_check', {'ok': True}, id='check', account='synthetic')
-        assert store.get('runtime_check', 'check')['ok']
+        # 自检持有的业务连接必须在删除临时目录前释放，Windows 不允许删除打开的库。
+        with closing(AIStore(root / 'business')) as store:
+            store.put('runtime_check', {'ok': True}, id='check', account='synthetic')
+            assert store.get('runtime_check', 'check')['ok']
         asyncio.run(_checkpoint(root))
         report['application_graph'] = 'deepagents-v3-one-call-no-query'
         index = SemanticIndex(root / 'vectors.sqlite3')