Schedule consultation bridges on demand and preserve delivery evidence

tests/test_decision_bridge.py

@@ -25,7 +25,8 @@ class DecisionBridgeTests(unittest.TestCase):
         cfg={'control':{},'codex':'codex','state_file':str(Path(root)/'worker.json'),
              'decision_fleet':{'runtime_root':str(Path(root)/'decisions'),'enabled':True,'max_active':1}}
         binding={'decision_id':'d','session_id':'s','thread_id':'11111111-1111-4111-8111-111111111111',
-                 'worktree':root,'status':'open','archive_status':'none'}
+                 'worktree':root,'status':'open','archive_status':'none',
+                 'demand':{'queued':1,'human_queued':1,'inflight':0,'uncertain':0}}
         return cfg,binding
 
     def test_only_bound_threads_and_bounded_startup_are_bridged(self):
@@ -65,6 +66,7 @@ class DecisionBridgeTests(unittest.TestCase):
                 self.assertIn('d',fleet.children)
                 self.assertTrue(all(p.poll() is None for p in child['processes'].values()))
                 atomic_json(state,{})
+                for process in child['processes'].values():process.code=0
                 fleet.tick([{**b,'status':'completed','archive_status':'archived'}])
                 self.assertEqual(fleet.children,{})
                 self.assertTrue(state.exists())

tests/test_decision_fleet.py

@@ -0,0 +1,76 @@
+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 decision_fleet
+from test_decision_bridge import Process
+import test_decision_bridge
+
+class DemandTests(unittest.TestCase):
+    def setup_fleet(self,root):return test_decision_bridge.DecisionBridgeTests().fixture(root)
+    def test_idle_bindings_start_nothing(self):
+        with tempfile.TemporaryDirectory() as root:
+            cfg,b=self.setup_fleet(root);b['demand']['queued']=0
+            f=decision_fleet.Fleet(cfg)
+            with patch.object(decision_fleet.subprocess,'Popen') as spawn:
+                self.assertEqual(f.tick([b])['active'],0);spawn.assert_not_called()
+    def test_one_message_cycle_releases_and_old_waiter_wins(self):
+        with tempfile.TemporaryDirectory() as root:
+            cfg,b=self.setup_fleet(root);f=decision_fleet.Fleet(cfg)
+            other={**b,'decision_id':'z','session_id':'z','thread_id':'22222222-2222-4222-8222-222222222222'}
+            with patch.object(decision_fleet.subprocess,'Popen',side_effect=lambda *a,**k:Process()) as spawn:
+                with patch.object(decision_fleet.time,'time',return_value=100):f.tick([b,other])
+                self.assertIn('--once',spawn.call_args.args[0])
+                for p in f.children['d']['processes'].values():p.code=0
+                with patch.object(decision_fleet.time,'time',return_value=110):f.tick([b,other])
+                self.assertEqual(set(f.children),{'z'});f.close()
+    def test_old_ordinary_precedes_fresh_human_after_restart(self):
+        with tempfile.TemporaryDirectory() as root:
+            cfg,b=self.setup_fleet(root);b['demand']={**b['demand'],'human_queued':0,'oldest_queued_at':1}
+            f=decision_fleet.Fleet(cfg)
+            other={**b,'decision_id':'a','session_id':'a','thread_id':'22222222-2222-4222-8222-222222222222','demand':{'queued':1,'human_queued':1}}
+            with patch.object(decision_fleet.subprocess,'Popen',side_effect=lambda *a,**k:Process()):
+                with patch.object(decision_fleet.time,'time',return_value=200):f.tick([b,other])
+                self.assertEqual(set(f.children),{'d'});f.close()
+                saved=json.loads((f.root/'schedule.json').read_text());self.assertEqual(saved['a']['waiting_since'],200)
+    def test_uncertain_after_restart_only_mirrors(self):
+        with tempfile.TemporaryDirectory() as root:
+            cfg,b=self.setup_fleet(root);b['demand']['uncertain']=1
+            f=decision_fleet.Fleet(cfg)
+            with patch.object(decision_fleet.subprocess,'Popen',side_effect=lambda *a,**k:Process()) as spawn:
+                result=f.tick([b]);self.assertEqual(set(f.children['d']['processes']),{'mirror'})
+                self.assertEqual(spawn.call_count,1);self.assertEqual(result['status'],'failed');f.close()
+    def test_fact_reply_waits_for_native_terminal_and_final_mirror(self):
+        with tempfile.TemporaryDirectory() as root:
+            cfg,b=self.setup_fleet(root);b['demand']['queued']=0
+            cfg['notification_state_file']=str(Path(root)/'notifier.json')
+            path=Path(cfg['notification_state_file'])
+            record={'decision_id':'d','event_kind':'fact_answered','processing_done':False}
+            decision_fleet.atomic_json(path,{'deliveries':{'receipt':record}})
+            f=decision_fleet.Fleet(cfg)
+            with patch.object(decision_fleet.subprocess,'Popen',side_effect=lambda *a,**k:Process()) as spawn:
+                f.tick([b]);self.assertEqual(set(f.children['d']['processes']),{'mirror'})
+                f.children['d']['processes']['mirror'].code=0
+                record.update(processing_done=True,completed_at=123)
+                decision_fleet.atomic_json(path,{'deliveries':{'receipt':record}})
+                f.tick([b]);self.assertEqual(spawn.call_count,2)
+                f.children['d']['processes']['mirror'].code=0
+                self.assertEqual(f.tick([b])['active'],0)
+                self.assertEqual(f.tick([b])['active'],0);self.assertEqual(spawn.call_count,2)
+                f.close()
+
+    def test_terminal_proof_clears_local_without_replaying(self):
+        with tempfile.TemporaryDirectory() as root:
+            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'}]}
+            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'))
+            self.assertFalse(decision_fleet.busy(path));self.assertIn('original',json.loads(path.read_text())['delivery_receipts'])
+
+if __name__=='__main__':unittest.main()

