atlas-iac/services/hermes/scripts/configure_agent_clients.py

#!/usr/bin/env python3
"""Apply persistent non-interactive settings for managed coding clients."""

from __future__ import annotations

import json
import os
import re
import shutil
import subprocess
from pathlib import Path
from typing import Any

import yaml


def configure_claude_state(path: Path) -> None:
    """Disable Claude's long-session resume chooser without losing state."""
    value: dict[str, Any] = {}
    if path.is_file():
        try:
            loaded = json.loads(path.read_text(encoding="utf-8"))
        except (OSError, json.JSONDecodeError):
            loaded = {}
        if isinstance(loaded, dict):
            value = loaded
    value["bypassPermissionsModeAccepted"] = True
    value["resumeReturnDismissed"] = True
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(value, indent=2) + "\n", encoding="utf-8")
    os.chmod(path, 0o600)


def configure_claude_settings(path: Path) -> None:
    """Layer unattended deny rules and a command hook onto Claude settings."""
    value: dict[str, Any] = {}
    if path.is_file():
        try:
            loaded = json.loads(path.read_text(encoding="utf-8"))
        except (OSError, json.JSONDecodeError):
            loaded = {}
        if isinstance(loaded, dict):
            value = loaded

    permissions = value.get("permissions")
    if not isinstance(permissions, dict):
        permissions = {}
    legacy_managed_denies = [
        "Bash(kubectl apply *)",
        "Bash(kubectl delete *)",
        "Bash(kubectl patch *)",
        "Bash(kubectl scale *)",
        "Bash(kubectl exec *)",
        "Bash(kubectl port-forward *)",
        "Bash(flux reconcile *)",
        "Bash(flux suspend *)",
        "Bash(flux resume *)",
        "Bash(vault kv *)",
    ]
    managed_denies = [
        "Bash(git push --force *)",
        "Bash(git reset --hard *)",
        "Bash(git clean -f *)",
    ]
    existing_denies = permissions.get("deny")
    if not isinstance(existing_denies, list):
        existing_denies = []
    previous_managed = {*legacy_managed_denies, *managed_denies}
    existing_denies = [
        str(item) for item in existing_denies if str(item) not in previous_managed
    ]
    permissions["deny"] = [
        *existing_denies,
        *managed_denies,
    ]
    value["permissions"] = permissions

    hooks = value.get("hooks")
    if not isinstance(hooks, dict):
        hooks = {}
    managed_hook = {
        "matcher": "Bash",
        "hooks": [
            {
                "type": "command",
                "command": "/opt/hermes/.venv/bin/python /opt/coordinator/claude_command_policy.py",
                "timeout": 10,
            }
        ],
    }
    existing_hooks = hooks.get("PreToolUse")
    if not isinstance(existing_hooks, list):
        existing_hooks = []
    existing_hooks = [
        item
        for item in existing_hooks
        if "claude_command_policy.py" not in json.dumps(item, sort_keys=True)
    ]
    hooks["PreToolUse"] = [*existing_hooks, managed_hook]
    value["hooks"] = hooks
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(value, indent=2) + "\n", encoding="utf-8")
    os.chmod(path, 0o600)


OWNER_PERMISSIONS_BEGIN = "# begin hermes-agent owner permissions"
OWNER_PERMISSIONS_END = "# end hermes-agent owner permissions"


def configure_codex_owner_permissions(path: Path) -> None:
    """Use the stable Codex 0.147 approval and sandbox configuration keys."""
    content = path.read_text(encoding="utf-8") if path.is_file() else ""
    content = re.sub(
        rf"(?ms)^{re.escape(OWNER_PERMISSIONS_BEGIN)}\n.*?^{re.escape(OWNER_PERMISSIONS_END)}\n?",
        "",
        content,
    )

    # Remove stale root-level values from the pre-0.147 Hermes migration.
    # Stop at the first TOML table so identically named nested keys remain intact.
    lines = content.splitlines()
    first_table = next(
        (index for index, line in enumerate(lines) if line.lstrip().startswith("[")),
        len(lines),
    )
    root = [
        line
        for line in lines[:first_table]
        if not re.match(
            r"^\s*(default_permissions|approval_policy|sandbox_mode)\s*=",
            line,
        )
    ]
    remainder = lines[first_table:]
    owner_block = [
        OWNER_PERMISSIONS_BEGIN,
        'approval_policy = "never"',
        'sandbox_mode = "danger-full-access"',
        OWNER_PERMISSIONS_END,
        "",
    ]
    rendered = "\n".join([*owner_block, *root, *remainder]).rstrip() + "\n"
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(rendered, encoding="utf-8")
    os.chmod(path, 0o600)


def validate_codex_runtime(codex_bin: str = "codex") -> None:
    """Fail pod initialization when Codex cannot load its config or auth."""
    resolved = shutil.which(codex_bin)
    if resolved is None:
        raise RuntimeError(f"Codex executable not found: {codex_bin}")
    result = subprocess.run(
        [resolved, "login", "status"],
        check=False,
        capture_output=True,
        text=True,
        timeout=20,
    )
    if result.returncode != 0:
        detail = (result.stderr or result.stdout or "unknown error").strip()
        raise RuntimeError(f"Codex runtime validation failed: {detail}")


def configure_codex_runtime(
    config_path: Path,
    migrate_fn=None,
    codex_home: Path | None = None,
) -> Any:
    """Expose Hermes tools to Codex and keep the owner lane unattended."""
    config = yaml.safe_load(config_path.read_text(encoding="utf-8")) or {}
    if migrate_fn is None:
        from hermes_cli.codex_runtime_plugin_migration import migrate

        migrate_fn = migrate
    if codex_home is None:
        codex_home = Path(
            os.environ.get(
                "CODEX_HOME",
                str(Path(os.environ.get("HOME", "/opt/data/home")) / ".codex"),
            )
        )
    report = migrate_fn(
        config,
        codex_home=codex_home,
        default_permission_profile=None,
    )
    if report.errors:
        raise RuntimeError("; ".join(str(item) for item in report.errors))
    configure_codex_owner_permissions(codex_home / "config.toml")
    print(report.summary())
    return report


def main() -> None:
    """Configure provider clients under the persistent Hermes home."""
    home = Path(os.environ.get("CLAUDE_CONFIG_DIR", "/opt/data/home/.claude"))
    configure_claude_state(home / ".claude.json")
    configure_claude_settings(home / "settings.json")
    hermes_home = Path(os.environ.get("HERMES_HOME", "/opt/data"))
    configure_codex_runtime(hermes_home / "config.yaml")
    validate_codex_runtime()


if __name__ == "__main__":
    main()