AI Flow

import importlib.util
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch

spec = importlib.util.spec_from_file_location('supervisor', Path(__file__).with_name('supervisor.py'))
s = importlib.util.module_from_spec(spec)
spec.loader.exec_module(s)

class PolicyTests(unittest.TestCase):
    def policy(self, **kw):
        args = dict(status={'type':'idle'}, turn={'id':'turn1','status':'completed'},
                    queue=[], guard={'decision':'continue','mayStop':False}, previous={})
        args.update(kw)
        return s.decision(**args)
    def test_completed_incomplete_run_continues(self):
        self.assertEqual(self.policy(), 'continue')
    def test_active_and_approval_waits_never_prompt(self):
        for status in ['active','notLoaded','systemError']:
            self.assertEqual(self.policy(status={'type':status,'activeFlags':['waitingOnApproval']}), 'waiting-for-idle')
    def test_no_duplicate_or_user_queue_interference(self):
        self.assertEqual(self.policy(queue=[{'id':'human'}]), 'user-queue-pending')
        self.assertEqual(self.policy(previous={'attemptedTurn':'turn1'}), 'waiting-for-queued-turn')
    def test_interruptions_and_failures_stop(self):
        for status in ['interrupted','failed','inProgress']:
            self.assertTrue(self.policy(turn={'id':'t','status':status}).startswith('paused'))
    def test_stop_instruction_and_completion(self):
        for text in ['pause until tomorrow', 'please stop', 'can u pause?', 'cancel this']:
            t={'id':'t','status':'completed','items':[{'type':'userMessage','content':[{'text':text}]}]}
            self.assertEqual(self.policy(turn=t), 'paused-user-instruction')
        self.assertEqual(self.policy(guard={'decision':'complete','mayStop':True}), 'complete')
    def test_invalid_guard_refuses(self):
        for g in [{},{'decision':'continue','mayStop':True},{'decision':'user-paused'}]:
            self.assertTrue(self.policy(guard=g).startswith('error'))

class ProgressTests(unittest.TestCase):
    def test_component_review_changes_count_but_activity_does_not(self):
        with tempfile.TemporaryDirectory() as d:
            p = Path(d)
            s.write(p/'run.json', {'currentStep':'components','updatedAt':'old'})
            s.write(p/'components.json', {'components':[{'reference':'Q1','review':{'status':'pending'}}]})
            before = s.progress_hash(p, {'remaining':['libraries']})
            s.write(p/'run.json', {'currentStep':'components','updatedAt':'new'})
            s.write(p/'components.json', {'updatedAt':'new','components':[{'reference':'Q1','review':{'status':'pending','checkedAt':'new'}}]})
            self.assertEqual(before, s.progress_hash(p, {'remaining':['libraries']}))
            s.write(p/'components.json', {'components':[{'reference':'Q1','review':{'status':'stale','geometryEvidence':['lead-hole-review.json']}}]})
            self.assertNotEqual(before, s.progress_hash(p, {'remaining':['libraries']}))
    def test_invalid_component_register_fails_closed(self):
        with tempfile.TemporaryDirectory() as d:
            p=Path(d);s.write(p/'run.json', {})
            (p/'components.json').write_text('{broken')
            with self.assertRaises(ValueError):s.progress_hash(p, {})

class FakeClient:
    calls=[]
    race=False
    read_count=0
    fail=False
    auto=False
    def __init__(self,*args): pass
    def close(self): pass
    def rpc(self,method,params):
        self.calls.append((method,params))
        if method=='thread/read':
            type(self).read_count+=1
            return {'thread':{'status':{'type':'active' if self.race and self.read_count>=3 else 'idle'}}}
        if method=='thread/turns/list':
            if self.auto and self.read_count>=3:
                sent=next(p for m,p in self.calls if m=='thread/queue/add')
                return {'data':[{'id':'own-next-turn','status':'inProgress','items':[{'type':'userMessage','clientId':sent['clientUserMessageId']}]}]}
            return {'data':[{'id':'t','status':'completed'}]}
        if method=='thread/queue/list':return {'data':[]}
        if method=='thread/queue/add':
            if self.fail:raise TimeoutError('uncertain dispatch')
            return {'queuedSubmission':{'id':'q'}}
        return {}

class DispatchTests(unittest.TestCase):
    def setUp(self):
        FakeClient.calls=[];FakeClient.race=False;FakeClient.read_count=0;FakeClient.fail=False;FakeClient.auto=False
    def run_check(self, path, state):
        with patch.object(s,'discover',return_value=({'socket':'test'},{})), patch.object(s,'Codex',FakeClient), patch.object(s,'guard',return_value={'decision':'continue','mayStop':False}), patch.object(s,'progress_hash',return_value='same'):
            return s.check(path,path,'exact-thread',state)
    def test_dispatch_exact_thread_and_once(self):
        with tempfile.TemporaryDirectory() as d:
            p=Path(d)
            action,state=self.run_check(p,{})
            self.assertEqual(action,'continuation-dispatched')
            self.assertEqual(state['dispatch'],'start-acknowledged')
            self.assertTrue(all(params['threadId']=='exact-thread' for _,params in FakeClient.calls))
            self.run_check(p,s.read(p/'state.json'))
            self.assertEqual(sum(m=='thread/queue/add' for m,_ in FakeClient.calls),1)
    def test_ambiguous_send_not_retried(self):
        with tempfile.TemporaryDirectory() as d:
            p=Path(d);FakeClient.fail=True
            with self.assertRaises(TimeoutError):self.run_check(p,{})
            action,_=self.run_check(p,s.read(p/'state.json'))
            self.assertEqual(action,'waiting-for-queued-turn')
            self.assertEqual(sum(m=='thread/queue/add' for m,_ in FakeClient.calls),1)
    def test_state_change_removes_only_own_message(self):
        with tempfile.TemporaryDirectory() as d:
            FakeClient.race=True
            action,_=self.run_check(Path(d),{})
            self.assertEqual(action,'paused-dispatch-race')
            self.assertIn(('thread/queue/delete',{'threadId':'exact-thread','queuedSubmissionId':'q'}),FakeClient.calls)
            self.assertFalse(any(m=='thread/queue/start' for m,_ in FakeClient.calls))
    def test_auto_start_is_recognized_without_deleting_or_restarting(self):
        with tempfile.TemporaryDirectory() as d:
            FakeClient.race=True;FakeClient.auto=True
            action,state=self.run_check(Path(d),{})
            self.assertEqual(action,'continuation-dispatched')
            self.assertEqual(state['dispatch'],'auto-start-verified')
            self.assertEqual(state['startedTurn'],'own-next-turn')
            self.assertFalse(any(m in ['thread/queue/start','thread/queue/delete'] for m,_ in FakeClient.calls))
    def test_pause_and_no_progress(self):
        with tempfile.TemporaryDirectory() as d:
            p=Path(d);s.write(p/'pause.json',{})
            self.assertEqual(self.run_check(p,{})[0],'paused')
            self.assertFalse(FakeClient.calls)
            (p/'pause.json').unlink()
            self.assertEqual(self.run_check(p,{'progressHash':'same','unchanged':2})[0],'paused-no-recorded-progress')
            self.assertFalse(any(m=='thread/queue/add' for m,_ in FakeClient.calls))

if __name__=='__main__':unittest.main()