tests/test_message_demand.py

@@ -0,0 +1,68 @@
+"""Transport evidence and demand survive storage reopen without resending."""
+import json
+import tempfile
+import time
+import unittest
+from pathlib import Path
+from agent_control.store import Store
+from agent_control.conversation import project
+from test_decision_workflow import bind
+from test_decisions import definition
+
+
+class DemandTests(unittest.TestCase):
+    def setUp(self):
+        self.tmp=tempfile.TemporaryDirectory()
+        self.path=Path(self.tmp.name)/'state.sqlite3'
+        self.store=Store(self.path)
+        self.store.decision(definition())
+        bind(self.store)
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def demand(self):
+        return Store(self.path).decision_bindings()['bindings'][0]['demand']
+
+    def test_persisted_queue_and_expired_claims_have_distinct_demand(self):
+        m=self.store.send(dict(session_id='consult',text='Question',idempotency_key='one'))
+        self.assertEqual(m['delivery_stage'],'queued')
+        self.assertEqual(self.demand()['queued'],1)
+        self.assertEqual(self.demand()['human_queued'],1)
+        self.assertIsNotNone(self.demand()['oldest_queued_at'])
+        claim=self.store.claim(dict(session_id='consult',worker_id='worker',lease_seconds=10))['message']
+        self.assertEqual(self.demand()['inflight'],1)
+        with self.store.db() as db:
+            db.execute('UPDATE messages SET lease_until=? WHERE id=?',(time.time()-1,m['id']))
+        self.assertEqual(self.demand()['queued'],1)
+        with self.store.db() as db:
+            db.execute('UPDATE messages SET started=1 WHERE id=?',(m['id'],))
+        self.assertEqual(self.demand()['uncertain'],1)
+        self.assertEqual(self.demand()['queued'],0)
+        self.assertEqual(Store(self.path).send(dict(session_id='consult',text='Question',idempotency_key='one'))['id'],m['id'])
+        self.assertEqual(len(self.store.messages('consult')),1)
+
+    def test_safe_projection_uses_host_evidence_not_local_start(self):
+        m=self.store.send(dict(session_id='consult',text='Question',idempotency_key='one'))
+        stages=[('claimed',None,None,'queued'),
+                ('claimed',dict(state='submitting'),None,'queued'),
+                ('claimed',dict(state='accepted',receipt_id='private'),None,'delivered'),
+                ('claimed',dict(state='observed',turn_id='private'),None,'processing'),
+                ('uncertain',dict(state='accepted'),None,'blocked'),
+                ('completed',dict(state='observed'),'reply','replied')]
+        for status,submission,response,expected in stages:
+            with self.store.db() as db:
+                db.execute('UPDATE messages SET status=?,started=1,submission=?,response_id=? WHERE id=?',
+                           (status,json.dumps(submission) if submission else None,response,m['id']))
+            view=Store(self.path).conversation('consult')[0]
+            self.assertEqual(view['delivery_stage'],expected)
+            self.assertNotIn('submission',view)
+            self.assertNotIn('private',json.dumps(view))
+
+    def test_mirrored_history_does_not_fabricate_delivery_proof(self):
+        row=dict(id='history',text='Question',source='human',origin='session_history',status='delivered',created=1)
+        self.assertNotIn('delivery_stage',project([row],[])[0])
+
+
+if __name__=='__main__':
+    unittest.main()

