v0.10: add automated Codex orchestration and deployment check

This commit is contained in:
kai
2026-04-24 13:38:22 +08:00
parent 77df1838b1
commit 450ecebcff
8 changed files with 460 additions and 39 deletions
+183
View File
@@ -0,0 +1,183 @@
#!/usr/bin/env python3
"""Check and repair a Deep Research deployment checkout."""
from __future__ import annotations
import argparse
import os
import shutil
import subprocess
import sys
from pathlib import Path
try:
import tomllib
except ModuleNotFoundError: # pragma: no cover - Python < 3.11 fallback.
tomllib = None # type: ignore[assignment]
REPO_ROOT = Path(__file__).resolve().parent.parent
CODEX_TEMPLATE = REPO_ROOT / "codex_adapter_templates" / "codex"
CODEX_ROOT = REPO_ROOT / ".codex"
OPENCODE_ROOT = REPO_ROOT / ".opencode"
AGENTS_SKILLS = REPO_ROOT / ".agents" / "skills"
OPENCODE_SKILLS = OPENCODE_ROOT / "skills"
REQUIRED_PATHS = [
"AGENTS.md",
"README.md",
"PLAN.md",
".opencode/opencode.json",
".codex/config.toml",
".codex/agents/dr-pm.toml",
".codex/commands/dr-run.md",
".agents/skills/search-strategy/SKILL.md",
"scripts/dr.py",
"scripts/install_codex_adapter.py",
]
REQUIRED_ENV_KEYS = [
"ZENMUX_API_KEY",
"TAVILY_API_KEY",
"BRAVE_API_KEY",
"EXA_API_KEY",
]
def rel(path: Path) -> str:
return str(path.relative_to(REPO_ROOT))
def copy_tree_contents(src: Path, dst: Path, *, force: bool) -> list[Path]:
written: list[Path] = []
if not src.exists():
raise RuntimeError(f"source not found: {rel(src)}")
dst.mkdir(parents=True, exist_ok=True)
for item in src.rglob("*"):
target = dst / item.relative_to(src)
if item.is_dir():
target.mkdir(parents=True, exist_ok=True)
continue
if target.exists() and not force:
continue
target.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(item, target)
written.append(target)
return written
def parse_env(path: Path) -> dict[str, str]:
values: dict[str, str] = {}
if not path.exists():
return values
for raw in path.read_text(encoding="utf-8").splitlines():
line = raw.strip()
if not line or line.startswith("#") or "=" not in line:
continue
key, value = line.split("=", 1)
values[key.strip()] = value.strip().strip('"').strip("'")
return values
def git_tracked(paths: list[str]) -> set[str]:
proc = subprocess.run(
["git", "ls-files", *paths],
cwd=REPO_ROOT,
text=True,
stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL,
check=False,
)
return set(proc.stdout.splitlines())
def check_toml(path: Path, issues: list[str]) -> None:
if not path.exists():
return
if tomllib is None:
issues.append("Python tomllib unavailable; skip TOML parse checks")
return
try:
tomllib.loads(path.read_text(encoding="utf-8"))
except Exception as exc: # noqa: BLE001 - deployment diagnostics.
issues.append(f"{rel(path)} TOML parse failed: {exc}")
def check_deployment() -> int:
issues: list[str] = []
warnings: list[str] = []
for item in REQUIRED_PATHS:
if not (REPO_ROOT / item).exists():
issues.append(f"missing required path: {item}")
tracked = git_tracked([".codex", ".opencode", ".agents/skills"])
for item in REQUIRED_PATHS:
if item.startswith((".codex/", ".opencode/", ".agents/")) and item not in tracked:
warnings.append(f"not tracked by git: {item}")
skill_files = sorted(AGENTS_SKILLS.glob("*/SKILL.md"))
if len(skill_files) < 10:
issues.append(f"expected at least 10 Codex skills, found {len(skill_files)}")
check_toml(CODEX_ROOT / "config.toml", issues)
for path in sorted((CODEX_ROOT / "agents").glob("*.toml")):
check_toml(path, issues)
env_values = {key: os.environ.get(key, "") for key in REQUIRED_ENV_KEYS}
env_values.update({k: v for k, v in parse_env(REPO_ROOT / "secrets.env").items() if not env_values.get(k)})
missing_env = [key for key in REQUIRED_ENV_KEYS if not env_values.get(key)]
if missing_env:
warnings.append("missing optional/required API keys for full automation: " + ", ".join(missing_env))
print("Deep Research deployment check")
print(f" repo: {REPO_ROOT}")
print(f" codex files tracked: {sum(1 for p in tracked if p.startswith('.codex/'))}")
print(f" opencode files tracked: {sum(1 for p in tracked if p.startswith('.opencode/'))}")
print(f" codex skills: {len(skill_files)}")
if warnings:
print("\nWarnings:")
for item in warnings:
print(f" - {item}")
if issues:
print("\nIssues:")
for item in issues:
print(f" - {item}")
return 1
print("\nDeployment check passed.")
return 0
def repair(force: bool) -> int:
try:
codex_written = copy_tree_contents(CODEX_TEMPLATE, CODEX_ROOT, force=force)
skills_written = copy_tree_contents(OPENCODE_SKILLS, AGENTS_SKILLS, force=force)
except PermissionError as exc:
print(f"repair failed: permission denied: {exc}", file=sys.stderr)
return 1
except RuntimeError as exc:
print(f"repair failed: {exc}", file=sys.stderr)
return 1
print("Repair completed.")
print(f" .codex files written: {len(codex_written)}")
print(f" .agents skills written: {len(skills_written)}")
return check_deployment()
def main() -> int:
parser = argparse.ArgumentParser(description="Check or repair Deep Research deployment files")
parser.add_argument("--repair", action="store_true", help="copy Codex templates and skills into hidden dirs")
parser.add_argument("--force", action="store_true", help="overwrite existing files during --repair")
args = parser.parse_args()
if args.repair:
return repair(force=args.force)
return check_deployment()
if __name__ == "__main__":
raise SystemExit(main())