#!/usr/bin/env python3 """Codex app-server transport and persistent-thread lifecycle primitives. Root runs and detached workers each own one client through this boundary. MMO callers control workers through the authenticated Agent-MCP kernel rather than connecting to a worker's private app-server process directly. """ from __future__ import annotations import base64 import contextlib import datetime as dt import hashlib import json import os import random import re import socket import struct import subprocess import tempfile import threading import time from collections import deque from collections.abc import Callable, Mapping, Sequence from pathlib import Path from typing import Any from mmo_util import ( append_jsonl, atomic_write_text, bounded_text, package_version, strict_json_loads, terminate_process_group, utc_now, ) from mmo_version import APP_SERVER_PROTOCOL_CODEX_VERSION class AppServerError(RuntimeError): """Codex app-server transport or protocol failure.""" class ControlDeliveryUnknown(AppServerError): """A worker control request may have been applied before transport failed.""" class ControlRequestRejected(AppServerError): """A worker explicitly rejected a delivered control request.""" # Initialization is local, while thread lifecycle requests may start or resume # required MCP servers. Keep the latter generous enough for slow external # integrations without turning individual model turns into wall-clock timers. APP_SERVER_INITIALIZE_TIMEOUT_SECONDS = 60.0 APP_SERVER_LIFECYCLE_TIMEOUT_SECONDS = 1200.0 APP_SERVER_RECOVERY_DELAYS_SECONDS = (1.0, 5.0, 15.0) APP_SERVER_GOAL_OBJECTIVE_MAX_CHARS = 4000 APP_SERVER_DYNAMIC_TOOL_TIMEOUT_SECONDS = APP_SERVER_LIFECYCLE_TIMEOUT_SECONDS + 60.0 PARTIAL_EVENT_WINDOW_BYTES = 32 * 1024 * 1024 APP_SERVER_PROTOCOL_FILE_COUNT = 401 APP_SERVER_PROTOCOL_SHA256 = "fcfeaf23728b96ab73916a21302eb7a16629e67ee99f7ee47b60fad6b6e5ee1a" APP_SERVER_OVERLOAD_RETRY_DELAYS_SECONDS = (0.1, 0.2, 0.4, 0.8) MAX_WEBSOCKET_MESSAGE_BYTES = 64 * 1024 * 1024 MAX_CONTROL_MESSAGE_BYTES = 4 * 1024 * 1024 APPROVAL_REQUEST_METHODS = frozenset( { "item/commandExecution/requestApproval", "item/fileChange/requestApproval", "item/permissions/requestApproval", "applyPatchApproval", "execCommandApproval", } ) PENDING_SERVER_REQUEST_METHODS = frozenset( { "item/tool/requestUserInput", "mcpServer/elicitation/request", *APPROVAL_REQUEST_METHODS, } ) UNSUPPORTED_SERVER_REQUEST_METHODS = frozenset( { "account/chatgptAuthTokens/refresh", "attestation/generate", } ) _RESPONSES_TOOL_NAME_PATTERN = re.compile(r"[^a-zA-Z0-9_-]+") _DYNAMIC_MCP_TOOL_PREFIX = "mmo_mcp__" def _responses_tool_name(value: str) -> str: """Return a non-empty Responses-compatible identifier component.""" normalized = _RESPONSES_TOOL_NAME_PATTERN.sub("_", value).strip("_") return normalized or "tool" def _flat_mcp_dynamic_tool_name(server: str, tool: str, *, salt: str = "") -> str: """Build the flat compatibility name for one namespaced MCP function.""" # Codex reserves the native ``mcp__`` namespace and rejects dynamic tool # declarations using it. Keep the shim visibly MMO-owned while preserving # the server/tool split in the model-facing name. base = f"{_DYNAMIC_MCP_TOOL_PREFIX}{_responses_tool_name(server)}__{_responses_tool_name(tool)}" if len(base) <= 128 and not salt: return base digest = hashlib.sha256(f"{server}\0{tool}\0{salt}".encode()).hexdigest()[:12] suffix = f"__{digest}" return base[: 128 - len(suffix)].rstrip("_") + suffix def _dynamic_tool_content_items(result: Mapping[str, Any]) -> list[dict[str, Any]]: """Translate one MCP call result into Codex dynamic-tool content items.""" items: list[dict[str, Any]] = [] decoded_text_values: list[Any] = [] content = result.get("content") if isinstance(content, list): for block in content: if not isinstance(block, Mapping): items.append( { "type": "inputText", "text": json.dumps(block, ensure_ascii=False, allow_nan=False), } ) continue block_type = block.get("type") if block_type == "text" and isinstance(block.get("text"), str): items.append({"type": "inputText", "text": block["text"]}) with contextlib.suppress(json.JSONDecodeError, ValueError): decoded_text_values.append(strict_json_loads(block["text"])) elif ( block_type == "image" and isinstance(block.get("data"), str) and isinstance(block.get("mimeType"), str) ): items.append( { "type": "inputImage", "imageUrl": f"data:{block['mimeType']};base64,{block['data']}", } ) elif ( block_type == "audio" and isinstance(block.get("data"), str) and isinstance(block.get("mimeType"), str) ): items.append( { "type": "inputAudio", "audioUrl": f"data:{block['mimeType']};base64,{block['data']}", } ) else: items.append( { "type": "inputText", "text": json.dumps( dict(block), ensure_ascii=False, separators=(",", ":"), allow_nan=False, ), } ) structured = result.get("structuredContent") structured_already_present = structured in decoded_text_values if isinstance(structured, Mapping) and len(structured) == 1: structured_already_present = ( structured_already_present or next(iter(structured.values())) in decoded_text_values ) if structured is not None and not structured_already_present: items.append( { "type": "inputText", "text": "structuredContent=" + json.dumps( structured, ensure_ascii=False, separators=(",", ":"), allow_nan=False, ), } ) if not items: items.append({"type": "inputText", "text": ""}) return items _PROVIDER_LIMIT_PATTERN = re.compile( r"(?:usage\s+limit|limit\s+(?:reached|exhausted)|quota|rate\s+limit)", re.IGNORECASE, ) _PROVIDER_RESET_PATTERN = re.compile( r"(?:limit\s+will\s+reset|reset)\s+at\s+" r"(?P\d{4}-\d{2}-\d{2}[ T]\d{2}:\d{2}:\d{2})" r"(?:\s*(?PZ|UTC|[+-]\d{2}:?\d{2}))?", re.IGNORECASE, ) _RETRYABLE_CODEX_TRANSPORT_ERRORS = frozenset( { "serverOverloaded", "internalServerError", "httpConnectionFailed", "responseStreamConnectionFailed", "responseStreamDisconnected", "responseTooManyFailedAttempts", } ) def normalize_turn_failure(turn: Mapping[str, Any]) -> dict[str, Any] | None: """Translate one authoritative failed turn into durable lifecycle evidence. Codex providers do not currently expose a common quota error enum. Preserve the complete error object, prefer its structured fields, and use the human message only to distinguish a provider limit from a generic failed turn. """ if turn.get("status") != "failed": return None raw_error = turn.get("error") error = dict(raw_error) if isinstance(raw_error, Mapping) else {} message_value = error.get("message") if error else raw_error message = str(message_value or "app-server turn failed") details = error.get("additionalDetails") searchable = "\n".join( value for value in (message, str(details) if details is not None else "") if value ) codex_error_info = error.get("codexErrorInfo") codex_error_kind = ( codex_error_info if isinstance(codex_error_info, str) else next(iter(codex_error_info), None) if isinstance(codex_error_info, Mapping) else None ) kind = "turn_failed" retryable = False if codex_error_kind == "usageLimitExceeded" or _PROVIDER_LIMIT_PATTERN.search(searchable): kind = "provider_usage_limited" retryable = True elif "failed to parse tool call arguments" in searchable.lower(): kind = "malformed_tool_arguments" retryable = True elif ( "stream disconnected" in searchable.lower() or codex_error_kind in _RETRYABLE_CODEX_TRANSPORT_ERRORS ): kind = "provider_transport" retryable = True retry_at: str | None = None retry_at_raw: str | None = None retry_at_timezone: str | None = None match = _PROVIDER_RESET_PATTERN.search(searchable) if match is not None: value = match.group("value") zone = match.group("zone") retry_at_raw = value + (f" {zone}" if zone else "") if zone is not None: retry_at_timezone = zone normalized_zone = "+00:00" if zone.upper() in {"Z", "UTC"} else zone if len(normalized_zone) == 5 and normalized_zone[3] != ":": normalized_zone = normalized_zone[:3] + ":" + normalized_zone[3:] with contextlib.suppress(ValueError): retry_at = dt.datetime.fromisoformat( value.replace(" ", "T") + normalized_zone ).isoformat() result: dict[str, Any] = { "kind": kind, "source": "turn/completed", "turn_id": turn.get("id"), "turn_status": "failed", "message": message, "codex_error_info": codex_error_info, "raw_error": raw_error, "retryable": retryable, "observed_at": utc_now(), } if retry_at_raw is not None: result["retry_at_raw"] = retry_at_raw if retry_at is not None and retry_at_timezone is not None: result["retry_at"] = retry_at result["retry_at_timezone"] = retry_at_timezone return result def app_server_message_thread_id(message: Mapping[str, Any]) -> str | None: """Return the thread identity carried by a notification or server request.""" raw_params = message.get("params") if not isinstance(raw_params, Mapping): return None for key in ("threadId", "conversationId"): value = raw_params.get(key) if isinstance(value, str): return value goal = raw_params.get("goal") if isinstance(goal, Mapping) and isinstance(goal.get("threadId"), str): return str(goal["threadId"]) thread = raw_params.get("thread") if isinstance(thread, Mapping) and isinstance(thread.get("id"), str): return str(thread["id"]) return None def app_server_socket_path(identity: str) -> Path: """Return a stable, private, portable AF_UNIX path for one logical host.""" runtime_root = os.environ.get("XDG_RUNTIME_DIR") if runtime_root: base = Path(runtime_root).expanduser().resolve() / "codex-mmo" else: owner = hashlib.sha256(f"{os.getuid()}:{Path.home().resolve()}".encode()).hexdigest()[:12] base = Path(tempfile.gettempdir()).resolve() / f"codex-mmo-{owner}" if base.exists() and (base.is_symlink() or not base.is_dir()): raise AppServerError(f"app-server socket root is unsafe: {base}") base.mkdir(parents=True, exist_ok=True, mode=0o700) with contextlib.suppress(OSError): base.chmod(0o700) name = hashlib.sha256(identity.encode("utf-8")).hexdigest()[:32] + ".sock" path = base / name if len(os.fsencode(path)) >= 104: raise AppServerError("app-server runtime directory is too long for AF_UNIX") return path def app_server_listen_command( binary: str, command_flags: Sequence[str], socket_path: Path, ) -> list[str]: return [ binary, *command_flags, "app-server", "--listen", f"unix://{socket_path}", ] def app_server_protocol_status(binary: str | None) -> dict[str, Any]: """Require the reviewed Codex 0.149 experimental app-server contract exactly.""" if not binary: return { "passed": False, "expected_codex_version": APP_SERVER_PROTOCOL_CODEX_VERSION, "expected_schema_sha256": APP_SERVER_PROTOCOL_SHA256, "error": "Codex binary not found", } try: version_result = subprocess.run( [binary, "--version"], text=True, capture_output=True, timeout=10, check=False, ) version_text = version_result.stdout.strip() observed_version = version_text.removeprefix("codex-cli ").strip() with tempfile.TemporaryDirectory(prefix="codex-mmo-app-schema-") as temporary: result = subprocess.run( [ binary, "app-server", "generate-json-schema", "--experimental", "--out", temporary, ], text=True, capture_output=True, timeout=30, check=False, ) schema_root = Path(temporary) files = sorted(path for path in schema_root.rglob("*") if path.is_file()) digest = hashlib.sha256() for path in files: digest.update(path.relative_to(schema_root).as_posix().encode("utf-8")) digest.update(b"\0") digest.update(path.read_bytes()) observed_sha256 = digest.hexdigest() version_matches = ( version_result.returncode == 0 and observed_version == APP_SERVER_PROTOCOL_CODEX_VERSION ) schema_matches = ( result.returncode == 0 and len(files) == APP_SERVER_PROTOCOL_FILE_COUNT and observed_sha256 == APP_SERVER_PROTOCOL_SHA256 ) return { "passed": version_matches and schema_matches, "expected_codex_version": APP_SERVER_PROTOCOL_CODEX_VERSION, "observed_codex_version": observed_version, "expected_schema_files": APP_SERVER_PROTOCOL_FILE_COUNT, "observed_schema_files": len(files), "expected_schema_sha256": APP_SERVER_PROTOCOL_SHA256, "observed_schema_sha256": observed_sha256, "version_matches": version_matches, "schema_matches": schema_matches, "schema_exit_code": result.returncode, "stderr": (version_result.stderr + result.stderr)[-2000:], } except (OSError, subprocess.SubprocessError) as exc: return { "passed": False, "expected_codex_version": APP_SERVER_PROTOCOL_CODEX_VERSION, "expected_schema_sha256": APP_SERVER_PROTOCOL_SHA256, "error": f"{type(exc).__name__}: {exc}", } def require_app_server_codex_version(binary: str) -> None: """Reject session admission unless the executable is the reviewed Codex release.""" try: result = subprocess.run( [binary, "--version"], text=True, capture_output=True, timeout=10, check=False, ) except (OSError, subprocess.SubprocessError) as exc: raise AppServerError(f"unable to verify Codex app-server version: {exc}") from exc observed = result.stdout.strip().removeprefix("codex-cli ").strip() if result.returncode != 0 or observed != APP_SERVER_PROTOCOL_CODEX_VERSION: raise AppServerError( "Codex app-server protocol version mismatch: expected " f"{APP_SERVER_PROTOCOL_CODEX_VERSION}, observed {observed or 'unavailable'}" ) def _exact_keys( value: Mapping[str, Any], *, required: set[str], allowed: set[str], label: str ) -> None: missing = sorted(required - set(value)) unexpected = sorted(set(value) - allowed) if missing or unexpected: details: list[str] = [] if missing: details.append("missing " + ", ".join(missing)) if unexpected: details.append("unexpected " + ", ".join(unexpected)) raise ValueError(f"{label} has invalid fields: {'; '.join(details)}") def _network_policy_amendment(value: Any, label: str) -> None: if not isinstance(value, Mapping): raise ValueError(f"{label} must be an object") _exact_keys(value, required={"action", "host"}, allowed={"action", "host"}, label=label) if value["action"] not in {"allow", "deny"} or not isinstance(value["host"], str): raise ValueError(f"{label} must contain an allow/deny action and string host") def _command_approval_decision(value: Any) -> None: if isinstance(value, str): if value not in {"accept", "acceptForSession", "decline", "cancel"}: raise ValueError("command approval decision is invalid") return if not isinstance(value, Mapping) or len(value) != 1: raise ValueError("command approval decision must be a supported string or object") if "acceptWithExecpolicyAmendment" in value: amendment = value["acceptWithExecpolicyAmendment"] if not isinstance(amendment, Mapping): raise ValueError("exec-policy amendment must be an object") _exact_keys( amendment, required={"execpolicy_amendment"}, allowed={"execpolicy_amendment"}, label="exec-policy amendment", ) rules = amendment["execpolicy_amendment"] if not isinstance(rules, list) or not all(isinstance(item, str) for item in rules): raise ValueError("execpolicy_amendment must be an array of strings") return if "applyNetworkPolicyAmendment" in value: amendment = value["applyNetworkPolicyAmendment"] if not isinstance(amendment, Mapping): raise ValueError("network-policy amendment wrapper must be an object") _exact_keys( amendment, required={"network_policy_amendment"}, allowed={"network_policy_amendment"}, label="network-policy amendment wrapper", ) _network_policy_amendment(amendment["network_policy_amendment"], "network-policy amendment") return raise ValueError("command approval decision object is invalid") def _legacy_approval_decision(value: Any) -> None: if isinstance(value, str): if value not in { "approved", "approved_for_session", "approved_mcp_policy_amendment", "timed_out", "abort", }: raise ValueError("legacy approval decision is invalid") return if not isinstance(value, Mapping) or len(value) != 1: raise ValueError("legacy approval decision must be a supported string or object") if "approved_execpolicy_amendment" in value: amendment = value["approved_execpolicy_amendment"] if not isinstance(amendment, Mapping): raise ValueError("approved exec-policy amendment must be an object") _exact_keys( amendment, required={"proposed_execpolicy_amendment"}, allowed={"proposed_execpolicy_amendment"}, label="approved exec-policy amendment", ) rules = amendment["proposed_execpolicy_amendment"] if not isinstance(rules, list) or not all(isinstance(item, str) for item in rules): raise ValueError("proposed_execpolicy_amendment must be an array of strings") return if "network_policy_amendment" in value: wrapper = value["network_policy_amendment"] if not isinstance(wrapper, Mapping): raise ValueError("legacy network-policy amendment wrapper must be an object") _exact_keys( wrapper, required={"network_policy_amendment"}, allowed={"network_policy_amendment"}, label="legacy network-policy amendment wrapper", ) _network_policy_amendment( wrapper["network_policy_amendment"], "legacy network-policy amendment" ) return if "denied" in value: denied = value["denied"] if not isinstance(denied, Mapping): raise ValueError("denied approval decision must be an object") _exact_keys( denied, required={"rejection"}, allowed={"rejection"}, label="denied approval decision", ) if not isinstance(denied["rejection"], str): raise ValueError("approval rejection must be a string") return raise ValueError("legacy approval decision object is invalid") def _permission_path(value: Any) -> None: if not isinstance(value, Mapping) or not isinstance(value.get("type"), str): raise ValueError("permission path must be a tagged object") path_type = value["type"] if path_type == "path": required, allowed, field = {"type", "path"}, {"type", "path"}, "path" elif path_type == "glob_pattern": required, allowed, field = {"type", "pattern"}, {"type", "pattern"}, "pattern" elif path_type == "special": required, allowed, field = {"type", "value"}, {"type", "value"}, "value" else: raise ValueError("permission path type is invalid") _exact_keys(value, required=required, allowed=allowed, label="permission path") if path_type != "special": if not isinstance(value[field], str): raise ValueError(f"permission path {field} must be a string") return special = value[field] if not isinstance(special, Mapping) or not isinstance(special.get("kind"), str): raise ValueError("special permission path must be a tagged object") kind = special["kind"] if kind in {"root", "minimal", "tmpdir", "slash_tmp"}: _exact_keys(special, required={"kind"}, allowed={"kind"}, label="special path") elif kind == "project_roots": _exact_keys( special, required={"kind"}, allowed={"kind", "subpath"}, label="project-roots special path", ) if ( "subpath" in special and special["subpath"] is not None and not isinstance(special["subpath"], str) ): raise ValueError("project-roots subpath must be a string or null") elif kind == "unknown": _exact_keys( special, required={"kind", "path"}, allowed={"kind", "path", "subpath"}, label="unknown special path", ) if not isinstance(special["path"], str) or ( "subpath" in special and special["subpath"] is not None and not isinstance(special["subpath"], str) ): raise ValueError("unknown special path values are invalid") else: raise ValueError("special permission path kind is invalid") def _permission_profile(value: Any) -> None: if not isinstance(value, Mapping): raise ValueError("permissions must be an object") _exact_keys( value, required=set(), allowed={"fileSystem", "network"}, label="permissions", ) filesystem = value.get("fileSystem") if filesystem is not None: if not isinstance(filesystem, Mapping): raise ValueError("permissions.fileSystem must be an object or null") _exact_keys( filesystem, required=set(), allowed={"entries", "globScanMaxDepth", "read", "write"}, label="permissions.fileSystem", ) entries = filesystem.get("entries") if entries is not None: if not isinstance(entries, list): raise ValueError("permissions.fileSystem.entries must be an array or null") for entry in entries: if not isinstance(entry, Mapping): raise ValueError("filesystem permission entry must be an object") _exact_keys( entry, required={"access", "path"}, allowed={"access", "path"}, label="filesystem permission entry", ) if entry["access"] not in {"read", "write", "deny"}: raise ValueError("filesystem permission access is invalid") _permission_path(entry["path"]) depth = filesystem.get("globScanMaxDepth") if depth is not None and ( not isinstance(depth, int) or isinstance(depth, bool) or depth < 1 ): raise ValueError("globScanMaxDepth must be a positive integer or null") for field in ("read", "write"): paths = filesystem.get(field) if paths is not None and ( not isinstance(paths, list) or not all(isinstance(item, str) for item in paths) ): raise ValueError(f"permissions.fileSystem.{field} must be strings or null") network = value.get("network") if network is not None: if not isinstance(network, Mapping): raise ValueError("permissions.network must be an object or null") _exact_keys( network, required=set(), allowed={"enabled"}, label="permissions.network", ) if network.get("enabled") is not None and not isinstance(network["enabled"], bool): raise ValueError("permissions.network.enabled must be a boolean or null") def validate_server_request_response(method: str, response: Mapping[str, Any]) -> None: """Validate one controller response against Codex app-server v2's method shape.""" if method == "item/tool/requestUserInput": _exact_keys(response, required={"answers"}, allowed={"answers"}, label=method) answers = response["answers"] if not isinstance(answers, Mapping) or not all(isinstance(key, str) for key in answers): raise ValueError("user-input answers must map string question IDs to answer objects") for answer in answers.values(): if not isinstance(answer, Mapping): raise ValueError("each user-input answer must be an object") _exact_keys( answer, required={"answers"}, allowed={"answers"}, label="user-input answer", ) values = answer["answers"] if not isinstance(values, list) or not all(isinstance(item, str) for item in values): raise ValueError("each user-input answer must contain an array of strings") return if method == "mcpServer/elicitation/request": _exact_keys( response, required={"action"}, allowed={"action", "content", "_meta"}, label=method, ) if response["action"] not in {"accept", "decline", "cancel"}: raise ValueError("MCP elicitation action is invalid") return if method in { "item/commandExecution/requestApproval", "item/fileChange/requestApproval", "applyPatchApproval", "execCommandApproval", }: _exact_keys(response, required={"decision"}, allowed={"decision"}, label=method) if method == "item/commandExecution/requestApproval": _command_approval_decision(response["decision"]) elif method == "item/fileChange/requestApproval": if response["decision"] not in {"accept", "acceptForSession", "decline", "cancel"}: raise ValueError("file-change approval decision is invalid") else: _legacy_approval_decision(response["decision"]) return if method == "item/permissions/requestApproval": _exact_keys( response, required={"permissions"}, allowed={"permissions", "scope", "strictAutoReview"}, label=method, ) _permission_profile(response["permissions"]) if response.get("scope", "turn") not in {"turn", "session"}: raise ValueError("permission grant scope is invalid") if response.get("strictAutoReview") is not None and not isinstance( response["strictAutoReview"], bool ): raise ValueError("strictAutoReview must be a boolean or null") return raise ValueError(f"unsupported pending app-server request method: {method}") class UnixWebSocket: """Minimal RFC 6455 client for Codex's Unix-domain app-server listener.""" def __init__(self, path: Path, *, timeout: float) -> None: if not path.is_absolute(): raise AppServerError("app-server Unix socket path must be absolute") if len(os.fsencode(path)) >= 104: raise AppServerError("app-server Unix socket path exceeds the portable AF_UNIX limit") self.path = path self._socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) self._socket.settimeout(timeout) self._buffer = bytearray() self._send_lock = threading.Lock() self._closed = False try: self._socket.connect(str(path)) self._handshake() self._socket.settimeout(None) except BaseException: self._socket.close() self._closed = True raise @property def closed(self) -> bool: return self._closed def _handshake(self) -> None: key = base64.b64encode(os.urandom(16)).decode("ascii") request = ( "GET / HTTP/1.1\r\n" "Host: localhost\r\n" "Upgrade: websocket\r\n" "Connection: Upgrade\r\n" f"Sec-WebSocket-Key: {key}\r\n" "Sec-WebSocket-Version: 13\r\n\r\n" ).encode("ascii") self._socket.sendall(request) response = bytearray() marker = b"\r\n\r\n" while marker not in response: chunk = self._socket.recv(4096) if not chunk: raise AppServerError("app-server closed during WebSocket handshake") response.extend(chunk) if len(response) > 64 * 1024: raise AppServerError("app-server WebSocket handshake exceeds 64 KiB") raw_headers, trailing = bytes(response).split(marker, 1) lines = raw_headers.decode("iso-8859-1").split("\r\n") status = lines[0].split(" ", 2) if lines else [] if len(status) < 2 or status[0] != "HTTP/1.1" or status[1] != "101": raise AppServerError( f"app-server rejected WebSocket upgrade: {lines[0] if lines else ''}" ) headers: dict[str, list[str]] = {} for line in lines[1:]: if ":" not in line: raise AppServerError("app-server returned a malformed WebSocket header") name, value = line.split(":", 1) name = name.strip().lower() if not name: raise AppServerError("app-server returned an empty WebSocket header name") headers.setdefault(name, []).append(value.strip()) expected = base64.b64encode( hashlib.sha1( # noqa: S324 - mandated by RFC 6455, not used for security (key + "258EAFA5-E914-47DA-95CA-C5AB0DC85B11").encode("ascii") ).digest() ).decode("ascii") if headers.get("sec-websocket-accept") != [expected]: raise AppServerError("app-server returned an invalid WebSocket accept digest") upgrade_tokens = { token.strip().casefold() for value in headers.get("upgrade", []) for token in value.split(",") if token.strip() } connection_tokens = { token.strip().casefold() for value in headers.get("connection", []) for token in value.split(",") if token.strip() } if "websocket" not in upgrade_tokens: raise AppServerError("app-server WebSocket upgrade header is missing") if "upgrade" not in connection_tokens: raise AppServerError("app-server WebSocket connection upgrade token is missing") if "sec-websocket-extensions" in headers or "sec-websocket-protocol" in headers: raise AppServerError("app-server negotiated an unrequested WebSocket feature") self._buffer.extend(trailing) def _recv_exact(self, size: int) -> bytes: while len(self._buffer) < size: chunk = self._socket.recv(max(4096, size - len(self._buffer))) if not chunk: raise AppServerError("app-server WebSocket transport closed") self._buffer.extend(chunk) value = bytes(self._buffer[:size]) del self._buffer[:size] return value @staticmethod def _frame(opcode: int, payload: bytes) -> bytes: if len(payload) > MAX_WEBSOCKET_MESSAGE_BYTES: raise AppServerError("app-server WebSocket message exceeds 64 MiB") if opcode >= 0x8 and len(payload) > 125: raise AppServerError("app-server WebSocket control frame exceeds 125 bytes") if opcode == 0x8 and len(payload) == 1: raise AppServerError("app-server WebSocket close payload is malformed") first = 0x80 | opcode length = len(payload) if length < 126: header = bytes((first, 0x80 | length)) elif length <= 0xFFFF: header = bytes((first, 0x80 | 126)) + struct.pack("!H", length) else: header = bytes((first, 0x80 | 127)) + struct.pack("!Q", length) mask = os.urandom(4) masked = bytes(value ^ mask[index & 3] for index, value in enumerate(payload)) return header + mask + masked def send_text(self, value: str) -> None: encoded = value.encode("utf-8") with self._send_lock: if self._closed: raise AppServerError("app-server WebSocket transport is closed") try: self._socket.sendall(self._frame(0x1, encoded)) except OSError as exc: raise AppServerError(f"app-server WebSocket write failed: {exc}") from exc def _send_control(self, opcode: int, payload: bytes = b"") -> None: with self._send_lock: if self._closed: return self._socket.sendall(self._frame(opcode, payload)) def _close_socket(self) -> None: if self._closed: return self._closed = True with contextlib.suppress(OSError): self._socket.shutdown(socket.SHUT_RDWR) with contextlib.suppress(OSError): self._socket.close() def _protocol_failure(self, message: str, *, close_code: int = 1002) -> None: with contextlib.suppress(OSError, AppServerError): self._send_control(0x8, struct.pack("!H", close_code)) self._close_socket() raise AppServerError(message) @staticmethod def _valid_close_code(code: int) -> bool: return ( code in { 1000, 1001, 1002, 1003, 1007, 1008, 1009, 1010, 1011, 1012, 1013, 1014, } or 3000 <= code <= 4999 ) def receive_text(self) -> str: fragments = bytearray() fragmented = False while True: first, second = self._recv_exact(2) final = bool(first & 0x80) if first & 0x70: self._protocol_failure("app-server WebSocket frame has unsupported RSV bits") opcode = first & 0x0F masked = bool(second & 0x80) if masked: self._protocol_failure("app-server sent a masked WebSocket server frame") length = second & 0x7F control = opcode >= 0x8 if control and (not final or length > 125): self._protocol_failure("app-server sent a malformed WebSocket control frame") if length == 126: length = struct.unpack("!H", self._recv_exact(2))[0] if length < 126: self._protocol_failure("app-server used a non-minimal WebSocket frame length") elif length == 127: encoded_length = self._recv_exact(8) if encoded_length[0] & 0x80: self._protocol_failure("app-server WebSocket frame length has its high bit set") length = struct.unpack("!Q", encoded_length)[0] if length <= 0xFFFF: self._protocol_failure("app-server used a non-minimal WebSocket frame length") if length > MAX_WEBSOCKET_MESSAGE_BYTES: self._protocol_failure( "app-server WebSocket message exceeds 64 MiB", close_code=1009 ) payload = self._recv_exact(length) if opcode == 0x8: if len(payload) == 1: self._protocol_failure("app-server sent a malformed WebSocket close payload") if payload: close_code = struct.unpack("!H", payload[:2])[0] if not self._valid_close_code(close_code): self._protocol_failure("app-server sent an invalid WebSocket close code") try: payload[2:].decode("utf-8") except UnicodeDecodeError: self._protocol_failure( "app-server sent a non-UTF-8 WebSocket close reason", close_code=1007, ) with contextlib.suppress(OSError, AppServerError): self._send_control(0x8, payload) self._close_socket() raise AppServerError("app-server WebSocket transport closed") if opcode == 0x9: self._send_control(0xA, payload) continue if opcode == 0xA: continue if opcode == 0x1: if fragmented: self._protocol_failure("app-server started a nested WebSocket message") fragments.extend(payload) fragmented = not final elif opcode == 0x0 and fragmented: fragments.extend(payload) fragmented = not final elif opcode == 0x2: self._protocol_failure( "app-server emitted unsupported WebSocket binary data", close_code=1003 ) else: self._protocol_failure(f"unsupported app-server WebSocket opcode: {opcode}") if len(fragments) > MAX_WEBSOCKET_MESSAGE_BYTES: self._protocol_failure( "app-server WebSocket message exceeds 64 MiB", close_code=1009 ) if not fragmented: try: return fragments.decode("utf-8") except UnicodeDecodeError as exc: with contextlib.suppress(OSError, AppServerError): self._send_control(0x8, struct.pack("!H", 1007)) self._close_socket() raise AppServerError("app-server emitted non-UTF-8 WebSocket text") from exc def close(self) -> None: if self._closed: return with contextlib.suppress(OSError, AppServerError): self._send_control(0x8, struct.pack("!H", 1000)) self._close_socket() class AppServerClient: """One protocol client connected to an isolated Unix app-server host.""" def __init__( self, *, socket_path: Path, cwd: Path, env: Mapping[str, str], events_path: Path, stderr_path: Path, approval_policy: str, command: Sequence[str] | None = None, on_message: Callable[[dict[str, Any]], None] | None = None, ) -> None: self.command = list(command) if command is not None else None self.socket_path = socket_path.expanduser().resolve() self.cwd = cwd self.env = dict(env) self.events_path = events_path self.stderr_path = stderr_path self.approval_policy = approval_policy self.on_message = on_message self.process: subprocess.Popen[bytes] | None = None self.transport: UnixWebSocket | None = None self.initialize_result: dict[str, Any] | None = None self._stderr: Any = None self._reader: threading.Thread | None = None self._write_lock = threading.Lock() self._condition = threading.Condition() self._responses: dict[int, dict[str, Any]] = {} self._server_requests: dict[str, dict[str, Any]] = {} self._dynamic_mcp_tools: dict[str, tuple[str, str]] = {} self._dynamic_call_threads: set[threading.Thread] = set() self._dynamic_call_lock = threading.Lock() self._next_id = 1 self._closed = False self._reader_error: str | None = None @property def pid(self) -> int | None: return self.process.pid if self.process is not None else None @property def alive(self) -> bool: process_alive = self.process is None or self.process.poll() is None return bool( process_alive and self.transport is not None and not self.transport.closed and not self._closed ) @property def pending_server_requests(self) -> list[dict[str, Any]]: with self._condition: return [dict(value) for value in self._server_requests.values()] def pending_server_requests_for_thread(self, thread_id: str) -> list[dict[str, Any]]: """Return only pending requests owned by one thread on a shared host.""" return [ request for request in self.pending_server_requests if app_server_message_thread_id(request) == thread_id ] def install_switchyard_mcp_bridge( self, *, timeout: float = APP_SERVER_LIFECYCLE_TIMEOUT_SECONDS, ) -> list[dict[str, Any]]: """Expose configured MCP tools as flat app-server dynamic functions. Codex 0.149 emits MCP tools as Responses namespace entries. Switchyard 0.2.0 accepts ordinary function tools but drops namespace entries while translating requests for OpenAI-compatible third-party providers. The app-server dynamic-tool protocol gives those providers an equivalent flat function surface without changing Codex or bypassing its MCP connection, allowlist, approval, and result handling. """ cursor: str | None = None seen_cursors: set[str] = set() rows: list[Mapping[str, Any]] = [] while True: params: dict[str, Any] = { "detail": "toolsAndAuthOnly", "limit": 100, } if cursor is not None: params["cursor"] = cursor response = self.request("mcpServerStatus/list", params, timeout=timeout) data = response.get("data") if isinstance(response, Mapping) else None if not isinstance(data, list) or not all(isinstance(row, Mapping) for row in data): raise AppServerError("mcpServerStatus/list returned invalid server data") rows.extend(data) next_cursor = response.get("nextCursor") if next_cursor is None: break if not isinstance(next_cursor, str) or next_cursor in seen_cursors: raise AppServerError("mcpServerStatus/list returned an invalid cursor sequence") seen_cursors.add(next_cursor) cursor = next_cursor routes: dict[str, tuple[str, str]] = {} specs: list[dict[str, Any]] = [] for row in rows: server = row.get("name") tools = row.get("tools") if not isinstance(server, str) or not isinstance(tools, Mapping): raise AppServerError("mcpServerStatus/list returned an invalid server row") for listed_name, raw_tool in sorted(tools.items(), key=lambda item: str(item[0])): if not isinstance(listed_name, str) or not isinstance(raw_tool, Mapping): raise AppServerError("mcpServerStatus/list returned an invalid tool row") tool = raw_tool.get("name", listed_name) if not isinstance(tool, str) or not tool: raise AppServerError("mcpServerStatus/list returned an invalid tool name") name = _flat_mcp_dynamic_tool_name(server, tool) if name in routes and routes[name] != (server, tool): name = _flat_mcp_dynamic_tool_name(server, tool, salt=listed_name) if name in routes: raise AppServerError(f"flat MCP dynamic tool name collision: {name}") input_schema = raw_tool.get("inputSchema", {"type": "object"}) if not isinstance(input_schema, Mapping): raise AppServerError(f"MCP tool {server}.{tool} has an invalid input schema") description = raw_tool.get("description") specs.append( { "type": "function", "name": name, "description": ( str(description) if isinstance(description, str) else f"Call {tool} on the {server} MCP server." ), "inputSchema": dict(input_schema), } ) routes[name] = (server, tool) self._dynamic_mcp_tools = routes return specs def _run_dynamic_mcp_tool_call( self, request_id: int | str, message: Mapping[str, Any], ) -> None: """Route one flat dynamic function through Codex's own MCP API.""" try: params = message.get("params") if not isinstance(params, Mapping): raise AppServerError("dynamic tool request params are invalid") thread_id = params.get("threadId") namespace = params.get("namespace") flat_name = params.get("tool") if not isinstance(thread_id, str) or not thread_id: raise AppServerError("dynamic tool request has no thread identity") if namespace is not None: raise AppServerError("flat MCP bridge received a namespaced dynamic tool call") if not isinstance(flat_name, str) or flat_name not in self._dynamic_mcp_tools: raise AppServerError(f"unknown flat MCP dynamic tool: {flat_name!r}") server, tool = self._dynamic_mcp_tools[flat_name] call_params: dict[str, Any] = { "threadId": thread_id, "server": server, "tool": tool, } if "arguments" in params: call_params["arguments"] = params["arguments"] result = self.request( "mcpServer/tool/call", call_params, timeout=APP_SERVER_DYNAMIC_TOOL_TIMEOUT_SECONDS, ) if not isinstance(result, Mapping): raise AppServerError("mcpServer/tool/call returned a non-object result") self.respond( request_id, { "contentItems": _dynamic_tool_content_items(result), "success": result.get("isError") is not True, }, ) except Exception as exc: with contextlib.suppress(Exception): self.respond( request_id, { "contentItems": [ { "type": "inputText", "text": f"MCP bridge error: {type(exc).__name__}: {exc}", } ], "success": False, }, ) finally: with self._dynamic_call_lock: self._dynamic_call_threads.discard(threading.current_thread()) def _dispatch_dynamic_mcp_tool_call( self, request_id: int | str, message: Mapping[str, Any], ) -> None: worker = threading.Thread( target=self._run_dynamic_mcp_tool_call, args=(request_id, dict(message)), name=f"codex-app-server-dynamic-tool-{request_id}", daemon=True, ) with self._dynamic_call_lock: self._dynamic_call_threads.add(worker) worker.start() @staticmethod def _server_request_key(request_id: int | str) -> str: prefix = "integer" if isinstance(request_id, int) else "string" return f"{prefix}:{request_id}" def start( self, *, timeout: float = 30.0, on_started: Callable[[int], None] | None = None, ) -> dict[str, Any]: if self.transport is not None or self.process is not None: raise AppServerError("app-server client has already been started") self.stderr_path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) self.socket_path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) if self.command is not None: if self.socket_path.exists() or self.socket_path.is_symlink(): raise AppServerError( f"refusing to replace an existing app-server socket: {self.socket_path}" ) self._stderr = self.stderr_path.open("ab", buffering=0) try: self.process = subprocess.Popen( self.command, stdin=subprocess.DEVNULL, stdout=self._stderr, stderr=self._stderr, cwd=self.cwd, env=self.env, start_new_session=True, close_fds=True, ) except BaseException: self._stderr.close() self._stderr = None raise try: # Publish the isolated process-group identity before performing any # potentially slow protocol or MCP initialization. This closes the # bootstrap interval in which an operator stop could otherwise # lose ownership of the newly spawned app-server descendants. if on_started is not None and self.process is not None: on_started(self.process.pid) deadline = time.monotonic() + timeout last_error = "socket not ready" while self.transport is None: if self.process is not None and self.process.poll() is not None: raise AppServerError( f"app-server exited before accepting connections: {self.process.returncode}" ) remaining = deadline - time.monotonic() if remaining <= 0: raise AppServerError( f"app-server Unix socket did not become ready: {last_error}" ) try: self.transport = UnixWebSocket( self.socket_path, timeout=min(1.0, remaining), ) except (AppServerError, OSError) as exc: last_error = f"{type(exc).__name__}: {exc}" time.sleep(min(0.05, remaining)) self._reader = threading.Thread( target=self._read_loop, name=f"codex-app-server-{self.pid or self.socket_path.name}", daemon=True, ) self._reader.start() result = self.request( "initialize", { "clientInfo": { "name": "codex-mmo", "title": "Codex Multi-Model Orchestrator", "version": package_version(), }, "capabilities": {"experimentalApi": True}, }, timeout=timeout, ) except BaseException: self.close() raise if not isinstance(result, dict): self.close() raise AppServerError("app-server initialize returned a non-object result") self.initialize_result = result self.notify("initialized", {}) return result def _record(self, direction: str, message: Mapping[str, Any]) -> None: append_jsonl( self.events_path, { "recorded_at": utc_now(), "direction": direction, "message": dict(message), }, ) def _read_loop(self) -> None: transport = self.transport if transport is None: raise AppServerError("app-server WebSocket transport was not initialized") try: while True: line = transport.receive_text() try: value = strict_json_loads(line) except (json.JSONDecodeError, ValueError) as exc: raise AppServerError(f"invalid app-server JSON: {exc}") from exc if not isinstance(value, dict): raise AppServerError("app-server emitted a non-object protocol message") if "jsonrpc" in value: raise AppServerError("app-server emitted an unexpected JSON-RPC version field") self._record("received", value) message_id = value.get("id") if message_id is not None and ("result" in value or "error" in value): if not isinstance(message_id, int) or isinstance(message_id, bool): raise AppServerError("app-server response id is not an integer") if ("result" in value) == ("error" in value): raise AppServerError( "app-server response must contain exactly one of result or error" ) if "error" in value: error = value["error"] if not isinstance(error, Mapping): raise AppServerError("app-server response error is not an object") code = error.get("code") if not isinstance(code, int) or isinstance(code, bool): raise AppServerError("app-server response error code is not an integer") if not isinstance(error.get("message"), str): raise AppServerError( "app-server response error message is not a string" ) with self._condition: self._responses[message_id] = value self._condition.notify_all() elif message_id is not None and isinstance(value.get("method"), str): if not ( isinstance(message_id, str) or (isinstance(message_id, int) and not isinstance(message_id, bool)) ): raise AppServerError("app-server request id must be an integer or string") key = self._server_request_key(message_id) method = str(value["method"]) if method == "currentTime/read": self.respond(message_id, {"currentTimeAt": int(time.time())}) elif method == "item/tool/call" and self._dynamic_mcp_tools: # A bridge call must run off the reader thread: its # mcpServer/tool/call request is answered by this same # reader, so handling it inline would deadlock. self._dispatch_dynamic_mcp_tool_call(message_id, value) if self.on_message is not None: with contextlib.suppress(Exception): self.on_message(value) elif method in APPROVAL_REQUEST_METHODS and self.approval_policy == "never": self._deny_approval(message_id, method) elif method in PENDING_SERVER_REQUEST_METHODS: with self._condition: self._server_requests[key] = value self._condition.notify_all() if self.on_message is not None: with contextlib.suppress(Exception): self.on_message(value) elif method in UNSUPPORTED_SERVER_REQUEST_METHODS: self.respond_error( message_id, -32601, f"{method} is deliberately not provided by Codex MMO", ) else: self.respond_error( message_id, -32601, f"unsupported app-server request method: {method}", ) elif value.get("method") == "serverRequest/resolved": params = value.get("params") request_id = params.get("requestId") if isinstance(params, Mapping) else None if isinstance(request_id, str) or ( isinstance(request_id, int) and not isinstance(request_id, bool) ): with self._condition: self._server_requests.pop(self._server_request_key(request_id), None) self._condition.notify_all() if self.on_message is not None: with contextlib.suppress(Exception): self.on_message(value) elif self.on_message is not None: with contextlib.suppress(Exception): self.on_message(value) except Exception as exc: self._reader_error = f"{type(exc).__name__}: {exc}" finally: with self._condition: self._closed = True self._condition.notify_all() def _deny_approval(self, request_id: Any, method: str) -> None: if method in { "item/commandExecution/requestApproval", "item/fileChange/requestApproval", }: result: dict[str, Any] = {"decision": "decline"} elif method == "item/permissions/requestApproval": self.respond_error(request_id, -32001, "permission escalation denied by MMO policy") return else: result = { "decision": { "denied": { "rejection": "denied by Codex MMO approval_policy=never", } } } self.respond(request_id, result) def _send(self, message: Mapping[str, Any]) -> None: transport = self.transport if transport is None or transport.closed or self._closed: raise AppServerError("app-server is not running") encoded = json.dumps( dict(message), ensure_ascii=False, separators=(",", ":"), allow_nan=False ) with self._write_lock: # Keep the durable trace in the exact order bytes are serialized # onto the app-server transport. Recording before this lock # allowed concurrent control/approval writers to publish an order # different from the wire order. self._record("sent", message) try: transport.send_text(encoded) except (AppServerError, OSError) as exc: raise AppServerError(f"app-server write failed: {exc}") from exc def request(self, method: str, params: Mapping[str, Any], *, timeout: float = 30.0) -> Any: deadline = time.monotonic() + timeout for attempt in range(len(APP_SERVER_OVERLOAD_RETRY_DELAYS_SECONDS) + 1): with self._condition: request_id = self._next_id self._next_id += 1 self._send({"id": request_id, "method": method, "params": dict(params)}) with self._condition: while request_id not in self._responses: if self._closed: detail = self._reader_error or "app-server transport closed" raise AppServerError(detail) remaining = deadline - time.monotonic() if remaining <= 0: self._responses.pop(request_id, None) raise AppServerError(f"app-server request timed out: {method}") self._condition.wait(timeout=remaining) response = self._responses.pop(request_id) if "error" not in response: return response.get("result") error = response["error"] code = int(error["code"]) message = str(error["message"]) overload = code == -32001 and message == "Server overloaded; retry later." if overload and attempt < len(APP_SERVER_OVERLOAD_RETRY_DELAYS_SECONDS): base_delay = APP_SERVER_OVERLOAD_RETRY_DELAYS_SECONDS[attempt] delay = base_delay * random.uniform(0.5, 1.0) if delay < deadline - time.monotonic(): time.sleep(delay) continue raise AppServerError(f"app-server {method} failed ({code}): {message}") raise AssertionError("app-server retry loop exhausted without returning") def notify(self, method: str, params: Mapping[str, Any]) -> None: self._send({"method": method, "params": dict(params)}) def respond(self, request_id: Any, result: Mapping[str, Any]) -> None: self._send({"id": request_id, "result": dict(result)}) if isinstance(request_id, str) or ( isinstance(request_id, int) and not isinstance(request_id, bool) ): with self._condition: self._server_requests.pop(self._server_request_key(request_id), None) def respond_error(self, request_id: Any, code: int, message: str) -> None: self._send( { "id": request_id, "error": {"code": int(code), "message": message}, } ) if isinstance(request_id, str) or ( isinstance(request_id, int) and not isinstance(request_id, bool) ): with self._condition: self._server_requests.pop(self._server_request_key(request_id), None) def close(self) -> None: """Detach this client without changing the app-server host or its goal.""" if self.transport is not None: self.transport.close() if self._reader is not None: self._reader.join(timeout=2.0) if self._stderr is not None: with contextlib.suppress(Exception): self._stderr.close() with self._condition: self._closed = True self._condition.notify_all() def stop_host(self, *, grace_seconds: float = 8.0) -> None: """Detach and retire the isolated app-server process this client started.""" process = self.process self.close() if process is not None: terminate_process_group(process.pid, grace_seconds=grace_seconds) with contextlib.suppress(Exception): process.wait(timeout=2.0) if self.socket_path.is_socket(): with contextlib.suppress(OSError): self.socket_path.unlink() def last_agent_message(turn: Any) -> str: if not isinstance(turn, Mapping): return "" messages = [ str(item.get("text", "")) for item in turn.get("items", []) if isinstance(item, Mapping) and item.get("type") == "agentMessage" ] return next((value for value in reversed(messages) if value.strip()), "") def bounded_goal_objective(value: str) -> str: """Return one deterministic app-server goal objective within its hard limit. The complete delegated or root prompt remains the first turn input. Long objectives retain a recognizable prefix plus a digest that lets operators correlate the bounded goal with those full instructions. """ objective = value.strip() if not objective: raise ValueError("goal objective must be a non-empty string") if len(objective) <= APP_SERVER_GOAL_OBJECTIVE_MAX_CHARS: return objective digest = hashlib.sha256(objective.encode("utf-8")).hexdigest() suffix = f"\n\n[Full instructions remain in the initial turn; sha256={digest}]" prefix_chars = APP_SERVER_GOAL_OBJECTIVE_MAX_CHARS - len(suffix) return objective[:prefix_chars].rstrip() + suffix def completed_turn_presentable_text(events_path: Path, turn_id: str) -> str: """Recover the last completed user-visible item for one exact root turn.""" if not events_path.is_file() or events_path.is_symlink(): return "" result = "" with events_path.open("r", encoding="utf-8", errors="replace") as handle: for raw in handle: try: record = strict_json_loads(raw) except (json.JSONDecodeError, ValueError): continue if not isinstance(record, Mapping) or record.get("direction") != "received": continue message = record.get("message") if not isinstance(message, Mapping) or message.get("method") != "item/completed": continue params = message.get("params") if not isinstance(params, Mapping) or params.get("turnId") != turn_id: continue item = params.get("item") if not isinstance(item, Mapping) or item.get("type") not in { "agentMessage", "plan", }: continue text = item.get("text") if isinstance(text, str) and text.strip(): result = text return result def turn_input(prompt: str, attachments: Sequence[str]) -> list[dict[str, Any]]: items: list[dict[str, Any]] = [{"type": "text", "text": prompt}] items.extend( {"type": "localImage", "path": path} for path in attachments if Path(path).suffix.lower() in {".png", ".jpg", ".jpeg", ".webp", ".gif"} ) return items def resumed_turns( thread: Mapping[str, Any], prior_turn_id: Any, *, turn_start_pending: bool = False, ) -> tuple[dict[str, Any] | None, str | None]: """Return an exact recovered terminal turn and any remaining active turn.""" turns = [turn for turn in thread.get("turns", []) if isinstance(turn, dict)] prior_index = next( ( index for index in range(len(turns) - 1, -1, -1) if isinstance(prior_turn_id, str) and turns[index].get("id") == prior_turn_id ), None, ) if turn_start_pending: appended = turns[(prior_index + 1) if isinstance(prior_index, int) else 0 :] candidate = appended[-1] if appended else None else: candidate = turns[prior_index] if isinstance(prior_index, int) else None terminal = ( candidate if isinstance(candidate, dict) and candidate.get("status") != "inProgress" else None ) active = next( (turn for turn in reversed(turns) if turn.get("status") == "inProgress"), None, ) active_id = active.get("id") if isinstance(active, dict) else None return terminal, str(active_id) if isinstance(active_id, str) else None class PersistentThreadHost: """Shared durable thread/turn state for root and delegated app-server hosts.""" def __init__( self, *, state: dict[str, Any] | None = None, state_lock: threading.RLock | None = None, on_state_change: Callable[[dict[str, Any], Mapping[str, Any]], None] | None = None, ) -> None: self.state = state if state is not None else {} self.state_lock = state_lock or threading.RLock() self.on_state_change = on_state_change self.state.setdefault("client", None) self.state.setdefault("thread_id", None) self.state.setdefault("active_turn_id", None) self.state.setdefault("last_turn_id", None) self.state.setdefault("turn_start_pending", False) self.state.setdefault("completed_turn", None) self.state.setdefault("turn_failure", None) self.state.setdefault("thread_status", None) self.state.setdefault("goal", None) self.state.setdefault("token_usage", None) self.state.setdefault("last_item", None) self.state.setdefault("turn_event", threading.Event()) @property def client(self) -> AppServerClient: with self.state_lock: client = self.state.get("client") if not isinstance(client, AppServerClient): raise AppServerError("app-server host has no connected client") return client @property def thread_id(self) -> str | None: with self.state_lock: value = self.state.get("thread_id") return str(value) if isinstance(value, str) else None @property def active_turn_id(self) -> str | None: with self.state_lock: value = self.state.get("active_turn_id") return str(value) if isinstance(value, str) else None @property def last_turn_id(self) -> str | None: with self.state_lock: value = self.state.get("last_turn_id") return str(value) if isinstance(value, str) else None @property def completed_turn(self) -> dict[str, Any] | None: with self.state_lock: value = self.state.get("completed_turn") return dict(value) if isinstance(value, Mapping) else None @property def turn_event(self) -> threading.Event: value = self.state["turn_event"] if not isinstance(value, threading.Event): raise AppServerError("app-server host turn event is invalid") return value def attach_client(self, client: AppServerClient) -> None: with self.state_lock: self.state["client"] = client def detach_client(self, client: AppServerClient | None = None) -> None: with self.state_lock: if client is None or self.state.get("client") is client: self.state["client"] = None def on_message(self, message: dict[str, Any]) -> None: method = message.get("method") raw_params = message.get("params") params: Mapping[str, Any] = raw_params if isinstance(raw_params, Mapping) else {} changes: dict[str, Any] = {} with self.state_lock: expected_thread_id = self.state.get("thread_id") message_thread_id = app_server_message_thread_id(message) if method == "thread/started": thread = params.get("thread") if isinstance(thread, Mapping) and isinstance(thread.get("id"), str): changes["thread_started"] = dict(thread) elif ( isinstance(expected_thread_id, str) and isinstance(message_thread_id, str) and message_thread_id != expected_thread_id ): return elif method == "turn/started": turn = params.get("turn") if isinstance(turn, Mapping) and isinstance(turn.get("id"), str): self.state["active_turn_id"] = turn["id"] self.state["last_turn_id"] = turn["id"] self.state["turn_start_pending"] = False self.state["turn_failure"] = None changes.update( active_turn_id=turn["id"], last_turn_id=turn["id"], turn_start_pending=False, turn_failure=None, ) elif method == "turn/completed": turn = params.get("turn") if isinstance(turn, Mapping): current_turn_id = self.state.get("active_turn_id") completed_turn_id = turn.get("id") if not isinstance(current_turn_id, str) or completed_turn_id == current_turn_id: self.state["completed_turn"] = dict(turn) self.state["active_turn_id"] = None if isinstance(completed_turn_id, str): self.state["last_turn_id"] = completed_turn_id self.state["turn_start_pending"] = False failure = normalize_turn_failure(turn) self.state["turn_failure"] = failure changes.update( active_turn_id=None, last_turn_id=( completed_turn_id if isinstance(completed_turn_id, str) else self.state.get("last_turn_id") ), turn_start_pending=False, last_turn_status=turn.get("status"), turn_failure=failure, ) self.turn_event.set() elif method == "thread/status/changed": status = params.get("status") self.state["thread_status"] = status changes["thread_status"] = status elif method == "thread/tokenUsage/updated": usage = params.get("tokenUsage") self.state["token_usage"] = usage changes["token_usage"] = usage elif method == "thread/goal/updated": goal = params.get("goal") if isinstance(goal, Mapping): self.state["goal"] = dict(goal) changes["goal"] = dict(goal) elif method == "thread/goal/cleared": self.state["goal"] = None changes["goal"] = None elif method in {"item/started", "item/completed"}: item = params.get("item") if isinstance(item, Mapping): self.state["last_item"] = dict(item) changes["last_item"] = dict(item) changes["last_item_event"] = method elif method in { "turn/diff/updated", "turn/plan/updated", "mcpServer/startupStatus/updated", "model/rerouted", "model/verification", "warning", "error", "account/rateLimits/updated", }: changes["last_observability_event"] = method changes["last_observability_params"] = dict(params) elif ( message.get("id") is not None and method in { "item/tool/requestUserInput", "mcpServer/elicitation/request", } | APPROVAL_REQUEST_METHODS ): client = self.state.get("client") changes["pending_request_count"] = len( client.pending_server_requests_for_thread(str(expected_thread_id)) if isinstance(client, AppServerClient) and isinstance(expected_thread_id, str) else [] ) elif method == "serverRequest/resolved": client = self.state.get("client") changes["pending_request_count"] = len( client.pending_server_requests_for_thread(str(expected_thread_id)) if isinstance(client, AppServerClient) and isinstance(expected_thread_id, str) else [] ) if self.on_state_change is not None: self.on_state_change(changes, message) def adopt_thread(self, thread: Mapping[str, Any]) -> None: """Switch the tracked logical root after its durable owner adopts a successor.""" thread_id = thread.get("id") if not isinstance(thread_id, str) or not thread_id: raise AppServerError("root successor has no thread identity") terminal, active_turn_id = resumed_turns(thread, None) turns = [value for value in thread.get("turns", []) if isinstance(value, Mapping)] last_turn = turns[-1] if turns else None with self.state_lock: self.state["thread_id"] = thread_id self.state["thread_status"] = thread.get("status") goal = thread.get("goal") self.state["goal"] = dict(goal) if isinstance(goal, Mapping) else None self.state["turn_start_pending"] = False self.state["completed_turn"] = terminal self.state["turn_failure"] = ( normalize_turn_failure(terminal) if isinstance(terminal, Mapping) else None ) self.state["active_turn_id"] = active_turn_id self.state["last_turn_id"] = ( terminal.get("id") if isinstance(terminal, Mapping) and isinstance(terminal.get("id"), str) else active_turn_id or ( last_turn.get("id") if isinstance(last_turn, Mapping) and isinstance(last_turn.get("id"), str) else None ) ) if terminal is not None: self.turn_event.set() else: self.turn_event.clear() def open_thread( self, mode: str, params: Mapping[str, Any], *, timeout: float, prior_turn_id: Any = None, expected_thread_id: str | None = None, interrupt_stale: bool = False, ) -> dict[str, Any]: method = { "start": "thread/start", "resume": "thread/resume", "fork": "thread/fork", }.get(mode) if method is None: raise ValueError(f"unsupported app-server thread mode: {mode}") response = self.client.request(method, params, timeout=timeout) thread = response.get("thread") if isinstance(response, Mapping) else None if not isinstance(thread, Mapping) or not isinstance(thread.get("id"), str): raise AppServerError("thread start/resume returned no thread identifier") thread_id = str(thread["id"]) if expected_thread_id is not None and thread_id != expected_thread_id: raise AppServerError("thread/resume returned the wrong thread identity") with self.state_lock: turn_start_pending = bool(self.state.get("turn_start_pending")) terminal, active_turn_id = resumed_turns( thread, prior_turn_id, turn_start_pending=turn_start_pending, ) with self.state_lock: self.state["thread_id"] = thread_id self.state["thread_status"] = thread.get("status") goal = thread.get("goal") self.state["goal"] = dict(goal) if isinstance(goal, Mapping) else None if terminal is not None: self.state["completed_turn"] = terminal self.state["turn_failure"] = normalize_turn_failure(terminal) self.state["active_turn_id"] = None self.state["last_turn_id"] = terminal.get("id") self.turn_event.set() else: self.state["turn_failure"] = None self.state["active_turn_id"] = active_turn_id if active_turn_id is not None: self.state["last_turn_id"] = active_turn_id self.state["turn_start_pending"] = False if terminal is None and interrupt_stale and active_turn_id is not None: interrupt_error: AppServerError | None = None try: self.client.request( "turn/interrupt", {"threadId": thread_id, "turnId": active_turn_id}, ) except AppServerError as exc: # The turn may have terminalized after thread/resume but before # the interrupt was admitted. Its completion notification, not # the racing request error, is authoritative. interrupt_error = exc if not self.turn_event.wait( timeout=min(timeout, APP_SERVER_INITIALIZE_TIMEOUT_SECONDS) ): if interrupt_error is not None: raise interrupt_error raise AppServerError("persisted active turn did not settle after interruption") with self.state_lock: settled = self.state.get("completed_turn") if not isinstance(settled, Mapping) or settled.get("id") != active_turn_id: raise AppServerError("persisted active turn settled with the wrong identity") if settled.get("status") != "interrupted": return dict(thread) with self.state_lock: self.state["active_turn_id"] = None self.state["completed_turn"] = None self.state["turn_failure"] = None self.turn_event.clear() elif ( interrupt_stale and isinstance(terminal, Mapping) and terminal.get("status") == "interrupted" ): with self.state_lock: self.state["active_turn_id"] = None self.state["completed_turn"] = None self.state["turn_failure"] = None self.turn_event.clear() return dict(thread) def set_goal( self, *, objective: str | None = None, status: str | None = None, token_budget: int | None = None, timeout: float = 60.0, ) -> dict[str, Any]: thread_id = self.thread_id if thread_id is None: raise AppServerError("app-server host has no persistent thread") params: dict[str, Any] = {"threadId": thread_id} if objective is not None: params["objective"] = objective if status is not None: params["status"] = status if token_budget is not None: params["tokenBudget"] = token_budget response = self.client.request("thread/goal/set", params, timeout=timeout) goal = response.get("goal") if isinstance(response, Mapping) else None if not isinstance(goal, Mapping): raise AppServerError("thread/goal/set returned no goal") with self.state_lock: self.state["goal"] = dict(goal) return dict(goal) def read_thread(self, *, include_turns: bool = False, timeout: float = 60.0) -> dict[str, Any]: thread_id = self.thread_id if thread_id is None: raise AppServerError("app-server host has no persistent thread") response = self.client.request( "thread/read", {"threadId": thread_id, "includeTurns": include_turns}, timeout=timeout, ) thread = response.get("thread") if isinstance(response, Mapping) else None if not isinstance(thread, Mapping) or thread.get("id") != thread_id: raise AppServerError("thread/read returned the wrong thread") return dict(thread) def complete_history(self, *, timeout: float = 60.0) -> list[dict[str, Any]]: """Read every persisted turn through the authoritative paginated API.""" thread_id = self.thread_id if thread_id is None: raise AppServerError("app-server host has no persistent thread") cursor: str | None = None turns: list[dict[str, Any]] = [] seen_cursors: set[str] = set() while True: params: dict[str, Any] = { "threadId": thread_id, "limit": 100, "sortDirection": "asc", "itemsView": "full", } if cursor is not None: params["cursor"] = cursor response = self.client.request("thread/turns/list", params, timeout=timeout) data = response.get("data") if isinstance(response, Mapping) else None if not isinstance(data, list) or not all(isinstance(item, Mapping) for item in data): raise AppServerError("thread/turns/list returned invalid turn data") turns.extend(dict(item) for item in data) next_cursor = response.get("nextCursor") if next_cursor is None: return turns if not isinstance(next_cursor, str) or next_cursor in seen_cursors: raise AppServerError("thread/turns/list returned an invalid cursor sequence") seen_cursors.add(next_cursor) cursor = next_cursor def start_turn( self, input_items: Sequence[Mapping[str, Any]], *, effort: Any, output_schema: Mapping[str, Any] | None = None, timeout: float = 60.0, ) -> str: thread_id = self.thread_id if thread_id is None: raise AppServerError("app-server host has no persistent thread") params: dict[str, Any] = { "threadId": thread_id, "input": [dict(item) for item in input_items], "effort": effort, } if output_schema is not None: params["outputSchema"] = dict(output_schema) with self.state_lock: self.state["active_turn_id"] = None self.state["turn_start_pending"] = True self.state["completed_turn"] = None self.state["turn_failure"] = None self.turn_event.clear() if self.on_state_change is not None: self.on_state_change( {"active_turn_id": None, "turn_start_pending": True}, {"method": "mmo/turn/reset", "params": {"threadId": thread_id}}, ) response = self.client.request("turn/start", params, timeout=timeout) turn = response.get("turn") if isinstance(response, Mapping) else None if not isinstance(turn, Mapping) or not isinstance(turn.get("id"), str): raise AppServerError("turn/start returned no turn identifier") turn_id = str(turn["id"]) with self.state_lock: completed_turn = self.state.get("completed_turn") self.state["last_turn_id"] = turn_id self.state["turn_start_pending"] = False if isinstance(completed_turn, Mapping) and completed_turn.get("id") == turn_id: self.state["active_turn_id"] = None changes = { "active_turn_id": None, "last_turn_id": turn_id, "turn_start_pending": False, } else: self.state["active_turn_id"] = turn_id changes = { "active_turn_id": turn_id, "last_turn_id": turn_id, "turn_start_pending": False, } if isinstance(completed_turn, Mapping): self.state["completed_turn"] = None self.turn_event.clear() if self.on_state_change is not None: self.on_state_change( changes, {"method": "mmo/turn/accepted", "params": {"threadId": thread_id}}, ) return turn_id def take_completed_turn(self) -> dict[str, Any] | None: with self.state_lock: value = self.state.get("completed_turn") self.state["completed_turn"] = None self.turn_event.clear() return dict(value) if isinstance(value, Mapping) else None def retain_partial_evidence( data: Mapping[str, Any], directory: Path, *, reason: str, events_filename: str = "events.jsonl", result_filename: str = "result.md", partial_filename: str = "partial-result.md", title: str = "Partial agent result", ) -> dict[str, Any]: """Materialize bounded readable evidence from an interrupted app-server job.""" for filename in (events_filename, result_filename, partial_filename): if Path(filename).name != filename: raise ValueError("partial-evidence filenames must be plain child names") partial_path = directory / partial_filename if partial_path.is_symlink(): raise RuntimeError("partial-evidence output cannot traverse a symlink") messages: deque[str] = deque(maxlen=100) observations: deque[str] = deque(maxlen=100) trace_window_truncated = False # Runtime state records include these paths for observability, but evidence # recovery uses only the canonical job/session children. A corrupted state # file must never turn partial-result recovery into an arbitrary file read. events_path = directory / events_filename if events_path.is_file() and not events_path.is_symlink(): with events_path.open("rb") as handle: size = events_path.stat().st_size start = max(0, size - PARTIAL_EVENT_WINDOW_BYTES) if start: trace_window_truncated = True handle.seek(start - 1) if handle.read(1) != b"\n": handle.readline() for raw in handle: try: envelope = strict_json_loads(raw) except (json.JSONDecodeError, ValueError): continue pending: list[Any] = [envelope] visited = 0 while pending and visited < 20_000: visited += 1 value = pending.pop() if isinstance(value, Mapping): if value.get("type") == "agentMessage" and isinstance( value.get("text"), str ): message, _truncated = bounded_text(str(value["text"]).strip(), 10_000) if message and message not in messages: messages.append(message) item_type = value.get("type") if isinstance(item_type, str) and item_type in { "commandExecution", "mcpToolCall", }: observation = json.dumps( value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), )[:4000] if observation not in observations: observations.append(observation) pending.extend(value.values()) elif isinstance(value, list): pending.extend(value) result_path = directory / result_filename if result_path.is_file() and not result_path.is_symlink(): candidate, _truncated = bounded_text( result_path.read_text(encoding="utf-8", errors="replace").strip(), 100_000 ) if candidate and candidate not in messages: messages.append(candidate) sections = [ f"# {title}", "", f"Terminal reason: {reason}", "", "This retained evidence is not a contract-valid final result. Review it directly or " "continue the persisted thread; it cannot be accepted or integrated as a completed result.", ] if messages: sections.extend(["", "## Completed assistant messages", "", "\n\n".join(messages)]) if observations: sections.extend( [ "", "## Completed tool observations", "", "\n".join(f"- `{item}`" for item in observations), ] ) if trace_window_truncated: sections.extend( [ "", "Older event records are omitted from this bounded summary; the complete durable " "JSONL trace remains available.", ] ) text = "\n".join(sections).strip() + "\n" atomic_write_text(partial_path, text, 0o600) return { "partial_result_path": str(partial_path), "partial_result_sha256": hashlib.sha256(text.encode("utf-8")).hexdigest(), "partial_message_count": len(messages), "partial_tool_observation_count": len(observations), "partial_trace_window_truncated": trace_window_truncated, "result_kind": "partial", "result_state": "unread", } def receive_control_request(connection: socket.socket) -> dict[str, Any]: """Read one bounded newline-framed local control request.""" raw = bytearray() while b"\n" not in raw: chunk = connection.recv(65536) if not chunk: break raw.extend(chunk) if len(raw) > MAX_CONTROL_MESSAGE_BYTES: raise ValueError("control request exceeds 4 MiB") request = strict_json_loads(bytes(raw).split(b"\n", 1)[0]) if not isinstance(request, dict): raise ValueError("control request must be an object") return request def send_control_response(connection: socket.socket, value: Mapping[str, Any]) -> None: """Write one strict newline-framed local control response.""" payload = ( json.dumps( dict(value), ensure_ascii=False, separators=(",", ":"), allow_nan=False, ).encode("utf-8") + b"\n" ) connection.sendall(payload) def serve_control_socket( socket_path: Path, *, stop_event: threading.Event, handler: Callable[[Mapping[str, Any]], Any], on_ready: Callable[[], None], backlog: int, connection_timeout: float | None, ) -> None: """Own the shared local control framing while a runner owns action semantics.""" if socket_path.exists() or socket_path.is_symlink(): if socket_path.is_symlink() or not socket_path.is_socket(): raise RuntimeError("control socket path is unsafe") socket_path.unlink() try: with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as server: server.bind(str(socket_path)) os.chmod(socket_path, 0o600) server.listen(backlog) server.settimeout(0.5) on_ready() while not stop_event.is_set(): try: connection, _ = server.accept() except TimeoutError: continue with connection: if connection_timeout is not None: connection.settimeout(connection_timeout) try: result = handler(receive_control_request(connection)) send_control_response(connection, {"ok": True, "result": result}) except Exception as exc: with contextlib.suppress(OSError): send_control_response( connection, {"ok": False, "error": f"{type(exc).__name__}: {exc}"}, ) finally: with contextlib.suppress(OSError): socket_path.unlink() def send_control_request( socket_path: Path, request: Mapping[str, Any], *, timeout: float = 30.0, ) -> dict[str, Any]: """Send one bounded request to a worker-owned Unix control socket.""" encoded = ( json.dumps( dict(request), ensure_ascii=False, separators=(",", ":"), allow_nan=False ).encode("utf-8") + b"\n" ) with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as client: client.settimeout(timeout) # A connect failure proves that the request was not delivered. Once the # socket is connected, a send/receive failure is ambiguous: the worker # may already have applied the mutation before its reply was lost. client.connect(str(socket_path)) try: client.sendall(encoded) chunks: list[bytes] = [] total = 0 while True: chunk = client.recv(65536) if not chunk: break chunks.append(chunk) total += len(chunk) if total > MAX_CONTROL_MESSAGE_BYTES: raise ControlDeliveryUnknown("worker control response exceeds 4 MiB") if b"\n" in chunk: break except ControlDeliveryUnknown: raise except OSError as exc: raise ControlDeliveryUnknown( f"worker control response was lost after delivery began: {exc}" ) from exc line = b"".join(chunks).split(b"\n", 1)[0] if not line: raise ControlDeliveryUnknown("worker control socket returned no response") try: value = strict_json_loads(line) except (json.JSONDecodeError, UnicodeDecodeError, ValueError) as exc: raise ControlDeliveryUnknown(f"worker control socket returned invalid JSON: {exc}") from exc if not isinstance(value, dict): raise ControlDeliveryUnknown("worker control socket returned a non-object response") if value.get("ok") is not True: raise ControlRequestRejected(str(value.get("error") or "worker control request failed")) return value