tests/test_mirror.py

@@ -123,5 +123,33 @@ class MirrorTests(unittest.TestCase):
             client.instance='two'
             with self.assertRaisesRegex(RuntimeError,'another control'): sync(Server(),client,cfg)
 
+    def test_lost_import_ack_retries_same_external_identity_without_model_execution(self):
+        class Server:
+            def history(self,tid):
+                return {'id':tid},[{'id':'turn','status':'completed','items':[
+                    {'id':'reply','type':'agentMessage','phase':'final_answer','text':'result'}]}]
+        class Client:
+            calls=[]
+            imported={}
+            lose_ack=True
+            def call(self,path,data=None):
+                if path.startswith('/api/messages'):return {'messages':[]}
+                if path=='/api/state':return {'instance_id':'one'}
+                self.calls.append(data)
+                for item in data['items']:self.imported[item['id']]=item
+                if self.lose_ack:
+                    self.lose_ack=False
+                    raise OSError('server persisted import but response was lost')
+                return {'deferred':[]}
+        with tempfile.TemporaryDirectory() as root:
+            cfg={'thread_id':'thread','session_id':'session','control':{},'state_file':str(Path(root)/'worker.json')}
+            client=Client()
+            with self.assertRaises(OSError):sync(Server(),client,cfg)
+            self.assertFalse((Path(root)/'mirror-state.json').exists())
+            sync(Server(),client,cfg);sync(Server(),client,cfg)
+            self.assertEqual(len(client.calls),2)
+            self.assertEqual(client.calls[0],client.calls[1])
+            self.assertEqual(list(client.imported),['reply'])
+
 
 if __name__=='__main__':unittest.main()

tests/test_notifier.py

@@ -10,6 +10,53 @@ import notifier
 
 
 class NotifierTests(unittest.TestCase):
+    def test_decisions_prefer_fresh_work_but_waiting_items_survive_restart_and_age(self):
+        normal={'notification_id':'ordinary','checkpoint':'half','task':{'id':'work'}}
+        decision={'notification_id':'choice','decision':{'id':'decision'}}
+        state={}
+        self.assertEqual(notifier.schedule_notifications([normal,decision],state,100)[0],decision)
+        # Persist/reload and a continuing stream of fresh decisions cannot starve old work.
+        state=json.loads(json.dumps(state))
+        fresh={'notification_id':'new-choice','decision':{'id':'new'}}
+        self.assertEqual(notifier.schedule_notifications([fresh,normal],state,221)[0],normal)
+        self.assertEqual(state['notification_waiting']['eta:ordinary'],100)
+
+    def test_waiting_age_does_not_bypass_uncertain_target_backpressure(self):
+        with tempfile.TemporaryDirectory() as root:
+            cfg,n,path=self.fixture(root);cfg['enforce_backpressure']=True
+            class Client:
+                def call(self,p,data=None):return {'notifications':[n]} if p.endswith('notifications') else {}
+            state={'notification_waiting':{'hook:note1':0}}
+            with patch.object(notifier,'target_ready',return_value=False),patch.object(notifier.subprocess,'run') as run:
+                notifier.deliver(Client(),cfg,state,path)
+                run.assert_not_called()
+            self.assertEqual(json.loads(path.read_text())['notification_waiting']['hook:note1'],0)
+
+    def test_uncertain_reconciliation_rejects_wrong_thread_even_with_exact_input(self):
+        import hashlib
+        with tempfile.TemporaryDirectory() as root:
+            cfg,n,path=self.fixture(root)
+            prompt=notifier.notification_text(n)
+            state={'deliveries':{'note1':{'thread_id':cfg['notification_targets']['control'],
+                'input_sha256':hashlib.sha256(prompt.encode()).hexdigest(),'state':'submitting'}}}
+            class Client:
+                acks=[]
+                def call(self,p,data=None):
+                    if p.endswith('notifications'):return {'notifications':[n]}
+                    self.acks.append(p);return {}
+            class Reader:
+                def __init__(self,*args):pass
+                def __enter__(self):return self
+                def __exit__(self,*args):pass
+                def history(self,thread):return {'id':'different-thread'},[{'id':'turn','items':[
+                    {'type':'userMessage','content':[{'type':'text','text':prompt}]}]}]
+            client=Client()
+            with patch.object(notifier,'AppServer',Reader),patch.object(notifier.subprocess,'run') as run:
+                with self.assertRaises(RuntimeError):notifier.deliver(client,cfg,state,path)
+                run.assert_not_called()
+            self.assertEqual(client.acks,[])
+            self.assertEqual(state['deliveries']['note1']['state'],'submitting')
+
     def fixture(self,root):
         cfg={'control':{},'notification_targets':{'control':'11111111-1111-4111-8111-111111111111'},'codex':'codex','cwd':root}
         n={'notification_id':'note1','target':'control','hook':{'source':'test','event_id':'incident','summary':'synthetic failure','severity':'warning','status':'firing'}}
