318 lines
20 KiB
Python
318 lines
20 KiB
Python
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 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'}}
|
|
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':[]}
|
|
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,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'}]},
|
|
'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()
|