"""Preflight checks for OpenOPC external-agent integration.""" from __future__ import annotations import os import shlex import shutil import subprocess import uuid from dataclasses import dataclass, field from pathlib import Path from typing import Any from opc.core.config import OPCConfig from opc.core.models import Task from opc.layer3_agent.adapters.base import ExternalAgentAdapter from opc.layer3_agent.adapters.registry import ADAPTER_CLASSES from opc.layer3_agent.skill_installer import ( install_collab_surface, opc_collab_executable, ) _HELP_COMMANDS: dict[str, tuple[str, ...]] = { "codex": ("exec", "--help"), "claude_code": ("--help",), "cursor": ("--help",), "opencode": ("run", "--help"), } _VERSION_COMMANDS: dict[str, tuple[str, ...]] = { "codex": ("--version",), "claude_code": ("--version",), "cursor": ("--version",), "opencode": ("--version",), } @dataclass class PathProbeResult: name: str path: str ok: bool error: str = "" def as_dict(self) -> dict[str, Any]: return { "name": self.name, "path": self.path, "ok": self.ok, "error": self.error, } @dataclass class ExternalAgentPreflightResult: agent: str enabled: bool command: str available: bool binary: str = "" version: str = "" launch_command: str = "" stdin_policy: str = "" collaboration_rpc_transport: str = "" isolated_home: str = "" collab_cli: str = "" issues: list[str] = field(default_factory=list) warnings: list[str] = field(default_factory=list) write_checks: list[PathProbeResult] = field(default_factory=list) @property def ok(self) -> bool: return bool(self.enabled and self.available and not self.issues) def as_dict(self) -> dict[str, Any]: return { "agent": self.agent, "enabled": self.enabled, "command": self.command, "available": self.available, "binary": self.binary, "version": self.version, "launch_command": self.launch_command, "stdin_policy": self.stdin_policy, "collaboration_rpc_transport": self.collaboration_rpc_transport, "isolated_home": self.isolated_home, "collab_cli": self.collab_cli, "issues": list(self.issues), "warnings": list(self.warnings), "write_checks": [check.as_dict() for check in self.write_checks], "ok": self.ok, } class ExternalAgentPreflightError(RuntimeError): """Raised when the workspace permission contract is not writable.""" def __init__(self, checks: list[PathProbeResult]) -> None: self.checks = checks failures = [check for check in checks if not check.ok] detail = "; ".join(f"{item.name}={item.path}: {item.error}" for item in failures) super().__init__(f"External agent workspace permission preflight failed: {detail}") def ensure_external_agent_surfaces( config: OPCConfig, *, opc_home: Path | None = None, ) -> dict[str, dict[str, str]]: """Provision isolated agent homes and the shared opc-collab shim. This is intentionally safe to run from ``opc init`` and ``opc status``: surfaces are idempotent and user config is mirrored into OpenOPC-owned isolated homes instead of mutating the user's real agent configuration. """ base_home = Path(opc_home) if opc_home else _get_opc_home() surfaces: dict[str, dict[str, str]] = {} for agent_name, adapter_cls in ADAPTER_CLASSES.items(): agent_config = config.agents.agents.get(agent_name) adapter = adapter_cls(config=agent_config) if agent_config and not agent_config.enabled: continue slug = adapter.agent_isolation_home_slug() if not slug: continue home, bin_dir = install_collab_surface(slug, opc_home=base_home) adapter.post_install_agent_home(str(home)) surfaces[agent_name] = { "home": str(home), "bin_dir": str(bin_dir), "collab_cli": str(opc_collab_executable(bin_dir)), } return surfaces def probe_external_agent_write_contract( *, workspace_path: str | Path, opc_home: Path | None = None, task: Task | None = None, project_db_path: str | Path | None = None, ) -> list[PathProbeResult]: """Check every path external agents must be able to write before launch.""" base_home = Path(opc_home) if opc_home else _get_opc_home() metadata = dict(getattr(task, "metadata", {}) or {}) probes: list[tuple[str, Path, bool]] = [ ("workspace", Path(workspace_path).expanduser(), False), ("opc_home", base_home.expanduser(), False), ("opc_memory", (base_home / "memory").expanduser(), False), ] for name, key in ( ("collab_workspace", "comms_workspace_root"), ("collab_root", "comms_root"), ("output_root", "output_root"), ("target_output_dir", "target_output_dir"), ): raw = str(metadata.get(key) or "").strip() if raw: probes.append((name, Path(raw).expanduser(), False)) if project_db_path: probes.append(("project_db", Path(project_db_path).expanduser(), True)) results: list[PathProbeResult] = [] seen: set[tuple[str, str]] = set() for name, path, is_file in probes: key = (name, str(path)) if key in seen: continue seen.add(key) results.append(_probe_writable_path(name, path, is_file=is_file)) return results def assert_external_agent_write_contract( *, workspace_path: str | Path, opc_home: Path | None = None, task: Task | None = None, project_db_path: str | Path | None = None, ) -> list[PathProbeResult]: checks = probe_external_agent_write_contract( workspace_path=workspace_path, opc_home=opc_home, task=task, project_db_path=project_db_path, ) if any(not check.ok for check in checks): raise ExternalAgentPreflightError(checks) return checks def run_external_agent_preflight( config: OPCConfig, *, project_id: str = "default", workspace_path: str | Path | None = None, opc_home: Path | None = None, probe_commands: bool = True, prepare_surfaces: bool = True, ) -> list[ExternalAgentPreflightResult]: base_home = Path(opc_home) if opc_home else _get_opc_home() if workspace_path is None: workspace = _get_project_workplace(project_id) else: workspace = Path(workspace_path) workspace.mkdir(parents=True, exist_ok=True) surfaces = ( ensure_external_agent_surfaces(config, opc_home=base_home) if prepare_surfaces else {} ) sample_task = _sample_preflight_task(project_id, workspace) project_db = base_home / "projects" / project_id / "tasks.db" write_checks = probe_external_agent_write_contract( workspace_path=workspace, opc_home=base_home, task=sample_task, project_db_path=project_db, ) rpc_transport, rpc_issue = _describe_collaboration_rpc_transport() results: list[ExternalAgentPreflightResult] = [] for agent_name, adapter_cls in ADAPTER_CLASSES.items(): agent_config = config.agents.agents.get(agent_name) adapter = adapter_cls(config=agent_config) command = adapter.configured_command() enabled = bool(adapter.config.enabled) binary = adapter.resolve_binary() if enabled else None result = ExternalAgentPreflightResult( agent=agent_name, enabled=enabled, command=command, available=bool(binary), binary=str(binary or ""), write_checks=list(write_checks), collaboration_rpc_transport=rpc_transport, ) surface = surfaces.get(agent_name, {}) result.isolated_home = surface.get("home", "") result.collab_cli = surface.get("collab_cli", "") if not enabled: result.warnings.append("disabled in config") results.append(result) continue cmd: list[str] = [] try: cmd, metadata = _build_sample_invocation(adapter, sample_task, str(workspace)) result.launch_command = str( metadata.get("display_command") or metadata.get("command") or shlex.join(cmd) ) result.stdin_policy = _describe_stdin_policy(adapter, cmd, metadata) except Exception as exc: result.warnings.append(f"sample invocation unavailable: {exc}") if not binary: result.issues.append(_missing_agent_issue(agent_name, command)) if rpc_issue: result.issues.append(rpc_issue) results.append(result) continue _add_windows_wrapper_warnings(command, str(binary), result.warnings) if rpc_issue: result.issues.append(rpc_issue) if probe_commands: result.version = _probe_version(agent_name, str(binary), result.warnings) if cmd: _probe_help_flags(agent_name, str(binary), cmd, result) _probe_isolated_home(agent_name, result) for check in write_checks: if not check.ok: result.issues.append(f"{check.name} is not writable: {check.path} ({check.error})") results.append(result) return results def _describe_collaboration_rpc_transport() -> tuple[str, str]: from opc.layer4_tools.collaboration_rpc import ( OPC_COLLAB_RPC_TRANSPORT, default_collaboration_rpc_transport, resolve_collaboration_rpc_transport, ) requested = str(os.environ.get(OPC_COLLAB_RPC_TRANSPORT, "")).strip().lower() try: resolved = resolve_collaboration_rpc_transport(requested or "auto") except Exception as exc: return (f"{requested or 'auto'}(unavailable)", str(exc)) if resolved == "tcp": return ("tcp(loopback)", "") if requested and requested != resolved: return (f"{resolved}({requested})", "") return (resolved, "") def _describe_stdin_policy( adapter: ExternalAgentAdapter, cmd: list[str], metadata: dict[str, Any], ) -> str: try: return adapter.stdin_policy_for_process(cmd, metadata) except Exception: explicit_policy = str(metadata.get("stdin_policy") or "").strip() return explicit_policy or "devnull" def _missing_agent_issue(agent_name: str, command: str) -> str: if agent_name == "cursor": editor = shutil.which("cursor") agent = shutil.which("cursor-agent") if editor and not agent: return "Cursor editor found, cursor-agent missing; install Cursor Agent CLI for headless execution" return f"command not found on PATH: {command}" def _add_windows_wrapper_warnings(command: str, binary: str, warnings: list[str]) -> None: if os.name != "nt": return path = Path(binary) if path.suffix.lower() == ".ps1": warnings.append( "PowerShell execution policy may block this .ps1 wrapper; use the matching .cmd command manually" ) return if path.suffix.lower() == ".cmd": ps1 = path.with_suffix(".ps1") if ps1.exists() and Path(command).suffix == "": warnings.append( f"PowerShell may prefer {ps1.name}; use `{path.name}` manually if script execution is blocked" ) def _sample_preflight_task(project_id: str, workspace: Path) -> Task: metadata = { "workspace_root": str(workspace), "comms_workspace_root": str(workspace), "comms_root": str(workspace / ".opc-comms"), "target_output_dir": str(workspace), } return Task( title="OpenOPC external agent preflight", description="Reply with OK. This task is only used to build a launch command.", assigned_to="owner", project_id=project_id, metadata=metadata, ) def _get_opc_home() -> Path: from opc.core.config import get_opc_home return get_opc_home() def _get_project_workplace(project_id: str) -> Path: from opc.core.config import get_project_workplace return get_project_workplace(project_id) def _build_sample_invocation( adapter: ExternalAgentAdapter, task: Task, workspace: str, ) -> tuple[list[str], dict[str, Any]]: if adapter.config.run_mode == "interactive" and adapter.supports_interactive(): return adapter.build_interactive_invocation(task, workspace_path=workspace) return adapter.build_invocation(task, workspace_path=workspace) def _probe_writable_path(name: str, path: Path, *, is_file: bool) -> PathProbeResult: try: if is_file: path.parent.mkdir(parents=True, exist_ok=True) if path.exists(): with path.open("ab"): pass else: _write_delete_probe(path.parent) else: path.mkdir(parents=True, exist_ok=True) _write_delete_probe(path) resolved = path.resolve(strict=False) return PathProbeResult(name=name, path=str(resolved), ok=True) except Exception as exc: return PathProbeResult(name=name, path=str(path), ok=False, error=str(exc)) def _write_delete_probe(directory: Path) -> None: probe_path = directory / f".opc-write-probe-{uuid.uuid4().hex}.tmp" probe_path.write_text("ok", encoding="utf-8") probe_path.unlink() def _probe_version(agent_name: str, binary: str, warnings: list[str]) -> str: args = _VERSION_COMMANDS.get(agent_name) if not args: return "" try: proc = subprocess.run( [binary, *args], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, timeout=2, check=False, ) except Exception as exc: warnings.append(f"version probe failed: {exc}") return "" text = (proc.stdout or proc.stderr or "").strip().splitlines() if not text and proc.returncode != 0: warnings.append(f"version probe exited {proc.returncode}") return text[0][:200] if text else "" def _probe_help_flags( agent_name: str, binary: str, launch_cmd: list[str], result: ExternalAgentPreflightResult, ) -> None: args = _HELP_COMMANDS.get(agent_name, ("--help",)) try: proc = subprocess.run( [binary, *args], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, timeout=3, check=False, ) except Exception as exc: result.warnings.append(f"help probe failed: {exc}") return help_text = f"{proc.stdout}\n{proc.stderr}" if proc.returncode not in {0, 1, 2} and not help_text.strip(): result.warnings.append(f"help probe exited {proc.returncode}") return for flag in _launch_flags(launch_cmd): if flag in {"--", "-"}: continue if flag not in help_text: result.warnings.append(f"launch flag not advertised by help output: {flag}") def _launch_flags(cmd: list[str]) -> list[str]: flags: list[str] = [] for item in cmd[1:]: value = str(item or "").strip() if not value.startswith("-"): continue flag = value.split("=", 1)[0] flags.append(flag) return list(dict.fromkeys(flags)) def _probe_isolated_home(agent_name: str, result: ExternalAgentPreflightResult) -> None: if not result.isolated_home: return home = Path(result.isolated_home) if not home.exists(): result.issues.append(f"isolated agent home was not created: {home}") skill = home / "skills" / "opc-collab" / "SKILL.md" if not skill.exists(): result.issues.append(f"opc-collab skill missing from isolated home: {skill}") if result.collab_cli and not Path(result.collab_cli).exists(): result.issues.append(f"opc-collab executable missing: {result.collab_cli}") for source, target in _known_user_config_mirrors(agent_name, home): if source.exists() and not target.exists(): result.warnings.append(f"user config exists but was not mirrored: {source}") def _known_user_config_mirrors(agent_name: str, home: Path) -> list[tuple[Path, Path]]: if agent_name == "codex": user_home = Path.home() / ".codex" return [ (user_home / "auth.json", home / "auth.json"), (user_home / "config.toml", home / "config.toml"), ] if agent_name == "claude_code": user_home = Path.home() / ".claude" return [ (user_home / ".credentials.json", home / ".credentials.json"), (user_home / "settings.json", home / "settings.json"), (user_home / "settings.local.json", home / "settings.local.json"), (user_home / "CLAUDE.md", home / "CLAUDE.md"), ] if agent_name == "opencode": candidates: list[tuple[Path, Path]] = [] raw_env = str(os.environ.get("OPENCODE_CONFIG_DIR") or "").strip() if raw_env: source_home = Path(raw_env).expanduser() else: xdg = str(os.environ.get("XDG_CONFIG_HOME") or "").strip() source_home = Path(xdg).expanduser() / "opencode" if xdg else Path.home() / ".config" / "opencode" for name in ("opencode.json", "opencode.jsonc"): candidates.append((source_home / name, home / name)) return candidates return []