@@ -165,13 +212,57 @@ class NotifierTests(unittest.TestCase):
                'event':{'kind':'fact_answered','request_id':'fact-one','answer':'The prerequisite is verified.'}}
             class Client:
                 def call(self,p,data=None):return {'notifications':[n]} if p=='/api/decision-notifications' else {'notifications':[]}
+            class Reader:
+                def __init__(self,*args):pass
+                def __enter__(self):return self
+                def __exit__(self,*args):pass
+                def call(self,method,params):
+                    return {'thread':{'id':thread}} if method=='thread/read' else {'data':[]}
             result=SimpleNamespace(returncode=0,stdout='Queued message 22222222-2222-4222-8222-222222222222 for thread '+thread+'.')
             state={}
-            with patch.object(notifier.subprocess,'run',return_value=result) as run:
+            with patch.object(notifier.subprocess,'run',return_value=result) as run,patch.object(notifier,'AppServer',Reader):
                 notifier.deliver(Client(),cfg,state,path);notifier.deliver(Client(),cfg,state,path)
                 self.assertEqual(run.call_count,1)
                 self.assertIn(thread,run.call_args.args[0])
                 self.assertNotIn(cfg['notification_targets']['control'],run.call_args.args[0])
+            record=state['deliveries']['decision:fact-answer']
+            self.assertEqual(record['decision_id'],'one')
+            self.assertEqual(record['event_kind'],'fact_answered')
+
+    def test_fact_answer_completion_reconciles_empty_outbox_without_resend(self):
+        import hashlib
+        with tempfile.TemporaryDirectory() as root:
+            cfg,_,path=self.fixture(root)
+            thread='44444444-4444-4444-8444-444444444444'
+            prompt='fact answer exact input'
+            record={'kind':'decision','decision_id':'one','event_kind':'fact_answered',
+                    'thread_id':thread,'input_sha256':hashlib.sha256(prompt.encode()).hexdigest(),
+                    'state':'accepted','acked':True}
+            state={'deliveries':{'decision:answer':record}}
+            class Reader:
+                body=prompt
+                identity=thread
+                def __init__(self,*args):pass
+                def __enter__(self):return self
+                def __exit__(self,*args):pass
+                def call(self,method,params):
+                    if method=='thread/read':return {'thread':{'id':self.identity}}
+                    return {'data':[{'id':'real-turn','status':'completed','completedAt':123,
+                        'items':[{'type':'userMessage','content':[{'type':'text','text':self.body}]},
+                                 {'type':'agentMessage','phase':'final_answer','text':'verified'}]}]}
+            client=SimpleNamespace(call=lambda *args:{'notifications':[]})
+            with patch.object(notifier,'AppServer',Reader),patch.object(notifier.subprocess,'run') as run:
+                Reader.body=prompt+' changed'
+                notifier.deliver(client,cfg,state,path)
+                self.assertFalse(record.get('processing_done'))
+                Reader.body=prompt;Reader.identity='wrong-thread'
+                with self.assertRaises(RuntimeError):notifier.deliver(client,cfg,state,path)
+                self.assertFalse(record.get('processing_done'))
+                Reader.identity=thread
+                notifier.deliver(client,cfg,state,path)
+                run.assert_not_called()
+            self.assertEqual(record['turn_id'],'real-turn')
+            self.assertTrue(json.loads(path.read_text())['deliveries']['decision:answer']['processing_done'])
 
     def test_review_notice_is_concise_and_submission_is_not_acceptance(self):
         n={'notification_id':'notice','decision':{'id':'one','title':'Plan','options':[{'id':'a','label':'Private option detail'}]},