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()