#!/usr/bin/env python3
"""Explicit real-agent proof; only dedicated temporary sessions are mutated."""
import asyncio
import json
import os
from pathlib import Path
import sys
import tempfile

REPO = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(REPO / 'server'))
from device_gateway.control.codex import CodexAdapter
from device_gateway.control.pi import PiAdapter
from device_gateway.control.models import Project, Settings
from device_gateway.control.projects import ProjectRegistry
from device_gateway.integrations.pi import PiSessionRegistry


async def main():
    root = Path(tempfile.mkdtemp(prefix='device-settings-040-'))
    os.environ['DEVICE_LAB_CONTROL_DIR'] = str(root / 'runtime')
    os.environ['DEVICE_LAB_GATEWAY_URL'] = 'http://127.0.0.1:1'
    projects = ProjectRegistry(root / 'projects.json')
    for agent in ('pi', 'codex'):
        path = root / agent
        path.mkdir()
        projects.add(Project(id=agent, name='Settings smoke', path=str(path)))
    result = {'proof_directory': str(root)}
    pi = PiAdapter(projects, PiSessionRegistry())
    codex = CodexAdapter(projects)
    try:
        sid = (await pi.rpc.call('create_smoke', {'cwd': str(root / 'pi')}))['session_id']
        await pi.rpc.call('input', {'cwd': str(root/'pi'), 'session_id': sid,
            'text':'Reply exactly PI_SETTINGS_INITIAL. Do not use tools.'})
        for _ in range(120):
            initial = await pi.rpc.call('smoke_result', {'session_id': sid})
            if initial['state'] == 'IDLE' and initial['text']: break
            await asyncio.sleep(1)
        assert initial['text'].strip() == 'PI_SETTINGS_INITIAL', 'Pi initial turn failed'
        current = await pi.get_settings(sid)
        models = current['available_models']
        chosen = next((m['id'] for m in models if m['id'] != current['model'] and m['id'].split('/')[0] == current['model'].split('/')[0]), current['model'])
        updated = await pi.update_settings(sid, Settings(model=chosen))
        assert updated['model'] == chosen and updated['session_id'] == sid
        await pi.send_input(sid, 'Reply exactly PI_SETTINGS_READY. Do not use tools.')
        for _ in range(120):
            check = await pi.rpc.call('smoke_result', {'session_id': sid})
            if check['state'] == 'IDLE' and check['text']:
                break
            await asyncio.sleep(1)
        assert check['text'].strip() == 'PI_SETTINGS_READY', 'Pi trivial turn failed'
        assert check['model'] == chosen.split('/', 1)[1]
        await pi.rpc.call('stop', {'session_id': sid})
        # SDK persists model changes in the same session, including after a restart.
        dormant = await pi.get_session(sid)
        assert dormant.session_id == sid and dormant.model == chosen
        result['pi'] = dict(status='PASS', session_id=sid, models=[m['id'] for m in models],
            selected=chosen, idempotent_only=chosen==current['model'], dormant_persistence=True, output=check['text'])
        print('Pi settings smoke PASS', flush=True)
        created = await codex.call('thread/start', dict(cwd=str(root/'codex'), approvalPolicy='on-request',
            sandbox='read-only', baseInstructions='Connectivity test. Never use tools. Reply exactly as requested.'))
        sid = created['thread']['id']
        # New threads do not have a readable rollout until their first turn.
        await codex.call('turn/start', {'threadId':sid,'input':[{'type':'text','text':'Reply exactly CODEX_SETTINGS_INITIAL'}]})
        for _ in range(120):
            initial=codex.live.get(sid,{})
            if initial.get('state') in ('DONE','FAILED'):break
            await asyncio.sleep(1)
        assert initial.get('state')=='DONE', 'Codex initial turn failed'
        current = await codex.get_settings(sid)
        models = current['available_models']
        # Keep the authenticated current model; vary effort using advertised metadata.
        chosen = next(m for m in models if m['id'] == current['model'])
        effort = next((e for e in chosen['available_efforts'] if e != current['effort']), current['effort'])
        updated = await codex.update_settings(sid, Settings(model=chosen['id'], effort=effort))
        assert updated['model'] == chosen['id'] and updated['effort'] == effort, 'Codex settings readback mismatch'
        await codex.send_input(sid, 'Reply exactly CODEX_SETTINGS_READY')
        for _ in range(120):
            check = await codex.get_session(sid)
            if check.state in ('DONE', 'FAILED'):
                break
            await asyncio.sleep(1)
        result['codex_turn_check'] = dict(state=check.state, text=check.recent_text)
        assert check.state == 'DONE' and check.recent_text.strip() == 'CODEX_SETTINGS_READY', 'Codex trivial turn failed'
        assert check.model == chosen['id'] and check.effort == effort
        await codex.call('thread/unsubscribe', {'threadId': sid})
        dormant = await codex.get_session(sid)
        readback = await codex.get_settings(sid)
        result['codex'] = dict(status='PASS', session_id=sid, models=[m['id'] for m in models],
            selected=chosen['id'], available_efforts=chosen['available_efforts'], effort=effort,
            dormant_state=dormant.state, dormant_settings=readback['model']==chosen['id'] and readback['effort']==effort,
            output=check.recent_text)
        print('Codex settings smoke PASS', flush=True)
    except Exception as exc:
        result['error'] = str(exc)
        print('Settings smoke failed:', str(exc), flush=True)
    finally:
        await pi.close()
        await codex.close()
        (root/'result.json').write_text(json.dumps(result, indent=2)+'\n')
        print(json.dumps(result, indent=2), flush=True)
    if 'error' in result:
        raise SystemExit(1)


if __name__ == '__main__':
    asyncio.run(main())