Reconcile proven completed inputs without replay and hide internal fact notices

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']