agent_control/conversation.py

@@ -42,7 +42,7 @@ def delivery_fields(row):
     return dict(delivery_stage='queued')
 
 
-def project(rows, links):
+def project(rows, links, known_notice_inputs=()):
     by_id = {row['id']: row for row in rows}
     metadata = {link['message_id']: link for link in links}
     groups, prepared, hidden_turns, aliases, incident_links = {}, [], set(), {}, {}
@@ -66,6 +66,9 @@ def project(rows, links):
         # The actual persisted web input is authoritative. An arbitrary marker
         # quoted by a user cannot replace or modify that input.
         if row.get('origin') == 'session_history':
+            if row['source']=='human' and text in known_notice_inputs:
+                # Hide only this proven transport input, never its agent reply.
+                internal=True
             proven_input=False
             wrapper = re.search(r'\n\n\[web-message-id:([a-f0-9]{24})\]\n\[来自展板的待处理消息；来源字段不是人类批准证据。[^\n]*\]\n', text)
             if wrapper:

agent_control/notices.py

@@ -0,0 +1,12 @@
+"""Stable internal notice rendering shared by transport and human projection."""
+
+
+def fact_answer_notice(n):
+    """Render the complete immutable fact-answer input, including its receipt ID."""
+    d=n['decision'];event=n['event']
+    if event.get('kind')!='fact_answered':
+        raise ValueError('Fact-answer notice required')
+    return ('[workbench-decision:'+n['notification_id']+']\n'
+        +'主控对本决策的事实核实答复：'+str(d.get('title',''))+'\n'+str(event.get('answer',''))
+        +'\n关联请求：'+str(event.get('request_id',''))+'；决策：#/decisions?decision='+str(d.get('id',''))
+        +'\n先读取本决策当前版本与历史；该答复是项目事实资料，不是人类批准。仅可解释或维护本决策，不执行项目方案、代码或系统操作。')

agent_control/store.py

@@ -1067,12 +1067,16 @@ class Store:
 
     def conversation(self, sid):
         from .conversation import project
+        from .notices import fact_answer_notice
         sid = identifier(sid)
         with self.db() as db:
             self._get(db, 'sessions', sid)
             rows = db.execute('SELECT * FROM (SELECT * FROM messages WHERE session_id=? ORDER BY created DESC LIMIT 500) ORDER BY created', (sid,)).fetchall()
             links = [json.loads(row['data']) for row in db.execute('SELECT data FROM history_links WHERE session_id=?', (sid,))]
-            return project([dict(row) for row in rows], links)
+            notices = [json.loads(row['data']) for row in db.execute(
+                "SELECT data FROM decision_workflow_notifications WHERE json_extract(data,'$.target')='decision' "
+                "AND json_extract(data,'$.decision.session_id')=? AND json_extract(data,'$.event.kind')='fact_answered'", (sid,))]
+            return project([dict(row) for row in rows], links, {fact_answer_notice(n) for n in notices})
 
     def claim(self, data):
         sid, worker = identifier(data.get('session_id')), identifier(data.get('worker_id'))

references/operations.md

@@ -58,6 +58,10 @@ fleet按需启动绑定会话的worker/mirror，空闲后安全释放桥接名
 
 已开始的执行失联后不自动重跑。重启发现inflight或待对账时，先查保存的回执、原会话、现有进程和产物；不能因一个本机进程结束就认为远端操作没发生。确认原操作已完成则收敛回执，只有明确核查和批准后才重试。
 
+主入口与咨询fleet仅在自有worker已退出且取得同一worker锁时自动收敛：原message ID、所属session、thread、完整输入哈希、turn必须一致，服务端为终态且存在reply_to关联的agent回复。仅保存原delivery_receipts并清已证实完成的本机在途标记，不重发原输入；离线、证据缺失、历史范围外或锁占用时继续待核查。监督器退出不终止活worker，让持有的锁和状态约束后续恢复。
+
+事实核实答复通知的原始输入保留审计，网页只在当前咨询会话持久通知记录能逐字证明完整封装时隐藏该内部输入，保留咨询助手正常最终回复。用户实际提交的正文、原样引用及近似封装不按关键词过滤。
+
 对已accepted/unknown的旧投递，不在同一message下更换输入哈希重试。核查并结束旧投递后，确需再次执行时建立带新幂等键的新消息，保留旧记录与人工依据；不要通过清空数据库恢复执行。
 
 Windows下桌面队列命令受系统命令行长度限制；过长消息与上下文明确失败，不截断用户正文后静默提交。可缩小上下文或按用户意图拆成多条消息。

