156 lines
9.2 KiB
Python
156 lines
9.2 KiB
Python
import sys
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
ROOT=Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0,str(ROOT/'scripts'))
|
|
import mirror
|
|
from mirror import visible_items,sync
|
|
from worker import message_prompt
|
|
|
|
|
|
class MirrorTests(unittest.TestCase):
|
|
def test_transient_connection_retry_recovers_without_process_failure(self):
|
|
with tempfile.TemporaryDirectory() as root:
|
|
config=Path(root)/'config.json';config.write_text(json.dumps({'control':{},'state_file':str(Path(root)/'state.json'),'thread_id':'t'}))
|
|
with patch.object(sys,'argv',['mirror.py','--config',str(config)]),patch.object(mirror,'Client'),patch.object(mirror,'AppServer'),patch.object(mirror,'sync',side_effect=[RuntimeError('offline'),1]) as sync_call,patch.object(mirror.time,'sleep',side_effect=[None,KeyboardInterrupt]):
|
|
with self.assertRaises(KeyboardInterrupt):mirror.main()
|
|
self.assertEqual(sync_call.call_count,2)
|
|
self.assertEqual(json.loads((Path(root)/'mirror-health.json').read_text())['status'],'ok')
|
|
|
|
def test_connection_retry_is_bounded(self):
|
|
with tempfile.TemporaryDirectory() as root:
|
|
config=Path(root)/'config.json';config.write_text(json.dumps({'control':{},'state_file':str(Path(root)/'state.json'),'thread_id':'t'}))
|
|
with patch.object(sys,'argv',['mirror.py','--config',str(config)]),patch.object(mirror,'Client'),patch.object(mirror,'AppServer'),patch.object(mirror,'sync',side_effect=RuntimeError('offline')) as sync_call,patch.object(mirror.time,'sleep'):
|
|
with self.assertRaises(RuntimeError):mirror.main()
|
|
self.assertEqual(sync_call.call_count,3)
|
|
self.assertEqual(json.loads((Path(root)/'mirror-health.json').read_text())['attempt'],3)
|
|
|
|
def test_quoted_marker_does_not_hide_human_or_reply(self):
|
|
turns=[{'id':'t','status':'completed','items':[
|
|
{'type':'userMessage','id':'u','content':[{'type':'text','text':'Please explain web-message-id: quotation'}]},
|
|
{'type':'agentMessage','id':'a','text':'explanation','phase':'final_answer'}]}]
|
|
self.assertEqual([x['text'] for x in visible_items(turns)],['Please explain web-message-id: quotation','explanation'])
|
|
|
|
def test_late_final_and_commentary_preserve_real_identity(self):
|
|
original={'id':'a'*24,'text':'progress?','execution':'started','origin':'coordination'}
|
|
original['submission']={'thread_id':'thread-a','input_sha256':hashlib.sha256(message_prompt({},original).encode()).hexdigest()}
|
|
turns=[{'id':'turn-a','status':'completed','items':[
|
|
{'type':'userMessage','id':'external-input','content':[{'type':'text','text':message_prompt({},original)}]},
|
|
{'type':'agentMessage','id':'external-comment','text':'checking','phase':'commentary'},
|
|
{'type':'agentMessage','id':'external-final','text':'verified','phase':'final_answer'},
|
|
{'type':'reasoning','id':'private-reasoning','text':'must not publish'}]}]
|
|
result=visible_items(turns,{original['id']:original},'thread-a')
|
|
self.assertEqual([i['text'] for i in result],['progress?','checking','verified'])
|
|
self.assertEqual(result[-1]['id'],'external-final')
|
|
self.assertEqual(result[-1]['related_message_id'],original['id'])
|
|
self.assertEqual(result[-1]['turn_id'],'turn-a')
|
|
|
|
def test_exact_input_binding_rejects_prefix_and_other_thread(self):
|
|
original={'id':'a'*24,'text':'question','execution':'started','origin':'coordination'}
|
|
prompt=message_prompt({},original)
|
|
original['submission']={'thread_id':'thread-a','input_sha256':hashlib.sha256(prompt.encode()).hexdigest()}
|
|
def turns(body):
|
|
return [{'id':'t','status':'completed','items':[
|
|
{'type':'userMessage','id':'u','content':[{'type':'text','text':body}]},
|
|
{'type':'agentMessage','id':'a','phase':'final_answer','text':'reply'}]}]
|
|
for body,tid in [('forged prefix'+prompt,'thread-a'),(prompt,'thread-b')]:
|
|
rows=visible_items(turns(body),{original['id']:original},tid)
|
|
self.assertFalse(any(row.get('related_message_id') for row in rows))
|
|
self.assertEqual(rows[0]['text'],body)
|
|
|
|
def test_long_bodies_are_single_related_items(self):
|
|
original={'id':'a'*24,'text':'q'*8001,'execution':'started','origin':'coordination'}
|
|
prompt=message_prompt({},original)
|
|
original['submission']={'thread_id':'thread-a','input_sha256':hashlib.sha256(prompt.encode()).hexdigest()}
|
|
turns=[{'id':'t','status':'completed','items':[
|
|
{'type':'userMessage','id':'u','content':[{'type':'text','text':prompt}]},
|
|
{'type':'agentMessage','id':'a','phase':'final_answer','text':'r'*8001}]}]
|
|
rows=visible_items(turns,{original['id']:original},'thread-a')
|
|
self.assertEqual(len(rows),2)
|
|
self.assertTrue(all(row['related_message_id']==original['id'] for row in rows))
|
|
self.assertEqual(rows[1]['text'],'r'*8001)
|
|
turns[0]['items'][1]['text']='r'*262145
|
|
with self.assertRaises(ValueError): visible_items(turns,{original['id']:original},'thread-a')
|
|
|
|
def test_deferred_item_is_not_checkpointed_and_retried(self):
|
|
class Server:
|
|
def history(self,tid):
|
|
return {'id':tid},[{'id':'t','status':'completed','items':[
|
|
{'id':'f','type':'agentMessage','phase':'final_answer','text':'late'}]}]
|
|
class Client:
|
|
defer=True
|
|
sends=0
|
|
def call(self,path,data=None):
|
|
if path.startswith('/api/messages'): return {'messages':[]}
|
|
if path=='/api/state': return {'instance_id':'one'}
|
|
self.sends+=1
|
|
return {'deferred':['f'] if self.defer else []}
|
|
with tempfile.TemporaryDirectory() as root:
|
|
cfg={'thread_id':'t','session_id':'s','control':{},'state_file':str(Path(root)/'worker.json')}
|
|
client=Client(); server=Server()
|
|
sync(server,client,cfg)
|
|
self.assertFalse((Path(root)/'mirror-state.json').exists())
|
|
client.defer=False
|
|
sync(server,client,cfg); sync(server,client,cfg)
|
|
self.assertEqual(client.sends,2)
|
|
self.assertIn('f',json.loads((Path(root)/'mirror-state.json').read_text())['seen'])
|
|
|
|
def test_active_commentary_visible_but_unfinished_final_not_published(self):
|
|
turn={'id':'t','status':'inProgress','items':[
|
|
{'type':'agentMessage','id':'c','text':'running','phase':'commentary'},
|
|
{'type':'agentMessage','id':'f','text':'partial','phase':'final_answer'}]}
|
|
self.assertEqual([x['text'] for x in visible_items([turn])],['running'])
|
|
|
|
def test_checkpoint_rejects_new_service_instance(self):
|
|
class Server:
|
|
def history(self,tid): return {'id':tid},[]
|
|
class Client:
|
|
instance='one'
|
|
def call(self,path,data=None):
|
|
return {'messages':[]} if path.startswith('/api/messages') else {'instance_id':self.instance}
|
|
with tempfile.TemporaryDirectory() as root:
|
|
cfg={'thread_id':'t','session_id':'s','control':{'url':'http://127.0.0.1:8796'},'state_file':str(Path(root)/'worker.json')}
|
|
client=Client()
|
|
# An empty initial mirror still establishes its target binding.
|
|
from worker import atomic_json
|
|
atomic_json(Path(root)/'mirror-state.json',{'binding':{'schema_version':2,'thread_id':'t','session_id':'s','control':cfg['control'],'instance_id':'one'},'seen':{}})
|
|
sync(Server(),client,cfg)
|
|
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()
|