from __future__ import annotations import asyncio import contextlib import os import json import logging import time from collections import OrderedDict from uuid import uuid4 from datetime import datetime, timezone from pathlib import Path from device_gateway.integrations.pi import normalize_activity, utf8_tail from device_gateway.display_text import device_final_text from .models import AgentSession, ControlError, Project from .rpc import JsonUnixSocket from .settings import idle_only, settings_view, validate_settings class CodexAdapter: agent_type = "codex" LAST_REPLY_CACHE_MAX = 128 LAST_REPLY_PAGE_SIZE = 8 LAST_REPLY_MAX_PAGES = 4 LAST_REPLY_TIMEOUT = 5 def __init__(self, projects, rpc=None, approvals=None, user_inputs=None): self.projects = projects self.on_activity = None self.on_terminal = None self.observed = set() self.approvals = approvals self.user_inputs = user_inputs self.instance = uuid4().hex self.approval_tasks = set() self.request_tasks = {} self.approval_owners = set() self.file_changes = {} self.live: dict[str, dict] = {} # Only revision + device-bounded reply, never retained native history. self.last_replies = OrderedDict() self.history_lock = asyncio.Lock() self.creating_threads={} self.ready_generation = 0 self.initialize_lock = asyncio.Lock() self.input_lock = asyncio.Lock() # A single long-lived socket client reuses the machine's existing daemon. # It never starts, restarts or kills that shared daemon. codex_home = Path(os.environ.get("CODEX_HOME", str(Path.home() / ".codex"))) socket = os.environ.get("GATEWAY_CODEX_SOCKET", str(codex_home / "app-server-control/app-server-control.sock")) self.rpc = rpc or JsonUnixSocket(socket, on_event=self.event, on_disconnect=self.disconnected) def disconnected(self): if self.approvals: for owner in self.approval_owners: self.approvals.cancel_owner(owner) if self.user_inputs: for owner in self.approval_owners: self.user_inputs.cancel_owner(owner) self.approval_owners.clear() for task in self.approval_tasks: task.cancel() self.request_tasks.clear() self.file_changes.clear() self.live.clear() self.last_replies.clear() self.observed.clear() self.ready_generation = 0 async def ready(self): async with self.initialize_lock: await self.rpc.start() if self.ready_generation != self.rpc.generation: await self.rpc.call("initialize", {"clientInfo": {"name": "device_lab_gateway", "version": "0.5.0"}, "capabilities": {"experimentalApi": True}}) await self.rpc.write({"method": "initialized"}) self.ready_generation = self.rpc.generation async def call(self, method, params): await self.ready() return await self.rpc.call(method, params) async def event(self, message): method, params = message["method"], message.get("params", {}) session_id = params.get("threadId") if "id" in message: if method in ("item/commandExecution/requestApproval", "item/fileChange/requestApproval", "item/tool/requestUserInput"): is_input=method=="item/tool/requestUserInput" if (self.user_inputs if is_input else self.approvals) is None: await self.rpc.write({"id": message["id"], "result": {"decision": "cancel"}}) else: # Waiting must never block the socket reader or other sessions. key = (self.rpc.generation, type(message["id"]).__name__, message["id"]) if key in self.request_tasks: return task = asyncio.create_task((self.request_user_input if is_input else self.request_approval)(message, self.rpc.generation)) self.approval_tasks.add(task) self.request_tasks[key] = (task, session_id, params.get("turnId")) task.add_done_callback(self.approval_tasks.discard) task.add_done_callback(lambda done, key=key: self.request_tasks.pop(key, None)) else: await self.rpc.write({"id": message["id"], "error": {"code": -32601, "message": "Gateway does not support this server request"}}) return if method == "serverRequest/resolved": key = (self.rpc.generation, type(params.get("requestId")).__name__, params.get("requestId")) entry = self.request_tasks.get(key) if entry: entry[0].cancel() if not session_id: return state = self.live.setdefault(session_id, {}) if method in ("turn/started", "turn/completed"): self.last_replies.pop(session_id, None) if method == "turn/started": state.update(final_items={}, final_order=[], terminal_sent=False, final_text="", turn_id=params.get("turn", {}).get("id"), state="RUNNING", started_at=datetime.now(timezone.utc), latest_activity="", recent_text="", waiting_reason=None) elif method == "turn/completed": turn = params.get("turn", {}) if state.get("turn_id") and state["turn_id"] != turn.get("id"): return if state.get("terminal_sent") and state.get("turn_id") == turn.get("id"): return speak_terminal = state.get("turn_id") == turn.get("id") and not state.get("terminal_sent", True) state["terminal_sent"] = True for item in turn.get("items", []): self.final_item(state, item) items = state.get("final_items", {}) candidates = [items[i] for i in state.get("final_order", []) if i in items] finals = [i for i in candidates if i.get("phase") == "final_answer"] selected = (finals or [i for i in candidates if i.get("phase") != "commentary"]) state["final_text"] = selected[-1]["text"] if selected else "" for task, sid, turn_id in list(self.request_tasks.values()): if sid == session_id and turn_id == params.get("turn", {}).get("id"): task.cancel() if self.approvals: self.approvals.cancel_session("codex", session_id) if self.user_inputs: self.user_inputs.cancel_session("codex",session_id) turn = params.get("turn", {}) state["state"] = "DONE" if turn.get("status") == "completed" else "FAILED" if state.get("started_at"): state["elapsed_seconds"] = int((datetime.now(timezone.utc)-state["started_at"]).total_seconds()) if self.on_terminal and speak_terminal: self.on_terminal("codex", session_id, state["state"], state["final_text"], name=state.get("name", session_id), latest_activity=state.get("latest_activity", "")) elif method == "thread/settings/updated": state.update({k: params.get("threadSettings", {}).get(k) for k in ("model", "effort")}) elif method == "thread/status/changed": status = params.get("status", {}).get("type") if status == "active": state["state"] = "RUNNING" elif status in ("notLoaded", "systemError"): for task, sid, _ in list(self.request_tasks.values()): if sid == session_id: task.cancel() if self.approvals: self.approvals.cancel_session("codex", session_id) if self.user_inputs: self.user_inputs.cancel_session("codex",session_id) if status == "notLoaded": self.live.pop(session_id, None) else: state["state"] = "FAILED" elif method == "item/agentMessage/delta": value = normalize_activity(params.get("delta", "")) if params.get("turnId") == state.get("turn_id") and not state.get("terminal_sent", True): iid = params.get("itemId", "message") items = state.setdefault("final_items", {}) if iid not in items: if len(items) >= 64: old = state["final_order"].pop(0); items.pop(old, None) items[iid] = {"text": "", "phase": None}; state.setdefault("final_order", []).append(iid) items[iid]["text"] = (items[iid]["text"] + value)[:16000] state["recent_text"] = utf8_tail(state.get("recent_text", "") + value, 12000) state["latest_activity"] = utf8_tail(state["recent_text"], 480) state["activity_type"] = "text" elif method in ("item/started", "item/completed"): item = params.get("item", {}) kind = item.get("type") if params.get("turnId") == state.get("turn_id") and not state.get("terminal_sent", True): self.final_item(state, item) if kind == "fileChange": key = (session_id, item.get("id")) if method == "item/started": self.file_changes[key] = [c.get("path", "") for c in item.get("changes", [])][:40] else: self.file_changes.pop(key, None) if kind not in ("agentMessage", "userMessage", "reasoning"): state["latest_activity"] = f"{kind}: {'started' if method.endswith('/started') else 'completed'}" state["activity_type"] = "tool" if self.on_activity: self.on_activity(session_id, method) @staticmethod def final_item(state, item): if item.get("type") != "agentMessage": return iid = item.get("id", "message") items = state.setdefault("final_items", {}) if iid not in items: if len(items) >= 64: old = state["final_order"].pop(0); items.pop(old, None) state.setdefault("final_order", []).append(iid) items[iid] = {"text": item.get("text", "")[:16000], "phase": item.get("phase")} async def request_approval(self, message, generation): params = message.get("params", {}) approval = None owner = f"codex:{self.instance}:{generation}" self.approval_owners.add(owner) try: # Fields verified against locally generated Codex 0.154.0 schemas. if not all(isinstance(params.get(k), str) and params[k] for k in ("threadId", "turnId", "itemId")): raise ControlError("malformed approval identity") session = await self.get_session(params["threadId"]) if self.rpc.generation != generation or owner not in self.approval_owners: return file_change = message["method"] == "item/fileChange/requestApproval" available = params.get("availableDecisions") # Render the offered native semantics exactly. No policy amendments # are synthesized; structured decisions remain desktop-only. choices=available if available is not None else ["accept","acceptForSession","decline","cancel"] choices=[d for d in choices if isinstance(d,str) and d in ("accept","acceptForSession","decline","cancel")] if not choices: raise ControlError("Unsupported native decisions") if file_change: # FileChangeRequestApprovalParams has no `changes` field. Paths # arrive in item/started; reason remains available as a fallback. paths = self.file_changes.get((session.session_id, params["itemId"]), []) detail = "\n".join(paths) or "File changes: " + params["itemId"] else: detail = params.get("command") or "Command: " + params["itemId"] if params.get("reason"): detail += "\n" + params["reason"] approval = self.approvals.create(agent="codex", session_id=session.session_id, project_id=session.project_id, kind="file_change" if file_change else "command", title="Edit files?" if file_change else "Run command?", detail=detail, owner=owner, request_identity=json.dumps([message["id"], params["threadId"], params["turnId"], params["itemId"], params.get("approvalId")]), choices=choices) state = await self.approvals.wait(approval) if state in ("APPROVED", "REJECTED", "CANCELLED"): await self.rpc.reply({"id": message["id"], "result": { "decision": approval.decision or "cancel"}}, generation) except asyncio.CancelledError: pass except Exception: with contextlib.suppress(Exception): await self.rpc.reply({"id": message["id"], "result": {"decision": "cancel"}}, generation) finally: if approval: self.approvals.cancel(approval.approval_id) async def request_user_input(self,message,generation): entry=None;owner=f"codex:{self.instance}:{generation}";self.approval_owners.add(owner) try: params=message['params'];session=await self.get_session(params['threadId']) if self.rpc.generation!=generation:return entry=self.user_inputs.create('codex',session.session_id,session.project_id, json.dumps([message['id'],params['turnId'],params['itemId']]),params['questions'],owner) result=await self.user_inputs.wait(entry) await self.rpc.reply({'id':message['id'],'result':result},generation) except asyncio.CancelledError: pass except Exception: with contextlib.suppress(Exception): await self.rpc.reply({'id':message['id'],'error':{'code':-32602,'message':'Unsupported native user input'}},generation) finally: if entry:self.user_inputs.cancel(entry) async def permissions(self,session_id): # thread/resume joins the native runtime without overriding its settings. await self.get_session(session_id) result=await self.call('thread/resume',{'threadId':session_id,'excludeTurns':True}) if result['thread']['id']!=session_id:raise ControlError('Thread identity changed') self.observed.add(session_id) reviewer=result.get('approvalsReviewer') return dict(current=reviewer,approvalPolicy=result.get('approvalPolicy'),sandboxPolicy=result.get('sandbox'), activePermissionProfile=result.get('activePermissionProfile'), available_modes=[{'id':'user','name':'Ask for approval'},{'id':'auto_review','name':'Approve for me'}] if reviewer else []) async def create_session(self,project,text,settings,on_created): async with self.input_lock: models=await self.list_models() model,effort=validate_settings(settings,settings_view('codex','',None,None,models)) params={'cwd':project.path,'model':model} if settings.permissions is not None: if settings.permissions not in ('user','auto_review'):raise ControlError('Unsupported native reviewer',400) params['approvalsReviewer']=settings.permissions result=await self.call('thread/start',params) sid=result['thread']['id'];on_created(sid) self.creating_threads[sid]=(result['thread'],time.monotonic()) while len(self.creating_threads)>64:self.creating_threads.pop(next(iter(self.creating_threads))) if result['thread']['cwd']!=project.path:raise ControlError('Native cwd mismatch') self.observed.add(sid) if effort is not None: await self.call('thread/settings/update',{'threadId':sid,'effort':effort}) turn=await self.call('turn/start',{'threadId':sid,'input':[{'type':'text','text':text}]}) self.live.setdefault(sid,{}).update(state='RUNNING',turn_id=turn['turn']['id']) return dict(accepted=True,agent='codex',session_id=sid,project_id=project.id) def convert(self, thread, project, recovered=""): sid = thread["id"] status = thread.get("status", {}).get("type", "notLoaded") overlay = self.live.get(sid, {}) live = status != "notLoaded" state = "RUNNING" if status == "active" else "FAILED" if status == "systemError" else "IDLE" if live else "DORMANT" # Native events can arrive while a metadata/history request is in flight. active_overlay = overlay.get("state") in ("RUNNING", "WAITING_APPROVAL", "WAITING_INPUT") effective_state = overlay.get("state", state) if live or active_overlay else "DORMANT" if status == "active" and effective_state not in ("WAITING_APPROVAL", "WAITING_INPUT"): effective_state = "RUNNING" final = overlay.get("final_text", "") if overlay.get("state") in ("DONE", "FAILED") else "" if effective_state in ("DORMANT", "IDLE") and not final: final = recovered session = AgentSession(agent="codex", session_id=sid, project_id=project.id, name=thread.get("name") or thread.get("preview", "")[:160] or sid, cwd=thread["cwd"], updated_at=datetime.fromtimestamp(thread["updatedAt"], timezone.utc).isoformat(), model=thread.get("model") or overlay.get("model"), effort=thread.get("reasoningEffort") or overlay.get("effort"), live=live, elapsed_seconds=int((datetime.now(timezone.utc)-overlay["started_at"]).total_seconds()) if overlay.get("started_at") and overlay.get("state")=="RUNNING" else overlay.get("elapsed_seconds",0), state=effective_state, latest_activity=overlay.get("latest_activity", ""), recent_text=utf8_tail(overlay.get("recent_text", ""), 12000), final_text=final, activity_type=overlay.get("activity_type"), waiting_reason=overlay.get("waiting_reason")) if sid in self.live: self.live[sid]["name"] = session.name if self.user_inputs:session=self.user_inputs.overlay(session) if self.approvals:session=self.approvals.overlay(session) if session.state in ("RUNNING", "WAITING_APPROVAL", "WAITING_INPUT"): session.final_text = "" return session async def read_last_reply(self, session_id): # Codex 0.154.0 generated schema: descending turns, full items. Read only; # do not resume/load a dormant runtime or hydrate the entire thread. cursor = None for _ in range(self.LAST_REPLY_MAX_PAGES): page = await self.call("thread/turns/list", dict(threadId=session_id, limit=self.LAST_REPLY_PAGE_SIZE, cursor=cursor, sortDirection="desc", itemsView="full")) if not isinstance(page, dict) or not isinstance(page.get("data"), list): raise ValueError("unsupported history response") for turn in page["data"][:self.LAST_REPLY_PAGE_SIZE]: if not isinstance(turn, dict) or not isinstance(turn.get("items"), list): raise ValueError("unsupported turn response") if turn.get("status") != "completed": continue if turn.get("itemsView", "full") != "full": raise ValueError("incomplete turn items") messages = [item for item in turn["items"] if isinstance(item, dict) and item.get("type") == "agentMessage" and item.get("phase") != "commentary" and isinstance(item.get("text"), str) and item["text"].strip()] finals = [item for item in messages if item.get("phase") == "final_answer"] if messages: return device_final_text((finals or messages)[-1]["text"]) next_cursor = page.get("nextCursor") if not next_cursor: break if not isinstance(next_cursor, str) or next_cursor == cursor: raise ValueError("unsupported history cursor") cursor = next_cursor return "" async def recover_last_reply(self, thread): sid, revision = thread["id"], thread["updatedAt"] async with self.history_lock: cached = self.last_replies.get(sid) if cached is not None and cached[0] == revision: self.last_replies.move_to_end(sid) return cached[1] try: text = await asyncio.wait_for(self.read_last_reply(sid), self.LAST_REPLY_TIMEOUT) except Exception as exc: # Enrichment failure must preserve selectable native metadata. # Cache empty results too; periodic pushes must not retry/spam. logging.getLogger("uvicorn.error.device_gateway.codex").warning( "Codex last reply unavailable session=%s category=%s", sid[:64], type(exc).__name__) text = "" self.last_replies[sid] = (revision, text) self.last_replies.move_to_end(sid) while len(self.last_replies) > self.LAST_REPLY_CACHE_MAX: self.last_replies.popitem(last=False) return text async def list_sessions(self, project: Project): sessions, cursor = [], None while True: page = await self.call("thread/list", {"cwd": project.path, "limit": 100, "cursor": cursor}) sessions.extend(self.convert(t, project) for t in page["data"] if t["cwd"] == project.path) cursor = page.get("nextCursor") if not cursor: return sessions async def get_session(self, session_id): try: thread = (await self.call("thread/read", {"threadId": session_id, "includeTurns": False}))['thread'] self.creating_threads.pop(session_id,None) except ControlError as exc: creating=self.creating_threads.get(session_id) # Native thread/start + turn/start have acknowledged this identity. # Rollout creation is asynchronous; bridge only that brief metadata # window, without creating or retrying any turn. if not creating or time.monotonic()-creating[1]>10 or not ('rollout' in exc.detail and ('empty' in exc.detail or 'no rollout found' in exc.detail)):raise thread=creating[0] for project in self.projects.list(): if project.path == thread["cwd"]: session = self.convert(thread, project) if session.state in ("DORMANT", "IDLE") and not any((session.final_text, session.recent_text, session.latest_activity)): recovered = await self.recover_last_reply(thread) # Re-read the live overlay after awaiting history. return self.convert(thread, project, recovered) return session raise ControlError("session not found in registered projects", 404) async def observe_session(self, session_id): # Rejoin an already loaded runtime to receive notifications from turns # started by other clients. Never load a dormant thread merely to watch. if session_id not in self.observed: session = await self.get_session(session_id) if session.live: result = await self.call("thread/resume", {"threadId": session_id, "excludeTurns": True}) if result["thread"]["id"] != session_id: raise ControlError("Codex observation changed thread identity") self.observed.add(session_id) async def _resume(self, session_id): session = await self.get_session(session_id) result = await self.call("thread/resume", {"threadId": session_id, "excludeTurns": True}) if result["thread"]["id"] != session_id: raise ControlError("Codex resume changed thread identity") self.observed.add(session_id) self.live.setdefault(session_id, {}).update(model=result.get("model"), effort=result.get("reasoningEffort")) return not session.live async def resume_session(self, session_id): async with self.input_lock: return await self._resume(session_id) async def send_input(self, session_id, text): async with self.input_lock: resumed = await self._resume(session_id) session = await self.get_session(session_id) if session.state in ("RUNNING", "WAITING_APPROVAL", "WAITING_INPUT"): raise ControlError("Codex turn is already running; wait before sending another input", 409) result = await self.call("turn/start", {"threadId": session_id, "input": [{"type": "text", "text": text}]}) overlay = self.live.setdefault(session_id, {}) if overlay.get("turn_id") != result["turn"]["id"]: overlay.update(final_items={}, final_order=[], final_text="", terminal_sent=False, turn_id=result["turn"]["id"], state="RUNNING", started_at=datetime.now(timezone.utc), latest_activity="", recent_text="", waiting_reason=None) return {"accepted": True, "agent": "codex", "session_id": session_id, "resumed": resumed, "turn_id": result["turn"]["id"]} async def list_models(self, session_id=None): result, cursor = [], None while True: page = await self.call("model/list", {"cursor": cursor, "limit": 100}) result.extend(dict(id=m["model"], name=m["displayName"], is_default=m["isDefault"], default_effort=m.get("defaultReasoningEffort"), available_efforts=[e["reasoningEffort"] for e in m.get("supportedReasoningEfforts", [])]) for m in page["data"] if not m.get("hidden")) cursor = page.get("nextCursor") if not cursor: return result async def _read_settings(self, session_id): # thread/read includes persisted model/reasoningEffort in 0.154.0. # Reading settings does not resume a dormant thread or mutate a busy one. await self.get_session(session_id) thread = (await self.call("thread/read", {"threadId": session_id}))["thread"] return {"model": thread.get("model"), "effort": thread.get("reasoningEffort")} async def get_settings(self, session_id): async with self.input_lock: current = await self._read_settings(session_id) view=settings_view("codex", session_id, current["model"], current.get("effort"), await self.list_models()) view["permissions"]=await self.permissions(session_id) if (await self.get_session(session_id)).live else dict(current=None,available_modes=[],status="Load session to read native permissions") return view async def update_settings(self, session_id, value, *, still_valid=lambda: True): async with self.input_lock: idle_only(await self.get_session(session_id)) current = await self._read_settings(session_id) models = await self.list_models() model,effort=(current["model"],current.get("effort")) if value.permissions is not None and value.model is None and value.effort is None else validate_settings(value,settings_view("codex",session_id,current["model"],current.get("effort"),models)) idle_only(await self.get_session(session_id)) session = await self.get_session(session_id) if not session.live: await self._resume(session_id) idle_only(await self.get_session(session_id)) if not still_valid(): raise ControlError("settings request cancelled", 409) params={"threadId":session_id} if value.model is not None or value.effort is not None:params.update(model=model,effort=effort) if value.permissions is not None: if value.permissions not in ("user","auto_review"):raise ControlError("Unsupported native reviewer",400) params["approvalsReviewer"]=value.permissions await self.call("thread/settings/update",params) current = await self._read_settings(session_id) self.live.setdefault(session_id, {}).update(model=current["model"], effort=current.get("effort")) view=settings_view("codex",session_id,current["model"],current.get("effort"),models) view["permissions"]=await self.permissions(session_id) if (await self.get_session(session_id)).live else dict(current=None,available_modes=[],status="Load session to read native permissions") return view async def close(self): self.disconnected() await asyncio.gather(*list(self.approval_tasks), return_exceptions=True) await self.rpc.close()