vance/codex-mmo
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
codex-mmo/scripts/validate_package.py
T

662 lines
24 KiB
Python
Raw Permalink Normal View History

😏
2026-08-24 08:11:59 -07:00
#!/usr/bin/env python3
"""Release acceptance validator for Codex Multi-Model Orchestrator.
The validator intentionally uses only the Python standard library. It checks the
static package, all profile compositions, the complete catalog baseline,
generated snapshots, source syntax, installable-profile safety, release hygiene,
and the offline integration suite.
"""
from __future__ import annotations
import argparse
import ast
import json
import os
import py_compile
import re
import stat
import subprocess
import sys
import tempfile
import time
from collections.abc import Iterable
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any
sys.dont_write_bytecode = True
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "libexec"))
from mmo_catalog import catalog_summary, local_inventory_report # noqa: E402
from mmo_eval import discover_suites, validate_suite # noqa: E402
from mmo_profiles import ( # noqa: E402
discover_profiles,
resolve_profile,
validate_profile_pack_tree,
)
from mmo_snapshot import compile_profile # noqa: E402
from mmo_util import ( # noqa: E402
atomic_write_json,
filtered_environment,
read_toml,
sha256_file,
strict_json_loads,
terminate_process_group,
utc_now,
)
from mmo_version import MMO_SCHEMA_VERSION # noqa: E402
EXPECTED_PROFILES = {
"adaptive-engineering",
"access-efficient-escalation-lab",
"bounded-research-organization-lab",
"codex-harness-team",
"competing-implementations-lab",
"contract-first-refactoring",
"high-confidence-debugging",
"incident-hypothesis-triage",
"research-backed-engineering",
"route-resilience-lab",
"secure-change",
"visual-engineering",
}
EXPECTED_ORCHESTRATION = {"mcp", "hybrid"}
EXPECTED_EVALUATION_SUITES = {
"access-efficient",
"adaptive-change",
"bounded-research",
"codex-harness",
"competing-implementations",
"contract-refactoring",
"debugging-confidence",
"incident-triage",
"research-currentness",
"route-resilience",
"security-assurance",
"visual-conformance",
}
REQUIRED_ROOT_FILES = {
"LICENSE",
"VERSION",
"README.md",
"CHANGELOG.md",
"VALIDATION.md",
"Makefile",
"PACKAGE-MANIFEST.json",
"install.sh",
"uninstall.sh",
}
REQUIRED_DOCS = {
"ACCEPTANCE.md",
"ARCHITECTURE.md",
"CATALOG.md",
"CLI.md",
"EVALUATION.md",
"EXTERNAL-VERACITY.md",
"INSTALLATION.md",
"OPTIMIZATION.md",
"PLAN-COVERAGE.md",
"ORCHESTRATION_BACKENDS.md",
"PROFILE_SCHEMA.md",
"PROFILES.md",
"PROVIDER_DRIVERS.md",
"SECURITY.md",
"TROUBLESHOOTING.md",
"TOOL_MCP.md",
}
IGNORED_PARTS = {
".git",
"__pycache__",
".pytest_cache",
".mypy_cache",
".ruff_cache",
"dist",
"build",
}
TEXT_SUFFIXES = {
"",
".md",
".txt",
".toml",
".json",
".py",
".sh",
".in",
".example",
".gitignore",
".license",
}
SECRET_PATTERNS = {
"private key": re.compile(r"-----BEGIN (?:RSA |EC |OPENSSH |DSA )?PRIVATE KEY-----"),
"OpenAI-style secret": re.compile(r"\bsk-[A-Za-z0-9_-]{20,}\b"),
"GitHub token": re.compile(r"\bgh[pousr]_[A-Za-z0-9]{30,}\b"),
"AWS access key": re.compile(r"\bAKIA[0-9A-Z]{16}\b"),
}
@dataclass
class Report:
started_at: str = field(default_factory=utc_now)
checks: dict[str, Any] = field(default_factory=dict)
errors: list[str] = field(default_factory=list)
warnings: list[str] = field(default_factory=list)
def check(self, name: str, passed: bool, **details: Any) -> None:
self.checks[name] = {"passed": bool(passed), **details}
if not passed:
self.errors.append(f"{name} failed")
def error(self, message: str) -> None:
self.errors.append(message)
def warning(self, message: str) -> None:
self.warnings.append(message)
def finish(self) -> dict[str, Any]:
return {
"schema_version": MMO_SCHEMA_VERSION,
"package": "codex-multimodel-orchestrator",
"version": (ROOT / "VERSION").read_text(encoding="utf-8").strip(),
"started_at": self.started_at,
"finished_at": utc_now(),
"passed": not self.errors and all(item.get("passed") for item in self.checks.values()),
"checks": self.checks,
"warnings": self.warnings,
"errors": self.errors,
}
def _source_files() -> list[Path]:
files: list[Path] = []
for path in ROOT.rglob("*"):
if any(part in IGNORED_PARTS for part in path.relative_to(ROOT).parts):
continue
if path.is_file() and not path.is_symlink():
files.append(path)
return sorted(files)
def _digest(paths: Iterable[Path]) -> dict[str, str]:
return {path.relative_to(ROOT).as_posix(): sha256_file(path) for path in sorted(paths)}
def _check_structure(report: Report) -> None:
missing_root = sorted(name for name in REQUIRED_ROOT_FILES if not (ROOT / name).is_file())
missing_docs = sorted(name for name in REQUIRED_DOCS if not (ROOT / "docs" / name).is_file())
report.check(
"required_files",
not missing_root and not missing_docs,
missing_root=missing_root,
missing_docs=missing_docs,
)
caches = sorted(
path.relative_to(ROOT).as_posix()
for path in ROOT.rglob("*")
if path.name == "__pycache__" or path.suffix in {".pyc", ".pyo"}
)
report.check("release_hygiene_no_bytecode", not caches, paths=caches)
symlinks = sorted(
path.relative_to(ROOT).as_posix() for path in ROOT.rglob("*") if path.is_symlink()
)
report.check("release_hygiene_no_symlinks", not symlinks, paths=symlinks)
special = sorted(
path.relative_to(ROOT).as_posix()
for path in ROOT.rglob("*")
if not any(part in IGNORED_PARTS for part in path.relative_to(ROOT).parts)
and not (
stat.S_ISREG(path.lstat().st_mode)
or stat.S_ISDIR(path.lstat().st_mode)
or stat.S_ISLNK(path.lstat().st_mode)
)
)
report.check("release_hygiene_no_special_files", not special, paths=special)
def _check_source_manifest(report: Report) -> None:
path = ROOT / "PACKAGE-MANIFEST.json"
errors: list[str] = []
try:
manifest = strict_json_loads(path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError, ValueError) as exc:
report.check("source_manifest", False, errors=[f"unable to read manifest: {exc}"])
return
if not isinstance(manifest, dict):
report.check("source_manifest", False, errors=["manifest root must be an object"])
return
expected_manifest_fields = {
"schema_version",
"package",
"version",
"source_date_epoch",
"manifest_excludes",
"file_count",
"total_bytes",
"files",
}
if set(manifest) != expected_manifest_fields:
errors.append("manifest root has missing or unexpected fields")
schema_version = manifest.get("schema_version")
if (
not isinstance(schema_version, int)
or isinstance(schema_version, bool)
or schema_version != MMO_SCHEMA_VERSION
):
errors.append(f"schema_version must be {MMO_SCHEMA_VERSION}")
if manifest.get("package") != "codex-multimodel-orchestrator":
errors.append("package identity is invalid")
version = (ROOT / "VERSION").read_text(encoding="utf-8").strip()
if manifest.get("version") != version:
errors.append("manifest version does not match VERSION")
if manifest.get("manifest_excludes") != ["PACKAGE-MANIFEST.json"]:
errors.append("manifest_excludes is invalid")
epoch = manifest.get("source_date_epoch")
if not isinstance(epoch, int) or isinstance(epoch, bool) or not 0 <= epoch <= 0xFFFFFFFF:
errors.append("source_date_epoch is invalid")
entries = manifest.get("files")
expected: dict[str, dict[str, Any]] = {}
if not isinstance(entries, list):
errors.append("files must be an array")
else:
for entry in entries:
if not isinstance(entry, dict) or not isinstance(entry.get("path"), str):
errors.append("files contains an invalid entry")
continue
relative = entry["path"]
if set(entry) != {"path", "mode", "size", "sha256"}:
errors.append(f"manifest entry has missing or unexpected fields: {relative}")
if entry.get("mode") not in {"0644", "0755"}:
errors.append(f"manifest entry has an invalid mode: {relative}")
size = entry.get("size")
if not isinstance(size, int) or isinstance(size, bool) or size < 0:
errors.append(f"manifest entry has an invalid size: {relative}")
digest = entry.get("sha256")
if (
not isinstance(digest, str)
or len(digest) != 64
or any(character not in "0123456789abcdef" for character in digest)
):
errors.append(f"manifest entry has an invalid hash: {relative}")
if relative in expected:
errors.append(f"duplicate manifest path: {relative}")
continue
expected[relative] = {
"mode": entry.get("mode"),
"size": entry.get("size"),
"sha256": entry.get("sha256"),
}
actual: dict[str, dict[str, Any]] = {}
for source in _source_files():
relative = source.relative_to(ROOT).as_posix()
if relative == "PACKAGE-MANIFEST.json":
continue
mode = "0755" if source.stat().st_mode & stat.S_IXUSR else "0644"
actual[relative] = {
"mode": mode,
"size": source.stat().st_size,
"sha256": sha256_file(source),
}
if expected != actual:
errors.append("manifest file inventory differs from the source tree")
file_count = manifest.get("file_count")
if not isinstance(file_count, int) or isinstance(file_count, bool) or file_count != len(actual):
errors.append("file_count differs from the source tree")
total_bytes = manifest.get("total_bytes")
if (
not isinstance(total_bytes, int)
or isinstance(total_bytes, bool)
or total_bytes != sum(item["size"] for item in actual.values())
):
errors.append("total_bytes differs from the source tree")
report.check(
"source_manifest",
not errors,
errors=errors,
expected_files=len(expected),
actual_files=len(actual),
missing=sorted(set(actual) - set(expected)),
extra=sorted(set(expected) - set(actual)),
changed=sorted(key for key in set(expected) & set(actual) if expected[key] != actual[key]),
)
def _check_syntax(report: Report) -> None:
python_errors: list[str] = []
python_files = [path for path in _source_files() if path.suffix == ".py"]
with tempfile.TemporaryDirectory(prefix="mmo-pycompile-") as temporary:
compiled = Path(temporary)
for index, path in enumerate(python_files):
try:
ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
py_compile.compile(str(path), cfile=str(compiled / f"{index}.pyc"), doraise=True)
except Exception as exc: # validation boundary
python_errors.append(f"{path.relative_to(ROOT)}: {type(exc).__name__}: {exc}")
report.check("python_syntax", not python_errors, files=len(python_files), errors=python_errors)
shell_errors: list[str] = []
shell_files = [path for path in _source_files() if path.suffix in {".sh", ".in"}]
for path in shell_files:
result = subprocess.run(
["bash", "-n", str(path)],
text=True,
capture_output=True,
check=False,
env=filtered_environment(),
)
if result.returncode:
shell_errors.append(f"{path.relative_to(ROOT)}: {result.stderr.strip()}")
report.check("shell_syntax", not shell_errors, files=len(shell_files), errors=shell_errors)
parse_errors: list[str] = []
toml_files = [path for path in _source_files() if path.suffix == ".toml"]
json_files = [path for path in _source_files() if path.suffix == ".json"]
for path in toml_files:
try:
read_toml(path)
except Exception as exc:
parse_errors.append(f"{path.relative_to(ROOT)}: {type(exc).__name__}: {exc}")
for path in json_files:
try:
strict_json_loads(path.read_text(encoding="utf-8"))
except Exception as exc:
parse_errors.append(f"{path.relative_to(ROOT)}: {type(exc).__name__}: {exc}")
report.check(
"configuration_syntax",
not parse_errors,
toml_files=len(toml_files),
json_files=len(json_files),
errors=parse_errors,
)
def _check_static_safety(report: Report) -> None:
unresolved: list[str] = []
markers: list[str] = []
secrets: list[str] = []
executable_profile_files: list[str] = []
for path in _source_files():
relative = path.relative_to(ROOT)
if relative.parts and relative.parts[0] == "profiles" and os.access(path, os.X_OK):
executable_profile_files.append(relative.as_posix())
suffix = path.suffix.lower()
if suffix not in TEXT_SUFFIXES and path.name not in {
"Makefile",
".gitignore",
"LICENSE",
"VERSION",
}:
continue
try:
text = path.read_text(encoding="utf-8")
except UnicodeDecodeError:
continue
# The validator contains the literal patterns it searches for. Exclude
# this file from content-marker scans so it does not report itself.
if path.name != "validate_package.py":
if "@@" in text and not (
relative.parts and relative.parts[0] == "bin" and path.suffix == ".in"
):
unresolved.append(relative.as_posix())
if re.search(r"(?m)\b(?:TODO|FIXME|XXX|HACK)\b", text):
markers.append(relative.as_posix())
for label, pattern in SECRET_PATTERNS.items():
if pattern.search(text):
secrets.append(f"{relative.as_posix()}: {label}")
report.check("no_unresolved_placeholders", not unresolved, paths=unresolved)
report.check("no_development_markers", not markers, paths=markers)
report.check("secret_scan", not secrets, findings=secrets)
report.check(
"profile_packs_are_non_executable",
not executable_profile_files,
paths=executable_profile_files,
)
def _sandbox_environment(temporary: Path) -> dict[str, str]:
config = temporary / "config"
state = temporary / "state"
base_home = temporary / "codex-home"
for path in (
config / "profiles.d",
config / "catalog.d",
config / "tool-mcp.d",
state,
base_home,
):
path.mkdir(parents=True, exist_ok=True)
settings = (ROOT / "config" / "settings.toml").read_text(encoding="utf-8")
settings = settings.replace(
'base_codex_home = "~/.codex"', f"base_codex_home = {json.dumps(str(base_home))}"
)
(config / "settings.toml").write_text(settings, encoding="utf-8")
(config / "credentials.env").write_text("", encoding="utf-8")
return {
**filtered_environment(),
"MMO_INSTALL_ROOT": str(ROOT),
"MMO_CONFIG_ROOT": str(config),
"MMO_STATE_ROOT": str(state),
"PYTHONDONTWRITEBYTECODE": "1",
"PYTHONPATH": os.pathsep.join((str(ROOT / "libexec"), str(ROOT / "tests"))),
}
def _check_profiles_and_catalog(report: Report) -> None:
previous_environment = dict(os.environ)
try:
with tempfile.TemporaryDirectory(prefix="mmo-acceptance-") as temporary_raw:
env = _sandbox_environment(Path(temporary_raw))
os.environ.clear()
os.environ.update(env)
profiles = discover_profiles()
profile_ids = set(profiles)
profile_errors: list[str] = []
modes: set[str] = set()
native_count = 0
mcp_count = 0
hybrid_count = 0
snapshots: dict[str, str] = {}
low_trust: list[str] = []
for profile_id in sorted(profile_ids):
directory = Path(profiles[profile_id]["path"])
tree_errors = validate_profile_pack_tree(directory)
if tree_errors:
profile_errors.extend(f"{profile_id}: {item}" for item in tree_errors)
continue
try:
resolved = resolve_profile(profile_id)
snapshot = compile_profile(profile_id)
except Exception as exc:
profile_errors.append(f"{profile_id}: {type(exc).__name__}: {exc}")
continue
snapshots[profile_id] = snapshot["manifest"]["snapshot_hash"]
mode = resolved["coordination"]["orchestration"]
modes.add(mode)
native_count += len(resolved["capabilities"]["native_agents"])
mcp_count += len(resolved["capabilities"]["mcp_agents"])
hybrid_count += int(mode == "hybrid")
root = resolved["profile"]["root"]
if resolved["agents"][root]["kind"] != "root":
profile_errors.append(f"{profile_id}: configured root is not kind=root")
if not resolved.get("smoke", {}).get("tasks"):
profile_errors.append(f"{profile_id}: no smoke tasks")
for agent_id, agent in resolved["agents"].items():
if agent["trust"] == "low":
low_trust.append(f"{profile_id}:{agent_id}")
if not (
agent["permissions"] == "read-only"
and agent["verification"] == "always"
and agent["backends"] == ["mcp"]
and agent["output_contract_schema"] is not None
):
profile_errors.append(
f"{profile_id}:{agent_id}: low-trust boundary is not mechanical"
)
report.check(
"profile_inventory",
profile_ids == EXPECTED_PROFILES,
expected=sorted(EXPECTED_PROFILES),
actual=sorted(profile_ids),
)
report.check(
"profile_semantics",
not profile_errors
and modes == EXPECTED_ORCHESTRATION
and native_count > 0
and mcp_count > 0
and hybrid_count > 0
and bool(low_trust),
errors=profile_errors,
orchestration_modes=sorted(modes),
native_agents=native_count,
mcp_agents=mcp_count,
hybrid_profiles=hybrid_count,
low_trust_agents=low_trust,
snapshots=snapshots,
)
summary = catalog_summary()
inventory = local_inventory_report()
inventory_counts = {
key: value["actual_count"] for key, value in inventory["inventories"].items()
}
catalog_ok = (
inventory["passed"]
and summary["routes"] == 21
and summary["models"] == 560
and summary["agent_compatible_models"] == 462
and summary["resources"] == 9
and inventory_counts.get("opencode-go") == 29
and inventory_counts.get("opencode-zen") == 64
and inventory_counts.get("openrouter") == 422
and inventory_counts.get("zai-api") == 35
and inventory_counts.get("zai-coding-plan") == 3
and inventory_counts.get("openai-codex") == 6
)
report.check(
"complete_model_catalog",
catalog_ok,
summary=summary,
inventory=inventory,
)
suite_results = {suite: validate_suite(suite) for suite in discover_suites()}
report.check(
"evaluation_suites",
set(suite_results) == EXPECTED_EVALUATION_SUITES
and all(item["valid"] for item in suite_results.values()),
expected=sorted(EXPECTED_EVALUATION_SUITES),
suites=suite_results,
)
finally:
os.environ.clear()
os.environ.update(previous_environment)
def _check_catalog_reproducibility(report: Report) -> None:
tracked = [ROOT / "config" / "catalog.toml", ROOT / "config" / "upstream-inventory.json"]
before = _digest(tracked)
result = subprocess.run(
[sys.executable, str(ROOT / "scripts" / "generate_catalog.py"), "--check"],
cwd=ROOT,
env=filtered_environment(extra={"PYTHONDONTWRITEBYTECODE": "1"}),
text=True,
capture_output=True,
check=False,
)
after = _digest(tracked)
report.check(
"catalog_generation_is_reproducible",
result.returncode == 0 and before == after,
exit_code=result.returncode,
changed=sorted(
key for key in set(before) | set(after) if before.get(key) != after.get(key)
),
stdout=result.stdout[-4000:],
stderr=result.stderr[-4000:],
)
def _run_tests(report: Report) -> None:
started = time.monotonic()
process = subprocess.Popen(
[sys.executable, str(ROOT / "scripts" / "run_tests.py")],
cwd=ROOT,
env=filtered_environment(
extra={
"PYTHONDONTWRITEBYTECODE": "1",
"PYTHONWARNINGS": "error::ResourceWarning",
"PYTHONPATH": os.pathsep.join((str(ROOT / "libexec"), str(ROOT / "tests"))),
}
),
text=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
start_new_session=True,
)
timed_out = False
try:
stdout, stderr = process.communicate(timeout=2100)
except subprocess.TimeoutExpired:
timed_out = True
terminate_process_group(process.pid)
stdout, stderr = process.communicate()
report.check(
"offline_integration_tests",
process.returncode == 0 and not timed_out,
exit_code=process.returncode,
timed_out=timed_out,
duration_seconds=time.monotonic() - started,
stdout=stdout[-12000:],
stderr=stderr[-12000:],
)
def _parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(description="Validate a Codex MMO source/release tree")
parser.add_argument(
"--skip-tests", action="store_true", help="skip the offline integration suite"
)
parser.add_argument("--report", type=Path, help="write the full JSON report to this path")
parser.add_argument("--json", action="store_true", help="emit the full JSON report on stdout")
return parser
def main() -> int:
args = _parser().parse_args()
report = Report()
_check_structure(report)
_check_source_manifest(report)
_check_syntax(report)
_check_static_safety(report)
_check_profiles_and_catalog(report)
_check_catalog_reproducibility(report)
if not args.skip_tests:
_run_tests(report)
result = report.finish()
if args.report:
atomic_write_json(args.report, result, 0o644)
if args.json:
print(json.dumps(result, indent=2, sort_keys=True, allow_nan=False))
else:
state = "PASS" if result["passed"] else "FAIL"
print(f"Codex MMO package validation: {state}")
for name, item in result["checks"].items():
print(f" {'PASS' if item['passed'] else 'FAIL'} {name}")
for warning in result["warnings"]:
print(f" WARN {warning}")
for error in result["errors"]:
print(f" ERROR {error}")
if args.report:
print(f"Report: {args.report}")
return 0 if result["passed"] else 1
if __name__ == "__main__":
raise SystemExit(main())
Reference in New Issue Copy Permalink