106 lines
5.6 KiB
Python
106 lines
5.6 KiB
Python
import itertools
|
|
import importlib.util
|
|
import io
|
|
import json
|
|
from pathlib import Path
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
from types import SimpleNamespace
|
|
|
|
ROOT=Path(__file__).resolve().parents[1]
|
|
SCRIPT=ROOT/'scripts'
|
|
sys.path.insert(0,str(SCRIPT))
|
|
import worker
|
|
|
|
|
|
class Process:
|
|
pid=12345
|
|
def __init__(self,events):
|
|
self.stdin=io.StringIO()
|
|
self.stdout=io.StringIO(''.join(json.dumps(e)+'\n' for e in events))
|
|
def wait(self,timeout=None): return 0
|
|
def poll(self): return 0
|
|
|
|
|
|
class WorkerTests(unittest.TestCase):
|
|
def test_desktop_queue_records_receipt_and_waits_for_final_hydration(self):
|
|
import appserver
|
|
with tempfile.TemporaryDirectory() as root:
|
|
cfg={'codex':'codex','cwd':root,'max_turn_seconds':1000}
|
|
message={'id':'a'*24,'text':'hello','source':'human'}
|
|
prompt=worker.message_prompt(cfg,message)
|
|
thread='11111111-1111-4111-8111-111111111111'
|
|
receipt='22222222-2222-4222-8222-222222222222'
|
|
class Reader:
|
|
calls=0
|
|
def __init__(self,*args):pass
|
|
def __enter__(self):return self
|
|
def __exit__(self,*args):pass
|
|
def history(self,tid):
|
|
self.calls+=1
|
|
if self.calls==1:return {'id':tid},[]
|
|
items=[{'type':'userMessage','content':[{'type':'text','text':prompt}]}]
|
|
if self.calls>2:items.append({'type':'agentMessage','text':'real reply','phase':'final_answer'})
|
|
return {'id':tid},[{'id':'observed-turn','status':'interrupted' if self.calls==2 else 'completed','items':items}]
|
|
receipts=[]
|
|
with patch.object(appserver,'AppServer',Reader),patch.object(worker.subprocess,'run',return_value=SimpleNamespace(returncode=0,stdout='Queued message '+receipt+' for thread '+thread+'.')),patch.object(worker.time,'sleep'),patch.object(worker.time,'monotonic',side_effect=itertools.count(0,15)):
|
|
result=worker.run_queued(cfg,message,thread,lambda:None,lambda value:None,receipts.append)
|
|
self.assertEqual(result['outcome'],'completed')
|
|
self.assertEqual(result['text'],'real reply')
|
|
self.assertEqual(next(r for r in receipts if r['state']=='accepted')['receipt_id'],receipt)
|
|
|
|
def test_progress_without_terminal_event_is_not_success(self):
|
|
with tempfile.TemporaryDirectory() as root:
|
|
proc=Process([{'type':'item.completed','item':{'type':'agent_message','text':'working'}}])
|
|
with patch.object(worker.subprocess,'Popen',return_value=proc):
|
|
result=worker.run_codex({'codex':'codex','cwd':root},{'id':'m1','text':'hello','source':'agent'},None,lambda:None)
|
|
self.assertEqual(result['outcome'],'failed')
|
|
|
|
def test_real_terminal_event_and_thread_saved_immediately(self):
|
|
with tempfile.TemporaryDirectory() as root:
|
|
saved=[]
|
|
proc=Process([{'type':'thread.started','thread_id':'thread-1'},
|
|
{'type':'item.completed','item':{'type':'agent_message','text':'ready'}},
|
|
{'type':'turn.completed'}])
|
|
with patch.object(worker.subprocess,'Popen',return_value=proc):
|
|
result=worker.run_codex({'codex':'codex','cwd':root},{'id':'m1','text':'hello','source':'agent'},None,lambda:None,on_thread=saved.append)
|
|
self.assertEqual(saved,['thread-1'])
|
|
self.assertEqual(result,{'text':'ready','thread_id':'thread-1','outcome':'completed'})
|
|
|
|
def test_once_completes_without_claiming_second_message(self):
|
|
with tempfile.TemporaryDirectory() as root:
|
|
calls=[]
|
|
class Client:
|
|
def __init__(self,cfg): pass
|
|
def call(self,path,data):
|
|
calls.append(path)
|
|
if path=='/api/worker/claim': return {'message':{'id':'m1','lease_id':'l1','text':'hello','source':'human'}}
|
|
return {}
|
|
cfg={'control':{},'session_id':'s1','cwd':root,'state_file':str(Path(root)/'state.json')}
|
|
with patch.object(worker,'Client',Client),patch.object(worker,'run_codex',return_value={'text':'reply','thread_id':'t1','outcome':'completed'}) as run:
|
|
worker.work(cfg,once=True)
|
|
self.assertEqual(calls,['/api/worker/claim','/api/worker/start','/api/worker/complete'])
|
|
self.assertEqual(run.call_count,1)
|
|
state=json.loads(Path(cfg['state_file']).read_text())
|
|
self.assertEqual(state['thread_id'],'t1')
|
|
self.assertNotIn('inflight',state)
|
|
|
|
def test_interrupted_execution_is_never_rerun(self):
|
|
with tempfile.TemporaryDirectory() as root:
|
|
cfg={'control':{},'session_id':'s1','cwd':root,'state_file':str(Path(root)/'state.json')}
|
|
worker.atomic_json(Path(cfg['state_file']),{'binding':{'control':{},'session_id':'s1','cwd':str(Path(root).resolve())},'inflight':{'message_id':'m1'}})
|
|
with patch.object(worker,'run_codex') as run:
|
|
with self.assertRaisesRegex(RuntimeError,'interrupted'): worker.work(cfg,once=True)
|
|
run.assert_not_called()
|
|
|
|
def test_changed_session_refuses_saved_thread(self):
|
|
with tempfile.TemporaryDirectory() as root:
|
|
cfg={'control':{},'session_id':'s2','cwd':root,'state_file':str(Path(root)/'state.json')}
|
|
worker.atomic_json(Path(cfg['state_file']),{'binding':{'control':{},'session_id':'s1','cwd':root},'thread_id':'t1'})
|
|
with self.assertRaisesRegex(RuntimeError,'another session'): worker.work(cfg,once=True)
|
|
|
|
|
|
if __name__=='__main__': unittest.main()
|