import json from pathlib import Path import sys import tempfile from types import SimpleNamespace import unittest from unittest.mock import patch sys.path.insert(0,str(Path(__file__).resolve().parents[1]/'scripts')) import notifier class NotifierTests(unittest.TestCase): 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'}} return cfg,n,Path(root)/'state.json' def test_ack_retry_never_resubmits_accepted_queue(self): with tempfile.TemporaryDirectory() as root: cfg,n,path=self.fixture(root) class Client: failed=False def call(self,p,data=None): if p.endswith('notifications'):return {'notifications':[n]} if not self.failed:self.failed=True;raise RuntimeError('ack offline') return {} c=Client();state={} result=SimpleNamespace(returncode=0,stdout='Queued message 22222222-2222-4222-8222-222222222222 for thread '+cfg['notification_targets']['control']+'.') with patch.object(notifier.subprocess,'run',return_value=result) as run: with self.assertRaises(RuntimeError):notifier.deliver(c,cfg,state,path) notifier.deliver(c,cfg,state,path) self.assertEqual(run.call_count,1) self.assertTrue(state['deliveries']['note1']['acked']) def test_uncertain_queue_never_blindly_replays(self): with tempfile.TemporaryDirectory() as root: cfg,n,path=self.fixture(root) class Client: def call(self,p,data=None):return {'notifications':[n]} class Reader: def __init__(self,*args):pass def __enter__(self):return self def __exit__(self,*args):pass def history(self,thread):return {'id':thread},[] state={} with patch.object(notifier.subprocess,'run',return_value=SimpleNamespace(returncode=1,stdout='')) as run,patch.object(notifier,'AppServer',Reader): with self.assertRaisesRegex(RuntimeError,'uncertain'):notifier.deliver(Client(),cfg,state,path) with self.assertRaisesRegex(RuntimeError,'reconciliation'):notifier.deliver(Client(),cfg,state,path) self.assertEqual(run.call_count,1) def test_unknown_target_does_not_block_other_target(self): with tempfile.TemporaryDirectory() as root: cfg,n,path=self.fixture(root) cfg['notification_targets']['human']='33333333-3333-4333-8333-333333333333' other={**n,'notification_id':'note2','target':'human'} class Client: def call(self,p,data=None):return {'notifications':[n,other]} if p.endswith('notifications') else {} results=[SimpleNamespace(returncode=1,stdout=''),SimpleNamespace(returncode=0,stdout='Queued message 22222222-2222-4222-8222-222222222222 for thread '+cfg['notification_targets']['human']+'.')] state={} with patch.object(notifier.subprocess,'run',side_effect=results) as run: with self.assertRaises(RuntimeError):notifier.deliver(Client(),cfg,state,path) self.assertEqual(run.call_count,2) self.assertTrue(state['deliveries']['note2']['acked']) self.assertEqual(state['deliveries']['note1']['state'],'submitting') def test_decision_response_goes_only_to_control_and_retries_ack_without_queue(self): with tempfile.TemporaryDirectory() as root: cfg,n,path=self.fixture(root) cfg['decision_notifications']=True item={'notification_id':'decision-note','target':'control','decision':{'id':'choice','title':'Choose plan','question':'Which plan?','status':'decided','requires_authorization':False,'options':[{'id':'a','label':'A','description':'Plan A'}],'response':{'id':'response','option_id':'a','comment':'Use A','authorization':False}}} calls=[] class Client: def call(self,p,data=None): calls.append(p) return {'notifications':[item] if p=='/api/decision-notifications' else []} result=SimpleNamespace(returncode=0,stdout='Queued message 22222222-2222-4222-8222-222222222222 for thread '+cfg['notification_targets']['control']+'.') state={} with patch.object(notifier.subprocess,'run',return_value=result) as run: notifier.deliver(Client(),cfg,state,path) notifier.deliver(Client(),cfg,state,path) self.assertEqual(run.call_count,1) self.assertEqual(calls.count('/api/decision-notifications/ack'),2) self.assertTrue(state['deliveries']['decision:decision-note']['acked']) def test_eta_checkpoint_only_goes_to_control_and_ack_is_not_review_completion(self): with tempfile.TemporaryDirectory() as root: cfg,n,path=self.fixture(root);cfg['eta_notifications']=True item={'notification_id':'eta-note','target':'control','task':{'id':'work','current_progress':'checking','status':'running'},'checkpoint':'half','revision':1} calls=[] class Client: def call(self,p,data=None): calls.append(p) return {'notifications':[item] if p=='/api/eta-notifications' else []} result=SimpleNamespace(returncode=0,stdout='Queued message 22222222-2222-4222-8222-222222222222 for thread '+cfg['notification_targets']['control']+'.') with patch.object(notifier.subprocess,'run',return_value=result):notifier.deliver(Client(),cfg,{},path) self.assertIn('/api/eta-notifications/ack',calls) self.assertFalse(any('/complete' in p for p in calls)) def test_internal_review_merges_control_and_human_hook_without_root_injection(self): with tempfile.TemporaryDirectory() as root: cfg,n,path=self.fixture(root) cfg['notification_targets']['human']='44444444-4444-4444-8444-444444444444' cfg['internal_event_thread_id']='33333333-3333-4333-8333-333333333333' other={**n,'notification_id':'note2','target':'human'} calls=[] class Client: def call(self,p,data=None): calls.append((p,data)) return {'notifications':[n,other]} if p.endswith('notifications') else {} result=SimpleNamespace(returncode=0,stdout='Queued message 22222222-2222-4222-8222-222222222222 for thread '+cfg['internal_event_thread_id']+'.') state={} with patch.object(notifier.subprocess,'run',return_value=result) as run: notifier.deliver(Client(),cfg,state,path) self.assertEqual(run.call_count,1) self.assertIn(cfg['internal_event_thread_id'],run.call_args.args[0]) self.assertNotIn(cfg['notification_targets']['control'],run.call_args.args[0]) self.assertEqual(len(state['deliveries']),1) self.assertEqual(len([x for x in calls if x[0]=='/api/hooks/ack']),2) def test_busy_target_keeps_backlog_until_real_completed_final(self): import hashlib with tempfile.TemporaryDirectory() as root: cfg,n,path=self.fixture(root);cfg['enforce_backpressure']=True notes=[{**n,'notification_id':'note'+str(i),'hook':{**n['hook'],'event_id':'incident'+str(i)}} for i in range(20)] class Client: def call(self,p,data=None):return {'notifications':notes} if p.endswith('notifications') else {} class Reader: turns=[] def __init__(self,*a):pass def __enter__(self):return self def __exit__(self,*a):pass def call(self,method,params): return {'thread':{'id':cfg['notification_targets']['control']}} if method=='thread/read' else {'data':self.turns} result=SimpleNamespace(returncode=0,stdout='Queued message 22222222-2222-4222-8222-222222222222 for thread '+cfg['notification_targets']['control']+'.') state={} with patch.object(notifier.subprocess,'run',return_value=result) as run,patch.object(notifier,'AppServer',Reader): for _ in range(5):notifier.deliver(Client(),cfg,state,path) self.assertEqual(run.call_count,1) prompt=run.call_args.args[0][-1] self.assertEqual(hashlib.sha256(prompt.encode()).hexdigest(),next(iter(state['deliveries'].values()))['input_sha256']) Reader.turns=[{'id':'turn','status':'completed','completedAt':1,'items':[{'type':'userMessage','content':[{'type':'text','text':prompt}]},{'type':'agentMessage','phase':'final_answer','text':'reviewed'}]}] notifier.deliver(Client(),cfg,state,path) self.assertEqual(run.call_count,2) def test_task_event_routes_to_registered_executor_before_deputy(self): with tempfile.TemporaryDirectory() as root: cfg,n,path=self.fixture(root) cfg.update(internal_event_thread_id='33333333-3333-4333-8333-333333333333',execution_notification_targets={'worker':'44444444-4444-4444-8444-444444444444'}) n['hook']['task_id']='owned-task' class Client: def call(self,p,data=None): if p=='/api/state':return {'tasks':[{'id':'owned-task','session_id':'worker'}]} return {'notifications':[n]} if p.endswith('notifications') else {} result=SimpleNamespace(returncode=0,stdout='Queued message 22222222-2222-4222-8222-222222222222 for thread '+cfg['execution_notification_targets']['worker']+'.') with patch.object(notifier.subprocess,'run',return_value=result) as run: notifier.deliver(Client(),cfg,{},path) self.assertIn(cfg['execution_notification_targets']['worker'],run.call_args.args[0]) self.assertNotIn(cfg['internal_event_thread_id'],run.call_args.args[0]) def test_decision_workflow_uses_bound_consultation_and_never_replays(self): with tempfile.TemporaryDirectory() as root: cfg,_,path=self.fixture(root);cfg['decision_notifications']=True thread='44444444-4444-4444-8444-444444444444' n={'notification_id':'fact-answer','target':'decision','decision':{'id':'one','title':'Plan','session_thread_id':thread}, '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':[]} 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: 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]) 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'}]}, 'event':{'kind':'reviewed','reason':'Suitable in scope','impact':'No migration needed','execution_plan':'Verify before rollout'}} text=notifier.notification_text(n) self.assertNotIn('Private option detail',text) self.assertIn('#/decisions?decision=one',text) n['event']={'kind':'submitted','revision':2} self.assertIn('不要把提交直接当接受',notifier.notification_text(n)) n['event']={'kind':'archive_required','revision':3} self.assertIn('不能只写数据库冒充宿主归档',notifier.notification_text(n)) def test_failed_decision_submission_marker_does_not_queue_or_leave_unknown_model_input(self): with tempfile.TemporaryDirectory() as root: cfg,_,path=self.fixture(root);cfg['decision_notifications']=True n={'notification_id':'submit','target':'control','decision':{'id':'one','title':'Plan'},'event':{'kind':'submitted','revision':1}} class Client: failed=False def call(self,p,data=None): if p=='/api/decision-notifications':return {'notifications':[n]} if p.endswith('/submitting') and not self.failed: self.failed=True;raise OSError('temporary marker interruption') return {'notifications':[]} c=Client();state={} result=SimpleNamespace(returncode=0,stdout='Queued message 22222222-2222-4222-8222-222222222222 for thread '+cfg['notification_targets']['control']+'.') with patch.object(notifier.subprocess,'run',return_value=result) as run: with self.assertRaises(RuntimeError):notifier.deliver(c,cfg,state,path) self.assertEqual(state['deliveries'],{}) run.assert_not_called() notifier.deliver(c,cfg,state,path) self.assertEqual(run.call_count,1) def test_consultation_target_cannot_point_to_primary_or_unbound_thread(self): with tempfile.TemporaryDirectory() as root: cfg,_,path=self.fixture(root);cfg['decision_notifications']=True n={'notification_id':'bad','target':'decision','decision':{'id':'one','session_thread_id':cfg['notification_targets']['control']},'event':{'kind':'fact_answered'}} class Client: def call(self,p,data=None):return {'notifications':[n]} if p=='/api/decision-notifications' else {'notifications':[]} with patch.object(notifier.subprocess,'run') as run: with self.assertRaises(RuntimeError):notifier.deliver(Client(),cfg,{},path) run.assert_not_called() def test_no_events_means_no_conversation_input(self): with tempfile.TemporaryDirectory() as root: cfg,n,path=self.fixture(root) c=SimpleNamespace(call=lambda *args:{'notifications':[]}) with patch.object(notifier.subprocess,'run') as run: notifier.deliver(c,cfg,{},path) run.assert_not_called() if __name__=='__main__':unittest.main()