from __future__ import annotations import asyncio import contextlib import os import json from uuid import uuid4 from datetime import datetime, timezone from pathlib import Path from device_gateway.integrations.pi import normalize_activity, utf8_tail from .models import AgentSession, ControlError, Project from .rpc import JsonUnixSocket from .settings import idle_only, settings_view, validate_settings class CodexAdapter: agent_type = "codex" def __init__(self, projects, rpc=None, approvals=None): self.projects = projects self.on_activity = None self.on_terminal = None self.observed = set() self.approvals = approvals self.instance = uuid4().hex self.approval_tasks = set() self.request_tasks = {} self.approval_owners = set() self.file_changes = {} self.live: dict[str, dict] = {} 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) self.approval_owners.clear() for task in self.approval_tasks: task.cancel() self.request_tasks.clear() self.file_changes.clear() self.live.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"): if 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_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 == "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) 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 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") # availableDecisions is a presentation hint: 0.154.0 can omit # decline while its response schema still accepts it. Never replace # a one-shot decision with a session/policy amendment. if available is not None and "accept" not in available: raise ControlError("approval does not offer one-shot acceptance") 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")])) state = await self.approvals.wait(approval) if state in ("APPROVED", "REJECTED", "CANCELLED"): await self.rpc.reply({"id": message["id"], "result": { "decision": "accept" if state == "APPROVED" else "decline" if state == "REJECTED" else "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) def convert(self, thread, project): 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" 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=overlay.get("state", state) if live else "DORMANT", latest_activity=overlay.get("latest_activity", ""), recent_text=utf8_tail(overlay.get("recent_text", ""), 12000), final_text=overlay.get("final_text", "") if live and overlay.get("state") in ("DONE", "FAILED") else "", activity_type=overlay.get("activity_type"), waiting_reason=overlay.get("waiting_reason")) if sid in self.live: self.live[sid]["name"] = session.name return self.approvals.overlay(session) if self.approvals else session 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): thread = (await self.call("thread/read", {"threadId": session_id}))['thread'] for project in self.projects.list(): if project.path == thread["cwd"]: return self.convert(thread, project) 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"): 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) return settings_view("codex", session_id, current["model"], current.get("effort"), await self.list_models()) 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 = 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) await self.call("thread/settings/update", {"threadId": session_id, "model": model, "effort": effort}) current = await self._read_settings(session_id) self.live.setdefault(session_id, {}).update(model=current["model"], effort=current.get("effort")) return settings_view("codex", session_id, current["model"], current.get("effort"), models) async def close(self): self.disconnected() await asyncio.gather(*list(self.approval_tasks), return_exceptions=True) await self.rpc.close()