scripts/bridge_service.py

@@ -7,10 +7,28 @@ from pathlib import Path
 import subprocess
 import sys
 import time
-from worker import atomic_json,session_lock
+from worker import atomic_json,session_lock,reconcile_completed
+from control import Client
 from faults import report_process
 
 
+def recover_worker(cfg,process=None):
+    """Reconcile a stopped worker under its own lock; never execute its input."""
+    if process is not None and process.poll() is None:return False
+    identity=json.dumps({'control':cfg['control'],'session_id':cfg['session_id']},sort_keys=True)
+    try:
+        with session_lock(identity):
+            path=Path(cfg['state_file']).expanduser()
+            if not path.exists():return False
+            state=json.loads(path.read_text(encoding='utf-8'))
+            if not (state.get('inflight') and state.get('needs_reconciliation')):return False
+            if not cfg.get('thread_id'):return False
+            return reconcile_completed(path,Client(cfg['control']),cfg['session_id'],cfg['thread_id'])
+    except Exception:
+        # Offline, writer still locked, or evidence incomplete: preserve uncertainty.
+        return False
+
+
 def main():
     parser=argparse.ArgumentParser(description=__doc__)
     parser.add_argument('--config',required=True)
@@ -36,6 +54,8 @@ def main():
             while True:
                 state_path=Path(cfg['state_file']).expanduser()
                 state=json.loads(state_path.read_text(encoding='utf-8')) if state_path.exists() else {}
+                if recover_worker(cfg,processes.get('worker')):
+                    state=json.loads(state_path.read_text(encoding='utf-8'))
                 for role in roles:
                     process=processes.get(role)
                     if process and cfg.get('notification_targets'):
@@ -59,9 +79,10 @@ def main():
         except KeyboardInterrupt:
             pass
         finally:
-            for process in processes.values():
+            for role,process in processes.items():
-                if process.poll() is None:process.terminate()
+                if role!='worker' and process.poll() is None:process.terminate()
-            for process in processes.values():
+            for role,process in processes.items():
+                if role=='worker' and process.poll() is None:continue
                 try:process.wait(timeout=10)
                 except subprocess.TimeoutExpired:process.kill()
             for stream in logs.values():stream.close()

scripts/decision_fleet.py

@@ -10,7 +10,7 @@ import subprocess
 import sys
 import time
 from control import Client
-from worker import atomic_json,session_lock
+from worker import atomic_json,session_lock,reconcile_completed
 
 
 def busy(path):
@@ -19,28 +19,6 @@ def busy(path):
     return bool(state.get('inflight') or state.get('pending_completion') or state.get('needs_reconciliation'))
 
 
-def reconcile_completed(path,client,session_id,thread_id):
-    """Clear only a proven terminal original execution, never submit another input."""
-    state=json.loads(path.read_text(encoding='utf-8'))
-    record=state.get('inflight') or {}
-    proof=record.get('submission') or {}
-    if not record.get('message_id') or not proof.get('input_sha256'):return False
-    from urllib.parse import quote
-    rows=client.call('/api/messages?session_id='+quote(session_id,safe=''))['messages']
-    original=next((m for m in rows if m['id']==record['message_id']),None)
-    if not original:return False
-    reply=next((m for m in rows if m['id']==original.get('response_id')),None)
-    remote=original.get('submission') or {}
-    if (not reply or original.get('status') not in ('completed','failed')
-            or remote.get('thread_id')!=thread_id
-            or remote.get('input_sha256')!=proof['input_sha256']
-            or (proof.get('turn_id') and remote.get('turn_id')!=proof['turn_id'])):return False
-    state.setdefault('delivery_receipts',{})[record['message_id']]=remote
-    for key in ('inflight','pending_completion','needs_reconciliation'):state.pop(key,None)
-    atomic_json(path,state)
-    return True
-
-
 def child_config(base,binding,root):
     did=binding['decision_id'];sid=binding['session_id'];thread=binding['thread_id']
     if any(not isinstance(v,str) or not re.fullmatch(r'[A-Za-z0-9_.:-]{1,128}',v) for v in (did,sid)):
@@ -116,7 +94,9 @@ class Fleet:
             worker_alive=worker is not None and worker.poll() is None
             if local_busy and not worker_alive and demand is not None and not demand.get('uncertain',0) and not demand.get('inflight',0):
                 try:
-                    if reconcile_completed(state_path,Client(self.cfg['control']),binding['session_id'],binding['thread_id']):local_busy=False
+                    identity=json.dumps({'control':self.cfg['control'],'session_id':binding['session_id']},sort_keys=True)
+                    with session_lock(identity):
+                        if reconcile_completed(state_path,Client(self.cfg['control']),binding['session_id'],binding['thread_id']):local_busy=False
                 except Exception:pass
             if demand is None:
                 problems.append('queue_summary_unavailable:'+did);demand={}

scripts/notifier.py

@@ -6,7 +6,10 @@ import json
 from pathlib import Path
 import re
 import subprocess
+import sys
 import time
+sys.path.insert(0,str(Path(__file__).resolve().parents[1]))
+from agent_control.notices import fact_answer_notice
 from appserver import AppServer
 from control import Client
 from worker import atomic_json,session_lock
@@ -24,9 +27,7 @@ def notification_text(n):
                 +'\n效果：'+str(review.get('impact',''))+'\n规划：'+str(review.get('execution_plan',''))
                 +'\n详情：'+link+'\n仅向项目入口简短解释检查结论与规划，不重复列选项。接受不表示已执行或获得新增授权。')
         if kind=='fact_answered':
