"""Opt-in SDK/Daemon lifecycle test; isolated from the production Daemon/device."""
import json
import os
from pathlib import Path
import re
import sys
import tempfile
import time
import unittest
from unittest.mock import patch
from urllib.request import urlopen

from watcherobot.cli import ensure_runtime, stop_runtime, _request_json


@unittest.skipUnless(os.environ.get('WATCHER_CODEX_RUN_INTEGRATION') == '1', 'Set WATCHER_CODEX_RUN_INTEGRATION=1 for isolated managed-runtime test')
class ManagedRuntimeTests(unittest.TestCase):
    def test_real_daemon_start_three_chats_three_rounds_restart_stop(self):
        with tempfile.TemporaryDirectory(prefix='codex-sdk-acceptance-') as temporary:
            base = Path(temporary)
            codex = base / 'codex-fixture'
            sessions = codex / 'sessions'
            sessions.mkdir(parents=True)
            paths = []
            for n in range(3):
                path = sessions / f'rollout-fixture-{n}.jsonl'
                path.write_text(json.dumps({'type': 'session_meta', 'payload': {'id': f'fixture-{n}', 'cwd': 'C:/fixture'}}) + '\n', encoding='utf-8')
                paths.append(path)
            (codex / 'session_index.jsonl').write_text(''.join(json.dumps({'id': f'fixture-{n}', 'thread_name': f'圆环任务{n + 1}'}, ensure_ascii=False) + '\n' for n in range(3)), encoding='utf-8')
            # Disable the SDK's default-port recovery probe as well as using
            # ephemeral child ports; a user's live Daemon must stay untouched.
            settings = {'WATCHER_RUNTIME_STATE_ROOT': str(base / 'runtime'), 'WATCHER_RUNTIME_INSTANCE_ROOT': str(base / 'instance'), 'WATCHER_RUNTIME_CONTROL_PORT': '0', 'WATCHER_CODEX_NO_BROWSER': '1', 'WATCHER_CODEX_DATA_HOME': str(codex)}
            with patch.dict(os.environ, settings):
                runtime, reused = ensure_runtime(state_root=base / 'runtime', ephemeral_ports=True)
                self.assertFalse(reused)
                try:
                    app_root = Path(__file__).resolve().parents[1]
                    def control(path, payload=None):
                        return _request_json(runtime.control_url, path, method='POST', payload=payload, timeout=15)
                    control('/daemon/application/select', {'application_dir': str(app_root), 'launcher': {'kind': 'python', 'executable': sys.executable}})
                    control('/daemon/application/start')
                    log_path = base / 'runtime' / 'logs' / 'applications' / 'com.orulink.codex_supervisor.jsonl'
                    def wait_preview():
                        end = time.monotonic() + 12
                        while time.monotonic() < end:
                            text = log_path.read_text(encoding='utf-8') if log_path.exists() else ''
                            urls = re.findall(r'Codex Supervisor preview: (http://127\.0\.0\.1:\d+)', text)
                            if urls:
                                try:
                                    with urlopen(urls[-1] + '/api/health', timeout=1) as response:
                                        if json.load(response)['ok']:
                                            return urls[-1]
                                except OSError:
                                    pass
                            time.sleep(.1)
                        self.fail('Managed Application did not expose a healthy preview')
                    url = wait_preview()
                    def wait_state(expected):
                        end = time.monotonic() + 8
                        while time.monotonic() < end:
                            with urlopen(url + '/api/state', timeout=2) as response:
                                state = json.load(response)
                            if len(state['sessions']) == 3 and all(item['status'] == expected for item in state['sessions']):
                                self.assertNotEqual(state['device']['state'], 'connected')
                                return state
                            time.sleep(.1)
                        self.fail('Three-chat lifecycle did not converge')
                    for round_number in range(1, 4):
                        for action, expected in (('task_started', 'running'), ('task_complete', 'waiting')):
                            for path in paths:
                                payload = {'type': action, 'turn_id': f'round-{round_number}', 'started_at' if action == 'task_started' else 'completed_at': time.time()}
                                with path.open('a', encoding='utf-8') as output:
                                    output.write(json.dumps({'type': 'event_msg', 'payload': payload}) + '\n')
                            state = wait_state(expected)
                            self.assertEqual([item['roundCountToday'] for item in state['sessions']], [round_number] * 3)
                    control('/daemon/application/restart')
                    url = wait_preview()
                    self.assertEqual(len(wait_state('waiting')['sessions']), 3)
                    control('/daemon/application/stop')
                    status = _request_json(runtime.control_url, '/daemon/status')['application']['state']
                    self.assertEqual(status, 'ended')
                    print('Managed SDK acceptance: 3 chats x 3 rounds, restart and graceful stop passed (device offline).')
                finally:
                    stop_runtime()
                    time.sleep(.5)


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