"""Decision bridge boundaries without spawning a model or modifying host sessions."""
import json
from pathlib import Path
import sys
import tempfile
import unittest
from unittest.mock import patch
sys.path.insert(0,str(Path(__file__).resolve().parents[1]/'scripts'))
import decision
import decision_fleet
from worker import atomic_json


class Process:
    pid=42
    code=None
    def poll(self):return self.code
    def terminate(self):self.code=0
    def wait(self,timeout=None):return self.code
    def kill(self):self.code=-1


class DecisionBridgeTests(unittest.TestCase):
    def fixture(self,root):
        cfg={'control':{},'codex':'codex','state_file':str(Path(root)/'worker.json'),
             'decision_fleet':{'runtime_root':str(Path(root)/'decisions'),'enabled':True,'max_active':1}}
        binding={'decision_id':'d','session_id':'s','thread_id':'11111111-1111-4111-8111-111111111111',
                 'worktree':root,'status':'open','archive_status':'none'}
        return cfg,binding

    def test_only_bound_threads_and_bounded_startup_are_bridged(self):
        with tempfile.TemporaryDirectory() as root:
            cfg,binding=self.fixture(root);fleet=decision_fleet.Fleet(cfg)
            second={**binding,'decision_id':'other','session_id':'other','thread_id':'22222222-2222-4222-8222-222222222222'}
            with patch.object(decision_fleet.subprocess,'Popen',side_effect=lambda *a,**k:Process()) as spawn:
                status=fleet.tick([binding,second]);fleet.tick([binding,second])
                self.assertEqual(spawn.call_count,2)
                self.assertEqual(status['waiting'],1)
                self.assertEqual(status['active'],1)
                self.assertTrue(all('app-server' not in call.args[0] for call in spawn.call_args_list))
                self.assertEqual(fleet.children['d']['cfg']['thread_id'],binding['thread_id'])
                self.assertNotIn('notification_targets',fleet.children['d']['cfg'])
                fleet.close()

    def test_binding_change_and_duplicate_target_never_redirect_running_work(self):
        with tempfile.TemporaryDirectory() as root:
            cfg,b=self.fixture(root);fleet=decision_fleet.Fleet(cfg)
            with patch.object(decision_fleet.subprocess,'Popen',side_effect=lambda *a,**k:Process()) as spawn:
                fleet.tick([b]);old=fleet.children['d']['cfg']['thread_id']
                result=fleet.tick([{**b,'thread_id':'22222222-2222-4222-8222-222222222222'}])
                self.assertEqual(result['status'],'failed')
                self.assertEqual(spawn.call_count,2)
                self.assertEqual(fleet.children['d']['cfg']['thread_id'],old)
                with self.assertRaises(ValueError):fleet.tick([b,{**b,'decision_id':'other'}])
                fleet.close()

    def test_archive_waits_for_pending_completion_and_does_not_kill_model(self):
        with tempfile.TemporaryDirectory() as root:
            cfg,b=self.fixture(root);fleet=decision_fleet.Fleet(cfg)
            with patch.object(decision_fleet.subprocess,'Popen',side_effect=lambda *a,**k:Process()):
                fleet.tick([b]);child=fleet.children['d'];state=Path(child['cfg']['state_file'])
                atomic_json(state,{'inflight':{'message_id':'message'}})
                result=fleet.tick([{**b,'status':'completed','archive_status':'archived'}])
                self.assertEqual(result['status'],'failed')
                self.assertIn('d',fleet.children)
                self.assertTrue(all(p.poll() is None for p in child['processes'].values()))
                atomic_json(state,{})
                fleet.tick([{**b,'status':'completed','archive_status':'archived'}])
                self.assertEqual(fleet.children,{})
                self.assertTrue(state.exists())

    def test_unknown_inflight_is_never_restarted(self):
        with tempfile.TemporaryDirectory() as root:
            cfg,b=self.fixture(root);fleet=decision_fleet.Fleet(cfg)
            with patch.object(decision_fleet.subprocess,'Popen',side_effect=lambda *a,**k:Process()) as spawn:
                fleet.tick([b]);child=fleet.children['d']
                atomic_json(Path(child['cfg']['state_file']),{'needs_reconciliation':True})
                child['processes']['worker'].code=1
                self.assertEqual(fleet.tick([b])['status'],'failed')
                self.assertEqual(spawn.call_count,2)
                fleet.close()

    def test_consultation_helper_cannot_choose_other_decision_or_review(self):
        calls=[]
        class Client:
            def __init__(self,*args):pass
            def call(self,*args):calls.append(args);return {}
        cfg={'control':{},'decision_id':'decision-one'}
        with patch.object(decision,'Client',Client):
            decision.execute(cfg,'show')
            decision.execute(cfg,'ask',{'question':'What is the dependency?'})
            decision.execute(cfg,'ask',{'question':'What is the dependency?'})
            self.assertEqual(calls[1][1]['idempotency_key'],calls[2][1]['idempotency_key'])
            self.assertEqual(calls[0][0],'/api/decisions/decision-one')
            with self.assertRaises(ValueError):decision.execute(cfg,'review',{})
            with self.assertRaises(ValueError):decision.execute({'control':{},'decision_id':'../other'},'show')
            with self.assertRaises(ValueError):decision.execute(cfg,'edit',{'title':'No revision'})


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