import asyncio import logging from pydantic import ValidationError from .catalog import Catalog from .models import ControlError,Input,Settings from .user_input import secret_question CAPABILITIES={'catalog','target-select','permissions','native-interaction','session-create'} logger=logging.getLogger('uvicorn.error.device_gateway.navigator') class Navigator: def __init__(self,projects,adapters,targets,operations,user_inputs,send,current,payload,invalidate): self.projects,self.adapters,self.targets,self.operations,self.user_inputs=projects,adapters,targets,operations,user_inputs self.send,self.current,self.payload,self.invalidate=send,current,payload,invalidate self.catalog=Catalog(projects,adapters) def valid(self,device,generation): return self.current(device) and self.targets.generation(device.device_id)==generation def adapter(self,agent): if agent not in self.adapters:raise ControlError('Adapter unavailable',503) return self.adapters[agent] async def deliver_target(self,device,value,rid=None): generation=value['generation'];target=value['target'] payload=await self.payload(target) from device_gateway.state import split_target selected=split_target(target) if selected and 'permissions' not in payload: adapter=self.adapter(selected[0]) payload['permissions']=await adapter.permissions(selected[1]) if selected[0]=='codex' and hasattr(adapter,'permissions') else dict(current='Native',available_modes=[]) if selected[0]=='pi' else dict(current=None,available_modes=[],status='Unsupported') if not self.valid(device,generation):return # Persisted CAS + ACK + snapshot in one send-lock batch prevents interleaving. async with device.send_lock: if not self.valid(device,generation):return if rid:await asyncio.wait_for(device.websocket.send_json(dict(type='target_selected',request_id=rid,target=target,target_generation=generation,project_id=value['project_id'])),3) await asyncio.wait_for(device.websocket.send_json(dict(payload,target_generation=generation,project_id=value['project_id'])),3) async def select(self,device,event): generation=event.get('expected_target_generation') if not self.valid(device,generation):raise ControlError('Stale target generation',409) project=self.projects.get(event.get('project_id')) agent=event.get('agent');sid=event.get('session_id') session=await self.adapter(agent).get_session(sid) if session.project_id!=project.id or session.cwd!=project.path:raise ControlError('Session disappeared',409) if not self.valid(device,generation):raise ControlError('Stale target generation',409) value=self.targets.set(device.device_id,f'agent:{agent}:{sid}',project.id,expected=generation) if value is None:raise ControlError('Stale target generation',409) self.invalidate(device) await self.deliver_target(device,value,event.get('request_id')) return value async def creation_options(self,device,event): project=self.projects.get(event.get('project_id'));agent=event.get('agent');adapter=self.adapter(agent) import time from uuid import uuid4 cursor=event.get('cursor',0);snapshot=event.get('snapshot');kind='models:'+agent if type(cursor) is not int or cursor<0 or cursor%4:raise ControlError('Stale cursor',409) if snapshot: data=self.catalog.snapshots.get(snapshot) if not data or data['connection']!=device.connection_id or data['kind']!=kind or data['project_id']!=project.id:raise ControlError('Stale snapshot',409) if time.monotonic()>=data['expires']:raise ControlError('Expired snapshot',410) models=data['items'] else: if cursor:raise ControlError('Stale cursor',409) models=await (adapter.list_models() if agent=='codex' else adapter.creation_models(project)) snapshot=uuid4().hex;data=dict(connection=device.connection_id,kind=kind,project_id=project.id,items=models,expires=time.monotonic()+60) self.catalog.snapshots[snapshot]=data while len(self.catalog.snapshots)>64:self.catalog.snapshots.popitem(last=False) if cursor>len(models):raise ControlError('Stale cursor',409) result=dict(type='creation_options',snapshot=snapshot,ttl_seconds=max(0,int(data['expires']-time.monotonic())),agent=agent,project_id=project.id,cursor=cursor,next_cursor=cursor+4 if cursor+43900:raise ControlError('Model catalog frame too large',413) return result async def create(self,device,event,text): project=self.projects.get(event.get('project_id'));agent=event.get('agent');adapter=self.adapter(agent) Input(text=text);settings=Settings.model_validate(event.get('settings',{}));oid=event.get('operation_id') payload=dict(project_id=project.id,agent=agent,settings=settings.model_dump(),text=text,expected_target_generation=event.get('expected_target_generation')) if not self.operations.begin(device.device_id,oid,payload):return self.operations.get(device.device_id,oid) stage='native_create' generation=event.get('expected_target_generation') if not self.valid(device,generation):return self.operations.write(device.device_id,oid,'FAILED',message='Stale target generation') def created(sid):self.operations.write(device.device_id,oid,'CREATED',agent=agent,session_id=sid,project_id=project.id) try: kwargs={'operation_id':oid} if agent=='pi' else {} result=await adapter.create_session(project,text,settings,created,**kwargs) result={k:result[k] for k in ('agent','session_id','project_id')} status=self.operations.write(device.device_id,oid,'DONE',**result) stage='target_delivery' if self.valid(device,generation): value=self.targets.set(device.device_id,f"agent:{agent}:{result['session_id']}",project.id,expected=generation) if value: self.invalidate(device);await self.deliver_target(device,value,oid) status=self.operations.write(device.device_id,oid,'DONE',**result,target_generation=value['generation']) return status except Exception as exc: # Native create/turn acceptance could already have happened. Keep # CREATED/UNKNOWN and native identity, and never blindly retry. status=self.operations.get(device.device_id,oid) logger.info('[SESSION_CREATE] stage=%s agent=%s state=%s error_status=%s', stage,agent,status['state'],exc.status if isinstance(exc,ControlError) else 'unavailable') return status async def voice_prepare(self,device,event): context=event.get('context') if not context:return None if not isinstance(context,dict):raise ControlError("Malformed voice context",400) if context.get('kind')=='message': generation=context.get('expected_target_generation') if 'target-select' not in device.capabilities or not self.valid(device,generation) or context.get('target')!=self.targets.get(device.device_id):raise ControlError('Stale voice target',409) from device_gateway.state import split_target parsed=split_target(context['target']) if not parsed:raise ControlError('No Agent selected',409) session=await self.adapter(parsed[0]).get_session(parsed[1]) if not self.valid(device,generation):raise ControlError('Stale voice target',409) return dict(agent=parsed[0],session_id=parsed[1],name=session.name,context=context) if 'native-interaction' not in device.capabilities:raise ControlError('Unsupported voice context',400) if context.get('kind')=='user_input': rid,qid=context.get('request_id'),context.get('question_id') if not self.user_inputs.valid(device.connection_id,rid,qid):raise ControlError('Stale native request',409) entry=self.user_inputs.entries[rid];question=entry['questions'][entry['index']] if secret_question(question):raise ControlError('Answer on desktop',409) if question.get('options') and not question.get('isOther'):raise ControlError('Select a native option',409) return dict(agent=entry['agent'],session_id=entry['session_id'],name=question['header'],context=context) if context.get('kind')=='new_session' and 'session-create' in device.capabilities: self.projects.get(context.get('project_id'));self.adapter(context.get('agent')) if not self.valid(device,context.get('expected_target_generation')):raise ControlError('Stale target generation',409) return dict(agent=context['agent'],session_id='',name='New Session',context=context) raise ControlError('Unsupported voice context',400) def voice_valid(self,device,context): if not context:return not self.user_inputs.pending() if context['kind']=='message':return not self.user_inputs.pending() and self.valid(device,context.get('expected_target_generation')) and context.get('target')==self.targets.get(device.device_id) if context['kind']=='user_input':return self.user_inputs.valid(device.connection_id,context.get('request_id'),context.get('question_id')) return self.valid(device,context.get('expected_target_generation')) async def voice_submit(self,device,context,text): if not self.voice_valid(device,context):raise ControlError('Stale voice context',409) if context['kind']=='message': from device_gateway.state import split_target agent,sid=split_target(context['target']);await self.adapter(agent).send_input(sid,text) elif context['kind']=='user_input': if not self.user_inputs.resolve(device.connection_id,context['request_id'],context['question_id'],text):raise ControlError('Stale native answer',409) else: result=await self.create(device,context,text) await self.send(device,dict(type='session_operation',**result)) if result['state']!='DONE':raise ControlError('Creation outcome unknown; query operation',409) async def handle(self,device,event): kind=event['type'];rid=event.get('request_id') capability='catalog' if kind.startswith('catalog_') else 'session-create' if kind.startswith(('session_','creation_')) else 'target-select' if kind.startswith('target_') else 'native-interaction' if capability not in device.capabilities:return if not isinstance(rid,str) or not 1<=len(rid)<=64:return try: if kind=='catalog_get':result=await self.catalog.page(device.connection_id,event.get('kind'),event.get('project_id'),event.get('snapshot'),event.get('cursor',0)) elif kind=='target_select':await self.select(device,event);return elif kind=='target_get':await self.deliver_target(device,self.targets.selection(device.device_id),rid);return elif kind=='creation_get':result=await self.creation_options(device,event) elif kind=='session_create':result=dict(type='session_operation',**await self.create(device,event,event.get('text'))) elif kind=='session_operation_get':result=dict(type='session_operation',**self.operations.get(device.device_id,event.get('operation_id'))) elif kind=='user_input_response': self.user_inputs.resolve(device.connection_id,event.get('request_id'),event.get('question_id'),event.get('answer'),event.get('choice_index'));return else:return if self.current(device):await self.send(device,dict(result,request_id=rid)) except (ControlError,ValidationError) as exc: if self.current(device):await self.send(device,dict(type='navigator_error',request_id=rid,message=exc.detail if isinstance(exc,ControlError) else 'Malformed request',status=exc.status if isinstance(exc,ControlError) else 400))