#!/usr/bin/env python3 """Deterministic Codex CLI stand-in used by the offline integration suite.""" from __future__ import annotations import base64 import datetime as dt import hashlib import json import os import re import signal import socket import struct import sys import termios import threading import time import uuid from pathlib import Path from typing import Any MODELS = [ "codex-auto-review", "gpt-5.2", "gpt-5.3-codex-spark", "gpt-5.4", "gpt-5.4-mini", "gpt-5.5", "gpt-5.6-luna", "gpt-5.6-sol", "gpt-5.6-terra", ] MODEL_METADATA = { "gpt-5.6-sol": (272000, ["low", "medium", "high", "xhigh", "max", "ultra"]), "gpt-5.6-terra": (272000, ["low", "medium", "high", "xhigh", "max", "ultra"]), "gpt-5.6-luna": (272000, ["low", "medium", "high", "xhigh", "max"]), } def value_for(schema: dict[str, Any]) -> Any: if "const" in schema: return schema["const"] if schema.get("enum"): return schema["enum"][0] if schema.get("oneOf"): return value_for(schema["oneOf"][0]) if schema.get("anyOf"): return value_for(schema["anyOf"][0]) kind = schema.get("type") if isinstance(kind, list): kind = next((item for item in kind if item != "null"), kind[0]) if kind == "object" or "properties" in schema: props = schema.get("properties", {}) keys = schema.get("required", list(props)) return {key: value_for(props[key]) for key in keys if key in props} if kind == "array": count = max(0, int(schema.get("minItems", 0))) return [value_for(schema.get("items", {})) for _ in range(count)] if kind == "string" or kind is None: if schema.get("pattern") == "^[0-9a-f]{64}$": return "0" * 64 if schema.get("format") in {"uri", "uri-reference"}: return "https://example.invalid/evidence" value = "fake verified value" minimum = int(schema.get("minLength", 0)) if len(value) < minimum: value += "x" * (minimum - len(value)) return value if kind == "integer": return max(0, int(schema.get("minimum", 0))) if kind == "number": return float(max(0, schema.get("minimum", 0))) if kind == "boolean": return True if kind == "null": return None return "fake verified value" def extract_schema(prompt: str) -> dict[str, Any] | None: markers = ( "against this JSON Schema:", "Return exactly one document satisfying this schema:", ) marker = next((candidate for candidate in markers if candidate in prompt), None) if marker is None: return None tail = prompt.split(marker, 1)[1] decoder = json.JSONDecoder() for match in re.finditer(r"\{", tail): try: value, _ = decoder.raw_decode(tail[match.start() :]) except json.JSONDecodeError: continue if isinstance(value, dict): return value return None def literal_result(prompt: str, cwd: Path) -> dict[str, Any] | None: marker = "Perform only the following runtime-generated literal operation." if marker not in prompt: return None tail = prompt.split(marker, 1)[1] decoder = json.JSONDecoder() task: dict[str, Any] | None = None for match in re.finditer(r"\{", tail): try: value, _ = decoder.raw_decode(tail[match.start() :]) except json.JSONDecodeError: continue if isinstance(value, dict) and value.get("operation") in { "locate", "references", "extract", "summarize_supplied", }: task = value break if task is None: return None operation = str(task["operation"]) if operation == "summarize_supplied": return { "operation": operation, "input_sha256": task["input_sha256"], "evidence": [], "summary_points": ["bounded literal summary"], } if operation == "extract": path = cwd / str(task["path"]) content = path.read_bytes() lines = content.decode("utf-8", errors="replace").splitlines() start = int(task["start_line"]) end = int(task["end_line"]) evidence = [ { "path": str(task["path"]), "sha256": hashlib.sha256(content).hexdigest(), "start_line": start, "end_line": end, "excerpt": "\n".join(lines[start - 1 : end]), } ] else: query = str(task["needle"] if operation == "locate" else task["symbol"]) evidence = [] for raw in task.get("paths", ["."]): candidate = (cwd / str(raw)).resolve() paths = [candidate] if candidate.is_file() else sorted(candidate.rglob("*")) for path in paths: if not path.is_file() or path.is_symlink(): continue content = path.read_bytes() lines = content.decode("utf-8", errors="replace").splitlines() for number, line in enumerate(lines, 1): if query not in line: continue evidence.append( { "path": path.relative_to(cwd).as_posix(), "sha256": hashlib.sha256(content).hexdigest(), "start_line": number, "end_line": number, "excerpt": line, } ) break if evidence: break if evidence: break return { "operation": operation, "input_sha256": None, "evidence": evidence, "summary_points": [], } def parse_output_path(args: list[str]) -> Path | None: try: return Path(args[args.index("--output-last-message") + 1]) except (ValueError, IndexError): return None def parse_cwd(args: list[str]) -> Path: try: return Path(args[args.index("-C") + 1]).resolve() except (ValueError, IndexError): return Path.cwd() def interactive_rollout(args: list[str]) -> tuple[str, Path]: home = Path(os.environ["CODEX_HOME"]).resolve() cwd = Path.cwd().resolve() resume_index: int | None = None if len(args) >= 2 and args[-2] == "resume": try: uuid.UUID(args[-1]) except ValueError: pass else: resume_index = len(args) - 2 if resume_index is not None: try: thread_id = args[resume_index + 1] except IndexError as exc: raise RuntimeError("fake resume requires a thread id") from exc matches = list((home / "sessions").rglob(f"*{thread_id}.jsonl")) if len(matches) != 1: raise RuntimeError(f"fake resume could not resolve thread {thread_id}") rollout = matches[0] with rollout.open("a", encoding="utf-8") as handle: handle.write( json.dumps( { "timestamp": dt.datetime.now(dt.UTC).isoformat(), "type": "event_msg", "payload": {"type": "fake_resume", "thread_id": thread_id}, }, sort_keys=True, ) + "\n" ) else: thread_id = str(uuid.uuid4()) now = dt.datetime.now(dt.UTC) directory = home / "sessions" / now.strftime("%Y") / now.strftime("%m") / now.strftime("%d") directory.mkdir(parents=True, exist_ok=True) rollout = directory / f"rollout-{now.strftime('%Y-%m-%dT%H-%M-%S')}-{thread_id}.jsonl" rollout.write_text( json.dumps( { "timestamp": now.isoformat(), "type": "session_meta", "payload": { "id": thread_id, "source": "cli", "cwd": str(cwd), "cli_version": "99.0.0-fake", "originator": "codex-tui", }, }, sort_keys=True, ) + "\n", encoding="utf-8", ) with (home / "fake-interactive-invocations.jsonl").open("a", encoding="utf-8") as handle: handle.write( json.dumps({"args": args, "cwd": str(cwd), "thread_id": thread_id}, sort_keys=True) + "\n" ) return thread_id, rollout def apply_directives(prompt: str, cwd: Path) -> None: sleep = re.search(r"FAKE_SLEEP(?:_SECONDS)?\s*=\s*([0-9.]+)", prompt) if sleep: time.sleep(min(float(sleep.group(1)), 30.0)) if "FAKE_ORPHAN_CHILD" in prompt and not (cwd / "fake-orphan.pid").exists(): child_pid = os.fork() if child_pid == 0: os.execl( sys.executable, sys.executable, "-c", "import time; time.sleep(60)", ) (cwd / "fake-orphan.pid").write_text(str(child_pid), encoding="utf-8") for match in re.finditer(r"^FAKE_WRITE\s+([^:]+)::(.*)$", prompt, re.MULTILINE): target = (cwd / match.group(1).strip()).resolve() target.parent.mkdir(parents=True, exist_ok=True) target.write_text(match.group(2).replace("\\n", "\n") + "\n", encoding="utf-8") if "Fix the Inventory.reserve contract completely" in prompt: target = cwd / "inventory.py" if target.is_file(): text = target.read_text(encoding="utf-8") needle = ' """Reserve positive stock atomically or report that it is unavailable."""\n' replacement = ( needle + " if isinstance(quantity, bool) or not isinstance(quantity, int) or quantity <= 0:\n return False\n" ) if replacement not in text: target.write_text(text.replace(needle, replacement), encoding="utf-8") def final_text(prompt: str, cwd: Path) -> str: if "FAKE_EMPTY_RESULT" in prompt: return "" literal = literal_result(prompt, cwd) if literal is not None: return json.dumps(literal, sort_keys=True) if "FAKE_INVALID_FIRST_RESULT" in prompt and "Repair only the final JSON result" not in prompt: return "not json" schema = extract_schema(prompt) if schema is not None: if schema.get("title") == "Reference versus real-browser render review": return json.dumps( { "verdict": "concerns", "image_artifacts": [], "mismatches": [], "blockers": [], }, sort_keys=True, ) return json.dumps(value_for(schema), sort_keys=True) markers = re.findall(r"\b(?:EVAL_[A-Z_]+|MMO_[A-Z_]+)\b", prompt) return "FAKE_CODEX_OK" + ("\n" + "\n".join(dict.fromkeys(markers)) if markers else "") def _fake_schema_bundle(args: list[str]) -> int: try: destination = Path(args[args.index("--out") + 1]) except (ValueError, IndexError): return 2 destination.mkdir(parents=True, exist_ok=True) methods = [ "thread/start", "thread/resume", "thread/fork", "thread/settings/update", "thread/compact/start", "turn/start", "turn/steer", "turn/interrupt", "mcpServerStatus/list", "mcpServer/tool/call", "item/tool/requestUserInput", "mcpServer/elicitation/request", "item/commandExecution/requestApproval", "item/fileChange/requestApproval", "item/permissions/requestApproval", "applyPatchApproval", "execCommandApproval", ] (destination / "ClientRequest.json").write_text( json.dumps({"methods": methods}, sort_keys=True), encoding="utf-8" ) (destination / "ServerRequest.json").write_text( json.dumps({"methods": methods[10:]}, sort_keys=True), encoding="utf-8" ) contracts = { "v2/ThreadStartParams.json": ( { "cwd", "sandbox", "approvalPolicy", "allowProviderModelFallback", "ephemeral", "historyMode", "dynamicTools", }, set(), ), "v2/ThreadResumeParams.json": ( {"threadId", "path", "cwd", "sandbox", "approvalPolicy", "excludeTurns"}, {"threadId"}, ), "v2/ThreadForkParams.json": ( { "threadId", "path", "cwd", "sandbox", "approvalPolicy", "ephemeral", "deferGoalContinuation", "excludeTurns", }, {"threadId"}, ), "v2/TurnStartParams.json": ( {"threadId", "input", "effort", "outputSchema"}, {"threadId", "input"}, ), "v2/TurnSteerParams.json": ( {"threadId", "expectedTurnId", "input"}, {"threadId", "expectedTurnId", "input"}, ), "v2/TurnInterruptParams.json": ( {"threadId", "turnId"}, {"threadId", "turnId"}, ), "v2/ThreadSettingsUpdateParams.json": ({"threadId", "effort"}, {"threadId"}), "v2/ThreadCompactStartParams.json": ({"threadId"}, {"threadId"}), "ToolRequestUserInputResponse.json": ({"answers"}, {"answers"}), "McpServerElicitationRequestResponse.json": ({"action", "content", "_meta"}, {"action"}), "CommandExecutionRequestApprovalResponse.json": ({"decision"}, {"decision"}), "FileChangeRequestApprovalResponse.json": ({"decision"}, {"decision"}), "PermissionsRequestApprovalResponse.json": ( {"permissions", "scope", "strictAutoReview"}, {"permissions"}, ), "ApplyPatchApprovalResponse.json": ({"decision"}, {"decision"}), "ExecCommandApprovalResponse.json": ({"decision"}, {"decision"}), } response_enums = { "McpServerElicitationRequestResponse.json": {"accept", "decline", "cancel"}, "CommandExecutionRequestApprovalResponse.json": { "accept", "acceptForSession", "decline", "cancel", "allow", "deny", }, "FileChangeRequestApprovalResponse.json": { "accept", "acceptForSession", "decline", "cancel", }, "PermissionsRequestApprovalResponse.json": { "turn", "session", "read", "write", "deny", "path", "glob_pattern", "special", "root", "minimal", "project_roots", "tmpdir", "slash_tmp", "unknown", }, "ApplyPatchApprovalResponse.json": { "approved", "approved_for_session", "timed_out", "abort", "allow", "deny", }, "ExecCommandApprovalResponse.json": { "approved", "approved_for_session", "timed_out", "abort", "allow", "deny", }, } for relative, (properties, required) in contracts.items(): path = destination / relative path.parent.mkdir(parents=True, exist_ok=True) path.write_text( json.dumps( { "type": "object", "properties": {key: {} for key in sorted(properties)}, "required": sorted(required), "definitions": { "compatibilityEnums": {"enum": sorted(response_enums.get(relative, set()))} }, }, sort_keys=True, ), encoding="utf-8", ) return 0 class _FakeWebSocketPeer: """Small RFC 6455 server peer used by the deterministic app-server fake.""" def __init__(self, connection: socket.socket) -> None: self.connection = connection self.buffer = bytearray() self.send_lock = threading.Lock() self.closed = False self._handshake() def _recv_exact(self, size: int) -> bytes: while len(self.buffer) < size: chunk = self.connection.recv(max(4096, size - len(self.buffer))) if not chunk: raise ConnectionError("fake WebSocket client disconnected") self.buffer.extend(chunk) value = bytes(self.buffer[:size]) del self.buffer[:size] return value def _handshake(self) -> None: marker = b"\r\n\r\n" while marker not in self.buffer: chunk = self.connection.recv(4096) if not chunk: raise ConnectionError("fake WebSocket handshake closed") self.buffer.extend(chunk) if len(self.buffer) > 64 * 1024: raise ValueError("fake WebSocket handshake is too large") raw, trailing = bytes(self.buffer).split(marker, 1) self.buffer = bytearray(trailing) headers: dict[str, str] = {} for line in raw.decode("iso-8859-1").split("\r\n")[1:]: if ":" not in line: continue name, value = line.split(":", 1) headers[name.strip().lower()] = value.strip() key = headers.get("sec-websocket-key") if not key: raise ValueError("fake WebSocket handshake has no key") digest = base64.b64encode( hashlib.sha1( # noqa: S324 - required by RFC 6455 (key + "258EAFA5-E914-47DA-95CA-C5AB0DC85B11").encode("ascii") ).digest() ).decode("ascii") self.connection.sendall( ( "HTTP/1.1 101 Switching Protocols\r\n" "Upgrade: websocket\r\n" "Connection: Upgrade\r\n" f"Sec-WebSocket-Accept: {digest}\r\n\r\n" ).encode("ascii") ) @staticmethod def _frame(opcode: int, payload: bytes) -> bytes: first = 0x80 | opcode length = len(payload) if length < 126: return bytes((first, length)) + payload if length <= 0xFFFF: return bytes((first, 126)) + struct.pack("!H", length) + payload return bytes((first, 127)) + struct.pack("!Q", length) + payload def send(self, value: dict[str, Any]) -> None: payload = json.dumps(value, ensure_ascii=False, separators=(",", ":")).encode() with self.send_lock: if self.closed: return try: self.connection.sendall(self._frame(0x1, payload)) except OSError: self.closed = True def receive(self) -> dict[str, Any]: fragments = bytearray() fragmented = False while True: first, second = self._recv_exact(2) final = bool(first & 0x80) if first & 0x70: raise ValueError("fake WebSocket client used reserved frame bits") opcode = first & 0x0F masked = bool(second & 0x80) if not masked: raise ValueError("fake WebSocket client frame is not masked") length = second & 0x7F if opcode >= 0x8 and (not final or length > 125): raise ValueError("malformed fake WebSocket control frame") if length == 126: length = struct.unpack("!H", self._recv_exact(2))[0] if length < 126: raise ValueError("non-minimal fake WebSocket frame length") elif length == 127: encoded_length = self._recv_exact(8) if encoded_length[0] & 0x80: raise ValueError("invalid fake WebSocket 64-bit frame length") length = struct.unpack("!Q", encoded_length)[0] if length <= 0xFFFF: raise ValueError("non-minimal fake WebSocket frame length") mask = self._recv_exact(4) payload = self._recv_exact(length) payload = bytes(value ^ mask[index & 3] for index, value in enumerate(payload)) if opcode == 0x8: with self.send_lock: self.connection.sendall(self._frame(0x8, payload)) raise ConnectionError("fake WebSocket client closed") if opcode == 0x9: with self.send_lock: self.connection.sendall(self._frame(0xA, payload)) continue if opcode == 0xA: continue if opcode == 0x1: if fragmented: raise ValueError("nested fake WebSocket message") fragments.extend(payload) fragmented = not final elif opcode == 0x0 and fragmented: fragments.extend(payload) fragmented = not final else: raise ValueError(f"unsupported fake WebSocket opcode {opcode}") if not fragmented: value = json.loads(fragments.decode()) if not isinstance(value, dict): raise ValueError("fake app-server request must be an object") return value def close(self) -> None: self.closed = True try: self.connection.close() except OSError: pass def _fake_app_server(args: list[str]) -> int: home = Path(os.environ["CODEX_HOME"]).resolve() state_lock = threading.Lock() peers_lock = threading.Lock() peers: list[_FakeWebSocketPeer] = [] threads: dict[str, dict[str, Any]] = {} active: dict[str, dict[str, Any]] = {} pending_requests: dict[str, tuple[str, threading.Event]] = {} def emit(value: dict[str, Any]) -> None: with peers_lock: current = list(peers) for peer in current: peer.send(value) def response(peer: _FakeWebSocketPeer, request_id: Any, result: Any) -> None: peer.send({"id": request_id, "result": result}) def error(peer: _FakeWebSocketPeer, request_id: Any, code: int, message: str) -> None: peer.send( { "id": request_id, "error": {"code": code, "message": message}, } ) def new_rollout( cwd: Path, *, source_path: Path | None = None, parent_thread_id: str | None = None, ) -> dict[str, Any]: thread_id = str(uuid.uuid4()) now = dt.datetime.now(dt.UTC) directory = home / "sessions" / now.strftime("%Y") / now.strftime("%m") / now.strftime("%d") directory.mkdir(parents=True, exist_ok=True) path = directory / f"rollout-{now.strftime('%Y-%m-%dT%H-%M-%S')}-{thread_id}.jsonl" turns: list[dict[str, Any]] = [] prompt_history: list[str] = [] if source_path is not None and source_path.is_file(): for line in source_path.read_text(encoding="utf-8", errors="replace").splitlines()[1:]: try: row = json.loads(line) except json.JSONDecodeError: continue if row.get("type") == "fake_turn" and isinstance(row.get("turn"), dict): turns.append(row["turn"]) elif row.get("type") == "fake_prompt" and isinstance(row.get("text"), str): prompt_history.append(row["text"]) path.write_text( json.dumps( { "timestamp": now.isoformat(), "type": "session_meta", "payload": { "id": thread_id, "source": "appServer", "cwd": str(cwd), "cli_version": "99.0.0-fake", "parent_thread_id": parent_thread_id, }, }, sort_keys=True, ) + "\n", encoding="utf-8", ) for turn in turns: with path.open("a", encoding="utf-8") as handle: handle.write(json.dumps({"type": "fake_turn", "turn": turn}) + "\n") for historical_prompt in prompt_history: with path.open("a", encoding="utf-8") as handle: handle.write(json.dumps({"type": "fake_prompt", "text": historical_prompt}) + "\n") value = { "id": thread_id, "sessionId": thread_id, "path": str(path), "cwd": str(cwd), "createdAt": int(now.timestamp()), "updatedAt": int(now.timestamp()), "source": "vscode", "threadSource": None, "ephemeral": False, "parentThreadId": parent_thread_id, "forkedFromId": None, "agentRole": None, "agentNickname": None, "turns": turns, "prompt_history": prompt_history, "status": {"type": "idle"}, "goal": None, "archived": False, } threads[thread_id] = value return value def load_thread(thread_id: str, path_value: Any = None) -> dict[str, Any] | None: if thread_id in threads: return threads[thread_id] candidates = [Path(path_value)] if isinstance(path_value, str) else [] candidates.extend((home / "sessions").rglob(f"*{thread_id}.jsonl")) matches = [path for path in candidates if path.is_file()] if not matches: return None path = matches[0].resolve() turns: list[dict[str, Any]] = [] prompt_history: list[str] = [] rows = path.read_text(encoding="utf-8", errors="replace").splitlines() parent_thread_id = None if rows: try: metadata = json.loads(rows[0]) except json.JSONDecodeError: metadata = None payload = metadata.get("payload") if isinstance(metadata, dict) else None if isinstance(payload, dict) and isinstance(payload.get("parent_thread_id"), str): parent_thread_id = payload["parent_thread_id"] for line in rows[1:]: try: row = json.loads(line) except json.JSONDecodeError: continue if row.get("type") == "fake_turn" and isinstance(row.get("turn"), dict): turns.append(row["turn"]) elif row.get("type") == "fake_prompt" and isinstance(row.get("text"), str): prompt_history.append(row["text"]) value = { "id": thread_id, "sessionId": thread_id, "path": str(path), "cwd": str(Path.cwd().resolve()), "createdAt": int(path.stat().st_mtime), "updatedAt": int(path.stat().st_mtime), "source": "vscode", "threadSource": None, "ephemeral": False, "parentThreadId": parent_thread_id, "forkedFromId": None, "agentRole": None, "agentNickname": None, "turns": turns, "prompt_history": prompt_history, "status": {"type": "idle"}, "goal": None, } threads[thread_id] = value return value def run_turn( thread: dict[str, Any], turn_id: str, prompt_parts: list[str], output_schema: Any ) -> None: turn_prompt = "\n".join(prompt_parts) prompt = turn_prompt interrupt = active[thread["id"]]["interrupt"] sleep_match = re.search(r"FAKE_SLEEP(?:_SECONDS)?\s*=\s*([0-9.]+)", prompt) if sleep_match: deadline = time.monotonic() + min(float(sleep_match.group(1)), 30.0) while time.monotonic() < deadline and not interrupt.wait(0.02): pass if "FAKE_REQUEST_USER_INPUT" in prompt and not interrupt.is_set(): request_id = f"fake-input-{turn_id}" wait = threading.Event() pending_requests[request_id] = (thread["id"], wait) emit( { "id": request_id, "method": "item/tool/requestUserInput", "params": { "threadId": thread["id"], "turnId": turn_id, "questions": [], }, } ) while not wait.wait(0.02) and not interrupt.is_set(): pass if "FAKE_REQUEST_COMMAND_APPROVAL" in prompt and not interrupt.is_set(): request_id = f"fake-approval-{turn_id}" wait = threading.Event() pending_requests[request_id] = (thread["id"], wait) emit( { "id": request_id, "method": "item/commandExecution/requestApproval", "params": { "threadId": thread["id"], "turnId": turn_id, "itemId": f"fake-command-{turn_id}", "reason": "synthetic approval request", }, } ) while not wait.wait(0.02) and not interrupt.is_set(): pass with state_lock: prompt = "\n".join(thread.get("prompt_history", [])) if "FAKE_APP_SERVER_CRASH_ONCE" in prompt: marker = home / "fake-app-server-crashed-once" if not marker.exists(): marker.write_text("crashed\n", encoding="utf-8") os._exit(70) if interrupt.is_set(): status = "interrupted" text = "" elif ( "FAKE_PROVIDER_USAGE_LIMIT" in prompt and not ( "FAKE_PROVIDER_USAGE_LIMIT_ONCE" in prompt and any( isinstance(prior.get("error"), dict) and "usage limit" in str(prior["error"].get("message", "")).lower() for prior in thread.get("turns", []) if isinstance(prior, dict) ) ) ) or "FAKE_PROVIDER_BAD_REQUEST" in prompt: status = "failed" text = "" else: apply_directives( re.sub(r"FAKE_SLEEP(?:_SECONDS)?\s*=\s*[0-9.]+", "", prompt), Path.cwd() ) text = ( json.dumps(value_for(output_schema), sort_keys=True) if isinstance(output_schema, dict) else final_text(prompt, Path.cwd()) ) status = "completed" turn: dict[str, Any] = { "id": turn_id, "status": status, "items": ([{"type": "agentMessage", "text": text}] if text else []), } if status == "failed": turn["error"] = ( { "message": "Provider rejected the request as invalid", "codexErrorInfo": "badRequest", } if "FAKE_PROVIDER_BAD_REQUEST" in prompt else { "message": ( "Provider usage limit reached; limit will reset at 2026-08-22 05:24:28" ) } ) multiplier = len(thread["turns"]) + 1 total_usage = { "inputTokens": 101 * multiplier, "outputTokens": 23 * multiplier, "cachedInputTokens": 7 * multiplier, "cacheWriteInputTokens": 3 * multiplier, "reasoningOutputTokens": 11 * multiplier, "totalTokens": 124 * multiplier, } emit( { "method": "thread/tokenUsage/updated", "params": { "threadId": thread["id"], "turnId": turn_id, "tokenUsage": {"total": total_usage, "last": total_usage}, }, } ) if any( marker in prompt for marker in ( "FAKE_GOAL_COMPLETE_BEFORE_TURN", "FAKE_GOAL_BUDGET_LIMIT_BEFORE_TURN", ) ): # The real worker starts the first turn while its goal is paused, # then activates the goal. Do not make these ordering fixtures # depend on host scheduling speed: wait until that activation is # observable before publishing the requested terminal goal state. deadline = time.monotonic() + 5.0 while time.monotonic() < deadline: with state_lock: goal = thread.get("goal") if isinstance(goal, dict) and goal.get("status") == "active": break time.sleep(0.01) with state_lock: goal = thread.get("goal") completed_goal = None if ( "FAKE_GOAL_COMPLETE_BEFORE_TURN" in prompt and isinstance(goal, dict) and goal.get("status") == "active" ): goal["status"] = "complete" goal["tokensUsed"] = total_usage["totalTokens"] goal["timeUsedSeconds"] = 0 completed_goal = dict(goal) elif ( "FAKE_GOAL_BUDGET_LIMIT_BEFORE_TURN" in prompt and isinstance(goal, dict) and goal.get("status") == "active" ): goal["status"] = "budgetLimited" goal["tokensUsed"] = total_usage["totalTokens"] goal["timeUsedSeconds"] = 0 completed_goal = dict(goal) if completed_goal is not None: emit( { "method": "thread/goal/updated", "params": {"threadId": thread["id"], "goal": completed_goal}, } ) time.sleep(0.75) thread["turns"].append(turn) with Path(thread["path"]).open("a", encoding="utf-8") as handle: handle.write(json.dumps({"type": "fake_turn", "turn": turn}, sort_keys=True) + "\n") crash_after_persist = "FAKE_APP_SERVER_CRASH_AFTER_PERSIST_ONCE" in prompt crash_during_finalization = ( "FAKE_APP_SERVER_CRASH_DURING_FINALIZATION_ONCE" in prompt and "Terminal serialization turn" in turn_prompt ) if crash_after_persist or crash_during_finalization: marker = home / ( "fake-app-server-crashed-during-finalization-once" if crash_during_finalization else "fake-app-server-crashed-after-persist-once" ) if not marker.exists(): marker.write_text("crashed\n", encoding="utf-8") os._exit(71) emit( { "method": "turn/completed", "params": {"threadId": thread["id"], "turn": turn}, } ) with state_lock: active.pop(thread["id"], None) thread["status"] = {"type": "idle"} goal = thread.get("goal") if status == "completed" and isinstance(goal, dict) and goal.get("status") == "active": goal["status"] = "complete" goal["tokensUsed"] = total_usage["totalTokens"] goal["timeUsedSeconds"] = 0 completed_goal = dict(goal) else: completed_goal = None emit( { "method": "thread/status/changed", "params": {"threadId": thread["id"], "status": {"type": "idle"}}, } ) if completed_goal is not None: emit( { "method": "thread/goal/updated", "params": {"threadId": thread["id"], "goal": completed_goal}, } ) def handle_message(peer: _FakeWebSocketPeer, message: dict[str, Any]) -> None: if "jsonrpc" in message: raise ValueError("Codex 0.149 app-server messages do not carry jsonrpc") request_id = message.get("id") method = message.get("method") raw_params = message.get("params") params: dict[str, Any] = raw_params if isinstance(raw_params, dict) else {} if method == "initialize": initialize_delay = os.environ.get("FAKE_CODEX_APP_SERVER_INITIALIZE_DELAY") if initialize_delay: time.sleep(min(float(initialize_delay), 30.0)) response( peer, request_id, {"serverInfo": {"name": "fake-codex", "version": "99.0.0"}}, ) elif method == "initialized": return elif method == "mcpServerStatus/list": response(peer, request_id, {"data": [], "nextCursor": None}) elif method == "mcpServer/tool/call": response( peer, request_id, {"content": [{"type": "text", "text": "fake MCP tool result"}]}, ) elif method == "thread/start": started_thread = new_rollout(Path(str(params.get("cwd") or Path.cwd())).resolve()) response(peer, request_id, {"thread": started_thread}) emit( { "method": "thread/started", "params": {"thread": started_thread}, } ) elif method == "thread/resume": thread_id = str(params.get("threadId", "")) resumed_thread = load_thread(thread_id, params.get("path")) if resumed_thread is None: error(peer, request_id, -32001, "thread not found") else: response(peer, request_id, {"thread": resumed_thread}) elif method == "thread/fork": source_id = str(params.get("threadId", "")) source = load_thread(source_id, params.get("path")) if source is None: error(peer, request_id, -32001, "source thread not found") else: forked_thread = new_rollout( Path(str(params.get("cwd") or Path.cwd())).resolve(), source_path=Path(source["path"]), parent_thread_id=source_id, ) response(peer, request_id, {"thread": forked_thread}) elif method == "thread/read": read_thread = load_thread(str(params.get("threadId", ""))) if read_thread is None: error(peer, request_id, -32001, "thread not found") else: response(peer, request_id, {"thread": read_thread}) elif method == "thread/turns/list": listed_thread = load_thread(str(params.get("threadId", ""))) if listed_thread is None: error(peer, request_id, -32001, "thread not found") else: response( peer, request_id, {"data": list(listed_thread["turns"]), "nextCursor": None}, ) elif method == "thread/list": for path in (home / "sessions").rglob("*.jsonl"): try: first = path.read_text(encoding="utf-8", errors="replace").splitlines()[0] metadata = json.loads(first) except (IndexError, OSError, json.JSONDecodeError): continue payload = metadata.get("payload") if isinstance(metadata, dict) else None discovered_thread_id = payload.get("id") if isinstance(payload, dict) else None if isinstance(discovered_thread_id, str) and discovered_thread_id not in threads: load_thread(discovered_thread_id, str(path)) ancestor = params.get("ancestorThreadId") archived = params.get("archived") data = [ value for value in threads.values() if ancestor is None or value.get("parentThreadId") == ancestor if archived is None or value.get("archived", False) is archived ] response(peer, request_id, {"data": data, "nextCursor": None}) elif method == "thread/archive": archived_thread = load_thread(str(params.get("threadId", ""))) if archived_thread is None: error(peer, request_id, -32001, "thread not found") else: archived_thread["archived"] = True response(peer, request_id, {}) elif method == "thread/goal/set": objective = params.get("objective") if isinstance(objective, str) and len(objective) > 4000: error( peer, request_id, -32602, "goal objective must be at most 4000 characters", ) return goal_thread = load_thread(str(params.get("threadId", ""))) if goal_thread is None: error(peer, request_id, -32001, "thread not found") return with state_lock: existing = goal_thread.get("goal") previous_goal_status = ( existing.get("status") if isinstance(existing, dict) else None ) goal = ( dict(existing) if isinstance(existing, dict) else { "objective": "", "status": "paused", "tokenBudget": None, "tokensUsed": 0, "timeUsedSeconds": 0, } ) if "objective" in params: goal["objective"] = params["objective"] if "tokenBudget" in params: goal["tokenBudget"] = params["tokenBudget"] if "status" in params: goal["status"] = params["status"] auto_continue = False if goal["status"] == "active" and goal_thread["id"] not in active: turns = goal_thread.get("turns", []) last_turn = turns[-1] if turns and isinstance(turns[-1], dict) else None auto_continue = bool( ( isinstance(last_turn, dict) and last_turn.get("status") in {"failed", "interrupted"} ) or previous_goal_status in {"blocked", "budgetLimited", "usageLimited"} ) if not auto_continue: # Ordinary fake turns stand in for a model that marks # its goal complete. Keep that baseline behavior while # exercising real automatic continuation after an # interrupted or retryable failed turn. goal["status"] = "complete" goal_thread["goal"] = goal response(peer, request_id, {"goal": goal}) emit( { "method": "thread/goal/updated", "params": {"threadId": goal_thread["id"], "goal": goal}, } ) if auto_continue: # Codex goals normally continue automatically. Model that # protocol behavior so recovery tests cannot pass merely # because the stand-in marks an idle active goal complete. turn_id = str(uuid.uuid4()) prompt_parts = ["Continue the active goal from retained thread context."] goal_thread.setdefault("prompt_history", []).extend(prompt_parts) with Path(goal_thread["path"]).open("a", encoding="utf-8") as handle: handle.write( json.dumps({"type": "fake_prompt", "text": prompt_parts[0]}) + "\n" ) active[goal_thread["id"]] = { "turn_id": turn_id, "interrupt": threading.Event(), "prompt_parts": prompt_parts, } goal_thread["status"] = {"type": "active"} emit( { "method": "turn/started", "params": { "threadId": goal_thread["id"], "turn": {"id": turn_id, "status": "inProgress", "items": []}, }, } ) emit( { "method": "thread/status/changed", "params": { "threadId": goal_thread["id"], "status": {"type": "active"}, }, } ) threading.Thread( target=run_turn, args=(goal_thread, turn_id, prompt_parts, None), daemon=True, ).start() elif method == "turn/start": turn_thread = load_thread(str(params.get("threadId", ""))) if turn_thread is None: error(peer, request_id, -32001, "thread not found") return turn_id = str(uuid.uuid4()) prompt_parts = [ str(item.get("text", "")) for item in params.get("input", []) if isinstance(item, dict) and item.get("type") == "text" ] current_prompt = "\n".join(prompt_parts) turn_thread.setdefault("prompt_history", []).append(current_prompt) with Path(turn_thread["path"]).open("a", encoding="utf-8") as handle: handle.write(json.dumps({"type": "fake_prompt", "text": current_prompt}) + "\n") active[turn_thread["id"]] = { "turn_id": turn_id, "interrupt": threading.Event(), "prompt_parts": prompt_parts, } turn_thread["status"] = {"type": "active"} started_message = { "method": "turn/started", "params": { "threadId": turn_thread["id"], "turn": {"id": turn_id, "status": "inProgress", "items": []}, }, } if "FAKE_NOTIFY_BEFORE_RESPONSE" in current_prompt: emit(started_message) response( peer, request_id, {"turn": {"id": turn_id, "status": "inProgress", "items": []}}, ) if "FAKE_NOTIFY_BEFORE_RESPONSE" not in current_prompt: emit(started_message) emit( { "method": "thread/status/changed", "params": { "threadId": turn_thread["id"], "status": {"type": "active"}, }, } ) threading.Thread( target=run_turn, args=(turn_thread, turn_id, prompt_parts, params.get("outputSchema")), daemon=True, ).start() elif method == "turn/steer": current = active.get(str(params.get("threadId", ""))) if current is None or current["turn_id"] != params.get("expectedTurnId"): error(peer, request_id, -32002, "active turn mismatch") else: additions = [ str(item.get("text", "")) for item in params.get("input", []) if isinstance(item, dict) and item.get("type") == "text" ] current["prompt_parts"].extend(additions) steered_thread = load_thread(str(params.get("threadId", ""))) if steered_thread is not None: steered_thread.setdefault("prompt_history", []).extend(additions) with Path(steered_thread["path"]).open("a", encoding="utf-8") as handle: for addition in additions: handle.write( json.dumps({"type": "fake_prompt", "text": addition}) + "\n" ) response(peer, request_id, {"turnId": current["turn_id"]}) elif method == "turn/interrupt": current = active.get(str(params.get("threadId", ""))) if current is not None and current["turn_id"] == params.get("turnId"): current["interrupt"].set() response(peer, request_id, {}) elif method in {"thread/settings/update", "thread/compact/start"}: response(peer, request_id, {}) elif request_id is not None and method is None: pending = pending_requests.pop(str(request_id), None) if pending is not None: pending[1].set() elif request_id is not None: error(peer, request_id, -32601, f"method not found: {method}") def serve_peer(connection: socket.socket) -> None: peer: _FakeWebSocketPeer | None = None try: peer = _FakeWebSocketPeer(connection) with peers_lock: peers.append(peer) while True: handle_message(peer, peer.receive()) except (ConnectionError, OSError, ValueError, json.JSONDecodeError): pass finally: if peer is not None: with peers_lock: if peer in peers: peers.remove(peer) peer.close() else: connection.close() try: listen_value = args[args.index("--listen") + 1] except (ValueError, IndexError): print("fake app-server requires --listen unix://PATH", file=sys.stderr) return 2 if not listen_value.startswith("unix://"): print("fake app-server supports only a Unix listener", file=sys.stderr) return 2 socket_path = Path(listen_value.removeprefix("unix://")) socket_path.parent.mkdir(parents=True, exist_ok=True) listener = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) listener.bind(str(socket_path)) os.chmod(socket_path, 0o600) listener.listen(16) while True: connection, _address = listener.accept() threading.Thread(target=serve_peer, args=(connection,), daemon=True).start() def main() -> int: args = sys.argv[1:] if args and args[0] in {"--version", "version"}: print("codex-cli 0.149.0") return 0 if args[:2] == ["debug", "models"]: print( json.dumps( { "models": [ { "slug": item, "display_name": item, "context_window": MODEL_METADATA.get(item, (272000, []))[0], "input_modalities": ["text", "image"], "supported_reasoning_levels": [ {"effort": value} for value in MODEL_METADATA.get( item, (272000, ["low", "medium", "high", "xhigh"]), )[1] ], } for item in MODELS ] } ) ) return 0 if args[:2] == ["login", "status"]: print("Logged in (fake)") return 0 if args and args[0] == "login": print("Login complete (fake)") return 0 if "app-server" in args: app_index = args.index("app-server") app_args = args[app_index + 1 :] if app_args and app_args[0] == "generate-json-schema": return _fake_schema_bundle(app_args[1:]) return _fake_app_server(app_args) if args and args[0] == "exec": # Codex 0.149's shared --image argument is variadic (num_args = 1..). # A bare '-' before the option terminator is therefore an image path, # not the prompt-from-stdin operand. option_end = args.index("--") if "--" in args else len(args) if "--image" in args and "-" in args[args.index("--image") + 1 : option_end]: print("stdin sentinel was consumed by variadic --image", file=sys.stderr) return 2 prompt = sys.stdin.read() cwd = parse_cwd(args) apply_directives(prompt, cwd) exit_match = re.search(r"FAKE_EXIT\s*=\s*(\d+)", prompt) code = int(exit_match.group(1)) if exit_match else 0 output = parse_output_path(args) text = final_text(prompt, cwd) if output is not None and code == 0: output.parent.mkdir(parents=True, exist_ok=True) output.write_text(text + "\n", encoding="utf-8") print( json.dumps( { "type": "turn.completed", "usage": { "input_tokens": 101, "output_tokens": 23, "cached_input_tokens": 7, "cache_write_input_tokens": 3, "reasoning_output_tokens": 11, }, "retry_count": 0, } ) ) if code: print("forced fake failure", file=sys.stderr) return code # Interactive mode is intentionally simple but permits launch tests. interactive_rollout(args) if os.environ.get("FAKE_TUI_PROBE") == "1": resized = False def on_resize(_signum: int, _frame: object) -> None: nonlocal resized resized = True signal.signal(signal.SIGWINCH, on_resize) def state(event: str) -> dict[str, object]: try: size = os.get_terminal_size(sys.stdin.fileno()) columns, lines = size.columns, size.lines except OSError: columns, lines = -1, -1 try: foreground_pgrp = os.tcgetpgrp(sys.stdin.fileno()) except OSError: foreground_pgrp = -1 return { "event": event, "stdin_isatty": sys.stdin.isatty(), "stdout_isatty": sys.stdout.isatty(), "stderr_isatty": sys.stderr.isatty(), "no_color": os.environ.get("NO_COLOR"), "term": os.environ.get("TERM"), "pgrp": os.getpgrp(), "foreground_pgrp": foreground_pgrp, "columns": columns, "lines": lines, } # Exercise raw ANSI passthrough in addition to environment inspection. # A wrapper that injects NO_COLOR or captures/re-emits output can make # the real Codex TUI appear monochrome or corrupt its control stream. print("\x1b[32mFAKE_TUI_COLOR\x1b[0m", flush=True) # Simulate a full-screen TUI entering a modified terminal mode and then # exiting without restoring it. The MMO launcher must return the user's # terminal to the attributes captured before Codex started. attributes = termios.tcgetattr(sys.stdin.fileno()) attributes[3] &= ~termios.ECHO termios.tcsetattr(sys.stdin.fileno(), termios.TCSANOW, attributes) print(json.dumps(state("ready"), sort_keys=True), flush=True) deadline = time.monotonic() + 5.0 while not resized and time.monotonic() < deadline: time.sleep(0.02) print( json.dumps(state("resized" if resized else "resize_timeout"), sort_keys=True), flush=True, ) return 0 print("FAKE_CODEX_INTERACTIVE") return 0 if __name__ == "__main__": raise SystemExit(main())