from __future__ import annotations import asyncio from datetime import datetime, timezone from pathlib import Path import pytest from fastapi.testclient import TestClient import device_gateway.app as gateway from device_gateway.control.models import ControlError, Project from device_gateway.control.projects import ProjectRegistry from device_gateway.control.pi import PiAdapter from device_gateway.control.codex import CodexAdapter from device_gateway.integrations.pi import PiSessionEvent, PiSessionRegistry def test_project_registry_persistence_and_validation(tmp_path): registry = ProjectRegistry(tmp_path / 'config/projects.json') p = Project(id='test', name='Test', path=str(tmp_path)) registry.add(p) assert ProjectRegistry(registry.path).get('test') == p with pytest.raises(ControlError, match='already registered'): registry.add(p) with pytest.raises(ControlError, match='existing absolute'): registry.add(Project(id='missing', name='Missing', path=str(tmp_path / 'absent'))) registry.delete('test') assert ProjectRegistry(registry.path).list() == [] registry.path.write_text('{broken') with pytest.raises(ControlError, match='invalid project registry'): registry.list() assert registry.path.read_text() == '{broken' class PiRPC: def __init__(self, cwd): self.cwd = cwd self.live = False self.resumes = 0 self.inputs = [] self.closed = False async def call(self, method, params): if method == 'list': return [dict(session_id='history', name='Saved', cwd=self.cwd, updated_at='2026-01-01T00:00:00+00:00', state='IDLE' if self.live else 'DORMANT', live=self.live)] assert params['session_id'] == 'history' resumed = not self.live if resumed: if not params.get('allow_resume', True): raise ControlError('live runtime unavailable', 409) self.resumes += 1 self.live = True if method == 'input': self.inputs.append(params['text']) await asyncio.sleep(0) return dict(accepted=True, agent='pi', session_id='history', resumed=resumed) async def close(self): self.closed = True self.live = False def test_static_overlay_resume_input_and_restart(tmp_path): async def exercise(): projects = ProjectRegistry(tmp_path / 'projects.json') project = projects.add(Project(id='p', name='P', path=str(tmp_path))) live = PiSessionRegistry() rpc = PiRPC(str(tmp_path)) adapter = PiAdapter(projects, live, rpc) assert (await adapter.get_session('history')).state == 'DORMANT' responses = await asyncio.gather(adapter.send_input('history', 'one'), adapter.send_input('history', 'two')) assert [r['resumed'] for r in responses] == [True, False] assert rpc.resumes == 1 and rpc.inputs == ['one', 'two'] live.apply(PiSessionEvent(event='agent_start', session_id='history', name='Saved', cwd=str(tmp_path))) assert (await adapter.list_sessions(project))[0].state == 'RUNNING' assert len(await adapter.list_sessions(project)) == 1 live.apply(PiSessionEvent(event='session_shutdown', session_id='history', name='Saved', cwd=str(tmp_path))) await adapter.close() assert (await adapter.list_sessions(project))[0].state == 'DORMANT' restarted = PiAdapter(ProjectRegistry(projects.path), PiSessionRegistry(), PiRPC(str(tmp_path))) assert (await restarted.get_session('history')).state == 'DORMANT' with pytest.raises(ControlError) as error: await restarted.send_input('unknown', 'never') assert error.value.status == 404 asyncio.run(exercise()) def test_live_session_without_endpoint_is_not_duplicated(tmp_path): async def exercise(): projects = ProjectRegistry(tmp_path / 'projects.json') projects.add(Project(id='p', name='P', path=str(tmp_path))) live = PiSessionRegistry() live.apply(PiSessionEvent(event='session_start', session_id='history', name='Saved', cwd=str(tmp_path))) rpc = PiRPC(str(tmp_path)); adapter = PiAdapter(projects, live, rpc) with pytest.raises(ControlError): await adapter.send_input('history', 'input') assert rpc.resumes == 0 and rpc.inputs == [] asyncio.run(exercise()) def test_control_http_routes_and_validation(tmp_path, monkeypatch): projects = ProjectRegistry(tmp_path / 'projects.json') adapter = PiAdapter(projects, PiSessionRegistry(), PiRPC(str(tmp_path))) monkeypatch.setattr(gateway, 'project_registry', projects) monkeypatch.setattr(gateway, 'adapters', {'pi': adapter}) with TestClient(gateway.app) as client: assert client.post('/api/projects', json={'id':'p','name':'P','path':str(tmp_path)}).status_code == 201 assert client.get('/api/projects').json()[0]['id'] == 'p' assert client.get('/api/projects/p/sessions').json()['sessions'][0]['state'] == 'DORMANT' url='/api/agents/pi/sessions/history/input' assert client.post(url,json={'text':' '}).status_code == 422 r=client.post(url,json={'text':'hello'}) assert r.status_code == 202 and r.json()['resumed'] assert client.post(url,json={'text':'again'}).json()['resumed'] is False assert client.post('/api/agents/unknown/sessions/history/input',json={'text':'no'}).status_code == 404 assert client.delete('/api/projects/p').status_code == 204 assert client.get('/api/agents/pi/sessions/history').status_code == 404 class CodexRPC: def __init__(self, cwd): self.cwd=cwd; self.generation=1; self.messages=[]; self.loaded=False; self.turns=0 async def start(self): pass async def write(self, message): self.messages.append(message) async def close(self): pass def thread(self, sid): return dict(id=sid,cwd=self.cwd,updatedAt=1,preview='Saved',status={'type':'idle' if self.loaded else 'notLoaded'}) async def call(self, method, params): self.messages.append((method,params)) if method=='thread/list': return {'data':[self.thread('second' if params.get('cursor') else 'first')], 'nextCursor':None if params.get('cursor') else 'next'} if method=='thread/read': return {'thread':self.thread(params['threadId'])} if method=='thread/resume': self.loaded=True return {'thread':self.thread(params['threadId']),'model':'test'} if method=='turn/start': self.turns+=1; return {'turn':{'id':str(self.turns)}} return {} def test_codex_pagination_input_streaming_and_approval_seam(tmp_path): async def exercise(): projects=ProjectRegistry(tmp_path/'projects.json') p=projects.add(Project(id='p',name='P',path=str(tmp_path))) rpc=CodexRPC(str(tmp_path)); adapter=CodexAdapter(projects,rpc) assert len(await adapter.list_sessions(p))==2 assert (await adapter.get_session('first')).state=='DORMANT' assert (await adapter.send_input('first','hello'))['resumed'] with pytest.raises(ControlError): await adapter.send_input('first','again') await adapter.event({'method':'turn/started','params':{'threadId':'first'}}) await adapter.event({'method':'item/agentMessage/delta','params':{'threadId':'first','delta':'SESSION_RESUMED'}}) assert (await adapter.get_session('first')).recent_text=='SESSION_RESUMED' with pytest.raises(ControlError) as error: await adapter.send_input('first','busy') assert error.value.status==409 await adapter.event({'method':'item/commandExecution/requestApproval','id':'approval-1','params':{'threadId':'first'}}) assert rpc.messages[-1]=={'id':'approval-1','result':{'decision':'cancel'}} await adapter.event({'method':'turn/completed','params':{'threadId':'first','turn':{'status':'completed'}}}) assert (await adapter.get_session('first')).state=='DONE' adapter.disconnected();rpc.loaded=False;rpc.generation+=1 assert (await adapter.get_session('first')).state=='DORMANT' assert sum(isinstance(m,tuple) and m[0]=='initialize' for m in rpc.messages)==2 asyncio.run(exercise()) def test_rpc_process_eof_fails_pending_and_restarts(tmp_path): from device_gateway.control.rpc import JsonProcess import sys script = tmp_path / 'host.py' script.write_text('''import sys, json for line in sys.stdin: request=json.loads(line) if request['method']=='crash': sys.exit(1) print(json.dumps({'id':request['id'],'result':request['params']}),flush=True) ''') async def exercise(): rpc=JsonProcess([sys.executable,str(script)]) try: assert await rpc.call('echo',{'value':1})=={'value':1} with pytest.raises(ControlError,match='disconnected'): await rpc.call('crash',{}) await rpc.reader assert await rpc.call('echo',{'value':2})=={'value':2} assert rpc.generation==2 finally: await rpc.close() assert rpc.process is None asyncio.run(exercise()) def test_codex_socket_reconnect_and_no_input_replay(tmp_path): from websockets.asyncio.server import unix_serve from device_gateway.control.rpc import JsonUnixSocket import json async def exercise(): socket=str(tmp_path/'codex.sock'); seen=[] async def handler(ws): async for payload in ws: request=json.loads(payload);seen.append(request['method']) if request['method']=='drop': await ws.close();return await ws.send(json.dumps({'id':request['id'],'result':{'ok':True}})) async with unix_serve(handler,socket,compression=None): rpc=JsonUnixSocket(socket) try: assert await rpc.call('read',{})=={'ok':True} with pytest.raises(ControlError,match='disconnected'): await rpc.call('drop',{}) assert await rpc.call('read',{})=={'ok':True} assert rpc.generation==2 assert seen==['read','drop','read'] finally:await rpc.close() asyncio.run(exercise()) def test_codex_settings_native_readback_busy_and_capabilities(tmp_path): from device_gateway.control.models import Settings class SettingsRPC(CodexRPC): model='native-a';effort='balanced';updates=0 def thread(self,sid): return {**super().thread(sid),'model':self.model,'reasoningEffort':self.effort} async def call(self,method,params): if method=='model/list': return {'data':[dict(model='native-a',displayName='A',isDefault=True, defaultReasoningEffort='balanced',supportedReasoningEfforts=[{'reasoningEffort':e} for e in ['balanced','custom-effort']]), dict(model='native-b',displayName='B',isDefault=False, defaultReasoningEffort='minimal',supportedReasoningEfforts=[{'reasoningEffort':'minimal'}])]} if method=='thread/settings/update': self.updates+=1;self.model=params['model'];self.effort=params['effort'];return {} return await super().call(method,params) async def exercise(): projects=ProjectRegistry(tmp_path/'p.json');projects.add(Project(id='p',name='P',path=str(tmp_path))) rpc=SettingsRPC(str(tmp_path));adapter=CodexAdapter(projects,rpc) current=await adapter.get_settings('first') assert not rpc.loaded and current['available_efforts']==['balanced','custom-effort'] result=await adapter.update_settings('first',Settings(model='native-b')) assert result['model']=='native-b' and result['effort']=='minimal' and rpc.updates==1 with pytest.raises(ControlError):await adapter.update_settings('first',Settings(effort='custom-effort')) adapter.live['first']['state']='RUNNING' with pytest.raises(ControlError) as e:await adapter.update_settings('first',Settings(model='native-a')) assert e.value.status==409 and rpc.updates==1 adapter.live['first']['state']='IDLE' with pytest.raises(ControlError):await adapter.update_settings('first',Settings(model='native-a'),still_valid=lambda:False) assert rpc.updates==1 await adapter.close() asyncio.run(exercise())