diff --git a/README.md b/README.md index 8a3009f..391d7d8 100644 --- a/README.md +++ b/README.md @@ -32,14 +32,15 @@ Our Zenith method keeps the useful parts of repeated review while making the loo Copy this prompt into Claude Code or Codex: ```text -/goal Read the readme at https://github.com/Intelligent-Internet/zenith, detect if using Claude Code, Codex, or both, install requirements, install and run Zenith (i.e. uv run zenith, as in the readme), and create a new skill called /zenith — when used (along with an additional prompt) it will call the skill: the minimum skill content should be: """First read .claude/orchestrator_prompt.md and treat it as your primary role, then use Zenith to run this mission.""" Afterwards, you can add information about the Zenith harness, based on the readme and the technical report (inside the repo), and info on how to start Zenith if it's not already running. Change the skill to use .codex when using it in Codex. If both harnesses are available, make sure to add the skill to both of them correctly. In fact, there might be more harness options (Hermes, for example). See what is supported in zenith/src/zenith_harness/providers.py, and for those that you detect are present, add their skills correctly. When finished, confirm to me that Zenith is installed, running, and ready, explain a bit about Zenith, and why and when to use it. +/goal Read the README at https://github.com/Intelligent-Internet/zenith, install its requirements and ACP adapter, then run `uv run zenith init --scope user --agent claude` for Claude Code or `uv run zenith init --scope user --agent codex` for Codex. If both hosts are installed, run both commands. Confirm that the user-scoped MCP server and `/zenith` skill were installed, and tell me to start a new host session before using `/zenith `. ``` This will: - install Zenith with its requirements - start Zenith using `uv` -- create a `/zenith` skill for each agent harness it detects +- register Zenith for every workspace in each detected host +- create a personal `/zenith` skill and install Zenith's agent assets Then, in Claude Code or Codex, type: @@ -76,19 +77,49 @@ npm install -g @agentclientprotocol/codex-acp command -v codex-acp ``` -**Initialize a workspace** +**Install for every workspace (recommended)** -Initialize the project workspace Zenith should operate on. This is your target app/repo, not the Zenith source checkout: +Run user-scope setup from the Zenith checkout: ```bash -# Claude Code, from this Zenith checkout -uv run zenith init --workspace-dir /path/to/your-app --agent claude +# Claude Code +uv run zenith init --scope user --agent claude -# Or Codex, from this Zenith checkout -uv run zenith init --workspace-dir /path/to/your-app --agent codex +# Codex +uv run zenith init --scope user --agent codex + +# If you use both, run both commands in either order. +``` + +User scope writes only Zenith's MCP registration and managed assets. It preserves existing +Codex/Claude preferences and does not select a model, reasoning effort, sandbox mode, or +persist ambient API/model environment variables. `CODEX_HOME` and `CLAUDE_CONFIG_DIR` are +honored when set. + +The MCP command contains the absolute path to this checkout. If you move or remove the +checkout, rerun the corresponding user-scope command from its new location. User scope is +currently supported for Claude Code and Codex; use project scope for Hermes. + +Restart the host or start a new session once after installation. You can then open any +workspace and run: + +```text +/zenith ``` -**Run a mission** +**Install for one project instead** + +Project scope remains the default and writes host configuration into the target workspace: + +```bash +uv run zenith init --scope project --workspace-dir /path/to/your-app --agent claude +uv run zenith init --scope project --workspace-dir /path/to/your-app --agent codex +uv run zenith init --scope project --workspace-dir /path/to/your-app --agent hermes +``` + +`--scope project` may be omitted for backward compatibility. + +**Run a project-scoped mission** Start your agent from the initialized project workspace: diff --git a/zenith/README.md b/zenith/README.md index 3a8d06d..af9c572 100644 --- a/zenith/README.md +++ b/zenith/README.md @@ -31,28 +31,44 @@ npm install -g @agentclientprotocol/codex-acp command -v codex-acp ``` -Initialize the project workspace Zenith should operate on. This is your target -app/repo, not the Zenith source checkout: +Install Zenith once for every workspace in your user account: ```bash -# Claude Code, from this Zenith checkout -uv run zenith init --workspace-dir /path/to/your-app --agent claude +# Claude Code +uv run zenith init --scope user --agent claude -# Or Codex, from this Zenith checkout -uv run zenith init --workspace-dir /path/to/your-app --agent codex +# Codex +uv run zenith init --scope user --agent codex + +# Run both commands if you use both hosts. ``` -Start your agent from the initialized project workspace: +This registers the user-scoped MCP server and installs a personal `/zenith` +skill, orchestrator prompt, agents, and playbooks. Existing model, reasoning, +sandbox, feature, and unrelated MCP settings are preserved. Ambient API/model +environment variables are not copied into the user configuration. -```bash -cd /path/to/your-app +`CODEX_HOME` and `CLAUDE_CONFIG_DIR` are honored. The generated MCP command +contains the absolute path to this checkout, so rerun setup after moving it. +Hermes currently supports project scope only. + +Restart Claude Code or Codex once, then use Zenith from any workspace: + +```text +/zenith +``` -claude -# or -codex +For repository-specific setup, initialize the target app/repo instead: + +```bash +uv run zenith init --scope project --workspace-dir /path/to/your-app --agent claude +uv run zenith init --scope project --workspace-dir /path/to/your-app --agent codex +uv run zenith init --scope project --workspace-dir /path/to/your-app --agent hermes ``` -Then ask the agent to read the generated orchestrator prompt: +Project scope is the backward-compatible default, so `--scope project` may be +omitted. Start the host in that workspace and ask it to read the generated +orchestrator prompt: ```text First read .claude/orchestrator_prompt.md and treat it as your primary role, then use Zenith to run this mission. diff --git a/zenith/src/zenith_harness/cli.py b/zenith/src/zenith_harness/cli.py index f4cd9f4..b52e201 100644 --- a/zenith/src/zenith_harness/cli.py +++ b/zenith/src/zenith_harness/cli.py @@ -3,6 +3,9 @@ import json import os import shutil +import stat +import tempfile +import tomllib from pathlib import Path import click @@ -38,6 +41,8 @@ "ZAI_BASE_URL", ) +USER_SCOPE_ORCHESTRATORS = ("claude", "codex") + @click.group() def cli() -> None: @@ -72,7 +77,14 @@ def cli() -> None: @click.option("--terminal-reviewer-provider", type=click.Choice(provider_names_for_role("worker")), default=None) @click.option("--terminal-reviewer-acp-command", default=None) @click.option("--zenith-home", type=click.Path(), default=None) -@click.option("--workspace-dir", "workspace_dir", type=click.Path(exists=True), default=".") +@click.option( + "--scope", + type=click.Choice(("project", "user")), + default="project", + show_default=True, + help="Install into one project or the current user's host configuration.", +) +@click.option("--workspace-dir", "workspace_dir", type=click.Path(exists=True), default=None) def init( agent: str | None, orchestrator_provider: str | None, @@ -83,17 +95,16 @@ def init( terminal_reviewer_provider: str | None, terminal_reviewer_acp_command: str | None, zenith_home: str | None, - workspace_dir: str, + scope: str, + workspace_dir: str | None, ) -> None: - """Initialize host-agent surface: MCP/codex config + provider agents + orchestrator prompt. + """Initialize Zenith's host-agent surface. - The project bucket is created by `start_project` at the first MCP call from - the host agent — `zenith init` stages the host-agent-facing files the agent - loads at startup (provider configs, subagent definitions, orchestrator - prompt, and the bundled skill surface under the provider's skill dirs). The - workspace stays clean of `.zenith/` until `start_project` runs. + Project scope (the default) stages MCP config and assets in one workspace. + User scope registers Zenith and installs its assets once for every workspace + in Claude Code or Codex. In both scopes, the project bucket is created lazily + by `start_project` at the first MCP call. """ - workspace = Path(workspace_dir).resolve() config = HarnessConfig.discover() loader = AssetLoader(config) selection = _resolve_selection( @@ -107,6 +118,27 @@ def init( terminal_reviewer_acp_command=terminal_reviewer_acp_command, ) + if scope == "user": + if workspace_dir is not None: + raise click.UsageError("--workspace-dir cannot be used with --scope user") + if selection.orchestrator.name not in USER_SCOPE_ORCHESTRATORS: + supported = ", ".join(USER_SCOPE_ORCHESTRATORS) + raise click.UsageError( + f"--scope user supports these orchestrators: {supported}; " + f"use --scope project for {selection.orchestrator.name}" + ) + storage_env = _storage_env( + zenith_home=zenith_home, + workspace=Path.cwd(), + selection=selection, + ) + _write_user_bootstrap_config(selection, storage_env) + _setup_user_provider_assets(loader, selection.orchestrator) + _echo_user_next_steps(selection) + return + + workspace = Path(workspace_dir or ".").resolve() + # 1) MCP / Codex config storage_env = _storage_env(zenith_home=zenith_home, workspace=workspace, selection=selection) _write_bootstrap_config(workspace, selection, storage_env) @@ -260,6 +292,21 @@ def _copy_skills(loader: AssetLoader, target: Path) -> None: shutil.copy2(skill_dir / "SKILL.md", dest / "SKILL.md") +def _echo_user_next_steps(selection: ProviderSelection) -> None: + provider = selection.orchestrator.name + host = {"claude": "Claude Code", "codex": "Codex"}.get(provider, provider) + click.echo( + f"\nInitialized v5 user scope: orchestrator={provider}, " + f"worker={selection.worker.name}, " + f"validator={selection.resolved_validation_worker.name}." + ) + click.echo("Zenith is available from every workspace for this user.") + click.echo("") + click.echo("Next:") + click.echo(f" 1. Restart {host} or start a new session.") + click.echo(" 2. Run: /zenith ") + + def _echo_next_steps(orchestrator: ProviderDefinition) -> None: prompt_path = orchestrator.orchestrator_prompt_output_path click.echo("") @@ -349,6 +396,255 @@ def _mcp_server_args() -> list[str]: ] +def _claude_user_paths() -> tuple[Path, Path]: + configured = os.environ.get("CLAUDE_CONFIG_DIR") + if configured: + root = Path(configured).expanduser().resolve() + return root, root / ".claude.json" + home = Path.home().resolve() + return home / ".claude", home / ".claude.json" + + +def _codex_user_paths() -> tuple[Path, Path]: + root = Path(os.environ.get("CODEX_HOME") or (Path.home() / ".codex")) + root = root.expanduser().resolve() + return root, root / "config.toml" + + +def _user_paths(provider: ProviderDefinition) -> tuple[Path, Path]: + if provider.name == "claude": + return _claude_user_paths() + if provider.name == "codex": + return _codex_user_paths() + raise ValueError(f"user-scope paths are not defined for {provider.name}") + + +def _write_text_atomic(path: Path, text: str) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + previous_mode = stat.S_IMODE(path.stat().st_mode) if path.exists() else None + temp_path: Path | None = None + try: + with tempfile.NamedTemporaryFile( + mode="w", + encoding="utf-8", + dir=path.parent, + prefix=f".{path.name}.zenith-", + delete=False, + ) as handle: + handle.write(text) + temp_path = Path(handle.name) + if previous_mode is not None: + temp_path.chmod(previous_mode) + os.replace(temp_path, path) + finally: + if temp_path is not None and temp_path.exists(): + temp_path.unlink() + + +def _user_server_config(selection: ProviderSelection, storage_env: dict[str, str]) -> dict: + return { + "type": "stdio", + "command": "uv", + "args": _mcp_server_args(), + "env": {**selection.env(), **storage_env}, + } + + +def _write_claude_user_config( + path: Path, + selection: ProviderSelection, + storage_env: dict[str, str], +) -> None: + if path.exists(): + try: + existing = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + raise click.ClickException(f"Cannot update invalid Claude config {path}: {exc}") from exc + else: + existing = {} + if not isinstance(existing, dict): + raise click.ClickException(f"Cannot update Claude config {path}: root must be an object") + mcp_servers = existing.setdefault("mcpServers", {}) + if not isinstance(mcp_servers, dict): + raise click.ClickException( + f"Cannot update Claude config {path}: mcpServers must be an object" + ) + mcp_servers["zenith"] = _user_server_config(selection, storage_env) + _write_text_atomic(path, json.dumps(existing, indent=2) + "\n") + click.echo(f"Wrote {path}") + + +def _toml_string(value: str) -> str: + return json.dumps(value, ensure_ascii=False) + + +def _toml_table_path(line: str) -> tuple[str, ...] | None: + stripped = line.strip() + if not stripped.startswith("[") or stripped.startswith("[["): + return None + try: + parsed = tomllib.loads(f"{stripped}\n") + except tomllib.TOMLDecodeError: + return None + path: list[str] = [] + current = parsed + while len(current) == 1: + key, value = next(iter(current.items())) + if not isinstance(value, dict): + return None + path.append(key) + current = value + return tuple(path) if not current else None + + +def _strip_toml_tables(text: str, table: tuple[str, ...]) -> str: + kept: list[str] = [] + removing = False + for line in text.splitlines(keepends=True): + path = _toml_table_path(line) + if path is not None: + removing = path[: len(table)] == table + if not removing: + kept.append(line) + return "".join(kept).rstrip() + + +def _parse_managed_block_lines( + text: str, + start: str, + end: str, +) -> tuple[list[str], tuple[int, int] | None]: + lines = text.splitlines(keepends=True) + start_lines: list[int] = [] + end_lines: list[int] = [] + for index, line in enumerate(lines): + stripped = line.strip() + if stripped == start: + start_lines.append(index) + elif stripped == end: + end_lines.append(index) + + if not start_lines and not end_lines: + return lines, None + if len(start_lines) != 1 or len(end_lines) != 1 or start_lines[0] >= end_lines[0]: + raise ValueError( + "malformed Zenith managed block: expected no markers or one forward pair; " + f"found {len(start_lines)} begin and {len(end_lines)} end markers" + ) + return lines, (start_lines[0], end_lines[0]) + + +def _write_codex_user_config( + path: Path, + selection: ProviderSelection, + storage_env: dict[str, str], +) -> None: + existing = path.read_text(encoding="utf-8") if path.exists() else "" + try: + tomllib.loads(existing) + except tomllib.TOMLDecodeError as exc: + raise click.ClickException(f"Cannot update invalid Codex config {path}: {exc}") from exc + + start = "# BEGIN zenith" + end = "# END zenith" + try: + existing_lines, managed_span = _parse_managed_block_lines(existing, start, end) + except ValueError as exc: + raise click.ClickException(f"Cannot update Codex config {path}: {exc}") from exc + if managed_span is None: + existing = _strip_toml_tables(existing, ("mcp_servers", "zenith")) + + server = _user_server_config(selection, storage_env) + env_lines = "\n".join( + f"{key} = {_toml_string(value)}" for key, value in server["env"].items() + ) + block = ( + f"{start}\n" + "[mcp_servers.zenith]\n" + 'command = "uv"\n' + f"args = {json.dumps(server['args'], ensure_ascii=False)}\n" + "startup_timeout_sec = 10\n" + "tool_timeout_sec = 1000000\n" + "\n" + "[mcp_servers.zenith.env]\n" + f"{env_lines}\n" + f"{end}\n" + ) + if managed_span is None: + updated = existing.rstrip() + if updated: + updated += "\n\n" + updated += block.rstrip() + "\n" + else: + start_line, end_line = managed_span + updated = ( + "".join(existing_lines[:start_line]) + + block + + "".join(existing_lines[end_line + 1 :]) + ) + try: + tomllib.loads(updated) + except tomllib.TOMLDecodeError as exc: + raise click.ClickException(f"Generated invalid Codex config for {path}: {exc}") from exc + _write_text_atomic(path, updated) + click.echo(f"Wrote {path}") + + +def _write_user_bootstrap_config( + selection: ProviderSelection, + storage_env: dict[str, str], +) -> None: + _, config_path = _user_paths(selection.orchestrator) + if selection.orchestrator.name == "claude": + _write_claude_user_config(config_path, selection, storage_env) + elif selection.orchestrator.name == "codex": + _write_codex_user_config(config_path, selection, storage_env) + else: + raise ValueError(f"unsupported user-scope provider: {selection.orchestrator.name}") + + +def _zenith_skill_body(prompt_path: Path) -> str: + return f'''--- +name: zenith +description: Run a long-horizon mission through the Zenith continuous-improvement harness. +--- + +# /zenith + +First read `{prompt_path}` and treat it as your primary role, then use the globally +registered Zenith MCP tools to run the mission supplied with this skill. + +If the Zenith tools are unavailable, ask the user to restart the host or start a new +session. Do not run workspace initialization merely because the current workspace has +no local Zenith configuration; project state begins with `start_project(brief, +workspace_dir)`. +''' + + +def _setup_user_provider_assets(loader: AssetLoader, provider: ProviderDefinition) -> None: + root, _ = _user_paths(provider) + agents_dir = root / "agents" + _copy_provider_agents(loader, agents_dir, provider.name) + click.echo(f"Installed {provider.name} subagents to {agents_dir}") + + skills_dir = root / "skills" + _copy_skills(loader, skills_dir) + click.echo(f"Installed bundled skills to {skills_dir}") + + shared_skills_dir = Path.home().resolve() / ".agents" / "skills" + _copy_skills(loader, shared_skills_dir) + click.echo(f"Installed bundled skills to {shared_skills_dir}") + + prompt_path = root / "orchestrator_prompt.md" + prompt = loader.load_prompt_file("orchestrator", "system_prompt.md") + _write_text_atomic(prompt_path, prompt) + click.echo(f"Wrote {prompt_path}") + + skill_path = skills_dir / "zenith" / "SKILL.md" + _write_text_atomic(skill_path, _zenith_skill_body(prompt_path)) + click.echo(f"Wrote {skill_path}") + + def _write_bootstrap_config( workspace: Path, selection: ProviderSelection, @@ -398,8 +694,7 @@ def _write_bootstrap_config( raise ValueError(f"unsupported config_format: {fmt}") -def _replace_managed_block(path: Path, start: str, end: str, block: str) -> None: - existing = path.read_text(encoding="utf-8") if path.exists() else "" +def _replace_managed_block_text(existing: str, start: str, end: str, block: str) -> str: if start in existing and end in existing: before, _, tail = existing.partition(start) _, _, after = tail.partition(end) @@ -414,6 +709,12 @@ def _replace_managed_block(path: Path, start: str, end: str, block: str) -> None if updated: updated += "\n\n" updated += block.rstrip() + "\n" + return updated + + +def _replace_managed_block(path: Path, start: str, end: str, block: str) -> None: + existing = path.read_text(encoding="utf-8") if path.exists() else "" + updated = _replace_managed_block_text(existing, start, end, block) path.write_text(updated, encoding="utf-8") diff --git a/zenith/tests/test_cli.py b/zenith/tests/test_cli.py index e5f186d..cf154ef 100644 --- a/zenith/tests/test_cli.py +++ b/zenith/tests/test_cli.py @@ -2,6 +2,9 @@ from __future__ import annotations import json +import os +import stat +import subprocess import tomllib from pathlib import Path @@ -35,6 +38,16 @@ def _expected_mcp_server_args() -> list[str]: ] +def _tree_snapshot(root: Path) -> dict[str, tuple[str, int, bytes]]: + snapshot: dict[str, tuple[str, int, bytes]] = {} + for path in (root, *sorted(root.rglob("*"))): + relative = str(path.relative_to(root)) + kind = "directory" if path.is_dir() else "file" + content = b"" if path.is_dir() else path.read_bytes() + snapshot[relative] = (kind, stat.S_IMODE(path.stat().st_mode), content) + return snapshot + + class TestInit: def test_stages_host_agent_surface_only( self, runner: CliRunner, workspace: Path, env: dict[str, str] @@ -147,6 +160,523 @@ def test_claude_init_forwards_only_allowed_model_env( assert mcp_env["ZAI_API_KEY"] == "zai-test-key" assert "DATABASE_URL" not in mcp_env + @pytest.mark.parametrize("agent", ["claude", "codex", "hermes"]) + def test_explicit_project_scope_matches_default( + self, + runner: CliRunner, + tmp_path: Path, + env: dict[str, str], + agent: str, + ) -> None: + default_workspace = tmp_path / "default" + explicit_workspace = tmp_path / "explicit" + default_workspace.mkdir() + explicit_workspace.mkdir() + + default_result = runner.invoke( + cli, + ["init", "--workspace-dir", str(default_workspace), "--agent", agent], + ) + explicit_result = runner.invoke( + cli, + [ + "init", + "--scope", + "project", + "--workspace-dir", + str(explicit_workspace), + "--agent", + agent, + ], + ) + assert default_result.exit_code == 0, default_result.output + assert explicit_result.exit_code == 0, explicit_result.output + + def files(root: Path) -> dict[str, bytes]: + return { + str(path.relative_to(root)): path.read_bytes() + for path in root.rglob("*") + if path.is_file() + } + + assert files(default_workspace) == files(explicit_workspace) + assert default_result.output.replace(str(default_workspace), "") == ( + explicit_result.output.replace(str(explicit_workspace), "") + ) + + +@pytest.fixture +def user_home(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: + home = tmp_path / "home" + home.mkdir() + monkeypatch.setenv("HOME", str(home)) + monkeypatch.delenv("CLAUDE_CONFIG_DIR", raising=False) + monkeypatch.delenv("CODEX_HOME", raising=False) + return home + + +class TestUserScopeInit: + def test_claude_installs_user_config_and_assets_without_freezing_model( + self, + runner: CliRunner, + workspace: Path, + user_home: Path, + env: dict[str, str], + monkeypatch: pytest.MonkeyPatch, + ) -> None: + config_path = user_home / ".claude.json" + config_path.write_text( + json.dumps( + { + "theme": "dark", + "projects": {"/existing": {"allowedTools": ["Read"]}}, + "mcpServers": { + "other": {"command": "other-server"}, + "zenith": {"command": "old-zenith"}, + }, + }, + indent=2, + ) + + "\n" + ) + monkeypatch.setenv("ANTHROPIC_MODEL", "must-not-be-persisted") + monkeypatch.setenv("ZAI_API_KEY", "must-not-be-persisted") + + result = runner.invoke(cli, ["init", "--scope", "user", "--agent", "claude"]) + assert result.exit_code == 0, result.output + + config = json.loads(config_path.read_text()) + assert config["theme"] == "dark" + assert config["projects"]["/existing"]["allowedTools"] == ["Read"] + assert config["mcpServers"]["other"]["command"] == "other-server" + server = config["mcpServers"]["zenith"] + assert server["command"] == "uv" + assert server["args"] == _expected_mcp_server_args() + assert server["env"]["ZENITH_ORCHESTRATOR_PROVIDER"] == "claude" + assert "ANTHROPIC_MODEL" not in server["env"] + assert "ZAI_API_KEY" not in server["env"] + + claude_root = user_home / ".claude" + assert (claude_root / "orchestrator_prompt.md").exists() + assert (claude_root / "agents" / "investigator.md").exists() + assert (claude_root / "skills" / "engineering-mission-playbook" / "SKILL.md").exists() + skill = (claude_root / "skills" / "zenith" / "SKILL.md").read_text() + assert "name: zenith" in skill + assert str(claude_root / "orchestrator_prompt.md") in skill + assert "Do not run workspace initialization" in skill + assert not (workspace / ".mcp.json").exists() + assert not (workspace / ".claude").exists() + assert "Restart Claude Code or start a new session" in result.output + + first_config = config_path.read_bytes() + first_skill = (claude_root / "skills" / "zenith" / "SKILL.md").read_bytes() + rerun = runner.invoke(cli, ["init", "--scope", "user", "--agent", "claude"]) + assert rerun.exit_code == 0, rerun.output + assert config_path.read_bytes() == first_config + assert (claude_root / "skills" / "zenith" / "SKILL.md").read_bytes() == first_skill + + def test_claude_config_dir_override_and_invalid_config_are_safe( + self, + runner: CliRunner, + user_home: Path, + env: dict[str, str], + monkeypatch: pytest.MonkeyPatch, + ) -> None: + config_root = user_home / "custom claude" + config_root.mkdir() + config_path = config_root / ".claude.json" + original = '{"mcpServers": []}\n' + config_path.write_text(original) + monkeypatch.setenv("CLAUDE_CONFIG_DIR", str(config_root)) + + result = runner.invoke(cli, ["init", "--scope", "user", "--agent", "claude"]) + assert result.exit_code != 0 + assert "mcpServers must be an object" in result.output + assert config_path.read_text() == original + assert not (config_root / "skills").exists() + assert not (config_root / "agents").exists() + + config_path.write_text("not json\n") + malformed = runner.invoke( + cli, ["init", "--scope", "user", "--agent", "claude"] + ) + assert malformed.exit_code != 0 + assert config_path.read_text() == "not json\n" + assert not (config_root / "skills").exists() + + config_path.write_text('{"theme": "light"}\n') + success = runner.invoke(cli, ["init", "--scope", "user", "--agent", "claude"]) + assert success.exit_code == 0, success.output + config = json.loads(config_path.read_text()) + assert config["theme"] == "light" + assert config["mcpServers"]["zenith"]["command"] == "uv" + assert (config_root / "skills" / "zenith" / "SKILL.md").exists() + assert (config_root / "agents" / "investigator.md").exists() + + def test_codex_preserves_preferences_and_adopts_existing_server( + self, + runner: CliRunner, + workspace: Path, + user_home: Path, + env: dict[str, str], + monkeypatch: pytest.MonkeyPatch, + ) -> None: + codex_root = user_home / "custom codex" + codex_root.mkdir() + monkeypatch.setenv("CODEX_HOME", str(codex_root)) + monkeypatch.setenv("ANTHROPIC_MODEL", "must-not-be-persisted") + config_path = codex_root / "config.toml" + config_path.write_text( + 'model = "user-model"\n' + 'model_reasoning_effort = "low"\n' + 'sandbox_mode = "workspace-write"\n' + "\n" + "[features]\n" + "memories = false\n" + "\n" + "[mcp_servers.other]\n" + 'command = "other-server"\n' + "\n" + "[mcp_servers.zenith]\n" + 'command = "old-zenith"\n' + "\n" + "[mcp_servers.zenith.env]\n" + 'OLD = "value"\n' + ) + sibling_skill = codex_root / "skills" / "personal" / "SKILL.md" + sibling_skill.parent.mkdir(parents=True) + sibling_skill.write_text("personal skill\n") + sibling_agent = codex_root / "agents" / "personal.toml" + sibling_agent.parent.mkdir(parents=True) + sibling_agent.write_text('name = "personal"\n') + + custom_home = user_home / 'state with "quotes" and ünicode' + result = runner.invoke( + cli, + [ + "init", + "--scope", + "user", + "--agent", + "codex", + "--zenith-home", + str(custom_home), + ], + ) + assert result.exit_code == 0, result.output + + text = config_path.read_text() + config = tomllib.loads(text) + assert config["model"] == "user-model" + assert config["model_reasoning_effort"] == "low" + assert config["sandbox_mode"] == "workspace-write" + assert config["features"]["memories"] is False + assert config["mcp_servers"]["other"]["command"] == "other-server" + server = config["mcp_servers"]["zenith"] + assert server["command"] == "uv" + assert server["args"] == _expected_mcp_server_args() + assert server["env"]["ZENITH_HOME"] == str(custom_home.resolve()) + assert "ANTHROPIC_MODEL" not in server["env"] + assert text.count("[mcp_servers.zenith]") == 1 + assert text.count("[mcp_servers.zenith.env]") == 1 + assert 'model = "gpt-5.5"' not in text + assert 'model_reasoning_effort = "xhigh"' not in text + + assert (codex_root / "orchestrator_prompt.md").exists() + assert (codex_root / "agents" / "investigator.toml").exists() + skill_path = codex_root / "skills" / "zenith" / "SKILL.md" + assert str(codex_root / "orchestrator_prompt.md") in skill_path.read_text() + assert not (workspace / ".codex").exists() + assert sibling_skill.read_text() == "personal skill\n" + assert sibling_agent.read_text() == 'name = "personal"\n' + + first = config_path.read_bytes() + rerun = runner.invoke( + cli, + [ + "init", + "--scope", + "user", + "--agent", + "codex", + "--zenith-home", + str(custom_home), + ], + ) + assert rerun.exit_code == 0, rerun.output + assert config_path.read_bytes() == first + + def test_codex_replaces_valid_managed_block_by_line_and_preserves_mode( + self, + runner: CliRunner, + user_home: Path, + env: dict[str, str], + ) -> None: + codex_root = user_home / ".codex" + codex_root.mkdir() + config_path = codex_root / "config.toml" + config_path.write_text( + 'model = "user-model"\n' + "\n" + "[features]\n" + "memories = false\n" + "\n" + " # BEGIN zenith \n" + "[mcp_servers.zenith]\n" + 'command = "old-zenith"\n' + "\n" + "[mcp_servers.zenith.env]\n" + 'OLD = "value"\n' + "\t# END zenith\t\n" + "\n" + "[mcp_servers.after]\n" + 'command = "after"\n' + ) + config_path.chmod(0o640) + + result = runner.invoke(cli, ["init", "--scope", "user", "--agent", "codex"]) + assert result.exit_code == 0, result.output + first = config_path.read_bytes() + config = tomllib.loads(first.decode()) + assert config["model"] == "user-model" + assert config["features"]["memories"] is False + assert config["mcp_servers"]["zenith"]["command"] == "uv" + assert config["mcp_servers"]["after"]["command"] == "after" + assert first.count(b"# BEGIN zenith") == 1 + assert first.count(b"# END zenith") == 1 + assert stat.S_IMODE(config_path.stat().st_mode) == 0o640 + + rerun = runner.invoke(cli, ["init", "--scope", "user", "--agent", "codex"]) + assert rerun.exit_code == 0, rerun.output + assert config_path.read_bytes() == first + assert stat.S_IMODE(config_path.stat().st_mode) == 0o640 + + @pytest.mark.parametrize( + "zenith_header, env_header", + [ + ('[mcp_servers."zenith"]', '[mcp_servers."zenith".env]'), + ("[mcp_servers.zenith] # old server", "[mcp_servers.zenith.env] # old env"), + ], + ) + def test_codex_adopts_equivalent_unmanaged_zenith_tables( + self, + runner: CliRunner, + user_home: Path, + env: dict[str, str], + zenith_header: str, + env_header: str, + ) -> None: + codex_root = user_home / ".codex" + codex_root.mkdir() + config_path = codex_root / "config.toml" + config_path.write_text( + 'title = "contains # BEGIN zenith but is not a boundary"\n' + "# # END zenith is part of a longer comment\n\n" + '[mcp_servers.before]\ncommand = "before"\n\n' + 'marker_text = "prefix # END zenith suffix"\n\n' + f'{zenith_header}\ncommand = "old"\n\n' + f'{env_header}\nOLD = "value"\n\n' + '[mcp_servers.after]\ncommand = "after"\n' + ) + config_path.chmod(0o600) + + result = runner.invoke(cli, ["init", "--scope", "user", "--agent", "codex"]) + assert result.exit_code == 0, result.output + text = config_path.read_text() + config = tomllib.loads(text) + assert config["title"] == "contains # BEGIN zenith but is not a boundary" + assert config["mcp_servers"]["before"]["command"] == "before" + assert config["mcp_servers"]["before"]["marker_text"] == "prefix # END zenith suffix" + assert config["mcp_servers"]["after"]["command"] == "after" + assert config["mcp_servers"]["zenith"]["command"] == "uv" + assert text.count("[mcp_servers.zenith]") == 1 + assert "# # END zenith is part of a longer comment" in text + assert stat.S_IMODE(config_path.stat().st_mode) == 0o600 + + first = config_path.read_bytes() + rerun = runner.invoke(cli, ["init", "--scope", "user", "--agent", "codex"]) + assert rerun.exit_code == 0, rerun.output + assert config_path.read_bytes() == first + assert stat.S_IMODE(config_path.stat().st_mode) == 0o600 + + def test_codex_invalid_toml_does_not_mutate_assets( + self, + runner: CliRunner, + user_home: Path, + env: dict[str, str], + monkeypatch: pytest.MonkeyPatch, + ) -> None: + codex_root = user_home / ".codex" + codex_root.mkdir() + config_path = codex_root / "config.toml" + content = "not = [valid\n" + config_path.write_text(content) + monkeypatch.setenv("CODEX_HOME", str(codex_root)) + + result = runner.invoke(cli, ["init", "--scope", "user", "--agent", "codex"]) + assert result.exit_code != 0 + assert "invalid Codex config" in result.output + assert config_path.read_text() == content + assert not (codex_root / "skills").exists() + assert not (codex_root / "agents").exists() + + @pytest.mark.parametrize( + "markers", + [ + pytest.param(("# BEGIN zenith",), id="begin-only"), + pytest.param(("# END zenith",), id="end-only"), + pytest.param(("# END zenith", "# BEGIN zenith"), id="reversed"), + pytest.param( + ("# BEGIN zenith", "# BEGIN zenith", "# END zenith"), + id="duplicate-begin", + ), + pytest.param( + ("# BEGIN zenith", "# END zenith", "# END zenith"), + id="duplicate-end", + ), + pytest.param( + ("# BEGIN zenith", "# END zenith", "# BEGIN zenith", "# END zenith"), + id="two-blocks", + ), + pytest.param( + ("# BEGIN zenith", "# BEGIN zenith", "# END zenith", "# END zenith"), + id="nested", + ), + pytest.param( + ( + "# BEGIN zenith", + "# BEGIN zenith", + "# END zenith", + "# BEGIN zenith", + "# END zenith", + "# END zenith", + ), + id="overlapping", + ), + ], + ) + def test_codex_rejects_malformed_managed_blocks_without_user_tree_mutation( + self, + runner: CliRunner, + user_home: Path, + env: dict[str, str], + monkeypatch: pytest.MonkeyPatch, + markers: tuple[str, ...], + ) -> None: + codex_root = user_home / ".codex" + codex_root.mkdir() + config_path = codex_root / "config.toml" + content = 'theme = "dark"\n' + "\n".join(markers) + "\n# keep me\n" + config_path.write_text(content) + config_path.chmod(0o640) + monkeypatch.setenv("CODEX_HOME", str(codex_root)) + existing_agent = codex_root / "agents" / "personal.toml" + existing_agent.parent.mkdir() + existing_agent.write_text('name = "personal"\n') + existing_skill = codex_root / "skills" / "personal" / "SKILL.md" + existing_skill.parent.mkdir(parents=True) + existing_skill.write_text("personal skill\n") + shared_skill = user_home / ".agents" / "skills" / "personal" / "SKILL.md" + shared_skill.parent.mkdir(parents=True) + shared_skill.write_text("shared personal skill\n") + before = _tree_snapshot(user_home) + + result = runner.invoke(cli, ["init", "--scope", "user", "--agent", "codex"]) + assert result.exit_code != 0 + assert "malformed Zenith managed block" in result.output + assert _tree_snapshot(user_home) == before + + @pytest.mark.parametrize("order", [("claude", "codex"), ("codex", "claude")]) + def test_claude_and_codex_user_installs_coexist( + self, + runner: CliRunner, + user_home: Path, + env: dict[str, str], + order: tuple[str, str], + ) -> None: + first = runner.invoke(cli, ["init", "--scope", "user", "--agent", order[0]]) + assert first.exit_code == 0, first.output + shared_root = user_home / ".agents" / "skills" + shared_before = { + str(path.relative_to(shared_root)): path.read_bytes() + for path in shared_root.rglob("*") + if path.is_file() + } + + for agent in order[1:]: + result = runner.invoke(cli, ["init", "--scope", "user", "--agent", agent]) + assert result.exit_code == 0, result.output + + claude = json.loads((user_home / ".claude.json").read_text()) + codex = tomllib.loads((user_home / ".codex" / "config.toml").read_text()) + assert claude["mcpServers"]["zenith"]["env"]["ZENITH_WORKER_PROVIDER"] == "claude" + assert codex["mcp_servers"]["zenith"]["env"]["ZENITH_WORKER_PROVIDER"] == "codex" + assert (user_home / ".claude" / "skills" / "zenith" / "SKILL.md").exists() + assert (user_home / ".codex" / "skills" / "zenith" / "SKILL.md").exists() + assert (user_home / ".agents" / "skills" / "scrutiny-validator" / "SKILL.md").exists() + shared_after = { + str(path.relative_to(shared_root)): path.read_bytes() + for path in shared_root.rglob("*") + if path.is_file() + } + assert shared_after == shared_before + + def test_user_scope_argument_errors_happen_before_writes( + self, + runner: CliRunner, + workspace: Path, + user_home: Path, + env: dict[str, str], + ) -> None: + conflicting = runner.invoke( + cli, + [ + "init", + "--scope", + "user", + "--workspace-dir", + str(workspace), + "--agent", + "codex", + ], + ) + assert conflicting.exit_code != 0 + assert "--workspace-dir cannot be used with --scope user" in conflicting.output + + unsupported = runner.invoke( + cli, ["init", "--scope", "user", "--agent", "hermes"] + ) + assert unsupported.exit_code != 0 + assert "use --scope project for hermes" in unsupported.output + assert list(user_home.iterdir()) == [] + + def test_registered_command_launches_from_another_workspace( + self, + runner: CliRunner, + user_home: Path, + env: dict[str, str], + tmp_path: Path, + ) -> None: + result = runner.invoke(cli, ["init", "--scope", "user", "--agent", "codex"]) + assert result.exit_code == 0, result.output + config = tomllib.loads((user_home / ".codex" / "config.toml").read_text()) + server = config["mcp_servers"]["zenith"] + unrelated = tmp_path / "unrelated workspace" + unrelated.mkdir() + + launched = subprocess.run( + [server["command"], *server["args"], "--help"], + cwd=unrelated, + env={**os.environ, **server["env"]}, + text=True, + capture_output=True, + check=False, + ) + assert launched.returncode == 0, launched.stderr + assert "Zenith MCP Server" in launched.stdout + assert not (unrelated / ".codex").exists() + assert not (unrelated / ".zenith").exists() + class TestListProjects: def test_empty(self, runner: CliRunner, env: dict[str, str]) -> None: