#!/usr/bin/env python3 """Install, update, or verify Bookmark Research with a selected client's native CLI.""" import argparse import hashlib import json import os import re import shlex import shutil import subprocess import sys import tempfile from pathlib import Path from urllib.parse import urlsplit from export_bundle import NAME, SOURCE_ROOT, VERSION_PATTERN, _copy_plan, read_plugin_manifest from host_assets import read_asset SELECTOR = NAME + "@" + NAME CODEX_HOST_FILES = ( "hosts/codex/delegate.md", "hosts/codex/prepare.py", "hosts/shared/research-call.py", ) def resolve_language(language="auto", environment=None): """Choose installer copy without changing the process locale or user settings.""" if language not in ("auto", "en", "zh"): raise ValueError("--lang must be auto, en, or zh") if language != "auto": return language environment = os.environ if environment is None else environment forwarded = environment.get("BOOKMARK_RESEARCH_INSTALL_LANG") if forwarded in ("en", "zh"): return forwarded locale = next((environment.get(key) for key in ("LC_ALL", "LC_MESSAGES", "LANG") if environment.get(key)), "en") return "zh" if locale.lower().startswith("zh") else "en" def _message(language, english, chinese): return chinese if language == "zh" else english class CodexCli: def __init__(self, binary="codex", timeout=60): self.binary = binary self.timeout = timeout def run(self, arguments): from host_clients import executable command = [executable(self.binary) or self.binary, *arguments, "--json"] result = subprocess.run(command, text=True, encoding="utf-8", errors="replace", capture_output=True, timeout=self.timeout) if result.returncode: raise RuntimeError("Codex command failed: " + " ".join(arguments[:3]) + "\n" + (result.stderr or result.stdout).strip()[-4000:]) try: value = json.loads(result.stdout) except json.JSONDecodeError as error: raise RuntimeError("Codex must support plugin commands with --json; update Codex CLI") from error if not isinstance(value, dict): raise RuntimeError("Unexpected Codex JSON response") return value def _marketplace(cli): value = cli.run(["plugin", "marketplace", "list"]) entries = value.get("marketplaces") if not isinstance(entries, list): raise RuntimeError("Codex did not return a marketplaces array") matches = [row for row in entries if row.get("name") == NAME] if len(matches) > 1: raise ValueError("Multiple bookmark-research marketplaces are in scope; resolve their names first") return matches[0] if matches else None def _installed(cli): value = cli.run(["plugin", "list"]) entries = value.get("installed") if not isinstance(entries, list): raise RuntimeError("Codex did not return an installed plugins array") return next((row for row in entries if row.get("pluginId") == SELECTOR), None) def _repository(value): """Normalize public GitHub shorthand while preserving other Git sources.""" if re.fullmatch(r"[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+", value): return "https://github.com/" + value.removesuffix(".git") + ".git" if value.startswith(("https://", "http://", "ssh://")): parsed = urlsplit(value) if parsed.password or parsed.query or parsed.fragment or (parsed.username and parsed.scheme != "ssh"): raise ValueError("Use a Git URL without credentials, query parameters, or fragments") if not parsed.hostname or not parsed.path.strip("/"): raise ValueError("Invalid Git repository URL") return value.rstrip("/").removesuffix(".git") + ".git" if re.fullmatch(r"git@[^\s:]+:[^\s]+", value): return value.removesuffix(".git") + ".git" raise ValueError("Source must be a local marketplace root, owner/repo, or a Git URL") def _source(value, ref=None): from npm_source import source as npm_source release = npm_source(value, ref) if release: return release value = str(value) if not value or any(ord(char) < 32 for char in value): raise ValueError("Source must be a nonempty path or Git repository") path = Path(value).expanduser() if path.exists() or value.startswith(("/", "./", "../", "~", ".\\", "..\\")): if ref: raise ValueError("--ref only applies to a Git marketplace; checkout local sources yourself") root = path.resolve() manifest = read_plugin_manifest(root) _copy_plan(root) for relative in CODEX_HOST_FILES: read_asset(root, relative) marketplace_path = root / ".agents/plugins/marketplace.json" marketplace = json.loads(marketplace_path.read_text(encoding="utf-8")) if not isinstance(marketplace, dict) or marketplace.get("name") != NAME: raise ValueError("The local marketplace must be named bookmark-research") plugin = next((entry for entry in marketplace.get("plugins", []) if entry.get("name") == NAME), None) # Repository catalogs distribute the published npm package. The source # installer deliberately exports this checkout; exported bundles keep # a local catalog so native installation of that bundle stays offline. source = plugin.get("source") if plugin else None # Source versions can advance before publication; the catalog keeps its # previous release until the new npm version is available. version = source.get("version") if isinstance(source, dict) else None published = {"source": "npm", "package": NAME, "version": version, "registry": "https://registry.npmjs.org"} pinned_release = isinstance(version, str) and VERSION_PATTERN.fullmatch(version) and source == published if source != {"source": "local", "path": "./"} and not pinned_release: raise ValueError("The marketplace must select its local root or a pinned bookmark-research npm release") return {"sourceType": "local", "source": str(root), "version": manifest["version"]} if ref and (ref.startswith("-") or not re.fullmatch(r"[A-Za-z0-9_./-]+", ref)): raise ValueError("--ref must be a Git branch, tag, or commit identifier") return {"sourceType": "git", "source": _repository(value), "ref": ref} def _configured_source(marketplace): source = marketplace.get("marketplaceSource") if source is None: return {"sourceType": "local", "source": str(Path(marketplace["root"]).resolve())} if not isinstance(source, dict) or source.get("sourceType") not in ("local", "git", "npm"): raise RuntimeError("Unsupported Codex marketplace source metadata") return source def _check_source(marketplace, requested): current = _configured_source(marketplace) if current["sourceType"] == requested["sourceType"] == "local": equal = Path(current["source"]).resolve() == Path(requested["source"]).resolve() elif current["sourceType"] == requested["sourceType"] == "git": equal = _repository(current["source"]) == requested["source"] elif current["sourceType"] == requested["sourceType"] == "npm": equal = current["source"] == requested["source"] and ( current == requested or requested.get("ref") != "latest" ) else: equal = False if not equal: raise ValueError("Marketplace bookmark-research already uses a different source: " + current["source"] + ". Keep it and run update, or explicitly resolve the source in Codex first.") def _cached_path(installed): version = installed.get("version") if not isinstance(version, str) or not re.fullmatch(r"[A-Za-z0-9.+_-]+", version): raise RuntimeError("Codex did not return a valid installed version") profile = Path(os.environ.get("CODEX_HOME") or Path.home() / ".codex").expanduser() return (profile / "plugins/cache" / NAME / NAME / version).resolve() def _runtime_check(root, timeout=30): manifest = read_plugin_manifest(root) for relative in CODEX_HOST_FILES: read_asset(root, relative) command = [sys.executable, "-B", str(root / "src/cli.py"), "doctor"] doctor = subprocess.run(command, cwd=root.parent, text=True, encoding="utf-8", errors="replace", capture_output=True, timeout=timeout) if doctor.returncode: raise RuntimeError("Installed runtime doctor failed: " + doctor.stderr.strip()[-2000:]) health = json.loads(doctor.stdout) if health.get("fts5") is not True: raise RuntimeError("Installed runtime requires SQLite FTS5") server = manifest.get("mcpServers", {}).get(NAME, {}) command = server.get("command") # Windows installs pin the detected interpreter; the repository manifest keeps python3. pinned = isinstance(command, str) and Path(command).is_absolute() and Path(command).is_file() if (command != "python3" and not pinned) or server.get("args") != ["src/cli.py", "serve"] or server.get("cwd") != ".": raise ValueError("Unexpected native MCP launch configuration; verify it with Codex directly") requests = [ {"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": { "protocolVersion": "2025-11-25", "capabilities": {}, "clientInfo": {"name": "bookmark-research-installer", "version": manifest["version"]}}}, {"jsonrpc": "2.0", "method": "notifications/initialized"}, {"jsonrpc": "2.0", "id": 2, "method": "tools/list"}, ] # Startup now monitors registered live sources. Tool discovery during an # installation check must use an empty store, not start syncing user data. with tempfile.TemporaryDirectory(prefix="bookmark-runtime-check-") as probe_data: environment = dict(os.environ, PYTHONDONTWRITEBYTECODE="1", BOOKMARK_RESEARCH_DATA_DIR=probe_data) mcp = subprocess.run([server["command"], *server["args"]], cwd=root, input="\n".join(json.dumps(row) for row in requests) + "\n", text=True, encoding="utf-8", errors="replace", capture_output=True, timeout=timeout, env=environment) if (Path(probe_data) / "index.sqlite3").exists(): raise RuntimeError("MCP discovery unexpectedly created a database") if mcp.returncode: raise RuntimeError("Installed MCP startup failed: " + mcp.stderr.strip()[-2000:]) responses = {row.get("id"): row for row in (json.loads(line) for line in mcp.stdout.splitlines())} tools = responses.get(2, {}).get("result", {}).get("tools", []) names = sorted(row["name"] for row in tools) if not {"search_bookmarks", "search_web", "fetch_web"} <= set(names): raise RuntimeError("Installed MCP did not expose the expected bookmark and research tools") if "result" not in responses.get(1, {}): raise RuntimeError("Installed MCP initialization failed") return {"doctor": health, "mcp_protocol": responses[1]["result"]["protocolVersion"], "mcp_tools": names, "network_checked": False, "database_created": False} def verify(cli, installed_path=None): installed = _installed(cli) if not installed or not installed.get("installed") or not installed.get("enabled"): raise ValueError("bookmark-research is not installed and enabled in this Codex profile") root = Path(installed_path).expanduser().resolve() if installed_path else _cached_path(installed) if not root.is_dir(): raise ValueError("Installed cache not found; pass --installed-path from codex plugin add --json") manifest = read_plugin_manifest(root) if manifest["version"] != installed.get("version"): raise ValueError("Installed cache version differs from Codex plugin list") runtime = _runtime_check(root, timeout=cli.timeout) return {"plugin": SELECTOR, "version": manifest["version"], "installed_path": str(root), "installed": True, "enabled": True, "verified": True, "runtime": runtime} def _getting_started(installed_path, timeout=30, language="auto"): """Read the installed runtime's preferences without initializing user data.""" language = resolve_language(language) message = lambda english, chinese: _message(language, english, chinese) command = [sys.executable, "-B", str(Path(installed_path) / "src/cli.py"), "config", "show"] guide = { "language": language, "first_prompt": message( 'Use Bookmark Research to read my canvas package at "/absolute/path/to/my-canvas-package". ' 'First list its sections, folders, and bookmark counts offline.', '用 Bookmark Research 读取我的书签画布包 "/absolute/path/to/my-canvas-package"，先离线列出栏目、文件夹和书签数量。'), "settings_command": command, "guide_url": "https://github.com/Browser-bookmark-hub/Bookmark-Research/blob/main/docs/" + message( "installation.en.md#first-use-and-data-locations", "installation.md#首次使用与数据位置"), "configuration": None, "configuration_summary": [], "configuration_error": None, } try: checked = subprocess.run(command, text=True, encoding="utf-8", errors="replace", capture_output=True, timeout=timeout) if checked.returncode: raise ValueError("Installed config show failed") configuration = json.loads(checked.stdout) settings = configuration["settings"] summary = [ message("Settings: ", "配置：") + (message("using saved preferences", "沿用已保存的设置") if configuration["config_exists"] else message("using defaults; no config file is required", "使用默认值，无需先创建配置文件")), message("Search: %s; %s results per target", "搜索：%s；每目标 %s 条结果") % ( " + ".join(settings["search"]["providers"]), settings["search"]["limit_per_target"]), message("Page reading: %s; length parameter %s characters; timeout %s seconds", "正文读取：%s；长度参数 %s 字符；超时 %s 秒") % ( settings["fetch"]["provider"], settings["fetch"]["max_characters"], settings["timeout_seconds"]), message("Archive ordinary page reads: ", "普通网页读取自动归档：") + ( message("enabled", "开启") if settings["archive"]["enabled"] else message("disabled", "关闭")), message("Archive directory: ", "归档目录：") + settings["archive"]["directory"], message("Config file: ", "配置文件：") + configuration["config_path"], ] guide.update(configuration=configuration, configuration_summary=summary) except (OSError, ValueError, KeyError, TypeError, subprocess.TimeoutExpired): # Installation is already verified. Keep an unreadable existing config # intact, and avoid exposing its contents or replacing it with defaults. guide["configuration_error"] = message( "Could not read existing settings. Run the config command below to diagnose and fix the file; the original was preserved.", "现有配置读取失败。请运行下方配置查询命令查看原因并修正文件；原文件已保留。") return guide def _print_getting_started(guide): """Print only installation-specific settings, leaving usage guides in README.""" language = resolve_language(guide.get("language", "auto")) message = lambda english, chinese: _message(language, english, chinese) configuration = guide.get("configuration") or {} if configuration.get("config_path"): print(message("Config file: ", "配置文件：") + configuration["config_path"]) archive = configuration.get("settings", {}).get("archive", {}) if archive.get("directory"): print(message("Archive directory: ", "归档目录：") + archive["directory"]) if guide.get("configuration_error"): print(guide["configuration_error"]) print(message("Inspect settings: ", "检查配置：") + shlex.join(guide["settings_command"])) def _print_install_summary(action, results, setup, language): """A short terminal receipt; --json retains the full diagnostic result.""" message = lambda english, chinese: _message(language, english, chinese) for row in results: host = row.get("host") or row.get("getting_started", {}).get("host", "codex") if row.get("error"): print(message("Failed: ", "失败：") + host + " — " + row["error"]) continue label = message("Updated", "更新完成") if action == "update" else message("Installed", "安装完成") print("Bookmark Research " + label + " · " + host) print(message("Version: ", "版本：") + row.get("version", "unknown")) if row.get("installed_path"): print(message("Plugin directory: ", "安装目录：") + row["installed_path"]) if row.get("verified"): print(message("Local runtime checks: passed", "本地运行检查：通过")) guide = row.get("getting_started") if guide: _print_getting_started(guide) else: doctor = row.get("runtime", {}).get("doctor", {}) if doctor.get("config_path"): print(message("Config file: ", "配置文件：") + doctor["config_path"]) if host == "codex": print(message("Start a new Codex thread to load this build.", "在 Codex 新建对话以加载此构建。")) else: print(message("Restart %s to load this build.", "重启 %s 以加载此构建。") % host) if setup is not None: if setup.get("configuration_retained"): print(message("Existing settings and credentials retained.", "已沿用原有配置与密钥。")) credentials = setup.get("credentials", {}).get("credentials") if isinstance(credentials, list): configured = [row["name"] for row in credentials if row.get("configured")] print(message("API keys configured: ", "已配置的 API Key：") + (", ".join(configured) if configured else message("none (optional)", "无（可选）"))) readiness = setup.get("readiness") if isinstance(readiness, dict): print(message("Network checks: performed", "网络检查：已执行") if readiness.get("network_checked") else message("Network checks: not performed this time", "网络检查：本次未执行")) checks = {row["id"]: row for row in (readiness or {}).get("checks", [])} for name in setup.get("needs_attention", []): check = checks.get(name, {}) status = check.get("status") detail = message("missing credentials", "缺少密钥") if status == "missing_credentials" else message("check failed", "检查失败") if check.get("error_kind"): detail += " (" + check["error_kind"] + ")" print(message("Needs attention: ", "需处理：") + name + " — " + detail) # A provider can verify fetching while its search operation fails. # Keep that partial failure visible even when the overall check passes. for name, check in checks.items(): for operation, outcome in check.get("operations", {}).items(): if outcome.get("status") == "failed": label = message(operation, {"search": "搜索", "fetch": "正文读取"}.get(operation, operation)) detail = outcome.get("error_kind") or message("check failed", "检查失败") print(message("Needs attention: ", "需处理：") + name + " / " + label + " — " + detail) if setup.get("completed") is not True: detail = setup.get("error") or setup.get("message") if setup.get("status") == "unsupported" and setup.get("completed") is None: print(detail) else: print(message("Setup incomplete", "配置未完成") + (": " + detail if detail else "")) first = next((row for row in results if row.get("verified") and not row.get("error")), None) if first and (setup or {}).get("status") != "unsupported": host = first.get("host") or first.get("getting_started", {}).get("host", "codex") command = [sys.executable, "-B", str(Path(first["installed_path"]) / "src/cli.py"), "setup", "--host", _setup_host(host), "--lang", language] print(message("Configure preferences / API keys: ", "修改配置／API Key：") + shlex.join(command)) def _pin_interpreter(): """Windows has no reliable python3 command, so Codex gets this interpreter's absolute path.""" return os.name == "nt" def _pinned_home(): data = Path(os.environ.get("XDG_DATA_HOME", str(Path.home() / ".local/share"))).expanduser() root = os.environ.get("BOOKMARK_RESEARCH_INSTALL_DIR") or str(data / "bookmark-research/installations") return Path(root).expanduser().resolve() / "codex-pinned" def _local_home(): profile = Path(os.environ.get("CODEX_HOME") or Path.home() / ".codex").expanduser().resolve() return _pinned_home().parent / ("codex-local-" + hashlib.sha256(str(profile).encode()).hexdigest()[:16]) def _export_source(origin, timeout, python, base, dry_run=False): """Prepare a clean, persistent marketplace; Codex owns registration and caching.""" from export_bundle import export_bundle from host_install import _checkout, _write_json bundle = base / "bundle" if origin["sourceType"] == "local": root = Path(origin["source"]).resolve() if base == root or root in base.parents or base in root.parents: raise ValueError("The managed installation directory must be separate from the source checkout") if dry_run: return {"sourceType": "local", "source": str(bundle)} base.mkdir(parents=True, exist_ok=True) if bundle.is_symlink(): raise ValueError("Managed bundle must not be a symlink: " + str(bundle)) with tempfile.TemporaryDirectory(prefix=".bookmark-stage-", dir=base) as temporary: stage, previous = Path(temporary) / "bundle", Path(temporary) / "previous" with _checkout(origin, timeout) as (root, _): export_bundle("codex", stage, root, python) _runtime_check(stage, timeout) if bundle.exists(): bundle.rename(previous) try: stage.rename(bundle) _write_json(base / "origin.json", origin) except BaseException: if bundle.exists(): shutil.rmtree(bundle) if previous.exists(): previous.rename(bundle) raise return _source(str(bundle)) def _export_origin(requested): if requested["sourceType"] == "local": for base in (_pinned_home(), _local_home()): if Path(requested["source"]) == base / "bundle": return json.loads((base / "origin.json").read_text(encoding="utf-8")) return None def _published_source(root): """Recognize this repository's release catalog without reading host settings.""" catalog = json.loads((Path(root) / ".agents/plugins/marketplace.json").read_text(encoding="utf-8")) plugin = next((row for row in catalog.get("plugins", []) if row.get("name") == NAME), {}) source = plugin.get("source", {}) if isinstance(source, dict) and source.get("source") == "npm": return source return None def manage(action, cli, source=None, ref=None, dry_run=False, installed_path=None, language="auto", replace_source=False): language = resolve_language(language) if action == "verify": return verify(cli, installed_path) current = _marketplace(cli) registered = _configured_source(current) if current else None origin = _export_origin(registered) if registered else None if action == "install": requested = _source(source if source is not None else SOURCE_ROOT, ref) requested = _export_origin(requested) or requested if current and not replace_source: _check_source({"marketplaceSource": origin or registered}, requested) if ref and (origin is None or ref != origin.get("ref")): # Native list JSON omits the registered ref. Equal commit IDs # cannot distinguish a moving branch from an immutable tag. raise ValueError("An existing Git marketplace retains its registered ref. Reinstall without --ref, or explicitly change the source in Codex first") if origin and ref is None and requested["sourceType"] == "git": if origin.get("sourceType") == "git" and requested["source"] == origin["source"] and origin.get("ref"): requested["ref"] = origin["ref"] else: if not current or not _installed(cli): raise ValueError("No existing bookmark-research installation; run install first") requested = origin or registered commands, migration = [], False legacy_git = registered is not None and registered["sourceType"] == "git" and not replace_source native_release = current and not origin and action == "update" and _published_source(current["root"]) if native_release: # Preserve native npm installations as releases; update must not replace # a published package with unpublished code from the catalog repository. origin = registered base = _local_home() if legacy_git: commands.append(["plugin", "marketplace", "upgrade", NAME]) commands.append(["plugin", "add", SELECTOR]) elif legacy_git: # Codex list omits the registered ref. Let the host keep and fetch it; # override only this plugin-add invocation with our filtered export. origin = registered base = _local_home() commands.append(["plugin", "marketplace", "upgrade", NAME]) requested = {"sourceType": "local", "source": str(base / "bundle")} commands.append(["plugin", "add", SELECTOR, "-c", "marketplaces.bookmark-research.source_type=\"local\"", "-c", "marketplaces.bookmark-research.source=" + json.dumps(requested["source"])]) else: if requested["sourceType"] == "local": requested = _source(requested["source"]) origin = {key: requested[key] for key in ("sourceType", "source", "ref") if key in requested} base = (Path(registered["source"]).parent if registered and _export_origin(registered) else _local_home()) migration = bool(registered and registered != {"sourceType": "local", "source": str(base / "bundle")}) if migration: # Only migrate this plugin's single-entry local catalog. catalog = json.loads((Path(current["root"]) / ".agents/plugins/marketplace.json").read_text(encoding="utf-8")) if len(catalog.get("plugins", [])) != 1 or catalog["plugins"][0].get("name") != NAME: raise ValueError("Cannot migrate a marketplace containing other plugins; resolve its source in Codex first") commands.append(["plugin", "marketplace", "remove", NAME]) requested = _export_source(origin, cli.timeout, sys.executable if _pin_interpreter() else "python3", base=base, dry_run=dry_run) if action == "install" or migration: commands.append(["plugin", "marketplace", "add", requested["source"]]) commands.append(["plugin", "add", SELECTOR]) plan = {"action": action, "source": requested, "dry_run": dry_run, "language": language, "commands": [[cli.binary, *command, "--json"] for command in commands]} plan["origin"] = origin if dry_run: return plan added = None for command in commands: try: result = cli.run(command) except (RuntimeError, subprocess.TimeoutExpired) as error: if migration and command[:3] == ["plugin", "marketplace", "add"]: try: cli.run(["plugin", "marketplace", "add", registered["source"]]) except (RuntimeError, subprocess.TimeoutExpired) as restore_error: raise RuntimeError(str(error) + "\nCould not restore the original marketplace: " + str(restore_error)) from error raise if command[:3] == ["plugin", "marketplace", "add"]: if result.get("marketplaceName") != NAME: raise RuntimeError("The selected source registered a different marketplace; bookmark-research was not installed") elif command[:3] == ["plugin", "marketplace", "upgrade"]: if result.get("errors"): raise RuntimeError("Marketplace upgrade reported errors; the plugin was not reinstalled") if _published_source(current["root"]): # A legacy root catalog can become a release catalog on upgrade. # Its npm entry already selects the filtered distribution. requested = registered commands[-1] = ["plugin", "add", SELECTOR] else: # Older pinned catalogs still need a filtered copy of their root. native_release = None requested = _export_source({"sourceType": "local", "source": current["root"]}, cli.timeout, sys.executable if _pin_interpreter() else "python3", base) commands[-1] = ["plugin", "add", SELECTOR, "-c", "marketplaces.bookmark-research.source_type=\"local\"", "-c", "marketplaces.bookmark-research.source=" + json.dumps(requested["source"])] plan["source"] = requested plan["commands"] = [[cli.binary, *args, "--json"] for args in commands] elif command[:2] == ["plugin", "add"]: added = result if not added or not isinstance(added.get("installedPath"), str) or added.get("pluginId") != SELECTOR: raise RuntimeError("Codex did not confirm the expected installed plugin path") verified = verify(cli, added["installedPath"]) if requested.get("version") and verified["version"] != requested["version"]: raise RuntimeError("Installed plugin version differs from the selected local source") if requested["sourceType"] == "local" and not native_release: source_root = Path(requested["source"]) cached_root = Path(verified["installed_path"]) for relative in [Path(".codex-plugin/plugin.json"), *_copy_plan(source_root, include_codex_metadata=True)]: if (source_root / relative).read_bytes() != (cached_root / relative).read_bytes(): raise RuntimeError("Installed cache differs from the selected source: " + relative.as_posix()) for relative in CODEX_HOST_FILES: if read_asset(source_root, relative) != read_asset(cached_root, relative): raise RuntimeError("Installed cache differs from the selected source: " + relative) expected = {p.relative_to(source_root) for p in source_root.rglob("*") if p.is_file()} actual = {p.relative_to(cached_root) for p in cached_root.rglob("*") if p.is_file()} if actual != expected: raise RuntimeError("Installed cache file list differs from the clean export: " + ", ".join(sorted(str(p) for p in actual ^ expected))) result = {**plan, **verified, "next_step": _message(language, "Start a new Codex thread to load the updated Skill and MCP tools.", "在 Codex 新建对话以加载更新后的 Skill 和 MCP 工具。")} if action == "install": result["getting_started"] = _getting_started(verified["installed_path"], cli.timeout, language) return result HOSTS = ("codex", "claude", "dsh") def _setup_host(host): return "claude_code" if host == "claude" else host def _normalize_hosts(values): """Accept repeated and comma-separated --host values; dedupe in order.""" if not values: return None hosts = [] for value in values: for name in str(value).split(","): name = name.strip() if not name: continue if name not in HOSTS: raise ValueError("--host must be codex, claude, or dsh: " + name) if name not in hosts: hosts.append(name) if not hosts: raise ValueError("--host requires at least one of codex, claude, or dsh") return hosts def _targets_from_args(args): hosts = args.host or ["codex"] others = [host for host in hosts if host != "codex"] if hosts == ["codex"]: if any((args.scope, args.project, args.profile, args.install_dir, args.claude, args.dsh)): raise ValueError("Host-specific options require --host claude or dsh") if others and getattr(args, "installed_path", None): raise ValueError("--installed-path is only supported for Codex") if len(hosts) == 1: # A single host keeps its historical option handling and native errors. return [{"host": hosts[0], "scope": args.scope, "project": args.project, "profile": args.profile}] # CLI paths for unselected hosts are harmless locators and are ignored. if (args.scope or args.project) and not {"claude"} & set(hosts): raise ValueError("--scope and --project require --host claude") if args.profile and "dsh" not in hosts: raise ValueError("--profile is only supported for DeepSeek Harness (dsh)") if args.install_dir and not others: raise ValueError("Host-specific options require --host claude or dsh") return [{"host": host, "scope": args.scope if host == "claude" else None, "project": args.project if host == "claude" else None, "profile": args.profile if host == "dsh" else None} for host in hosts] def _run_target(args, target, language): host = target["host"] if host == "codex": return manage(args.action, CodexCli(args.codex, args.timeout), source=getattr(args, "source", None), ref=getattr(args, "ref", None), dry_run=getattr(args, "dry_run", False), installed_path=getattr(args, "installed_path", None), language=language, replace_source=getattr(args, "replace_source", False)) from host_install import manage_host if getattr(args, "installed_path", None): raise ValueError("--installed-path is only supported for Codex") result = manage_host(args.action, host, binary=getattr(args, host), timeout=args.timeout, source=getattr(args, "source", None), ref=getattr(args, "ref", None), dry_run=getattr(args, "dry_run", False), scope=target.get("scope"), project=target.get("project"), profile=target.get("profile"), install_dir=args.install_dir) result["language"] = language return result def _run_setup(args, installed, host, language, guided): from settings import Settings existing_config = Settings().path.is_file() interactive = guided and (getattr(args, "interaction", "auto") == "always" or not existing_config) command = [sys.executable, "-B", str(installed / "src/cli.py"), "setup", "--lang", language, "--host", _setup_host(host), "--interactive" if interactive else "--non-interactive"] if args.preferences: command.extend(["--input", args.preferences]) if args.skip_checks: command.append("--skip-checks") if args.test_retrieval: command.append("--test-retrieval") if not (installed / "src/onboarding.py").is_file(): requested = interactive or bool(args.preferences) or args.test_retrieval return {"completed": False if requested else None, "status": "unsupported", "message": _message(language, "This version does not support guided setup; install a newer version to configure it.", "此版本不支持配置向导；请安装较新版本后配置。")} # The installed version owns its settings schema. Do not use the # bootstrap checkout's runtime or impose a timeout on human input. completed = subprocess.run(command, text=True, encoding="utf-8", errors="replace", stdout=subprocess.PIPE) try: setup = json.loads(completed.stdout) if not isinstance(setup, dict): raise ValueError("Setup result must be an object") except ValueError: setup = {"error": _message(language, "Setup did not return a valid result", "配置向导未返回有效结果")} setup["completed"] = completed.returncode == 0 and "error" not in setup setup["exit_code"] = completed.returncode setup["configuration_retained"] = existing_config and not interactive and not args.preferences return setup def _guide(installed, host, timeout, language): guide = {**_getting_started(installed, timeout, language), "host": host} if (installed / "src/onboarding.py").is_file(): guide["setup_command"] = ["python3", str(installed / "src/cli.py"), "setup", "--host", _setup_host(host), "--lang", language] return guide def main(argv=None): sys.path.insert(0, str(SOURCE_ROOT / "src")) from textio import utf8_stdio utf8_stdio() language_parser = argparse.ArgumentParser(add_help=False) language_parser.add_argument("--lang", choices=("auto", "en", "zh"), default="auto") preliminary, _ = language_parser.parse_known_args(argv) language = resolve_language(preliminary.lang) message = lambda english, chinese: _message(language, english, chinese) parser = argparse.ArgumentParser(description=message(__doc__, "通过所选宿主的原生 CLI 安装、更新或验证 Bookmark Research。")) language_help = message("Installer language; auto follows LC_ALL, LC_MESSAGES, then LANG", "安装器语言；auto 依次读取 LC_ALL、LC_MESSAGES、LANG") parser.add_argument("--lang", choices=("auto", "en", "zh"), default=argparse.SUPPRESS, help=language_help) subparsers = parser.add_subparsers(dest="action", required=True) for action in ("install", "update", "verify"): command = subparsers.add_parser(action) command.add_argument("--lang", choices=("auto", "en", "zh"), default=argparse.SUPPRESS, help=language_help) command.add_argument("--json", action="store_true", help=message( "Print the full installation result as JSON", "以 JSON 输出完整安装结果")) command.add_argument("--host", action="append", help=message( "Target client: codex, claude, or dsh; repeat or comma-separate for several. Guided selection in a terminal, otherwise codex", "目标宿主：codex、claude 或 dsh；可重复或用逗号指定多个。终端中引导选择，非交互时默认 codex")) command.add_argument("--codex", default="codex", help=message("Codex CLI executable, optionally an absolute path", "Codex CLI 程序名或绝对路径")) for host in ("claude", "dsh"): detail = (" (default: a usable dsh on PATH, then a DeepSeek Harness Desktop installation)", "（默认依次使用 PATH 中可用的 dsh 与 DeepSeek Harness 桌面版自带的 CLI）") if host == "dsh" else ("", "") command.add_argument("--" + host, help=message("Path to the " + host + " CLI" + detail[0], host + " CLI 路径" + detail[1])) command.add_argument("--scope", choices=("user", "project", "local"), help=message( "Claude scope; default user (local is Claude-only)", "Claude 作用域，默认 user；local 仅限 Claude")) command.add_argument("--project", help=message("Project directory for project/local scope", "project/local 作用域对应的项目目录")) command.add_argument("--profile", help=message("Required DeepSeek Harness (dsh) profile name", "DeepSeek Harness (dsh) 必填的 profile 名称")) command.add_argument("--install-dir", help=message("Managed export/receipt root for non-Codex hosts", "非 Codex 宿主的持久包与安装记录根目录")) command.add_argument("--timeout", type=int, default=60, help=message("Timeout for each native command in seconds (1-300)", "每条原生命令的超时秒数（1–300）")) if action == "install": interaction = command.add_mutually_exclusive_group() interaction.add_argument("--interactive", dest="interaction", action="store_const", const="always", help=message("Always open the terminal wizard", "始终打开终端向导")) interaction.add_argument("--non-interactive", dest="interaction", action="store_const", const="never", help=message("Do not prompt; add --json for machine-readable output", "不询问；需要机器可读结果时添加 --json")) command.set_defaults(interaction="auto") command.add_argument("--preferences", help=message("Preferences JSON file (no credentials)", "偏好 JSON 文件（不含密钥）")) command.add_argument("--skip-checks", action="store_true", help=message("Skip network readiness checks", "跳过联网就绪检查")) command.add_argument("--test-retrieval", action="store_true", help=message("Test sample search/read; uses retrieval quota", "实际搜索与阅读自检；使用检索额度")) command.add_argument("--source", help=message("npm:bookmark-research[@VERSION], local source, or Git repository; Python entry defaults to this source tree", "npm:bookmark-research[@VERSION]、本地源码或 Git 仓库；Python 入口默认当前源码目录")) command.add_argument("--replace-source", action="store_true", help=message( "Replace this plugin's existing Codex source with the selected build", "用所选构建替换此插件在 Codex 中登记的安装来源")) command.add_argument("--ref", help=message("Optional Git tag, branch, or commit; local sources must be checked out separately", "可选 Git tag、分支或提交；本地来源需自行 checkout")) if action == "verify": command.add_argument("--installed-path", help=message("Override the native cache path returned by codex plugin add --json", "指定 codex plugin add --json 返回的安装缓存路径")) else: command.add_argument("--dry-run", action="store_true", help=message("Read current registration and print commands without installing", "读取登记状态并预览命令，不执行安装")) args = parser.parse_args(argv) if not 1 <= args.timeout <= 300: parser.error(message("--timeout must be between 1 and 300", "--timeout 必须在 1 到 300 之间")) try: args.host = _normalize_hosts(args.host) guided = False targets = None if args.action == "install": if args.skip_checks and args.test_retrieval: raise ValueError("--skip-checks cannot be combined with --test-retrieval") sys.path.insert(0, str(SOURCE_ROOT / "src")) if args.preferences: from settings import Settings if args.preferences == "-": raise ValueError("--preferences requires a file path; use cli.py setup --input - for stdin") path = Path(args.preferences).expanduser().resolve() if path.stat().st_size > 1024 * 1024: raise ValueError("Preferences file exceeds the size limit") changes = json.loads(path.read_text(encoding="utf-8"), object_pairs_hook=Settings._unique_object) if not isinstance(changes, dict) or not changes: raise ValueError("Preferences must be a nonempty JSON object without credentials") # Validate before registering a plugin or changing user settings. Settings._validate(Settings._merge(Settings().load(), changes)) args.preferences = str(path) if not args.dry_run: from onboarding import Console with Console.open(args.interaction, language) as console: if console: guided = True targets = console.install_targets(args) if targets is None: targets = _targets_from_args(args) single = len(targets) == 1 results = [] for target in targets: try: results.append(_run_target(args, target, language)) except (OSError, ValueError, RuntimeError, subprocess.TimeoutExpired) as error: if single: raise results.append({"host": target["host"], "error": str(error), "verified": False}) if args.json: print(message("Warning: Operation failed for " + target["host"] + ": " + str(error), "警告：操作失败，宿主 " + target["host"] + "：" + str(error)), file=sys.stderr) verified = [(target["host"], result) for target, result in zip(targets, results) if args.action == "install" and result.get("verified") and "error" not in result] setup = None if verified: first_host, first = verified[0] setup = _run_setup(args, Path(first["installed_path"]), first_host, language, guided) if args.action in ("install", "update") and not args.dry_run: for target, result in zip(targets, results): if not result.get("verified") or result.get("error"): continue host = target["host"] result["getting_started"] = _guide(Path(result["installed_path"]), host, args.timeout, language) if single: result = results[0] if setup is not None: result["setup"] = setup else: result = {"results": results, "language": language} if setup is not None: result["setup"] = setup if args.action == "install" and getattr(args, "dry_run", False) and args.preferences: result["preferences_file"] = str(Path(args.preferences).expanduser().resolve()) result["preferences_applied"] = False except KeyboardInterrupt: print(json.dumps({"cancelled": True, "message": message( "Setup cancelled; completed installation steps are retained.", "已取消；已完成的安装步骤予以保留。")}, ensure_ascii=False), file=sys.stderr) return 130 except (OSError, ValueError, RuntimeError, subprocess.TimeoutExpired) as error: print(json.dumps({"error": str(error), "verified": False}, ensure_ascii=False), file=sys.stderr) return 1 if args.json or args.action == "verify" or getattr(args, "dry_run", False): print(json.dumps(result, ensure_ascii=False, indent=2), flush=True) else: _print_install_summary(args.action, results, setup, language) if (setup or {}).get("exit_code") == 130: return 130 if any("error" in row for row in results): return 1 return 1 if (setup or {}).get("completed") is False else 0 if __name__ == "__main__": raise SystemExit(main())