vance/codex-mmo
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
codex-mmo/libexec/mmo_app_server.py
T
vance 53df0eed10 😏
2026-08-24 08:11:59 -07:00

2340 lines
97 KiB
Python
Raw Permalink Blame History

#!/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<value>\d{4}-\d{2}-\d{2}[ T]\d{2}:\d{2}:\d{2})"
r"(?:\s*(?P<zone>Z|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
Reference in New Issue View Git Blame Copy Permalink