439 lines
18 KiB
Python
439 lines
18 KiB
Python
"""Check an existing checkout and synchronize the local Codex MCP configuration."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import hashlib
|
||
import json
|
||
import os
|
||
from pathlib import Path
|
||
import re
|
||
import shutil
|
||
import subprocess
|
||
import sys
|
||
import tempfile
|
||
import tomllib
|
||
from urllib.parse import urlsplit
|
||
|
||
ROOT = Path(__file__).resolve().parents[1]
|
||
|
||
|
||
class McpSyncValidationError(ValueError):
|
||
"""The synchronizer could not safely update an already parsed configuration."""
|
||
|
||
|
||
def launch_text(english: str, chinese: str) -> str:
|
||
return chinese if os.environ.get("ARTEMIS_LAUNCH_LANG") == "zh-CN" else english
|
||
|
||
|
||
def git(root: Path, *args: str, timeout: int = 15) -> str:
|
||
env = {**os.environ, "GIT_TERMINAL_PROMPT": "0", "GCM_INTERACTIVE": "Never"}
|
||
result = subprocess.run(
|
||
["git", "-C", str(root), *args],
|
||
capture_output=True,
|
||
text=True,
|
||
encoding="utf-8",
|
||
errors="replace",
|
||
env=env,
|
||
timeout=timeout,
|
||
)
|
||
if result.returncode:
|
||
# Remote diagnostics can contain URLs/userinfo. Keep them out of launcher logs.
|
||
raise RuntimeError(
|
||
f"git {args[0]} failed (exit {result.returncode}); check network, credentials or repository state in Git."
|
||
)
|
||
return result.stdout.strip()
|
||
|
||
|
||
def update_source(root: Path, *, check_only: bool = False) -> int:
|
||
"""Return 10 for an available/applied fast-forward, 1 on failure; preserve local work."""
|
||
if not shutil.which("git"):
|
||
print(
|
||
launch_text(
|
||
"[ERROR] Install Git before upgrading; local source is preserved.",
|
||
"[提示] 未安装 Git，无法升级；当前源码保持不变。",
|
||
)
|
||
)
|
||
return 1
|
||
if not (root / ".git").exists():
|
||
print(
|
||
launch_text(
|
||
"[SKIP] This older source ZIP has no Git metadata. Use a new release ZIP or Git clone for automatic updates; local files are preserved.",
|
||
"[跳过] 旧版源码包不含 Git 信息。请使用新版源码包或 Git 克隆；当前文件保持不变。",
|
||
)
|
||
)
|
||
return 1
|
||
try:
|
||
if not check_only and git(root, "status", "--porcelain"):
|
||
print(
|
||
launch_text(
|
||
"[ERROR] Local changes exist; commit or back up first.",
|
||
"[错误] 发现本地修改；请先提交或备份后再升级。",
|
||
)
|
||
)
|
||
return 1
|
||
branch = git(root, "symbolic-ref", "--quiet", "--short", "HEAD")
|
||
remote = git(root, "config", "--get", f"branch.{branch}.remote")
|
||
merge_ref = git(root, "config", "--get", f"branch.{branch}.merge")
|
||
if remote == "." or remote.startswith("-") or not merge_ref.startswith("refs/heads/"):
|
||
raise RuntimeError(
|
||
"No remote tracking branch; configure an upstream before enabling source updates."
|
||
)
|
||
url = git(root, "remote", "get-url", remote)
|
||
parsed = urlsplit(url)
|
||
if parsed.scheme in {"http", "https"} and (parsed.username or parsed.password):
|
||
raise RuntimeError(
|
||
"Use a clean Git remote URL and a credential manager, not embedded credentials."
|
||
)
|
||
print(
|
||
launch_text(
|
||
"[CHECK] Checking the current branch's configured upstream...",
|
||
"[检查] 正在检查当前分支的远程更新...",
|
||
)
|
||
)
|
||
git(root, "fetch", "--no-tags", remote, merge_ref, timeout=5 if check_only else 45)
|
||
target = git(root, "rev-parse", "FETCH_HEAD")
|
||
current = git(root, "rev-parse", "HEAD")
|
||
if current == target:
|
||
print(launch_text("[OK] Source is up to date.", "[完成] 源码已是最新版。"))
|
||
return 0
|
||
if check_only:
|
||
print(
|
||
launch_text(
|
||
"[UPDATE] A different upstream revision is available. Run update.bat (update-cn.bat for China mirrors).",
|
||
"[更新] 远程版本有变化；请运行 update-cn.bat 或 update.bat 升级。",
|
||
)
|
||
)
|
||
return 10
|
||
if git(root, "status", "--porcelain"):
|
||
print(
|
||
launch_text(
|
||
"[ERROR] Local changes appeared during the check. Upgrade stopped; files preserved.",
|
||
"[错误] 检查期间出现本地修改；已停止升级，保留现有文件。",
|
||
)
|
||
)
|
||
return 1
|
||
ancestor = git(root, "merge-base", "HEAD", target)
|
||
if ancestor != current:
|
||
print(
|
||
launch_text(
|
||
"[ERROR] Local branch is ahead or diverged. Upgrade stopped; no merge/rebase performed.",
|
||
"[错误] 本地分支领先或分叉；已停止升级，不自动合并。",
|
||
)
|
||
)
|
||
return 1
|
||
git(root, "merge", "--ff-only", target, timeout=30)
|
||
print(
|
||
launch_text(
|
||
f"[UPDATED] Source {current[:12]} -> {target[:12]}. Preparing the updated installation.",
|
||
f"[已更新] 源码 {current[:12]} -> {target[:12]}，准备同步新版环境。",
|
||
)
|
||
)
|
||
return 10
|
||
except (OSError, RuntimeError, subprocess.TimeoutExpired) as exc:
|
||
print(
|
||
launch_text(
|
||
f"[WARN] {exc if isinstance(exc, RuntimeError) else 'Git check failed or timed out.'} Local source is preserved.",
|
||
"[提示] Git 检查/更新失败或超时；请检查网络、凭据和仓库状态。当前源码保持不变。",
|
||
)
|
||
)
|
||
return 1
|
||
|
||
|
||
def _atomic_write(path: Path, text: str) -> None:
|
||
if path.is_file() and path.read_text(encoding="utf-8") == text:
|
||
return
|
||
backup = path.with_name(path.name + ".artemis-update.bak")
|
||
if path.is_file() and not backup.exists():
|
||
shutil.copy2(path, backup)
|
||
fd, name = tempfile.mkstemp(prefix=".artemis-update-", dir=path.parent)
|
||
try:
|
||
with os.fdopen(fd, "w", encoding="utf-8", newline="\n") as stream:
|
||
stream.write(text)
|
||
os.replace(name, path)
|
||
finally:
|
||
Path(name).unlink(missing_ok=True)
|
||
|
||
|
||
def _is_standard_local_artemis(config: dict) -> bool:
|
||
if config.get("url") or config.get("type") not in (None, "stdio"):
|
||
return False
|
||
args = config.get("args", [])
|
||
# Running this launcher selects this checkout, including after a user moves
|
||
# a release ZIP to a new directory. Do not take over custom wrappers/SSE.
|
||
return args in (["-m", "mcp_server"], ["-m", "mcp_server.server"])
|
||
|
||
|
||
def _updated_entry(config: dict, template: dict, root: Path, names: tuple[str, ...]) -> dict:
|
||
updated = {**config, "command": template["command"], "args": template["args"]}
|
||
if "cwd" in template or "cwd" in config:
|
||
updated["cwd"] = str(root)
|
||
environment = config.get("env", {})
|
||
if not isinstance(environment, dict):
|
||
raise ValueError("The existing Artemis MCP environment is not an object")
|
||
updated["env"] = {**environment, **template["env"]}
|
||
# Earlier releases made Artemis mandatory for every Codex session. Migrate
|
||
# that setting while preserving enabled=false and all tool approval policies.
|
||
updated["required"] = False
|
||
if "enabled_tools" in template or "enabled_tools" in config:
|
||
updated["enabled_tools"] = list(names)
|
||
# Per-tool approval/enablement policies belong to the user. Leave `tools`
|
||
# untouched, including entries outside the current exported tool list.
|
||
# Rebuilding it both changes policy and sends nested tables to the scalar
|
||
# serializer, which produces JSON syntax rather than valid TOML.
|
||
for key in ("type", "required", "startup_timeout_sec", "connectionTimeoutMs"):
|
||
if key in template and key not in updated:
|
||
updated[key] = template[key]
|
||
return updated
|
||
|
||
|
||
def _toml_value(value: object) -> str:
|
||
if isinstance(value, bool):
|
||
return "true" if value else "false"
|
||
return json.dumps(value, ensure_ascii=False)
|
||
|
||
|
||
def _toml_inline_table(values: dict) -> str:
|
||
return "{ " + ", ".join(f"{key} = {_toml_value(value)}" for key, value in values.items()) + " }"
|
||
|
||
|
||
def _replace_toml_tables(raw: str, assignments: dict[str, dict]) -> str:
|
||
"""Update only Artemis TOML tables; preserve every unrelated section and setting."""
|
||
lines = raw.splitlines(keepends=True)
|
||
output: list[str] = []
|
||
section = ""
|
||
found: dict[str, set[str]] = {name: set() for name in assignments}
|
||
seen: set[str] = set()
|
||
|
||
def append_missing() -> None:
|
||
if section in assignments:
|
||
for key, value in assignments[section].items():
|
||
if key not in found[section]:
|
||
rendered = _toml_inline_table(value) if key == "env" else _toml_value(value)
|
||
output.append(f"{key} = {rendered}\n")
|
||
found[section].add(key)
|
||
|
||
index = 0
|
||
while index < len(lines):
|
||
line = lines[index]
|
||
if line.lstrip().startswith("["):
|
||
append_missing()
|
||
header = re.sub(r"[\s\"']", "", line.split("#", 1)[0]).strip()
|
||
section = header[1:-1] if header.startswith("[") and header.endswith("]") else ""
|
||
seen.add(section)
|
||
match = re.match(r"\s*(\w+)\s*=", line) if section in assignments else None
|
||
if match and match[1] in assignments[section]:
|
||
key = match[1]
|
||
original = line
|
||
while True:
|
||
try:
|
||
tomllib.loads(original)
|
||
break
|
||
except tomllib.TOMLDecodeError:
|
||
index += 1
|
||
if index >= len(lines):
|
||
raise ValueError("Cannot safely parse existing MCP setting")
|
||
original += lines[index]
|
||
value = assignments[section][key]
|
||
rendered = _toml_inline_table(value) if key == "env" else _toml_value(value)
|
||
output.append(f"{key} = {rendered}\n")
|
||
found[section].add(key)
|
||
else:
|
||
output.append(line)
|
||
index += 1
|
||
append_missing()
|
||
for name, values in assignments.items():
|
||
if name not in seen:
|
||
if output and not output[-1].endswith("\n"):
|
||
output.append("\n")
|
||
output.append(f"\n[{name}]\n")
|
||
for key, value in values.items():
|
||
rendered = _toml_inline_table(value) if key == "env" else _toml_value(value)
|
||
output.append(f"{key} = {rendered}\n")
|
||
return "".join(output)
|
||
|
||
|
||
def refresh_codex(
|
||
path: Path,
|
||
root: Path,
|
||
python: str,
|
||
names: tuple[str, ...],
|
||
template: dict | None = None,
|
||
install_missing: bool = False,
|
||
) -> bool | None:
|
||
if not path.is_file() and not install_missing:
|
||
return False
|
||
raw = path.read_text(encoding="utf-8") if path.is_file() else ""
|
||
parsed = tomllib.loads(raw)
|
||
config = parsed.get("mcp_servers", {}).get("artemis")
|
||
if config is None and install_missing:
|
||
if template is None:
|
||
raise ValueError("A standard MCP template is required for installation")
|
||
updated = template
|
||
elif isinstance(config, dict) and _is_standard_local_artemis(config):
|
||
if template is None:
|
||
template = {
|
||
"command": python,
|
||
"args": ["-m", "mcp_server"],
|
||
"cwd": str(root),
|
||
"env": {"PYTHONPATH": str(root)},
|
||
"enabled_tools": list(names),
|
||
}
|
||
updated = _updated_entry(config, template, root, names)
|
||
else:
|
||
return None
|
||
top = {
|
||
key: value
|
||
for key, value in updated.items()
|
||
if key != "env" and (not config or config.get(key) != value)
|
||
}
|
||
environment = updated["env"]
|
||
if config and config.get("env") == environment:
|
||
env_updates = {}
|
||
else:
|
||
env_updates = {
|
||
key: value
|
||
for key, value in environment.items()
|
||
if not config or config.get("env", {}).get(key) != value
|
||
}
|
||
env_header = "mcp_servers.artemis.env"
|
||
has_env_table = re.search(r"(?m)^\s*\[mcp_servers\.artemis\.env\]\s*(?:#.*)?$", raw) is not None
|
||
if config and "env" in config and not has_env_table and env_updates:
|
||
top["env"] = environment
|
||
env_updates = {}
|
||
assignments = {"mcp_servers.artemis": top}
|
||
if env_updates:
|
||
assignments[env_header] = env_updates
|
||
if not top and not env_updates:
|
||
return False
|
||
try:
|
||
result = _replace_toml_tables(raw, assignments)
|
||
actual = tomllib.loads(result)["mcp_servers"]["artemis"]
|
||
if any(actual.get(key) != value for key, value in top.items()) or any(
|
||
actual.get("env", {}).get(key) != value for key, value in env_updates.items()
|
||
):
|
||
raise ValueError("Updated MCP settings did not match the requested values")
|
||
except (ValueError, KeyError, TypeError) as exc:
|
||
raise McpSyncValidationError(
|
||
"Could not safely generate the MCP configuration update"
|
||
) from exc
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
_atomic_write(path, result)
|
||
return result != raw
|
||
|
||
|
||
def refresh_mcp(root: Path) -> int:
|
||
"""Synchronize only Codex; never inspect or modify other clients."""
|
||
sys.path.insert(0, str(root))
|
||
from artemis.interfaces.cli.commands.mcp import ARTEMIS_MCP_TOOLS, _get_config_snippet
|
||
|
||
codex_home = Path(os.environ.get("CODEX_HOME") or (Path.home() / ".codex"))
|
||
path = codex_home / "config.toml"
|
||
print(launch_text(f"[CHECK] Codex MCP configuration: {path}", f"[检查] Codex MCP 配置：{path}"))
|
||
if not path.is_file() and not codex_home.is_dir():
|
||
print(
|
||
launch_text(
|
||
"[SKIP] Codex configuration directory not found. Install with: uv run artemis mcp --install codex",
|
||
"[跳过] 未发现 Codex 配置目录。可执行 uv run artemis mcp --install codex 安装。",
|
||
)
|
||
)
|
||
return 0
|
||
existed = path.is_file()
|
||
template = _get_config_snippet("codex", sys.executable, str(root))["mcp_servers"]["artemis"]
|
||
try:
|
||
changed = refresh_codex(
|
||
path, root, sys.executable, ARTEMIS_MCP_TOOLS, template=template, install_missing=True
|
||
)
|
||
except (OSError, ValueError, TypeError, AttributeError) as exc:
|
||
if isinstance(exc, OSError):
|
||
reason = launch_text(
|
||
"Cannot read/write the configuration; check file permissions.",
|
||
"无法读写配置，请检查文件权限。",
|
||
)
|
||
elif isinstance(exc, McpSyncValidationError):
|
||
reason = launch_text(
|
||
"The synchronizer generated an invalid update from a readable configuration. This is an ARTEMIS sync error; do not delete your Codex configuration.",
|
||
"原配置可正常解析，但同步程序生成的更新未通过校验。这是 ARTEMIS 同步错误，请勿删除 Codex 配置。",
|
||
)
|
||
else:
|
||
reason = launch_text(
|
||
"Invalid TOML or Artemis configuration structure; check the file in Codex.",
|
||
"TOML 格式或 ARTEMIS 配置结构无效，请在 Codex 中检查配置文件。",
|
||
)
|
||
print(
|
||
launch_text(
|
||
f"[ERROR] Cannot safely synchronize {path}; original preserved. {reason}",
|
||
f"[错误] 无法安全同步 {path}；原配置已保留。{reason}",
|
||
)
|
||
)
|
||
return 1
|
||
if changed:
|
||
print(
|
||
launch_text(
|
||
f"[{'UPDATED' if existed else 'INSTALLED'}] Codex MCP: {path}",
|
||
f"[{'已更新' if existed else '已安装'}] Codex MCP：{path}",
|
||
)
|
||
)
|
||
elif changed is None:
|
||
print(
|
||
launch_text(
|
||
"[SKIP] Custom Codex Artemis MCP command preserved; check it manually.",
|
||
"[跳过] Codex 使用自定义 ARTEMIS MCP 启动方式，已保留，请手动检查。",
|
||
)
|
||
)
|
||
return 0
|
||
else:
|
||
print(
|
||
launch_text(
|
||
"[OK] Codex Artemis MCP is already current.", "[完成] Codex ARTEMIS MCP 已是最新。"
|
||
)
|
||
)
|
||
print(
|
||
launch_text(
|
||
"[ACTION] Reload MCP or restart Codex to activate the updated tools.",
|
||
"[操作] 请在 Codex 中重新加载 MCP 或重启 Codex，使新版进程和工具签名生效。",
|
||
)
|
||
)
|
||
return 0
|
||
|
||
|
||
def frontend_fingerprint(root: Path) -> str:
|
||
ui = root / "apps/showcase_ui"
|
||
files = list((ui / "src").rglob("*")) + list((ui / "public").rglob("*"))
|
||
files += [
|
||
ui / name
|
||
for name in (
|
||
"package.json",
|
||
"package-lock.json",
|
||
"angular.json",
|
||
"tsconfig.json",
|
||
"tsconfig.app.json",
|
||
)
|
||
]
|
||
digest = hashlib.sha256()
|
||
for path in sorted(p for p in files if p.is_file()):
|
||
digest.update(path.relative_to(ui).as_posix().encode())
|
||
digest.update(path.read_bytes())
|
||
return digest.hexdigest()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
parser = argparse.ArgumentParser(description=__doc__)
|
||
parser.add_argument(
|
||
"action", choices=["source", "source-check", "mcp", "frontend-check", "frontend-mark"]
|
||
)
|
||
parser.add_argument("--root", type=Path, default=ROOT)
|
||
args = parser.parse_args()
|
||
if args.action in {"source", "source-check"}:
|
||
sys.exit(update_source(args.root, check_only=args.action == "source-check"))
|
||
elif args.action == "mcp":
|
||
sys.exit(refresh_mcp(args.root))
|
||
else:
|
||
marker = args.root / "apps/showcase_ui/node_modules/.artemis-source-hash"
|
||
current = frontend_fingerprint(args.root)
|
||
if args.action == "frontend-mark":
|
||
marker.parent.mkdir(parents=True, exist_ok=True)
|
||
marker.write_text(current, encoding="ascii")
|
||
elif not marker.is_file() or marker.read_text(encoding="ascii").strip() != current:
|
||
sys.exit(10)
|