chenmingxuan
e1ea099592
Install and update artemis-use with task diagnostics and plugin guidance. Preserve cross-device file receipts, review canonical reports, guard stage transitions and reduce small history summaries. Improve exploration lifecycle, request recovery and browser live connections.
274 lines
11 KiB
Python
274 lines
11 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]
|
||
|
||
|
||
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 refresh_codex(
|
||
path: Path,
|
||
root: Path,
|
||
python: str,
|
||
names: tuple[str, ...],
|
||
template: dict | None = None,
|
||
install_missing: bool = False,
|
||
) -> bool | None:
|
||
# The frozen source updater also runs before any project dependencies exist.
|
||
# Only the MCP action may import installed application code.
|
||
if str(ROOT) not in sys.path:
|
||
sys.path.insert(0, str(ROOT))
|
||
from artemis.integrations.codex_mcp import refresh_codex as synchronize
|
||
|
||
return synchronize(path, root, python, names, template, install_missing)
|
||
|
||
|
||
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
|
||
|
||
from artemis.integrations.codex_skill import sync_skill_safely
|
||
from artemis.integrations.codex_mcp import McpSyncValidationError
|
||
|
||
sync_skill_safely(root)
|
||
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)
|