| 1 | from __future__ import annotations |
| 2 | |
| 3 | import os |
| 4 | import shutil |
| 5 | from abc import ABC, abstractmethod |
| 6 | from pathlib import Path |
| 7 | from typing import Any |
| 8 | |
| 9 | from helpers import files |
| 10 | |
| 11 | |
| 12 | class TerminalAgentAdapter(ABC): |
| 13 | """Contract every terminal coding agent status adapter must implement.""" |
| 14 | |
| 15 | id: str = "" |
| 16 | title: str = "" |
| 17 | binary: str = "" |
| 18 | install_hint: str = "" |
| 19 | description: str = "" |
| 20 | |
| 21 | # --- environment ----------------------------------------------------- |
| 22 | |
| 23 | def data_dir(self) -> Path: |
| 24 | """Plugin-owned private directory for this adapter (auth, state).""" |
| 25 | path = Path( |
| 26 | files.get_abs_path("usr", "plugins", "_orchestrator", "data", self.id) |
| 27 | ) |
| 28 | path.mkdir(parents=True, exist_ok=True) |
| 29 | return path |
| 30 | |
| 31 | def resolve_binary(self, config: dict[str, Any] | None = None) -> str: |
| 32 | cfg = config if isinstance(config, dict) else {} |
| 33 | configured = str(cfg.get("binary") or "").strip() |
| 34 | return configured or self.binary |
| 35 | |
| 36 | def is_installed(self, config: dict[str, Any] | None = None) -> bool: |
| 37 | binary = self.resolve_binary(config) |
| 38 | if not binary: |
| 39 | return False |
| 40 | if os.path.isabs(binary): |
| 41 | return Path(binary).is_file() and os.access(binary, os.X_OK) |
| 42 | return shutil.which(binary) is not None |
| 43 | |
| 44 | # --- authentication ---------------------------------------------------- |
| 45 | |
| 46 | @abstractmethod |
| 47 | def auth_status(self, config: dict[str, Any] | None = None) -> dict[str, Any]: |
| 48 | """Return {connected: bool, mode: 'plugin'|'external'|'', auth_path: str}.""" |
| 49 | |
| 50 | def supports_device_login(self) -> bool: |
| 51 | return False |
| 52 | |
| 53 | def start_device_login(self) -> dict[str, Any]: |
| 54 | raise NotImplementedError(f"{self.id} does not support device login.") |
| 55 | |
| 56 | def poll_device_login(self, payload: dict[str, Any]) -> dict[str, Any]: |
| 57 | raise NotImplementedError(f"{self.id} does not support device login.") |
| 58 | |
| 59 | def can_disconnect(self, config: dict[str, Any] | None = None) -> bool: |
| 60 | return False |
| 61 | |
| 62 | def disconnect(self, config: dict[str, Any] | None = None) -> dict[str, Any]: |
| 63 | raise NotImplementedError(f"{self.id} does not support disconnect.") |