from __future__ import annotations import fcntl import json import os import pty import select import shutil import struct import subprocess import sys import termios import time import tomllib import unittest from pathlib import Path from typing import Any from unittest import mock import mmo_codex_home import mmo_runtime from common import ROOT, RuntimeSandbox from mmo_runtime import create_session, finish_session from mmo_util import read_toml, toml_dumps from mmo_version import MMO_SCHEMA_VERSION class TuiAndModelMetadataTests(unittest.TestCase): def _config(self, home: str) -> dict[str, Any]: with (Path(home) / "config.toml").open("rb") as handle: return tomllib.load(handle) def _catalog(self, config: dict[str, Any]) -> dict[str, dict[str, Any]]: path = Path(config["model_catalog_json"]) payload = json.loads(path.read_text(encoding="utf-8")) return {item["slug"]: item for item in payload["models"]} def test_switchyard_bindings_have_exact_codex_model_metadata(self) -> None: expected: dict[str, dict[str, Any]] = { "routine_engineer": { "context_window": 1000000, "default_reasoning_level": "high", "apply_patch_tool_type": "freeform", }, "literal_scout": { "context_window": 32768, "default_reasoning_level": None, "apply_patch_tool_type": "freeform", }, } with RuntimeSandbox() as box: session = create_session(profile="access-efficient-escalation-lab", cwd=box.workspace) try: for agent_id, required in expected.items(): home_record = session["homes"][agent_id] config = self._config(home_record["home"]) self.assertIn("model_catalog_json", config) self.assertEqual( config["model_catalog_json"], home_record["model_catalog_json"] ) entries = self._catalog(config) selected = config["model"] self.assertIn(selected, entries) metadata = entries[selected] self.assertEqual(metadata["slug"], selected) self.assertEqual(metadata["context_window"], required["context_window"]) self.assertEqual(metadata["max_context_window"], required["context_window"]) self.assertEqual( metadata["default_reasoning_level"], required["default_reasoning_level"], ) self.assertEqual( metadata["apply_patch_tool_type"], required["apply_patch_tool_type"], ) self.assertEqual(metadata["shell_type"], "shell_command") self.assertTrue(metadata["supported_in_api"]) self.assertIn("text", metadata["input_modalities"]) self.assertEqual(metadata["default_reasoning_summary"], "none") self.assertNotIn("base_instructions", metadata) self.assertTrue(metadata["model_messages"]["instructions_template"]) self.assertEqual( set(metadata["model_messages"]), { "instructions_template", "instructions_variables", "approvals", "collaboration_modes", "auto_review", "permissions", "token_budget", }, ) self.assertNotIn("node_repl_auto_review_required", metadata) self.assertNotIn("node_repl_disabled", metadata) self.assertEqual( metadata["supports_reasoning_summary_parameter"], required["default_reasoning_level"] is not None and agent_id != "routine_engineer", ) self.assertNotIn("used_fallback_model_metadata", metadata) self.assertIn("truncation_policy", metadata) finally: finish_session(session["session_id"], exit_code=0) def test_function_only_model_omits_custom_patch_but_keeps_shell(self) -> None: with RuntimeSandbox() as box: profile = box.root / "function-only-model" shutil.copytree(ROOT / "profiles" / "access-efficient-escalation-lab", profile) profile_path = profile / "profile.toml" profile_data = read_toml(profile_path) profile_data["id"] = "function-only-model" profile_data["catalog"] = "catalog.toml" profile_path.write_text(toml_dumps(profile_data), encoding="utf-8") (profile / "catalog.toml").write_text( toml_dumps( { "schema_version": MMO_SCHEMA_VERSION, "models": { "opencode_go_openai_chat__deepseek_v4_flash": { "supports_custom_tools": False, } }, } ), encoding="utf-8", ) session = create_session(profile=profile, cwd=box.workspace) try: home = session["homes"]["routine_engineer"]["home"] config = self._config(home) metadata = self._catalog(config)[config["model"]] self.assertEqual(metadata["shell_type"], "shell_command") self.assertIsNone(metadata["apply_patch_tool_type"]) finally: finish_session(session["session_id"], exit_code=0) def test_hybrid_catalog_preserves_bundled_models_and_adds_external_bindings(self) -> None: with RuntimeSandbox() as box: mmo_codex_home._BUNDLED_CODEX_CATALOGS.clear() profile = box.root / "hybrid-native-external" shutil.copytree(ROOT / "profiles" / "adaptive-engineering", profile) profile_path = profile / "profile.toml" profile_data = read_toml(profile_path) profile_data["id"] = "hybrid-native-external" profile_data["agents"]["implementation_specialist"]["backends"] = [ "mcp", "native", ] specialist = profile_data["agents"]["implementation_specialist"] specialist["permissions"] = "read-only" specialist["execution_mode"] = "turn" specialist.pop("goal_token_budget") specialist.pop("max_goal_token_budget") profile_data["coordination"]["max_active_writers"] = 0 profile_path.write_text(toml_dumps(profile_data), encoding="utf-8") original = mmo_runtime.bundled_codex_catalog_for_profile with mock.patch( "mmo_runtime.bundled_codex_catalog_for_profile", wraps=original ) as load_catalog: session = create_session(profile=profile, cwd=box.workspace) try: # The active Codex catalog is queried once per session, not once # for every generated agent home. load_catalog.assert_called_once() self.assertEqual(load_catalog.call_args.args[1], Path(session["codex_binary"])) root_home = session["homes"][session["root_agent"]]["home"] config = self._config(root_home) self.assertIn("model_catalog_json", config) entries = self._catalog(config) # model_catalog_json is a startup replacement, so MMO preserves # the exact bundled rows before adding external route bindings. self.assertIn("gpt-5.6-terra", entries) self.assertIn(config["model"], entries) native_dir = Path(root_home) / "agents" native_configs = [ tomllib.loads(path.read_text(encoding="utf-8")) for path in native_dir.glob("*.toml") ] external = [ item["model"] for item in native_configs if item["model"].startswith("mmo-") ] self.assertTrue(external) self.assertTrue(all(slug in entries for slug in external)) patch_types = {entries[slug]["apply_patch_tool_type"] for slug in external} self.assertEqual(patch_types, {"freeform"}) finally: finish_session(session["session_id"], exit_code=0) def test_builtin_only_process_uses_codex_native_catalog(self) -> None: with RuntimeSandbox() as box: session = create_session(profile="codex-harness-team", cwd=box.workspace) try: config = self._config(session["homes"][session["root_agent"]]["home"]) self.assertNotIn("model_catalog_json", config) finally: finish_session(session["session_id"], exit_code=0) def test_interactive_launch_preserves_tty_with_redirected_stderr(self) -> None: with RuntimeSandbox() as box: old = {key: os.environ.get(key) for key in ("NO_COLOR", "TERM", "FAKE_TUI_PROBE")} os.environ.pop("NO_COLOR", None) os.environ["TERM"] = "xterm-256color" os.environ["FAKE_TUI_PROBE"] = "1" try: master, slave = pty.openpty() fcntl.ioctl(slave, termios.TIOCSWINSZ, struct.pack("HHHH", 24, 80, 0, 0)) environment = { **os.environ, "PYTHONPATH": os.pathsep.join([str(ROOT / "libexec"), str(ROOT / "tests")]), } process = subprocess.Popen( [ sys.executable, str(ROOT / "tests" / "helpers" / "tui_probe_launcher.py"), "access-efficient-escalation-lab", str(box.workspace), ], stdin=slave, stdout=slave, stderr=subprocess.PIPE, env=environment, start_new_session=True, close_fds=True, ) os.close(slave) buffer = b"" all_output = b"" records: list[dict] = [] resized = False deadline = time.monotonic() + 30 while time.monotonic() < deadline: readable, _, _ = select.select([master], [], [], 0.25) if readable: try: chunk = os.read(master, 65536) except OSError: break if not chunk: break all_output += chunk buffer += chunk while b"\n" in buffer: raw, buffer = buffer.split(b"\n", 1) text = raw.decode("utf-8", errors="replace").strip().strip("\r") if not text.startswith("{"): continue try: record = json.loads(text) except json.JSONDecodeError: continue records.append(record) if record.get("event") == "ready" and not resized: fcntl.ioctl( master, termios.TIOCSWINSZ, struct.pack("HHHH", 42, 132, 0, 0), ) resized = True if any( item.get("event") in {"launcher_returned", "launcher_error"} for item in records ): break _unused_stdout, stderr_output = process.communicate(timeout=10) exit_code = int(process.returncode or 0) os.close(master) diagnostic = { "records": records, "output": all_output.decode("utf-8", errors="replace"), "stderr": stderr_output.decode("utf-8", errors="replace"), } self.assertEqual(exit_code, 0, diagnostic) self.assertIn(b"\x1b[32mFAKE_TUI_COLOR\x1b[0m", all_output, diagnostic) ready = next((item for item in records if item.get("event") == "ready"), None) resize = next((item for item in records if item.get("event") == "resized"), None) returned = next( (item for item in records if item.get("event") == "launcher_returned"), None, ) self.assertIsNotNone(ready, diagnostic) self.assertIsNotNone(resize, diagnostic) self.assertIsNotNone(returned, diagnostic) assert ready is not None and resize is not None and returned is not None self.assertTrue(ready["stdin_isatty"]) self.assertTrue(ready["stdout_isatty"]) self.assertFalse(ready["stderr_isatty"]) self.assertIsNone(ready["no_color"]) self.assertEqual(ready["term"], "xterm-256color") self.assertEqual(ready["pgrp"], ready["foreground_pgrp"]) self.assertEqual((resize["columns"], resize["lines"]), (132, 42)) self.assertEqual(resize["pgrp"], resize["foreground_pgrp"]) self.assertEqual(returned["exit_code"], 0) self.assertTrue(returned["echo_enabled"], diagnostic) self.assertTrue(returned["attributes_restored"], diagnostic) self.assertEqual(returned["pgrp"], returned["foreground_pgrp"]) finally: for key, value in old.items(): if value is None: os.environ.pop(key, None) else: os.environ[key] = value if __name__ == "__main__": unittest.main()