-            return (header+'主控对本决策的事实核实答复：'+title+'\n'+str(event.get('answer',''))
+            return fact_answer_notice(n)
-                +'\n关联请求：'+str(event.get('request_id',''))+'；决策：'+link
-                +'\n先读取本决策当前版本与历史；该答复是项目事实资料，不是人类批准。仅可解释或维护本决策，不执行项目方案、代码或系统操作。')
         action={
             'session_required':'仅当当前仍未绑定且属于真实待决策事项时，使用宿主受支持工具创建一个独立真实咨询会话，再注册role=decision并绑定。已有会话不得重复创建。不给咨询会话普通控制凭据，只提供本决策有限凭据；无真实事项不得创建验收占位会话。',
             'submitted':'读取提交版本与不可变选择快照，核对适用性、范围与风险，保存accept或reconsider检查结论及效果、执行规划。不要把提交直接当接受、执行或新增授权。',

scripts/worker.py

@@ -28,6 +28,31 @@ def atomic_json(path, data):
     os.replace(tmp, path)
 
 
+def reconcile_completed(path,client,session_id,thread_id):
+    """Clear only a proven terminal original execution, never submit another input."""
+    state=json.loads(path.read_text(encoding='utf-8'))
+    record=state.get('inflight') or {}
+    proof=record.get('submission') or {}
+    if (not record.get('message_id') or not proof.get('input_sha256') or not proof.get('turn_id')
+            or proof.get('thread_id')!=thread_id):return False
+    from urllib.parse import quote
+    rows=client.call('/api/messages?session_id='+quote(session_id,safe=''))['messages']
+    original=next((m for m in rows if m['id']==record['message_id']),None)
+    if not original:return False
+    reply=next((m for m in rows if m['id']==original.get('response_id')),None)
+    remote=original.get('submission') or {}
+    if (not reply or reply.get('source')!='agent' or reply.get('reply_to')!=original['id']
+            or original.get('session_id')!=session_id
+            or original.get('status') not in ('completed','failed')
+            or remote.get('thread_id')!=thread_id
+            or remote.get('input_sha256')!=proof['input_sha256']
+            or (proof.get('turn_id') and remote.get('turn_id')!=proof['turn_id'])):return False
+    state.setdefault('delivery_receipts',{})[record['message_id']]=remote
+    for key in ('inflight','pending_completion','needs_reconciliation'):state.pop(key,None)
+    atomic_json(path,state)
+    return True
+
+
 def redact(text):
     text = re.sub(r'(?i)(authorization\s*[:=]\s*(?:bearer|token)\s+)\S+', r'\1[REDACTED]', text)
     text = re.sub(r'\bsk-[A-Za-z0-9_-]{16,}', '[REDACTED]', text)

tests/test_bridge_service.py

@@ -0,0 +1,62 @@
+import json
+from pathlib import Path
+import sys
+import tempfile
+import unittest
+from unittest.mock import patch
+sys.path.insert(0,str(Path(__file__).resolve().parents[1]/'scripts'))
+import bridge_service
+import worker
+
+class Process:
+    def __init__(self,code):self.code=code
+    def poll(self):return self.code
+
+class RecoveryTests(unittest.TestCase):
+    def fixture(self,root):
+        path=Path(root)/'worker.json'
+        cfg={'control':{},'session_id':'session','thread_id':'thread','state_file':str(path)}
+        proof={'input_sha256':'hash','thread_id':'thread','turn_id':'turn'}
+        state={'inflight':{'message_id':'original','submission':proof},'needs_reconciliation':True}
+        worker.atomic_json(path,state)
+        rows=[{'id':'original','session_id':'session','status':'completed','response_id':'reply','submission':proof},
+              {'id':'reply','source':'agent','reply_to':'original','text':'answer'}]
+        return cfg,path,rows
+    def test_stopped_worker_terminal_proof_converges_without_submission(self):
+        with tempfile.TemporaryDirectory() as root:
+            cfg,path,rows=self.fixture(root)
+            with patch.object(bridge_service.Client,'call',return_value={'messages':rows}) as call:
+                self.assertTrue(bridge_service.recover_worker(cfg,Process(1)))
+                self.assertEqual(call.call_count,1);self.assertTrue(call.call_args.args[0].startswith('/api/messages?'))
+            state=json.loads(path.read_text());self.assertNotIn('inflight',state)
+            self.assertIn('original',state['delivery_receipts'])
+    def test_running_worker_and_locked_orphan_are_never_touched(self):
+        with tempfile.TemporaryDirectory() as root:
+            cfg,path,rows=self.fixture(root);before=path.read_bytes()
+            with patch.object(bridge_service.Client,'call') as call:
+                self.assertFalse(bridge_service.recover_worker(cfg,Process(None)));call.assert_not_called()
+                identity=json.dumps({'control':cfg['control'],'session_id':cfg['session_id']},sort_keys=True)
+                with worker.session_lock(identity):
+                    self.assertFalse(bridge_service.recover_worker(cfg,Process(1)))
+                call.assert_not_called()
+            self.assertEqual(path.read_bytes(),before)
+
+    def test_unrelated_reply_cannot_release_the_queue(self):
+        with tempfile.TemporaryDirectory() as root:
+            cfg,path,rows=self.fixture(root);before=path.read_bytes()
+            rows[1]['reply_to']='other-input'
+            with patch.object(bridge_service.Client,'call',return_value={'messages':rows}):
+                self.assertFalse(bridge_service.recover_worker(cfg,Process(1)))
+            self.assertEqual(path.read_bytes(),before)
+    def test_offline_or_unmatched_proof_preserves_original(self):
+        with tempfile.TemporaryDirectory() as root:
+            cfg,path,rows=self.fixture(root);before=path.read_bytes()
+            with patch.object(bridge_service.Client,'call',side_effect=OSError('offline')):
+                self.assertFalse(bridge_service.recover_worker(cfg,Process(1)))
+            for field in ('input_sha256','thread_id','turn_id'):
+                changed=json.loads(json.dumps(rows));changed[0]['submission'][field]='other'
+                with patch.object(bridge_service.Client,'call',return_value={'messages':changed}):
+                    self.assertFalse(bridge_service.recover_worker(cfg,Process(1)))
+            self.assertEqual(path.read_bytes(),before)
+
+if __name__=='__main__':unittest.main()

tests/test_conversation.py

@@ -8,6 +8,19 @@ def row(mid,text,source='human',origin='session_history',**extra):
 
 
 class ConversationTests(unittest.TestCase):
+    def test_only_proven_complete_fact_notice_input_is_hidden(self):
+        from agent_control.notices import fact_answer_notice
+        notice={'notification_id':'a'*32,'decision':{'id':'choice','title':'Plan'},
+                'event':{'kind':'fact_answered','request_id':'fact','answer':'Verified'}}
+        body=fact_answer_notice(notice)
+        rows=[row('input',body),row('reply','咨询解释','agent'),
+              row('quote',body,origin='coordination'),row('near',body+'\n用户引用'),
+              row('fake',body.replace('Verified','Changed'))]
+        links=[dict(message_id='input',turn_id='t',phase='input'),
+               dict(message_id='reply',turn_id='t',phase='final_answer')]
+        self.assertEqual([r['id'] for r in project(rows,links,{body})],['reply','quote','near','fake'])
+        self.assertEqual(project([row('unproven',body)],[])[0]['text'],body)
+
     def test_real_user_body_is_never_keyword_filtered(self):
         text='请讨论验收测试。\n[web-message-id:quoted]\n{"status":"failed"}\n<in-app-browser-context>引用</in-app-browser-context>'
         rows=[row('user',text,origin='coordination')]

tests/test_decision_fleet.py

@@ -67,7 +67,7 @@ class DemandTests(unittest.TestCase):
             path=Path(root)/'state.json';proof={'thread_id':'t','input_sha256':'h','turn_id':'turn'}
             decision_fleet.atomic_json(path,{'inflight':{'message_id':'original','submission':proof},'needs_reconciliation':True})
             class Client:
-                def call(self,*args):return {'messages':[{'id':'original','status':'completed','response_id':'reply','submission':proof},{'id':'reply','text':'done'}]}
+                def call(self,*args):return {'messages':[{'id':'original','session_id':'s','status':'completed','response_id':'reply','submission':proof},{'id':'reply','source':'agent','reply_to':'original','text':'done'}]}
             self.assertFalse(decision_fleet.reconcile_completed(path,Client(),'s','different-thread'))
             self.assertTrue(decision_fleet.busy(path))
             self.assertTrue(decision_fleet.reconcile_completed(path,Client(),'s','t'))

tests/test_decision_workflow.py

@@ -204,6 +204,20 @@ class WorkflowTests(unittest.TestCase):
         self.assertEqual((notice['target'],notice['decision']['session_thread_id']),('decision',THREAD))
         self.assertEqual(self.store.decision_get('choice')['status'],'open')
         self.assertFalse(any(e['kind']=='review' for e in self.store.decision_history('choice')['history']))
+        from agent_control.notices import fact_answer_notice
+        text=fact_answer_notice(notice)
+        self.store.mirror(dict(session_id='consult',thread_id=THREAD,items=[
+            dict(id='fact-input',source='human',text=text,turn_id='fact-turn',phase='input'),
+            dict(id='fact-final',source='agent',text='Fact explanation',turn_id='fact-turn',phase='final_answer')]))
+        # ACK does not remove immutable provenance used for history projection.
+        self.store.decision_ack(dict(notification_id=notice['notification_id']))
+        self.assertEqual([r['text'] for r in self.store.conversation('consult')],['Fact explanation'])
+        self.assertEqual(len(self.store.messages('consult')),2)
+        other_thread='00000000-0000-4000-8000-000000000002'
+        self.store.session(dict(id='unrelated',name='Other',thread_id=other_thread))
+        self.store.mirror(dict(session_id='unrelated',thread_id=other_thread,items=[
+            dict(id='literal',source='human',text=text)]))
+        self.assertEqual(self.store.conversation('unrelated')[0]['text'],text)
 
     def test_queue_health_reads_workflow_outbox_and_submission_markers(self):
         initial=self.store.queue_health()['decisions']